beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From al...@apache.org
Subject [beam] branch master updated: [BEAM-11994] Update ShortIdMap's maps to only use valid keyable fields in the MonitoringInfo (urn and labels).
Date Thu, 01 Jul 2021 16:01:22 GMT
This is an automated email from the ASF dual-hosted git repository.

altay 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 7cfd330  [BEAM-11994] Update ShortIdMap's maps to only use valid keyable fields in
the MonitoringInfo (urn and labels).
     new e4d27c1  Merge pull request #14804 from ajamato/bq_metrics_process_wide_short_ids
7cfd330 is described below

commit 7cfd330e47eb04da7cd6eec314fd1ca8f82635d1
Author: Alex Amato <ajamato@google.com>
AuthorDate: Thu Apr 8 20:56:01 2021 -0700

    [BEAM-11994] Update ShortIdMap's maps to only use valid keyable fields in the MonitoringInfo
(urn and labels).
---
 .../beam/runners/core/metrics/ShortIdMap.java      |   3 +
 .../beam/runners/core/metrics/ShortIdMapTest.java  | 133 +++++++++++++++++++++
 2 files changed, 136 insertions(+)

diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ShortIdMap.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ShortIdMap.java
index 03ac54a..ef9481f 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ShortIdMap.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ShortIdMap.java
@@ -30,6 +30,9 @@ public class ShortIdMap {
 
   public synchronized String getOrCreateShortId(MonitoringInfo info) {
     Preconditions.checkNotNull(info);
+    Preconditions.checkArgument(info.getPayload().isEmpty());
+    Preconditions.checkArgument(!info.hasStartTime());
+
     String shortId = monitoringInfoMap.inverse().get(info);
     if (shortId == null) {
       shortId = "metric" + counter++;
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/ShortIdMapTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/ShortIdMapTest.java
new file mode 100644
index 0000000..0a1eb12
--- /dev/null
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/ShortIdMapTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.beam.runners.core.metrics;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.sdk.values.KV;
+import org.junit.Test;
+
+public class ShortIdMapTest {
+
+  @Test
+  public void testShortIdAssignment() throws Exception {
+    ShortIdMap shortIdMap = new ShortIdMap();
+    List<KV<String, MetricsApi.MonitoringInfo>> testCases = new ArrayList<>();
+
+    SimpleMonitoringInfoBuilder builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_DISTRIBUTION_INT64);
+    testCases.add(KV.of("metric0", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.ELEMENT_COUNT);
+    testCases.add(KV.of("metric1", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_DISTRIBUTION_DOUBLE);
+    testCases.add(KV.of("metric2", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn("TestingSentinelUrn");
+    builder.setType("TestingSentinelType");
+    testCases.add(KV.of("metric3", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.FINISH_BUNDLE_MSECS);
+    testCases.add(KV.of("metric4", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64);
+    testCases.add(KV.of("metric5", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64);
+    builder.setLabel(MonitoringInfoConstants.Labels.NAME, "metricNumber7");
+    builder.setLabel(MonitoringInfoConstants.Labels.NAMESPACE, "myNamespace");
+    builder.setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "myPtransform");
+    testCases.add(KV.of("metric6", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64);
+    builder.setLabel(MonitoringInfoConstants.Labels.NAME, "metricNumber8");
+    builder.setLabel(MonitoringInfoConstants.Labels.NAMESPACE, "myNamespace");
+    builder.setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "myPtransform");
+    testCases.add(KV.of("metric7", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.API_REQUEST_COUNT);
+    builder.setLabel(MonitoringInfoConstants.Labels.SERVICE, "BigQuery");
+    testCases.add(KV.of("metric8", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.API_REQUEST_COUNT);
+    builder.setLabel(MonitoringInfoConstants.Labels.SERVICE, "Storage");
+    testCases.add(KV.of("metric9", builder.build()));
+
+    // Validate that modifying the payload, but using the same URN/labels
+    // does not change the shortId assignment.
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_DISTRIBUTION_DOUBLE);
+    testCases.add(KV.of("metric2", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64);
+    builder.setLabel(MonitoringInfoConstants.Labels.NAME, "metricNumber7");
+    builder.setLabel(MonitoringInfoConstants.Labels.NAMESPACE, "myNamespace");
+    builder.setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "myPtransform");
+    testCases.add(KV.of("metric6", builder.build()));
+
+    // Verify each short ID is assigned properly.
+    Set<String> expectedShortIds = new HashSet<String>();
+    for (KV<String, MetricsApi.MonitoringInfo> entry : testCases) {
+      assertEquals(entry.getKey(), shortIdMap.getOrCreateShortId(entry.getValue()));
+      expectedShortIds.add(entry.getKey());
+    }
+
+    HashMap<String, MetricsApi.MonitoringInfo> actualRecoveredInfos = new HashMap<>();
+    for (String expectedShortId : expectedShortIds) {
+      actualRecoveredInfos.put(expectedShortId, shortIdMap.get(expectedShortId));
+    }
+    // Retrieve all of the MonitoringInfos by short id, and verify that the
+    // metadata (everything but the payload) matches the originals
+    assertEquals(expectedShortIds, actualRecoveredInfos.keySet());
+    for (KV<String, MetricsApi.MonitoringInfo> entry : testCases) {
+      // Clear payloads of both expected and actual before comparing
+      MetricsApi.MonitoringInfo expectedMonitoringInfo = entry.getValue();
+      MetricsApi.MonitoringInfo.Builder expected =
+          MetricsApi.MonitoringInfo.newBuilder(expectedMonitoringInfo);
+      expected.clearPayload();
+
+      MetricsApi.MonitoringInfo.Builder actual =
+          MetricsApi.MonitoringInfo.newBuilder(actualRecoveredInfos.get(entry.getKey()));
+      actual.clearPayload();
+      assertEquals(expected.build(), actual.build());
+    }
+
+    // Verify each short ID is assigned properly, in reverse.
+    for (int i = testCases.size() - 1; i > 0; i--) {
+      assertEquals(
+          testCases.get(i).getKey(), shortIdMap.getOrCreateShortId(testCases.get(i).getValue()));
+    }
+  }
+}

Mime
View raw message