OpampActivator.java

/*
 * Copyright Splunk Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.splunk.opentelemetry.opamp;

import static io.opentelemetry.opamp.client.internal.request.service.HttpRequestService.DEFAULT_DELAY_BETWEEN_REQUESTS;
import static io.opentelemetry.opamp.client.internal.request.service.HttpRequestService.DEFAULT_DELAY_BETWEEN_RETRIES;
import static io.opentelemetry.sdk.autoconfigure.AutoConfigureUtil.getConfig;
import static io.opentelemetry.sdk.autoconfigure.AutoConfigureUtil.getResource;
import static java.util.logging.Level.WARNING;

import com.google.auto.service.AutoService;
import io.opentelemetry.javaagent.extension.AgentListener;
import io.opentelemetry.opamp.client.OpampClient;
import io.opentelemetry.opamp.client.OpampClientBuilder;
import io.opentelemetry.opamp.client.internal.connectivity.http.OkHttpSender;
import io.opentelemetry.opamp.client.internal.request.delay.PeriodicDelay;
import io.opentelemetry.opamp.client.internal.request.service.HttpRequestService;
import io.opentelemetry.opamp.client.internal.response.MessageData;
import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk;
import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
import io.opentelemetry.sdk.resources.Resource;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.logging.Logger;
import opamp.proto.ServerErrorResponse;
import org.jetbrains.annotations.Nullable;

@AutoService(AgentListener.class)
public class OpampActivator implements AgentListener {
  private static final Logger logger = Logger.getLogger(OpampActivator.class.getName());

  private static final String OP_AMP_ENABLED_PROPERTY = "splunk.opamp.enabled";
  private static final String OP_AMP_ENDPOINT = "splunk.opamp.endpoint";
  private static final String OP_AMP_POLLING_INTERVAL = "splunk.opamp.polling.interval";

  @Override
  public void afterAgent(AutoConfiguredOpenTelemetrySdk autoConfiguredOpenTelemetrySdk) {
    ConfigProperties config = getConfig(autoConfiguredOpenTelemetrySdk);
    if (!config.getBoolean(OP_AMP_ENABLED_PROPERTY, false)) {
      return;
    }

    Resource resource = getResource(autoConfiguredOpenTelemetrySdk);
    long pollingDuration =
        config.getLong(
            OP_AMP_POLLING_INTERVAL, DEFAULT_DELAY_BETWEEN_REQUESTS.getNextDelay().toMillis());

    String endpoint = config.getString(OP_AMP_ENDPOINT);
    OpampClient client =
        startOpampClient(
            endpoint,
            resource,
            pollingDuration,
            new OpampClient.Callbacks() {
              @Override
              public void onConnect(OpampClient opampClient) {}

              @Override
              public void onConnectFailed(OpampClient opampClient, @Nullable Throwable throwable) {
                logger.log(WARNING, "Connection to OpAMP server failed", throwable);
              }

              @Override
              public void onErrorResponse(
                  OpampClient opampClient, ServerErrorResponse serverErrorResponse) {
                logger.log(WARNING, "OpAMP server returned error " + serverErrorResponse);
              }

              @Override
              public void onMessage(OpampClient opampClient, MessageData messageData) {}
            });
    Runtime.getRuntime()
        .addShutdownHook(
            new Thread(
                () -> {
                  try {
                    client.close();
                  } catch (IOException e) {
                    logger.log(WARNING, "Error shutting down OpAMP client", e);
                  }
                }));
  }

  static OpampClient startOpampClient(
      String endpoint,
      Resource resource,
      long pollingDurationMillis,
      OpampClient.Callbacks callbacks) {

    OpampClientBuilder builder = OpampClient.builder();
    // TODO: Uncomment once we are able to report our effective config
    // builder.enableEffectiveConfigReporting();
    if (endpoint != null) {
      PeriodicDelay pollingDelay =
          PeriodicDelay.ofFixedDuration(Duration.ofMillis(pollingDurationMillis));
      OkHttpSender okhttp = OkHttpSender.create(endpoint);
      HttpRequestService httpSender =
          HttpRequestService.create(okhttp, pollingDelay, DEFAULT_DELAY_BETWEEN_RETRIES);
      builder.setRequestService(httpSender);
    }
    addIdentifyingAttributes(builder, resource);

    return builder.build(callbacks);
  }

  static void addIdentifyingAttributes(OpampClientBuilder builder, Resource res) {
    res.getAttributes()
        .forEach(
            (key, value) -> {
              switch (key.getType()) {
                case STRING:
                  builder.putIdentifyingAttribute(key.getKey(), (String) value);
                  break;
                case LONG:
                  builder.putIdentifyingAttribute(key.getKey(), (long) value);
                  break;
                case DOUBLE:
                  builder.putIdentifyingAttribute(key.getKey(), (double) value);
                  break;
                case BOOLEAN:
                  builder.putIdentifyingAttribute(key.getKey(), (boolean) value);
                  break;
                case VALUE:
                  builder.putIdentifyingAttribute(key.getKey(), value.toString());
                  break;
                case STRING_ARRAY:
                  {
                    List<String> typedValueList = (List<String>) value;
                    String[] array = typedValueList.toArray(new String[] {});
                    builder.putIdentifyingAttribute(key.getKey(), array);
                    break;
                  }
                case LONG_ARRAY:
                  {
                    List<Long> typedValueList = (List<Long>) value;
                    long[] primitiveArray = new long[typedValueList.size()];
                    for (int i = 0; i < typedValueList.size(); i++) {
                      primitiveArray[i] = typedValueList.get(i);
                    }
                    builder.putIdentifyingAttribute(key.getKey(), primitiveArray);
                    break;
                  }
                case DOUBLE_ARRAY:
                  {
                    List<Double> typedValueList = (List<Double>) value;
                    double[] primitiveArray = new double[typedValueList.size()];
                    for (int i = 0; i < typedValueList.size(); i++) {
                      primitiveArray[i] = typedValueList.get(i);
                    }
                    builder.putIdentifyingAttribute(key.getKey(), primitiveArray);
                    break;
                  }
                case BOOLEAN_ARRAY:
                  {
                    List<Boolean> typedValueList = (List<Boolean>) value;
                    boolean[] primitiveArray = new boolean[typedValueList.size()];
                    for (int i = 0; i < typedValueList.size(); i++) {
                      primitiveArray[i] = typedValueList.get(i);
                    }
                    builder.putIdentifyingAttribute(key.getKey(), primitiveArray);
                    break;
                  }
              }
            });
  }
}