Pekko http

instrumentation/apache-pekko-http-2.13_1/README.md

@@ -0,0 +1,6 @@
+## Pekko Http Instrumentation
+
+This module adds support for `pekko-http`. 
+
+Pekko Http is a fork of Akka Http 10.2.0. Consequently, this instrumentation is a direct lift of the existing `akka-http-2.13_10.1.8` instrumentation module. 
+

instrumentation/apache-pekko-http-2.13_1/build.gradle

@@ -0,0 +1,37 @@
+apply plugin: 'scala'
+
+isScalaProjectEnabled(project, "scala-2.13")
+
+sourceSets.test.scala.srcDir "src/test/java"
+sourceSets.test.java.srcDirs = []
+
+jar {
+    manifest { attributes 'Implementation-Title': 'com.newrelic.instrumentation.apache-pekko-http-2.13_1' }
+}
+
+dependencies {
+    implementation(project(":agent-bridge"))
+    implementation(project(":newrelic-weaver-api"))
+    implementation(project(":newrelic-weaver-scala-api"))
+    implementation("org.apache.pekko:pekko-http_2.13:1.0.1")
+    implementation("org.apache.pekko:pekko-stream_2.13:1.0.1")
+    implementation("org.apache.pekko:pekko-actor_2.13:1.0.1")
+
+    testImplementation(project(":instrumentation:apache-pekko-1")) { transitive = false }
+    testImplementation(project(":instrumentation:scala-2.13.0")) { transitive = false }
+    testImplementation("com.jayway.restassured:rest-assured:2.7.0")
+    testImplementation("jakarta.xml.ws:jakarta.xml.ws-api:2.3.3")
+}
+
+verifyInstrumentation {
+    passesOnly('org.apache.pekko:pekko-http_2.13:[1.0.0,)') {
+        implementation("org.apache.pekko:pekko-stream_2.13:1.0.0")
+    }
+    excludeRegex 'org.apache.pekko:pekko-http_2.13:.*(RC|M)[0-9]*$'
+    excludeRegex 'org.apache.pekko:pekko-http_2.13:.*-[0-9a-f]{8}$'
+}
+
+site {
+    title 'Pekko Http'
+    type 'Framework'
+}

instrumentation/apache-pekko-http-2.13_1/src/main/scala/com/agent/instrumentation/org.apache.pekko/http/Function0Wrapper.java

@@ -0,0 +1,28 @@
+/*
+ *
+ *  * Copyright 2024 New Relic Corporation. All rights reserved.
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package com.agent.instrumentation.org.apache.pekko.http;
+
+import scala.Function0;
+import scala.concurrent.Future;
+import scala.runtime.AbstractFunction0;
+
+public class Function0Wrapper<T> extends AbstractFunction0<Future<T>> {
+
+    private final Function0<Future<T>> original;
+
+    public Function0Wrapper(Function0<Future<T>> original) {
+        this.original = original;
+    }
+
+    @Override
+    public Future<T> apply() {
+        Future<T> result = original.apply();
+        return new FutureWrapper<>(result);
+    }
+
+}

instrumentation/apache-pekko-http-2.13_1/src/main/scala/com/agent/instrumentation/org.apache.pekko/http/Function1Wrapper.java

@@ -0,0 +1,38 @@
+/*
+ *
+ *  * Copyright 2024 New Relic Corporation. All rights reserved.
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package com.agent.instrumentation.org.apache.pekko.http;
+
+import com.newrelic.agent.bridge.AgentBridge;
+import com.newrelic.api.agent.Token;
+import com.newrelic.api.agent.Trace;
+import com.newrelic.api.agent.weaver.Weaver;
+import scala.Function1;
+import scala.runtime.AbstractFunction1;
+
+public class Function1Wrapper<T, U> extends AbstractFunction1<T, U> {
+
+    private final Function1<T, U> original;
+    private final Token token;
+
+    public Function1Wrapper(Function1<T, U> original, Token token) {
+        this.original = original;
+        this.token = token;
+    }
+
+    @Override
+    @Trace(async = true)
+    public U apply(T v1) {
+        try {
+            token.linkAndExpire();
+        } catch (Throwable t) {
+            AgentBridge.instrumentation.noticeInstrumentationError(t, Weaver.getImplementationTitle());
+        }
+        return original.apply(v1);
+    }
+
+}

instrumentation/apache-pekko-http-2.13_1/src/main/scala/com/agent/instrumentation/org.apache.pekko/http/FutureWrapper.java

@@ -0,0 +1,156 @@
+/*
+ *
+ *  * Copyright 2024 New Relic Corporation. All rights reserved.
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package com.agent.instrumentation.org.apache.pekko.http;
+
+import com.newrelic.agent.bridge.AgentBridge;
+import com.newrelic.api.agent.NewRelic;
+import com.newrelic.api.agent.weaver.Weaver;
+import scala.$less$colon$less;
+import scala.Function1;
+import scala.Function2;
+import scala.Option;
+import scala.PartialFunction;
+import scala.Tuple2;
+import scala.concurrent.Awaitable;
+import scala.concurrent.CanAwait;
+import scala.concurrent.ExecutionContext;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+import scala.reflect.ClassTag;
+import scala.util.Try;
+
+import java.util.concurrent.TimeoutException;
+
+public class FutureWrapper<T> implements Future<T> {
+
+    private final Future<T> original;
+
+    public FutureWrapper(Future<T> original) {
+        this.original = original;
+    }
+
+    @Override
+    public <U> void onComplete(Function1<Try<T>, U> f, ExecutionContext executor) {
+        try {
+            f = new Function1Wrapper<>(f, NewRelic.getAgent().getTransaction().getToken());
+        } catch (Throwable t) {
+            AgentBridge.instrumentation.noticeInstrumentationError(t, Weaver.getImplementationTitle());
+        }
+        original.onComplete(f, executor);
+    }
+
+    @Override
+    public boolean isCompleted() {
+        return original.isCompleted();
+    }
+
+    @Override
+    public Option<Try<T>> value() {
+        return original.value();
+    }
+
+    @Override
+    public Future<Throwable> failed() {
+        return original.failed();
+    }
+
+    @Override
+    public <U> void foreach(Function1<T, U> f, ExecutionContext executor) {
+        original.foreach(f, executor);
+    }
+
+    @Override
+    public <S> Future<S> transform(Function1<T, S> s, Function1<Throwable, Throwable> f, ExecutionContext executor) {
+        return original.transform(s, f, executor);
+    }
+
+    @Override
+    public <S> Future<S> transform(Function1<Try<T>, Try<S>> f, ExecutionContext executor) {
+        return original.transform(f, executor);
+    }
+
+    @Override
+    public <S> Future<S> transformWith(Function1<Try<T>, Future<S>> f, ExecutionContext executor) {
+        return original.transformWith(f, executor);
+    }
+
+    @Override
+    public <S> Future<S> map(Function1<T, S> f, ExecutionContext executor) {
+        return original.map(f, executor);
+    }
+
+    @Override
+    public <S> Future<S> flatMap(Function1<T, Future<S>> f, ExecutionContext executor) {
+        return original.flatMap(f, executor);
+    }
+
+    @Override
+    public <S> Future<S> flatten($less$colon$less<T, Future<S>> ev) {
+        return original.flatten(ev);
+    }
+
+    @Override
+    public Future<T> filter(Function1<T, Object> p, ExecutionContext executor) {
+        return original.filter(p, executor);
+    }
+
+    @Override
+    public Future<T> withFilter(Function1<T, Object> p, ExecutionContext executor) {
+        return original.withFilter(p, executor);
+    }
+
+    @Override
+    public <S> Future<S> collect(PartialFunction<T, S> pf, ExecutionContext executor) {
+        return original.collect(pf, executor);
+    }
+
+    @Override
+    public <U> Future<U> recover(PartialFunction<Throwable, U> pf, ExecutionContext executor) {
+        return original.recover(pf, executor);
+    }
+
+    @Override
+    public <U> Future<U> recoverWith(PartialFunction<Throwable, Future<U>> pf, ExecutionContext executor) {
+        return original.recoverWith(pf, executor);
+    }
+
+    @Override
+    public <U> Future<Tuple2<T, U>> zip(Future<U> that) {
+        return original.zip(that);
+    }
+
+    @Override
+    public <U, R> Future<R> zipWith(Future<U> that, Function2<T, U, R> f, ExecutionContext executor) {
+        return original.zipWith(that, f, executor);
+    }
+
+    @Override
+    public <U> Future<U> fallbackTo(Future<U> that) {
+        return original.fallbackTo(that);
+    }
+
+    @Override
+    public <S> Future<S> mapTo(ClassTag<S> tag) {
+        return original.mapTo(tag);
+    }
+
+    @Override
+    public <U> Future<T> andThen(PartialFunction<Try<T>, U> pf, ExecutionContext executor) {
+        return original.andThen(pf, executor);
+    }
+
+    @Override
+    public Awaitable<T> ready(Duration atMost, CanAwait permit) throws InterruptedException, TimeoutException {
+        return original.ready(atMost, permit);
+    }
+
+    @Override
+    public T result(Duration atMost, CanAwait permit) throws TimeoutException, InterruptedException {
+        return original.result(atMost, permit);
+    }
+}

instrumentation/apache-pekko-http-2.13_1/src/main/scala/com/agent/instrumentation/org.apache.pekko/http/InboundWrapper.scala

@@ -0,0 +1,35 @@
+/*
+ *
+ *  * Copyright 2024 New Relic Corporation. All rights reserved.
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package com.agent.instrumentation.org.apache.pekko.http
+
+import org.apache.pekko.http.scaladsl.model.HttpRequest
+import com.newrelic.api.agent.{ExtendedInboundHeaders, HeaderType}
+
+import java.util
+import scala.jdk.javaapi.CollectionConverters
+
+class InboundWrapper(request: HttpRequest) extends ExtendedInboundHeaders {
+
+  def getHeaderType: HeaderType = {
+    HeaderType.HTTP
+  }
+
+  def getHeader(name: String): String = {
+    request.headers.find(header => header.is(name.toLowerCase)).map(header => header.value).orNull
+  }
+
+  override def getHeaders(name: String): util.List[String] = {
+    val headers = request.headers.filter(header => header.is(name.toLowerCase)).map(header => header.value)
+    if (headers.isEmpty) {
+      return null
+    }
+    CollectionConverters.asJava(headers)
+  }
+
+}
+

instrumentation/apache-pekko-http-2.13_1/src/main/scala/com/agent/instrumentation/org.apache.pekko/http/OutboundWrapper.scala

@@ -0,0 +1,28 @@
+/*
+ *
+ *  * Copyright 2024 New Relic Corporation. All rights reserved.
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package com.agent.instrumentation.org.apache.pekko.http
+
+import org.apache.pekko.http.scaladsl.model.HttpResponse
+import org.apache.pekko.http.scaladsl.model.headers.RawHeader
+import com.newrelic.api.agent.{HeaderType, OutboundHeaders}
+
+class OutboundWrapper(var response: HttpResponse) extends OutboundHeaders {
+
+  def getHeaderType: HeaderType = {
+    HeaderType.HTTP
+  }
+
+  def setHeader(name: String, value: String): Unit = {
+    response = response.addHeader(new RawHeader(name, value))
+  }
+
+  def getHeader(name: String): String = {
+    response.headers.find(header => header.is(name.toLowerCase)).map(header => header.value).orNull
+  }
+
+}