Add Scala concurrent plugin

apm-agent-plugins/apm-scala-concurrent-plugin/pom.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>apm-agent-plugins</artifactId>
+        <groupId>co.elastic.apm</groupId>
+        <version>1.15.1-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>apm-scala-concurrent-plugin</artifactId>
+    <name>${project.groupId}:${project.artifactId}</name>
+
+    <properties>
+        <apm-agent-parent.base.dir>${project.basedir}/../..</apm-agent-parent.base.dir>
+    </properties>
+
+    <build>
+        <sourceDirectory>src/main/scala</sourceDirectory>
+        <testSourceDirectory>src/test/scala</testSourceDirectory>
+        <plugins>
+            <plugin>
+                <groupId>net.alchim31.maven</groupId>
+                <artifactId>scala-maven-plugin</artifactId>
+                <version>4.3.0</version>
+                <configuration>
+                    <scalaVersion>2.12.10</scalaVersion>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

apm-agent-plugins/apm-scala-concurrent-plugin/src/main/scala/co/elastic/apm/agent/concurrent/FutureInstrumentation.scala

@@ -0,0 +1,41 @@
+package co.elastic.apm.agent.concurrent
+
+import java.util
+
+import co.elastic.apm.agent.bci.ElasticApmInstrumentation
+import net.bytebuddy.asm.Advice
+import net.bytebuddy.description.`type`.TypeDescription
+import net.bytebuddy.description.method.MethodDescription
+import net.bytebuddy.matcher.ElementMatcher
+import net.bytebuddy.matcher.ElementMatchers._
+
+import scala.collection.JavaConverters._
+import scala.util.Try
+
+class FutureInstrumentation extends ElasticApmInstrumentation {
+
+  override def getTypeMatcher: ElementMatcher[_ >: TypeDescription] =
+    hasSuperType[TypeDescription](named("scala.concurrent.Future"))
+      .or(hasSuperType(named("scala.concurrent.impl.Promise")))
+      .or(hasSuperType(named("scala.concurrent.impl.Promise$Transformation")))
+      .or(hasSuperType(named("scala.concurrent.Future$")))
+
+  override def getMethodMatcher: ElementMatcher[_ >: MethodDescription] =
+    named[MethodDescription]("onComplete").and(returns(classOf[Unit])).and(takesArguments(classOf[Try[_] => _]))
+//    .or(named[MethodDescription]("transform").and(returns(classOf[Future[_]])).and(takesArguments(classOf[Try[_] => Try[_]])))
+//    .or(named[MethodDescription]("transformWith").and(returns(classOf[Future[_]])).and(takesArguments(classOf[Try[_] => Future[_]])))
+    .and(not(isTypeInitializer[MethodDescription]()))
+
+  override def getInstrumentationGroupNames: util.Collection[String] = List("concurrent", "future").asJavaCollection
+
+  @Advice.OnMethodEnter(suppress = classOf[Throwable])
+  def onComplete(@Advice.Argument(value = 0, readOnly = true) callback: Try[_] => _): Unit = {
+    val active = ElasticApmInstrumentation.getActive
+    val tracer = ElasticApmInstrumentation.tracer
+    if (active != null && tracer != null && tracer.isWrappingAllowedOnThread) {
+      active.setDiscard(false)
+      tracer.avoidWrappingOnThread()
+    }
+  }
+
+}

apm-agent-plugins/apm-scala-concurrent-plugin/src/test/scala/co/elastic/apm/agent/concurrent/FutureInstrumentationSpec.scala

@@ -0,0 +1,64 @@
+package co.elastic.apm.agent.concurrent
+
+import java.util.concurrent.Executors
+
+import co.elastic.apm.agent.AbstractInstrumentationTest
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContext, Future}
+
+class FutureInstrumentationSpec extends AbstractInstrumentationTest {
+
+  @Test
+  def testWithDefaultConfig(): Unit = {
+    new TestFutureTraceMethods().invokeAsync()
+    assertThat(AbstractInstrumentationTest.reporter.getTransactions().toArray).hasSize(1)
+    assertThat(AbstractInstrumentationTest.reporter.getSpans().toArray).hasSize(4)
+  }
+
+  private class TestFutureTraceMethods {
+
+    /**
+     * Calling this method results in this method call tree:
+     *
+     *                      main thread                         |           worker thread
+     * -------------------------------------------------------------------------------------------
+     * invokeAsync                                              |
+     *      |                                                   |
+     *      --- blockingMethodOnMainThread                      |
+     *                     |                                    |
+     *                     --- nonBlockingMethodOnMainThread    |
+     *                                      |                   |
+     *                                      --------------------------> methodOnWorkerThread
+     *                                                          |                |
+     *                                                          |                --- longMethod
+     *                                                          |
+     */
+    def invokeAsync(): Unit = blockingMethodOnMainThread()
+
+    private def blockingMethodOnMainThread(): Unit = {
+      try {
+        Await.result(nonBlockingMethodOnMainThread(), 10.seconds)
+      } catch {
+        case e: Exception => e.printStackTrace()
+      }
+    }
+
+    private def nonBlockingMethodOnMainThread(): Future[Unit] =
+      Future(methodOnWorkerThread())(ExecutionContext.fromExecutor(Executors.newFixedThreadPool(1)))
+
+    private def methodOnWorkerThread(): Unit = longMethod()
+
+    private def longMethod(): Unit = {
+      try {
+        Thread.sleep(100)
+      } catch {
+        case e: InterruptedException => e.printStackTrace()
+      }
+    }
+
+  }
+
+}

apm-agent-plugins/pom.xml

@@ -40,6 +40,7 @@
         <module>apm-jms-plugin</module>
         <module>apm-hibernate-search-plugin</module>
         <module>apm-redis-plugin</module>
+        <module>apm-scala-concurrent-plugin</module>
         <module>apm-error-logging-plugin</module>
         <module>apm-jmx-plugin</module>
         <module>apm-mule4-plugin</module>