Common: add benchmarking module (close #370)

build.sbt

@@ -233,3 +233,8 @@ lazy val fs2 = project
   .enablePlugins(BuildInfoPlugin)
   .settings(BuildSettings.dockerSettings)
   .enablePlugins(BuildInfoPlugin, JavaAppPackaging, DockerPlugin)
+
+lazy val bench = project
+  .in(file("modules/bench"))
+  .dependsOn(fs2 % "test->test")
+  .enablePlugins(JmhPlugin)

modules/bench/build.sbt

@@ -0,0 +1,6 @@
+sourceDirectory in Jmh := (sourceDirectory in Test).value
+classDirectory in Jmh := (classDirectory in Test).value
+dependencyClasspath in Jmh := (dependencyClasspath in Test).value
+// rewire tasks, so that 'jmh:run' automatically invokes 'jmh:compile' (otherwise a clean 'jmh:run' would fail)
+compile in Jmh := (compile in Jmh).dependsOn(compile in Test).value
+run in Jmh := (run in Jmh).dependsOn(Keys.compile in Jmh).evaluated

modules/bench/src/test/scala/com.snowplowanalytics.snowplow.enrich.bench/EnrichBench.scala

@@ -0,0 +1,55 @@
+/*
+ * Copyright (c) 2020 Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package com.snowplowanalytics.snowplow.enrich.bench
+
+import org.openjdk.jmh.annotations._
+import java.util.concurrent.TimeUnit
+
+import cats.effect.{IO, Clock}
+
+import com.snowplowanalytics.iglu.client.Client
+
+import com.snowplowanalytics.snowplow.enrich.common.enrichments.EnrichmentRegistry
+import com.snowplowanalytics.snowplow.enrich.common.loaders.ThriftLoader
+
+import com.snowplowanalytics.snowplow.enrich.fs2.{Enrich, EnrichSpec, Payload}
+
+@State(Scope.Thread)
+@BenchmarkMode(Array(Mode.AverageTime, Mode.Throughput))
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+class EnrichBench {
+
+  implicit val ioClock: Clock[IO] = Clock.create[IO]
+
+  @Benchmark
+  def measureEnrichWithMinimalPayload(state: EnrichBench.BenchState) = {
+    Enrich.enrichWith[IO](IO.pure(EnrichmentRegistry()), Client.IgluCentral, None, (_: Option[Long]) => IO.unit)(state.raw).unsafeRunSync()
+  }
+
+  @Benchmark
+  def measureToCollectorPayload(state: EnrichBench.BenchState) = {
+    ThriftLoader.toCollectorPayload(state.raw.data, Enrich.processor)
+  }
+}
+
+object EnrichBench {
+  @State(Scope.Benchmark)
+  class BenchState {
+    var raw: Payload[IO, Array[Byte]] = _
+
+    @Setup(Level.Trial)
+    def setup(): Unit = {
+      raw = EnrichSpec.payload[IO]
+    }
+  }
+}

modules/bench/src/test/scala/com.snowplowanalytics.snowplow.enrich.bench/EtlPipelineBench.scala

@@ -0,0 +1,85 @@
+/*
+ * Copyright (c) 2020 Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package com.snowplowanalytics.snowplow.enrich.bench
+
+import org.openjdk.jmh.annotations._
+
+import java.util.concurrent.TimeUnit
+
+import cats.Id
+import cats.data.Validated
+
+import cats.effect.{IO, Clock}
+
+import io.circe.Json
+
+import com.snowplowanalytics.iglu.client.{Resolver, Client, CirceValidator}
+
+import com.snowplowanalytics.snowplow.enrich.common.EtlPipeline
+import com.snowplowanalytics.snowplow.enrich.common.adapters.AdapterRegistry
+import com.snowplowanalytics.snowplow.enrich.common.enrichments.EnrichmentRegistry
+
+import com.snowplowanalytics.snowplow.enrich.fs2.{Enrich, EnrichSpec}
+
+import org.joda.time.DateTime
+
+@State(Scope.Thread)
+@BenchmarkMode(Array(Mode.AverageTime, Mode.Throughput))
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+class EtlPipelineBench {
+
+  private implicit val ioClock: Clock[IO] = Clock.create[IO]
+
+  private implicit val idClock: Clock[Id] = new Clock[Id] {
+    final def realTime(unit: TimeUnit): Id[Long] =
+      unit.convert(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
+    final def monotonic(unit: TimeUnit): Id[Long] =
+      unit.convert(System.nanoTime(), TimeUnit.NANOSECONDS)
+  }
+
+  @Benchmark
+  def measureProcessEventsIO(state: EtlPipelineBench.BenchState) = {
+    val payload = EnrichSpec.colllectorPayload
+    EtlPipeline.processEvents[IO](state.adapterRegistry, state.enrichmentRegistryIo, Client.IgluCentral, Enrich.processor, state.dateTime, Validated.Valid(Some(payload))).unsafeRunSync()
+  }
+
+  @Benchmark
+  def measureProcessEventsId(state: EtlPipelineBench.BenchState) = {
+    val payload = EnrichSpec.colllectorPayload
+    EtlPipeline.processEvents[Id](state.adapterRegistry, state.enrichmentRegistryId, state.clientId, Enrich.processor, state.dateTime, Validated.Valid(Some(payload)))
+  }
+}
+
+object EtlPipelineBench {
+
+
+  @State(Scope.Benchmark)
+  class BenchState {
+    var dateTime: DateTime = _
+    var adapterRegistry: AdapterRegistry = _
+    var enrichmentRegistryId: EnrichmentRegistry[Id] = _
+    var enrichmentRegistryIo: EnrichmentRegistry[IO] = _
+    var clientId: Client[Id, Json] = _
+    var clientIO: Client[IO, Json] = _
+
+    @Setup(Level.Trial)
+    def setup(): Unit = {
+      dateTime = DateTime.parse("2010-06-30T01:20+02:00")
+      adapterRegistry = new AdapterRegistry()
+      enrichmentRegistryId = EnrichmentRegistry[Id]()
+      enrichmentRegistryIo = EnrichmentRegistry[IO]()
+      clientId = Client[Id, Json](Resolver(List(), None), CirceValidator)
+      clientIO = Client[IO, Json](Resolver(List(), None), CirceValidator)
+    }
+  }
+}

modules/bench/src/test/scala/com.snowplowanalytics.snowplow.enrich.bench/ThriftLoaderBench.scala

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) 2020 Snowplow Analytics Ltd. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package com.snowplowanalytics.snowplow.enrich.bench
+
+import org.openjdk.jmh.annotations._
+
+import java.util.concurrent.TimeUnit
+
+import com.snowplowanalytics.snowplow.enrich.common.loaders.ThriftLoader
+import com.snowplowanalytics.snowplow.enrich.fs2.{Enrich, EnrichSpec}
+
+@State(Scope.Thread)
+@BenchmarkMode(Array(Mode.AverageTime, Mode.Throughput))
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+class ThriftLoaderBench {
+
+  @Benchmark
+  def measureToCollectorPayload(state: EnrichBench.BenchState) =
+    ThriftLoader.toCollectorPayload(state.raw.data, Enrich.processor)
+}
+
+object ThriftLoaderBench {
+  @State(Scope.Benchmark)
+  class BenchState {
+    var data: Array[Byte] = _
+
+    @Setup(Level.Trial)
+    def setup(): Unit = {
+      data = EnrichSpec.colllectorPayload.toRaw
+    }
+  }
+}
+

project/Dependencies.scala

@@ -169,8 +169,8 @@ object Dependencies {
     val fs2PubSub        = "com.permutive"                    %% "fs2-google-pubsub-grpc"            % V.fs2PubSub
     val fs2              = "co.fs2"                           %% "fs2-core"                          % V.fs2
     val fs2Io            = "co.fs2"                           %% "fs2-io"                            % V.fs2
-    val monocle          = "com.github.julien-truffaut"       %% "monocle-core"                      % V.monocle
-    val monocleMacro     = "com.github.julien-truffaut"       %% "monocle-macro"                     % V.monocle
+    val monocle          = "com.github.julien-truffaut"       %% "monocle-core"                      % V.monocle         % Test
+    val monocleMacro     = "com.github.julien-truffaut"       %% "monocle-macro"                     % V.monocle         % Test
     val http4sClient     = "org.http4s"                       %% "http4s-blaze-client"               % V.http4s
     val log4cats         = "io.chrisdavenport"                %% "log4cats-slf4j"                    % V.log4cats
     val catsRetry        = "com.github.cb372"                 %% "cats-retry"                        % V.catsRetry

project/plugins.sbt

@@ -7,3 +7,4 @@ addSbtPlugin("org.foundweekends" % "sbt-bintray" % "0.5.4")
 addSbtPlugin("org.scoverage" % "sbt-coveralls" % "1.2.7")
 addSbtPlugin("org.scoverage" % "sbt-scoverage" % "1.6.1")
 addSbtPlugin("net.virtual-void" % "sbt-dependency-graph" % "0.10.0-RC1")
+addSbtPlugin("pl.project13.scala" % "sbt-jmh" % "0.4.0")