Merge branch 'aws-observability:main' into update-bedrock

yiyuan-he · web-flow · commit 7e5e8d61eeb4 · 2024-10-11T15:25:28.000-07:00
diff --git a/awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsApplicationSignalsCustomizerProvider.java b/awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsApplicationSignalsCustomizerProvider.java
@@ -17,7 +17,6 @@
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
-import io.opentelemetry.api.metrics.MeterProvider;
 import io.opentelemetry.contrib.awsxray.AlwaysRecordSampler;
 import io.opentelemetry.contrib.awsxray.ResourceHolder;
 import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter;
@@ -164,14 +163,16 @@ private SdkTracerProviderBuilder customizeTracerProviderBuilder(
       MetricReader metricReader =
           PeriodicMetricReader.builder(metricsExporter).setInterval(exportInterval).build();
 
-      MeterProvider meterProvider =
+      SdkMeterProvider meterProvider =
           SdkMeterProvider.builder()
               .setResource(ResourceHolder.getResource())
               .registerMetricReader(metricReader)
               .build();
+
       // Construct and set application signals metrics processor
       SpanProcessor spanMetricsProcessor =
-          AwsSpanMetricsProcessorBuilder.create(meterProvider, ResourceHolder.getResource())
+          AwsSpanMetricsProcessorBuilder.create(
+                  meterProvider, ResourceHolder.getResource(), meterProvider::forceFlush)
               .build();
       tracerProviderBuilder.addSpanProcessor(spanMetricsProcessor);
     }
diff --git a/awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessor.java b/awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessor.java
@@ -22,12 +22,14 @@
 import io.opentelemetry.api.metrics.LongHistogram;
 import io.opentelemetry.api.trace.StatusCode;
 import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.trace.ReadWriteSpan;
 import io.opentelemetry.sdk.trace.ReadableSpan;
 import io.opentelemetry.sdk.trace.SpanProcessor;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import java.util.Map;
+import java.util.function.Supplier;
 import javax.annotation.concurrent.Immutable;
 
 /**
@@ -64,29 +66,38 @@ public final class AwsSpanMetricsProcessor implements SpanProcessor {
 
   private final MetricAttributeGenerator generator;
   private final Resource resource;
+  private final Supplier<CompletableResultCode> forceFlushAction;
 
   /** Use {@link AwsSpanMetricsProcessorBuilder} to construct this processor. */
   static AwsSpanMetricsProcessor create(
       LongHistogram errorHistogram,
       LongHistogram faultHistogram,
       DoubleHistogram latencyHistogram,
       MetricAttributeGenerator generator,
-      Resource resource) {
+      Resource resource,
+      Supplier<CompletableResultCode> forceFlushAction) {
     return new AwsSpanMetricsProcessor(
-        errorHistogram, faultHistogram, latencyHistogram, generator, resource);
+        errorHistogram, faultHistogram, latencyHistogram, generator, resource, forceFlushAction);
   }
 
   private AwsSpanMetricsProcessor(
       LongHistogram errorHistogram,
       LongHistogram faultHistogram,
       DoubleHistogram latencyHistogram,
       MetricAttributeGenerator generator,
-      Resource resource) {
+      Resource resource,
+      Supplier<CompletableResultCode> forceFlushAction) {
     this.errorHistogram = errorHistogram;
     this.faultHistogram = faultHistogram;
     this.latencyHistogram = latencyHistogram;
     this.generator = generator;
     this.resource = resource;
+    this.forceFlushAction = forceFlushAction;
+  }
+
+  @Override
+  public CompletableResultCode forceFlush() {
+    return forceFlushAction.get();
   }
 
   @Override
diff --git a/awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessorBuilder.java b/awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessorBuilder.java
@@ -22,7 +22,9 @@
 import io.opentelemetry.api.metrics.LongHistogram;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.resources.Resource;
+import java.util.function.Supplier;
 
 /** A builder for {@link AwsSpanMetricsProcessor} */
 public final class AwsSpanMetricsProcessorBuilder {
@@ -42,18 +44,29 @@ public final class AwsSpanMetricsProcessorBuilder {
   private final MeterProvider meterProvider;
   private final Resource resource;
 
+  // ForceFlush action provided from {@link SdkMeterProvider#forceFlush()} so that when the
+  // application exits The spanMetricProcessor calls the meterProvder.forceFlush to flush
+  // any remaining metrics before shutdown
+  private final Supplier<CompletableResultCode> forceFlushAction;
+
   // Optional builder elements
   private MetricAttributeGenerator generator = DEFAULT_GENERATOR;
   private String scopeName = DEFAULT_SCOPE_NAME;
 
   public static AwsSpanMetricsProcessorBuilder create(
-      MeterProvider meterProvider, Resource resource) {
-    return new AwsSpanMetricsProcessorBuilder(meterProvider, resource);
+      MeterProvider meterProvider,
+      Resource resource,
+      Supplier<CompletableResultCode> forceFlushAction) {
+    return new AwsSpanMetricsProcessorBuilder(meterProvider, resource, forceFlushAction);
   }
 
-  private AwsSpanMetricsProcessorBuilder(MeterProvider meterProvider, Resource resource) {
+  private AwsSpanMetricsProcessorBuilder(
+      MeterProvider meterProvider,
+      Resource resource,
+      Supplier<CompletableResultCode> forceFlushAction) {
     this.meterProvider = meterProvider;
     this.resource = resource;
+    this.forceFlushAction = forceFlushAction;
   }
 
   /**
@@ -86,6 +99,6 @@ public AwsSpanMetricsProcessor build() {
         meter.histogramBuilder(LATENCY).setUnit(LATENCY_UNITS).build();
 
     return AwsSpanMetricsProcessor.create(
-        errorHistogram, faultHistogram, latencyHistogram, generator, resource);
+        errorHistogram, faultHistogram, latencyHistogram, generator, resource, forceFlushAction);
   }
 }
diff --git a/awsagentprovider/src/test/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessorTest.java b/awsagentprovider/src/test/java/software/amazon/opentelemetry/javaagent/providers/AwsSpanMetricsProcessorTest.java
@@ -77,6 +77,12 @@ private enum ExpectedStatusMetric {
   private MetricAttributeGenerator generatorMock;
   private AwsSpanMetricsProcessor awsSpanMetricsProcessor;
 
+  // Mock forceFlush function that returns success when invoked similar
+  // to the default implementation of forceFlush.
+  private CompletableResultCode forceFlushAction() {
+    return CompletableResultCode.ofSuccess();
+  }
+
   @BeforeEach
   public void setUpMocks() {
     errorHistogramMock = mock(LongHistogram.class);
@@ -90,7 +96,8 @@ public void setUpMocks() {
             faultHistogramMock,
             latencyHistogramMock,
             generatorMock,
-            testResource);
+            testResource,
+            this::forceFlushAction);
   }
 
   @Test
diff --git a/instrumentation/jmx-metrics/src/main/resources/README.md b/instrumentation/jmx-metrics/src/main/resources/README.md
@@ -10,4 +10,7 @@ OTEL_EXPERIMENTAL_METRICS_VIEW_CONFIG: classpath:/jmx/view.yaml
 
 ### rules/*.yaml
 The rules are a translation of the JMX Metric Gatherer's [target systems](https://github.com/open-telemetry/opentelemetry-java-contrib/tree/main/jmx-metrics/src/main/resources/target-systems)
-based on the [JMX metric rule YAML schema](https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/instrumentation/jmx-metrics/javaagent/README.md#basic-syntax).
+based on the [JMX metric rule YAML schema](https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/instrumentation/jmx-metrics/javaagent/README.md#basic-syntax).
+
+### SystemCpuLoad
+The `SystemCpuLoad` metric is deprecated and Java versions 14+ now use `CpuLoad`. However, to avoid emitting double metrics, we stick to using `SystemCpuLoad` as it still works on newer versions.
diff --git a/instrumentation/jmx-metrics/src/main/resources/jmx/rules/jvm.yaml b/instrumentation/jmx-metrics/src/main/resources/jmx/rules/jvm.yaml
@@ -8,7 +8,7 @@ rules:
       LoadedClassCount:
         metric: loaded
         desc: Number of loaded classes
-  - bean: java.lang:type=GarbageCollector,*
+  - bean: java.lang:type=GarbageCollector,name=*
     prefix: jvm.gc.collections.
     type: counter
     metricAttribute:
@@ -27,13 +27,31 @@ rules:
     prefix: jvm.memory.
     type: gauge
     mapping:
+      HeapMemoryUsage.init:
+        metric: heap.init
+        desc: The initial amount of memory that the JVM requests from the operating system for the heap
       HeapMemoryUsage.used:
         metric: heap.used
         desc: The current heap usage
+      HeapMemoryUsage.committed:
+        metric: heap.committed
+        desc: The amount of memory that is guaranteed to be available for the heap
+      HeapMemoryUsage.max:
+        metric: heap.max
+        desc: The maximum amount of memory can be used for the heap
+      NonHeapMemoryUsage.init:
+        metric: nonheap.init
+        desc: The initial amount of memory that the JVM requests from the operating system for non-heap purposes
       NonHeapMemoryUsage.used:
         metric: nonheap.used
-        desc: The current non-heap usage
-  - bean: java.lang:type=MemoryPool,*
+        desc: The current non-heap memory usage
+      NonHeapMemoryUsage.committed:
+        metric: nonheap.committed
+        desc: The amount of memory that is guaranteed to be available for non-heap purposes
+      NonHeapMemoryUsage.max:
+        metric: nonheap.max
+        desc: The maximum amount of memory can be used for non-heap purposes
+  - bean: java.lang:type=MemoryPool,name=*
     unit: by
     prefix: jvm.memory.pool.
     type: gauge
@@ -43,26 +61,64 @@ rules:
       CollectionUsage.used:
         metric: used_after_last_gc
         desc: Memory used after the most recent gc event
+      Usage.init:
+        metric: init
+        desc: The initial amount of memory that the JVM requests from the operating system for the memory pool
       Usage.used:
         metric: used
-        desc: Current memory pool used
+        desc: The current memory pool memory usage
+      Usage.committed:
+        metric: committed
+        desc: The amount of memory that is guaranteed to be available for the memory pool
+      Usage.max:
+        metric: max
+        desc: The maximum amount of memory can be used for the memory pool
   - bean: java.lang:type=Threading
     unit: "1"
-    prefix: jvm.threads.
     type: gauge
     mapping:
       ThreadCount:
-        metric: count
+        metric: jvm.threads.count
         desc: Number of threads
-  - bean: java.lang:type=OperatingSystem,*
-    prefix: jvm.cpu.
+      DaemonThreadCount:
+        metric: jvm.daemon_threads.count
+        desc: Number of daemon threads
+  - bean: java.lang:type=OperatingSystem
     type: gauge
     mapping:
+      TotalSwapSpaceSize:
+        metric: jvm.system.swap.space.total
+        desc: The host swap memory size in bytes
+        unit: by
+      FreeSwapSpaceSize:
+        metric: jvm.system.swap.space.free
+        desc: The amount of available swap memory in bytes
+        unit: by
+      TotalPhysicalMemorySize:
+        metric: jvm.system.physical.memory.total
+        desc: The total physical memory size in host
+        unit: by
+      FreePhysicalMemorySize:
+        metric: jvm.system.physical.memory.free
+        desc: The amount of free physical memory in host
+        unit: by
+      AvailableProcessors:
+        metric: jvm.system.available.processors
+        desc: The number of available processors
+        unit: "1"
+      SystemCpuLoad:
+        metric: jvm.system.cpu.utilization
+        desc: The current load of CPU in host
+        unit: "1"
       ProcessCpuTime:
-        metric: time
+        metric: jvm.cpu.time
         unit: ns
         desc: CPU time used
       ProcessCpuLoad:
-        metric: recent_utilization
+        metric: jvm.cpu.recent_utilization
         unit: "1"
         desc: Recent CPU utilization for the process
+      OpenFileDescriptorCount:
+        metric: jvm.open_file_descriptor.count
+        desc: The number of opened file descriptors
+        unit: "1"
diff --git a/instrumentation/jmx-metrics/src/main/resources/jmx/rules/kafka-consumer.yaml b/instrumentation/jmx-metrics/src/main/resources/jmx/rules/kafka-consumer.yaml
@@ -0,0 +1,51 @@
+---
+rules:
+  - bean: kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*
+    metricAttribute:
+      client-id: param(client-id)
+    mapping:
+      fetch-rate:
+        metric: kafka.consumer.fetch-rate
+        type: gauge
+        desc: The number of fetch requests for all topics per second
+        unit: "1"
+      records-lag-max:
+        metric: kafka.consumer.records-lag-max
+        type: gauge
+        desc: Number of messages the consumer lags behind the producer
+        unit: "1"
+      bytes-consumed-rate:
+        metric: kafka.consumer.total.bytes-consumed-rate
+        type: gauge
+        desc: The average number of bytes consumed for all topics per second
+        unit: by
+      fetch-size-avg:
+        metric: kafka.consumer.total.fetch-size-avg
+        type: gauge
+        desc: The average number of bytes fetched per request for all topics
+        unit: by
+      records-consumed-rate:
+        metric: kafka.consumer.total.records-consumed-rate
+        type: gauge
+        desc: The average number of records consumed for all topics per second
+        unit: "1"
+  - bean: kafka.consumer:type=consumer-fetch-manager-metrics,client-id=*,topic=*
+    metricAttribute:
+      client-id: param(client-id)
+      topic: param(topic)
+    mapping:
+      bytes-consumed-rate:
+        metric: kafka.consumer.bytes-consumed-rate
+        type: gauge
+        desc: The average number of bytes consumed per second
+        unit: by
+      fetch-size-avg:
+        metric: kafka.consumer.fetch-size-avg
+        type: gauge
+        desc: The average number of bytes fetched per request
+        unit: by
+      records-consumed-rate:
+        metric: kafka.consumer.records-consumed-rate
+        type: gauge
+        desc: The average number of records consumed per second
+        unit: "1"
diff --git a/instrumentation/jmx-metrics/src/main/resources/jmx/rules/kafka-producer.yaml b/instrumentation/jmx-metrics/src/main/resources/jmx/rules/kafka-producer.yaml
@@ -0,0 +1,61 @@
+---
+rules:
+  - bean: kafka.producer:type=producer-metrics,client-id=*
+    metricAttribute:
+      client-id: param(client-id)
+    mapping:
+      io-wait-time-ns-avg:
+        metric: kafka.producer.io-wait-time-ns-avg
+        type: gauge
+        desc: The average length of time the I/O thread spent waiting for a socket ready for reads or writes
+        unit: ns
+      outgoing-byte-rate:
+        metric: kafka.producer.outgoing-byte-rate
+        type: gauge
+        desc: The average number of outgoing bytes sent per second to all servers
+        unit: by
+      request-latency-avg:
+        metric: kafka.producer.request-latency-avg
+        type: gauge
+        desc: The average request latency
+        unit: ms
+      request-rate:
+        metric: kafka-producer.request-rate
+        type: gauge
+        desc: The average number of requests sent per second
+        unit: "1"
+      response-rate:
+        metric: kafka.producer.response-rate
+        type: gauge
+        desc: Responses received per second
+        unit: "1"
+  - bean: kafka.producer:type=producer-topic-metrics,client-id=*,topic=*
+    metricAttribute:
+      client-id: param(client-id)
+      topic: param(topic)
+    mapping:
+      byte-rate:
+        metric: kafka.producer.byte-rate
+        type: gauge
+        desc: The average number of bytes sent per second for a topic
+        unit: by
+      compression-rate:
+        metric: kafka.producer.compression-rate
+        type: gauge
+        desc: The average compression rate of record batches for a topic
+        unit: "1"
+      record-error-rate:
+        metric: kafka.producer.record-error-rate
+        type: gauge
+        desc: The average per-second number of record sends that resulted in errors for a topic
+        unit: "1"
+      record-retry-rate:
+        metric: kafka.producer.record-retry-rate
+        type: gauge
+        desc: The average per-second number of retried record sends for a topic
+        unit: "1"
+      record-send-rate:
+        metric: kafka.producer.record-send-rate
+        type: gauge
+        desc: The average number of records sent per second for a topic
+        unit: "1"
diff --git a/instrumentation/jmx-metrics/src/main/resources/jmx/rules/kafka.yaml b/instrumentation/jmx-metrics/src/main/resources/jmx/rules/kafka.yaml
diff --git a/instrumentation/jmx-metrics/src/main/resources/jmx/rules/tomcat.yaml b/instrumentation/jmx-metrics/src/main/resources/jmx/rules/tomcat.yaml
