OpAMP HTTP service (#1928)

Co-authored-by: Lauri Tulmin <[email protected]>

Author: LikeTheSalad

opamp-client/build.gradle.kts

@@ -12,8 +12,10 @@ description = "Client implementation of the OpAMP spec."
 otelJava.moduleName.set("io.opentelemetry.contrib.opamp.client")
 
 dependencies {
+  implementation("com.squareup.okhttp3:okhttp")
   annotationProcessor("com.google.auto.value:auto-value")
   compileOnly("com.google.auto.value:auto-value-annotations")
+  testImplementation("org.mockito:mockito-inline")
 }
 
 val opampReleaseInfo = tasks.register<Download>("opampLastReleaseInfo") {

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/connectivity/http/HttpErrorException.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.connectivity.http;
+
+public class HttpErrorException extends Exception {
+  private final int errorCode;
+
+  private static final long serialVersionUID = 1L;
+
+  public int getErrorCode() {
+    return errorCode;
+  }
+
+  /**
+   * Constructs an HTTP error related exception.
+   *
+   * @param errorCode The HTTP error code.
+   * @param message The HTTP error message associated with the code.
+   */
+  public HttpErrorException(int errorCode, String message) {
+    super(message);
+    this.errorCode = errorCode;
+  }
+}

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/connectivity/http/HttpSender.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.connectivity.http;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.concurrent.CompletableFuture;
+
+public interface HttpSender {
+
+  CompletableFuture<Response> send(BodyWriter writer, int contentLength);
+
+  interface BodyWriter {
+    void writeTo(OutputStream outputStream) throws IOException;
+  }
+
+  interface Response extends Closeable {
+    int statusCode();
+
+    String statusMessage();
+
+    InputStream bodyInputStream();
+
+    String getHeader(String name);
+  }
+}

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/connectivity/http/OkHttpSender.java

@@ -0,0 +1,135 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.connectivity.http;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.CompletableFuture;
+import okhttp3.Call;
+import okhttp3.Callback;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.RequestBody;
+import okio.BufferedSink;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+public final class OkHttpSender implements HttpSender {
+  private final OkHttpClient client;
+  private final String url;
+
+  public static OkHttpSender create(String url) {
+    return create(url, new OkHttpClient());
+  }
+
+  public static OkHttpSender create(String url, OkHttpClient client) {
+    return new OkHttpSender(url, client);
+  }
+
+  private static final String CONTENT_TYPE = "application/x-protobuf";
+  private static final MediaType MEDIA_TYPE = MediaType.parse(CONTENT_TYPE);
+
+  private OkHttpSender(String url, OkHttpClient client) {
+    this.url = url;
+    this.client = client;
+  }
+
+  @Override
+  public CompletableFuture<Response> send(BodyWriter writer, int contentLength) {
+    CompletableFuture<Response> future = new CompletableFuture<>();
+    okhttp3.Request.Builder builder = new okhttp3.Request.Builder().url(url);
+    builder.addHeader("Content-Type", CONTENT_TYPE);
+
+    RequestBody body = new RawRequestBody(writer, contentLength, MEDIA_TYPE);
+    builder.post(body);
+
+    client
+        .newCall(builder.build())
+        .enqueue(
+            new Callback() {
+              @Override
+              public void onResponse(@NotNull Call call, @NotNull okhttp3.Response response) {
+                if (response.isSuccessful() && response.body() != null) {
+                  future.complete(new OkHttpResponse(response));
+                } else {
+                  future.completeExceptionally(
+                      new HttpErrorException(response.code(), response.message()));
+                }
+              }
+
+              @Override
+              public void onFailure(@NotNull Call call, @NotNull IOException e) {
+                future.completeExceptionally(e);
+              }
+            });
+
+    return future;
+  }
+
+  private static class OkHttpResponse implements Response {
+    private final okhttp3.Response response;
+
+    private OkHttpResponse(okhttp3.Response response) {
+      if (response.body() == null) {
+        throw new IllegalStateException("null response body not expected");
+      }
+      this.response = response;
+    }
+
+    @Override
+    public int statusCode() {
+      return response.code();
+    }
+
+    @Override
+    public String statusMessage() {
+      return response.message();
+    }
+
+    @Override
+    public InputStream bodyInputStream() {
+      return response.body().byteStream();
+    }
+
+    @Override
+    public String getHeader(String name) {
+      return response.headers().get(name);
+    }
+
+    @Override
+    public void close() {
+      response.close();
+    }
+  }
+
+  private static class RawRequestBody extends RequestBody {
+    private final BodyWriter writer;
+    private final int contentLength;
+    private final MediaType contentType;
+
+    private RawRequestBody(BodyWriter writer, int contentLength, MediaType contentType) {
+      this.writer = writer;
+      this.contentLength = contentLength;
+      this.contentType = contentType;
+    }
+
+    @Nullable
+    @Override
+    public MediaType contentType() {
+      return contentType;
+    }
+
+    @Override
+    public long contentLength() {
+      return contentLength;
+    }
+
+    @Override
+    public void writeTo(@NotNull BufferedSink bufferedSink) throws IOException {
+      writer.writeTo(bufferedSink.outputStream());
+    }
+  }
+}

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/connectivity/http/RetryAfterParser.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.connectivity.http;
+
+import io.opentelemetry.opamp.client.internal.tools.SystemTime;
+import java.time.Duration;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Locale;
+import java.util.Optional;
+import java.util.regex.Pattern;
+
+public final class RetryAfterParser {
+  private final SystemTime systemTime;
+  private static final Pattern SECONDS_PATTERN = Pattern.compile("\\d+");
+  private static final Pattern DATE_PATTERN =
+      Pattern.compile(
+          "[A-Za-z]{3}, [0-3][0-9] [A-Za-z]{3} [0-9]{4} [0-2][0-9]:[0-5][0-9]:[0-5][0-9] GMT");
+  private static final DateTimeFormatter DATE_FORMAT =
+      DateTimeFormatter.ofPattern("EEE, dd MMM yyyy HH:mm:ss z", Locale.US);
+
+  public static RetryAfterParser getInstance() {
+    return new RetryAfterParser(SystemTime.getInstance());
+  }
+
+  RetryAfterParser(SystemTime systemTime) {
+    this.systemTime = systemTime;
+  }
+
+  public Optional<Duration> tryParse(String value) {
+    Duration duration = null;
+    if (SECONDS_PATTERN.matcher(value).matches()) {
+      duration = Duration.ofSeconds(Long.parseLong(value));
+    } else if (DATE_PATTERN.matcher(value).matches()) {
+      long difference = toMilliseconds(value) - systemTime.getCurrentTimeMillis();
+      if (difference > 0) {
+        duration = Duration.ofMillis(difference);
+      }
+    }
+    return Optional.ofNullable(duration);
+  }
+
+  private static long toMilliseconds(String value) {
+    return ZonedDateTime.parse(value, DATE_FORMAT).toInstant().toEpochMilli();
+  }
+}

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/request/Request.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.request;
+
+import com.google.auto.value.AutoValue;
+import opamp.proto.AgentToServer;
+
+/** Wrapper class for "AgentToServer" request body. */
+@AutoValue
+public abstract class Request {
+  public abstract AgentToServer getAgentToServer();
+
+  public static Request create(AgentToServer agentToServer) {
+    return new AutoValue_Request(agentToServer);
+  }
+}

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/request/delay/AcceptsDelaySuggestion.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.request.delay;
+
+import java.time.Duration;
+
+/**
+ * A {@link PeriodicDelay} implementation that wants to accept delay time suggestions, as explained
+ * <a
+ * href="https://github.com/open-telemetry/opamp-spec/blob/main/specification.md#throttling">here</a>,
+ * must implement this interface.
+ */
+public interface AcceptsDelaySuggestion {
+  void suggestDelay(Duration delay);
+}

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/request/delay/FixedPeriodicDelay.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.request.delay;
+
+import java.time.Duration;
+
+final class FixedPeriodicDelay implements PeriodicDelay {
+  private final Duration duration;
+
+  public FixedPeriodicDelay(Duration duration) {
+    this.duration = duration;
+  }
+
+  @Override
+  public Duration getNextDelay() {
+    return duration;
+  }
+
+  @Override
+  public void reset() {}
+}

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/request/delay/PeriodicDelay.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.request.delay;
+
+import java.time.Duration;
+
+public interface PeriodicDelay {
+  static PeriodicDelay ofFixedDuration(Duration duration) {
+    return new FixedPeriodicDelay(duration);
+  }
+
+  Duration getNextDelay();
+
+  void reset();
+}