From a9f922a50566c2659af55ac38feead0ac5b520a1 Mon Sep 17 00:00:00 2001 From: exceptionfactory Date: Tue, 8 Sep 2026 12:51:37 -0500 Subject: [PATCH] NIFI-16321 Added Attribute Handling for Counters and Gauges - Upgraded NiFi API from 2.11.0 to 2.12.0 - Added new adjustCounter and recordGauge methods to ProcessSession classes --- .../groovyx/flow/ProcessSessionWrap.java | 10 +++ .../controller/metrics/CounterRecord.java | 30 +++++++ .../nifi/controller/metrics/GaugeRecord.java | 30 +++++++ .../repository/AbstractRepositoryContext.java | 5 +- .../repository/RepositoryContext.java | 3 +- .../repository/StandardProcessSession.java | 90 ++++++++++++++----- .../WeakHashMapProcessSessionFactory.java | 10 +++ .../StandardProcessSessionTest.java | 66 ++++++++++++-- .../repository/BatchingSessionFactory.java | 10 +++ .../BatchingSessionFactoryTest.java | 40 ++++++++- .../apache/nifi/util/GaugeMeasurement.java | 26 ++++++ .../java/org/apache/nifi/util/MetricKey.java | 28 ++++++ .../apache/nifi/util/MockProcessSession.java | 60 +++++++------ .../apache/nifi/util/SharedSessionState.java | 90 ++++++++++++++++--- .../util/StandardProcessorTestRunner.java | 10 +++ .../java/org/apache/nifi/util/TestRunner.java | 18 ++++ .../util/TestStandardProcessorTestRunner.java | 67 ++++++++++++-- .../SystemTestComponentMetricReporter.java | 6 +- .../processors/tests/system/UpdateMetric.java | 7 +- .../metrics/ComponentMetricReporterIT.java | 9 +- pom.xml | 2 +- 21 files changed, 522 insertions(+), 95 deletions(-) create mode 100644 nifi-mock/src/main/java/org/apache/nifi/util/GaugeMeasurement.java create mode 100644 nifi-mock/src/main/java/org/apache/nifi/util/MetricKey.java diff --git a/nifi-extension-bundles/nifi-groovyx-bundle/nifi-groovyx-processors/src/main/java/org/apache/nifi/processors/groovyx/flow/ProcessSessionWrap.java b/nifi-extension-bundles/nifi-groovyx-bundle/nifi-groovyx-processors/src/main/java/org/apache/nifi/processors/groovyx/flow/ProcessSessionWrap.java index d0a67ed2d479..d86075b1a4a5 100644 --- a/nifi-extension-bundles/nifi-groovyx-bundle/nifi-groovyx-processors/src/main/java/org/apache/nifi/processors/groovyx/flow/ProcessSessionWrap.java +++ b/nifi-extension-bundles/nifi-groovyx-bundle/nifi-groovyx-processors/src/main/java/org/apache/nifi/processors/groovyx/flow/ProcessSessionWrap.java @@ -312,11 +312,21 @@ public void adjustCounter(String name, long delta, boolean immediate) { session.adjustCounter(name, delta, immediate); } + @Override + public void adjustCounter(final String name, final long delta, final Map attributes, final CommitTiming commitTiming) { + session.adjustCounter(name, delta, attributes, commitTiming); + } + @Override public void recordGauge(final String name, final double value, final CommitTiming commitTiming) { session.recordGauge(name, value, commitTiming); } + @Override + public void recordGauge(final String name, final double value, final Map attributes, final CommitTiming commitTiming) { + session.recordGauge(name, value, attributes, commitTiming); + } + /** * @return FlowFile that is next highest priority FlowFile to process. Otherwise returns null. */ diff --git a/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/CounterRecord.java b/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/CounterRecord.java index ed8a070f0714..7358817fda2e 100644 --- a/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/CounterRecord.java +++ b/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/CounterRecord.java @@ -17,19 +17,49 @@ package org.apache.nifi.controller.metrics; import java.time.Instant; +import java.util.Map; +import java.util.Objects; /** * Single measurement for a named Counter recorded during processing * * @param name Counter Name * @param value Counter Value + * @param attributes Map of keys and values associated with the Counter measurement, which may be empty but not null * @param recorded Timestamp when the Component recorded the Counter value * @param componentMetricContext Context for Component Metric record */ public record CounterRecord( String name, long value, + Map attributes, Instant recorded, ComponentMetricContext componentMetricContext ) { + public CounterRecord { + attributes = Map.copyOf(Objects.requireNonNull(attributes, "Attributes required")); + } + + /** + * Counter Record constructor for compatibility with earlier versions + * + * @param name Counter Name + * @param value Counter Value + * @param recorded Timestamp when the Processor recorded the Counter value + * @param componentMetricContext Context for Component Metric record + */ + public CounterRecord( + final String name, + final long value, + final Instant recorded, + final ComponentMetricContext componentMetricContext + ) { + this( + name, + value, + Map.of(), + recorded, + componentMetricContext + ); + } } diff --git a/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/GaugeRecord.java b/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/GaugeRecord.java index 32971f1cb12e..6d8d9de883d9 100644 --- a/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/GaugeRecord.java +++ b/nifi-framework-api/src/main/java/org/apache/nifi/controller/metrics/GaugeRecord.java @@ -17,19 +17,49 @@ package org.apache.nifi.controller.metrics; import java.time.Instant; +import java.util.Map; +import java.util.Objects; /** * Single measurement for a named Gauge recorded during processing * * @param name Gauge Name * @param value Gauge Value + * @param attributes Map of keys and values associated with the Gauge measurement, which may be empty but not null * @param recorded Timestamp when the Processor recorded the Gauge value * @param componentMetricContext Context for Component Metric record */ public record GaugeRecord( String name, double value, + Map attributes, Instant recorded, ComponentMetricContext componentMetricContext ) { + public GaugeRecord { + attributes = Map.copyOf(Objects.requireNonNull(attributes, "Attributes required")); + } + + /** + * Gauge Record constructor for compatibility with earlier versions + * + * @param name Gauge Name + * @param value Gauge Value + * @param recorded Timestamp when the Processor recorded the Gauge value + * @param componentMetricContext Context for Component Metric record + */ + public GaugeRecord( + final String name, + final double value, + final Instant recorded, + final ComponentMetricContext componentMetricContext + ) { + this( + name, + value, + Map.of(), + recorded, + componentMetricContext + ); + } } diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java index 8430a77f9575..29c8d6817e1b 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/AbstractRepositoryContext.java @@ -42,6 +42,7 @@ import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Predicate; @@ -172,11 +173,11 @@ private boolean pollFromSelfLoopsOnly() { } @Override - public void adjustCounter(final String name, final long delta) { + public void adjustCounter(final String name, final long delta, final Map attributes) { counterRepo.adjustCounter(componentNameCounterContext, name, delta); counterRepo.adjustCounter(componentTypeCounterContext, name, delta); - final CounterRecord counterRecord = new CounterRecord(name, delta, Instant.now(), componentMetricContext); + final CounterRecord counterRecord = new CounterRecord(name, delta, attributes, Instant.now(), componentMetricContext); componentMetricReporter.recordCounter(counterRecord); } diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java index 6d5ce99b0925..30cdc53caa5d 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/RepositoryContext.java @@ -34,6 +34,7 @@ import java.util.Collection; import java.util.List; +import java.util.Map; import java.util.function.Predicate; public interface RepositoryContext { @@ -69,7 +70,7 @@ public interface RepositoryContext { long getNextFlowFileSequence(); - void adjustCounter(String name, long delta); + void adjustCounter(String name, long delta, Map attributes); void recordGauge(GaugeRecord gaugeRecord); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java index 0349ccd79cc7..1070d8976671 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java @@ -165,8 +165,8 @@ public class StandardProcessSession implements ProcessSession, ProvenanceEventEn private final String connectableDescription; private final PerformanceTracker performanceTracker; - private Map countersOnCommit; - private Map immediateCounters; + private Map countersOnCommit; + private Map immediateCounters; private List gaugeRecordsSessionCommitted; private final Set removedFlowFiles = new HashSet<>(); @@ -692,8 +692,9 @@ protected void commit(final Checkpoint checkpoint, final boolean asynchronous) { } } - for (final Map.Entry entry : checkpoint.countersOnCommit.entrySet()) { - context.adjustCounter(entry.getKey(), entry.getValue()); + for (final Map.Entry entry : checkpoint.countersOnCommit.entrySet()) { + final CounterKey counterKey = entry.getKey(); + context.adjustCounter(counterKey.name(), entry.getValue(), counterKey.attributes()); } for (final GaugeRecord gaugeRecord : checkpoint.gaugeRecordsSessionCommitted) { @@ -889,23 +890,36 @@ private LoadBalanceStatus getLoadBalanceStatus(final FlowFileQueue flowFileQueue return loadBalanceStatus; } - private Map combineCounters(final Map first, final Map second) { - final boolean firstEmpty = first == null || first.isEmpty(); - final boolean secondEmpty = second == null || second.isEmpty(); + private Map combineCounters(final Map first, final Map second) { + final Map firstValues = getCounterValues(first); + final Map secondValues = getCounterValues(second); - if (firstEmpty && secondEmpty) { - return null; + if (firstValues == null) { + return secondValues; } - if (firstEmpty) { - return second; + if (secondValues == null) { + return firstValues; } - if (secondEmpty) { - return first; + + secondValues.forEach((name, value) -> firstValues.merge(name, value, Long::sum)); + return firstValues; + } + + /** + * Reduce Counter measurements to values keyed by Counter name, summing the measurements recorded for a name with + * differing attributes, since FlowFile Events track Counter values by name alone. + * + * @param counters Counter measurements which may be null or empty + * @return Counter values keyed by Counter name, or null when no measurements were recorded + */ + private Map getCounterValues(final Map counters) { + if (counters == null || counters.isEmpty()) { + return null; } - final Map combined = new HashMap<>(first); - second.forEach((key, value) -> combined.merge(key, value, Long::sum)); - return combined; + final Map counterValues = new HashMap<>(); + counters.forEach((counterKey, value) -> counterValues.merge(counterKey.name(), value, Long::sum)); + return counterValues; } private void addEventType(final Map map, final String id, final ProvenanceEventType eventType) { @@ -1410,7 +1424,7 @@ protected synchronized void rollback(final boolean penalize, final boolean rollb final ProcessSessionEvent flowFileEvent = ProcessSessionEventBuilder.forComponent(context.getComponentMetricContext()) .bytesRead(bytesRead) .bytesWritten(bytesWritten) - .counters(immediateCounters) + .counters(getCounterValues(immediateCounters)) .build(); // update event repository @@ -2009,11 +2023,17 @@ private void handleConflictingId(final FlowFileRecord flowFile, final Connection @Override public void recordGauge(final String name, final double value, final CommitTiming commitTiming) { + recordGauge(name, value, Map.of(), commitTiming); + } + + @Override + public void recordGauge(final String name, final double value, final Map attributes, final CommitTiming commitTiming) { Objects.requireNonNull(name, "Gauge Name required"); + Objects.requireNonNull(attributes, "Gauge Attributes required"); Objects.requireNonNull(commitTiming, "Commit Timing required"); final Instant recorded = Instant.now(); - final GaugeRecord gaugeRecord = new GaugeRecord(name, value, recorded, context.getComponentMetricContext()); + final GaugeRecord gaugeRecord = new GaugeRecord(name, value, Map.copyOf(attributes), recorded, context.getComponentMetricContext()); if (CommitTiming.NOW == commitTiming) { context.recordGauge(gaugeRecord); @@ -2027,6 +2047,17 @@ public void recordGauge(final String name, final double value, final CommitTimin @Override public void adjustCounter(final String name, final long delta, final boolean immediate) { + adjustCounter(name, delta, Map.of(), immediate ? CommitTiming.NOW : CommitTiming.SESSION_COMMITTED); + } + + @Override + public void adjustCounter(final String name, final long delta, final Map attributes, final CommitTiming commitTiming) { + Objects.requireNonNull(name, "Counter Name required"); + Objects.requireNonNull(attributes, "Counter Attributes required"); + Objects.requireNonNull(commitTiming, "Commit Timing required"); + + final boolean immediate = CommitTiming.NOW == commitTiming; + // If we are adjusting the counter immediately, allow it even if the task is terminated. The contract states: // "the counter will be updated immediately, without regard to whether the session is committed or rolled back" // so we need to ensure that we allow adjusting the counter even after the task is terminated. @@ -2034,7 +2065,7 @@ public void adjustCounter(final String name, final long delta, final boolean imm verifyTaskActive(); } - final Map counters; + final Map counters; if (immediate) { if (immediateCounters == null) { immediateCounters = new HashMap<>(); @@ -2047,13 +2078,17 @@ public void adjustCounter(final String name, final long delta, final boolean imm counters = countersOnCommit; } + // Measurements are aggregated for each distinct combination of Counter name and attributes + final Map counterAttributes = Map.copyOf(attributes); + final CounterKey counterKey = new CounterKey(name, counterAttributes); + // Set current value or adjust when found - counters.compute(name, (currentName, currentValue) -> + counters.compute(counterKey, (currentKey, currentValue) -> currentValue == null ? delta : currentValue + delta ); if (immediate) { - context.adjustCounter(name, delta); + context.adjustCounter(name, delta, counterAttributes); } } @@ -4071,6 +4106,15 @@ private interface ConnectionPoller { List poll(Connection connection, Set expiredRecords); } + /** + * Key for aggregating Counter measurements recorded under the same Counter name with the same attributes + * + * @param name Counter name + * @param attributes Immutable Map of keys and values associated with the Counter measurement + */ + private record CounterKey(String name, Map attributes) { + } + protected static class Checkpoint { private long processingTime = 0L; @@ -4085,8 +4129,8 @@ protected static class Checkpoint { private Map processedConnections; private Map connectionMetricContexts; - private Map countersOnCommit; - private Map immediateCounters; + private Map countersOnCommit; + private Map immediateCounters; private List gaugeRecordsSessionCommitted; diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/WeakHashMapProcessSessionFactory.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/WeakHashMapProcessSessionFactory.java index 9aadf1a9f0ab..325d1749035a 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/WeakHashMapProcessSessionFactory.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/WeakHashMapProcessSessionFactory.java @@ -145,11 +145,21 @@ public void adjustCounter(final String name, final long delta, final boolean imm delegate.adjustCounter(name, delta, immediate); } + @Override + public void adjustCounter(final String name, final long delta, final Map attributes, final CommitTiming commitTiming) { + delegate.adjustCounter(name, delta, attributes, commitTiming); + } + @Override public void recordGauge(final String name, final double value, final CommitTiming commitTiming) { delegate.recordGauge(name, value, commitTiming); } + @Override + public void recordGauge(final String name, final double value, final Map attributes, final CommitTiming commitTiming) { + delegate.recordGauge(name, value, attributes, commitTiming); + } + @Override public FlowFile get() { return delegate.get(); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java index 1a57bf49b3f2..ad04a8ecfd64 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java @@ -54,6 +54,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.List; +import java.util.Map; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; @@ -92,6 +93,12 @@ class StandardProcessSessionTest { private static final double GAUGE_VALUE = 64.5; + private static final String COUNTER_NAME = "onTrigger"; + + private static final long COUNTER_DELTA = 5; + + private static final Map METRIC_ATTRIBUTES = Map.of("service.name", "Processing", "deployment.environment", "production"); + private static final String INPUT_CONNECTION_ID = "input-connection-id"; private static final String OUTPUT_CONNECTION_ID = "output-connection-id"; private static final String BACK_PRESSURE_DATA_SIZE_THRESHOLD = "1 MB"; @@ -366,26 +373,52 @@ void testExportToOutputStreamFlowFileEventBytes() throws IOException { @Test void testRecordGaugeNow() { session.recordGauge(GAUGE_NAME, GAUGE_VALUE, CommitTiming.NOW); + session.recordGauge(GAUGE_NAME, GAUGE_VALUE, METRIC_ATTRIBUTES, CommitTiming.NOW); - verify(repositoryContext).recordGauge(gaugeRecordCaptor.capture()); - final GaugeRecord gaugeRecord = gaugeRecordCaptor.getValue(); - - assertEquals(GAUGE_NAME, gaugeRecord.name()); - assertEquals(GAUGE_VALUE, gaugeRecord.value()); + assertGaugeRecordsMatched(); } @Test void testRecordGaugeSessionCommitted() { session.recordGauge(GAUGE_NAME, GAUGE_VALUE, CommitTiming.SESSION_COMMITTED); + session.recordGauge(GAUGE_NAME, GAUGE_VALUE, METRIC_ATTRIBUTES, CommitTiming.SESSION_COMMITTED); + + verify(repositoryContext, never()).recordGauge(any()); setRepositoryContext(); session.commit(); - verify(repositoryContext).recordGauge(gaugeRecordCaptor.capture()); - final GaugeRecord gaugeRecord = gaugeRecordCaptor.getValue(); + assertGaugeRecordsMatched(); + } + + @Test + void testAdjustCounterNow() { + session.adjustCounter(COUNTER_NAME, COUNTER_DELTA, true); + verify(repositoryContext).adjustCounter(eq(COUNTER_NAME), eq(COUNTER_DELTA), eq(Map.of())); - assertEquals(GAUGE_NAME, gaugeRecord.name()); - assertEquals(GAUGE_VALUE, gaugeRecord.value()); + session.adjustCounter(COUNTER_NAME, COUNTER_DELTA, METRIC_ATTRIBUTES, CommitTiming.NOW); + verify(repositoryContext).adjustCounter(eq(COUNTER_NAME), eq(COUNTER_DELTA), eq(METRIC_ATTRIBUTES)); + } + + @Test + void testAdjustCounterSessionCommitted() throws IOException { + session.adjustCounter(COUNTER_NAME, COUNTER_DELTA, false); + session.adjustCounter(COUNTER_NAME, COUNTER_DELTA, METRIC_ATTRIBUTES, CommitTiming.SESSION_COMMITTED); + session.adjustCounter(COUNTER_NAME, COUNTER_DELTA, METRIC_ATTRIBUTES, CommitTiming.SESSION_COMMITTED); + + verify(repositoryContext, never()).adjustCounter(any(), anyLong(), any()); + + setRepositoryContext(); + session.commit(); + + // Measurements recorded for the same Counter name with differing attributes are aggregated separately + verify(repositoryContext).adjustCounter(eq(COUNTER_NAME), eq(COUNTER_DELTA), eq(Map.of())); + verify(repositoryContext).adjustCounter(eq(COUNTER_NAME), eq(COUNTER_DELTA * 2), eq(METRIC_ATTRIBUTES)); + + // FlowFile Events track Counter values by name alone, summing measurements recorded with differing attributes + verify(flowFileEventRepository).updateRepository(flowFileEventCaptor.capture()); + final ProcessSessionEvent flowFileEvent = flowFileEventCaptor.getValue(); + assertEquals(Map.of(COUNTER_NAME, COUNTER_DELTA * 3), flowFileEvent.getCounters()); } @Test @@ -408,6 +441,21 @@ void testCreateLineage() { assertEquals(secondFlowFileId, secondFlowFile.getLineageStartIndex()); } + private void assertGaugeRecordsMatched() { + verify(repositoryContext, times(2)).recordGauge(gaugeRecordCaptor.capture()); + final List gaugeRecords = gaugeRecordCaptor.getAllValues(); + + final GaugeRecord firstGaugeRecord = gaugeRecords.getFirst(); + assertEquals(GAUGE_NAME, firstGaugeRecord.name()); + assertEquals(GAUGE_VALUE, firstGaugeRecord.value()); + assertEquals(Map.of(), firstGaugeRecord.attributes()); + + final GaugeRecord secondGaugeRecord = gaugeRecords.getLast(); + assertEquals(GAUGE_NAME, secondGaugeRecord.name()); + assertEquals(GAUGE_VALUE, secondGaugeRecord.value()); + assertEquals(METRIC_ATTRIBUTES, secondGaugeRecord.attributes()); + } + private void assertFlowFileEventMatched(final long bytesRead, final long bytesWritten) throws IOException { verify(flowFileEventRepository).updateRepository(flowFileEventCaptor.capture()); final ProcessSessionEvent flowFileEvent = flowFileEventCaptor.getValue(); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/BatchingSessionFactory.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/BatchingSessionFactory.java index 11ce752ea5b2..b3a373d85d32 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/BatchingSessionFactory.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/BatchingSessionFactory.java @@ -116,11 +116,21 @@ public void adjustCounter(String name, long delta, boolean immediate) { session.adjustCounter(name, delta, immediate); } + @Override + public void adjustCounter(final String name, final long delta, final Map attributes, final CommitTiming commitTiming) { + session.adjustCounter(name, delta, attributes, commitTiming); + } + @Override public void recordGauge(final String name, final double value, final CommitTiming commitTiming) { session.recordGauge(name, value, commitTiming); } + @Override + public void recordGauge(final String name, final double value, final Map attributes, final CommitTiming commitTiming) { + session.recordGauge(name, value, attributes, commitTiming); + } + @Override public FlowFile get() { return session.get(); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/BatchingSessionFactoryTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/BatchingSessionFactoryTest.java index eced182de10b..99b0a9e88547 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/BatchingSessionFactoryTest.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/BatchingSessionFactoryTest.java @@ -30,8 +30,13 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import java.util.List; +import java.util.Map; + import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -43,6 +48,12 @@ class BatchingSessionFactoryTest { private static final double GAUGE_VALUE = 64.5; + private static final String COUNTER_NAME = "recorded"; + + private static final long COUNTER_DELTA = 5; + + private static final Map METRIC_ATTRIBUTES = Map.of("service.name", "Processing", "deployment.environment", "production"); + private final TaskTermination taskTermination = () -> false; @Mock @@ -76,10 +87,31 @@ void testCreateSessionRecordGauge() { assertNotNull(session); session.recordGauge(GAUGE_NAME, GAUGE_VALUE, CommitTiming.NOW); + session.recordGauge(GAUGE_NAME, GAUGE_VALUE, METRIC_ATTRIBUTES, CommitTiming.NOW); + + verify(repositoryContext, times(2)).recordGauge(gaugeRecordCaptor.capture()); + final List gaugeRecords = gaugeRecordCaptor.getAllValues(); + + final GaugeRecord firstGaugeRecord = gaugeRecords.getFirst(); + assertEquals(GAUGE_NAME, firstGaugeRecord.name()); + assertEquals(GAUGE_VALUE, firstGaugeRecord.value()); + assertEquals(Map.of(), firstGaugeRecord.attributes()); + + final GaugeRecord secondGaugeRecord = gaugeRecords.getLast(); + assertEquals(GAUGE_NAME, secondGaugeRecord.name()); + assertEquals(GAUGE_VALUE, secondGaugeRecord.value()); + assertEquals(METRIC_ATTRIBUTES, secondGaugeRecord.attributes()); + } + + @Test + void testCreateSessionAdjustCounter() { + final ProcessSession session = factory.createSession(); + assertNotNull(session); + + session.adjustCounter(COUNTER_NAME, COUNTER_DELTA, true); + verify(repositoryContext).adjustCounter(eq(COUNTER_NAME), eq(COUNTER_DELTA), eq(Map.of())); - verify(repositoryContext).recordGauge(gaugeRecordCaptor.capture()); - final GaugeRecord gaugeRecord = gaugeRecordCaptor.getValue(); - assertEquals(GAUGE_NAME, gaugeRecord.name()); - assertEquals(GAUGE_VALUE, gaugeRecord.value()); + session.adjustCounter(COUNTER_NAME, COUNTER_DELTA, METRIC_ATTRIBUTES, CommitTiming.NOW); + verify(repositoryContext).adjustCounter(eq(COUNTER_NAME), eq(COUNTER_DELTA), eq(METRIC_ATTRIBUTES)); } } diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/GaugeMeasurement.java b/nifi-mock/src/main/java/org/apache/nifi/util/GaugeMeasurement.java new file mode 100644 index 000000000000..0faa7d0b66e8 --- /dev/null +++ b/nifi-mock/src/main/java/org/apache/nifi/util/GaugeMeasurement.java @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.util; + +/** + * Gauge measurement associated with a metric key + * + * @param key Metric key containing the Gauge name and attributes + * @param value Gauge value + */ +record GaugeMeasurement(MetricKey key, double value) { +} diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MetricKey.java b/nifi-mock/src/main/java/org/apache/nifi/util/MetricKey.java new file mode 100644 index 000000000000..1e9587ec5f8b --- /dev/null +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MetricKey.java @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.util; + +import java.util.Map; + +/** + * Key for tracking Counter and Gauge measurements recorded under the same name with the same attributes + * + * @param name Counter or Gauge name + * @param attributes Immutable Map of keys and values associated with the measurement + */ +record MetricKey(String name, Map attributes) { +} diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessSession.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessSession.java index ee2b10a6f651..13b5d20e6881 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessSession.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessSession.java @@ -77,8 +77,8 @@ public class MockProcessSession implements ProcessSession { private final Map currentVersions = new HashMap<>(); private final Map originalVersions = new HashMap<>(); private final SharedSessionState sharedState; - private final Map counterMap = new HashMap<>(); - private final Map> namedGaugeValues = new HashMap<>(); + private final Map counterMap = new HashMap<>(); + private final List gaugeMeasurementsSessionCommitted = new ArrayList<>(); private final Map readRecursionSet = new HashMap<>(); private final Set writeRecursionSet = new HashSet<>(); private final MockProvenanceReporter provenanceReporter; @@ -140,32 +140,41 @@ public static Builder builder(final SharedSessionState sharedState, final Proces @Override public void adjustCounter(final String name, final long delta, final boolean immediate) { - if (immediate) { - sharedState.adjustCounter(name, delta); - return; - } + adjustCounter(name, delta, Map.of(), immediate ? CommitTiming.NOW : CommitTiming.SESSION_COMMITTED); + } - Long counter = counterMap.get(name); - if (counter == null) { - counter = delta; - counterMap.put(name, counter); - return; - } + @Override + public void adjustCounter(final String name, final long delta, final Map attributes, final CommitTiming commitTiming) { + Objects.requireNonNull(name, "Counter Name required"); + Objects.requireNonNull(attributes, "Counter Attributes required"); + Objects.requireNonNull(commitTiming, "Commit Timing required"); - counter = counter + delta; - counterMap.put(name, counter); + final MetricKey counterKey = new MetricKey(name, Map.copyOf(attributes)); + + if (CommitTiming.NOW == commitTiming) { + sharedState.adjustCounter(counterKey, delta); + } else { + counterMap.merge(counterKey, delta, Long::sum); + } } @Override public void recordGauge(final String name, final double value, final CommitTiming commitTiming) { + recordGauge(name, value, Map.of(), commitTiming); + } + + @Override + public void recordGauge(final String name, final double value, final Map attributes, final CommitTiming commitTiming) { + Objects.requireNonNull(name, "Gauge Name required"); + Objects.requireNonNull(attributes, "Gauge Attributes required"); + Objects.requireNonNull(commitTiming, "Commit Timing required"); + + final MetricKey gaugeKey = new MetricKey(name, Map.copyOf(attributes)); + if (CommitTiming.NOW == commitTiming) { - sharedState.recordGauge(name, value); + sharedState.recordGauge(gaugeKey, value); } else { - namedGaugeValues.compute(name, (gaugeName, values) -> { - final List gaugeValues = Objects.requireNonNullElseGet(values, ArrayList::new); - gaugeValues.add(value); - return gaugeValues; - }); + gaugeMeasurementsSessionCommitted.add(new GaugeMeasurement(gaugeKey, value)); } } @@ -337,21 +346,18 @@ private void commitInternal() { originalVersions.clear(); created.clear(); - for (final Map.Entry entry : counterMap.entrySet()) { + for (final Map.Entry entry : counterMap.entrySet()) { sharedState.adjustCounter(entry.getKey(), entry.getValue()); } - for (final Map.Entry> namedGaugeEntry : namedGaugeValues.entrySet()) { - final String name = namedGaugeEntry.getKey(); - final List gaugeValues = namedGaugeEntry.getValue(); - for (final Double gaugeValue : gaugeValues) { - sharedState.recordGauge(name, gaugeValue); - } + for (final GaugeMeasurement gaugeMeasurement : gaugeMeasurementsSessionCommitted) { + sharedState.recordGauge(gaugeMeasurement.key(), gaugeMeasurement.value()); } sharedState.addProvenanceEvents(provenanceReporter.getEvents()); provenanceReporter.clear(); counterMap.clear(); + gaugeMeasurementsSessionCommitted.clear(); } @Override diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/SharedSessionState.java b/nifi-mock/src/main/java/org/apache/nifi/util/SharedSessionState.java index 1b65aa8b02f8..14d60bc3c42d 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/SharedSessionState.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/SharedSessionState.java @@ -23,9 +23,11 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; -import java.util.Objects; +import java.util.Map; +import java.util.Queue; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicLong; @@ -36,8 +38,8 @@ public class SharedSessionState { @SuppressWarnings("unused") private final Processor processor; private final AtomicLong flowFileIdGenerator; - private final ConcurrentMap counterMap = new ConcurrentHashMap<>(); - private final ConcurrentMap> namedGaugeValues = new ConcurrentHashMap<>(); + private final ConcurrentMap counterMap = new ConcurrentHashMap<>(); + private final Queue gaugeMeasurements = new ConcurrentLinkedQueue<>(); // list of provenance events as they were in the provenance repository (events emitted with force=true or committed with the session) private final List events = new ArrayList<>(); @@ -73,10 +75,14 @@ public long nextFlowFileId() { } public void adjustCounter(final String name, final long delta) { - AtomicLong counter = counterMap.get(name); + adjustCounter(new MetricKey(name, Map.of()), delta); + } + + void adjustCounter(final MetricKey counterKey, final long delta) { + AtomicLong counter = counterMap.get(counterKey); if (counter == null) { counter = new AtomicLong(0L); - final AtomicLong existingCounter = counterMap.putIfAbsent(name, counter); + final AtomicLong existingCounter = counterMap.putIfAbsent(counterKey, counter); if (existingCounter != null) { counter = existingCounter; } @@ -85,20 +91,80 @@ public void adjustCounter(final String name, final long delta) { counter.addAndGet(delta); } + /** + * Get the value recorded for the named Counter, summing the measurements recorded with differing attributes + * + * @param name Counter Name + * @return Counter value, or null when the named Counter was not used + */ public Long getCounterValue(final String name) { - final AtomicLong counterValue = counterMap.get(name); + Long counterValue = null; + + for (final Map.Entry counterEntry : counterMap.entrySet()) { + if (counterEntry.getKey().name().equals(name)) { + final long recorded = counterEntry.getValue().get(); + counterValue = counterValue == null ? recorded : counterValue + recorded; + } + } + + return counterValue; + } + + /** + * Get the value recorded for the named Counter with the specified attributes + * + * @param name Counter Name + * @param attributes Map of keys and values associated with the Counter + * @return Counter value, or null when the named Counter was not used with the specified attributes + */ + public Long getCounterValue(final String name, final Map attributes) { + final AtomicLong counterValue = counterMap.get(new MetricKey(name, Map.copyOf(attributes))); return counterValue == null ? null : counterValue.get(); } public void recordGauge(final String name, final double value) { - namedGaugeValues.compute(name, (gaugeName, values) -> { - final List gaugeValues = Objects.requireNonNullElseGet(values, ArrayList::new); - gaugeValues.add(value); - return gaugeValues; - }); + recordGauge(new MetricKey(name, Map.of()), value); + } + + void recordGauge(final MetricKey gaugeKey, final double value) { + gaugeMeasurements.add(new GaugeMeasurement(gaugeKey, value)); } + /** + * Get list of values recorded for the named Gauge, including the measurements recorded with differing attributes + * + * @param name Gauge Name + * @return List of recorded values, or empty when the named Gauge was not used + */ public List getGaugeValues(final String name) { - return namedGaugeValues.getOrDefault(name, List.of()); + final List gaugeValues = new ArrayList<>(); + + for (final GaugeMeasurement gaugeMeasurement : gaugeMeasurements) { + if (gaugeMeasurement.key().name().equals(name)) { + gaugeValues.add(gaugeMeasurement.value()); + } + } + + return gaugeValues; + } + + /** + * Get list of values recorded for the named Gauge with the specified attributes + * + * @param name Gauge Name + * @param attributes Map of keys and values associated with the Gauge + * @return List of recorded values, or empty when the named Gauge was not used with the specified attributes + */ + public List getGaugeValues(final String name, final Map attributes) { + final MetricKey gaugeKey = new MetricKey(name, Map.copyOf(attributes)); + final List gaugeValues = new ArrayList<>(); + + for (final GaugeMeasurement gaugeMeasurement : gaugeMeasurements) { + if (gaugeMeasurement.key().equals(gaugeKey)) { + gaugeValues.add(gaugeMeasurement.value()); + } + } + + return List.copyOf(gaugeValues); } } diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java index 94491390a652..21440449354b 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java @@ -578,11 +578,21 @@ public Long getCounterValue(final String name) { return sharedState.getCounterValue(name); } + @Override + public Long getCounterValue(final String name, final Map attributes) { + return sharedState.getCounterValue(name, attributes); + } + @Override public List getGaugeValues(final String name) { return sharedState.getGaugeValues(name); } + @Override + public List getGaugeValues(final String name, final Map attributes) { + return sharedState.getGaugeValues(name, attributes); + } + @Override public int getRemovedCount() { int count = 0; diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java index 5259eb741101..f43e7f9321d2 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunner.java @@ -562,6 +562,15 @@ void assertAttributes( */ Long getCounterValue(String name); + /** + * Get the value recorded for the named Counter with the specified attributes + * + * @param name Counter Name + * @param attributes Map of keys and values associated with the Counter + * @return Counter value, or null when the named Counter was not used with the specified attributes + */ + Long getCounterValue(String name, Map attributes); + /** * Get list of values recorded for the named Gauge * @@ -570,6 +579,15 @@ void assertAttributes( */ List getGaugeValues(String name); + /** + * Get list of values recorded for the named Gauge with the specified attributes + * + * @param name Gauge Name + * @param attributes Map of keys and values associated with the Gauge + * @return List of recorded values, or empty when the named Gauge was not used with the specified attributes + */ + List getGaugeValues(String name, Map attributes); + /** * @return the number of FlowFiles that have been removed from the system */ diff --git a/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java b/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java index 0d1e48914183..570a9f7dd2e8 100644 --- a/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java +++ b/nifi-mock/src/test/java/org/apache/nifi/util/TestStandardProcessorTestRunner.java @@ -42,6 +42,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -49,7 +50,15 @@ public class TestStandardProcessorTestRunner { private static final String NAMED_GAUGE = "Processing Time"; - private static final double NAMED_GAUGE_VALUE = 120.35; + private static final double FIRST_NAMED_GAUGE_VALUE = 120.35; + private static final double SECOND_NAMED_GAUGE_VALUE = 121.35; + private static final double THIRD_NAMED_GAUGE_VALUE = 122.35; + private static final List NAMED_GAUGE_VALUES = List.of(FIRST_NAMED_GAUGE_VALUE, SECOND_NAMED_GAUGE_VALUE, THIRD_NAMED_GAUGE_VALUE); + private static final List ATTRIBUTED_NAMED_GAUGE_VALUES = List.of(FIRST_NAMED_GAUGE_VALUE, THIRD_NAMED_GAUGE_VALUE); + private static final String NAMED_COUNTER = "Processed Records"; + private static final long NAMED_COUNTER_DELTA = 5; + private static final Map METRIC_ATTRIBUTES = Map.of("service.name", "Processing", "deployment.environment", "production"); + private static final Map UNRECORDED_METRIC_ATTRIBUTES = Map.of("service.name", "Unrecorded"); @Test public void testProcessContextPassedToOnStoppedMethods() { @@ -110,16 +119,45 @@ public void testAllConditionsMetComplex() { } @Test - public void testRecordGauge() { - final RecordGaugeProcessor processor = new RecordGaugeProcessor(); + public void testRecordGaugeNow() { + final RecordMetricProcessor processor = new RecordMetricProcessor(CommitTiming.NOW); final TestRunner runner = TestRunners.newTestRunner(processor); runner.run(); - final List gaugeValues = runner.getGaugeValues(NAMED_GAUGE); - assertFalse(gaugeValues.isEmpty()); - final Double firstValue = gaugeValues.getFirst(); - assertEquals(NAMED_GAUGE_VALUE, firstValue); + assertGaugeValues(runner); + } + + @Test + public void testRecordGaugeSessionCommitted() { + final RecordMetricProcessor processor = new RecordMetricProcessor(CommitTiming.SESSION_COMMITTED); + final TestRunner runner = TestRunners.newTestRunner(processor); + + runner.run(); + + assertGaugeValues(runner); + } + + private void assertGaugeValues(final TestRunner runner) { + assertEquals(NAMED_GAUGE_VALUES, runner.getGaugeValues(NAMED_GAUGE)); + assertEquals(ATTRIBUTED_NAMED_GAUGE_VALUES, runner.getGaugeValues(NAMED_GAUGE, METRIC_ATTRIBUTES)); + assertEquals(List.of(SECOND_NAMED_GAUGE_VALUE), runner.getGaugeValues(NAMED_GAUGE, Map.of())); + assertTrue(runner.getGaugeValues(NAMED_GAUGE, UNRECORDED_METRIC_ATTRIBUTES).isEmpty()); + } + + @Test + public void testAdjustCounter() { + final RecordMetricProcessor processor = new RecordMetricProcessor(CommitTiming.NOW); + final TestRunner runner = TestRunners.newTestRunner(processor); + + runner.run(); + + // Measurements recorded without attributes and with attributes are both summed for the Counter name + assertEquals(NAMED_COUNTER_DELTA * 2, runner.getCounterValue(NAMED_COUNTER)); + + assertEquals(NAMED_COUNTER_DELTA, runner.getCounterValue(NAMED_COUNTER, METRIC_ATTRIBUTES)); + assertEquals(NAMED_COUNTER_DELTA, runner.getCounterValue(NAMED_COUNTER, Map.of())); + assertNull(runner.getCounterValue(NAMED_COUNTER, UNRECORDED_METRIC_ATTRIBUTES)); } @Test @@ -233,10 +271,21 @@ public void testProcessorInvalidWhenControllerServiceDisabled() { runner.assertValid(); } - private static class RecordGaugeProcessor extends AbstractProcessor { + private static class RecordMetricProcessor extends AbstractProcessor { + private final CommitTiming gaugeCommitTiming; + + private RecordMetricProcessor(final CommitTiming gaugeCommitTiming) { + this.gaugeCommitTiming = gaugeCommitTiming; + } + @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { - session.recordGauge(NAMED_GAUGE, NAMED_GAUGE_VALUE, CommitTiming.NOW); + session.recordGauge(NAMED_GAUGE, FIRST_NAMED_GAUGE_VALUE, METRIC_ATTRIBUTES, gaugeCommitTiming); + session.recordGauge(NAMED_GAUGE, SECOND_NAMED_GAUGE_VALUE, Map.of(), gaugeCommitTiming); + session.recordGauge(NAMED_GAUGE, THIRD_NAMED_GAUGE_VALUE, METRIC_ATTRIBUTES, gaugeCommitTiming); + + session.adjustCounter(NAMED_COUNTER, NAMED_COUNTER_DELTA, true); + session.adjustCounter(NAMED_COUNTER, NAMED_COUNTER_DELTA, METRIC_ATTRIBUTES, CommitTiming.NOW); } } diff --git a/nifi-system-tests/nifi-system-test-component-metric-reporter-bundle/nifi-system-test-component-metric-reporter/src/main/java/org/apache/nifi/controller/metrics/SystemTestComponentMetricReporter.java b/nifi-system-tests/nifi-system-test-component-metric-reporter-bundle/nifi-system-test-component-metric-reporter/src/main/java/org/apache/nifi/controller/metrics/SystemTestComponentMetricReporter.java index a0cd473c168a..68ac500bc01e 100644 --- a/nifi-system-tests/nifi-system-test-component-metric-reporter-bundle/nifi-system-test-component-metric-reporter/src/main/java/org/apache/nifi/controller/metrics/SystemTestComponentMetricReporter.java +++ b/nifi-system-tests/nifi-system-test-component-metric-reporter-bundle/nifi-system-test-component-metric-reporter/src/main/java/org/apache/nifi/controller/metrics/SystemTestComponentMetricReporter.java @@ -41,7 +41,8 @@ public void recordGauge(final GaugeRecord gaugeRecord) { logger.info("Recording Gauge [{}] Value [{}]", gaugeRecord.name(), gaugeRecord.value()); final ComponentMetricContext componentMetricContext = gaugeRecord.componentMetricContext(); - final String formatted = "Gauge [%s] Value [%s] Component ID [%s]".formatted(gaugeRecord.name(), gaugeRecord.value(), componentMetricContext.id()); + final String id = componentMetricContext.id(); + final String formatted = "Gauge [%s] Value [%s] Attributes %s Component ID [%s]".formatted(gaugeRecord.name(), gaugeRecord.value(), gaugeRecord.attributes(), id); final String filename = "%s.GaugeRecord.%d.log".formatted(LOG_FILE_PREFIX, System.nanoTime()); final Path log = USER_DIRECTORY.resolve(filename); @@ -57,7 +58,8 @@ public void recordCounter(final CounterRecord counterRecord) { logger.info("Recording Counter [{}] Value [{}]", counterRecord.name(), counterRecord.value()); final ComponentMetricContext componentMetricContext = counterRecord.componentMetricContext(); - final String formatted = "Counter [%s] Value [%s] Component ID [%s]".formatted(counterRecord.name(), counterRecord.value(), componentMetricContext.id()); + final String id = componentMetricContext.id(); + final String formatted = "Counter [%s] Value [%s] Attributes %s Component ID [%s]".formatted(counterRecord.name(), counterRecord.value(), counterRecord.attributes(), id); final String filename = "%s.CounterRecord.%d.log".formatted(LOG_FILE_PREFIX, System.nanoTime()); final Path log = USER_DIRECTORY.resolve(filename); diff --git a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/UpdateMetric.java b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/UpdateMetric.java index 93201461a997..82138bd55150 100644 --- a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/UpdateMetric.java +++ b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/UpdateMetric.java @@ -24,6 +24,7 @@ import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.metrics.CommitTiming; +import java.util.Map; import java.util.Set; public class UpdateMetric extends AbstractProcessor { @@ -35,6 +36,8 @@ public class UpdateMetric extends AbstractProcessor { private static final Set RELATIONSHIPS = Set.of(SUCCESS); + private static final Map METRIC_ATTRIBUTES = Map.of("service.name", "UpdateMetric"); + @Override public Set getRelationships() { return RELATIONSHIPS; @@ -42,11 +45,11 @@ public Set getRelationships() { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { - session.adjustCounter("onTrigger", 1, true); + session.adjustCounter("onTrigger", 1, METRIC_ATTRIBUTES, CommitTiming.NOW); final Runtime runtime = Runtime.getRuntime(); final long freeMemory = runtime.freeMemory(); - session.recordGauge("freeMemory", freeMemory, CommitTiming.NOW); + session.recordGauge("freeMemory", freeMemory, METRIC_ATTRIBUTES, CommitTiming.NOW); FlowFile flowFile = session.get(); if (flowFile == null) { diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/metrics/ComponentMetricReporterIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/metrics/ComponentMetricReporterIT.java index ca06c2f6e4cd..3c44861ffc33 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/metrics/ComponentMetricReporterIT.java +++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/metrics/ComponentMetricReporterIT.java @@ -40,6 +40,8 @@ public class ComponentMetricReporterIT extends NiFiSystemIT { private static final String COUNTER_RECORD_LOG = "%s.CounterRecord".formatted(REPORTER_CLASS); + private static final String METRIC_ATTRIBUTE = "service.name=UpdateMetric"; + @Override protected Map getNifiPropertiesOverrides() { return Map.of(REPORTER_IMPLEMENTATION, REPORTER_CLASS); @@ -52,8 +54,8 @@ void testUpdateMetricReported() throws NiFiClientException, IOException, Interru getClientUtil().waitForStoppedProcessor(updateMetric.getId()); final String componentId = updateMetric.getId(); - assertLogComponentIdFound(GAUGE_RECORD_LOG, componentId); - assertLogComponentIdFound(COUNTER_RECORD_LOG, componentId); + assertLogRecordFound(GAUGE_RECORD_LOG, componentId); + assertLogRecordFound(COUNTER_RECORD_LOG, componentId); } private Optional findReportedLog(final String fileNameSearch) throws IOException { @@ -63,7 +65,7 @@ private Optional findReportedLog(final String fileNameSearch) throws IOExc } } - private void assertLogComponentIdFound(final String fileNameSearch, final String componentId) throws IOException { + private void assertLogRecordFound(final String fileNameSearch, final String componentId) throws IOException { final Optional reportedLogFound = findReportedLog(fileNameSearch); assertTrue(reportedLogFound.isPresent(), "Component Metric Reporter [%s] log not found".formatted(fileNameSearch)); @@ -71,5 +73,6 @@ private void assertLogComponentIdFound(final String fileNameSearch, final String final String log = Files.readString(reportedLog); assertTrue(log.contains(componentId), "Update Metric ID [%s] not found in log [%s]".formatted(componentId, log)); + assertTrue(log.contains(METRIC_ATTRIBUTE), "Update Metric Attribute [%s] not found in log [%s]".formatted(METRIC_ATTRIBUTE, log)); } } diff --git a/pom.xml b/pom.xml index e9def3588231..90a1c22ee343 100644 --- a/pom.xml +++ b/pom.xml @@ -118,7 +118,7 @@ v24.14.1 - 2.11.0 + 2.12.0 2.4.0