APM Metering API

docs/changelog/99832.yaml

@@ -0,0 +1,5 @@
+pr: 99832
+summary: APM Metering API
+area: Infra/Core
+type: enhancement
+issues: []

modules/apm/src/main/java/org/elasticsearch/telemetry/apm/APM.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.telemetry.TelemetryProvider;
 import org.elasticsearch.telemetry.apm.internal.APMAgentSettings;
 import org.elasticsearch.telemetry.apm.internal.APMTelemetryProvider;
+import org.elasticsearch.telemetry.apm.internal.metrics.APMMeter;
 import org.elasticsearch.telemetry.apm.internal.tracing.APMTracer;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -97,7 +98,9 @@ public Collection<Object> createComponents(
         apmAgentSettings.syncAgentSystemProperties(settings);
         apmAgentSettings.addClusterSettingsListeners(clusterService, telemetryProvider.get());
 
-        return List.of(apmTracer);
+        final APMMeter apmMeter = telemetryProvider.get().getMeter();
+
+        return List.of(apmTracer, apmMeter);
     }
 
     @Override

modules/apm/src/main/java/org/elasticsearch/telemetry/apm/internal/APMAgentSettings.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.telemetry.apm.internal.metrics.APMMeter;
 import org.elasticsearch.telemetry.apm.internal.tracing.APMTracer;
 
 import java.security.AccessController;
@@ -40,14 +41,21 @@ public class APMAgentSettings {
      * Sensible defaults that Elasticsearch configures. This cannot be done via the APM agent
      * config file, as then their values could not be overridden dynamically via system properties.
      */
-    static Map<String, String> APM_AGENT_DEFAULT_SETTINGS = Map.of("transaction_sample_rate", "0.2");
+    static Map<String, String> APM_AGENT_DEFAULT_SETTINGS = Map.of(
+        "transaction_sample_rate",
+        "0.2",
+        "enable_experimental_instrumentations",
+        "true"
+    );
 
     public void addClusterSettingsListeners(ClusterService clusterService, APMTelemetryProvider apmTelemetryProvider) {
         final ClusterSettings clusterSettings = clusterService.getClusterSettings();
         final APMTracer apmTracer = apmTelemetryProvider.getTracer();
+        final APMMeter apmMeter = apmTelemetryProvider.getMeter();
 
         clusterSettings.addSettingsUpdateConsumer(APM_ENABLED_SETTING, enabled -> {
             apmTracer.setEnabled(enabled);
+            apmMeter.setEnabled(enabled);
             // The agent records data other than spans, e.g. JVM metrics, so we toggle this setting in order to
             // minimise its impact to a running Elasticsearch.
             this.setAgentSetting("recording", Boolean.toString(enabled));
@@ -159,7 +167,7 @@ public void setAgentSetting(String key, String value) {
 
     public static final Setting<Boolean> APM_ENABLED_SETTING = Setting.boolSetting(
         APM_SETTING_PREFIX + "enabled",
-        false,
+        true,
         OperatorDynamic,
         NodeScope
     );

modules/apm/src/main/java/org/elasticsearch/telemetry/apm/internal/APMTelemetryProvider.java

@@ -10,19 +10,27 @@
 
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.telemetry.TelemetryProvider;
+import org.elasticsearch.telemetry.apm.internal.metrics.APMMeter;
 import org.elasticsearch.telemetry.apm.internal.tracing.APMTracer;
 
 public class APMTelemetryProvider implements TelemetryProvider {
     private final Settings settings;
     private final APMTracer apmTracer;
+    private final APMMeter apmMeter;
 
     public APMTelemetryProvider(Settings settings) {
         this.settings = settings;
         apmTracer = new APMTracer(settings);
+        apmMeter = new APMMeter(settings);
     }
 
     @Override
     public APMTracer getTracer() {
         return apmTracer;
     }
+
+    @Override
+    public APMMeter getMeter() {
+        return apmMeter;
+    }
 }

modules/apm/src/main/java/org/elasticsearch/telemetry/apm/internal/metrics/APMMeter.java

@@ -0,0 +1,190 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.telemetry.apm.internal.metrics;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.metrics.Meter;
+
+import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.common.component.AbstractLifecycleComponent;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.telemetry.metric.DoubleCounter;
+import org.elasticsearch.telemetry.metric.DoubleGauge;
+import org.elasticsearch.telemetry.metric.DoubleHistogram;
+import org.elasticsearch.telemetry.metric.DoubleUpDownCounter;
+import org.elasticsearch.telemetry.metric.LongCounter;
+import org.elasticsearch.telemetry.metric.LongGauge;
+import org.elasticsearch.telemetry.metric.LongHistogram;
+import org.elasticsearch.telemetry.metric.LongUpDownCounter;
+
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.function.Supplier;
+
+import static org.elasticsearch.telemetry.apm.internal.APMAgentSettings.APM_ENABLED_SETTING;
+
+public class APMMeter extends AbstractLifecycleComponent implements org.elasticsearch.telemetry.metric.Meter {
+    private final Instruments instruments;
+
+    private final Supplier<Meter> otelMeterSupplier;
+    private final Supplier<Meter> noopMeterSupplier;
+
+    private volatile boolean enabled;
+
+    public APMMeter(Settings settings) {
+        this(settings, APMMeter::otelMeter, APMMeter::noopMeter);
+    }
+
+    public APMMeter(Settings settings, Supplier<Meter> otelMeterSupplier, Supplier<Meter> noopMeterSupplier) {
+        this.enabled = APM_ENABLED_SETTING.get(settings);
+        this.otelMeterSupplier = otelMeterSupplier;
+        this.noopMeterSupplier = noopMeterSupplier;
+        this.instruments = new Instruments(enabled ? createApmServices() : noopMeterSupplier.get());
+        setEnabled(enabled);
+    }
+
+    public void setEnabled(boolean enabled) {
+        this.enabled = enabled;
+        setupApmServices(this.enabled);
+    }
+
+    private void setupApmServices(boolean enabled) {
+        if (enabled) {
+            instruments.setProvider(createApmServices());
+        } else {
+            destroyApmServices();
+        }
+    }
+
+    @Override
+    protected void doStart() {
+        if (enabled) {
+            createApmServices();
+        }
+    }
+
+    @Override
+    protected void doStop() {
+        destroyApmServices();
+    }
+
+    @Override
+    protected void doClose() {}
+
+    @Override
+    public DoubleCounter registerDoubleCounter(String name, String description, String unit) {
+        return instruments.registerDoubleCounter(name, description, unit);
+    }
+
+    @Override
+    public DoubleCounter getDoubleCounter(String name) {
+        return instruments.getDoubleCounter(name);
+    }
+
+    @Override
+    public DoubleUpDownCounter registerDoubleUpDownCounter(String name, String description, String unit) {
+        return instruments.registerDoubleUpDownCounter(name, description, unit);
+    }
+
+    @Override
+    public DoubleUpDownCounter getDoubleUpDownCounter(String name) {
+        return instruments.getDoubleUpDownCounter(name);
+    }
+
+    @Override
+    public DoubleGauge registerDoubleGauge(String name, String description, String unit) {
+        return instruments.registerDoubleGauge(name, description, unit);
+    }
+
+    @Override
+    public DoubleGauge getDoubleGauge(String name) {
+        return instruments.getDoubleGauge(name);
+    }
+
+    @Override
+    public DoubleHistogram registerDoubleHistogram(String name, String description, String unit) {
+        return instruments.registerDoubleHistogram(name, description, unit);
+    }
+
+    @Override
+    public DoubleHistogram getDoubleHistogram(String name) {
+        return instruments.getDoubleHistogram(name);
+    }
+
+    @Override
+    public LongCounter registerLongCounter(String name, String description, String unit) {
+        return instruments.registerLongCounter(name, description, unit);
+    }
+
+    @Override
+    public LongCounter getLongCounter(String name) {
+        return instruments.getLongCounter(name);
+    }
+
+    @Override
+    public LongUpDownCounter registerLongUpDownCounter(String name, String description, String unit) {
+        return instruments.registerLongUpDownCounter(name, description, unit);
+    }
+
+    @Override
+    public LongUpDownCounter getLongUpDownCounter(String name) {
+        return instruments.getLongUpDownCounter(name);
+    }
+
+    @Override
+    public LongGauge registerLongGauge(String name, String description, String unit) {
+        return instruments.registerLongGauge(name, description, unit);
+    }
+
+    @Override
+    public LongGauge getLongGauge(String name) {
+        return instruments.getLongGauge(name);
+    }
+
+    @Override
+    public LongHistogram registerLongHistogram(String name, String description, String unit) {
+        return instruments.registerLongHistogram(name, description, unit);
+    }
+
+    @Override
+    public LongHistogram getLongHistogram(String name) {
+        return instruments.getLongHistogram(name);
+    }
+
+    Meter createApmServices() {
+        assert this.enabled;
+
+        SetOnce<Meter> provider = new SetOnce<>();
+        AccessController.doPrivileged((PrivilegedAction<Void>) () -> {
+            provider.set(otelMeterSupplier.get());
+            return null;
+        });
+        return provider.get();
+    }
+
+    private void destroyApmServices() {
+        instruments.setProvider(noopMeterSupplier.get());
+    }
+
+    private static Meter noopMeter() {
+        return OpenTelemetry.noop().getMeter("noop");
+    }
+
+    private static Meter otelMeter() {
+        var openTelemetry = GlobalOpenTelemetry.get();
+        var meter = openTelemetry.getMeter("elasticsearch");
+        return meter;
+    }
+
+    // scope for testing
+    Instruments getInstruments() {
+        return instruments;
+    }
+}

modules/apm/src/main/java/org/elasticsearch/telemetry/apm/internal/metrics/AbstractInstrument.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.telemetry.apm.internal.metrics;
+
+import io.opentelemetry.api.metrics.Meter;
+
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.telemetry.metric.Instrument;
+
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * An instrument that contains the name, description and unit.  The delegate may be replaced when
+ * the provider is updated.
+ * Subclasses should implement the builder, which is used on initialization and provider updates.
+ * @param <T> delegated instrument
+ */
+public abstract class AbstractInstrument<T> implements Instrument {
+    private final AtomicReference<T> delegate;
+    private final String name;
+    private final String description;
+    private final String unit;
+
+    public AbstractInstrument(Meter meter, String name, String description, String unit) {
+        this.name = Objects.requireNonNull(name);
+        this.description = Objects.requireNonNull(description);
+        this.unit = Objects.requireNonNull(unit);
+        this.delegate = new AtomicReference<>(buildInstrument(meter));
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    public String getUnit() {
+        return unit.toString();
+    }
+
+    T getInstrument() {
+        return delegate.get();
+    }
+
+    String getDescription() {
+        return description;
+    }
+
+    void setProvider(@Nullable Meter meter) {
+        delegate.set(buildInstrument(Objects.requireNonNull(meter)));
+    }
+
+    abstract T buildInstrument(Meter meter);
+}

modules/apm/src/main/java/org/elasticsearch/telemetry/apm/internal/metrics/DoubleCounterAdapter.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.telemetry.apm.internal.metrics;
+
+import io.opentelemetry.api.metrics.Meter;
+
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * DoubleGaugeAdapter wraps an otel ObservableDoubleMeasurement
+ */
+public class DoubleCounterAdapter extends AbstractInstrument<io.opentelemetry.api.metrics.DoubleCounter>
+    implements
+        org.elasticsearch.telemetry.metric.DoubleCounter {
+
+    public DoubleCounterAdapter(Meter meter, String name, String description, String unit) {
+        super(meter, name, description, unit);
+    }
+
+    io.opentelemetry.api.metrics.DoubleCounter buildInstrument(Meter meter) {
+        return Objects.requireNonNull(meter)
+            .counterBuilder(getName())
+            .ofDoubles()
+            .setDescription(getDescription())
+            .setUnit(getUnit())
+            .build();
+    }
+
+    @Override
+    public void increment() {
+        getInstrument().add(1d);
+    }
+
+    @Override
+    public void incrementBy(double inc) {
+        assert inc >= 0;
+        getInstrument().add(inc);
+    }
+
+    @Override
+    public void incrementBy(double inc, Map<String, Object> attributes) {
+        assert inc >= 0;
+        getInstrument().add(inc, OtelHelper.fromMap(attributes));
+    }
+}