You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jo...@apache.org on 2023/11/15 18:58:46 UTC
(beam) branch master updated: Implement a handful of perworkermetrics the bigquery sink (#28903). (#29098)
This is an automated email from the ASF dual-hosted git repository.
johncasey 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 2012107a0fa Implement a handful of perworkermetrics the bigquery sink (#28903). (#29098)
2012107a0fa is described below
commit 2012107a0fa2bb3fedf1b5aedcb49445534b2dad
Author: JayajP <ja...@google.com>
AuthorDate: Wed Nov 15 10:58:36 2023 -0800
Implement a handful of perworkermetrics the bigquery sink (#28903). (#29098)
* Implement a handful of perworker bigquery sink metrics
* Spotless check
* Address comments and spotless fixes
* Spotless
---
.../apache/beam/sdk/metrics/DelegatingCounter.java | 35 ++-
.../beam/sdk/metrics/DelegatingHistogram.java | 32 ++-
.../org/apache/beam/sdk/util/FluentBackoff.java | 57 +++-
.../sdk/io/gcp/bigquery/BigQuerySinkMetrics.java | 268 ++++++++++++++++++
.../beam/sdk/io/gcp/bigquery/RetryManager.java | 48 ++++
.../gcp/bigquery/StorageApiFinalizeWritesDoFn.java | 13 +-
.../bigquery/StorageApiFlushAndFinalizeDoFn.java | 29 +-
.../bigquery/StorageApiWriteUnshardedRecords.java | 52 +++-
.../bigquery/StorageApiWritesShardedRecords.java | 47 +++-
.../beam/sdk/io/gcp/bigquery/TableDestination.java | 6 +
.../io/gcp/bigquery/BigQuerySinkMetricsTest.java | 304 +++++++++++++++++++++
.../beam/sdk/io/gcp/bigquery/RetryManagerTest.java | 21 +-
12 files changed, 886 insertions(+), 26 deletions(-)
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DelegatingCounter.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DelegatingCounter.java
index 82808d8cb63..a0b2e3b3467 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DelegatingCounter.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DelegatingCounter.java
@@ -25,14 +25,40 @@ import org.apache.beam.sdk.annotations.Internal;
public class DelegatingCounter implements Metric, Counter, Serializable {
private final MetricName name;
private final boolean processWideContainer;
+ private final boolean perWorkerCounter;
+ /**
+ * Create a {@code DelegatingCounter} with {@code perWorkerCounter} and {@code
+ * processWideContainer} set to false.
+ *
+ * @param name Metric name for this metric.
+ */
public DelegatingCounter(MetricName name) {
- this(name, false);
+ this(name, false, false);
}
+ /**
+ * Create a {@code DelegatingCounter} with {@code perWorkerCounter} set to false.
+ *
+ * @param name Metric name for this metric.
+ * @param processWideContainer Whether this Counter is stored in the ProcessWide container or the
+ * current thread's container.
+ */
public DelegatingCounter(MetricName name, boolean processWideContainer) {
+ this(name, processWideContainer, false);
+ }
+
+ /**
+ * @param name Metric name for this metric.
+ * @param processWideContainer Whether this Counter is stored in the ProcessWide container or the
+ * current thread's container.
+ * @param perWorkerCounter Whether this Counter refers to a perWorker metric or not.
+ */
+ public DelegatingCounter(
+ MetricName name, boolean processWideContainer, boolean perWorkerCounter) {
this.name = name;
this.processWideContainer = processWideContainer;
+ this.perWorkerCounter = perWorkerCounter;
}
/** Increment the counter. */
@@ -48,7 +74,12 @@ public class DelegatingCounter implements Metric, Counter, Serializable {
this.processWideContainer
? MetricsEnvironment.getProcessWideContainer()
: MetricsEnvironment.getCurrentContainer();
- if (container != null) {
+ if (container == null) {
+ return;
+ }
+ if (perWorkerCounter) {
+ container.getPerWorkerCounter(name).inc(n);
+ } else {
container.getCounter(name).inc(n);
}
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DelegatingHistogram.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DelegatingHistogram.java
index 74e3cf5719f..b877db69b0b 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DelegatingHistogram.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/DelegatingHistogram.java
@@ -27,12 +27,37 @@ public class DelegatingHistogram implements Metric, Histogram, Serializable {
private final MetricName name;
private final HistogramData.BucketType bucketType;
private final boolean processWideContainer;
+ private final boolean perWorkerHistogram;
+ /**
+ * Create a {@code DelegatingHistogram} with {@code perWorkerHistogram} set to false.
+ *
+ * @param name Metric name for this metric.
+ * @param bucketType Histogram bucketing strategy.
+ * @param processWideContainer Whether this Counter is stored in the ProcessWide container or the
+ * current thread's container.
+ */
public DelegatingHistogram(
MetricName name, HistogramData.BucketType bucketType, boolean processWideContainer) {
+ this(name, bucketType, processWideContainer, false);
+ }
+
+ /**
+ * @param name Metric name for this metric.
+ * @param bucketType Histogram bucketing strategy.
+ * @param processWideContainer Whether this Counter is stored in the ProcessWide container or the
+ * current thread's container.
+ * @param perWorkerHistogram Whether this Histogram refers to a perWorker metric or not.
+ */
+ public DelegatingHistogram(
+ MetricName name,
+ HistogramData.BucketType bucketType,
+ boolean processWideContainer,
+ boolean perWorkerHistogram) {
this.name = name;
this.bucketType = bucketType;
this.processWideContainer = processWideContainer;
+ this.perWorkerHistogram = perWorkerHistogram;
}
@Override
@@ -41,7 +66,12 @@ public class DelegatingHistogram implements Metric, Histogram, Serializable {
processWideContainer
? MetricsEnvironment.getProcessWideContainer()
: MetricsEnvironment.getCurrentContainer();
- if (container != null) {
+ if (container == null) {
+ return;
+ }
+ if (perWorkerHistogram) {
+ container.getPerWorkerHistogram(name, bucketType).update(value);
+ } else {
container.getHistogram(name, bucketType).update(value);
}
}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java
index 0d45ff0a57f..ce10885b068 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/FluentBackoff.java
@@ -19,6 +19,8 @@ package org.apache.beam.sdk.util;
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.NoOpCounter;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects;
import org.joda.time.Duration;
@@ -38,12 +40,14 @@ public final class FluentBackoff {
private static final Duration DEFAULT_MAX_BACKOFF = Duration.standardDays(1000);
private static final int DEFAULT_MAX_RETRIES = Integer.MAX_VALUE;
private static final Duration DEFAULT_MAX_CUM_BACKOFF = Duration.standardDays(1000);
+ private static final Counter DEFAULT_THROTTLED_TIME_COUNTER = NoOpCounter.getInstance();
private final double exponent;
private final Duration initialBackoff;
private final Duration maxBackoff;
private final Duration maxCumulativeBackoff;
private final int maxRetries;
+ private final Counter throttledTimeCounter;
/**
* By default the {@link BackOff} created by this builder will use exponential backoff (base
@@ -65,7 +69,8 @@ public final class FluentBackoff {
DEFAULT_MIN_BACKOFF,
DEFAULT_MAX_BACKOFF,
DEFAULT_MAX_CUM_BACKOFF,
- DEFAULT_MAX_RETRIES);
+ DEFAULT_MAX_RETRIES,
+ DEFAULT_THROTTLED_TIME_COUNTER);
/**
* Instantiates a {@link BackOff} that will obey the current configuration.
@@ -87,7 +92,12 @@ public final class FluentBackoff {
public FluentBackoff withExponent(double exponent) {
checkArgument(exponent > 0, "exponent %s must be greater than 0", exponent);
return new FluentBackoff(
- exponent, initialBackoff, maxBackoff, maxCumulativeBackoff, maxRetries);
+ exponent,
+ initialBackoff,
+ maxBackoff,
+ maxCumulativeBackoff,
+ maxRetries,
+ throttledTimeCounter);
}
/**
@@ -104,7 +114,12 @@ public final class FluentBackoff {
"initialBackoff %s must be at least 1 millisecond",
initialBackoff);
return new FluentBackoff(
- exponent, initialBackoff, maxBackoff, maxCumulativeBackoff, maxRetries);
+ exponent,
+ initialBackoff,
+ maxBackoff,
+ maxCumulativeBackoff,
+ maxRetries,
+ throttledTimeCounter);
}
/**
@@ -119,7 +134,12 @@ public final class FluentBackoff {
checkArgument(
maxBackoff.getMillis() > 0, "maxBackoff %s must be at least 1 millisecond", maxBackoff);
return new FluentBackoff(
- exponent, initialBackoff, maxBackoff, maxCumulativeBackoff, maxRetries);
+ exponent,
+ initialBackoff,
+ maxBackoff,
+ maxCumulativeBackoff,
+ maxRetries,
+ throttledTimeCounter);
}
/**
@@ -136,7 +156,12 @@ public final class FluentBackoff {
"maxCumulativeBackoff %s must be at least 1 millisecond",
maxCumulativeBackoff);
return new FluentBackoff(
- exponent, initialBackoff, maxBackoff, maxCumulativeBackoff, maxRetries);
+ exponent,
+ initialBackoff,
+ maxBackoff,
+ maxCumulativeBackoff,
+ maxRetries,
+ throttledTimeCounter);
}
/**
@@ -151,7 +176,22 @@ public final class FluentBackoff {
public FluentBackoff withMaxRetries(int maxRetries) {
checkArgument(maxRetries >= 0, "maxRetries %s cannot be negative", maxRetries);
return new FluentBackoff(
- exponent, initialBackoff, maxBackoff, maxCumulativeBackoff, maxRetries);
+ exponent,
+ initialBackoff,
+ maxBackoff,
+ maxCumulativeBackoff,
+ maxRetries,
+ throttledTimeCounter);
+ }
+
+ public FluentBackoff withThrottledTimeCounter(Counter throttledTimeCounter) {
+ return new FluentBackoff(
+ exponent,
+ initialBackoff,
+ maxBackoff,
+ maxCumulativeBackoff,
+ maxRetries,
+ throttledTimeCounter);
}
@Override
@@ -206,6 +246,7 @@ public final class FluentBackoff {
// Update state and return backoff.
currentCumulativeBackoff = currentCumulativeBackoff.plus(Duration.millis(nextBackoffMillis));
currentRetry += 1;
+ backoffConfig.throttledTimeCounter.inc(nextBackoffMillis);
return nextBackoffMillis;
}
@@ -229,11 +270,13 @@ public final class FluentBackoff {
Duration initialBackoff,
Duration maxBackoff,
Duration maxCumulativeBackoff,
- int maxRetries) {
+ int maxRetries,
+ Counter throttledTimeCounter) {
this.exponent = exponent;
this.initialBackoff = initialBackoff;
this.maxBackoff = maxBackoff;
this.maxRetries = maxRetries;
this.maxCumulativeBackoff = maxCumulativeBackoff;
+ this.throttledTimeCounter = throttledTimeCounter;
}
}
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetrics.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetrics.java
new file mode 100644
index 00000000000..24323fce689
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetrics.java
@@ -0,0 +1,268 @@
+/*
+ * 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.sdk.io.gcp.bigquery;
+
+import io.grpc.Status;
+import java.time.Instant;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.io.gcp.bigquery.RetryManager.Operation.Context;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.DelegatingCounter;
+import org.apache.beam.sdk.metrics.DelegatingHistogram;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.util.HistogramData;
+
+/**
+ * Helper class to create perworker metrics for BigQuery Sink stages.
+ *
+ * <p>In general metrics be in the namespace 'BigQuerySink' and have their name formatted as:
+ *
+ * <p>'{baseName}-{metricLabelKey1}:{metricLabelVal1};...{metricLabelKeyN}:{metricLabelValN};'
+ */
+public class BigQuerySinkMetrics {
+ private static Boolean supportMetricsDeletion = false;
+
+ private static final String METRICS_NAMESPACE = "BigQuerySink";
+
+ // Status codes
+ private static final String UNKNOWN = Status.Code.UNKNOWN.toString();
+ public static final String OK = Status.Code.OK.toString();
+ public static final String PAYLOAD_TOO_LARGE = "PayloadTooLarge";
+
+ // Base Metric names
+ private static final String RPC_REQUESTS = "RpcRequests";
+ private static final String RPC_LATENCY = "RpcLatency";
+ private static final String APPEND_ROWS_ROW_STATUS = "AppendRowsRowStatus";
+ private static final String THROTTLED_TIME = "ThrottledTime";
+
+ // StorageWriteAPI Method names
+ enum RpcMethod {
+ APPEND_ROWS,
+ FLUSH_ROWS,
+ FINALIZE_STREAM
+ }
+
+ // Status of a BigQuery row from the AppendRows RPC call.
+ enum RowStatus {
+ SUCCESSFUL,
+ RETRIED,
+ FAILED
+ }
+
+ // Metric labels
+ private static final String TABLE_ID_LABEL = "TableId";
+ private static final String RPC_STATUS_LABEL = "RpcStatus";
+ private static final String RPC_METHOD = "Method";
+ private static final String ROW_STATUS = "RowStatus";
+
+ // Delimiters
+ private static final char LABEL_DELIMITER = ';';
+ private static final char METRIC_KV_DELIMITER = ':';
+ private static final char METRIC_NAME_DELIMITER = '-';
+
+ /**
+ * Returns a metric name that merges the baseName with metricLables formatted as.
+ *
+ * <p>'{baseName}-{metricLabelKey1}:{metricLabelVal1};...{metricLabelKeyN}:{metricLabelValN};'
+ */
+ private static String createLabeledMetricName(
+ String baseName, NavigableMap<String, String> metricLabels) {
+ StringBuilder nameBuilder = new StringBuilder(baseName + METRIC_NAME_DELIMITER);
+
+ metricLabels.forEach(
+ (labelKey, labelVal) ->
+ nameBuilder.append(labelKey + METRIC_KV_DELIMITER + labelVal + LABEL_DELIMITER));
+ return nameBuilder.toString();
+ }
+
+ /**
+ * @param method StorageWriteAPI method associated with this metric.
+ * @param rpcStatus RPC return status.
+ * @param tableId Table pertaining to the write method. Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @return Counter in namespace BigQuerySink and name
+ * 'RpcRequests-Method:{method}RpcStatus:{status};TableId:{tableId}' TableId label is dropped
+ * if 'supportsMetricsDeletion' is not enabled.
+ */
+ private static Counter createRPCRequestCounter(
+ RpcMethod method, String rpcStatus, String tableId) {
+ NavigableMap<String, String> metricLabels = new TreeMap<String, String>();
+ metricLabels.put(RPC_STATUS_LABEL, rpcStatus);
+ metricLabels.put(RPC_METHOD, method.toString());
+ if (BigQuerySinkMetrics.supportMetricsDeletion) {
+ metricLabels.put(TABLE_ID_LABEL, tableId);
+ }
+
+ String fullMetricName = createLabeledMetricName(RPC_REQUESTS, metricLabels);
+ MetricName metricName = MetricName.named(METRICS_NAMESPACE, fullMetricName);
+ return new DelegatingCounter(metricName, false, true);
+ }
+
+ /**
+ * Creates an Histogram metric to record RPC latency. Metric will have name.
+ *
+ * <p>'RpcLatency-Method:{method};'
+ *
+ * @param method StorageWriteAPI method associated with this metric.
+ * @return Histogram with exponential buckets with a sqrt(2) growth factor.
+ */
+ private static Histogram createRPCLatencyHistogram(RpcMethod method) {
+ NavigableMap<String, String> metricLabels = new TreeMap<String, String>();
+ metricLabels.put(RPC_METHOD, method.toString());
+ String fullMetricName = createLabeledMetricName(RPC_LATENCY, metricLabels);
+ MetricName metricName = MetricName.named(METRICS_NAMESPACE, fullMetricName);
+
+ HistogramData.BucketType buckets = HistogramData.ExponentialBuckets.of(1, 34);
+
+ return new DelegatingHistogram(metricName, buckets, false, true);
+ }
+
+ /**
+ * Records an RPC operation's duration in a PerWorkerHistogram.
+ *
+ * @param c Retry manager context, used to get the operation start and end time.
+ * @param method StorageWriteAPI write method.
+ */
+ private static void updateRpcLatencyMetric(@Nonnull Context<?> c, RpcMethod method) {
+ @Nullable Instant operationStartTime = c.getOperationStartTime();
+ @Nullable Instant operationEndTime = c.getOperationEndTime();
+ if (operationStartTime == null || operationEndTime == null) {
+ return;
+ }
+ long timeElapsed = java.time.Duration.between(operationStartTime, operationEndTime).toMillis();
+ if (timeElapsed > 0) {
+ BigQuerySinkMetrics.createRPCLatencyHistogram(method).update(timeElapsed);
+ }
+ }
+
+ /**
+ * @param rowStatus Status of these BigQuery rows.
+ * @param rpcStatus rpcStatus
+ * @param tableId Table pertaining to the write method. Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ * @return Metric that tracks the status of BigQuery rows after making an AppendRows RPC call.
+ */
+ public static Counter appendRowsRowStatusCounter(
+ RowStatus rowStatus, String rpcStatus, String tableId) {
+ NavigableMap<String, String> metricLabels = new TreeMap<String, String>();
+ metricLabels.put(RPC_STATUS_LABEL, rpcStatus);
+ metricLabels.put(ROW_STATUS, rowStatus.toString());
+ if (BigQuerySinkMetrics.supportMetricsDeletion) {
+ metricLabels.put(TABLE_ID_LABEL, tableId);
+ }
+
+ String fullMetricName = createLabeledMetricName(APPEND_ROWS_ROW_STATUS, metricLabels);
+ MetricName metricName = MetricName.named(METRICS_NAMESPACE, fullMetricName);
+ return new DelegatingCounter(metricName, false, true);
+ }
+
+ /**
+ * @param method StorageWriteAPI write method.
+ * @return Counter that tracks throttled time due to RPC retries.
+ */
+ public static Counter throttledTimeCounter(RpcMethod method) {
+ NavigableMap<String, String> metricLabels = new TreeMap<String, String>();
+ metricLabels.put(RPC_METHOD, method.toString());
+ String fullMetricName = createLabeledMetricName(THROTTLED_TIME, metricLabels);
+ MetricName metricName = MetricName.named(METRICS_NAMESPACE, fullMetricName);
+
+ return new DelegatingCounter(metricName, false, true);
+ }
+
+ /**
+ * Converts a Throwable to a gRPC Status code.
+ *
+ * @param t Throwable.
+ * @return gRPC status code string or 'UNKNOWN' if 't' is null or does not map to a gRPC error.
+ */
+ public static String throwableToGRPCCodeString(@Nullable Throwable t) {
+ if (t == null) {
+ return BigQuerySinkMetrics.UNKNOWN;
+ }
+ return Status.fromThrowable(t).getCode().toString();
+ }
+
+ /**
+ * Records RpcRequests counter and RpcLatency histogram for this RPC call. If
+ * 'SupportMetricsDeletion' is enabled, RpcRequests counter will have tableId label set to {@code
+ * UNKNOWN}. RpcRequets counter will have RpcStatus label set to {@code OK}.
+ *
+ * @param c Context of successful RPC call.
+ * @param method StorageWriteAPI method associated with this metric.
+ */
+ public static void reportSuccessfulRpcMetrics(@Nullable Context<?> c, RpcMethod method) {
+ reportSuccessfulRpcMetrics(c, method, UNKNOWN);
+ }
+
+ /**
+ * Records RpcRequests counter and RpcLatency histogram for this RPC call. RpcRequets counter will
+ * have RpcStatus label set to {@code OK}.
+ *
+ * @param c Context of successful RPC call.
+ * @param method StorageWriteAPI method associated with this metric.
+ * @param tableId Table pertaining to the write method. Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ public static void reportSuccessfulRpcMetrics(
+ @Nullable Context<?> c, RpcMethod method, String tableId) {
+ if (c == null) {
+ return;
+ }
+ createRPCRequestCounter(method, OK, tableId).inc(1);
+ updateRpcLatencyMetric(c, method);
+ }
+
+ /**
+ * Records RpcRequests counter and RpcLatency histogram for this RPC call. If
+ * 'SupportMetricsDeletion' is enabled, RpcRequests counter will have tableId label set to {@code
+ * UNKNOWN}. RpcRequets counter will have a RpcStatus label set from the gRPC error.
+ *
+ * @param c Context of successful RPC call.
+ * @param method StorageWriteAPI method associated with this metric.
+ */
+ public static void reportFailedRPCMetrics(@Nullable Context<?> c, RpcMethod method) {
+ reportFailedRPCMetrics(c, method, UNKNOWN);
+ }
+
+ /**
+ * Records RpcRequests counter and RpcLatency histogram for this RPC call. RpcRequets counter will
+ * have a RpcStatus label set from the gRPC error.
+ *
+ * @param c Context of successful RPC call.
+ * @param method StorageWriteAPI method associated with this metric.
+ * @param tableId Table pertaining to the write method. Only included in the metric key if
+ * 'supportsMetricsDeletion' is enabled.
+ */
+ public static void reportFailedRPCMetrics(
+ @Nullable Context<?> c, RpcMethod method, String tableId) {
+ if (c == null) {
+ return;
+ }
+ String statusCode = throwableToGRPCCodeString(c.getError());
+ createRPCRequestCounter(method, statusCode, tableId).inc(1);
+ updateRpcLatencyMetric(c, method);
+ }
+
+ public static void setSupportMetricsDeletion(Boolean supportMetricsDeletion) {
+ BigQuerySinkMetrics.supportMetricsDeletion = supportMetricsDeletion;
+ }
+}
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/RetryManager.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/RetryManager.java
index 245d0528423..ae2eafa7a2b 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/RetryManager.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/RetryManager.java
@@ -22,6 +22,7 @@ import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Pr
import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
+import java.time.Instant;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
@@ -33,6 +34,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.beam.sdk.io.gcp.bigquery.RetryManager.Operation.Context;
+import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.util.BackOff;
import org.apache.beam.sdk.util.BackOffUtils;
import org.apache.beam.sdk.util.FluentBackoff;
@@ -85,10 +87,24 @@ class RetryManager<ResultT, ContextT extends Context<ResultT>> {
.backoff();
}
+ RetryManager(
+ Duration initialBackoff, Duration maxBackoff, int maxRetries, Counter throttledTimeCounter) {
+ this.operations = Queues.newArrayDeque();
+ backoff =
+ FluentBackoff.DEFAULT
+ .withInitialBackoff(initialBackoff)
+ .withMaxBackoff(maxBackoff)
+ .withMaxRetries(maxRetries)
+ .withThrottledTimeCounter(throttledTimeCounter)
+ .backoff();
+ }
+
static class Operation<ResultT, ContextT extends Context<ResultT>> {
static class Context<ResultT> {
private @Nullable Throwable error = null;
private @Nullable ResultT result = null;
+ private @Nullable Instant operationStartTime = null;
+ private @Nullable Instant operationEndTime = null;
public void setError(@Nullable Throwable error) {
this.error = error;
@@ -105,6 +121,22 @@ class RetryManager<ResultT, ContextT extends Context<ResultT>> {
public @Nullable ResultT getResult() {
return result;
}
+
+ public void setOperationStartTime(@Nullable Instant operationStartTime) {
+ this.operationStartTime = operationStartTime;
+ }
+
+ public @Nullable Instant getOperationStartTime() {
+ return operationStartTime;
+ }
+
+ public void setOperationEndTime(@Nullable Instant operationEndTime) {
+ this.operationEndTime = operationEndTime;
+ }
+
+ public @Nullable Instant getOperationEndTime() {
+ return operationEndTime;
+ }
}
private final Function<ContextT, ApiFuture<ResultT>> runOperation;
@@ -129,6 +161,8 @@ class RetryManager<ResultT, ContextT extends Context<ResultT>> {
}
void run(Executor executor) {
+ this.context.setOperationStartTime(Instant.now());
+ this.context.setOperationEndTime(null);
this.future = runOperation.apply(context);
this.callback = new Callback<>(hasSucceeded);
ApiFutures.addCallback(future, callback, executor);
@@ -146,6 +180,7 @@ class RetryManager<ResultT, ContextT extends Context<ResultT>> {
private final Function<ResultT, Boolean> hasSucceeded;
@Nullable private Throwable failure = null;
boolean failed = false;
+ @Nullable Instant operationEndTime = null;
Callback(Function<ResultT, Boolean> hasSucceeded) {
this.waiter = new CountDownLatch(1);
@@ -163,6 +198,7 @@ class RetryManager<ResultT, ContextT extends Context<ResultT>> {
@Override
public void onFailure(Throwable t) {
synchronized (this) {
+ operationEndTime = Instant.now();
failure = t;
failed = true;
}
@@ -172,6 +208,7 @@ class RetryManager<ResultT, ContextT extends Context<ResultT>> {
@Override
public void onSuccess(ResultT result) {
synchronized (this) {
+ operationEndTime = Instant.now();
if (hasSucceeded.apply(result)) {
failure = null;
} else {
@@ -194,6 +231,13 @@ class RetryManager<ResultT, ContextT extends Context<ResultT>> {
return failed;
}
}
+
+ @Nullable
+ Instant getOperationEndTime() {
+ synchronized (this) {
+ return operationEndTime;
+ }
+ }
}
void addOperation(
@@ -256,6 +300,10 @@ class RetryManager<ResultT, ContextT extends Context<ResultT>> {
while (!this.operations.isEmpty()) {
Operation<ResultT, ContextT> operation = this.operations.element();
boolean failed = operation.await();
+ @Nullable Callback<?> callback = operation.callback;
+ if (callback != null) {
+ operation.context.setOperationEndTime(callback.getOperationEndTime());
+ }
if (failed) {
Throwable failure = Preconditions.checkStateNotNull(operation.callback).getFailure();
operation.context.setError(failure);
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFinalizeWritesDoFn.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFinalizeWritesDoFn.java
index 6ce58a10ef8..853a63fca8c 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFinalizeWritesDoFn.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFinalizeWritesDoFn.java
@@ -104,7 +104,12 @@ class StorageApiFinalizeWritesDoFn extends DoFn<KV<String, String>, Void> {
DatasetService datasetService = getDatasetService(pipelineOptions);
RetryManager<FinalizeWriteStreamResponse, Context<FinalizeWriteStreamResponse>> retryManager =
- new RetryManager<>(Duration.standardSeconds(1), Duration.standardMinutes(1), 3);
+ new RetryManager<>(
+ Duration.standardSeconds(1),
+ Duration.standardMinutes(1),
+ 3,
+ BigQuerySinkMetrics.throttledTimeCounter(
+ BigQuerySinkMetrics.RpcMethod.FINALIZE_STREAM));
retryManager.addOperation(
c -> {
finalizeOperationsSent.inc();
@@ -115,6 +120,9 @@ class StorageApiFinalizeWritesDoFn extends DoFn<KV<String, String>, Void> {
Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null));
LOG.error("Finalize of stream " + streamId + " failed with " + firstContext.getError());
finalizeOperationsFailed.inc();
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ firstContext, BigQuerySinkMetrics.RpcMethod.FINALIZE_STREAM);
+
return RetryType.RETRY_ALL_OPERATIONS;
},
c -> {
@@ -126,6 +134,9 @@ class StorageApiFinalizeWritesDoFn extends DoFn<KV<String, String>, Void> {
rowsFinalized.inc(response.getRowCount());
finalizeOperationsSucceeded.inc();
+ BigQuerySinkMetrics.reportSuccessfulRpcMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.FINALIZE_STREAM);
+
commitStreams.computeIfAbsent(tableId, d -> Lists.newArrayList()).add(streamId);
},
new Context<>());
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFlushAndFinalizeDoFn.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFlushAndFinalizeDoFn.java
index bb2bfba85fb..333a0c0b36b 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFlushAndFinalizeDoFn.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiFlushAndFinalizeDoFn.java
@@ -142,7 +142,11 @@ public class StorageApiFlushAndFinalizeDoFn extends DoFn<KV<String, Operation>,
if (offset >= 0) {
Instant now = Instant.now();
RetryManager<FlushRowsResponse, Context<FlushRowsResponse>> retryManager =
- new RetryManager<>(Duration.standardSeconds(1), Duration.standardMinutes(1), 3);
+ new RetryManager<>(
+ Duration.standardSeconds(1),
+ Duration.standardMinutes(1),
+ 3,
+ BigQuerySinkMetrics.throttledTimeCounter(BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS));
retryManager.addOperation(
// runOperation
c -> {
@@ -155,11 +159,15 @@ public class StorageApiFlushAndFinalizeDoFn extends DoFn<KV<String, Operation>,
},
// onError
contexts -> {
- Throwable error =
- Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null)).getError();
+ Context<FlushRowsResponse> failedContext =
+ Preconditions.checkArgumentNotNull(Iterables.getFirst(contexts, null));
+ Throwable error = failedContext.getError();
LOG.warn(
"Flush of stream " + streamId + " to offset " + offset + " failed with " + error);
flushOperationsFailed.inc();
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ failedContext, BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS);
+
if (error instanceof ApiException) {
Code statusCode = ((ApiException) error).getStatusCode().getCode();
if (statusCode.equals(Code.ALREADY_EXISTS)) {
@@ -181,6 +189,8 @@ public class StorageApiFlushAndFinalizeDoFn extends DoFn<KV<String, Operation>,
},
// onSuccess
c -> {
+ BigQuerySinkMetrics.reportSuccessfulRpcMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.FLUSH_ROWS);
flushOperationsSucceeded.inc();
},
new Context<>());
@@ -198,10 +208,16 @@ public class StorageApiFlushAndFinalizeDoFn extends DoFn<KV<String, Operation>,
// or we would end up with duplicates.
if (operation.finalizeStream) {
RetryManager<FinalizeWriteStreamResponse, Context<FinalizeWriteStreamResponse>> retryManager =
- new RetryManager<>(Duration.standardSeconds(1), Duration.standardMinutes(1), 3);
+ new RetryManager<>(
+ Duration.standardSeconds(1),
+ Duration.standardMinutes(1),
+ 3,
+ BigQuerySinkMetrics.throttledTimeCounter(
+ BigQuerySinkMetrics.RpcMethod.FINALIZE_STREAM));
retryManager.addOperation(
c -> {
finalizeOperationsSent.inc();
+
return datasetService.finalizeWriteStream(streamId);
},
contexts -> {
@@ -214,7 +230,10 @@ public class StorageApiFlushAndFinalizeDoFn extends DoFn<KV<String, Operation>,
finalizeOperationsFailed.inc();
@Nullable
Context<FinalizeWriteStreamResponse> firstContext = Iterables.getFirst(contexts, null);
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ firstContext, BigQuerySinkMetrics.RpcMethod.FINALIZE_STREAM);
@Nullable Throwable error = firstContext == null ? null : firstContext.getError();
+
if (error instanceof ApiException) {
Code statusCode = ((ApiException) error).getStatusCode().getCode();
if (statusCode.equals(Code.NOT_FOUND)) {
@@ -224,6 +243,8 @@ public class StorageApiFlushAndFinalizeDoFn extends DoFn<KV<String, Operation>,
return RetryType.RETRY_ALL_OPERATIONS;
},
r -> {
+ BigQuerySinkMetrics.reportSuccessfulRpcMetrics(
+ r, BigQuerySinkMetrics.RpcMethod.FINALIZE_STREAM);
finalizeOperationsSucceeded.inc();
},
new Context<>());
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java
index 21c2a485c27..444d87f62da 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java
@@ -265,6 +265,7 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
class DestinationState {
private final String tableUrn;
+ private final String shortTableUrn;
private String streamName = "";
private @Nullable AppendClientInfo appendClientInfo = null;
private long currentOffset = 0;
@@ -295,6 +296,7 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
public DestinationState(
String tableUrn,
+ String shortTableUrn,
MessageConverter<ElementT> messageConverter,
DatasetService datasetService,
boolean useDefaultStream,
@@ -305,6 +307,7 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
boolean includeCdcColumns)
throws Exception {
this.tableUrn = tableUrn;
+ this.shortTableUrn = shortTableUrn;
this.pendingMessages = Lists.newArrayList();
this.pendingTimestamps = Lists.newArrayList();
this.maybeDatasetService = datasetService;
@@ -608,7 +611,13 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
failedRow, "Row payload too large. Maximum size " + maxRequestSize),
timestamp);
}
- rowsSentToFailedRowsCollection.inc(inserts.getSerializedRowsCount());
+ int numRowsFailed = inserts.getSerializedRowsCount();
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.FAILED,
+ BigQuerySinkMetrics.PAYLOAD_TOO_LARGE,
+ shortTableUrn)
+ .inc(numRowsFailed);
+ rowsSentToFailedRowsCollection.inc(numRowsFailed);
return 0;
}
@@ -650,11 +659,17 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
contexts -> {
AppendRowsContext failedContext =
Preconditions.checkStateNotNull(Iterables.getFirst(contexts, null));
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ failedContext, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, shortTableUrn);
+ String errorCode =
+ BigQuerySinkMetrics.throwableToGRPCCodeString(failedContext.getError());
+
if (failedContext.getError() != null
&& failedContext.getError() instanceof Exceptions.AppendSerializtionError) {
Exceptions.AppendSerializtionError error =
Preconditions.checkStateNotNull(
(Exceptions.AppendSerializtionError) failedContext.getError());
+
Set<Integer> failedRowIndices = error.getRowIndexToErrorMessage().keySet();
for (int failedIndex : failedRowIndices) {
// Convert the message to a TableRow and send it to the failedRows collection.
@@ -677,7 +692,11 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
LOG.error("Failed to insert row and could not parse the result!", e);
}
}
- rowsSentToFailedRowsCollection.inc(failedRowIndices.size());
+ int numRowsFailed = failedRowIndices.size();
+ rowsSentToFailedRowsCollection.inc(numRowsFailed);
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.FAILED, errorCode, shortTableUrn)
+ .inc(numRowsFailed);
// Remove the failed row from the payload, so we retry the batch without the failed
// rows.
@@ -692,6 +711,10 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
}
failedContext.protoRows = retryRows.build();
failedContext.timestamps = retryTimestamps;
+ int numRowsRetried = failedContext.protoRows.getSerializedRowsCount();
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.RETRIED, errorCode, shortTableUrn)
+ .inc(numRowsRetried);
// Since we removed rows, we need to update the insert offsets for all remaining
// rows.
@@ -762,11 +785,26 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
throw new RuntimeException(e);
}
+ int numRowsRetried = failedContext.protoRows.getSerializedRowsCount();
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.RETRIED, errorCode, shortTableUrn)
+ .inc(numRowsRetried);
+
appendFailures.inc();
return RetryType.RETRY_ALL_OPERATIONS;
},
c -> {
- recordsAppended.inc(c.protoRows.getSerializedRowsCount());
+ int numRecordsAppended = c.protoRows.getSerializedRowsCount();
+ recordsAppended.inc(numRecordsAppended);
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.SUCCESSFUL,
+ BigQuerySinkMetrics.OK,
+ shortTableUrn)
+ .inc(numRecordsAppended);
+
+ BigQuerySinkMetrics.reportSuccessfulRpcMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, shortTableUrn);
+
if (successfulRowsReceiver != null) {
for (int i = 0; i < c.protoRows.getSerializedRowsCount(); ++i) {
ByteString rowBytes = c.protoRows.getSerializedRowsList().get(i);
@@ -903,7 +941,12 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
for (DestinationState destinationState :
Preconditions.checkStateNotNull(destinations).values()) {
RetryManager<AppendRowsResponse, AppendRowsContext> retryManager =
- new RetryManager<>(Duration.standardSeconds(1), Duration.standardSeconds(10), 1000);
+ new RetryManager<>(
+ Duration.standardSeconds(1),
+ Duration.standardSeconds(10),
+ 1000,
+ BigQuerySinkMetrics.throttledTimeCounter(
+ BigQuerySinkMetrics.RpcMethod.APPEND_ROWS));
retryManagers.add(retryManager);
numRowsWritten +=
destinationState.flush(retryManager, failedRowsReceiver, successfulRowsReceiver);
@@ -974,6 +1017,7 @@ public class StorageApiWriteUnshardedRecords<DestinationT, ElementT>
messageConverter = messageConverters.get(destination, dynamicDestinations, datasetService);
return new DestinationState(
tableDestination1.getTableUrn(bigQueryOptions),
+ tableDestination1.getShortTableUrn(),
messageConverter,
datasetService,
useDefaultStream,
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java
index f4982396e9d..d3042984638 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java
@@ -440,6 +440,7 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
return tableDestination1;
});
final String tableId = tableDestination.getTableUrn(bigQueryOptions);
+ final String shortTableId = tableDestination.getShortTableUrn();
final DatasetService datasetService = getDatasetService(pipelineOptions);
Coder<DestinationT> destinationCoder = dynamicDestinations.getDestinationCoder();
@@ -545,6 +546,11 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
new BigQueryStorageApiInsertError(failedRow.getValue(), errorMessage),
failedRow.getTimestamp());
rowsSentToFailedRowsCollection.inc();
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.FAILED,
+ BigQuerySinkMetrics.PAYLOAD_TOO_LARGE,
+ shortTableId)
+ .inc(1);
},
autoUpdateSchema,
ignoreUnknownValues,
@@ -628,6 +634,10 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
// The first context is always the one that fails.
AppendRowsContext failedContext =
Preconditions.checkStateNotNull(Iterables.getFirst(failedContexts, null));
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ failedContext, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS);
+ String errorCode =
+ BigQuerySinkMetrics.throwableToGRPCCodeString(failedContext.getError());
// AppendSerializationError means that BigQuery detected errors on individual rows, e.g.
// a row not conforming
@@ -639,6 +649,7 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
Exceptions.AppendSerializtionError error =
Preconditions.checkArgumentNotNull(
(Exceptions.AppendSerializtionError) failedContext.getError());
+
Set<Integer> failedRowIndices = error.getRowIndexToErrorMessage().keySet();
for (int failedIndex : failedRowIndices) {
// Convert the message to a TableRow and send it to the failedRows collection.
@@ -651,7 +662,11 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
failedRow, error.getRowIndexToErrorMessage().get(failedIndex)),
timestamp);
}
- rowsSentToFailedRowsCollection.inc(failedRowIndices.size());
+ int failedRows = failedRowIndices.size();
+ rowsSentToFailedRowsCollection.inc(failedRows);
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.FAILED, errorCode, shortTableId)
+ .inc(failedRows);
// Remove the failed row from the payload, so we retry the batch without the failed
// rows.
@@ -666,6 +681,10 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
}
failedContext.protoRows = retryRows.build();
failedContext.timestamps = timestamps;
+ int retriedRows = failedContext.protoRows.getSerializedRowsCount();
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.RETRIED, errorCode, shortTableId)
+ .inc(retriedRows);
// Since we removed rows, we need to update the insert offsets for all remaining rows.
long offset = failedContext.offset;
@@ -679,6 +698,7 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
Throwable error = Preconditions.checkStateNotNull(failedContext.getError());
Status.Code statusCode = Status.fromThrowable(error).getCode();
+
// This means that the offset we have stored does not match the current end of
// the stream in the Storage API. Usually this happens because a crash or a bundle
// failure
@@ -711,6 +731,10 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
clearClients.accept(failedContexts);
}
appendFailures.inc();
+ int retriedRows = failedContext.protoRows.getSerializedRowsCount();
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.RETRIED, errorCode, shortTableId)
+ .inc(retriedRows);
boolean explicitStreamFinalized =
failedContext.getError() instanceof StreamFinalizedException;
@@ -755,7 +779,12 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
new Operation(
context.offset + context.protoRows.getSerializedRowsCount() - 1,
false)));
- flushesScheduled.inc(context.protoRows.getSerializedRowsCount());
+ int flushedRows = context.protoRows.getSerializedRowsCount();
+ flushesScheduled.inc(flushedRows);
+ BigQuerySinkMetrics.reportSuccessfulRpcMetrics(
+ context, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, shortTableId);
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.SUCCESSFUL, BigQuerySinkMetrics.OK, shortTableId);
if (successfulRowsTag != null) {
for (int i = 0; i < context.protoRows.getSerializedRowsCount(); ++i) {
@@ -769,7 +798,11 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
Instant now = Instant.now();
List<AppendRowsContext> contexts = Lists.newArrayList();
RetryManager<AppendRowsResponse, AppendRowsContext> retryManager =
- new RetryManager<>(Duration.standardSeconds(1), Duration.standardSeconds(10), 1000);
+ new RetryManager<>(
+ Duration.standardSeconds(1),
+ Duration.standardSeconds(10),
+ 1000,
+ BigQuerySinkMetrics.throttledTimeCounter(BigQuerySinkMetrics.RpcMethod.APPEND_ROWS));
int numAppends = 0;
for (SplittingIterable.Value splitValue : messages) {
// Handle the case of a row that is too large.
@@ -794,7 +827,13 @@ public class StorageApiWritesShardedRecords<DestinationT extends @NonNull Object
failedRow, "Row payload too large. Maximum size " + maxRequestSize),
timestamp);
}
- rowsSentToFailedRowsCollection.inc(splitValue.getProtoRows().getSerializedRowsCount());
+ int numRowsFailed = splitValue.getProtoRows().getSerializedRowsCount();
+ rowsSentToFailedRowsCollection.inc(numRowsFailed);
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.FAILED,
+ BigQuerySinkMetrics.PAYLOAD_TOO_LARGE,
+ shortTableId)
+ .inc(numRowsFailed);
} else {
++numAppends;
// RetryManager
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestination.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestination.java
index 3f9f1b750cc..df96e1bc226 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestination.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableDestination.java
@@ -136,6 +136,12 @@ public class TableDestination implements Serializable {
table.getProjectId(), table.getDatasetId(), table.getTableId());
}
+ /** Return shortened tablespec in datasets/[dataset]/tables/[table] format. */
+ public String getShortTableUrn() {
+ TableReference table = getTableReference();
+ return String.format("datasets/%s/tables/%s", table.getDatasetId(), table.getTableId());
+ }
+
public TableReference getTableReference() {
return BigQueryHelpers.parseTableSpec(tableSpec);
}
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetricsTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetricsTest.java
new file mode 100644
index 00000000000..c937c6b299b
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQuerySinkMetricsTest.java
@@ -0,0 +1,304 @@
+/*
+ * 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.sdk.io.gcp.bigquery;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.equalTo;
+
+import com.google.cloud.bigquery.storage.v1.AppendRowsResponse;
+import com.google.cloud.bigquery.storage.v1.Exceptions;
+import io.grpc.Status;
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.beam.runners.core.metrics.CounterCell;
+import org.apache.beam.runners.core.metrics.MetricsContainerImpl;
+import org.apache.beam.sdk.io.gcp.bigquery.RetryManager.Operation.Context;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Histogram;
+import org.apache.beam.sdk.metrics.MetricName;
+import org.apache.beam.sdk.metrics.MetricsEnvironment;
+import org.apache.beam.sdk.util.HistogramData;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
+import org.hamcrest.collection.IsMapContaining;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link BigQuerySinkMetrics}. */
+@RunWith(JUnit4.class)
+public class BigQuerySinkMetricsTest {
+
+ public static class TestHistogram implements Histogram {
+ public List<Double> values = Lists.newArrayList();
+ private MetricName metricName = MetricName.named("namespace", "name");
+
+ @Override
+ public void update(double value) {
+ values.add(value);
+ }
+
+ @Override
+ public MetricName getName() {
+ return metricName;
+ }
+ }
+
+ public static class TestMetricsContainer extends MetricsContainerImpl {
+
+ // public TestHistogram testHistogram = new TestHistogram();
+ public ConcurrentHashMap<KV<MetricName, HistogramData.BucketType>, TestHistogram>
+ perWorkerHistograms =
+ new ConcurrentHashMap<KV<MetricName, HistogramData.BucketType>, TestHistogram>();
+ public ConcurrentHashMap<MetricName, CounterCell> perWorkerCounters =
+ new ConcurrentHashMap<MetricName, CounterCell>();
+
+ public TestMetricsContainer() {
+ super("TestStep");
+ }
+
+ @Override
+ public Histogram getPerWorkerHistogram(
+ MetricName metricName, HistogramData.BucketType bucketType) {
+ perWorkerHistograms.computeIfAbsent(KV.of(metricName, bucketType), kv -> new TestHistogram());
+ return perWorkerHistograms.get(KV.of(metricName, bucketType));
+ // return testHistogram;
+ }
+
+ @Override
+ public Counter getPerWorkerCounter(MetricName metricName) {
+ perWorkerCounters.computeIfAbsent(metricName, name -> new CounterCell(name));
+ return perWorkerCounters.get(metricName);
+ }
+
+ @Override
+ public void reset() {
+ // testHistogram.values.clear();
+ perWorkerHistograms.clear();
+ perWorkerCounters.clear();
+ }
+ }
+
+ @Test
+ public void testAppendRowsRowStatusCounter() throws Exception {
+ // Setup
+ TestMetricsContainer testContainer = new TestMetricsContainer();
+ MetricsEnvironment.setCurrentContainer(testContainer);
+
+ BigQuerySinkMetrics.setSupportMetricsDeletion(false);
+ Counter deletesDisabledCounter =
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.SUCCESSFUL, "rpcStatus", "tableId");
+ deletesDisabledCounter.inc();
+ MetricName deletesDisabledCounterName =
+ MetricName.named(
+ "BigQuerySink", "AppendRowsRowStatus-RowStatus:SUCCESSFUL;RpcStatus:rpcStatus;");
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(deletesDisabledCounterName));
+ assertThat(
+ testContainer.perWorkerCounters.get(deletesDisabledCounterName).getCumulative(),
+ equalTo(1L));
+
+ BigQuerySinkMetrics.setSupportMetricsDeletion(true);
+ testContainer.reset();
+ Counter deletesEnabledCounter =
+ BigQuerySinkMetrics.appendRowsRowStatusCounter(
+ BigQuerySinkMetrics.RowStatus.SUCCESSFUL, "rpcStatus", "tableId");
+ deletesEnabledCounter.inc();
+ MetricName deletesEnabledCounterName =
+ MetricName.named(
+ "BigQuerySink",
+ "AppendRowsRowStatus-RowStatus:SUCCESSFUL;RpcStatus:rpcStatus;TableId:tableId;");
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(deletesEnabledCounterName));
+ assertThat(
+ testContainer.perWorkerCounters.get(deletesEnabledCounterName).getCumulative(),
+ equalTo(1L));
+ }
+
+ @Test
+ public void testThrowableToGRPCCodeString() throws Exception {
+ Throwable nullThrowable = null;
+ assertThat(BigQuerySinkMetrics.throwableToGRPCCodeString(nullThrowable), equalTo("UNKNOWN"));
+
+ Throwable nonGrpcError = new IndexOutOfBoundsException("Test Error");
+ assertThat(BigQuerySinkMetrics.throwableToGRPCCodeString(nonGrpcError), equalTo("UNKNOWN"));
+
+ int notFoundVal = Status.Code.NOT_FOUND.value();
+ Throwable grpcError =
+ new Exceptions.AppendSerializtionError(notFoundVal, "Test Error", "Stream name", null);
+ assertThat(BigQuerySinkMetrics.throwableToGRPCCodeString(grpcError), equalTo("NOT_FOUND"));
+ }
+
+ @Test
+ public void testThrottledTimeCounter() throws Exception {
+ // Setup
+ TestMetricsContainer testContainer = new TestMetricsContainer();
+ MetricsEnvironment.setCurrentContainer(testContainer);
+
+ // Test throttleCounter metric.
+ Counter appendRowsThrottleCounter =
+ BigQuerySinkMetrics.throttledTimeCounter(BigQuerySinkMetrics.RpcMethod.APPEND_ROWS);
+ appendRowsThrottleCounter.inc(1);
+ assertThat(
+ appendRowsThrottleCounter.getName().getName(),
+ equalTo("ThrottledTime-Method:APPEND_ROWS;"));
+ MetricName counterName = MetricName.named("BigQuerySink", "ThrottledTime-Method:APPEND_ROWS;");
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(counterName));
+ assertThat(testContainer.perWorkerCounters.get(counterName).getCumulative(), equalTo(1L));
+ }
+
+ @Test
+ public void testReportSuccessfulRpcMetrics() throws Exception {
+ // Setup
+ TestMetricsContainer testContainer = new TestMetricsContainer();
+ MetricsEnvironment.setCurrentContainer(testContainer);
+ Context<AppendRowsResponse> c = new Context<AppendRowsResponse>();
+ Instant t1 = Instant.now();
+ c.setOperationStartTime(t1);
+ c.setOperationEndTime(t1.plusMillis(3));
+
+ // Test disabled SupportMetricsDeletion
+ BigQuerySinkMetrics.setSupportMetricsDeletion(false);
+ BigQuerySinkMetrics.reportSuccessfulRpcMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, "tableId");
+ MetricName counterNameDisabledDeletes =
+ MetricName.named("BigQuerySink", "RpcRequests-Method:APPEND_ROWS;RpcStatus:OK;");
+ MetricName histogramName = MetricName.named("BigQuerySink", "RpcLatency-Method:APPEND_ROWS;");
+ HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(1, 34);
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(counterNameDisabledDeletes));
+ assertThat(
+ testContainer.perWorkerCounters.get(counterNameDisabledDeletes).getCumulative(),
+ equalTo(1L));
+ assertThat(
+ testContainer.perWorkerHistograms.get(KV.of(histogramName, bucketType)).values,
+ containsInAnyOrder(Double.valueOf(3.0)));
+
+ // Test enable SupportMetricsDeletion.
+ BigQuerySinkMetrics.setSupportMetricsDeletion(true);
+ testContainer.reset();
+ BigQuerySinkMetrics.reportSuccessfulRpcMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, "tableId");
+ MetricName counterNameEnabledDeletes =
+ MetricName.named(
+ "BigQuerySink", "RpcRequests-Method:APPEND_ROWS;RpcStatus:OK;TableId:tableId;");
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(counterNameEnabledDeletes));
+ assertThat(
+ testContainer.perWorkerCounters.get(counterNameEnabledDeletes).getCumulative(),
+ equalTo(1L));
+ assertThat(
+ testContainer.perWorkerHistograms.get(KV.of(histogramName, bucketType)).values,
+ containsInAnyOrder(Double.valueOf(3.0)));
+ }
+
+ @Test
+ public void testReportFailedRPCMetrics_KnownGrpcError() throws Exception {
+ // Setup
+ TestMetricsContainer testContainer = new TestMetricsContainer();
+ MetricsEnvironment.setCurrentContainer(testContainer);
+ Context<AppendRowsResponse> c = new Context<AppendRowsResponse>();
+ Instant t1 = Instant.now();
+ c.setOperationStartTime(t1);
+ c.setOperationEndTime(t1.plusMillis(5));
+ int notFoundVal = Status.Code.NOT_FOUND.value();
+ Throwable grpcError =
+ new Exceptions.AppendSerializtionError(notFoundVal, "Test Error", "Stream name", null);
+ c.setError(grpcError);
+
+ // Test disabled SupportMetricsDeletion
+ BigQuerySinkMetrics.setSupportMetricsDeletion(false);
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, "tableId");
+ MetricName counterNameDisabledDeletes =
+ MetricName.named("BigQuerySink", "RpcRequests-Method:APPEND_ROWS;RpcStatus:NOT_FOUND;");
+ MetricName histogramName = MetricName.named("BigQuerySink", "RpcLatency-Method:APPEND_ROWS;");
+ HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(1, 34);
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(counterNameDisabledDeletes));
+ assertThat(
+ testContainer.perWorkerCounters.get(counterNameDisabledDeletes).getCumulative(),
+ equalTo(1L));
+ assertThat(
+ testContainer.perWorkerHistograms,
+ IsMapContaining.hasKey(KV.of(histogramName, bucketType)));
+ assertThat(
+ testContainer.perWorkerHistograms.get(KV.of(histogramName, bucketType)).values,
+ containsInAnyOrder(Double.valueOf(5.0)));
+
+ // Test enable SupportMetricsDeletion
+ BigQuerySinkMetrics.setSupportMetricsDeletion(true);
+ testContainer.reset();
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, "tableId");
+ MetricName counterNameEnabledDeletes =
+ MetricName.named(
+ "BigQuerySink", "RpcRequests-Method:APPEND_ROWS;RpcStatus:NOT_FOUND;TableId:tableId;");
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(counterNameEnabledDeletes));
+ assertThat(
+ testContainer.perWorkerCounters.get(counterNameEnabledDeletes).getCumulative(),
+ equalTo(1L));
+ assertThat(
+ testContainer.perWorkerHistograms.get(KV.of(histogramName, bucketType)).values,
+ containsInAnyOrder(Double.valueOf(5.0)));
+ }
+
+ @Test
+ public void testReportFailedRPCMetrics_UnknownGrpcError() throws Exception {
+ // Setup
+ BigQuerySinkMetrics.setSupportMetricsDeletion(false);
+ TestMetricsContainer testContainer = new TestMetricsContainer();
+ MetricsEnvironment.setCurrentContainer(testContainer);
+ Context<AppendRowsResponse> c = new Context<AppendRowsResponse>();
+ Instant t1 = Instant.now();
+ c.setOperationStartTime(t1);
+ c.setOperationEndTime(t1.plusMillis(15));
+ Throwable nonGrpcError = new IndexOutOfBoundsException("Test Error");
+ c.setError(nonGrpcError);
+
+ // Test disabled SupportMetricsDeletion
+ BigQuerySinkMetrics.setSupportMetricsDeletion(false);
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, "tableId");
+ MetricName counterNameDisabledDeletes =
+ MetricName.named("BigQuerySink", "RpcRequests-Method:APPEND_ROWS;RpcStatus:UNKNOWN;");
+ MetricName histogramName = MetricName.named("BigQuerySink", "RpcLatency-Method:APPEND_ROWS;");
+ HistogramData.BucketType bucketType = HistogramData.ExponentialBuckets.of(1, 34);
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(counterNameDisabledDeletes));
+ assertThat(
+ testContainer.perWorkerCounters.get(counterNameDisabledDeletes).getCumulative(),
+ equalTo(1L));
+ assertThat(
+ testContainer.perWorkerHistograms.get(KV.of(histogramName, bucketType)).values,
+ containsInAnyOrder(Double.valueOf(15.0)));
+
+ // Test enable SupportMetricsDeletion
+ BigQuerySinkMetrics.setSupportMetricsDeletion(true);
+ testContainer.reset();
+ BigQuerySinkMetrics.reportFailedRPCMetrics(
+ c, BigQuerySinkMetrics.RpcMethod.APPEND_ROWS, "tableId");
+ MetricName counterNameEnabledDeletes =
+ MetricName.named(
+ "BigQuerySink", "RpcRequests-Method:APPEND_ROWS;RpcStatus:UNKNOWN;TableId:tableId;");
+ assertThat(testContainer.perWorkerCounters, IsMapContaining.hasKey(counterNameEnabledDeletes));
+ assertThat(
+ testContainer.perWorkerCounters.get(counterNameEnabledDeletes).getCumulative(),
+ equalTo(1L));
+ assertThat(
+ testContainer.perWorkerHistograms.get(KV.of(histogramName, bucketType)).values,
+ containsInAnyOrder(Double.valueOf(15.0)));
+ }
+}
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/RetryManagerTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/RetryManagerTest.java
index 958fd356344..804552fc83c 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/RetryManagerTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/RetryManagerTest.java
@@ -22,9 +22,11 @@ import static org.junit.Assert.assertEquals;
import com.google.api.core.ApiFutures;
import java.util.List;
import java.util.Map;
+import org.apache.beam.runners.core.metrics.CounterCell;
import org.apache.beam.sdk.io.gcp.bigquery.RetryManager.Operation;
import org.apache.beam.sdk.io.gcp.bigquery.RetryManager.Operation.Context;
import org.apache.beam.sdk.io.gcp.bigquery.RetryManager.RetryType;
+import org.apache.beam.sdk.metrics.MetricName;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;
import org.joda.time.Duration;
@@ -44,8 +46,10 @@ public class RetryManagerTest {
@Test
public void testNoFailures() throws Exception {
List<Context> contexts = Lists.newArrayList();
+ MetricName metricName = MetricName.named("TEST_NAMESPACE", "THROTTLED_COUNTER");
+ CounterCell throttleTimeCounter = new CounterCell(metricName);
RetryManager<String, RetryManagerTest.Context> retryManager =
- new RetryManager<>(Duration.millis(1), Duration.millis(1), 5);
+ new RetryManager<>(Duration.millis(1), Duration.millis(1), 5, throttleTimeCounter);
for (int i = 0; i < 5; ++i) {
Context context = new Context();
contexts.add(context);
@@ -74,6 +78,8 @@ public class RetryManagerTest {
assertEquals(1, c.numSucceeded);
assertEquals(0, c.numFailed);
});
+
+ assertEquals(0L, (long) throttleTimeCounter.getCumulative());
}
@Test
@@ -82,8 +88,11 @@ public class RetryManagerTest {
Map<String, Integer> expectedStarts = Maps.newHashMap();
Map<String, Integer> expectedFailures = Maps.newHashMap();
+ MetricName metricName = MetricName.named("TEST_NAMESPACE", "THROTTLED_COUNTER");
+ CounterCell throttleTimeCounter = new CounterCell(metricName);
+
RetryManager<String, RetryManagerTest.Context> retryManager =
- new RetryManager<>(Duration.millis(1), Duration.millis(1), 50);
+ new RetryManager<>(Duration.millis(1), Duration.millis(1), 50, throttleTimeCounter);
for (int i = 0; i < 5; ++i) {
final int index = i;
String value = "yes " + i;
@@ -129,14 +138,18 @@ public class RetryManagerTest {
assertEquals(1, e.getValue().numSucceeded);
assertEquals((int) expectedFailures.get(e.getKey()), e.getValue().numFailed);
});
+ // Each operation backsoff once and each backoff is 1ms.
+ assertEquals(5L, (long) throttleTimeCounter.getCumulative());
}
@Test
public void testDontRetry() throws Exception {
List<Context> contexts = Lists.newArrayList();
+ MetricName metricName = MetricName.named("TEST_NAMESPACE", "THROTTLED_COUNTER");
+ CounterCell throttleTimeCounter = new CounterCell(metricName);
RetryManager<String, RetryManagerTest.Context> retryManager =
- new RetryManager<>(Duration.millis(1), Duration.millis(1), 50);
+ new RetryManager<>(Duration.millis(1), Duration.millis(1), 50, throttleTimeCounter);
for (int i = 0; i < 5; ++i) {
Context context = new Context();
contexts.add(context);
@@ -172,6 +185,8 @@ public class RetryManagerTest {
assertEquals(0, c.numSucceeded);
assertEquals(1, c.numFailed);
});
+
+ assertEquals(0L, (long) throttleTimeCounter.getCumulative());
}
@Test