[BEAM-11994] Update BigQueryServicesImpl to capture API_REQUEST_COUNT metrics for streaming inserts mode

runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/GcpResourceIdentifiers.java

@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+/**
+ * Helper functions to generate resource labels strings for GCP entitites These can be used on
+ * MonitoringInfo 'resource' labels. See example entities:
+ *
+ * <p>https://s.apache.org/beam-gcp-debuggability For GCP entities, populate the RESOURCE label with
+ * the aip.dev/122 format: https://google.aip.dev/122 If an official GCP format does not exist, try
+ * to use the following format. //whatever.googleapis.com/parents/{parentId}/whatevers/{whateverId}
+ */
+public class GcpResourceIdentifiers {
+
+  public static String bigQueryTable(String projectId, String datasetId, String tableId) {
+    return String.format(
+        "//bigquery.googleapis.com/projects/%s/datasets/%s/tables/%s",
+        projectId, datasetId, tableId);
+  }
+}

runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ServiceCallMetric.java

@@ -0,0 +1,107 @@
+/*
+ * 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 java.util.HashMap;
+import java.util.Map;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+
+/*
+ * Metric class which records Service API call metrics.
+ * This class will capture a request count metric for the specified
+ * request_count_urn and base_labels.
+ * When Call() is invoked the status must be provided, which will
+ * be converted to a canonical GCP status code, if possible.
+ */
+public class ServiceCallMetric {
+
+  public static final Map<Integer, String> CANONICAL_STATUS_MAP =
+      ImmutableMap.<Integer, String>builder()
+          .put(200, "ok")
+          .put(400, "out_of_range")
+          .put(401, "unauthenticated")
+          .put(403, "permission_denied")
+          .put(404, "not_found")
+          .put(409, "already_exists")
+          .put(429, "resource_exhausted")
+          .put(499, "cancelled")
+          .put(500, "internal")
+          .put(501, "not_implemented")
+          .put(503, "unavailable")
+          .put(504, "deadline_exceeded")
+          .build();
+
+  public static final String CANONICAL_STATUS_UNKNOWN = "unknown";
+
+  public static final Map<String, String> STATUS_NORMALIZATION_MAP =
+      ImmutableMap.<String, String>builder()
+          .put("outofrange", "out_of_range")
+          .put("permissiondenied", "permission_denied")
+          .put("notfound", "not_found")
+          .put("alreadyexists", "already_exists")
+          .put("resourceexhausted", "resource_exhausted")
+          .put("notimplemented", "not_implemented")
+          .put("unavailable", "unavailable")
+          .put("deadlineexceeded", "deadline_exceeded")
+          .build();
+
+  private HashMap<String, String> labels;
+  private final String requestCountUrn;
+
+  public ServiceCallMetric(String requestCountUrn, HashMap<String, String> baseLabels) {
+    this.requestCountUrn = requestCountUrn;
+    this.labels = baseLabels;
+  }
+
+  public void call(int httpStatusCode) {
+    String canonicalStatusCode = ServiceCallMetric.convertToCanonicalStatusString(httpStatusCode);
+    call(canonicalStatusCode);
+  }
+
+  public void call(String statusCode) {
+    labels.put(
+        MonitoringInfoConstants.Labels.STATUS,
+        ServiceCallMetric.convertToCanonicalStatusString(statusCode));
+    // MonitoringInfoMetricName will copy labels. So its safe to reuse this reference.
+    MonitoringInfoMetricName name = MonitoringInfoMetricName.named(requestCountUrn, labels);
+    Counter counter = LabeledMetrics.counter(name, true);
+    counter.inc();
+  }
+
+  /** Converts an http status code to a canonical GCP status code string. */
+  public static String convertToCanonicalStatusString(int httpStatusCode) {
+    return CANONICAL_STATUS_MAP.getOrDefault(httpStatusCode, CANONICAL_STATUS_UNKNOWN);
+  }
+
+  /**
+   * Converts an status code string to a canonical GCP status code string. This is used to make
+   * strings like "notFound" to "not_found". If a mapping cannot be created known, then
+   * statusCode.toLowerCase() will be returned.
+   */
+  public static String convertToCanonicalStatusString(String statusCode) {
+    if (statusCode == null) {
+      return CANONICAL_STATUS_UNKNOWN;
+    }
+    String normalizedStatus = STATUS_NORMALIZATION_MAP.get(statusCode.toLowerCase());
+    if (normalizedStatus != null) {
+      return normalizedStatus;
+    }
+    return statusCode.toLowerCase();
+  }
+}

runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/GcpResourceIdentifiersTest.java

@@ -0,0 +1,31 @@
+/*
+ * 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 org.junit.Assert;
+import org.junit.Test;
+
+public class GcpResourceIdentifiersTest {
+  @Test
+  public void testBigQueryTable() {
+    String resource = GcpResourceIdentifiers.bigQueryTable("myProject", "myDataset", "myTableId");
+    Assert.assertEquals(
+        "//bigquery.googleapis.com/projects/myProject/datasets/myDataset/tables/myTableId",
+        resource);
+  }
+}

runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/ServiceCallMetricTest.java

@@ -0,0 +1,69 @@
+/*
+ * 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 java.util.HashMap;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ServiceCallMetricTest {
+
+  @Test
+  public void testCall() {
+    // Test that its on the ProcessWideMetricContainer.
+    MetricsContainerImpl container = new MetricsContainerImpl(null);
+    MetricsEnvironment.setProcessWideContainer(container);
+
+    String urn = MonitoringInfoConstants.Urns.API_REQUEST_COUNT;
+    HashMap<String, String> labels = new HashMap<String, String>();
+    labels.put("key", "value");
+    ServiceCallMetric metric =
+        new ServiceCallMetric(MonitoringInfoConstants.Urns.API_REQUEST_COUNT, labels);
+
+    // Http Status Code
+    metric.call(200);
+    labels.put(MonitoringInfoConstants.Labels.STATUS, "ok");
+    MonitoringInfoMetricName name = MonitoringInfoMetricName.named(urn, labels);
+    Assert.assertEquals(1, (long) container.getCounter(name).getCumulative());
+
+    // Normalize the status by lower casing and mapping to a canonical name with underscores.
+    metric.call("notFound");
+    labels.put(MonitoringInfoConstants.Labels.STATUS, "not_found");
+    name = MonitoringInfoMetricName.named(urn, labels);
+    Assert.assertEquals(1, (long) container.getCounter(name).getCumulative());
+
+    // Normalize the status by lower casing and mapping to a canonical name with underscores.
+    metric.call("PERMISSIONDENIED");
+    labels.put(MonitoringInfoConstants.Labels.STATUS, "permission_denied");
+    name = MonitoringInfoMetricName.named(urn, labels);
+    Assert.assertEquals(1, (long) container.getCounter(name).getCumulative());
+
+    // Accept other string codes passed in, even if they aren't in the canonical map.
+    metric.call("something_else");
+    labels.put(MonitoringInfoConstants.Labels.STATUS, "something_else");
+    name = MonitoringInfoMetricName.named(urn, labels);
+    Assert.assertEquals(1, (long) container.getCounter(name).getCumulative());
+
+    // Map unknown numeric codes to "unknown"
+    metric.call(123);
+    labels.put(MonitoringInfoConstants.Labels.STATUS, "unknown");
+    name = MonitoringInfoMetricName.named(urn, labels);
+    Assert.assertEquals(1, (long) container.getCounter(name).getCumulative());
+  }
+}