mosche commented on code in PR #22157:
URL: https://github.com/apache/beam/pull/22157#discussion_r926330199
##########
runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java:
##########
@@ -33,61 +37,71 @@
import org.apache.beam.sdk.metrics.MetricResult;
import org.apache.beam.sdk.metrics.MetricResults;
import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
-import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Strings;
+import
org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Streams;
/**
- * An adapter between the {@link MetricsContainerStepMap} and Codahale's
{@link Metric} interface.
+ * An adapter between the {@link MetricsContainerStepMap} and the Dropwizard
{@link Metric}
+ * interface.
*/
-public class SparkBeamMetric implements Metric {
+public class SparkBeamMetric extends BeamMetricSet {
+
private static final String ILLEGAL_CHARACTERS = "[^A-Za-z0-9-]";
- static Map<String, ?> renderAll(MetricResults metricResults) {
- Map<String, Object> metrics = new HashMap<>();
- MetricQueryResults metricQueryResults = metricResults.allMetrics();
- for (MetricResult<Long> metricResult : metricQueryResults.getCounters()) {
- metrics.put(renderName(metricResult), metricResult.getAttempted());
+ @Override
+ public Map<String, Gauge<Double>> getValue(String prefix, MetricFilter
filter) {
+ MetricResults metricResults =
+ asAttemptedOnlyMetricResults(MetricsAccumulator.getInstance().value());
+ Map<String, Gauge<Double>> metrics = new HashMap<>();
+ MetricQueryResults allMetrics = metricResults.allMetrics();
+ for (MetricResult<Long> metricResult : allMetrics.getCounters()) {
+ putFiltered(metrics, filter, renderName(prefix, metricResult),
metricResult.getAttempted());
}
- for (MetricResult<DistributionResult> metricResult :
metricQueryResults.getDistributions()) {
+ for (MetricResult<DistributionResult> metricResult :
allMetrics.getDistributions()) {
DistributionResult result = metricResult.getAttempted();
- metrics.put(renderName(metricResult) + ".count", result.getCount());
- metrics.put(renderName(metricResult) + ".sum", result.getSum());
- metrics.put(renderName(metricResult) + ".min", result.getMin());
- metrics.put(renderName(metricResult) + ".max", result.getMax());
- metrics.put(renderName(metricResult) + ".mean", result.getMean());
+ String baseName = renderName(prefix, metricResult);
+ putFiltered(metrics, filter, baseName + ".count", result.getCount());
+ putFiltered(metrics, filter, baseName + ".sum", result.getSum());
+ putFiltered(metrics, filter, baseName + ".min", result.getMin());
+ putFiltered(metrics, filter, baseName + ".max", result.getMax());
+ putFiltered(metrics, filter, baseName + ".mean", result.getMean());
}
- for (MetricResult<GaugeResult> metricResult :
metricQueryResults.getGauges()) {
- metrics.put(renderName(metricResult),
metricResult.getAttempted().getValue());
+ for (MetricResult<GaugeResult> metricResult : allMetrics.getGauges()) {
+ putFiltered(
+ metrics,
+ filter,
+ renderName(prefix, metricResult),
+ metricResult.getAttempted().getValue());
}
return metrics;
}
- Map<String, ?> renderAll() {
- MetricResults metricResults =
- asAttemptedOnlyMetricResults(MetricsAccumulator.getInstance().value());
- return renderAll(metricResults);
- }
-
@VisibleForTesting
- static String renderName(MetricResult<?> metricResult) {
+ @SuppressWarnings("nullness") // ok to have nullable elements on stream
+ static String renderName(String prefix, MetricResult<?> metricResult) {
MetricKey key = metricResult.getKey();
MetricName name = key.metricName();
- String step = key.stepName();
-
- ArrayList<String> pieces = new ArrayList<>();
+ return Streams.concat(
+ Stream.of(prefix), // prefix is not cleaned, should it be?
+ Stream.of(stripSuffix(cleanPart(key.stepName()))),
+ Stream.of(name.getNamespace(),
name.getName()).map(SparkBeamMetric::cleanPart))
+ .filter(not(Strings::isNullOrEmpty))
+ .collect(Collectors.joining("."));
+ }
- if (step != null) {
- step = step.replaceAll(ILLEGAL_CHARACTERS, "_");
- if (step.endsWith("_")) {
- step = step.substring(0, step.length() - 1);
- }
- pieces.add(step);
- }
+ private static @Nullable String cleanPart(@Nullable String str) {
Review Comment:
👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
