preview of the meter api

TRACING.md

@@ -158,6 +158,6 @@ explicitly opening a scope via the `Tracer`.
 [otel]: https://opentelemetry.io/
 [thread-context]: ./server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java
 [w3c]: https://www.w3.org/TR/trace-context/
-[tracing]: ./server/src/main/java/org/elasticsearch/tracing
+[tracing]: ./server/src/main/java/org/elasticsearch/telemetry
 [agent-config]: https://www.elastic.co/guide/en/apm/agent/java/master/configuration.html
 [agent]: https://www.elastic.co/guide/en/apm/agent/java/current/index.html

distribution/tools/server-cli/src/main/java/org/elasticsearch/server/cli/APMJvmOptions.java

@@ -50,10 +50,12 @@ class APMJvmOptions {
         // by the agent. Don't disable writing to a log file, as the agent will then
         // require extra Security Manager permissions when it tries to do something
         // else, and it's just painful.
-        "log_file", "_AGENT_HOME_/../../logs/apm.log",
-
+        "log_file", "/Users/przemyslawgomulka/workspace/pgomulka/apm.log",
+        "log_level", "debug",
         // ES does not use auto-instrumentation.
-        "instrument", "false"
+        "instrument", "false",
+        "experimental", "true",
+        "enable_experimental_instrumentations", "true"
         );
 
     /**
@@ -82,7 +84,7 @@ class APMJvmOptions {
         // is doing, leave this value alone.
         "log_level", "error",
         "application_packages", "org.elasticsearch,org.apache.lucene",
-        "metrics_interval", "120s",
+        "metrics_interval", "5s",
         "breakdown_metrics", "false",
         "central_config", "false"
         );
@@ -316,9 +318,7 @@ static Path findAgentJar(String installDir) throws IOException, UserException {
         }
 
         try (var apmStream = Files.list(apmModule)) {
-            final List<Path> paths = apmStream.filter(
-                path -> path.getFileName().toString().matches("elastic-apm-agent-\\d+\\.\\d+\\.\\d+\\.jar")
-            ).toList();
+            final List<Path> paths = apmStream.filter(path -> path.getFileName().toString().matches("elastic-apm-agent-.*.jar")).toList();
 
             if (paths.size() > 1) {
                 throw new UserException(

gradle/verification-metadata.xml

@@ -74,6 +74,11 @@
             <sha256 value="3be90268557494f88fe252caf6358d4107aa9349852b8704b1e0ff4831cb74d6" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="co.elastic.apm" name="elastic-apm-agent" version="1.37.0">
+         <artifact name="elastic-apm-agent-1.37.0.jar">
+            <sha256 value="10496390d1dcc7c558ae63e115beff9477157e3d219fce686ef5ca59c744dfea" origin="Generated by Gradle"/>
+         </artifact>
+      </component>
       <component group="co.elastic.logging" name="ecs-logging-core" version="1.2.0">
          <artifact name="ecs-logging-core-1.2.0.jar">
             <sha256 value="0d6318af29848ea588b1c244834e3e762315c68de2bcfc9f1adf3f0633d22d37" origin="Generated by Gradle"/>

modules/apm/build.gradle

@@ -14,11 +14,19 @@ esplugin {
 
 def otelVersion = '1.17.0'
 
+repositories {
+  maven {
+    name "sonatype-nexus-snapshots"
+    url "https://oss.sonatype.org/content/repositories/snapshots"
+  }
+}
 dependencies {
   implementation "io.opentelemetry:opentelemetry-api:${otelVersion}"
   implementation "io.opentelemetry:opentelemetry-context:${otelVersion}"
   implementation "io.opentelemetry:opentelemetry-semconv:${otelVersion}-alpha"
-  runtimeOnly    "co.elastic.apm:elastic-apm-agent:1.36.0"
+  implementation "org.apache.logging.log4j:log4j-api:${versions.log4j}"
+
+  runtimeOnly    "co.elastic.apm:elastic-apm-agent:1.42.1-SNAPSHOT"
 }
 
 tasks.named("dependencyLicenses").configure {

modules/apm/src/main/java/module-info.java

@@ -16,4 +16,5 @@
     requires io.opentelemetry.api;
 
     exports org.elasticsearch.telemetry.apm;
+    exports org.elasticsearch.telemetry.apm.internal.metrics;
 }

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.APMMetric;
 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 APMMetric apmMetric = telemetryProvider.get().getMetric();
+
+        return List.of(apmTracer, apmMetric);
     }
 
     @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.APMMetric;
 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 APMMetric apmMetric = apmTelemetryProvider.getMetric();
 
         clusterSettings.addSettingsUpdateConsumer(APM_ENABLED_SETTING, enabled -> {
             apmTracer.setEnabled(enabled);
+            apmMetric.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));

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.APMMetric;
 import org.elasticsearch.telemetry.apm.internal.tracing.APMTracer;
 
 public class APMTelemetryProvider implements TelemetryProvider {
     private final Settings settings;
     private final APMTracer apmTracer;
+    private final APMMetric apmMetric;
 
     public APMTelemetryProvider(Settings settings) {
         this.settings = settings;
         apmTracer = new APMTracer(settings);
+        apmMetric = new APMMetric(settings);
     }
 
     @Override
     public APMTracer getTracer() {
         return apmTracer;
     }
+
+    @Override
+    public APMMetric getMetric() {
+        return apmMetric;
+    }
 }

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

@@ -0,0 +1,197 @@
+/*
+ * 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.elasticsearch.common.component.AbstractLifecycleComponent;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.LazyInitializable;
+import org.elasticsearch.telemetry.MetricName;
+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.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.elasticsearch.telemetry.apm.internal.APMAgentSettings.APM_ENABLED_SETTING;
+
+public class APMMetric extends AbstractLifecycleComponent implements org.elasticsearch.telemetry.metric.Metric {
+
+    private final InstrumentRegistrar<DoubleCounter> doubleCounters = new InstrumentRegistrar<>();
+    private final InstrumentRegistrar<DoubleUpDownCounter> doubleUpDownCounters = new InstrumentRegistrar<>();
+    private final InstrumentRegistrar<DoubleGauge> doubleGauges = new InstrumentRegistrar<>();
+    private final InstrumentRegistrar<DoubleHistogram> doubleHistograms = new InstrumentRegistrar<>();
+    private final InstrumentRegistrar<LongCounter> longCounters = new InstrumentRegistrar<>();
+    private final InstrumentRegistrar<LongUpDownCounter> longUpDownCounters = new InstrumentRegistrar<>();
+    private final InstrumentRegistrar<LongGauge> longGauges = new InstrumentRegistrar<>();
+    private final InstrumentRegistrar<LongHistogram> longHistograms = new InstrumentRegistrar<>();
+    private volatile boolean enabled;
+    private AtomicReference<APMServices> services = new AtomicReference<>();
+
+    private final List<SwitchableInstrument> allRegisteredInstruments = new ArrayList<>();
+
+    record APMServices(Meter meter, OpenTelemetry openTelemetry) {}
+
+    // TODO remove duplication between APMTracer and APMMetric. enabled, create apm services etc
+    public APMMetric(Settings settings) {
+        this.enabled = APM_ENABLED_SETTING.get(settings);
+    }
+
+    public void setEnabled(boolean enabled) {
+        this.enabled = enabled;
+        if (enabled) {
+            createApmServices();
+            allRegisteredInstruments.forEach(si -> si.setEnabled(true));
+        } else {
+            allRegisteredInstruments.forEach(si -> si.setEnabled(false));
+            destroyApmServices();
+        }
+
+    }
+
+    @Override
+    protected void doStart() {
+        if (enabled) {
+            createApmServices();
+        }
+    }
+
+    @Override
+    protected void doStop() {
+        destroyApmServices();
+    }
+
+    @Override
+    protected void doClose() {}
+
+    @Override
+    public <T> DoubleCounter registerDoubleCounter(MetricName name, String description, T unit) {
+        var counter = OtelDoubleCounter.build(()->services.get().meter, name, description, unit);
+        doubleCounters.register(counter);
+        allRegisteredInstruments.add(counter);
+        return counter;
+    }
+
+    @Override
+    public <T> DoubleUpDownCounter registerDoubleUpDownCounter(MetricName name, String description, T unit) {
+        var counter = OtelDoubleUpDownCounter.build(services.get().meter, name, description, unit);
+        doubleUpDownCounters.register(counter);
+        allRegisteredInstruments.add(counter);
+
+        return counter;
+    }
+
+    @Override
+    public <T> DoubleGauge registerDoubleGauge(MetricName name, String description, T unit) {
+        var gauge = OtelDoubleGauge.build(services.get().meter, name, description, unit);
+        doubleGauges.register(gauge);
+        allRegisteredInstruments.add(gauge);
+
+        return gauge;
+    }
+
+    @Override
+    public <T> DoubleHistogram registerDoubleHistogram(MetricName name, String description, T unit) {
+        var histogram = OtelDoubleHistogram.build(services.get().meter, name, description, unit);
+        doubleHistograms.register(histogram);
+        allRegisteredInstruments.add(histogram);
+        return histogram;
+    }
+
+    @Override
+    public <T> LongCounter registerLongCounter(MetricName name, String description, T unit) {
+        var lazyCounter = new LazyInitializable<>(
+            () -> services.get().meter.counterBuilder(name.getRawName()).setDescription(description).setUnit(unit.toString()).build()
+        );
+        var counter = OtelLongCounter.build(lazyCounter,services.get().meter, name, description, unit);
+        longCounters.register(counter);
+        allRegisteredInstruments.add(counter);
+
+        return counter;
+    }
+
+    @Override
+    public <T> LongUpDownCounter registerLongUpDownCounter(MetricName name, String description, T unit) {
+
+        var lazyCounter = new LazyInitializable<>(
+            () -> services.get().meter.upDownCounterBuilder(name.getRawName()).setDescription(description).setUnit(unit.toString()).build()
+        );
+        var counter = OtelLongUpDownCounter.build(lazyCounter,services.get().meter, name, description, unit);
+        longUpDownCounters.register(counter);
+        allRegisteredInstruments.add(counter);
+
+        return counter;
+    }
+
+    @Override
+    public <T> LongGauge registerLongGauge(MetricName name, String description, T unit) {
+
+        var lazyGauge = new LazyInitializable<>(
+            () -> services.get().meter.gaugeBuilder(name.getRawName())
+                .ofLongs()
+                .setDescription(description)
+                .setUnit(unit.toString())
+                .buildObserver()
+        );
+
+        var gauge = OtelLongGauge.build(lazyGauge, services.get().meter, name, description, unit);
+        longGauges.register(gauge);
+        allRegisteredInstruments.add(gauge);
+
+        return gauge;
+    }
+
+    @Override
+    public <T> LongHistogram registerLongHistogram(MetricName name, String description, T unit) {
+
+        var lazyHistogram = new LazyInitializable<>(
+            () -> services.get().meter.histogramBuilder(name.getRawName())
+                .ofLongs()
+                .setDescription(description)
+                .setUnit(unit.toString())
+                .build()
+        );
+
+        var histogram = OtelLongHistogram.build(lazyHistogram,services.get().meter, name, description, unit);
+        longHistograms.register(histogram);
+        allRegisteredInstruments.add(histogram);
+
+        return histogram;
+    }
+
+    void createApmServices() {
+        assert this.enabled;
+        assert this.services.get() == null;
+
+        AccessController.doPrivileged((PrivilegedAction<Void>) () -> {
+            var openTelemetry = GlobalOpenTelemetry.get();
+            var meter = openTelemetry.getMeter("elasticsearch");
+
+            this.services.set(new APMServices(meter, openTelemetry));
+            return null;
+        });
+    }
+
+    private void destroyApmServices() {
+        this.services.set(null);
+    }
+
+}

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

@@ -0,0 +1,35 @@
+/*
+ * 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 org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.telemetry.MetricName;
+import org.elasticsearch.telemetry.metric.Instrument;
+
+import java.util.Map;
+
+public class InstrumentRegistrar<T extends Instrument> {
+    private final Map<MetricName, T> registered = ConcurrentCollections.newConcurrentMap();
+
+    void register(T instrument) {
+        registered.compute(instrument.getName(), (k, v) -> {
+            if (v != null) {
+                throw new IllegalStateException(
+                    instrument.getClass().getSimpleName() + "[" + instrument.getName().getRawName() + "] already registered"
+                );
+            }
+
+            return instrument;
+        });
+    }
+
+    T get(MetricName name) {
+        return registered.get(name);
+    }
+}