Merge pull request #67 from mmatloka/feature/prometheus

Prometheus backend

6 files changed, 261 insertions, 2 deletions

diff --git a/build.sbt b/build.sbt
index 33b51a9..775a64c 100644
--- a/build.sbt
+++ b/build.sbt
@@ -58,6 +58,7 @@ lazy val rootProject = (project in file("."))
     circe,
     json4s,
     braveBackend,
+    prometheusBackend,
     tests
   )
 
@@ -188,6 +189,17 @@ lazy val braveBackend: Project = (project in file("metrics/brave-backend"))
     )
   ).dependsOn(core)
 
+lazy val prometheusBackend: Project = (project in file("metrics/prometheus-backend"))
+  .settings(commonSettings: _*)
+  .settings(
+    name := "prometheus-backend",
+    libraryDependencies ++= Seq(
+      "io.prometheus" % "simpleclient" % "0.3.0",
+      scalaTest % "test"
+    )
+  )
+  .dependsOn(core)
+
 lazy val tests: Project = (project in file("tests"))
   .settings(commonSettings: _*)
   .settings(
diff --git a/docs/backends/prometheus.rst b/docs/backends/prometheus.rst
new file mode 100644
index 0000000..9de7886
--- /dev/null
+++ b/docs/backends/prometheus.rst
@@ -0,0 +1,30 @@
+.. _prometheus_backend:
+
+Prometheus backend
+=============
+
+To use, add the following dependency to your project::
+
+  "com.softwaremill.sttp" %% "prometheus-backend" % "1.1.6"
+
+This backend depends on `Prometheus JVM Client <https://github.com/prometheus/client_java>`_. Keep in mind this backend registers histograms and gathers request times, but you have to expose those metrics to `Prometheus <https://prometheus.io/>`_ e.g. using  `prometheus-akka-http <https://github.com/lonelyplanet/prometheus-akka-http>`_.
+
+The Prometheus backend wraps any other backend, for example::
+
+  implicit val sttpBackend = PrometheusBackend(AkkaHttpBackend())
+
+It gathers request execution times in ``Histogram``. It uses by default ``sttp_request_latency`` name, defined in ``PrometheusBackend.DefaultHistogramName``. It is possible to define custom histograms name by passing function mapping request to histogram name::
+
+  implicit val sttpBackend = PrometheusBackend(AkkaHttpBackend(), request => Some(request.uri.host))
+
+You can disable request histograms by passing ``None`` returning function::
+
+  implicit val sttpBackend = PrometheusBackend(AkkaHttpBackend(), _ => None)
+
+This backend also offers ``Gauge`` with currently in-progress requests number. It uses by default ``sttp_requests_in_progress`` name, defined in ``PrometheusBackend.DefaultRequestsInProgressGaugeName``. It is possible to define custom gauge name by passing function mapping request to gauge name::
+
+   implicit val sttpBackend = PrometheusBackend(AkkaHttpBackend(), requestToInProgressGaugeNameMapper = request => Some(request.uri.host))
+
+You can disable request in-progress gauges by passing ``None`` returning function::
+
+  implicit val sttpBackend = PrometheusBackend(AkkaHttpBackend(), requestToInProgressGaugeNameMapper = _ => None)
\ No newline at end of file
diff --git a/docs/backends/summary.rst b/docs/backends/summary.rst
index 9fb9dec..11e9b55 100644
--- a/docs/backends/summary.rst
+++ b/docs/backends/summary.rst
@@ -17,7 +17,7 @@ Below is a summary of all the backends. See the sections on individual backend i
 ================================ ============================ ================================================
 Class                            Response wrapper             Supported stream type
 ================================ ============================ ================================================
-``HttpURLConnectionBackend``     None (``Id``)                n/a 
+``HttpURLConnectionBackend``     None (``Id``)                n/a
 ``TryHttpURLConnectionBackend``  ``scala.util.Try``           n/a
 ``AkkaHttpBackend``              ``scala.concurrent.Future``  ``akka.stream.scaladsl.Source[ByteString, Any]``
 ``AsyncHttpClientFutureBackend`` ``scala.concurrent.Future``  n/a
@@ -33,4 +33,5 @@ Class                            Response wrapper             Supported stream t
 There are also backends which wrap other backends to provide additional functionality. These include:
 
 * ``TryBackend``, which safely wraps any exceptions thrwon by a synchronous backend in ``scala.util.Try``
-* ``BraveBackend``, for Zipkin-compatible distributed tracing. See the :ref:`dedicated section <brave_backend>`.
\ No newline at end of file
+* ``BraveBackend``, for Zipkin-compatible distributed tracing. See the :ref:`dedicated section <brave_backend>`.
+* ``PrometheusBackend``, for gathering Prometheus-format metrics. See the :ref:`dedicated section <prometheus_backend>`.
\ No newline at end of file
diff --git a/docs/index.rst b/docs/index.rst
index 1f97373..0ea1cda 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -68,6 +68,7 @@ For more examples, see the :ref:`usage examples <usage_examples>` section. Or ex
    backends/asynchttpclient
    backends/okhttp
    backends/brave
+   backends/prometheus
    backends/custom
 
 .. toctree::
diff --git a/metrics/prometheus-backend/src/main/scala/com/softwaremill/sttp/prometheus/PrometheusBackend.scala b/metrics/prometheus-backend/src/main/scala/com/softwaremill/sttp/prometheus/PrometheusBackend.scala
new file mode 100644
index 0000000..173d9c6
--- /dev/null
+++ b/metrics/prometheus-backend/src/main/scala/com/softwaremill/sttp/prometheus/PrometheusBackend.scala
@@ -0,0 +1,69 @@
+package com.softwaremill.sttp.prometheus
+
+import java.util.concurrent.ConcurrentHashMap
+
+import com.softwaremill.sttp.{FollowRedirectsBackend, MonadError, Request, Response, SttpBackend}
+import io.prometheus.client.{Gauge, Histogram}
+
+import scala.collection.mutable
+import scala.language.higherKinds
+import scala.collection.JavaConverters._
+
+class PrometheusBackend[R[_], S] private (delegate: SttpBackend[R, S],
+                                          requestToHistogramNameMapper: Request[_, S] => Option[String],
+                                          requestToInProgressGaugeNameMapper: Request[_, S] => Option[String])
+    extends SttpBackend[R, S] {
+
+  private[this] val histograms: mutable.Map[String, Histogram] = new ConcurrentHashMap[String, Histogram]().asScala
+  private[this] val gauges: mutable.Map[String, Gauge] = new ConcurrentHashMap[String, Gauge]().asScala
+
+  override def send[T](request: Request[T, S]): R[Response[T]] = {
+    val requestTimer: Option[Histogram.Timer] = for {
+      histogramName: String <- requestToHistogramNameMapper(request)
+      histogram: Histogram = histograms.getOrElseUpdate(histogramName, createNewHistogram(histogramName))
+    } yield histogram.startTimer()
+
+    val gauge: Option[Gauge] = for {
+      gaugeName: String <- requestToInProgressGaugeNameMapper(request)
+    } yield gauges.getOrElseUpdate(gaugeName, createNewGauge(gaugeName))
+
+    gauge.foreach(_.inc())
+
+    responseMonad.handleError(
+      responseMonad.map(delegate.send(request)) { response =>
+        requestTimer.foreach(_.observeDuration())
+        gauge.foreach(_.dec())
+        response
+      }
+    ) {
+      case e: Exception =>
+        requestTimer.foreach(_.observeDuration())
+        gauge.foreach(_.dec())
+        responseMonad.error(e)
+    }
+  }
+
+  override def close(): Unit = delegate.close()
+
+  override def responseMonad: MonadError[R] = delegate.responseMonad
+
+  private[this] def createNewHistogram(name: String): Histogram = Histogram.build().name(name).help(name).register()
+
+  private[this] def createNewGauge(name: String): Gauge = Gauge.build().name(name).help(name).register()
+}
+
+object PrometheusBackend {
+
+  val DefaultHistogramName = "sttp_request_latency"
+  val DefaultRequestsInProgressGaugeName = "sttp_requests_in_progress"
+
+  def apply[R[_], S](delegate: SttpBackend[R, S],
+                     requestToHistogramNameMapper: Request[_, S] => Option[String] = (_: Request[_, S]) =>
+                       Some(DefaultHistogramName),
+                     requestToInProgressGaugeNameMapper: Request[_, S] => Option[String] = (_: Request[_, S]) =>
+                       Some(DefaultRequestsInProgressGaugeName)): SttpBackend[R, S] = {
+    // redirects should be handled before prometheus
+    new FollowRedirectsBackend(
+      new PrometheusBackend(delegate, requestToHistogramNameMapper, requestToInProgressGaugeNameMapper))
+  }
+}
diff --git a/metrics/prometheus-backend/src/test/scala/com/softwaremill/sttp/prometheus/PrometheusBackendTest.scala b/metrics/prometheus-backend/src/test/scala/com/softwaremill/sttp/prometheus/PrometheusBackendTest.scala
new file mode 100644
index 0000000..33477a0
--- /dev/null
+++ b/metrics/prometheus-backend/src/test/scala/com/softwaremill/sttp/prometheus/PrometheusBackendTest.scala
@@ -0,0 +1,146 @@
+package com.softwaremill.sttp.prometheus
+
+import java.lang
+import java.util.concurrent.CountDownLatch
+
+import com.softwaremill.sttp.testing.SttpBackendStub
+import com.softwaremill.sttp.{HttpURLConnectionBackend, Id, sttp, _}
+import io.prometheus.client.CollectorRegistry
+import org.scalatest.concurrent.{Eventually, IntegrationPatience}
+import org.scalatest.{BeforeAndAfter, FlatSpec, Matchers, OptionValues}
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+
+class PrometheusBackendTest extends FlatSpec with Matchers with BeforeAndAfter with Eventually with OptionValues {
+
+  before {
+    CollectorRegistry.defaultRegistry.clear()
+  }
+
+  it should "use default histogram name" in {
+    // given
+    val backendStub = SttpBackendStub(HttpURLConnectionBackend()).whenAnyRequest.thenRespondOk()
+    val backend = PrometheusBackend[Id, Nothing](backendStub)
+    val requestsNumber = 10
+
+    // when
+    (0 until requestsNumber).foreach(_ => backend.send(sttp.get(uri"http://127.0.0.1/foo")))
+
+    // then
+    getMetricVale(s"${PrometheusBackend.DefaultHistogramName}_count").value shouldBe requestsNumber
+  }
+
+  it should "use mapped request to histogram name" in {
+    // given
+    val customHistogramName = "my_custom_histogram"
+    val backend =
+      PrometheusBackend[Id, Nothing](SttpBackendStub(HttpURLConnectionBackend()), _ => Some(customHistogramName))
+    val requestsNumber = 5
+
+    // when
+    (0 until requestsNumber).foreach(_ => backend.send(sttp.get(uri"http://127.0.0.1/foo")))
+
+    // then
+    getMetricVale(s"${PrometheusBackend.DefaultHistogramName}_count") shouldBe empty
+    getMetricVale(s"${customHistogramName}_count").value shouldBe requestsNumber
+  }
+
+  it should "disable histograms" in {
+    // given
+    val backend =
+      PrometheusBackend[Id, Nothing](SttpBackendStub(HttpURLConnectionBackend()), _ => None)
+    val requestsNumber = 6
+
+    // when
+    (0 until requestsNumber).foreach(_ => backend.send(sttp.get(uri"http://127.0.0.1/foo")))
+
+    // then
+    getMetricVale(s"${PrometheusBackend.DefaultHistogramName}_count") shouldBe empty
+  }
+
+  it should "use default gauge name" in {
+    // given
+    val requestsNumber = 10
+    val countDownLatch = new CountDownLatch(1)
+    val backendStub = SttpBackendStub.asynchronousFuture.whenAnyRequest.thenRespondWrapped {
+      Future {
+        countDownLatch.await()
+        Response(Right(""), 200, "", Nil, Nil)
+      }
+    }
+    val backend = PrometheusBackend[Future, Nothing](backendStub)
+
+    // when
+    (0 until requestsNumber).foreach(_ => backend.send(sttp.get(uri"http://127.0.0.1/foo")))
+
+    // then
+    eventually {
+      getMetricVale(PrometheusBackend.DefaultRequestsInProgressGaugeName).value shouldBe requestsNumber
+    }
+
+    countDownLatch.countDown()
+    eventually {
+      getMetricVale(PrometheusBackend.DefaultRequestsInProgressGaugeName).value shouldBe 0
+    }
+  }
+
+  it should "use mapped request to gauge name" in {
+    // given
+    val customGaugeName = "my_custom_gauge"
+    val requestsNumber = 10
+    val countDownLatch = new CountDownLatch(1)
+    val backendStub = SttpBackendStub.asynchronousFuture.whenAnyRequest.thenRespondWrapped {
+      Future {
+        countDownLatch.await()
+        Response(Right(""), 200, "", Nil, Nil)
+      }
+    }
+    val backend =
+      PrometheusBackend[Future, Nothing](backendStub, requestToInProgressGaugeNameMapper = _ => Some(customGaugeName))
+
+    // when
+    (0 until requestsNumber).foreach(_ => backend.send(sttp.get(uri"http://127.0.0.1/foo")))
+
+    // then
+    eventually {
+      getMetricVale(PrometheusBackend.DefaultRequestsInProgressGaugeName) shouldBe empty
+      getMetricVale(customGaugeName).value shouldBe requestsNumber
+    }
+
+    countDownLatch.countDown()
+    eventually {
+      getMetricVale(PrometheusBackend.DefaultRequestsInProgressGaugeName) shouldBe empty
+      getMetricVale(customGaugeName).value shouldBe 0
+    }
+  }
+
+  it should "disable gauge" in {
+    // given
+    val requestsNumber = 10
+    val countDownLatch = new CountDownLatch(1)
+    val backendStub = SttpBackendStub.asynchronousFuture.whenAnyRequest.thenRespondWrapped {
+      Future {
+        countDownLatch.await()
+        Response(Right(""), 200, "", Nil, Nil)
+      }
+    }
+    val backend = PrometheusBackend[Future, Nothing](backendStub, requestToInProgressGaugeNameMapper = _ => None)
+
+    // when
+    (0 until requestsNumber).foreach(_ => backend.send(sttp.get(uri"http://127.0.0.1/foo")))
+
+    // then
+    getMetricVale(PrometheusBackend.DefaultRequestsInProgressGaugeName) shouldBe empty
+
+    countDownLatch.countDown()
+    eventually {
+      getMetricVale(s"${PrometheusBackend.DefaultHistogramName}_count").value shouldBe requestsNumber
+      getMetricVale(PrometheusBackend.DefaultRequestsInProgressGaugeName) shouldBe empty
+    }
+  }
+
+  private[this] def getMetricVale(name: String): Option[lang.Double] =
+    Option(CollectorRegistry.defaultRegistry.getSampleValue(name))
+
+}