Fix Tomcat metric definitions to aggregate multiple MBeans. (open-tel…

…emetry#1366)

jmx-metrics/README.md

@@ -35,6 +35,10 @@ Kafka metric-gathering scripts determined by the comma-separated list in `otel.j
 it will then run the scripts on the desired interval length of `otel.jmx.interval.milliseconds` and
 export the resulting metrics.
 
+Some metrics (e.g. `tomcat.sessions`) are configured to query multiple MBeans. By default, only the value in the first MBean
+is recorded for the metric and all other values are dropped. To aggregate the MBean values together, set the
+`otel.jmx.aggregate.across.mbeans` property to `true`.
+
 For custom metrics and unsupported targets, you can provide your own MBean querying scripts to produce
 OpenTelemetry instruments:
 

jmx-metrics/src/main/groovy/io/opentelemetry/contrib/jmxmetrics/GroovyRunner.java

@@ -70,7 +70,8 @@ public class GroovyRunner {
     Binding binding = new Binding();
     binding.setVariable("log", logger);
 
-    OtelHelper otelHelper = new OtelHelper(jmxClient, this.groovyMetricEnvironment);
+    OtelHelper otelHelper =
+        new OtelHelper(jmxClient, this.groovyMetricEnvironment, config.aggregateAcrossMBeans);
     binding.setVariable("otel", otelHelper);
 
     for (final Script script : scripts) {

jmx-metrics/src/main/groovy/io/opentelemetry/contrib/jmxmetrics/InstrumentHelper.groovy

@@ -45,6 +45,7 @@ class InstrumentHelper {
     private final Map<String, Closure> labelFuncs
     private final Closure instrument
     private final GroovyMetricEnvironment metricEnvironment
+    private final boolean aggregateAcrossMBeans
 
     /**
      * An InstrumentHelper provides the ability to easily create and update {@link io.opentelemetry.api.metrics.Instrument}
@@ -63,8 +64,9 @@ class InstrumentHelper {
      *        (e.g. new OtelHelper().&doubleValueRecorder)
      * @param metricenvironment - The {@link GroovyMetricEnvironment} used to register callbacks onto the SDK meter for
      *        batch callbacks used to handle {@link CompositeData}
+     * @param aggregateAcrossMBeans - Whether to aggregate multiple MBeans together before recording.
      */
-    InstrumentHelper(MBeanHelper mBeanHelper, String instrumentName, String description, String unit, Map<String, Closure<?>> labelFuncs, Map<String, Map<String, Closure<?>>> MBeanAttributes, Closure<?> instrument, GroovyMetricEnvironment metricEnvironment) {
+    InstrumentHelper(MBeanHelper mBeanHelper, String instrumentName, String description, String unit, Map<String, Closure<?>> labelFuncs, Map<String, Map<String, Closure<?>>> MBeanAttributes, Closure<?> instrument, GroovyMetricEnvironment metricEnvironment, boolean aggregateAcrossMBeans) {
         this.mBeanHelper = mBeanHelper
         this.instrumentName = instrumentName
         this.description = description
@@ -73,6 +75,7 @@ class InstrumentHelper {
         this.mBeanAttributes = MBeanAttributes
         this.instrument = instrument
         this.metricEnvironment = metricEnvironment
+        this.aggregateAcrossMBeans = aggregateAcrossMBeans
     }
 
     void update() {
@@ -181,19 +184,39 @@ class InstrumentHelper {
         return labels
     }
 
+    private static String getAggregationKey(String instrumentName, Map<String, String> labels) {
+        def labelsKey = labels.sort().collect { key, value -> "${key}:${value}" }.join(";")
+        return "${instrumentName}/${labelsKey}"
+    }
+
     // Create a closure for simple attributes that will retrieve mbean information on
     // callback to ensure that metrics are collected on request
     private Closure prepareUpdateClosure(List<GroovyMBean> mbeans, attributes) {
         return { result ->
+            def aggregations = [:] as Map<String, Aggregation>
+            boolean requireAggregation = aggregateAcrossMBeans && mbeans.size() > 1 && instrumentIsValue(instrument)
             [mbeans, attributes].combinations().each { pair ->
                 def (mbean, attribute) = pair
                 def value = MBeanHelper.getBeanAttribute(mbean, attribute)
                 if (value != null) {
                     def labels = getLabels(mbean, labelFuncs, mBeanAttributes[attribute])
-                    logger.fine("Recording ${instrumentName} - ${instrument.method} w/ ${value} - ${labels}")
-                    recordDataPoint(instrument, result, value, GroovyMetricEnvironment.mapToAttributes(labels))
+                    if (requireAggregation) {
+                        def key = getAggregationKey(instrumentName, labels)
+                        if (aggregations[key] == null) {
+                            aggregations[key] = new Aggregation(labels)
+                        }
+                        logger.fine("Aggregating ${mbean.name()} ${instrumentName} - ${instrument.method} w/ ${value} - ${labels}")
+                        aggregations[key].add(value)
+                    } else {
+                        logger.fine("Recording ${mbean.name()} ${instrumentName} - ${instrument.method} w/ ${value} - ${labels}")
+                        recordDataPoint(instrument, result, value, GroovyMetricEnvironment.mapToAttributes(labels))
+                    }
                 }
             }
+            aggregations.each { entry ->
+                logger.fine("Recording ${instrumentName} - ${instrument.method} - w/ ${entry.value.value} - ${entry.value.labels}")
+                recordDataPoint(instrument, result, entry.value.value, GroovyMetricEnvironment.mapToAttributes(entry.value.labels))
+            }
         }
     }
 
@@ -252,6 +275,14 @@ class InstrumentHelper {
         ].contains(inst.method)
     }
 
+    @PackageScope
+    static boolean instrumentIsValue(inst) {
+        return [
+            "doubleValueCallback",
+            "longValueCallback"
+        ].contains(inst.method)
+    }
+
     @PackageScope
     static boolean instrumentIsCounter(inst) {
         return [
@@ -261,4 +292,18 @@ class InstrumentHelper {
           "longUpDownCounter"
         ].contains(inst.method)
     }
+
+    static class Aggregation {
+        private final Map<String, String> labels
+        private def value
+
+        Aggregation(Map<String, String> labels) {
+            this.labels = labels
+            this.value = 0.0
+        }
+
+        void add(value) {
+            this.value += value
+        }
+    }
 }

jmx-metrics/src/main/groovy/io/opentelemetry/contrib/jmxmetrics/JmxConfig.java

@@ -31,6 +31,7 @@ class JmxConfig {
   static final String JMX_PASSWORD = PREFIX + "jmx.password";
   static final String JMX_REMOTE_PROFILE = PREFIX + "jmx.remote.profile";
   static final String JMX_REALM = PREFIX + "jmx.realm";
+  static final String JMX_AGGREGATE_ACROSS_MBEANS = PREFIX + "jmx.aggregate.across.mbeans";
 
   // These properties need to be copied into System Properties if provided via the property
   // file so that they are available to the JMX Connection builder
@@ -77,6 +78,8 @@ class JmxConfig {
   final boolean registrySsl;
   final Properties properties;
 
+  final boolean aggregateAcrossMBeans;
+
   JmxConfig(final Properties props) {
     properties = new Properties();
     // putAll() instead of using constructor defaults
@@ -112,6 +115,8 @@ class JmxConfig {
     realm = properties.getProperty(JMX_REALM);
 
     registrySsl = Boolean.valueOf(properties.getProperty(REGISTRY_SSL));
+    aggregateAcrossMBeans =
+        Boolean.parseBoolean(properties.getProperty(JMX_AGGREGATE_ACROSS_MBEANS));
 
     // For the list of System Properties, if they have been set in the properties file
     // they need to be set in Java System Properties.

jmx-metrics/src/main/groovy/io/opentelemetry/contrib/jmxmetrics/OtelHelper.groovy

@@ -22,10 +22,12 @@ class OtelHelper {
 
     private final JmxClient jmxClient
     private final GroovyMetricEnvironment groovyMetricEnvironment
+    private final boolean aggregateAcrossMBeans
 
-    OtelHelper(JmxClient jmxClient, GroovyMetricEnvironment groovyMetricEnvironment) {
+    OtelHelper(JmxClient jmxClient, GroovyMetricEnvironment groovyMetricEnvironment, boolean aggregateAcrossMBeans) {
         this.jmxClient = jmxClient
         this.groovyMetricEnvironment = groovyMetricEnvironment
+        this.aggregateAcrossMBeans = aggregateAcrossMBeans
     }
 
     /**
@@ -99,7 +101,7 @@ class OtelHelper {
      * attribute value(s).  The parameters map to the InstrumentHelper constructor.
      */
     InstrumentHelper instrument(MBeanHelper mBeanHelper, String instrumentName, String description, String unit, Map<String, Closure> labelFuncs, Map<String, Map<String, Closure>> attributes, Closure otelInstrument) {
-        def instrumentHelper = new InstrumentHelper(mBeanHelper, instrumentName, description, unit, labelFuncs, attributes, otelInstrument, groovyMetricEnvironment)
+        def instrumentHelper = new InstrumentHelper(mBeanHelper, instrumentName, description, unit, labelFuncs, attributes, otelInstrument, groovyMetricEnvironment, aggregateAcrossMBeans)
         instrumentHelper.update()
         return instrumentHelper
     }

jmx-metrics/src/main/resources/target-systems/tomcat.groovy

@@ -15,10 +15,10 @@
  */
 
 
-def beantomcatmanager = otel.mbean("Catalina:type=Manager,host=localhost,context=*")
-otel.instrument(beantomcatmanager, "tomcat.sessions", "The number of active sessions.", "sessions", "activeSessions", otel.&doubleValueCallback)
+def beantomcatmanager = otel.mbeans("Catalina:type=Manager,host=localhost,context=*")
+otel.instrument(beantomcatmanager, "tomcat.sessions", "The number of active sessions.", "sessions", "activeSessions", otel.&longValueCallback)
 
-def beantomcatrequestProcessor = otel.mbean("Catalina:type=GlobalRequestProcessor,name=*")
+def beantomcatrequestProcessor = otel.mbeans("Catalina:type=GlobalRequestProcessor,name=*")
 otel.instrument(beantomcatrequestProcessor, "tomcat.errors", "The number of errors encountered.", "errors",
   ["proto_handler" : { mbean -> mbean.name().getKeyProperty("name") }],
   "errorCount", otel.&longCounterCallback)
@@ -37,15 +37,15 @@ otel.instrument(beantomcatrequestProcessor, "tomcat.traffic",
   ["bytesReceived":["direction" : {"received"}], "bytesSent": ["direction" : {"sent"}]],
   otel.&longCounterCallback)
 
-def beantomcatconnectors = otel.mbean("Catalina:type=ThreadPool,name=*")
+def beantomcatconnectors = otel.mbeans("Catalina:type=ThreadPool,name=*")
 otel.instrument(beantomcatconnectors, "tomcat.threads", "The number of threads", "threads",
   ["proto_handler" : { mbean -> mbean.name().getKeyProperty("name") }],
   ["currentThreadCount":["state":{"idle"}],"currentThreadsBusy":["state":{"busy"}]], otel.&longValueCallback)
 
-def beantomcatnewmanager = otel.mbean("Tomcat:type=Manager,host=localhost,context=*")
-otel.instrument(beantomcatnewmanager, "tomcat.sessions", "The number of active sessions.", "sessions", "activeSessions", otel.&doubleValueCallback)
+def beantomcatnewmanager = otel.mbeans("Tomcat:type=Manager,host=localhost,context=*")
+otel.instrument(beantomcatnewmanager, "tomcat.sessions", "The number of active sessions.", "sessions", "activeSessions", otel.&longValueCallback)
 
-def beantomcatnewrequestProcessor = otel.mbean("Tomcat:type=GlobalRequestProcessor,name=*")
+def beantomcatnewrequestProcessor = otel.mbeans("Tomcat:type=GlobalRequestProcessor,name=*")
 otel.instrument(beantomcatnewrequestProcessor, "tomcat.errors", "The number of errors encountered.", "errors",
     ["proto_handler" : { mbean -> mbean.name().getKeyProperty("name") }],
     "errorCount", otel.&longCounterCallback)
@@ -64,7 +64,7 @@ otel.instrument(beantomcatnewrequestProcessor, "tomcat.traffic",
     ["bytesReceived":["direction" : {"received"}], "bytesSent": ["direction" : {"sent"}]],
     otel.&longCounterCallback)
 
-def beantomcatnewconnectors = otel.mbean("Tomcat:type=ThreadPool,name=*")
+def beantomcatnewconnectors = otel.mbeans("Tomcat:type=ThreadPool,name=*")
 otel.instrument(beantomcatnewconnectors, "tomcat.threads", "The number of threads", "threads",
     ["proto_handler" : { mbean -> mbean.name().getKeyProperty("name") }],
     ["currentThreadCount":["state":{"idle"}],"currentThreadsBusy":["state":{"busy"}]], otel.&longValueCallback)

jmx-metrics/src/test/java/io/opentelemetry/contrib/jmxmetrics/InstrumenterHelperTest.java

@@ -96,7 +96,7 @@ void setupOtel() {
     metricReader = InMemoryMetricReader.create();
     meterProvider = SdkMeterProvider.builder().registerMetricReader(metricReader).build();
     metricEnvironment = new GroovyMetricEnvironment(meterProvider, "otel.test");
-    otel = new OtelHelper(jmxClient, metricEnvironment);
+    otel = new OtelHelper(jmxClient, metricEnvironment, false);
   }
 
   @AfterEach
@@ -429,7 +429,36 @@ void doubleValueCallback() throws Exception {
     }
 
     @Test
-    void doubleValueCallbackMultipleMBeans() throws Exception {
+    void doubleValueCallbackMBeans() throws Exception {
+      String instrumentMethod = "doubleValueCallback";
+      String thingName = "multiple:type=" + instrumentMethod + ".Thing";
+      MBeanHelper mBeanHelper = registerThings(thingName);
+
+      String instrumentName = "multiple." + instrumentMethod + ".gauge";
+      String description = "multiple double gauge description";
+
+      updateWithHelper(
+          mBeanHelper,
+          instrumentMethod,
+          instrumentName,
+          description,
+          "Double",
+          new HashMap<>(),
+          /* aggregateAcrossMBeans= */ true);
+
+      assertThat(metricReader.collectAllMetrics())
+          .satisfiesExactly(
+              metric ->
+                  assertThat(metric)
+                      .hasName(instrumentName)
+                      .hasDescription(description)
+                      .hasUnit("1")
+                      .hasDoubleGaugeSatisfying(
+                          gauge -> gauge.hasPointsSatisfying(assertDoublePoint())));
+    }
+
+    @Test
+    void doubleValueCallbackListMBeans() throws Exception {
       String instrumentMethod = "doubleValueCallback";
       ArrayList<String> thingNames = new ArrayList<>();
       for (int i = 0; i < 4; i++) {
@@ -515,6 +544,12 @@ void longValueCallback() throws Exception {
                           gauge -> gauge.hasPointsSatisfying(assertLongPoints())));
     }
 
+    @SuppressWarnings("unchecked")
+    private Consumer<DoublePointAssert>[] assertDoublePoint() {
+      return Stream.<Consumer<DoublePointAssert>>of(point -> point.hasValue(123.456 * 4))
+          .toArray(Consumer[]::new);
+    }
+
     @SuppressWarnings("unchecked")
     private Consumer<DoublePointAssert>[] assertDoublePoints() {
       return Stream.<Consumer<DoublePointAssert>>of(
@@ -679,11 +714,29 @@ void updateWithHelper(
       String instrumentName,
       String description,
       String attribute) {
-    Closure<?> instrument = (Closure<?>) Eval.me("otel", otel, "otel.&" + instrumentMethod);
     Map<String, Closure<?>> labelFuncs = new HashMap<>();
     labelFuncs.put("labelOne", (Closure<?>) Eval.me("{ unused -> 'labelOneValue' }"));
     labelFuncs.put(
         "labelTwo", (Closure<?>) Eval.me("{ mbean -> mbean.name().getKeyProperty('thing') }"));
+    updateWithHelper(
+        mBeanHelper,
+        instrumentMethod,
+        instrumentName,
+        description,
+        attribute,
+        labelFuncs,
+        /* aggregateAcrossMBeans= */ false);
+  }
+
+  void updateWithHelper(
+      MBeanHelper mBeanHelper,
+      String instrumentMethod,
+      String instrumentName,
+      String description,
+      String attribute,
+      Map<String, Closure<?>> labelFuncs,
+      boolean aggregateAcrossMBeans) {
+    Closure<?> instrument = (Closure<?>) Eval.me("otel", otel, "otel.&" + instrumentMethod);
     InstrumentHelper instrumentHelper =
         new InstrumentHelper(
             mBeanHelper,
@@ -693,7 +746,8 @@ void updateWithHelper(
             labelFuncs,
             Collections.singletonMap(attribute, null),
             instrument,
-            metricEnvironment);
+            metricEnvironment,
+            aggregateAcrossMBeans);
     instrumentHelper.update();
   }
 
@@ -714,7 +768,8 @@ void updateWithHelperMultiAttribute(
             labelFuncs,
             attributes,
             instrument,
-            metricEnvironment);
+            metricEnvironment,
+            false);
     instrumentHelper.update();
   }
 

jmx-metrics/src/test/java/io/opentelemetry/contrib/jmxmetrics/JmxConfigTest.java

@@ -52,6 +52,7 @@ void defaultValues() {
     assertThat(config.remoteProfile).isNull();
     assertThat(config.realm).isNull();
     assertThat(config.properties.getProperty("otel.metric.export.interval")).isEqualTo("10000");
+    assertThat(config.aggregateAcrossMBeans).isFalse();
   }
 
   @Test
@@ -87,6 +88,7 @@ void specifiedValues() {
     assertThat(config.password).isEqualTo("myPassword");
     assertThat(config.remoteProfile).isEqualTo("myRemoteProfile");
     assertThat(config.realm).isEqualTo("myRealm");
+    assertThat(config.aggregateAcrossMBeans).isFalse();
   }
 
   @Test
@@ -109,6 +111,7 @@ void propertiesFile() {
     assertThat(config.password).isEqualTo("myPassw\\ord");
     assertThat(config.remoteProfile).isEqualTo("SASL/DIGEST-MD5");
     assertThat(config.realm).isEqualTo("myRealm");
+    assertThat(config.aggregateAcrossMBeans).isTrue();
 
     // These properties are set from the config file loading into JmxConfig
     assertThat(System.getProperty("javax.net.ssl.keyStore")).isEqualTo("/my/key/store");

jmx-metrics/src/test/java/io/opentelemetry/contrib/jmxmetrics/OtelHelperAsynchronousMetricTest.java

@@ -29,7 +29,7 @@ class OtelHelperAsynchronousMetricTest {
   void setUp() {
     metricReader = InMemoryMetricReader.create();
     meterProvider = SdkMeterProvider.builder().registerMetricReader(metricReader).build();
-    otel = new OtelHelper(null, new GroovyMetricEnvironment(meterProvider, "otel.test"));
+    otel = new OtelHelper(null, new GroovyMetricEnvironment(meterProvider, "otel.test"), false);
   }
 
   @Test

jmx-metrics/src/test/java/io/opentelemetry/contrib/jmxmetrics/OtelHelperJmxTest.java

@@ -132,7 +132,7 @@ private static JMXServiceURL setupServer(Map<String, String> env) throws Excepti
   }
 
   private static OtelHelper setupHelper(JmxConfig config) throws Exception {
-    return new OtelHelper(new JmxClient(config), new GroovyMetricEnvironment(config));
+    return new OtelHelper(new JmxClient(config), new GroovyMetricEnvironment(config), false);
   }
 
   private static void verifyClient(Properties props) throws Exception {

jmx-metrics/src/test/java/io/opentelemetry/contrib/jmxmetrics/OtelHelperSynchronousMetricTest.java

@@ -33,7 +33,7 @@ class OtelHelperSynchronousMetricTest {
   void setUp() {
     metricReader = InMemoryMetricReader.create();
     meterProvider = SdkMeterProvider.builder().registerMetricReader(metricReader).build();
-    otel = new OtelHelper(null, new GroovyMetricEnvironment(meterProvider, "otel.test"));
+    otel = new OtelHelper(null, new GroovyMetricEnvironment(meterProvider, "otel.test"), false);
   }
 
   @Test

jmx-metrics/src/test/resources/all.properties

@@ -19,3 +19,4 @@ javax.net.ssl.keyStoreType=JKS
 javax.net.ssl.trustStore=/my/trust/store
 javax.net.ssl.trustStorePassword=def456
 javax.net.ssl.trustStoreType=JKS
+otel.jmx.aggregate.across.mbeans=true