diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystem.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystem.java index 1bee44eb38c0..5eca4e9e2c2f 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystem.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystem.java @@ -77,25 +77,58 @@ class GcsFileSystem extends FileSystem { private final GcsOptions options; - /** Number of copy operations performed. */ - private Counter numCopies; + /** The {@code _count} and {@code _msec} counter pair for a single operation. */ + private static class OpMetrics { + private final Counter count; + private final Counter msec; + + OpMetrics(String operation) { + this.count = Metrics.counter(GcsUtil.METRIC_NAMESPACE, "gcs_op_" + operation + "_count"); + this.msec = Metrics.counter(GcsUtil.METRIC_NAMESPACE, "gcs_op_" + operation + "_msec"); + } + } - /** Number of renames operations performed. */ - private Counter numRenames; + /** + * Per-operation metrics, or null when {@link GcsOptions#getGcsPerformanceMetrics()} is off, which + * is the default. Every recording site tolerates null, so nothing is emitted unless asked for. + * + *

For {@link #open} and {@link #create} the elapsed time covers only channel setup, not the + * transfer; the bytes moved are counted by the {@code gcs_http_*} wire-byte counters. + */ + private @Nullable OpMetrics copyMetrics; - /** Time spent performing copies. */ - private Counter copyTimeMsec; + private @Nullable OpMetrics renameMetrics; + private @Nullable OpMetrics deleteMetrics; + private @Nullable OpMetrics matchGlobMetrics; + private @Nullable OpMetrics matchNonGlobMetrics; + private @Nullable OpMetrics openMetrics; + private @Nullable OpMetrics createMetrics; - /** Time spent performing renames. */ - private Counter renameTimeMsec; + /** Object listing pages fetched while expanding globs. */ + private @Nullable Counter matchGlobPages; GcsFileSystem(GcsOptions options) { this.options = checkNotNull(options, "options"); if (options.getGcsPerformanceMetrics()) { - numCopies = Metrics.counter(GcsFileSystem.class, "num_copies"); - copyTimeMsec = Metrics.counter(GcsFileSystem.class, "copy_time_msec"); - numRenames = Metrics.counter(GcsFileSystem.class, "num_renames"); - renameTimeMsec = Metrics.counter(GcsFileSystem.class, "rename_time_msec"); + copyMetrics = new OpMetrics("copy"); + renameMetrics = new OpMetrics("rename"); + deleteMetrics = new OpMetrics("delete"); + matchGlobMetrics = new OpMetrics("match_glob"); + matchNonGlobMetrics = new OpMetrics("match_nonglob"); + openMetrics = new OpMetrics("open"); + createMetrics = new OpMetrics("create"); + matchGlobPages = Metrics.counter(GcsUtil.METRIC_NAMESPACE, "gcs_op_match_glob_pages"); + } + } + + /** + * Records a finished operation. Called from a finally block so that failed operations, which are + * often the slow ones, are counted too. + */ + private static void record(@Nullable OpMetrics metrics, long count, Stopwatch stopwatch) { + if (metrics != null) { + metrics.count.inc(count); + metrics.msec.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS)); } } @@ -152,12 +185,22 @@ protected WritableByteChannel create(GcsResourceId resourceId, CreateOptions cre builder.setUploadBufferSizeBytes( ((GcsCreateOptions) createOptions).gcsUploadBufferSizeBytes()); } - return options.getGcsUtil().create(resourceId.getGcsPath(), builder.build()); + Stopwatch stopwatch = Stopwatch.createStarted(); + try { + return options.getGcsUtil().create(resourceId.getGcsPath(), builder.build()); + } finally { + record(createMetrics, 1, stopwatch); + } } @Override protected ReadableByteChannel open(GcsResourceId resourceId) throws IOException { - return options.getGcsUtil().open(resourceId.getGcsPath()); + Stopwatch stopwatch = Stopwatch.createStarted(); + try { + return options.getGcsUtil().open(resourceId.getGcsPath()); + } finally { + record(openMetrics, 1, stopwatch); + } } @Override @@ -167,19 +210,23 @@ protected void rename( MoveOptions... moveOptions) throws IOException { Stopwatch stopwatch = Stopwatch.createStarted(); - options - .getGcsUtil() - .rename(toFilenames(srcResourceIds), toFilenames(destResourceIds), moveOptions); - stopwatch.stop(); - if (options.getGcsPerformanceMetrics()) { - numRenames.inc(srcResourceIds.size()); - renameTimeMsec.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS)); + try { + options + .getGcsUtil() + .rename(toFilenames(srcResourceIds), toFilenames(destResourceIds), moveOptions); + } finally { + record(renameMetrics, srcResourceIds.size(), stopwatch); } } @Override protected void delete(Collection resourceIds) throws IOException { - options.getGcsUtil().remove(toFilenames(resourceIds)); + Stopwatch stopwatch = Stopwatch.createStarted(); + try { + options.getGcsUtil().remove(toFilenames(resourceIds)); + } finally { + record(deleteMetrics, resourceIds.size(), stopwatch); + } } @Override @@ -202,11 +249,10 @@ protected GcsResourceId matchNewResource(String singleResourceSpec, boolean isDi protected void copy(List srcResourceIds, List destResourceIds) throws IOException { Stopwatch stopwatch = Stopwatch.createStarted(); - options.getGcsUtil().copy(toFilenames(srcResourceIds), toFilenames(destResourceIds)); - stopwatch.stop(); - if (options.getGcsPerformanceMetrics()) { - numCopies.inc(srcResourceIds.size()); - copyTimeMsec.inc(stopwatch.elapsed(TimeUnit.MILLISECONDS)); + try { + options.getGcsUtil().copy(toFilenames(srcResourceIds), toFilenames(destResourceIds)); + } finally { + record(copyMetrics, srcResourceIds.size(), stopwatch); } } @@ -265,26 +311,35 @@ MatchResult expand(GcsPath gcsPattern) throws IOException { prefix, p.toString()); - String pageToken = null; - List results = new ArrayList<>(); - do { - Objects objects = options.getGcsUtil().listObjects(gcsPattern.getBucket(), prefix, pageToken); - if (objects.getItems() == null) { - break; - } + Stopwatch stopwatch = Stopwatch.createStarted(); + try { + String pageToken = null; + List results = new ArrayList<>(); + do { + Objects objects = + options.getGcsUtil().listObjects(gcsPattern.getBucket(), prefix, pageToken); + if (matchGlobPages != null) { + matchGlobPages.inc(); + } + if (objects.getItems() == null) { + break; + } - // Filter objects based on the regex. - for (StorageObject o : objects.getItems()) { - String name = o.getName(); - // Skip directories, which end with a slash. - if (p.matcher(name).matches() && !name.endsWith("/")) { - LOG.debug("Matched object: {}", name); - results.add(toMetadata(o)); + // Filter objects based on the regex. + for (StorageObject o : objects.getItems()) { + String name = o.getName(); + // Skip directories, which end with a slash. + if (p.matcher(name).matches() && !name.endsWith("/")) { + LOG.debug("Matched object: {}", name); + results.add(toMetadata(o)); + } } - } - pageToken = objects.getNextPageToken(); - } while (pageToken != null); - return MatchResult.create(Status.OK, results); + pageToken = objects.getNextPageToken(); + } while (pageToken != null); + return MatchResult.create(Status.OK, results); + } finally { + record(matchGlobMetrics, 1, stopwatch); + } } /** @@ -295,7 +350,17 @@ MatchResult expand(GcsPath gcsPattern) throws IOException { */ @VisibleForTesting List matchNonGlobs(List gcsPaths) throws IOException { - List results = options.getGcsUtil().getObjects(gcsPaths); + if (gcsPaths.isEmpty()) { + // match() always calls this, so recording here would bury the real calls in empty ones. + return ImmutableList.of(); + } + Stopwatch stopwatch = Stopwatch.createStarted(); + List results; + try { + results = options.getGcsUtil().getObjects(gcsPaths); + } finally { + record(matchNonGlobMetrics, gcsPaths.size(), stopwatch); + } ImmutableList.Builder ret = ImmutableList.builder(); for (StorageObjectOrIOException result : results) { diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java index 5ed97d935c66..070cf74d7c11 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtil.java @@ -49,9 +49,23 @@ import org.checkerframework.checker.nullness.qual.Nullable; public class GcsUtil { + /** + * Namespace for every GCS metric. The namespace is dropped when Dataflow exports counters to + * Cloud Monitoring, so the layer is carried by the metric name instead: {@code gcs_http_*} for + * transport-level counters and {@code gcs_op_*} for operation-level ones. + */ + public static final String METRIC_NAMESPACE = "Gcs"; + @VisibleForTesting GcsUtilV1 delegate; @VisibleForTesting @Nullable GcsUtilV2 delegateV2; + /** + * @deprecated no {@link GcsUtil} API accepts this type, so an instance cannot be used for + * anything. GCS counters are configured from {@link + * org.apache.beam.sdk.extensions.gcp.options.GcsOptions} when the {@link GcsUtil} is + * constructed. Scheduled for removal. + */ + @Deprecated public static class GcsCountersOptions { final GcsUtilV1.GcsCountersOptions delegate; diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java index 5a6a87dc4b79..a04f688ec7a3 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilV1.java @@ -50,6 +50,7 @@ import com.google.cloud.hadoop.util.ResilientOperation; import com.google.cloud.hadoop.util.RetryDeterminer; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import java.io.Closeable; import java.io.FileNotFoundException; import java.io.IOException; import java.nio.channels.SeekableByteChannel; @@ -85,13 +86,20 @@ import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath; import org.apache.beam.sdk.io.fs.MoveOptions; import org.apache.beam.sdk.io.fs.MoveOptions.StandardMoveOptions; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.MetricName; import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.metrics.MetricsContainer; +import org.apache.beam.sdk.metrics.MetricsEnvironment; import org.apache.beam.sdk.options.DefaultValueFactory; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.util.FluentBackoff; import org.apache.beam.sdk.util.MoreFutures; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.Cache; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.CacheBuilder; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.cache.RemovalNotification; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; 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.Sets; @@ -107,19 +115,35 @@ }) class GcsUtilV1 { + /** Describes which GCS counters this {@link GcsUtilV1} emits. */ @AutoValue public abstract static class GcsCountersOptions { public abstract @Nullable String getReadCounterPrefix(); public abstract @Nullable String getWriteCounterPrefix(); + /** + * Whether to emit the {@code gcs_*} performance counters, which are reported under {@link + * GcsUtil#METRIC_NAMESPACE} and are not per bucket. Set from {@link + * GcsOptions#getGcsPerformanceMetrics()}. + */ + public abstract boolean getPerformanceMetricsEnabled(); + public boolean hasAnyPrefix() { return getWriteCounterPrefix() != null || getReadCounterPrefix() != null; } public static GcsCountersOptions create( @Nullable String readCounterPrefix, @Nullable String writeCounterPrefix) { - return new AutoValue_GcsUtilV1_GcsCountersOptions(readCounterPrefix, writeCounterPrefix); + return create(readCounterPrefix, writeCounterPrefix, false); + } + + public static GcsCountersOptions create( + @Nullable String readCounterPrefix, + @Nullable String writeCounterPrefix, + boolean performanceMetricsEnabled) { + return new AutoValue_GcsUtilV1_GcsCountersOptions( + readCounterPrefix, writeCounterPrefix, performanceMetricsEnabled); } } @@ -153,7 +177,8 @@ public GcsUtilV1 create(PipelineOptions options) { : null, gcsOptions.getEnableBucketWriteMetricCounter() ? gcsOptions.getGcsWriteCounterPrefix() - : null), + : null, + Boolean.TRUE.equals(gcsOptions.getGcsPerformanceMetrics())), gcsOptions.getGoogleCloudStorageReadOptions()); } } @@ -213,6 +238,17 @@ public boolean shouldRetry(IOException e) { private GoogleCloudStorage googleCloudStorage; private GoogleCloudStorageOptions googleCloudStorageOptions; + private final Cache readStorageByContainer = + CacheBuilder.newBuilder() + .weakKeys() + .removalListener( + (RemovalNotification notification) -> { + GoogleCloudStorage storage = notification.getValue(); + if (storage != null) { + storage.close(); + } + }) + .build(); private final int rewriteDataOpBatchLimit; @@ -223,29 +259,6 @@ public boolean shouldRetry(IOException e) { @VisibleForTesting @Nullable AtomicInteger numRewriteTokensUsed; - @VisibleForTesting - GcsUtilV1( - Storage storageClient, - HttpRequestInitializer httpRequestInitializer, - ExecutorService executorService, - Boolean shouldUseGrpc, - Credentials credentials, - @Nullable Integer uploadBufferSizeBytes, - @Nullable Integer rewriteDataOpBatchLimit, - GcsCountersOptions gcsCountersOptions, - GcsOptions gcsOptions) { - this( - storageClient, - httpRequestInitializer, - executorService, - shouldUseGrpc, - credentials, - uploadBufferSizeBytes, - rewriteDataOpBatchLimit, - gcsCountersOptions, - gcsOptions.getGoogleCloudStorageReadOptions()); - } - @VisibleForTesting GcsUtilV1( Storage storageClient, @@ -526,54 +539,79 @@ private static Consumer createCounterConsumer(String counterNamePrefix, } private WritableByteChannel wrapInCounting( - WritableByteChannel writableByteChannel, String bucket) { + WritableByteChannel writableByteChannel, + String bucket, + @Nullable MetricsContainer container) { if (writableByteChannel instanceof CountingWritableByteChannel) { return writableByteChannel; } - return Optional.ofNullable(gcsCountersOptions.getWriteCounterPrefix()) - .map( - prefix -> { - LOG.debug( - "wrapping writable byte channel using counter name prefix {} and bucket {}", - prefix, - bucket); - return new CountingWritableByteChannel( - writableByteChannel, createCounterConsumer(prefix, bucket)); - }) - .orElse(writableByteChannel); - } - - private SeekableByteChannel wrapInCounting( - SeekableByteChannel seekableByteChannel, String bucket) { - if (seekableByteChannel instanceof CountingSeekableByteChannel - || !gcsCountersOptions.hasAnyPrefix()) { - return seekableByteChannel; - } - return new CountingSeekableByteChannel( - seekableByteChannel, - Optional.ofNullable(gcsCountersOptions.getReadCounterPrefix()) + Consumer writeConsumer = + Optional.ofNullable(gcsCountersOptions.getWriteCounterPrefix()) .map( prefix -> { LOG.debug( - "wrapping seekable byte channel with \"bytes read\" counter name prefix {}" - + " and bucket {}", + "wrapping writable byte channel using counter name prefix {} and bucket {}", prefix, bucket); return createCounterConsumer(prefix, bucket); }) - .orElse(null), - Optional.ofNullable(gcsCountersOptions.getWriteCounterPrefix()) + .orElse(null); + + if (gcsCountersOptions.getPerformanceMetricsEnabled() && container != null) { + Counter perfWriteCounter = + container.getCounter( + MetricName.named(GcsUtil.METRIC_NAMESPACE, "gcs_http_write_wire_bytes_sent")); + Consumer perfConsumer = perfWriteCounter::inc; + writeConsumer = writeConsumer == null ? perfConsumer : writeConsumer.andThen(perfConsumer); + } + + if (writeConsumer == null) { + return writableByteChannel; + } + + return new CountingWritableByteChannel(writableByteChannel, writeConsumer); + } + + private SeekableByteChannel wrapInCounting( + SeekableByteChannel seekableByteChannel, + String bucket, + @Nullable MetricsContainer container) { + if (seekableByteChannel instanceof CountingSeekableByteChannel) { + return seekableByteChannel; + } + + // SeekableByteChannel is only returned by GcsUtilV1.open(...) for reading immutable GCS objects + // (GoogleCloudStorageReadChannel throws NonWritableChannelException on write). All GCS writes + // go through GcsUtilV1.create(...), which returns a WritableByteChannel. Therefore, only a read + // counter consumer is needed here. + Consumer readConsumer = + Optional.ofNullable(gcsCountersOptions.getReadCounterPrefix()) .map( prefix -> { LOG.debug( - "wrapping seekable byte channel with \"bytes written\" counter name prefix {}" + "wrapping seekable byte channel with \"bytes read\" counter name prefix {}" + " and bucket {}", prefix, bucket); return createCounterConsumer(prefix, bucket); }) - .orElse(null)); + .orElse(null); + + if (gcsCountersOptions.getPerformanceMetricsEnabled() && container != null) { + Counter perfReadCounter = + container.getCounter( + MetricName.named(GcsUtil.METRIC_NAMESPACE, "gcs_http_read_wire_bytes_received")); + Consumer perfConsumer = perfReadCounter::inc; + readConsumer = readConsumer == null ? perfConsumer : readConsumer.andThen(perfConsumer); + } + + if (readConsumer == null) { + return seekableByteChannel; + } + + return CountingSeekableByteChannel.createWithBytesReadConsumer( + seekableByteChannel, readConsumer); } /** @@ -615,11 +653,38 @@ SeekableByteChannel open(GcsPath path, GoogleCloudStorageReadOptions readOptions ServiceCallMetric serviceCallMetric = new ServiceCallMetric(MonitoringInfoConstants.Urns.API_REQUEST_COUNT, baseLabels); try { + GoogleCloudStorage gcpStorage = this.googleCloudStorage; + MetricsContainer container = null; + if (gcsCountersOptions.getPerformanceMetricsEnabled()) { + container = MetricsEnvironment.getCurrentContainer(); + if (container != null) { + final MetricsContainer currentContainer = container; + try { + gcpStorage = + readStorageByContainer.get( + currentContainer, + () -> { + HttpRequestInitializer scopedInitializer = + Transport.withMetricsContainer( + this.httpRequestInitializer, currentContainer, false); + return createGoogleCloudStorage( + googleCloudStorageOptions, + this.storageClient, + this.credentials, + scopedInitializer); + }); + } catch (ExecutionException e) { + if (e.getCause() instanceof IOException) { + throw (IOException) e.getCause(); + } + throw new IOException(e); + } + } + } SeekableByteChannel channel = - googleCloudStorage.open( - new StorageResourceId(path.getBucket(), path.getObject()), readOptions); + gcpStorage.open(new StorageResourceId(path.getBucket(), path.getObject()), readOptions); serviceCallMetric.call("ok"); - return wrapInCounting(channel, path.getBucket()); + return wrapInCounting(channel, path.getBucket(), container); } catch (IOException e) { if (e.getCause() instanceof GoogleJsonResponseException) { serviceCallMetric.call(((GoogleJsonResponseException) e.getCause()).getDetails().getCode()); @@ -701,9 +766,18 @@ public WritableByteChannel create(GcsPath path, CreateOptions options) throws IO } GoogleCloudStorageOptions newGoogleCloudStorageOptions = googleCloudStorageOptions.toBuilder().setWriteChannelOptions(wcOptions).build(); + HttpRequestInitializer scopedInitializer = this.httpRequestInitializer; + MetricsContainer container = null; + if (gcsCountersOptions.getPerformanceMetricsEnabled()) { + container = MetricsEnvironment.getCurrentContainer(); + if (container != null) { + scopedInitializer = + Transport.withMetricsContainer(this.httpRequestInitializer, container, true); + } + } GoogleCloudStorage gcpStorage = createGoogleCloudStorage( - newGoogleCloudStorageOptions, this.storageClient, this.credentials); + newGoogleCloudStorageOptions, this.storageClient, this.credentials, scopedInitializer); StorageResourceId resourceId = new StorageResourceId( path.getBucket(), @@ -735,7 +809,7 @@ public WritableByteChannel create(GcsPath path, CreateOptions options) throws IO try { WritableByteChannel channel = gcpStorage.create(resourceId, createBuilder.build()); serviceCallMetric.call("ok"); - return wrapInCounting(channel, path.getBucket()); + return wrapInCounting(channel, path.getBucket(), container); } catch (IOException e) { if (e.getCause() instanceof GoogleJsonResponseException) { serviceCallMetric.call(((GoogleJsonResponseException) e.getCause()).getDetails().getCode()); @@ -744,10 +818,21 @@ public WritableByteChannel create(GcsPath path, CreateOptions options) throws IO } } - @SuppressFBWarnings("LG_LOST_LOGGER_DUE_TO_WEAK_REFERENCE") + @VisibleForTesting GoogleCloudStorage createGoogleCloudStorage( GoogleCloudStorageOptions options, Storage storage, Credentials credentials) throws IOException { + return createGoogleCloudStorage(options, storage, credentials, this.httpRequestInitializer); + } + + @VisibleForTesting + @SuppressFBWarnings("LG_LOST_LOGGER_DUE_TO_WEAK_REFERENCE") + GoogleCloudStorage createGoogleCloudStorage( + GoogleCloudStorageOptions options, + Storage storage, + Credentials credentials, + @Nullable HttpRequestInitializer httpRequestInitializer) + throws IOException { // Suppress log spams in gcsio 3.0 if (overwriteLog.compareAndSet(false, true)) { java.util.logging.Logger.getLogger("com.google.cloud.hadoop.gcsio.GoogleCloudStorageImpl") @@ -949,9 +1034,21 @@ private static void executeBatches(List batches) throws IOExcept TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>())); + MetricsContainer container = MetricsEnvironment.getCurrentContainer(); List> futures = new ArrayList<>(); for (final BatchInterface batch : batches) { - futures.add(MoreFutures.runAsync(batch::execute, executor)); + futures.add( + MoreFutures.runAsync( + () -> { + if (container != null) { + try (Closeable scope = MetricsEnvironment.scopedMetricsContainer(container)) { + batch.execute(); + } + } else { + batch.execute(); + } + }, + executor)); } try { diff --git a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/Transport.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/Transport.java index ea31e6c91801..6e100a6148d8 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/Transport.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/Transport.java @@ -21,7 +21,12 @@ import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings.isNullOrEmpty; import com.google.api.client.googleapis.javanet.GoogleNetHttpTransport; +import com.google.api.client.http.HttpExecuteInterceptor; +import com.google.api.client.http.HttpIOExceptionHandler; +import com.google.api.client.http.HttpRequest; import com.google.api.client.http.HttpRequestInitializer; +import com.google.api.client.http.HttpResponse; +import com.google.api.client.http.HttpResponseInterceptor; import com.google.api.client.http.HttpTransport; import com.google.api.client.json.JsonFactory; import com.google.api.client.json.gson.GsonFactory; @@ -39,6 +44,9 @@ import javax.annotation.Nullable; import org.apache.beam.sdk.extensions.gcp.auth.NullCredentialInitializer; import org.apache.beam.sdk.extensions.gcp.options.GcsOptions; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.metrics.MetricName; +import org.apache.beam.sdk.metrics.MetricsContainer; import org.apache.beam.sdk.util.ReleaseInfo; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; @@ -119,6 +127,106 @@ public static Storage.Builder newStorageClient(GcsOptions options) { return storageBuilder; } + /** + * Wraps an {@link HttpRequestInitializer} so that HTTP execute and response interceptors + * increment {@link Counter} instances pre-bound to the given {@link MetricsContainer}. This + * guarantees that GCS HTTP metrics are attributed directly to the step that created the channel, + * even when requests execute on background worker threads. + * + *

+ * + *

Note that {@code request_count_unbounded} counts metadata GETs as well as full object reads, + * since neither carries a Range header. + */ + public static HttpRequestInitializer withMetricsContainer( + HttpRequestInitializer base, @Nullable MetricsContainer container, boolean isWrite) { + if (container == null) { + return base; + } + + String prefix = isWrite ? "gcs_http_write_" : "gcs_http_read_"; + + // Pre-resolve counters on the calling thread (e.g. DoFn thread) while it is in the target step + Counter requestCount = + container.getCounter(MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "request_count")); + Counter rangeRequestCount = + isWrite + ? null + : container.getCounter( + MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "request_count_ranged")); + Counter unboundedStreamCount = + isWrite + ? null + : container.getCounter( + MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "request_count_unbounded")); + // Requests that are not a GET, so that the shape counters above add up to request_count. + Counter otherRequestCount = + isWrite + ? null + : container.getCounter( + MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "request_count_other")); + Counter status2xx = + container.getCounter(MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "status_2xx")); + // 3xx is not an error for GCS: a resumable upload answers 308 to every chunk but the last. + Counter status3xx = + container.getCounter(MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "status_3xx")); + Counter status4xx = + container.getCounter(MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "status_4xx")); + Counter status5xx = + container.getCounter(MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "status_5xx")); + Counter statusOther = + container.getCounter(MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "status_other")); + Counter noResponse = + container.getCounter( + MetricName.named(GcsUtil.METRIC_NAMESPACE, prefix + "request_no_response")); + + return request -> { + base.initialize(request); + HttpExecuteInterceptor existingExecuteInterceptor = request.getInterceptor(); + request.setInterceptor( + req -> { + if (existingExecuteInterceptor != null) { + existingExecuteInterceptor.intercept(req); + } + recordRequestMetrics( + req, requestCount, rangeRequestCount, unboundedStreamCount, otherRequestCount); + }); + + HttpResponseInterceptor existingResponseInterceptor = request.getResponseInterceptor(); + request.setResponseInterceptor( + res -> { + if (existingResponseInterceptor != null) { + existingResponseInterceptor.interceptResponse(res); + } + recordResponseMetrics(res, status2xx, status3xx, status4xx, status5xx, statusOther); + }); + + // An attempt that throws before a response is received never reaches the response + // interceptor, so it is counted here instead. The existing handler decides whether the + // request is retried, this only observes it. + HttpIOExceptionHandler existingIOExceptionHandler = request.getIOExceptionHandler(); + request.setIOExceptionHandler( + (req, supportsRetry) -> { + noResponse.inc(); + return existingIOExceptionHandler != null + && existingIOExceptionHandler.handleIOException(req, supportsRetry); + }); + }; + } + private static HttpRequestInitializer httpRequestInitializerFromOptions(GcsOptions options) { // Do not log the code 404. Code up the stack will deal with 404's if needed, // and logging it by default clutters the output during file staging. @@ -148,12 +256,59 @@ private static HttpRequestInitializer httpRequestInitializerFromOptions(GcsOptio retryHttpRequestInitializer.setWriteTimeout(writeTimeout); } Credentials credential = options.getGcpCredential(); - if (credential == null) { - return new ChainingHttpRequestInitializer( - new NullCredentialInitializer(), retryHttpRequestInitializer); + HttpRequestInitializer credentialsInitializer = + credential == null + ? new NullCredentialInitializer() + : new HttpCredentialsAdapter(credential); + + return new ChainingHttpRequestInitializer(credentialsInitializer, retryHttpRequestInitializer); + } + + private static void recordRequestMetrics( + HttpRequest req, + Counter requestCount, + @Nullable Counter rangeRequestCount, + @Nullable Counter unboundedStreamCount, + @Nullable Counter otherRequestCount) { + String method = req.getRequestMethod(); + requestCount.inc(); + if ("GET".equalsIgnoreCase(method)) { + String range = req.getHeaders() != null ? req.getHeaders().getRange() : null; + if (range != null) { + if (rangeRequestCount != null) { + rangeRequestCount.inc(); + } + } else { + if (unboundedStreamCount != null) { + unboundedStreamCount.inc(); + } + } + } else if (otherRequestCount != null) { + // Not a GET, e.g. the POST of a batched metadata lookup. Counted so that the three shape + // counters add up to requestCount. + otherRequestCount.inc(); + } + } + + private static void recordResponseMetrics( + HttpResponse res, + Counter status2xx, + Counter status3xx, + Counter status4xx, + Counter status5xx, + Counter statusOther) { + int code = res.getStatusCode(); + if (code >= 200 && code < 300) { + status2xx.inc(); + } else if (code >= 300 && code < 400) { + // Not an error: a resumable upload answers 308 Resume Incomplete to every chunk but the last. + status3xx.inc(); + } else if (code >= 400 && code < 500) { + status4xx.inc(); + } else if (code >= 500 && code < 600) { + status5xx.inc(); } else { - return new ChainingHttpRequestInitializer( - new HttpCredentialsAdapter(credential), retryHttpRequestInitializer); + statusOther.inc(); } } } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystemTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystemTest.java index daa419abb576..c4f716df0827 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystemTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystemTest.java @@ -20,9 +20,12 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.contains; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -30,11 +33,13 @@ import com.google.api.services.storage.model.Objects; import com.google.api.services.storage.model.StorageObject; +import java.io.Closeable; import java.io.FileNotFoundException; import java.io.IOException; import java.math.BigInteger; import java.util.ArrayList; import java.util.List; +import org.apache.beam.runners.core.metrics.MetricsContainerImpl; import org.apache.beam.sdk.extensions.gcp.options.GcsOptions; import org.apache.beam.sdk.extensions.gcp.util.GcsUtil; import org.apache.beam.sdk.extensions.gcp.util.GcsUtil.StorageObjectOrIOException; @@ -42,6 +47,8 @@ import org.apache.beam.sdk.io.fs.MatchResult; import org.apache.beam.sdk.io.fs.MatchResult.Status; import org.apache.beam.sdk.metrics.Lineage; +import org.apache.beam.sdk.metrics.MetricName; +import org.apache.beam.sdk.metrics.MetricsEnvironment; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.FluentIterable; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; @@ -268,4 +275,58 @@ private List toFilenames(MatchResult matchResult) throws IOException { .transform(metadata -> ((GcsResourceId) metadata.resourceId()).getGcsPath().toString()) .toList(); } + + private GcsFileSystem fileSystemWithMetrics(boolean enabled) { + GcsOptions gcsOptions = PipelineOptionsFactory.as(GcsOptions.class); + gcsOptions.setGcsUtil(mockGcsUtil); + gcsOptions.setGcsPerformanceMetrics(enabled); + return new GcsFileSystem(gcsOptions); + } + + private static long counter(MetricsContainerImpl container, String name) { + return container.getCounter(MetricName.named(GcsUtil.METRIC_NAMESPACE, name)).getCumulative(); + } + + private static List resourceIds(String... uris) { + return FluentIterable.from(uris) + .transform(uri -> GcsResourceId.fromGcsPath(GcsPath.fromUri(uri))) + .toList(); + } + + @Test + public void testOperationMetricsAreOffByDefault() throws Exception { + MetricsContainerImpl container = new MetricsContainerImpl(null); + try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(container)) { + fileSystemWithMetrics(false) + .rename(resourceIds("gs://bucket/from"), resourceIds("gs://bucket/to")); + } + assertEquals(0L, counter(container, "gcs_op_rename_count")); + assertEquals(0L, counter(container, "gcs_op_rename_msec")); + } + + @Test + public void testRenameRecordsObjectCountWhenEnabled() throws Exception { + MetricsContainerImpl container = new MetricsContainerImpl(null); + try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(container)) { + fileSystemWithMetrics(true) + .rename( + resourceIds("gs://bucket/a", "gs://bucket/b"), + resourceIds("gs://bucket/c", "gs://bucket/d")); + } + assertEquals(2L, counter(container, "gcs_op_rename_count")); + } + + @Test + public void testFailedOperationIsStillRecorded() throws Exception { + doThrow(new IOException("boom")).when(mockGcsUtil).copy(any(), any()); + + MetricsContainerImpl container = new MetricsContainerImpl(null); + try (Closeable ignored = MetricsEnvironment.scopedMetricsContainer(container)) { + fileSystemWithMetrics(true).copy(resourceIds("gs://bucket/a"), resourceIds("gs://bucket/b")); + fail("Expected the copy to propagate the IOException"); + } catch (IOException expected) { + // The point of the test is that the finally block still recorded the attempt. + } + assertEquals(1L, counter(container, "gcs_op_copy_count")); + } } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java index 2f77f15dcffc..52bb877bb96c 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcsUtilTest.java @@ -1703,7 +1703,8 @@ public static GcsUtilV1Mock createMock(PipelineOptions options) { : null, gcsOptions.getEnableBucketWriteMetricCounter() ? gcsOptions.getGcsWriteCounterPrefix() - : null), + : null, + Boolean.TRUE.equals(gcsOptions.getGcsPerformanceMetrics())), gcsOptions.getGoogleCloudStorageReadOptions()); } @@ -1731,7 +1732,10 @@ private GcsUtilV1Mock( @Override GoogleCloudStorage createGoogleCloudStorage( - GoogleCloudStorageOptions options, Storage storage, Credentials credentials) { + GoogleCloudStorageOptions options, + Storage storage, + Credentials credentials, + @Nullable HttpRequestInitializer httpRequestInitializer) { return googleCloudStorage; } } diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/TransportTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/TransportTest.java index a290d1d78b6b..49c636eaa415 100644 --- a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/TransportTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/TransportTest.java @@ -21,15 +21,22 @@ import static org.hamcrest.Matchers.greaterThan; import static org.hamcrest.Matchers.greaterThanOrEqualTo; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import com.google.api.client.http.GenericUrl; import com.google.api.client.http.HttpRequest; +import com.google.api.client.http.HttpRequestInitializer; +import com.google.api.client.testing.http.MockHttpTransport; +import com.google.api.client.testing.http.MockLowLevelHttpResponse; import com.google.api.services.storage.Storage; import java.io.IOException; import java.util.Arrays; import java.util.Collections; +import org.apache.beam.runners.core.metrics.MetricsContainerImpl; import org.apache.beam.sdk.extensions.gcp.auth.TestCredential; import org.apache.beam.sdk.extensions.gcp.options.GcsOptions; import org.apache.beam.sdk.extensions.gcp.options.GcsOptions.GcsCustomAuditEntries; +import org.apache.beam.sdk.metrics.MetricName; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.util.ReleaseInfo; import org.junit.Test; @@ -90,4 +97,103 @@ public void testUserAgentAndCustomAuditInGcsRequestHeaders() throws IOException request.getHeaders().getHeaderStringValues("x-goog-custom-audit-status"), Collections.singletonList("ok")); } + + private static final String READ_PREFIX = "gcs_http_read_"; + private static final String WRITE_PREFIX = "gcs_http_write_"; + + private static long counter(MetricsContainerImpl container, String name) { + return container.getCounter(MetricName.named(GcsUtil.METRIC_NAMESPACE, name)).getCumulative(); + } + + /** + * Executes one request through a metrics-wrapped initializer against a mock transport, and + * returns the container the counters were recorded against. An empty {@code range} sends no Range + * header. + */ + private static MetricsContainerImpl executeRequest( + boolean isWrite, String method, int statusCode, String range) throws IOException { + MetricsContainerImpl container = new MetricsContainerImpl(null); + MockHttpTransport transport = + new MockHttpTransport.Builder() + .setLowLevelHttpResponse(new MockLowLevelHttpResponse().setStatusCode(statusCode)) + .build(); + HttpRequest request = + transport + .createRequestFactory(Transport.withMetricsContainer(req -> {}, container, isWrite)) + .buildRequest(method, new GenericUrl("https://storage.googleapis.com/test"), null); + if (!range.isEmpty()) { + request.getHeaders().setRange(range); + } + // Observe the status that was actually returned, rather than throwing or following it. + request.setThrowExceptionOnExecuteError(false); + request.setFollowRedirects(false); + request.execute(); + return container; + } + + @Test + public void testReadMetricsCountUnboundedGets() throws IOException { + MetricsContainerImpl container = executeRequest(false, "GET", 200, ""); + + assertEquals(1, counter(container, READ_PREFIX + "request_count")); + assertEquals(1, counter(container, READ_PREFIX + "request_count_unbounded")); + assertEquals(0, counter(container, READ_PREFIX + "request_count_ranged")); + assertEquals(0, counter(container, READ_PREFIX + "request_count_other")); + assertEquals(1, counter(container, READ_PREFIX + "status_2xx")); + assertEquals(0, counter(container, READ_PREFIX + "request_no_response")); + } + + @Test + public void testReadMetricsCountRangedGets() throws IOException { + MetricsContainerImpl container = executeRequest(false, "GET", 206, "bytes=0-9"); + + assertEquals(1, counter(container, READ_PREFIX + "request_count")); + assertEquals(1, counter(container, READ_PREFIX + "request_count_ranged")); + assertEquals(0, counter(container, READ_PREFIX + "request_count_unbounded")); + assertEquals(0, counter(container, READ_PREFIX + "request_count_other")); + // 206 Partial Content is still a success. + assertEquals(1, counter(container, READ_PREFIX + "status_2xx")); + } + + @Test + public void testReadMetricsClassifyNonGetsAsOther() throws IOException { + MetricsContainerImpl container = executeRequest(false, "POST", 200, ""); + + assertEquals(1, counter(container, READ_PREFIX + "request_count")); + assertEquals(1, counter(container, READ_PREFIX + "request_count_other")); + assertEquals(0, counter(container, READ_PREFIX + "request_count_ranged")); + assertEquals(0, counter(container, READ_PREFIX + "request_count_unbounded")); + } + + @Test + public void testWriteMetricsAreNotClassifiedByRequestShape() throws IOException { + MetricsContainerImpl container = executeRequest(true, "POST", 200, ""); + + assertEquals(1, counter(container, WRITE_PREFIX + "request_count")); + assertEquals(1, counter(container, WRITE_PREFIX + "status_2xx")); + // Writes are POSTs and PUTs by construction, so the shape counters are never allocated. + assertEquals(0, counter(container, WRITE_PREFIX + "request_count_ranged")); + assertEquals(0, counter(container, WRITE_PREFIX + "request_count_unbounded")); + assertEquals(0, counter(container, WRITE_PREFIX + "request_count_other")); + } + + @Test + public void testResponsesAreCountedByStatusClass() throws IOException { + // A resumable upload answers 308 to every chunk but the last, so 3xx is not an error. + assertEquals(1, counter(executeRequest(true, "PUT", 308, ""), WRITE_PREFIX + "status_3xx")); + assertEquals(1, counter(executeRequest(false, "GET", 404, ""), READ_PREFIX + "status_4xx")); + assertEquals(1, counter(executeRequest(false, "GET", 503, ""), READ_PREFIX + "status_5xx")); + + // Each of those is still exactly one request, and none of them lands in 2xx. + MetricsContainerImpl notFound = executeRequest(false, "GET", 404, ""); + assertEquals(1, counter(notFound, READ_PREFIX + "request_count")); + assertEquals(0, counter(notFound, READ_PREFIX + "status_2xx")); + } + + @Test + public void testInitializerIsUnchangedWithoutAContainer() { + HttpRequestInitializer base = request -> {}; + assertSame(base, Transport.withMetricsContainer(base, null, false)); + assertSame(base, Transport.withMetricsContainer(base, null, true)); + } } diff --git a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java index d0ea19ffdf85..35155c8cec62 100644 --- a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java +++ b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java @@ -216,10 +216,10 @@ private Set> fillMetricSuppliers( if (gatherGcsPerformanceMetrics) { metricSuppliers.add( reader -> { - MetricsReader actualReader = - reader.withNamespace("org.apache.beam.sdk.extensions.gcp.storage.GcsFileSystem"); - long numRenames = actualReader.getCounterMetric("num_renames"); - long renameTimeMsec = actualReader.getCounterMetric("rename_time_msec"); + // Namespace and names are defined by GcsUtil.METRIC_NAMESPACE / GcsFileSystem. + MetricsReader actualReader = reader.withNamespace("Gcs"); + long numRenames = actualReader.getCounterMetric("gcs_op_rename_count"); + long renameTimeMsec = actualReader.getCounterMetric("gcs_op_rename_msec"); double remamePerSec = (numRenames < 0 || renameTimeMsec < 0) ? -1 : numRenames / (renameTimeMsec / 1e3); return NamedTestResult.create(uuid, timestamp, "rename_per_sec", remamePerSec);