Direct cloud logging

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -757,6 +757,7 @@ class BeamModulePlugin implements Plugin<Project> {
         google_cloud_datastore_v1_proto_client      : "com.google.cloud.datastore:datastore-v1-proto-client:2.34.0",   // [bomupgrader] sets version
         google_cloud_firestore                      : "com.google.cloud:google-cloud-firestore", // google_cloud_platform_libraries_bom sets version
         google_cloud_kms                            : "com.google.cloud:google-cloud-kms", // google_cloud_platform_libraries_bom sets version
+        google_cloud_logging                        : "com.google.cloud:google-cloud-logging", // google_cloud_platform_libraries_bom sets version
         google_cloud_pubsub                         : "com.google.cloud:google-cloud-pubsub", // google_cloud_platform_libraries_bom sets version
         // [bomupgrader] the BOM version is set by scripts/tools/bomupgrader.py. If update manually, also update
         // libraries-bom version on sdks/java/container/license_scripts/dep_urls_java.yaml
@@ -772,7 +773,6 @@ class BeamModulePlugin implements Plugin<Project> {
         google_http_client_apache_v2                : "com.google.http-client:google-http-client-apache-v2", // google_cloud_platform_libraries_bom sets version
         google_http_client_gson                     : "com.google.http-client:google-http-client-gson", // google_cloud_platform_libraries_bom sets version
         google_http_client_jackson                  : "com.google.http-client:google-http-client-jackson:1.29.2",
-        google_http_client_gson                     : "com.google.http-client:google-http-client-gson", // google_cloud_platform_libraries_bom sets version
         google_http_client_protobuf                 : "com.google.http-client:google-http-client-protobuf", // google_cloud_platform_libraries_bom sets version
         google_oauth_client                         : "com.google.oauth-client:google-oauth-client:$google_oauth_clients_version",
         google_oauth_client_java6                   : "com.google.oauth-client:google-oauth-client-java6:$google_oauth_clients_version",

runners/google-cloud-dataflow-java/build.gradle

@@ -110,6 +110,8 @@ dependencies {
   implementation library.java.google_http_client
   implementation library.java.google_http_client_gson
   permitUnusedDeclared library.java.google_http_client_gson // BEAM-11761
+  implementation library.java.google_cloud_logging
+  permitUnusedDeclared library.java.google_cloud_logging // BEAM-11761
   implementation library.java.hamcrest
   implementation library.java.jackson_annotations
   implementation library.java.jackson_core
@@ -164,7 +166,9 @@ def legacyPipelineOptions = [
   "--region=${gcpRegion}",
   "--tempRoot=${dataflowValidatesTempRoot}",
   "--dataflowWorkerJar=${dataflowLegacyWorkerJar}",
-  "--experiments=enable_lineage"
+  "--experiments=enable_lineage",
+  "--defaultWorkerDirectLoggerLevel=INFO",
+  "--workerDirectLoggerBufferByteLimit=1",
 ]
 
 // For the following test tasks using legacy worker, set workerHarnessContainerImage to empty to
@@ -230,7 +234,7 @@ def createLegacyWorkerValidatesRunnerTest = { Map args ->
 
     // Increase test parallelism up to the number of Gradle workers. By default this is equal
     // to the number of CPU cores, but can be increased by setting --max-workers=N.
-    maxParallelForks Integer.MAX_VALUE
+    maxParallelForks 1
     classpath = configurations.validatesRunner
     testClassesDirs = files(project(":sdks:java:core").sourceSets.test.output.classesDirs) +
       files(project(project.path).sourceSets.test.output.classesDirs)
@@ -650,7 +654,7 @@ task googleCloudPlatformLegacyWorkerIntegrationTest(type: Test, dependsOn: copyG
     "--tempRoot=${dataflowPostCommitTempRoot}",
     "--dataflowWorkerJar=${dataflowLegacyWorkerJar}",
     "--workerHarnessContainerImage=",
-    "--firestoreDb=${firestoreDb}",
+    "--firestoreDb=${firestoreDb}"
   ])
 
   include '**/*IT.class'
@@ -842,6 +846,18 @@ tasks.register("examplesJavaRunnerV2PreCommit", Test.class) {
   useJUnit { }
 }
 
+tasks.register("loggingTest", Test.class) {
+  group = "Verification"
+  dependsOn buildAndPushDockerJavaContainer
+  systemProperty "beamTestPipelineOptions", JsonOutput.toJson(legacyPipelineOptions + ['--streaming', ''],)
+  include '**/WordCountIT.class'
+
+  maxParallelForks 4
+  classpath = configurations.examplesJavaIntegrationTest
+  testClassesDirs = files(project(":examples:java").sourceSets.test.output.classesDirs)
+  useJUnit { }
+}
+
 tasks.register("examplesJavaDistrolessRunnerV2PreCommit", Test.class) {
   group = "Verification"
   dependsOn buildAndPushDistrolessDockerJavaContainer

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowWorkerLoggingOptions.java

@@ -66,6 +66,28 @@ enum Level {
 
   void setDefaultWorkerLogLevel(Level level);
 
+  /**
+   * Controls the log level given to messages printed to {@code System.err}.
+   *
+   * <p>Note that the message may be filtered depending on the {@link #getDefaultWorkerLogLevel
+   * defaultWorkerLogLevel} or if a {@code System.err} override is specified via {@link
+   * #getWorkerLogLevelOverrides workerLogLevelOverrides}.
+   */
+  @Description(
+      "Controls the threshold log level at which logs are uploaded directly to Cloud Logging instead of "
+          + "disk-based logging. Logs that are at this level or below are uploaded directly. "
+          + "OFF means no logs will be sent directly.")
+  @Default.Enum("OFF")
+  Level getDefaultWorkerDirectLoggerLevel();
+
+  void setDefaultWorkerDirectLoggerLevel(Level level);
+
+  @Description("The maximum buffer size used for direct cloud logging records")
+  @Default.Long(100L * 1024 * 1024)
+  Long getWorkerDirectLoggerBufferByteLimit();
+
+  void setWorkerDirectLoggerBufferByteLimit(Long value);
+
   /**
    * Controls the log level given to messages printed to {@code System.out}.
    *

runners/google-cloud-dataflow-java/worker/build.gradle

@@ -38,6 +38,7 @@ def sdk_provided_dependencies = [
         library.java.avro,
         library.java.google_http_client,
         library.java.google_http_client_gson,
+        library.java.google_cloud_logging,
         library.java.jackson_annotations,
         library.java.jackson_core,
         library.java.jackson_databind,
@@ -193,6 +194,7 @@ dependencies {
 
     implementation library.java.google_auth_library_credentials
     implementation library.java.proto_google_common_protos
+    implementation library.java.google_cloud_logging
 
     // Conscrypt shouldn't be included here because Conscrypt won't work when being shaded.
     // (Context: https://github.com/apache/beam/pull/13846)