You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2025/01/07 02:42:30 UTC
(beam) branch master updated: Adds an SdkHarnessOption that controls whether logging is redirected through the FnApi (#33418)
This is an automated email from the ASF dual-hosted git repository.
yhu 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 e657d6c8cec Adds an SdkHarnessOption that controls whether logging is redirected through the FnApi (#33418)
e657d6c8cec is described below
commit e657d6c8cec8ee4d52566fee32ce752bb5798f69
Author: Tim Heckman <60...@users.noreply.github.com>
AuthorDate: Tue Jan 7 02:42:23 2025 +0000
Adds an SdkHarnessOption that controls whether logging is redirected through the FnApi (#33418)
* Add an SdkHarnessOption that controls whether logging is redirected through the FnApi logging service. Redirection through the logging service is enabled by default.
* Add an SdkHarnessOption that controls whether logging is redirected through the FnApi logging service. Redirection through the FnApi is enabled by default.
* include license in new files
* fix ManagedChannel dep
* fix invalid conversions
* fix missing imports
* fix type mismatch
* fix up tests
* continue to use anyOf when logViaFnApi is enabled
* More comments on the new SdkHarnessOption. DataflowRunner.run() forces
the option to 'enabled'.
---
.../beam/runners/dataflow/DataflowRunner.java | 3 ++
.../apache/beam/sdk/options/SdkHarnessOptions.java | 10 ++++
.../jmh/logging/BeamFnLoggingClientBenchmark.java | 6 ++-
.../java/org/apache/beam/fn/harness/FnHarness.java | 14 +++--
.../fn/harness/logging/BeamFnLoggingClient.java | 5 +-
.../beam/fn/harness/logging/LoggingClient.java | 25 +++++++++
.../fn/harness/logging/LoggingClientFactory.java | 59 ++++++++++++++++++++++
.../data/PCollectionConsumerRegistryTest.java | 7 +--
8 files changed, 118 insertions(+), 11 deletions(-)
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
index ce99958c57f..9ca6e95ed95 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
@@ -103,6 +103,7 @@ import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.ExperimentalOptions;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsValidator;
+import org.apache.beam.sdk.options.SdkHarnessOptions;
import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.runners.PTransformOverride;
@@ -1252,6 +1253,8 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> {
experiments.add("use_portable_job_submission");
}
options.setExperiments(ImmutableList.copyOf(experiments));
+ // Ensure that logging via the FnApi is enabled
+ options.as(SdkHarnessOptions.class).setEnableLogViaFnApi(true);
}
logWarningIfPCollectionViewHasNonDeterministicKeyCoder(pipeline);
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java
index 78ea34503e5..2981046a0a4 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java
@@ -110,6 +110,16 @@ public interface SdkHarnessOptions extends PipelineOptions, MemoryMonitorOptions
void setLogMdc(boolean value);
+ /** This option controls whether logging will be redirected through the FnApi. */
+ @Description(
+ "Controls whether logging will be redirected through the FnApi. In normal usage, setting "
+ + "this to a non-default value will cause log messages to be dropped.")
+ @Default.Boolean(true)
+ @Hidden
+ boolean getEnableLogViaFnApi();
+
+ void setEnableLogViaFnApi(boolean enableLogViaFnApi);
+
/**
* Size (in MB) of each grouping table used to pre-combine elements. Larger values may reduce the
* amount of data shuffled. If unset, defaults to 100 MB.
diff --git a/sdks/java/harness/jmh/src/main/java/org/apache/beam/fn/harness/jmh/logging/BeamFnLoggingClientBenchmark.java b/sdks/java/harness/jmh/src/main/java/org/apache/beam/fn/harness/jmh/logging/BeamFnLoggingClientBenchmark.java
index b9e4b20db00..745e1f07864 100644
--- a/sdks/java/harness/jmh/src/main/java/org/apache/beam/fn/harness/jmh/logging/BeamFnLoggingClientBenchmark.java
+++ b/sdks/java/harness/jmh/src/main/java/org/apache/beam/fn/harness/jmh/logging/BeamFnLoggingClientBenchmark.java
@@ -24,6 +24,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.beam.fn.harness.logging.BeamFnLoggingClient;
import org.apache.beam.fn.harness.logging.BeamFnLoggingMDC;
+import org.apache.beam.fn.harness.logging.LoggingClient;
+import org.apache.beam.fn.harness.logging.LoggingClientFactory;
import org.apache.beam.fn.harness.logging.QuotaEvent;
import org.apache.beam.model.fnexecution.v1.BeamFnApi;
import org.apache.beam.model.fnexecution.v1.BeamFnLoggingGrpc;
@@ -80,7 +82,7 @@ public class BeamFnLoggingClientBenchmark {
/** Setup a simple logging service and configure the {@link BeamFnLoggingClient}. */
@State(Scope.Benchmark)
public static class ManageLoggingClientAndService {
- public final BeamFnLoggingClient loggingClient;
+ public final LoggingClient loggingClient;
public final CallCountLoggingService loggingService;
public final Server server;
@@ -98,7 +100,7 @@ public class BeamFnLoggingClientBenchmark {
.build();
server.start();
loggingClient =
- BeamFnLoggingClient.createAndStart(
+ LoggingClientFactory.createAndStart(
PipelineOptionsFactory.create(),
apiServiceDescriptor,
managedChannelFactory::forDescriptor);
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java
index 9df9f12bc52..3c8784d0ee4 100644
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java
@@ -39,7 +39,8 @@ import org.apache.beam.fn.harness.control.HarnessMonitoringInfosInstructionHandl
import org.apache.beam.fn.harness.control.ProcessBundleHandler;
import org.apache.beam.fn.harness.data.BeamFnDataGrpcClient;
import org.apache.beam.fn.harness.debug.DataSampler;
-import org.apache.beam.fn.harness.logging.BeamFnLoggingClient;
+import org.apache.beam.fn.harness.logging.LoggingClient;
+import org.apache.beam.fn.harness.logging.LoggingClientFactory;
import org.apache.beam.fn.harness.state.BeamFnStateGrpcClientCache;
import org.apache.beam.fn.harness.status.BeamFnStatusClient;
import org.apache.beam.fn.harness.stream.HarnessStreamObserverFactories;
@@ -62,6 +63,7 @@ import org.apache.beam.sdk.metrics.MetricsEnvironment;
import org.apache.beam.sdk.options.ExecutorOptions;
import org.apache.beam.sdk.options.ExperimentalOptions;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.SdkHarnessOptions;
import org.apache.beam.sdk.util.construction.CoderTranslation;
import org.apache.beam.sdk.util.construction.PipelineOptionsTranslation;
import org.apache.beam.vendor.grpc.v1p60p1.com.google.protobuf.TextFormat;
@@ -283,8 +285,8 @@ public class FnHarness {
// The logging client variable is not used per se, but during its lifetime (until close()) it
// intercepts logging and sends it to the logging service.
- try (BeamFnLoggingClient logging =
- BeamFnLoggingClient.createAndStart(
+ try (LoggingClient logging =
+ LoggingClientFactory.createAndStart(
options, loggingApiServiceDescriptor, channelFactory::forDescriptor)) {
LOG.info("Fn Harness started");
// Register standard file systems.
@@ -410,7 +412,11 @@ public class FnHarness {
outboundObserverFactory,
executorService,
handlers);
- CompletableFuture.anyOf(control.terminationFuture(), logging.terminationFuture()).get();
+ if (options.as(SdkHarnessOptions.class).getEnableLogViaFnApi()) {
+ CompletableFuture.anyOf(control.terminationFuture(), logging.terminationFuture()).get();
+ } else {
+ control.terminationFuture().get();
+ }
if (beamFnStatusClient != null) {
beamFnStatusClient.close();
}
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
index 7812d8c0bc3..112104e4d25 100644
--- a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java
@@ -68,7 +68,7 @@ import org.slf4j.MDC;
/**
* Configures {@link java.util.logging} to send all {@link LogRecord}s via the Beam Fn Logging API.
*/
-public class BeamFnLoggingClient implements AutoCloseable {
+public class BeamFnLoggingClient implements LoggingClient {
private static final String ROOT_LOGGER_NAME = "";
private static final ImmutableMap<Level, BeamFnApi.LogEntry.Severity.Enum> LOG_LEVEL_MAP =
ImmutableMap.<Level, BeamFnApi.LogEntry.Severity.Enum>builder()
@@ -119,7 +119,7 @@ public class BeamFnLoggingClient implements AutoCloseable {
*/
private @Nullable Thread logEntryHandlerThread = null;
- public static BeamFnLoggingClient createAndStart(
+ static BeamFnLoggingClient createAndStart(
PipelineOptions options,
Endpoints.ApiServiceDescriptor apiServiceDescriptor,
Function<Endpoints.ApiServiceDescriptor, ManagedChannel> channelFactory) {
@@ -383,6 +383,7 @@ public class BeamFnLoggingClient implements AutoCloseable {
}
}
+ @Override
public CompletableFuture<?> terminationFuture() {
checkNotNull(bufferedLogConsumer, "BeamFnLoggingClient not fully started");
return bufferedLogConsumer;
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/LoggingClient.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/LoggingClient.java
new file mode 100644
index 00000000000..3c1972ec643
--- /dev/null
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/LoggingClient.java
@@ -0,0 +1,25 @@
+/*
+ * 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.fn.harness.logging;
+
+import java.util.concurrent.CompletableFuture;
+
+public interface LoggingClient extends AutoCloseable {
+
+ CompletableFuture<?> terminationFuture();
+}
diff --git a/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/LoggingClientFactory.java b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/LoggingClientFactory.java
new file mode 100644
index 00000000000..f61b8d3c4ba
--- /dev/null
+++ b/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/LoggingClientFactory.java
@@ -0,0 +1,59 @@
+/*
+ * 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.fn.harness.logging;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.beam.model.pipeline.v1.Endpoints;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.SdkHarnessOptions;
+import org.apache.beam.vendor.grpc.v1p60p1.io.grpc.ManagedChannel;
+
+/**
+ * A factory for {@link LoggingClient}s. Provides {@link BeamFnLoggingClient} if the logging service
+ * is enabled, otherwise provides a no-op client.
+ */
+public class LoggingClientFactory {
+
+ private LoggingClientFactory() {}
+
+ /**
+ * A factory for {@link LoggingClient}s. Provides {@link BeamFnLoggingClient} if the logging
+ * service is enabled, otherwise provides a no-op client.
+ */
+ public static LoggingClient createAndStart(
+ PipelineOptions options,
+ Endpoints.ApiServiceDescriptor apiServiceDescriptor,
+ Function<Endpoints.ApiServiceDescriptor, ManagedChannel> channelFactory) {
+ if (options.as(SdkHarnessOptions.class).getEnableLogViaFnApi()) {
+ return BeamFnLoggingClient.createAndStart(options, apiServiceDescriptor, channelFactory);
+ } else {
+ return new NoOpLoggingClient();
+ }
+ }
+
+ static final class NoOpLoggingClient implements LoggingClient {
+ @Override
+ public CompletableFuture<?> terminationFuture() {
+ return CompletableFuture.completedFuture(new Object());
+ }
+
+ @Override
+ public void close() throws Exception {}
+ }
+}
diff --git a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/data/PCollectionConsumerRegistryTest.java b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/data/PCollectionConsumerRegistryTest.java
index 245c87f3e19..96e68586b3a 100644
--- a/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/data/PCollectionConsumerRegistryTest.java
+++ b/sdks/java/harness/src/test/java/org/apache/beam/fn/harness/data/PCollectionConsumerRegistryTest.java
@@ -47,8 +47,9 @@ import org.apache.beam.fn.harness.control.BundleProgressReporter;
import org.apache.beam.fn.harness.control.ExecutionStateSampler;
import org.apache.beam.fn.harness.control.ExecutionStateSampler.ExecutionStateTracker;
import org.apache.beam.fn.harness.debug.DataSampler;
-import org.apache.beam.fn.harness.logging.BeamFnLoggingClient;
import org.apache.beam.fn.harness.logging.BeamFnLoggingMDC;
+import org.apache.beam.fn.harness.logging.LoggingClient;
+import org.apache.beam.fn.harness.logging.LoggingClientFactory;
import org.apache.beam.model.fnexecution.v1.BeamFnApi;
import org.apache.beam.model.fnexecution.v1.BeamFnApi.ProcessBundleDescriptor;
import org.apache.beam.model.fnexecution.v1.BeamFnLoggingGrpc;
@@ -647,8 +648,8 @@ public class PCollectionConsumerRegistryTest {
// Start the test within the logging context. This reroutes logging through to the boiler-plate
// that was set up
// earlier.
- try (BeamFnLoggingClient ignored =
- BeamFnLoggingClient.createAndStart(
+ try (LoggingClient ignored =
+ LoggingClientFactory.createAndStart(
PipelineOptionsFactory.create(),
apiServiceDescriptor,
(Endpoints.ApiServiceDescriptor descriptor) -> channel)) {