Repository: nifi Updated Branches: refs/heads/master 410176ed2 -> 7bbb5a823
http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java index f939273..c8f5126 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java @@ -17,91 +17,92 @@ package org.apache.nifi.controller.status.history; -import java.util.List; -import java.util.concurrent.TimeUnit; - import org.apache.nifi.controller.status.ProcessorStatus; import org.apache.nifi.controller.status.history.MetricDescriptor.Formatter; +import java.util.List; +import java.util.concurrent.TimeUnit; + public enum ProcessorStatusDescriptor { - BYTES_READ(new StandardMetricDescriptor<ProcessorStatus>( + BYTES_READ( "bytesRead", "Bytes Read (5 mins)", "The total number of bytes read from the Content Repository by this Processor in the past 5 minutes", Formatter.DATA_SIZE, - s -> s.getBytesRead())), + ProcessorStatus::getBytesRead), - BYTES_WRITTEN(new StandardMetricDescriptor<ProcessorStatus>( + BYTES_WRITTEN( "bytesWritten", "Bytes Written (5 mins)", "The total number of bytes written to the Content Repository by this Processor in the past 5 minutes", Formatter.DATA_SIZE, - s -> s.getBytesWritten())), + ProcessorStatus::getBytesWritten), - BYTES_TRANSFERRED(new StandardMetricDescriptor<ProcessorStatus>( + BYTES_TRANSFERRED( "bytesTransferred", "Bytes Transferred (5 mins)", "The total number of bytes read from or written to the Content Repository by this Processor in the past 5 minutes", Formatter.DATA_SIZE, - s -> s.getBytesRead() + s.getBytesWritten())), + s -> s.getBytesRead() + s.getBytesWritten()), - INPUT_BYTES(new StandardMetricDescriptor<ProcessorStatus>( + INPUT_BYTES( "inputBytes", "Bytes In (5 mins)", "The cumulative size of all FlowFiles that this Processor has pulled from its queues in the past 5 minutes", Formatter.DATA_SIZE, - s -> s.getInputBytes())), + ProcessorStatus::getInputBytes), - INPUT_COUNT(new StandardMetricDescriptor<ProcessorStatus>( + INPUT_COUNT( "inputCount", "FlowFiles In (5 mins)", "The number of FlowFiles that this Processor has pulled from its queues in the past 5 minutes", Formatter.COUNT, - s -> Long.valueOf(s.getInputCount()))), + s -> Long.valueOf(s.getInputCount())), - OUTPUT_BYTES(new StandardMetricDescriptor<ProcessorStatus>( + OUTPUT_BYTES( "outputBytes", "Bytes Out (5 mins)", "The cumulative size of all FlowFiles that this Processor has transferred to downstream queues in the past 5 minutes", Formatter.DATA_SIZE, - s -> s.getOutputBytes())), + ProcessorStatus::getOutputBytes), - OUTPUT_COUNT(new StandardMetricDescriptor<ProcessorStatus>( + OUTPUT_COUNT( "outputCount", "FlowFiles Out (5 mins)", "The number of FlowFiles that this Processor has transferred to downstream queues in the past 5 minutes", Formatter.COUNT, - s -> Long.valueOf(s.getOutputCount()))), + s -> Long.valueOf(s.getOutputCount())), - TASK_COUNT(new StandardMetricDescriptor<ProcessorStatus>( + TASK_COUNT( "taskCount", "Tasks (5 mins)", "The number of tasks that this Processor has completed in the past 5 minutes", Formatter.COUNT, - s -> Long.valueOf(s.getInvocations()))), + s -> Long.valueOf(s.getInvocations())), - TASK_MILLIS(new StandardMetricDescriptor<ProcessorStatus>( + TASK_MILLIS( "taskMillis", "Total Task Duration (5 mins)", "The total number of thread-milliseconds that the Processor has used to complete its tasks in the past 5 minutes", Formatter.DURATION, - s -> TimeUnit.MILLISECONDS.convert(s.getProcessingNanos(), TimeUnit.NANOSECONDS))), + s -> TimeUnit.MILLISECONDS.convert(s.getProcessingNanos(), TimeUnit.NANOSECONDS)), - TASK_NANOS(new StandardMetricDescriptor<ProcessorStatus>( + TASK_NANOS( "taskNanos", "Total Task Time (nanos)", "The total number of thread-nanoseconds that the Processor has used to complete its tasks in the past 5 minutes", Formatter.COUNT, - ProcessorStatus::getProcessingNanos), false), + ProcessorStatus::getProcessingNanos, + false), - FLOWFILES_REMOVED(new StandardMetricDescriptor<ProcessorStatus>( + FLOWFILES_REMOVED( "flowFilesRemoved", "FlowFiles Removed (5 mins)", "The total number of FlowFiles removed by this Processor in the last 5 minutes", Formatter.COUNT, - s -> Long.valueOf(s.getFlowFilesRemoved()))), + s -> Long.valueOf(s.getFlowFilesRemoved())), - AVERAGE_LINEAGE_DURATION(new StandardMetricDescriptor<ProcessorStatus>( + AVERAGE_LINEAGE_DURATION( "averageLineageDuration", "Average Lineage Duration (5 mins)", "The average amount of time that a FlowFile took to process (from receipt until this Processor finished processing it) in the past 5 minutes.", @@ -114,23 +115,24 @@ public enum ProcessorStatusDescriptor { int count = 0; for (final StatusSnapshot snapshot : values) { - final long removed = snapshot.getStatusMetrics().get(FLOWFILES_REMOVED.getDescriptor()).longValue(); - final long outputCount = snapshot.getStatusMetrics().get(OUTPUT_COUNT.getDescriptor()).longValue(); + final long removed = snapshot.getStatusMetric(FLOWFILES_REMOVED.getDescriptor()).longValue(); + final long outputCount = snapshot.getStatusMetric(OUTPUT_COUNT.getDescriptor()).longValue(); final long processed = removed + outputCount; count += processed; - final long avgMillis = snapshot.getStatusMetrics().get(AVERAGE_LINEAGE_DURATION.getDescriptor()).longValue(); + final long avgMillis = snapshot.getStatusMetric(AVERAGE_LINEAGE_DURATION.getDescriptor()).longValue(); final long totalMillis = avgMillis * processed; millis += totalMillis; } return count == 0 ? 0 : millis / count; } - } - )), + }, + true + ), - AVERAGE_TASK_NANOS(new StandardMetricDescriptor<ProcessorStatus>( + AVERAGE_TASK_NANOS( "averageTaskNanos", "Average Task Duration (nanoseconds)", "The average number of nanoseconds it took this Processor to complete a task, over the past 5 minutes", @@ -143,12 +145,12 @@ public enum ProcessorStatusDescriptor { int invocations = 0; for (final StatusSnapshot snapshot : values) { - final Long taskNanos = snapshot.getStatusMetrics().get(TASK_NANOS.getDescriptor()); + final Long taskNanos = snapshot.getStatusMetric(TASK_NANOS.getDescriptor()); if (taskNanos != null) { procNanos += taskNanos.longValue(); } - final Long taskInvocations = snapshot.getStatusMetrics().get(TASK_COUNT.getDescriptor()); + final Long taskInvocations = snapshot.getStatusMetric(TASK_COUNT.getDescriptor()); if (taskInvocations != null) { invocations += taskInvocations.intValue(); } @@ -160,22 +162,38 @@ public enum ProcessorStatusDescriptor { return procNanos / invocations; } - })); + }, + true + ); private final MetricDescriptor<ProcessorStatus> descriptor; private final boolean visible; - private ProcessorStatusDescriptor(final MetricDescriptor<ProcessorStatus> descriptor) { - this(descriptor, true); + ProcessorStatusDescriptor(final String field, final String label, final String description, + final MetricDescriptor.Formatter formatter, final ValueMapper<ProcessorStatus> valueFunction) { + + this(field, label, description, formatter, valueFunction, true); } - private ProcessorStatusDescriptor(final MetricDescriptor<ProcessorStatus> descriptor, final boolean visible) { - this.descriptor = descriptor; + ProcessorStatusDescriptor(final String field, final String label, final String description, + final MetricDescriptor.Formatter formatter, final ValueMapper<ProcessorStatus> valueFunction, final boolean visible) { + + this.descriptor = new StandardMetricDescriptor<>(this::ordinal, field, label, description, formatter, valueFunction); this.visible = visible; } + ProcessorStatusDescriptor(final String field, final String label, final String description, + final MetricDescriptor.Formatter formatter, final ValueMapper<ProcessorStatus> valueFunction, + final ValueReducer<StatusSnapshot, Long> reducer, final boolean visible) { + + this.descriptor = new StandardMetricDescriptor<>(this::ordinal, field, label, description, formatter, valueFunction, reducer); + this.visible = visible; + } + + + public String getField() { return descriptor.getField(); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java index 5875249..6b131d2 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java @@ -24,43 +24,49 @@ import java.util.List; import java.util.concurrent.TimeUnit; public enum RemoteProcessGroupStatusDescriptor { - SENT_BYTES(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentBytes", + SENT_BYTES( + "sentBytes", "Bytes Sent (5 mins)", "The cumulative size of all FlowFiles that have been successfully sent to the remote system in the past 5 minutes", Formatter.DATA_SIZE, - s -> s.getSentContentSize())), + RemoteProcessGroupStatus::getSentContentSize), - SENT_COUNT(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentCount", + SENT_COUNT( + "sentCount", "FlowFiles Sent (5 mins)", "The number of FlowFiles that have been successfully sent to the remote system in the past 5 minutes", Formatter.COUNT, - s -> s.getSentCount().longValue())), + s -> s.getSentCount().longValue()), - RECEIVED_BYTES(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedBytes", + RECEIVED_BYTES( + "receivedBytes", "Bytes Received (5 mins)", "The cumulative size of all FlowFiles that have been received from the remote system in the past 5 minutes", Formatter.DATA_SIZE, - s -> s.getReceivedContentSize())), + RemoteProcessGroupStatus::getReceivedContentSize), - RECEIVED_COUNT(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedCount", + RECEIVED_COUNT( + "receivedCount", "FlowFiles Received (5 mins)", "The number of FlowFiles that have been received from the remote system in the past 5 minutes", Formatter.COUNT, - s -> s.getReceivedCount().longValue())), + s -> s.getReceivedCount().longValue()), - RECEIVED_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedBytesPerSecond", + RECEIVED_BYTES_PER_SECOND( + "receivedBytesPerSecond", "Received Bytes Per Second", "The data rate at which data was received from the remote system in the past 5 minutes in terms of Bytes Per Second", Formatter.DATA_SIZE, - s -> s.getReceivedContentSize().longValue() / 300L)), + s -> s.getReceivedContentSize().longValue() / 300L), - SENT_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentBytesPerSecond", + SENT_BYTES_PER_SECOND( + "sentBytesPerSecond", "Sent Bytes Per Second", "The data rate at which data was received from the remote system in the past 5 minutes in terms of Bytes Per Second", Formatter.DATA_SIZE, - s -> s.getSentContentSize().longValue() / 300L)), + s -> s.getSentContentSize().longValue() / 300L), - TOTAL_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("totalBytesPerSecond", + TOTAL_BYTES_PER_SECOND("totalBytesPerSecond", "Total Bytes Per Second", "The sum of the send and receive data rate from the remote system in the past 5 minutes in terms of Bytes Per Second", Formatter.DATA_SIZE, @@ -69,9 +75,9 @@ public enum RemoteProcessGroupStatusDescriptor { public Long getValue(final RemoteProcessGroupStatus status) { return Long.valueOf((status.getReceivedContentSize().longValue() + status.getSentContentSize().longValue()) / 300L); } - })), + }), - AVERAGE_LINEAGE_DURATION(new StandardMetricDescriptor<RemoteProcessGroupStatus>( + AVERAGE_LINEAGE_DURATION( "averageLineageDuration", "Average Lineage Duration (5 mins)", "The average amount of time that a FlowFile took to process from receipt to drop in the past 5 minutes. For Processors that do not terminate FlowFiles, this value will be 0.", @@ -84,22 +90,29 @@ public enum RemoteProcessGroupStatusDescriptor { int count = 0; for (final StatusSnapshot snapshot : values) { - final long sent = snapshot.getStatusMetrics().get(SENT_COUNT.getDescriptor()).longValue(); + final long sent = snapshot.getStatusMetric(SENT_COUNT.getDescriptor()).longValue(); count += sent; - final long avgMillis = snapshot.getStatusMetrics().get(AVERAGE_LINEAGE_DURATION.getDescriptor()).longValue(); + final long avgMillis = snapshot.getStatusMetric(AVERAGE_LINEAGE_DURATION.getDescriptor()).longValue(); final long totalMillis = avgMillis * sent; millis += totalMillis; } return count == 0 ? 0 : millis / count; } - })); + }); + private final MetricDescriptor<RemoteProcessGroupStatus> descriptor; - private RemoteProcessGroupStatusDescriptor(final MetricDescriptor<RemoteProcessGroupStatus> descriptor) { - this.descriptor = descriptor; + RemoteProcessGroupStatusDescriptor(final String field, final String label, final String description, + final MetricDescriptor.Formatter formatter, final ValueMapper<RemoteProcessGroupStatus> valueFunction) { + this.descriptor = new StandardMetricDescriptor<>(this::ordinal, field, label, description, formatter, valueFunction); + } + + RemoteProcessGroupStatusDescriptor(final String field, final String label, final String description, + final MetricDescriptor.Formatter formatter, final ValueMapper<RemoteProcessGroupStatus> valueFunction, final ValueReducer<StatusSnapshot, Long> reducer) { + this.descriptor = new StandardMetricDescriptor<>(this::ordinal, field, label, description, formatter, valueFunction, reducer); } public String getField() { http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardMetricDescriptor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardMetricDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardMetricDescriptor.java index 6970fce..de5e0b5 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardMetricDescriptor.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardMetricDescriptor.java @@ -20,6 +20,7 @@ import java.util.List; public class StandardMetricDescriptor<T> implements MetricDescriptor<T> { + private final IndexableMetric indexableMetric; private final String field; private final String label; private final String description; @@ -27,12 +28,14 @@ public class StandardMetricDescriptor<T> implements MetricDescriptor<T> { private final ValueMapper<T> valueMapper; private final ValueReducer<StatusSnapshot, Long> reducer; - public StandardMetricDescriptor(final String field, final String label, final String description, final MetricDescriptor.Formatter formatter, final ValueMapper<T> valueFunction) { - this(field, label, description, formatter, valueFunction, null); + public StandardMetricDescriptor(final IndexableMetric indexableMetric, final String field, final String label, final String description, + final MetricDescriptor.Formatter formatter, final ValueMapper<T> valueFunction) { + this(indexableMetric, field, label, description, formatter, valueFunction, null); } - public StandardMetricDescriptor(final String field, final String label, final String description, - final MetricDescriptor.Formatter formatter, final ValueMapper<T> valueFunction, final ValueReducer<StatusSnapshot, Long> reducer) { + public StandardMetricDescriptor(final IndexableMetric indexableMetric, final String field, final String label, final String description, + final MetricDescriptor.Formatter formatter, final ValueMapper<T> valueFunction, final ValueReducer<StatusSnapshot, Long> reducer) { + this.indexableMetric = indexableMetric; this.field = field; this.label = label; this.description = description; @@ -42,6 +45,11 @@ public class StandardMetricDescriptor<T> implements MetricDescriptor<T> { } @Override + public int getMetricIdentifier() { + return indexableMetric.getIndex(); + } + + @Override public String getField() { return field; } @@ -100,7 +108,7 @@ public class StandardMetricDescriptor<T> implements MetricDescriptor<T> { public Long reduce(final List<StatusSnapshot> values) { long sum = 0; for (final StatusSnapshot snapshot : values) { - sum += snapshot.getStatusMetrics().get(StandardMetricDescriptor.this); + sum += snapshot.getStatusMetric(StandardMetricDescriptor.this); } return sum; http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusHistory.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusHistory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusHistory.java index 9299b0d..96b3759 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusHistory.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusHistory.java @@ -16,18 +16,25 @@ */ package org.apache.nifi.controller.status.history; -import java.util.ArrayList; -import java.util.Collections; import java.util.Date; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; public class StandardStatusHistory implements StatusHistory { + private final List<StatusSnapshot> snapshots; + private final Date generated; + private final Map<String, String> componentDetails; + + public StandardStatusHistory(final List<StatusSnapshot> snapshots, final Map<String, String> componentDetails, final Date generated) { + this.snapshots = snapshots; + this.generated = generated; + this.componentDetails = componentDetails; + } - private final List<StatusSnapshot> snapshots = new ArrayList<>(); - private final Date generated = new Date(); - private final Map<String, String> componentDetails = new LinkedHashMap<>(); + @Override + public List<StatusSnapshot> getStatusSnapshots() { + return snapshots; + } @Override public Date getDateGenerated() { @@ -36,19 +43,6 @@ public class StandardStatusHistory implements StatusHistory { @Override public Map<String, String> getComponentDetails() { - return Collections.unmodifiableMap(componentDetails); - } - - public void setComponentDetail(final String detailName, final String detailValue) { - componentDetails.put(detailName, detailValue); - } - - @Override - public List<StatusSnapshot> getStatusSnapshots() { - return Collections.unmodifiableList(snapshots); - } - - public void addStatusSnapshot(final StatusSnapshot snapshot) { - snapshots.add(snapshot); + return componentDetails; } } http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusSnapshot.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusSnapshot.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusSnapshot.java index abaf899..4662753 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusSnapshot.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StandardStatusSnapshot.java @@ -17,7 +17,7 @@ package org.apache.nifi.controller.status.history; import java.util.Date; -import java.util.LinkedHashMap; +import java.util.HashMap; import java.util.LinkedHashSet; import java.util.List; import java.util.Map; @@ -25,25 +25,63 @@ import java.util.Set; public class StandardStatusSnapshot implements StatusSnapshot { - private final Map<MetricDescriptor<?>, Long> metricValues = new LinkedHashMap<>(); + private final Set<MetricDescriptor<?>> metricDescriptors; + private final long[] values; + + private Map<MetricDescriptor<?>, Long> counterValues = null; private Date timestamp = new Date(); + + public StandardStatusSnapshot(final Set<MetricDescriptor<?>> metricDescriptors) { + this.metricDescriptors = metricDescriptors; + values = new long[metricDescriptors.size()]; + } + + private StandardStatusSnapshot(final Set<MetricDescriptor<?>> metricDescriptors, final long[] values) { + this.metricDescriptors = metricDescriptors; + this.values = values; + } + @Override public Date getTimestamp() { return timestamp; } - public void setTimestamp(final Date timestamp) { - this.timestamp = timestamp; + @Override + public Set<MetricDescriptor<?>> getMetricDescriptors() { + return metricDescriptors; } @Override - public Map<MetricDescriptor<?>, Long> getStatusMetrics() { - return metricValues; + public Long getStatusMetric(final MetricDescriptor<?> descriptor) { + return values[descriptor.getMetricIdentifier()]; + } + + public void setTimestamp(final Date timestamp) { + this.timestamp = timestamp; } + public void addStatusMetric(final MetricDescriptor<?> metric, final Long value) { - metricValues.put(metric, value); + values[metric.getMetricIdentifier()] = value; + } + + public void addCounterStatusMetric(final MetricDescriptor<?> metric, final Long value) { + if (counterValues == null) { + counterValues = new HashMap<>(); + } + + counterValues.put(metric, value); + } + + public StandardStatusSnapshot withoutCounters() { + if (counterValues == null) { + return this; + } + + final StandardStatusSnapshot without = new StandardStatusSnapshot(metricDescriptors, values); + without.setTimestamp(timestamp); + return without; } @Override @@ -52,16 +90,16 @@ public class StandardStatusSnapshot implements StatusSnapshot { @Override public StatusSnapshot reduce(final List<StatusSnapshot> values) { Date reducedTimestamp = null; - final Set<MetricDescriptor<?>> allDescriptors = new LinkedHashSet<>(metricValues.keySet()); + final Set<MetricDescriptor<?>> allDescriptors = new LinkedHashSet<>(metricDescriptors); for (final StatusSnapshot statusSnapshot : values) { if (reducedTimestamp == null) { reducedTimestamp = statusSnapshot.getTimestamp(); } - allDescriptors.addAll(statusSnapshot.getStatusMetrics().keySet()); + allDescriptors.addAll(statusSnapshot.getMetricDescriptors()); } - final StandardStatusSnapshot reduced = new StandardStatusSnapshot(); + final StandardStatusSnapshot reduced = new StandardStatusSnapshot(allDescriptors); if (reducedTimestamp != null) { reduced.setTimestamp(reducedTimestamp); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java index a126589..9ac3909 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java @@ -16,6 +16,10 @@ */ package org.apache.nifi.controller.status.history; +import org.apache.nifi.web.api.dto.status.StatusDescriptorDTO; +import org.apache.nifi.web.api.dto.status.StatusHistoryDTO; +import org.apache.nifi.web.api.dto.status.StatusSnapshotDTO; + import java.util.ArrayList; import java.util.Collection; import java.util.Date; @@ -27,10 +31,6 @@ import java.util.List; import java.util.Map; import java.util.Set; -import org.apache.nifi.web.api.dto.status.StatusDescriptorDTO; -import org.apache.nifi.web.api.dto.status.StatusHistoryDTO; -import org.apache.nifi.web.api.dto.status.StatusSnapshotDTO; - public class StatusHistoryUtil { public static StatusHistoryDTO createStatusHistoryDTO(final StatusHistory statusHistory) { @@ -43,7 +43,7 @@ public class StatusHistoryUtil { final StatusSnapshotDTO snapshotDto = StatusHistoryUtil.createStatusSnapshotDto(snapshot); snapshotDtos.add(snapshotDto); metricNames.addAll(snapshotDto.getStatusMetrics().keySet()); - metricDescriptors.addAll(snapshot.getStatusMetrics().keySet()); + metricDescriptors.addAll(snapshot.getMetricDescriptors()); } // We need to ensure that the 'aggregate snapshot' has an entry for every metric, including counters. @@ -94,9 +94,7 @@ public class StatusHistoryUtil { final Set<MetricDescriptor<?>> allDescriptors = new LinkedHashSet<>(); for (final StatusSnapshot statusSnapshot : statusHistory.getStatusSnapshots()) { - for (final MetricDescriptor<?> metricDescriptor : statusSnapshot.getStatusMetrics().keySet()) { - allDescriptors.add(metricDescriptor); - } + allDescriptors.addAll(statusSnapshot.getMetricDescriptors()); } for (final MetricDescriptor<?> metricDescriptor : allDescriptors) { @@ -111,8 +109,8 @@ public class StatusHistoryUtil { dto.setTimestamp(statusSnapshot.getTimestamp()); final Map<String, Long> statusMetrics = new HashMap<>(); - for (final Map.Entry<MetricDescriptor<?>, Long> entry : statusSnapshot.getStatusMetrics().entrySet()) { - statusMetrics.put(entry.getKey().getField(), entry.getValue()); + for (final MetricDescriptor<?> descriptor : statusSnapshot.getMetricDescriptors()) { + statusMetrics.put(descriptor.getField(), statusSnapshot.getStatusMetric(descriptor)); } dto.setStatusMetrics(statusMetrics); http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java index 6b90030..5336d17 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java @@ -16,43 +16,64 @@ */ package org.apache.nifi.controller.status.history; -import java.util.Date; -import java.util.List; -import java.util.Map; - import org.apache.nifi.controller.status.ConnectionStatus; import org.apache.nifi.controller.status.ProcessGroupStatus; import org.apache.nifi.controller.status.ProcessorStatus; import org.apache.nifi.controller.status.RemoteProcessGroupStatus; -import org.apache.nifi.controller.status.history.MetricDescriptor.Formatter; -import org.apache.nifi.util.ComponentStatusReport; -import org.apache.nifi.util.ComponentStatusReport.ComponentType; +import org.apache.nifi.util.ComponentMetrics; import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.util.RingBuffer; -import org.apache.nifi.util.RingBuffer.ForEachEvaluator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Arrays; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + public class VolatileComponentStatusRepository implements ComponentStatusRepository { + private static final Logger logger = LoggerFactory.getLogger(VolatileComponentStatusRepository.class); + + private static final Set<MetricDescriptor<?>> DEFAULT_PROCESSOR_METRICS = Arrays.stream(ProcessorStatusDescriptor.values()) + .map(ProcessorStatusDescriptor::getDescriptor) + .collect(Collectors.toSet()); + private static final Set<MetricDescriptor<?>> DEFAULT_CONNECTION_METRICS = Arrays.stream(ConnectionStatusDescriptor.values()) + .map(ConnectionStatusDescriptor::getDescriptor) + .collect(Collectors.toSet()); + private static final Set<MetricDescriptor<?>> DEFAULT_GROUP_METRICS = Arrays.stream(ProcessGroupStatusDescriptor.values()) + .map(ProcessGroupStatusDescriptor::getDescriptor) + .collect(Collectors.toSet()); + private static final Set<MetricDescriptor<?>> DEFAULT_RPG_METRICS = Arrays.stream(RemoteProcessGroupStatusDescriptor.values()) + .map(RemoteProcessGroupStatusDescriptor::getDescriptor) + .collect(Collectors.toSet()); + public static final String NUM_DATA_POINTS_PROPERTY = "nifi.components.status.repository.buffer.size"; public static final int DEFAULT_NUM_DATA_POINTS = 288; // 1 day worth of 5-minute snapshots - private final RingBuffer<Capture> captures; - private final Logger logger = LoggerFactory.getLogger(VolatileComponentStatusRepository.class); + private final Map<String, ComponentStatusHistory> componentStatusHistories = new HashMap<>(); + private final RingBuffer<Date> timestamps; + private final RingBuffer<List<GarbageCollectionStatus>> gcStatuses; + private final int numDataPoints; private volatile long lastCaptureTime = 0L; /** * Default no args constructor for service loading only */ - public VolatileComponentStatusRepository(){ - captures = null; + public VolatileComponentStatusRepository() { + numDataPoints = DEFAULT_NUM_DATA_POINTS; + gcStatuses = null; + timestamps = null; } public VolatileComponentStatusRepository(final NiFiProperties nifiProperties) { - final int numDataPoints = nifiProperties.getIntegerProperty(NUM_DATA_POINTS_PROPERTY, DEFAULT_NUM_DATA_POINTS); - captures = new RingBuffer<>(numDataPoints); + numDataPoints = nifiProperties.getIntegerProperty(NUM_DATA_POINTS_PROPERTY, DEFAULT_NUM_DATA_POINTS); + gcStatuses = new RingBuffer<>(numDataPoints); + timestamps = new RingBuffer<>(numDataPoints); } @Override @@ -62,214 +83,116 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit @Override public synchronized void capture(final ProcessGroupStatus rootGroupStatus, final List<GarbageCollectionStatus> gcStatus, final Date timestamp) { - final ComponentStatusReport statusReport = ComponentStatusReport.fromProcessGroupStatus(rootGroupStatus, ComponentType.PROCESSOR, - ComponentType.CONNECTION, ComponentType.PROCESS_GROUP, ComponentType.REMOTE_PROCESS_GROUP); + final Date evicted = timestamps.add(timestamp); + if (evicted != null) { + componentStatusHistories.values().forEach(history -> history.expireBefore(evicted)); + } + + capture(rootGroupStatus, timestamp); + gcStatuses.add(gcStatus); - captures.add(new Capture(timestamp, statusReport, gcStatus)); logger.debug("Captured metrics for {}", this); lastCaptureTime = Math.max(lastCaptureTime, timestamp.getTime()); } - @Override - public Date getLastCaptureDate() { - return new Date(lastCaptureTime); - } - @Override - public StatusHistory getProcessorStatusHistory(final String processorId, final Date start, final Date end, final int preferredDataPoints, final boolean includeCounters) { - final StandardStatusHistory history = new StandardStatusHistory(); - history.setComponentDetail(COMPONENT_DETAIL_ID, processorId); - - captures.forEach(new ForEachEvaluator<Capture>() { - @Override - public boolean evaluate(final Capture capture) { - final ComponentStatusReport statusReport = capture.getStatusReport(); - final ProcessorStatus status = statusReport.getProcessorStatus(processorId); - if (status == null) { - return true; - } + private void capture(final ProcessGroupStatus groupStatus, final Date timestamp) { + // Capture status for the ProcessGroup + final ComponentDetails groupDetails = ComponentDetails.forProcessGroup(groupStatus); + final StatusSnapshot groupSnapshot = ComponentMetrics.createSnapshot(groupStatus, timestamp); + updateStatusHistory(groupSnapshot, groupDetails, timestamp); - history.setComponentDetail(COMPONENT_DETAIL_GROUP_ID, status.getGroupId()); - history.setComponentDetail(COMPONENT_DETAIL_NAME, status.getName()); - history.setComponentDetail(COMPONENT_DETAIL_TYPE, status.getType()); + // Capture statuses for the Processors + for (final ProcessorStatus processorStatus : groupStatus.getProcessorStatus()) { + final ComponentDetails componentDetails = ComponentDetails.forProcessor(processorStatus); + final StatusSnapshot snapshot = ComponentMetrics.createSnapshot(processorStatus, timestamp); + updateStatusHistory(snapshot, componentDetails, timestamp); + } - final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(); - snapshot.setTimestamp(capture.getCaptureDate()); + // Capture statuses for the Connections + for (final ConnectionStatus connectionStatus : groupStatus.getConnectionStatus()) { + final ComponentDetails componentDetails = ComponentDetails.forConnection(connectionStatus); + final StatusSnapshot snapshot = ComponentMetrics.createSnapshot(connectionStatus, timestamp); + updateStatusHistory(snapshot, componentDetails, timestamp); + } - for (final ProcessorStatusDescriptor descriptor : ProcessorStatusDescriptor.values()) { - if (descriptor.isVisible()) { - snapshot.addStatusMetric(descriptor.getDescriptor(), descriptor.getDescriptor().getValueFunction().getValue(status)); - } - } + // Capture statuses for the RPG's + for (final RemoteProcessGroupStatus rpgStatus : groupStatus.getRemoteProcessGroupStatus()) { + final ComponentDetails componentDetails = ComponentDetails.forRemoteProcessGroup(rpgStatus); + final StatusSnapshot snapshot = ComponentMetrics.createSnapshot(rpgStatus, timestamp); + updateStatusHistory(snapshot, componentDetails, timestamp); + } - if (includeCounters) { - final Map<String, Long> counters = status.getCounters(); - if (counters != null) { - for (final Map.Entry<String, Long> entry : counters.entrySet()) { - final String counterName = entry.getKey(); + // Capture statuses for the child groups + for (final ProcessGroupStatus childStatus : groupStatus.getProcessGroupStatus()) { + capture(childStatus, timestamp); + } + } - final String label = entry.getKey() + " (5 mins)"; - final MetricDescriptor<ProcessorStatus> metricDescriptor = new StandardMetricDescriptor<>(entry.getKey(), label, label, Formatter.COUNT, - s -> s.getCounters() == null ? null : s.getCounters().get(counterName)); - snapshot.addStatusMetric(metricDescriptor, entry.getValue()); - } - } - } + private void updateStatusHistory(final StatusSnapshot statusSnapshot, final ComponentDetails componentDetails, final Date timestamp) { + final String componentId = componentDetails.getComponentId(); + final ComponentStatusHistory procHistory = componentStatusHistories.computeIfAbsent(componentId, id -> new ComponentStatusHistory(componentDetails, numDataPoints)); + procHistory.update(statusSnapshot, componentDetails); + } - history.addStatusSnapshot(snapshot); - return true; - } - }); - return history; + @Override + public Date getLastCaptureDate() { + return new Date(lastCaptureTime); } @Override - public StatusHistory getConnectionStatusHistory(final String connectionId, final Date start, final Date end, final int preferredDataPoints) { - final StandardStatusHistory history = new StandardStatusHistory(); - history.setComponentDetail(COMPONENT_DETAIL_ID, connectionId); - - captures.forEach(new ForEachEvaluator<Capture>() { - @Override - public boolean evaluate(final Capture capture) { - final ComponentStatusReport statusReport = capture.getStatusReport(); - final ConnectionStatus status = statusReport.getConnectionStatus(connectionId); - if (status == null) { - return true; - } - - history.setComponentDetail(COMPONENT_DETAIL_GROUP_ID, status.getGroupId()); - history.setComponentDetail(COMPONENT_DETAIL_NAME, status.getName()); - history.setComponentDetail(COMPONENT_DETAIL_SOURCE_NAME, status.getSourceName()); - history.setComponentDetail(COMPONENT_DETAIL_DESTINATION_NAME, status.getDestinationName()); - - final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(); - snapshot.setTimestamp(capture.getCaptureDate()); - - for (final ConnectionStatusDescriptor descriptor : ConnectionStatusDescriptor.values()) { - snapshot.addStatusMetric(descriptor.getDescriptor(), descriptor.getDescriptor().getValueFunction().getValue(status)); - } - - history.addStatusSnapshot(snapshot); - return true; - } - }); + public StatusHistory getProcessorStatusHistory(final String processorId, final Date start, final Date end, final int preferredDataPoints, final boolean includeCounters) { + return getStatusHistory(processorId, includeCounters, DEFAULT_PROCESSOR_METRICS); + } - return history; + @Override + public StatusHistory getConnectionStatusHistory(final String connectionId, final Date start, final Date end, final int preferredDataPoints) { + return getStatusHistory(connectionId, true, DEFAULT_CONNECTION_METRICS); } @Override public StatusHistory getProcessGroupStatusHistory(final String processGroupId, final Date start, final Date end, final int preferredDataPoints) { - final StandardStatusHistory history = new StandardStatusHistory(); - history.setComponentDetail(COMPONENT_DETAIL_ID, processGroupId); - - captures.forEach(new ForEachEvaluator<Capture>() { - @Override - public boolean evaluate(final Capture capture) { - final ComponentStatusReport statusReport = capture.getStatusReport(); - final ProcessGroupStatus status = statusReport.getProcessGroupStatus(processGroupId); - if (status == null) { - return true; - } - - history.setComponentDetail(COMPONENT_DETAIL_NAME, status.getName()); - - final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(); - snapshot.setTimestamp(capture.getCaptureDate()); - - for (final ProcessGroupStatusDescriptor descriptor : ProcessGroupStatusDescriptor.values()) { - snapshot.addStatusMetric(descriptor.getDescriptor(), descriptor.getDescriptor().getValueFunction().getValue(status)); - } - - history.addStatusSnapshot(snapshot); - return true; - } - }); - - return history; + return getStatusHistory(processGroupId, true, DEFAULT_GROUP_METRICS); } @Override public StatusHistory getRemoteProcessGroupStatusHistory(final String remoteGroupId, final Date start, final Date end, final int preferredDataPoints) { - final StandardStatusHistory history = new StandardStatusHistory(); - history.setComponentDetail(COMPONENT_DETAIL_ID, remoteGroupId); - - captures.forEach(new ForEachEvaluator<Capture>() { - @Override - public boolean evaluate(final Capture capture) { - final ComponentStatusReport statusReport = capture.getStatusReport(); - final RemoteProcessGroupStatus status = statusReport.getRemoteProcessGroupStatus(remoteGroupId); - if (status == null) { - return true; - } - - history.setComponentDetail(COMPONENT_DETAIL_GROUP_ID, status.getGroupId()); - history.setComponentDetail(COMPONENT_DETAIL_NAME, status.getName()); - history.setComponentDetail(COMPONENT_DETAIL_URI, status.getTargetUri()); + return getStatusHistory(remoteGroupId, true, DEFAULT_RPG_METRICS); + } - final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(); - snapshot.setTimestamp(capture.getCaptureDate()); - for (final RemoteProcessGroupStatusDescriptor descriptor : RemoteProcessGroupStatusDescriptor.values()) { - snapshot.addStatusMetric(descriptor.getDescriptor(), descriptor.getDescriptor().getValueFunction().getValue(status)); - } - - history.addStatusSnapshot(snapshot); - return true; - } - }); + private synchronized StatusHistory getStatusHistory(final String componentId, final boolean includeCounters, final Set<MetricDescriptor<?>> defaultMetricDescriptors) { + final ComponentStatusHistory history = componentStatusHistories.get(componentId); + if (history == null) { + return null; + } - return history; + final List<Date> dates = timestamps.asList(); + return history.toStatusHistory(dates, includeCounters, defaultMetricDescriptors); } + @Override public GarbageCollectionHistory getGarbageCollectionHistory(final Date start, final Date end) { final StandardGarbageCollectionHistory history = new StandardGarbageCollectionHistory(); - captures.forEach(new ForEachEvaluator<Capture>() { - @Override - public boolean evaluate(final Capture capture) { - if (capture.getCaptureDate().before(start)) { - return true; - } - if (capture.getCaptureDate().after(end)) { - return false; + gcStatuses.forEach(statusSet -> { + for (final GarbageCollectionStatus gcStatus : statusSet) { + if (gcStatus.getTimestamp().before(start)) { + continue; } - - final List<GarbageCollectionStatus> statuses = capture.getGarbageCollectionStatus(); - if (statuses != null) { - statuses.stream().forEach(history::addGarbageCollectionStatus); + if (gcStatus.getTimestamp().after(end)) { + continue; } - return true; + history.addGarbageCollectionStatus(gcStatus); } + + return true; }); return history; } - - private static class Capture { - private final Date captureDate; - private final ComponentStatusReport statusReport; - private final List<GarbageCollectionStatus> gcStatus; - - public Capture(final Date date, final ComponentStatusReport statusReport, final List<GarbageCollectionStatus> gcStatus) { - this.captureDate = date; - this.statusReport = statusReport; - this.gcStatus = gcStatus; - } - - public Date getCaptureDate() { - return captureDate; - } - - public ComponentStatusReport getStatusReport() { - return statusReport; - } - - public List<GarbageCollectionStatus> getGarbageCollectionStatus() { - return gcStatus; - } - } } http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java index 1a98ee3..e68aba8 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java @@ -16,10 +16,6 @@ */ package org.apache.nifi.controller.tasks; -import java.io.IOException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; - import org.apache.nifi.components.state.StateManager; import org.apache.nifi.connectable.Connectable; import org.apache.nifi.connectable.ConnectableType; @@ -35,8 +31,8 @@ import org.apache.nifi.controller.repository.StandardProcessSessionFactory; import org.apache.nifi.controller.repository.WeakHashMapProcessSessionFactory; import org.apache.nifi.controller.repository.metrics.StandardFlowFileEvent; import org.apache.nifi.controller.scheduling.ConnectableProcessContext; -import org.apache.nifi.controller.scheduling.RepositoryContextFactory; import org.apache.nifi.controller.scheduling.LifecycleState; +import org.apache.nifi.controller.scheduling.RepositoryContextFactory; import org.apache.nifi.controller.scheduling.SchedulingAgent; import org.apache.nifi.encrypt.StringEncryptor; import org.apache.nifi.logging.ComponentLog; @@ -51,6 +47,10 @@ import org.apache.nifi.util.Connectables; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + /** * Continually runs a <code>{@link Connectable}</code> component as long as the component has work to do. * {@link #invoke()} ()} will return <code>{@link InvocationResult}</code> telling if the component should be yielded. @@ -267,10 +267,10 @@ public class ConnectableTask { final long processingNanos = System.nanoTime() - startNanos; try { - final StandardFlowFileEvent procEvent = new StandardFlowFileEvent(connectable.getIdentifier()); + final StandardFlowFileEvent procEvent = new StandardFlowFileEvent(); procEvent.setProcessingNanos(processingNanos); procEvent.setInvocations(invocationCount); - repositoryContext.getFlowFileEventRepository().updateRepository(procEvent); + repositoryContext.getFlowFileEventRepository().updateRepository(procEvent, connectable.getIdentifier()); } catch (final IOException e) { logger.error("Unable to update FlowFileEvent Repository for {}; statistics may be inaccurate. Reason for failure: {}", connectable.getRunnableComponent(), e.toString()); logger.error("", e); http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java index edfa355..e173cc6 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java @@ -957,7 +957,7 @@ public final class StandardProcessGroup implements ProcessGroup { public Collection<ProcessorNode> getProcessors() { readLock.lock(); try { - return processors.values(); + return new HashSet<>(processors.values()); } finally { readLock.unlock(); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/ComponentMetrics.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/ComponentMetrics.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/ComponentMetrics.java new file mode 100644 index 0000000..9c1505c --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/ComponentMetrics.java @@ -0,0 +1,188 @@ +/* + * 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 org.apache.nifi.controller.status.ConnectionStatus; +import org.apache.nifi.controller.status.ProcessGroupStatus; +import org.apache.nifi.controller.status.ProcessorStatus; +import org.apache.nifi.controller.status.RemoteProcessGroupStatus; +import org.apache.nifi.controller.status.history.ConnectionStatusDescriptor; +import org.apache.nifi.controller.status.history.MetricDescriptor; +import org.apache.nifi.controller.status.history.ProcessGroupStatusDescriptor; +import org.apache.nifi.controller.status.history.ProcessorStatusDescriptor; +import org.apache.nifi.controller.status.history.RemoteProcessGroupStatusDescriptor; +import org.apache.nifi.controller.status.history.StandardMetricDescriptor; +import org.apache.nifi.controller.status.history.StandardStatusSnapshot; +import org.apache.nifi.controller.status.history.StatusSnapshot; + +import java.util.Date; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +public class ComponentMetrics { + private static final Set<MetricDescriptor<?>> PROCESSOR_METRICS; + private static final Set<MetricDescriptor<?>> CONNECTION_METRICS; + private static final Set<MetricDescriptor<?>> PROCESS_GROUP_METRICS; + private static final Set<MetricDescriptor<?>> RPG_METRICS; + + static { + PROCESSOR_METRICS = new HashSet<>(); + CONNECTION_METRICS = new HashSet<>(); + PROCESS_GROUP_METRICS = new HashSet<>(); + RPG_METRICS = new HashSet<>(); + + for (final ProcessorStatusDescriptor descriptor : ProcessorStatusDescriptor.values()) { + PROCESSOR_METRICS.add(descriptor.getDescriptor()); + } + + for (final ConnectionStatusDescriptor descriptor : ConnectionStatusDescriptor.values()) { + CONNECTION_METRICS.add(descriptor.getDescriptor()); + } + + for (final ProcessGroupStatusDescriptor descriptor : ProcessGroupStatusDescriptor.values()) { + PROCESS_GROUP_METRICS.add(descriptor.getDescriptor()); + } + + for (final RemoteProcessGroupStatusDescriptor descriptor : RemoteProcessGroupStatusDescriptor.values()) { + RPG_METRICS.add(descriptor.getDescriptor()); + } + } + + + public static StatusSnapshot createSnapshot(final ProcessorStatus status, final Date timestamp) { + if (isEmpty(status)) { + return null; + } + + final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(PROCESSOR_METRICS); + snapshot.setTimestamp(timestamp); + + for (final ProcessorStatusDescriptor descriptor : ProcessorStatusDescriptor.values()) { + if (descriptor.isVisible()) { + snapshot.addStatusMetric(descriptor.getDescriptor(), descriptor.getDescriptor().getValueFunction().getValue(status)); + } + } + + final Map<String, Long> counters = status.getCounters(); + if (counters != null) { + for (final Map.Entry<String, Long> entry : counters.entrySet()) { + final String counterName = entry.getKey(); + + final String label = entry.getKey() + " (5 mins)"; + final MetricDescriptor<ProcessorStatus> metricDescriptor = new StandardMetricDescriptor<>(() -> 0, entry.getKey(), label, label, MetricDescriptor.Formatter.COUNT, + s -> s.getCounters() == null ? null : s.getCounters().get(counterName)); + + snapshot.addCounterStatusMetric(metricDescriptor, entry.getValue()); + } + } + + return snapshot; + } + + public static boolean isEmpty(final ProcessorStatus status) { + for (final ProcessorStatusDescriptor descriptor : ProcessorStatusDescriptor.values()) { + if (descriptor.isVisible()) { + final Long value = descriptor.getDescriptor().getValueFunction().getValue(status); + if (value != null && value > 0) { + return false; + } + } + } + + return true; + } + + + public static StatusSnapshot createSnapshot(final ConnectionStatus status, final Date timestamp) { + if (isEmpty(status)) { + return null; + } + + final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(CONNECTION_METRICS); + snapshot.setTimestamp(timestamp); + + for (final ConnectionStatusDescriptor descriptor : ConnectionStatusDescriptor.values()) { + snapshot.addStatusMetric(descriptor.getDescriptor(), descriptor.getDescriptor().getValueFunction().getValue(status)); + } + + return snapshot; + } + + public static boolean isEmpty(final ConnectionStatus status) { + for (final ConnectionStatusDescriptor descriptor : ConnectionStatusDescriptor.values()) { + final Long value = descriptor.getDescriptor().getValueFunction().getValue(status); + if (value != null && value > 0) { + return false; + } + } + + return true; + } + + public static StatusSnapshot createSnapshot(final ProcessGroupStatus status, final Date timestamp) { + if (isEmpty(status)) { + return null; + } + + final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(PROCESS_GROUP_METRICS); + snapshot.setTimestamp(timestamp); + + for (final ProcessGroupStatusDescriptor descriptor : ProcessGroupStatusDescriptor.values()) { + snapshot.addStatusMetric(descriptor.getDescriptor(), descriptor.getDescriptor().getValueFunction().getValue(status)); + } + + return snapshot; + } + + private static boolean isEmpty(final ProcessGroupStatus status) { + for (final ProcessGroupStatusDescriptor descriptor : ProcessGroupStatusDescriptor.values()) { + final Long value = descriptor.getDescriptor().getValueFunction().getValue(status); + if (value != null && value > 0) { + return false; + } + } + + return true; + } + + public static StatusSnapshot createSnapshot(final RemoteProcessGroupStatus status, final Date timestamp) { + if (isEmpty(status)) { + return null; + } + + final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(RPG_METRICS); + snapshot.setTimestamp(timestamp); + + for (final RemoteProcessGroupStatusDescriptor descriptor : RemoteProcessGroupStatusDescriptor.values()) { + snapshot.addStatusMetric(descriptor.getDescriptor(), descriptor.getDescriptor().getValueFunction().getValue(status)); + } + + return snapshot; + } + + private static boolean isEmpty(final RemoteProcessGroupStatus status) { + for (final RemoteProcessGroupStatusDescriptor descriptor : RemoteProcessGroupStatusDescriptor.values()) { + final Long value = descriptor.getDescriptor().getValueFunction().getValue(status); + if (value != null && value > 0) { + return false; + } + } + + return true; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/ComponentStatusReport.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/ComponentStatusReport.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/ComponentStatusReport.java deleted file mode 100644 index ca31467..0000000 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/ComponentStatusReport.java +++ /dev/null @@ -1,137 +0,0 @@ -/* - * 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.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - -import org.apache.nifi.controller.status.ConnectionStatus; -import org.apache.nifi.controller.status.PortStatus; -import org.apache.nifi.controller.status.ProcessGroupStatus; -import org.apache.nifi.controller.status.ProcessorStatus; -import org.apache.nifi.controller.status.RemoteProcessGroupStatus; - -/** - * ComponentStatusReport is a util class that can be used to "flatten" a ProcessGroupStatus into a collection of Map's so that retrieval of a Status for a particular component is very efficient - */ -public class ComponentStatusReport { - - private final Map<String, ProcessGroupStatus> groupMap = new HashMap<>(); - private final Map<String, ProcessorStatus> processorMap = new HashMap<>(); - private final Map<String, ConnectionStatus> connectionMap = new HashMap<>(); - private final Map<String, RemoteProcessGroupStatus> remoteGroupMap = new HashMap<>(); - private final Map<String, PortStatus> inputPortMap = new HashMap<>(); - private final Map<String, PortStatus> outputPortMap = new HashMap<>(); - - private ComponentStatusReport() { - } - - public static ComponentStatusReport createEmpty() { - return new ComponentStatusReport(); - } - - public static ComponentStatusReport fromProcessGroupStatus(final ProcessGroupStatus status) { - return fromProcessGroupStatus(status, ComponentType.values()); - } - - public static ComponentStatusReport fromProcessGroupStatus(final ProcessGroupStatus status, final ComponentType... componentTypes) { - final Set<ComponentType> componentTypeSet = new HashSet<>(); - for (final ComponentType type : componentTypes) { - componentTypeSet.add(type); - } - - final ComponentStatusReport report = new ComponentStatusReport(); - report.populate(status, componentTypeSet); - return report; - } - - private void populate(final ProcessGroupStatus status, final Set<ComponentType> componentTypes) { - if (componentTypes.contains(ComponentType.PROCESS_GROUP)) { - groupMap.put(status.getId(), status); - } - - if (componentTypes.contains(ComponentType.PROCESSOR)) { - for (final ProcessorStatus procStatus : status.getProcessorStatus()) { - processorMap.put(procStatus.getId(), procStatus); - } - } - - if (componentTypes.contains(ComponentType.CONNECTION)) { - for (final ConnectionStatus connStatus : status.getConnectionStatus()) { - connectionMap.put(connStatus.getId(), connStatus); - } - } - - if (componentTypes.contains(ComponentType.REMOTE_PROCESS_GROUP)) { - for (final RemoteProcessGroupStatus rpgStatus : status.getRemoteProcessGroupStatus()) { - remoteGroupMap.put(rpgStatus.getId(), rpgStatus); - } - } - - if (componentTypes.contains(ComponentType.INPUT_PORT)) { - for (final PortStatus portStatus : status.getInputPortStatus()) { - inputPortMap.put(portStatus.getId(), portStatus); - } - } - - if (componentTypes.contains(ComponentType.OUTPUT_PORT)) { - for (final PortStatus portStatus : status.getOutputPortStatus()) { - outputPortMap.put(portStatus.getId(), portStatus); - } - } - - for (final ProcessGroupStatus childStatus : status.getProcessGroupStatus()) { - populate(childStatus, componentTypes); - } - } - - public ProcessGroupStatus getProcessGroupStatus(final String groupId) { - return groupMap.get(groupId); - } - - public ProcessorStatus getProcessorStatus(final String processorId) { - return processorMap.get(processorId); - } - - public ConnectionStatus getConnectionStatus(final String connectionId) { - return connectionMap.get(connectionId); - } - - public RemoteProcessGroupStatus getRemoteProcessGroupStatus(final String remoteGroupId) { - return remoteGroupMap.get(remoteGroupId); - } - - public PortStatus getInputPortStatus(final String portId) { - return inputPortMap.get(portId); - } - - public PortStatus getOutputPortStatus(final String portId) { - return outputPortMap.get(portId); - } - - public static enum ComponentType { - - PROCESSOR, - INPUT_PORT, - OUTPUT_PORT, - PROCESS_GROUP, - CONNECTION, - REMOTE_PROCESS_GROUP; - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestRingBufferEventRepository.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestRingBufferEventRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestRingBufferEventRepository.java index 2bc158f..778efbe 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestRingBufferEventRepository.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestRingBufferEventRepository.java @@ -16,17 +16,15 @@ */ package org.apache.nifi.controller.repository; -import org.apache.nifi.controller.repository.StandardRepositoryStatusReport; import org.apache.nifi.controller.repository.metrics.RingBufferEventRepository; -import org.apache.nifi.controller.repository.FlowFileEvent; +import org.junit.Test; +import org.testng.Assert; + import java.io.IOException; import java.util.Collections; import java.util.Map; import java.util.concurrent.TimeUnit; -import org.junit.Test; -import org.testng.Assert; - public class TestRingBufferEventRepository { @Test @@ -34,15 +32,15 @@ public class TestRingBufferEventRepository { final RingBufferEventRepository repo = new RingBufferEventRepository(5); long insertNanos = 0L; for (int i = 0; i < 1000000; i++) { - final FlowFileEvent event = generateEvent("ABC"); + final FlowFileEvent event = generateEvent(); final long insertStart = System.nanoTime(); - repo.updateRepository(event); + repo.updateRepository(event, "ABC"); insertNanos += System.nanoTime() - insertStart; } final long queryStart = System.nanoTime(); - final StandardRepositoryStatusReport report = repo.reportTransferEvents(System.currentTimeMillis() - 2 * 60000); + final StandardRepositoryStatusReport report = repo.reportTransferEvents(System.currentTimeMillis()); final long queryNanos = System.nanoTime() - queryStart; System.out.println(report); System.out.println("Insert: " + TimeUnit.MILLISECONDS.convert(insertNanos, TimeUnit.NANOSECONDS)); @@ -55,37 +53,32 @@ public class TestRingBufferEventRepository { final FlowFileEventRepository repo = new RingBufferEventRepository(5); String id1 = "component1"; String id2 = "component2"; - repo.updateRepository(generateEvent(id1)); - repo.updateRepository(generateEvent(id2)); - RepositoryStatusReport report = repo.reportTransferEvents(System.currentTimeMillis() - 2 * 60000); + repo.updateRepository(generateEvent(), id1); + repo.updateRepository(generateEvent(), id2); + RepositoryStatusReport report = repo.reportTransferEvents(System.currentTimeMillis()); FlowFileEvent entry = report.getReportEntry(id1); Assert.assertNotNull(entry); entry = report.getReportEntry(id2); Assert.assertNotNull(entry); repo.purgeTransferEvents(id1); - report = repo.reportTransferEvents(System.currentTimeMillis() - 2 * 60000); + report = repo.reportTransferEvents(System.currentTimeMillis()); entry = report.getReportEntry(id1); Assert.assertNull(entry); entry = report.getReportEntry(id2); Assert.assertNotNull(entry); repo.purgeTransferEvents(id2); - report = repo.reportTransferEvents(System.currentTimeMillis() - 2 * 60000); + report = repo.reportTransferEvents(System.currentTimeMillis()); entry = report.getReportEntry(id2); Assert.assertNull(entry); repo.close(); } - private FlowFileEvent generateEvent(final String id) { + private FlowFileEvent generateEvent() { return new FlowFileEvent() { @Override - public String getComponentIdentifier() { - return id; - } - - @Override public int getFlowFilesIn() { return 1; } http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/metrics/TestSecondPrecisionEventContainer.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/metrics/TestSecondPrecisionEventContainer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/metrics/TestSecondPrecisionEventContainer.java new file mode 100644 index 0000000..f8b4559 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/metrics/TestSecondPrecisionEventContainer.java @@ -0,0 +1,135 @@ +/* + * 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.controller.repository.metrics; + +import org.apache.nifi.controller.repository.FlowFileEvent; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class TestSecondPrecisionEventContainer { + + @Test + public void testUpdateOncePerSecond() { + final SecondPrecisionEventContainer container = new SecondPrecisionEventContainer(5); + final long startTime = System.currentTimeMillis(); + + final StandardFlowFileEvent event = new StandardFlowFileEvent(); + event.setBytesRead(100L); + event.setBytesWritten(100L); + + for (int i=0; i < 5; i++) { + for (int j=0; j < 300; j++) { + container.addEvent(event, startTime + (i * 300_000) + (j * 1000)); + } + + final long timestamp = startTime + 300_000 * i + 300_000; + final FlowFileEvent result = container.generateReport(timestamp); + assertEquals("Failure at i=" + i, 300 * 100, result.getBytesRead()); + assertEquals("Failure at i=" + i, 300 * 100, result.getBytesWritten()); + } + } + + @Test + public void testExpiresOnReportGeneration() { + final SecondPrecisionEventContainer container = new SecondPrecisionEventContainer(5); + final long startTime = System.currentTimeMillis(); + + final StandardFlowFileEvent event = new StandardFlowFileEvent(); + event.setBytesRead(100L); + event.setBytesWritten(100L); + + for (int j=0; j < 100; j++) { + container.addEvent(event, startTime + (j * 1000)); + } + + final FlowFileEvent resultAt5Mins = container.generateReport(startTime + 300_000); + assertEquals(100 * 100, resultAt5Mins.getBytesRead()); + assertEquals(100 * 100, resultAt5Mins.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus50Seconds = container.generateReport(startTime + 350_000); + assertEquals(50 * 100, resultAt5MinsPlus50Seconds.getBytesRead()); + assertEquals(50 * 100, resultAt5MinsPlus50Seconds.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus99Seconds = container.generateReport(startTime + 399_000); + assertEquals(100, resultAt5MinsPlus99Seconds.getBytesRead()); + assertEquals(100, resultAt5MinsPlus99Seconds.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus100Seconds = container.generateReport(startTime + 400_000); + assertEquals(0, resultAt5MinsPlus100Seconds.getBytesRead()); + assertEquals(0, resultAt5MinsPlus100Seconds.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus101Seconds = container.generateReport(startTime + 401_000); + assertEquals(0, resultAt5MinsPlus101Seconds.getBytesRead()); + assertEquals(0, resultAt5MinsPlus101Seconds.getBytesWritten()); + + final FlowFileEvent resultsAt5MinsPlus300seconds = container.generateReport(startTime + 600_000); + assertEquals(0, resultsAt5MinsPlus300seconds.getBytesRead()); + assertEquals(0, resultsAt5MinsPlus300seconds.getBytesWritten()); + + final FlowFileEvent resultsAt5MinsPlus600seconds = container.generateReport(startTime + 900_000); + assertEquals(0, resultsAt5MinsPlus600seconds.getBytesRead()); + assertEquals(0, resultsAt5MinsPlus600seconds.getBytesWritten()); + } + + @Test + public void testExpiresOnReportGenerationWithSkipsBetweenUpdates() { + final SecondPrecisionEventContainer container = new SecondPrecisionEventContainer(5); + final long startTime = System.currentTimeMillis(); + + final StandardFlowFileEvent event = new StandardFlowFileEvent(); + event.setBytesRead(100L); + event.setBytesWritten(100L); + + for (int j=0; j < 20; j++) { + container.addEvent(event, startTime + (j * 5000)); + } + + final FlowFileEvent resultAt5Mins = container.generateReport(startTime + 300_000); + assertEquals(20 * 100, resultAt5Mins.getBytesRead()); + assertEquals(20 * 100, resultAt5Mins.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus50Seconds = container.generateReport(startTime + 350_000); + assertEquals(10 * 100, resultAt5MinsPlus50Seconds.getBytesRead()); + assertEquals(10 * 100, resultAt5MinsPlus50Seconds.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus94Seconds = container.generateReport(startTime + 394_000); + assertEquals(100, resultAt5MinsPlus94Seconds.getBytesRead()); + assertEquals(100, resultAt5MinsPlus94Seconds.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus95Seconds = container.generateReport(startTime + 395_000); + assertEquals(100, resultAt5MinsPlus95Seconds.getBytesRead()); + assertEquals(100, resultAt5MinsPlus95Seconds.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus100Seconds = container.generateReport(startTime + 400_000); + assertEquals(0, resultAt5MinsPlus100Seconds.getBytesRead()); + assertEquals(0, resultAt5MinsPlus100Seconds.getBytesWritten()); + + final FlowFileEvent resultAt5MinsPlus101Seconds = container.generateReport(startTime + 401_000); + assertEquals(0, resultAt5MinsPlus101Seconds.getBytesRead()); + assertEquals(0, resultAt5MinsPlus101Seconds.getBytesWritten()); + + final FlowFileEvent resultsAt5MinsPlus300seconds = container.generateReport(startTime + 600_000); + assertEquals(0, resultsAt5MinsPlus300seconds.getBytesRead()); + assertEquals(0, resultsAt5MinsPlus300seconds.getBytesWritten()); + + final FlowFileEvent resultsAt5MinsPlus600seconds = container.generateReport(startTime + 900_000); + assertEquals(0, resultsAt5MinsPlus600seconds.getBytesRead()); + assertEquals(0, resultsAt5MinsPlus600seconds.getBytesWritten()); + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/7bbb5a82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/TestMetricRollingBuffer.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/TestMetricRollingBuffer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/TestMetricRollingBuffer.java new file mode 100644 index 0000000..338fc62 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/status/history/TestMetricRollingBuffer.java @@ -0,0 +1,126 @@ +/* + * 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.controller.status.history; + +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class TestMetricRollingBuffer { + private static final Set<MetricDescriptor<?>> PROCESSOR_METRICS = Arrays.stream(ProcessorStatusDescriptor.values()) + .map(ProcessorStatusDescriptor::getDescriptor) + .collect(Collectors.toSet()); + + @Test + public void testBufferGrows() { + final int bufferCapacity = 1000; + final MetricRollingBuffer buffer = new MetricRollingBuffer(bufferCapacity); + + final long startTime = System.currentTimeMillis(); + final List<Date> timestamps = new ArrayList<>(); + + int iterations = 1440; + for (int i=0; i < iterations; i++) { + final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(PROCESSOR_METRICS); + snapshot.setTimestamp(new Date(startTime + i * 1000)); + timestamps.add(snapshot.getTimestamp()); + + snapshot.addStatusMetric(ProcessorStatusDescriptor.BYTES_WRITTEN.getDescriptor(), Long.valueOf(i)); + + buffer.update(snapshot); + } + + assertEquals(bufferCapacity, buffer.size()); + + final List<StatusSnapshot> snapshots = buffer.getSnapshots(timestamps, true, PROCESSOR_METRICS); + assertEquals(iterations, snapshots.size()); + + final int expectedEmptyCount = iterations - bufferCapacity; + final long emptyCount = snapshots.stream().filter(snapshot -> snapshot instanceof EmptyStatusSnapshot).count(); + assertEquals(expectedEmptyCount, emptyCount); + + for (int i=0; i < iterations; i++) { + final StatusSnapshot snapshot = snapshots.get(i); + if (i < expectedEmptyCount) { + assertTrue("Snapshot at i=" + i + " is not an EmptyStatusSnapshot", snapshot instanceof EmptyStatusSnapshot); + } else { + assertEquals(Long.valueOf(i), snapshot.getStatusMetric(ProcessorStatusDescriptor.BYTES_WRITTEN.getDescriptor())); + assertFalse(snapshot instanceof EmptyStatusSnapshot); + } + } + } + + @Test + public void testBufferShrinks() { + // Cause buffer to grow + final int bufferCapacity = 1000; + final MetricRollingBuffer buffer = new MetricRollingBuffer(bufferCapacity); + + final long startTime = System.currentTimeMillis(); + + int iterations = 1440; + for (int i=0; i < iterations; i++) { + final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(PROCESSOR_METRICS); + snapshot.setTimestamp(new Date(startTime + i * 1000)); + + snapshot.addStatusMetric(ProcessorStatusDescriptor.BYTES_WRITTEN.getDescriptor(), Long.valueOf(i)); + buffer.update(snapshot); + } + + assertEquals(bufferCapacity, buffer.size()); + + // Expire data ensure that the buffer shrinks + final long lastTimestamp = startTime + 1440 * 1000; + buffer.expireBefore(new Date(lastTimestamp - 144_001L)); + assertEquals(144, buffer.size()); + + buffer.expireBefore(new Date(lastTimestamp - 16_001L)); + assertEquals(16, buffer.size()); + + buffer.expireBefore(new Date(lastTimestamp)); + assertEquals(0, buffer.size()); + + // Ensure that we can now properly add data again + long insertStart = lastTimestamp + 10_000L; + final List<Date> timestamps = new ArrayList<>(); + for (int i=0; i < 4; i++) { + final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(PROCESSOR_METRICS); + snapshot.setTimestamp(new Date(insertStart + i * 1000)); + timestamps.add(snapshot.getTimestamp()); + + snapshot.addStatusMetric(ProcessorStatusDescriptor.BYTES_WRITTEN.getDescriptor(), Long.valueOf(i)); + buffer.update(snapshot); + } + + assertEquals(4, buffer.size()); + final List<StatusSnapshot> snapshots = buffer.getSnapshots(timestamps, true, PROCESSOR_METRICS); + assertEquals(4, snapshots.size()); + for (int i=0; i < 4; i++) { + final StatusSnapshot snapshot = snapshots.get(i); + assertEquals(Long.valueOf(i), snapshot.getStatusMetric(ProcessorStatusDescriptor.BYTES_WRITTEN.getDescriptor())); + } + } +}
