[opamp-client] Add the ability to receive custom messages

opamp-client/src/main/java/io/opentelemetry/opamp/client/OpampClientBuilder.java

@@ -17,9 +17,11 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.UUID;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -28,13 +30,15 @@
 import opamp.proto.AnyValue;
 import opamp.proto.ArrayValue;
 import opamp.proto.ComponentHealth;
+import opamp.proto.CustomCapabilities;
 import opamp.proto.KeyValue;
 import opamp.proto.RemoteConfigStatus;
 
 /** Builds an {@link OpampClient} instance. */
 public final class OpampClientBuilder {
   private final Map<String, AnyValue> identifyingAttributes = new HashMap<>();
   private final Map<String, AnyValue> nonIdentifyingAttributes = new HashMap<>();
+  private final Set<String> customCapabilities = new HashSet<>();
   private long capabilities = 0;
   private RequestService service =
       HttpRequestService.create(OkHttpSender.create("http://localhost:4320/v1/opamp"));
@@ -391,6 +395,21 @@ public OpampClientBuilder enableHealthReporting(@Nonnull ComponentHealth initial
     return this;
   }
 
+  /**
+   * Adds a custom capability supported by this Client for messages received from the Server. The
+   * configured capabilities are reported in {@code AgentToServer.custom_capabilities}. Note that
+   * incoming custom messages are not filtered, it is the user's responsibility to check the
+   * capability
+   *
+   * @param capability The reverse FQDN identifying the custom capability.
+   * @return this
+   */
+  @CanIgnoreReturnValue
+  public OpampClientBuilder addCustomCapability(String capability) {
+    customCapabilities.add(Objects.requireNonNull(capability));
+    return this;
+  }
+
   public OpampClient build(OpampClient.Callbacks callbacks) {
     List<KeyValue> protoIdentifyingAttributes = new ArrayList<>();
     List<KeyValue> protoNonIdentifyingAttributes = new ArrayList<>();
@@ -414,6 +433,7 @@ public OpampClient build(OpampClient.Callbacks callbacks) {
                     .non_identifying_attributes(protoNonIdentifyingAttributes)
                     .build()),
             new State.Capabilities(capabilities),
+            new State.CustomCapabilities(createCustomCapabilities()),
             new State.Health(initialHealth),
             new State.InstanceUid(instanceUid),
             new State.Flags(0L),
@@ -425,6 +445,16 @@ private void enableCapability(AgentCapabilities capability) {
     capabilities = capabilities | capability.getValue();
   }
 
+  @Nullable
+  private CustomCapabilities createCustomCapabilities() {
+    if (customCapabilities.isEmpty()) {
+      return null;
+    }
+    return new CustomCapabilities.Builder()
+        .capabilities(new ArrayList<>(customCapabilities))
+        .build();
+  }
+
   private static State.EffectiveConfig createEffectiveConfigNoop() {
     return new State.EffectiveConfig() {
       @Nullable

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/impl/OpampClientImpl.java

@@ -12,6 +12,7 @@
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.AgentDescriptionAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.AgentDisconnectAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.CapabilitiesAppender;
+import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.CustomCapabilitiesAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.EffectiveConfigAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.FlagsAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.HealthAppender;
@@ -79,6 +80,7 @@ public final class OpampClientImpl
     // Compressable fields init
     List<Field> compressableFields = new ArrayList<>();
     compressableFields.add(Field.AGENT_DESCRIPTION);
+    compressableFields.add(Field.CUSTOM_CAPABILITIES);
     compressableFields.add(Field.HEALTH);
     compressableFields.add(Field.EFFECTIVE_CONFIG);
     compressableFields.add(Field.REMOTE_CONFIG_STATUS);
@@ -94,6 +96,7 @@ public static OpampClientImpl create(
             RemoteConfigStatusAppender.create(state.remoteConfigStatus),
             SequenceNumberAppender.create(state.sequenceNum),
             CapabilitiesAppender.create(state.capabilities),
+            CustomCapabilitiesAppender.create(state.customCapabilities),
             HealthAppender.create(state.health),
             InstanceUidAppender.create(state.instanceUid),
             FlagsAppender.create(state.flags),
@@ -206,6 +209,11 @@ private void handleResponsePayload(ServerToAgent response) {
       messageBuilder.setRemoteConfig(response.remote_config);
     }
 
+    if (response.custom_message != null) {
+      notifyOnMessage = true;
+      messageBuilder.setCustomMessage(response.custom_message);
+    }
+
     if (notifyOnMessage) {
       callbacks.onMessage(this, messageBuilder.build());
     }

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/impl/OpampClientState.java

@@ -19,17 +19,20 @@ public final class OpampClientState {
   public final State.SequenceNum sequenceNum;
   public final State.AgentDescription agentDescription;
   public final State.Capabilities capabilities;
+  public final State.CustomCapabilities customCapabilities;
   public final State.Health health;
   public final State.InstanceUid instanceUid;
   public final State.Flags flags;
   public final State.EffectiveConfig effectiveConfig;
   private final List<State<?>> items;
 
+  @SuppressWarnings("TooManyParameters")
   public OpampClientState(
       State.RemoteConfigStatus remoteConfigStatus,
       State.SequenceNum sequenceNum,
       State.AgentDescription agentDescription,
       State.Capabilities capabilities,
+      State.CustomCapabilities customCapabilities,
       State.Health health,
       State.InstanceUid instanceUid,
       State.Flags flags,
@@ -38,6 +41,7 @@ public OpampClientState(
     this.sequenceNum = sequenceNum;
     this.agentDescription = agentDescription;
     this.capabilities = capabilities;
+    this.customCapabilities = customCapabilities;
     this.health = health;
     this.instanceUid = instanceUid;
     this.flags = flags;
@@ -48,6 +52,7 @@ public OpampClientState(
     providedItems.add(sequenceNum);
     providedItems.add(agentDescription);
     providedItems.add(capabilities);
+    providedItems.add(customCapabilities);
     providedItems.add(health);
     providedItems.add(instanceUid);
     providedItems.add(flags);

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/impl/recipe/AgentToServerAppenders.java

@@ -9,6 +9,7 @@
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.AgentDisconnectAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.AgentToServerAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.CapabilitiesAppender;
+import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.CustomCapabilitiesAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.EffectiveConfigAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.FlagsAppender;
 import io.opentelemetry.opamp.client.internal.impl.recipe.appenders.HealthAppender;
@@ -30,6 +31,7 @@ public final class AgentToServerAppenders {
   public final RemoteConfigStatusAppender remoteConfigStatusAppender;
   public final SequenceNumberAppender sequenceNumberAppender;
   public final CapabilitiesAppender capabilitiesAppender;
+  public final CustomCapabilitiesAppender customCapabilitiesAppender;
   public final HealthAppender healthAppender;
   public final InstanceUidAppender instanceUidAppender;
   public final FlagsAppender flagsAppender;
@@ -43,6 +45,7 @@ public AgentToServerAppenders(
       RemoteConfigStatusAppender remoteConfigStatusAppender,
       SequenceNumberAppender sequenceNumberAppender,
       CapabilitiesAppender capabilitiesAppender,
+      CustomCapabilitiesAppender customCapabilitiesAppender,
       HealthAppender healthAppender,
       InstanceUidAppender instanceUidAppender,
       FlagsAppender flagsAppender,
@@ -52,6 +55,7 @@ public AgentToServerAppenders(
     this.remoteConfigStatusAppender = remoteConfigStatusAppender;
     this.sequenceNumberAppender = sequenceNumberAppender;
     this.capabilitiesAppender = capabilitiesAppender;
+    this.customCapabilitiesAppender = customCapabilitiesAppender;
     this.healthAppender = healthAppender;
     this.instanceUidAppender = instanceUidAppender;
     this.flagsAppender = flagsAppender;
@@ -63,6 +67,7 @@ public AgentToServerAppenders(
     appenders.put(Field.REMOTE_CONFIG_STATUS, remoteConfigStatusAppender);
     appenders.put(Field.SEQUENCE_NUM, sequenceNumberAppender);
     appenders.put(Field.CAPABILITIES, capabilitiesAppender);
+    appenders.put(Field.CUSTOM_CAPABILITIES, customCapabilitiesAppender);
     appenders.put(Field.HEALTH, healthAppender);
     appenders.put(Field.INSTANCE_UID, instanceUidAppender);
     appenders.put(Field.FLAGS, flagsAppender);

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/impl/recipe/appenders/CustomCapabilitiesAppender.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.impl.recipe.appenders;
+
+import java.util.function.Supplier;
+import opamp.proto.AgentToServer;
+import opamp.proto.CustomCapabilities;
+
+/**
+ * This class is internal and is hence not for public use. Its APIs are unstable and can change at
+ * any time. TODO AgentToServer.custom_message support
+ */
+public final class CustomCapabilitiesAppender implements AgentToServerAppender {
+  private final Supplier<CustomCapabilities> customCapabilities;
+
+  public static CustomCapabilitiesAppender create(Supplier<CustomCapabilities> customCapabilities) {
+    return new CustomCapabilitiesAppender(customCapabilities);
+  }
+
+  private CustomCapabilitiesAppender(Supplier<CustomCapabilities> customCapabilities) {
+    this.customCapabilities = customCapabilities;
+  }
+
+  @Override
+  public void appendTo(AgentToServer.Builder builder) {
+    builder.custom_capabilities(customCapabilities.get());
+  }
+}

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

@@ -18,6 +18,7 @@ public enum Field {
   SEQUENCE_NUM,
   AGENT_DESCRIPTION,
   CAPABILITIES,
+  CUSTOM_CAPABILITIES,
   HEALTH,
   EFFECTIVE_CONFIG,
   REMOTE_CONFIG_STATUS,

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/response/MessageData.java

@@ -9,6 +9,7 @@
 import io.opentelemetry.opamp.client.OpampClient;
 import javax.annotation.Nullable;
 import opamp.proto.AgentRemoteConfig;
+import opamp.proto.CustomMessage;
 
 /**
  * Data class provided in {@link OpampClient.Callbacks#onMessage(OpampClient, MessageData)} with
@@ -19,6 +20,9 @@ public abstract class MessageData {
   @Nullable
   public abstract AgentRemoteConfig getRemoteConfig();
 
+  @Nullable
+  public abstract CustomMessage getCustomMessage();
+
   public static Builder builder() {
     return new AutoValue_MessageData.Builder();
   }
@@ -27,6 +31,8 @@ public static Builder builder() {
   public abstract static class Builder {
     public abstract Builder setRemoteConfig(AgentRemoteConfig remoteConfig);
 
+    public abstract Builder setCustomMessage(CustomMessage customMessage);
+
     public abstract MessageData build();
   }
 }

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/state/State.java

@@ -77,6 +77,17 @@ public Field getFieldType() {
     }
   }
 
+  final class CustomCapabilities extends InMemoryState<opamp.proto.CustomCapabilities> {
+    public CustomCapabilities(@Nullable opamp.proto.CustomCapabilities initialValue) {
+      super(initialValue);
+    }
+
+    @Override
+    public Field getFieldType() {
+      return Field.CUSTOM_CAPABILITIES;
+    }
+  }
+
   final class Health extends InMemoryState<ComponentHealth> {
     public Health(@Nullable ComponentHealth initialValue) {
       super(initialValue);