Update TCK tests for OTel metrics types and units

tck/pom.xml

@@ -32,7 +32,7 @@
         <checkstyle.methodNameFormat>^_?[a-z][a-zA-Z0-9_]*$</checkstyle.methodNameFormat>
         <microprofile-config-api.version>3.1</microprofile-config-api.version>
         <microprofile-metrics-api.version>4.0</microprofile-metrics-api.version>
-        <microprofile-telemetry-api.version>2.0</microprofile-telemetry-api.version>
+        <microprofile-telemetry-api.version>2.0-RC2</microprofile-telemetry-api.version>
     </properties>
 
     <dependencyManagement>

tck/src/main/java/org/eclipse/microprofile/fault/tolerance/tck/metrics/common/util/TimeUtils.java

@@ -51,6 +51,25 @@ public static Matcher<Long> approxMillis(final long originalMillis) {
         return Matchers.allOf(greaterThan(nanos - error), lessThan(nanos + error));
     }
 
+    /**
+     * Check that a floating point time in seconds is within 20% of an expected time in milliseconds
+     * <p>
+     * Note that this method applies any timeout scaling configured in TCKConfig, does the millseconds to nanoseconds
+     * conversion and creates a {@link Matcher} to do the check.
+     * <p>
+     * Useful for checking the results from Telemetry Histograms.
+     *
+     * @param originalMillis
+     *            the expected time in milliseconds
+     * @return a {@link Matcher} which matches against a time in seconds
+     */
+    public static Matcher<Double> approxMillisFromSeconds(final long originalMillis) {
+        long millis = TCKConfig.getConfig().getTimeoutInMillis(originalMillis);
+        double seconds = millis / 1_000d;
+        double error = seconds * 0.2;
+        return Matchers.closeTo(seconds, error);
+    }
+
     /**
      * Check that a nanosecond time is less than an expected time in milliseconds
      * <p>

tck/src/main/java/org/eclipse/microprofile/fault/tolerance/tck/telemetryMetrics/BulkheadTelemetryTest.java

@@ -19,7 +19,7 @@
 package org.eclipse.microprofile.fault.tolerance.tck.telemetryMetrics;
 
 import static java.util.concurrent.TimeUnit.MINUTES;
-import static org.eclipse.microprofile.fault.tolerance.tck.metrics.common.util.TimeUtils.approxMillis;
+import static org.eclipse.microprofile.fault.tolerance.tck.metrics.common.util.TimeUtils.approxMillisFromSeconds;
 import static org.eclipse.microprofile.fault.tolerance.tck.telemetryMetrics.util.TelemetryMetricDefinition.BulkheadResult.ACCEPTED;
 import static org.eclipse.microprofile.fault.tolerance.tck.telemetryMetrics.util.TelemetryMetricDefinition.BulkheadResult.REJECTED;
 import static org.eclipse.microprofile.fault.tolerance.tck.telemetryMetrics.util.TelemetryMetricDefinition.InvocationResult.EXCEPTION_THROWN;
@@ -32,7 +32,6 @@
 import static org.testng.Assert.assertTrue;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -115,17 +114,14 @@ private CompletableFuture<Void> newWaitingFuture() {
 
     @Test(groups = "main")
     public void bulkheadMetricTest() throws InterruptedException, ExecutionException, TimeoutException {
-        System.out.println("GREP + bulkheadMetricTest start");
         TelemetryMetricGetter m = new TelemetryMetricGetter(BulkheadMetricBean.class, "waitFor");
         m.baselineMetrics();
-        System.out.println("GREP + bulkheadMetricTest after baseline");
 
         CompletableFuture<Void> waitingFuture = newWaitingFuture();
 
         Future<?> f1 = async.run(() -> bulkheadBean.waitFor(waitingFuture));
         Future<?> f2 = async.run(() -> bulkheadBean.waitFor(waitingFuture));
 
-        System.out.println("GREP + bulkheadMetricTest before fail");
         bulkheadBean.waitForRunningExecutions(2);
         assertThat("executions running", m.getBulkheadExecutionsRunning().value(), is(2L));
 
@@ -202,28 +198,23 @@ public void bulkheadMetricHistogramTest() throws InterruptedException, Execution
         f1.get(1, MINUTES);
         f2.get(1, MINUTES);
 
-        Long executionTimesCount = m.getBulkheadRunningDuration().getHistogramCount().get();
-        assertThat("histogram count", executionTimesCount, is(2L)); // Rejected executions
-                                                                    // not recorded in
+        HistogramPointData executionTimesPoint = m.getBulkheadRunningDuration()
+                .getHistogramPoint()
+                .orElseThrow(() -> new AssertionError("No data reported for ft.bulkhead.runningDuration"));
 
-        Collection<HistogramPointData> executionTimesPoints = m.getBulkheadRunningDuration().getHistogramPoints();
-        double time = executionTimesPoints.stream()
-                .mapToDouble(points -> points.getSum())
-                .sum();
+        double time = executionTimesPoint.getSum();
+        long count = executionTimesPoint.getCount();
 
-        long count = executionTimesPoints.stream()
-                .mapToLong(points -> points.getCount())
-                .sum();
-
-        assertThat("mean", Math.round(time / count), approxMillis(1000)); // histogram
+        assertEquals(count, 2L, "histogram count"); // Rejected executions not recorded in histogram
+        assertThat("mean", time / count, approxMillisFromSeconds(1000));
 
         // Now let's put some quick results through the bulkhead
         bulkheadBean.waitForHistogram(CompletableFuture.completedFuture(null));
         bulkheadBean.waitForHistogram(CompletableFuture.completedFuture(null));
 
         // Should have 4 results, ~0ms * 2 and ~1000ms * 2
-        executionTimesCount = m.getBulkheadRunningDuration().getHistogramCount().get();
-        assertThat("histogram count", executionTimesCount, is(4L));
+        m.getBulkheadRunningDuration().assertBucketCounts(1000, 1000, 0, 0);
+        m.getBulkheadRunningDuration().assertBoundaries();
     }
 
     @Test(groups = "main")
@@ -236,6 +227,7 @@ public void bulkheadMetricAsyncTest() throws InterruptedException, ExecutionExce
         Future<?> f1 = bulkheadBean.waitForAsync(waitingFuture);
         Future<?> f2 = bulkheadBean.waitForAsync(waitingFuture);
         bulkheadBean.waitForRunningExecutions(2);
+        long startTime = System.nanoTime();
 
         Future<?> f3 = bulkheadBean.waitForAsync(waitingFuture);
         Future<?> f4 = bulkheadBean.waitForAsync(waitingFuture);
@@ -248,6 +240,10 @@ public void bulkheadMetricAsyncTest() throws InterruptedException, ExecutionExce
 
         Thread.sleep(config.getTimeoutInMillis(1000));
         waitingFuture.complete(null);
+        long durationms = (System.nanoTime() - startTime) / 1_000_000;
+        durationms /= config.getBaseMultiplier(); // This value is used with approxMillis which always applies the
+                                                  // baseMultiplier
+                                                  // so preemptively divide it by the baseMultiplier here
 
         f1.get(1, MINUTES);
         f2.get(1, MINUTES);
@@ -258,10 +254,9 @@ public void bulkheadMetricAsyncTest() throws InterruptedException, ExecutionExce
         assertThat("accepted calls", m.getBulkheadCalls(ACCEPTED).delta(), is(4L));
         assertThat("rejections", m.getBulkheadCalls(REJECTED).delta(), is(1L));
 
-        Long queueWaits = m.getBulkheadWaitingDuration().getHistogramCount().get();
-
         // Expect 2 * wait for 0ms, 2 * wait for durationms
-        assertThat("waiting duration histogram counts", queueWaits, is(4L));
+        m.getBulkheadWaitingDuration().assertBucketCounts(0, 0, durationms, durationms);
+        m.getBulkheadWaitingDuration().assertBoundaries();
 
         // General metrics should be updated
         assertThat("successful invocations", m.getInvocations(VALUE_RETURNED, InvocationFallback.NOT_DEFINED).delta(),

tck/src/main/java/org/eclipse/microprofile/fault/tolerance/tck/telemetryMetrics/TimeoutTelemetryTest.java

@@ -113,8 +113,8 @@ public void testTimeoutHistogram() {
                                                                                                            // timeout
                                                                                                            // after 2000
 
-        Long histogramCount = m.getTimeoutExecutionDuration().getHistogramCount().get();
-        assertThat("Histogram count", histogramCount, is(2L));
+        m.getTimeoutExecutionDuration().assertBucketCounts(300, 2000);
+        m.getTimeoutExecutionDuration().assertBoundaries();
     }
 
     @Test(dependsOnGroups = "main")

tck/src/main/java/org/eclipse/microprofile/fault/tolerance/tck/telemetryMetrics/util/InMemoryMetricReader.java

@@ -19,17 +19,34 @@
  *******************************************************************************/
 package org.eclipse.microprofile.fault.tolerance.tck.telemetryMetrics.util;
 
+import static io.opentelemetry.sdk.metrics.data.AggregationTemporality.CUMULATIVE;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.lessThan;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
 import java.util.Collection;
 import java.util.List;
-import java.util.NoSuchElementException;
+import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Optional;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.metrics.InstrumentType;
 import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.data.GaugeData;
+import io.opentelemetry.sdk.metrics.data.HistogramPointData;
 import io.opentelemetry.sdk.metrics.data.LongPointData;
 import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.data.MetricDataType;
+import io.opentelemetry.sdk.metrics.data.PointData;
+import io.opentelemetry.sdk.metrics.data.SumData;
 import io.opentelemetry.sdk.metrics.export.CollectionRegistration;
 import io.opentelemetry.sdk.metrics.export.MetricReader;
 import jakarta.enterprise.context.ApplicationScoped;
@@ -71,61 +88,144 @@ public CompletableResultCode shutdown() {
         return CompletableResultCode.ofSuccess();
     }
 
+    /**
+     * Get the metric for the given {@code id}.
+     * <p>
+     * If the metric exists but is the wrong type, an assertionError is thrown.
+     *
+     * @param id
+     *            the metric ID
+     * @return the metric data, or an empty {@code Optional}
+     * @throws AssertionError
+     *             if the metric exists but has the wrong type
+     */
+    public Optional<MetricData> getMetric(TelemetryMetricID id) {
+        Optional<MetricData> result = getMetric(id.name);
+        result.ifPresent(md -> validateMetricType(md, id));
+        return result;
+    }
+
+    private Optional<MetricData> getMetric(String name) {
+        Collection<MetricData> allMetrics = collectionRegistration.collectAllMetrics();
+        List<MetricData> matchingMetrics = allMetrics.stream()
+                .filter(md -> md.getName().equals(name))
+                .collect(Collectors.toList());
+
+        assertThat("More than one MetricData found for name: " + name,
+                matchingMetrics, hasSize(lessThan(2)));
+
+        return matchingMetrics.isEmpty() ? Optional.empty() : Optional.of(matchingMetrics.get(0));
+    }
+
     public long readLongData(TelemetryMetricID id) {
-        @SuppressWarnings("unchecked")
-        List<LongPointData> longData = (List<LongPointData>) getPointData(id);
+        return getMetric(id)
+                .flatMap(md -> getLongPointData(md, id))
+                .map(LongPointData::getValue)
+                .orElse(0L);
+    }
 
-        return longData.stream()
-                .mapToLong(LongPointData::getValue)
-                .sum();
+    public String getUnit(String metricName) {
+        return getMetric(metricName)
+                .orElseThrow(() -> new IllegalStateException("No metric found for name: " + metricName))
+                .getUnit();
     }
 
-    protected List<?> getPointData(TelemetryMetricID id) {
-        Collection<MetricData> allMetrics = collectionRegistration.collectAllMetrics();
+    public static Optional<LongPointData> getLongPointData(MetricData md, TelemetryMetricID id) {
+        switch (md.getType()) {
+            case LONG_GAUGE :
+                GaugeData<LongPointData> gaugeData = md.getLongGaugeData();
+                return getGaugePointData(gaugeData, id);
+            case LONG_SUM :
+                SumData<LongPointData> sumData = md.getLongSumData();
+                return getSumPointData(sumData, id);
+            default :
+                throw new IllegalStateException("Metric " + id.name + " does not have long type data");
+        }
+    }
 
-        return allMetrics.stream()
-                .filter(
-                        md -> md.getName().equals(id.name))
-                .flatMap(
-                        md -> md.getData().getPoints().stream())
-                .filter(
-                        point -> id.attributes.asMap().keySet().stream()
-                                .allMatch(key -> point.getAttributes().asMap().containsKey(key)
-                                        && id.attributes.asMap().get(key)
-                                                .equals(point.getAttributes().asMap().get(key))))
+    public static Optional<HistogramPointData> getHistogramPointData(MetricData md, TelemetryMetricID id) {
+        assertEquals(md.getType(), MetricDataType.HISTOGRAM, "Metric " + id.name + " is not a histogram");
+        assertEquals(md.getHistogramData().getAggregationTemporality(), AggregationTemporality.CUMULATIVE,
+                "Metric " + id.name + " has wrong temporality");
+
+        List<HistogramPointData> data = md.getHistogramData().getPoints().stream()
+                .filter(hasAllAttributes(id.attributes))
                 .collect(Collectors.toList());
+
+        assertThat("Found more than one data point for metric: " + id, data, hasSize(lessThan(2)));
+
+        return data.isEmpty() ? Optional.empty() : Optional.of(data.get(0));
     }
 
-    public Optional<LongPointData> getGaugueMetricLatestValue(TelemetryMetricID id) {
-        Collection<MetricData> allMetrics = collectionRegistration.collectAllMetrics();
+    private static <T extends PointData> Optional<T> getGaugePointData(GaugeData<T> gaugeData, TelemetryMetricID id) {
+        List<T> data = gaugeData.getPoints().stream()
+                .filter(hasAllAttributes(id.attributes))
+                .collect(Collectors.toList());
+        assertThat("Found more than one data point for metric: " + id, data, hasSize(lessThan(2)));
 
-        Optional<LongPointData> gague = allMetrics.stream()
-                .filter(
-                        md -> md.getName().equals(id.name))
-                .flatMap(md -> md.getLongGaugeData().getPoints().stream())
-                .filter(point -> id.attributes.asMap().keySet().stream()
-                        .allMatch(key -> point.getAttributes().asMap().containsKey(key)
-                                && id.attributes.asMap().get(key)
-                                        .equals(point.getAttributes().asMap().get(key))))
-                // feeding the points into Long.compare in reverse order will return the largest first.
-                .sorted((pointOne, pointTwo) -> Long.compare(pointTwo.getEpochNanos(), pointOne.getEpochNanos()))
-                .findFirst();
-
-        return gague;
+        return data.isEmpty() ? Optional.empty() : Optional.of(data.get(0));
     }
 
-    public String getUnit(String metricName) {
-        try {
-            Collection<MetricData> allMetrics = collectionRegistration.collectAllMetrics();
-            Optional<MetricData> mathcingData = allMetrics.stream()
-                    .filter(
-                            md -> md.getName().equals(metricName))
-                    .findAny();
-
-            return mathcingData.get().getUnit();
-        } catch (NoSuchElementException e) {
-            // If we didn't find anything throwing an exception to fail the test is reasonable
-            throw new RuntimeException("Found no results for " + metricName);
+    private static <T extends PointData> Optional<T> getSumPointData(SumData<T> sumData, TelemetryMetricID id) {
+        assertEquals(sumData.getAggregationTemporality(), CUMULATIVE, "Wrong temporality type for metric " + id.name);
+
+        List<T> data = sumData.getPoints().stream()
+                .filter(hasAllAttributes(id.attributes))
+                .collect(Collectors.toList());
+        assertThat("Found more than one data point for metric: " + id, data, hasSize(lessThan(2)));
+
+        return data.isEmpty() ? Optional.empty() : Optional.of(data.get(0));
+    }
+
+    /**
+     * Returns a predicate which checks whether a {@code PointData} contains all of the given {@code Attributes}.
+     * <p>
+     * Permits package access for testing
+     *
+     * @param attributes
+     *            the attributes to check for
+     * @return a predicate which returns {@code true} if {@link PointData#getAttributes()} returns a superset of
+     *         {@code attributes}, otherwise {@code false}
+     */
+    static Predicate<PointData> hasAllAttributes(Attributes attributes) {
+        return pointData -> {
+            for (Entry<AttributeKey<?>, Object> e : attributes.asMap().entrySet()) {
+                if (!pointData.getAttributes().asMap().containsKey(e.getKey())) {
+                    return false;
+                }
+                if (!Objects.equals(pointData.getAttributes().get(e.getKey()), e.getValue())) {
+                    return false;
+                }
+            }
+            return true;
+        };
+    }
+
+    private static void validateMetricType(MetricData md, TelemetryMetricID id) {
+        switch (id.type) {
+            case COUNTER :
+                assertEquals(md.getType(), MetricDataType.LONG_SUM,
+                        "Wrong type for metric " + id.name);
+                assertTrue(md.getLongSumData().isMonotonic(),
+                        "Metric is not monotonic: " + id.name);
+                break;
+            case UPDOWNCOUNTER :
+                assertEquals(md.getType(), MetricDataType.LONG_SUM,
+                        "Wrong type for metric " + id.name);
+                assertFalse(md.getLongSumData().isMonotonic(),
+                        "Metric should not be monotonic: " + id.name);
+                break;
+            case GAUGE :
+                assertEquals(md.getType(), MetricDataType.LONG_GAUGE,
+                        "Wrong type for metric " + id.name);
+                break;
+            case HISTOGRAM :
+                assertEquals(md.getType(), MetricDataType.HISTOGRAM,
+                        "Wrong type for metric " + id.name);
+                break;
+            default :
+                // Shouldn't happen because we validate this in the constructor
+                throw new IllegalStateException("Invalid metric type: " + id.type);
         }
     }
 }