path: root/kamon-newrelic/src/test/scala/kamon/newrelic/MetricReporterSpec.scala

/*
 * =========================================================================================
 * Copyright © 2013-2014 the kamon project <http://kamon.io/>
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
 * except in compliance with the License. You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the
 * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied. See the License for the specific language governing permissions
 * and limitations under the License.
 * =========================================================================================
 */

package kamon.newrelic

import akka.actor.{ ActorRef, ActorSystem }
import akka.io.IO
import akka.testkit._
import com.typesafe.config.ConfigFactory
import kamon.metric.{ TraceMetrics, Metrics }
import kamon.{ Kamon, AkkaExtensionSwap }
import kamon.metric.Subscriptions.TickMetricSnapshot
import org.scalatest.{ Matchers, WordSpecLike }
import spray.can.Http
import spray.http.Uri.Query
import spray.http._
import spray.httpx.encoding.Deflate
import spray.httpx.{ RequestBuilding, SprayJsonSupport }
import scala.concurrent.duration._
import spray.json._

class MetricReporterSpec extends TestKitBase with WordSpecLike with Matchers with RequestBuilding with SprayJsonSupport {
  import kamon.newrelic.JsonProtocol._

  implicit lazy val system: ActorSystem = ActorSystem("metric-reporter-spec", ConfigFactory.parseString(
    """
      |akka {
      |  loggers = ["akka.testkit.TestEventListener"]
      |  loglevel = "INFO"
      |}
      |kamon {
      |  metric {
      |    tick-interval = 1 hour
      |  }
      |}
      |
    """.stripMargin))

  val agentSettings = Agent.Settings("1111111111", "kamon", "test-host", 1, 1, 30 seconds, 1D)
  val baseQuery = Query(
    "license_key" -> agentSettings.licenseKey,
    "marshal_format" -> "json",
    "protocol_version" -> "12")
  val baseCollectorUri = Uri("http://collector-1.newrelic.com/agent_listener/invoke_raw_method").withQuery(baseQuery)

  "the MetricReporter" should {
    "report metrics to New Relic upon arrival" in new FakeTickSnapshotsFixture {
      val httpManager = setHttpManager(TestProbe())
      val metricReporter = system.actorOf(MetricReporter.props(agentSettings, 9999, baseCollectorUri))

      metricReporter ! firstSnapshot
      val metricPost = httpManager.expectMsgType[HttpRequest]

      metricPost.method should be(HttpMethods.POST)
      metricPost.uri should be(rawMethodUri("collector-1.newrelic.com", "metric_data"))
      metricPost.encoding should be(HttpEncodings.deflate)

      val postedBatch = Deflate.decode(metricPost).entity.asString.parseJson.convertTo[MetricBatch]
      postedBatch.runID should be(9999)
      postedBatch.timeSliceMetrics.from should be(1415587618)
      postedBatch.timeSliceMetrics.to should be(1415587678)

      val metrics = postedBatch.timeSliceMetrics.metrics
      metrics(MetricID("Apdex", None)).callCount should be(3)
      metrics(MetricID("WebTransaction", None)).callCount should be(3)
      metrics(MetricID("HttpDispatcher", None)).callCount should be(3)
    }

    "accumulate metrics if posting fails" in new FakeTickSnapshotsFixture {
      val httpManager = setHttpManager(TestProbe())
      val metricReporter = system.actorOf(MetricReporter.props(agentSettings, 9999, baseCollectorUri))

      metricReporter ! firstSnapshot
      val request = httpManager.expectMsgType[HttpRequest]
      httpManager.reply(Timedout(request))

      metricReporter ! secondSnapshot
      val metricPost = httpManager.expectMsgType[HttpRequest]

      metricPost.method should be(HttpMethods.POST)
      metricPost.uri should be(rawMethodUri("collector-1.newrelic.com", "metric_data"))
      metricPost.encoding should be(HttpEncodings.deflate)

      val postedBatch = Deflate.decode(metricPost).entity.asString.parseJson.convertTo[MetricBatch]
      postedBatch.runID should be(9999)
      postedBatch.timeSliceMetrics.from should be(1415587618)
      postedBatch.timeSliceMetrics.to should be(1415587738)

      val metrics = postedBatch.timeSliceMetrics.metrics
      metrics(MetricID("Apdex", None)).callCount should be(6)
      metrics(MetricID("WebTransaction", None)).callCount should be(6)
      metrics(MetricID("HttpDispatcher", None)).callCount should be(6)
    }
  }

  def setHttpManager(probe: TestProbe): TestProbe = {
    AkkaExtensionSwap.swap(system, Http, new IO.Extension {
      def manager: ActorRef = probe.ref
    })
    probe
  }

  def rawMethodUri(host: String, methodName: String): Uri = {
    Uri(s"http://$host/agent_listener/invoke_raw_method").withQuery(
      "method" -> methodName,
      "run_id" -> "9999",
      "license_key" -> "1111111111",
      "marshal_format" -> "json",
      "protocol_version" -> "12")
  }

  def compactJsonEntity(jsonString: String): HttpEntity = {
    import spray.json._

    val compactJson = jsonString.parseJson.compactPrint
    HttpEntity(ContentTypes.`application/json`, compactJson)
  }

  trait FakeTickSnapshotsFixture {
    val testTraceID = TraceMetrics("example-trace")
    val recorder = Kamon(Metrics).register(testTraceID, TraceMetrics.Factory).get
    val collectionContext = Kamon(Metrics).buildDefaultCollectionContext

    def collectRecorder = recorder.collect(collectionContext)

    recorder.elapsedTime.record(1000000)
    recorder.elapsedTime.record(2000000)
    recorder.elapsedTime.record(3000000)
    val firstSnapshot = TickMetricSnapshot(1415587618000L, 1415587678000L, Map(testTraceID -> collectRecorder))

    recorder.elapsedTime.record(6000000)
    recorder.elapsedTime.record(5000000)
    recorder.elapsedTime.record(4000000)
    val secondSnapshot = TickMetricSnapshot(1415587678000L, 1415587738000L, Map(testTraceID -> collectRecorder))
  }
}