You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2024/02/13 15:12:14 UTC
(beam) branch master updated: Only populate DataflowHistogramValue::OutlierStats when overflow/underflow values exist. (#30246)
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 0ade62cd083 Only populate DataflowHistogramValue::OutlierStats when overflow/underflow values exist. (#30246)
0ade62cd083 is described below
commit 0ade62cd08361e8769eed1f0148234f57a7ba6f6
Author: JayajP <ja...@google.com>
AuthorDate: Tue Feb 13 07:12:07 2024 -0800
Only populate DataflowHistogramValue::OutlierStats when overflow/underflow values exist. (#30246)
* Only populate outlier stats when overflow/underflow values exist.
* Fix StreamingStepMetricsContainerTest
---
.../MetricsToPerStepNamespaceMetricsConverter.java | 73 ++++++++++++++--------
...ricsToPerStepNamespaceMetricsConverterTest.java | 9 +--
.../worker/StreamingStepMetricsContainerTest.java | 10 +--
3 files changed, 50 insertions(+), 42 deletions(-)
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java
index 7b72d650778..8f9cbd350a2 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverter.java
@@ -62,6 +62,35 @@ public class MetricsToPerStepNamespaceMetricsConverter {
.setValueInt64(value));
}
+ /**
+ * Adds {@code outlierStats} to {@code outputHistogram} if {@code inputHistogram} has recorded
+ * overflow or underflow values.
+ *
+ * @param inputHistogram
+ * @param outputHistogram
+ */
+ private static void addOutlierStatsToHistogram(
+ HistogramData inputHistogram, DataflowHistogramValue outputHistogram) {
+ long overflowCount = inputHistogram.getTopBucketCount();
+ long underflowCount = inputHistogram.getBottomBucketCount();
+ if (underflowCount == 0 && overflowCount == 0) {
+ return;
+ }
+
+ OutlierStats outlierStats = new OutlierStats();
+ if (underflowCount > 0) {
+ outlierStats
+ .setUnderflowCount(underflowCount)
+ .setUnderflowMean(inputHistogram.getBottomBucketMean());
+ }
+ if (overflowCount > 0) {
+ outlierStats
+ .setOverflowCount(overflowCount)
+ .setOverflowMean(inputHistogram.getTopBucketMean());
+ }
+ outputHistogram.setOutlierStats(outlierStats);
+ }
+
/**
* @param metricName The {@link MetricName} that represents this Histogram.
* @param value The histogram value. Currently we only support converting histograms that use
@@ -70,8 +99,8 @@ public class MetricsToPerStepNamespaceMetricsConverter {
* Otherwise returns an empty optional.
*/
private static Optional<MetricValue> convertHistogramToMetricValue(
- MetricName metricName, HistogramData value) {
- if (value.getTotalCount() == 0L) {
+ MetricName metricName, HistogramData inputHistogram) {
+ if (inputHistogram.getTotalCount() == 0L) {
return Optional.empty();
}
@@ -81,32 +110,33 @@ public class MetricsToPerStepNamespaceMetricsConverter {
return Optional.empty();
}
- DataflowHistogramValue histogramValue = new DataflowHistogramValue();
- int numberOfBuckets = value.getBucketType().getNumBuckets();
+ DataflowHistogramValue outputHistogram = new DataflowHistogramValue();
+ int numberOfBuckets = inputHistogram.getBucketType().getNumBuckets();
- if (value.getBucketType() instanceof HistogramData.LinearBuckets) {
- HistogramData.LinearBuckets buckets = (HistogramData.LinearBuckets) value.getBucketType();
+ if (inputHistogram.getBucketType() instanceof HistogramData.LinearBuckets) {
+ HistogramData.LinearBuckets buckets =
+ (HistogramData.LinearBuckets) inputHistogram.getBucketType();
Linear linearOptions =
new Linear()
.setNumberOfBuckets(numberOfBuckets)
.setWidth(buckets.getWidth())
.setStart(buckets.getStart());
- histogramValue.setBucketOptions(new BucketOptions().setLinear(linearOptions));
- } else if (value.getBucketType() instanceof HistogramData.ExponentialBuckets) {
+ outputHistogram.setBucketOptions(new BucketOptions().setLinear(linearOptions));
+ } else if (inputHistogram.getBucketType() instanceof HistogramData.ExponentialBuckets) {
HistogramData.ExponentialBuckets buckets =
- (HistogramData.ExponentialBuckets) value.getBucketType();
+ (HistogramData.ExponentialBuckets) inputHistogram.getBucketType();
Base2Exponent expoenntialOptions =
new Base2Exponent().setNumberOfBuckets(numberOfBuckets).setScale(buckets.getScale());
- histogramValue.setBucketOptions(new BucketOptions().setExponential(expoenntialOptions));
+ outputHistogram.setBucketOptions(new BucketOptions().setExponential(expoenntialOptions));
} else {
return Optional.empty();
}
- histogramValue.setCount(value.getTotalCount());
- List<Long> bucketCounts = new ArrayList<>(value.getBucketType().getNumBuckets());
+ outputHistogram.setCount(inputHistogram.getTotalCount());
+ List<Long> bucketCounts = new ArrayList<>(inputHistogram.getBucketType().getNumBuckets());
- for (int i = 0; i < value.getBucketType().getNumBuckets(); i++) {
- bucketCounts.add(value.getCount(i));
+ for (int i = 0; i < inputHistogram.getBucketType().getNumBuckets(); i++) {
+ bucketCounts.add(inputHistogram.getCount(i));
}
// Remove trailing 0 buckets.
@@ -117,22 +147,15 @@ public class MetricsToPerStepNamespaceMetricsConverter {
bucketCounts.remove(i);
}
- histogramValue.setBucketCounts(bucketCounts);
-
- OutlierStats outlierStats =
- new OutlierStats()
- .setOverflowCount(value.getTopBucketCount())
- .setOverflowMean(value.getTopBucketMean())
- .setUnderflowCount(value.getBottomBucketCount())
- .setUnderflowMean(value.getBottomBucketMean());
+ outputHistogram.setBucketCounts(bucketCounts);
- histogramValue.setOutlierStats(outlierStats);
+ addOutlierStatsToHistogram(inputHistogram, outputHistogram);
return Optional.of(
new MetricValue()
.setMetric(labeledName.get().getBaseName())
.setMetricLabels(labeledName.get().getMetricLabels())
- .setValueHistogram(histogramValue));
+ .setValueHistogram(outputHistogram));
}
/**
@@ -147,9 +170,9 @@ public class MetricsToPerStepNamespaceMetricsConverter {
String stepName, Map<MetricName, Long> counters, Map<MetricName, HistogramData> histograms) {
Map<String, PerStepNamespaceMetrics> metricsByNamespace = new HashMap<>();
-
for (Entry<MetricName, Long> entry : counters.entrySet()) {
MetricName metricName = entry.getKey();
+
Optional<MetricValue> metricValue = convertCounterToMetricValue(metricName, entry.getValue());
if (!metricValue.isPresent()) {
continue;
diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java
index 0a4cd06cf8c..4e5108399f6 100644
--- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java
+++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/MetricsToPerStepNamespaceMetricsConverterTest.java
@@ -258,18 +258,11 @@ public class MetricsToPerStepNamespaceMetricsConverterTest {
Linear linearOptions1 = new Linear().setNumberOfBuckets(10).setWidth(10.0).setStart(0.0);
BucketOptions bucketOptions1 = new BucketOptions().setLinear(linearOptions1);
- OutlierStats outlierStats1 =
- new OutlierStats()
- .setUnderflowCount(0L)
- .setUnderflowMean(0.0)
- .setOverflowCount(0L)
- .setOverflowMean(0.0);
DataflowHistogramValue linearHistogram1 =
new DataflowHistogramValue()
.setCount(1L)
.setBucketOptions(bucketOptions1)
- .setBucketCounts(bucketCounts1)
- .setOutlierStats(outlierStats1);
+ .setBucketCounts(bucketCounts1);
Map<String, String> histogramLabelMap = new HashMap<>();
histogramLabelMap.put("label2", "val2");
diff --git a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainerTest.java b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainerTest.java
index c586f2b8781..6aecafbb10d 100644
--- a/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainerTest.java
+++ b/runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingStepMetricsContainerTest.java
@@ -35,7 +35,6 @@ import com.google.api.services.dataflow.model.DataflowHistogramValue;
import com.google.api.services.dataflow.model.DistributionUpdate;
import com.google.api.services.dataflow.model.Linear;
import com.google.api.services.dataflow.model.MetricValue;
-import com.google.api.services.dataflow.model.OutlierStats;
import com.google.api.services.dataflow.model.PerStepNamespaceMetrics;
import java.time.Clock;
import java.time.Duration;
@@ -252,18 +251,11 @@ public class StreamingStepMetricsContainerTest {
Linear linearOptions = new Linear().setNumberOfBuckets(10).setWidth(10.0).setStart(0.0);
BucketOptions bucketOptions = new BucketOptions().setLinear(linearOptions);
- OutlierStats outlierStats =
- new OutlierStats()
- .setUnderflowCount(0L)
- .setUnderflowMean(0.0)
- .setOverflowCount(0L)
- .setOverflowMean(0.0);
DataflowHistogramValue linearHistogram =
new DataflowHistogramValue()
.setCount(1L)
.setBucketOptions(bucketOptions)
- .setBucketCounts(bucketCounts)
- .setOutlierStats(outlierStats);
+ .setBucketCounts(bucketCounts);
MetricValue expectedHistogram =
new MetricValue()