[Dataflow Java Runner] Add support for sending logs directly to 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

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

@@ -66,6 +66,38 @@ enum Level {
 
   void setDefaultWorkerLogLevel(Level level);
 
+  /**
+   * Controls the log level for which messages are uploaded to Cloud Logging. If a message is
+   * configured to be sent to both directly to cloud logging and default disk-based logging it will
+   * just be sent to disk-based logging. This allows for configuration such as
+   * "--defaultWorkerLogLevel=WARN --defaultWorkerDirectLoggerLevel=INFO where INFO logs will be
+   * directly sent to cloud logging and WARN logs and higher will be sent to disk-based logging.
+   *
+   * <p>Note that this is just the default and may be overridden for specific classes with
+   * --workerDirectLogLevelOverrides.
+   */
+  @Description(
+      "Controls the default direct to Cloud Logging level of all logs without a log level override."
+          + "If a message is configured to be sent to both directly to cloud logging and default disk-based logging "
+          + "it will just be sent to disk-based logging.")
+  @Default.Enum("OFF")
+  Level getDefaultWorkerDirectLoggerLevel();
+
+  void setDefaultWorkerDirectLoggerLevel(Level level);
+
+  @Description(
+      "The maximum buffered bytes for records in the queue that are being sent directly to Cloud Logging.")
+  @Default.Long(100L * 1024 * 1024)
+  Long getWorkerDirectLoggerBufferByteLimit();
+
+  void setWorkerDirectLoggerBufferByteLimit(Long value);
+
+  @Description("The maximum buffered elements in the queue being sent directly to Cloud Logging.")
+  @Default.Long(1_000_000)
+  Long getWorkerDirectLoggerBufferElementLimit();
+
+  void setWorkerDirectLoggerBufferElementLimit(Long value);
+
   /**
    * Controls the log level given to messages printed to {@code System.out}.
    *
@@ -121,6 +153,48 @@ enum Level {
 
   void setWorkerLogLevelOverrides(WorkerLogLevelOverrides value);
 
+  /**
+   * This option controls the direct log levels for specifically named loggers. If a message is
+   * configured to be sent to both directly to cloud logging and default disk-based logging it will
+   * just be sent to disk-based logging. If an override only exists for a logger for direct logging,
+   * the --defaultWorkerLogLevel will be used for the non-direct configuration for the logger.
+   *
+   * <p>Later options with equivalent names override earlier options.
+   *
+   * <p>See {@link WorkerLogLevelOverrides} for more information on how to configure logging on a
+   * per {@link Class}, {@link Package}, or name basis. If used from the command line, the expected
+   * format is {"Name":"Level",...}, further details on {@link WorkerLogLevelOverrides#from}.
+   */
+  @Description(
+      "This option controls the direct log levels for specifically named loggers. "
+          + "The expected format is {\"Name\":\"Level\",...}. The Dataflow worker supports a logging "
+          + "hierarchy based off of names that are '.' separated. For example, by specifying the value "
+          + "{\"a.b.c.Foo\":\"DEBUG\"}, the logger for the class 'a.b.c.Foo' will be configured to "
+          + "output logs at the DEBUG level. Similarly, by specifying the value {\"a.b.c\":\"WARN\"}, "
+          + "all loggers underneath the 'a.b.c' package will be configured to output logs at the WARN "
+          + "level. System.out and System.err levels are configured via loggers of the corresponding "
+          + "name. Also, note that when multiple overrides are specified, the exact name followed by "
+          + "the closest parent takes precedence. Note that if an override is just provided for the direct log level "
+          + "for a logger, the default non-direct log level will be used for non-direct logs.")
+  WorkerLogLevelOverrides getWorkerDirectLogLevelOverrides();
+
+  void setWorkerDirectLogLevelOverrides(WorkerLogLevelOverrides value);
+
+  @Default.Boolean(true)
+  @Description(
+      "If true, when there are errors with sending logs directly to Cloud Logging, the logs will fallback to "
+          + "disk-based logging. If false, such logs will be dropped.")
+  Boolean getDirectLoggingFallbackToDiskOnErrors();
+
+  void setDirectLoggingFallbackToDiskOnErrors(Boolean value);
+
+  @Default.Integer(10)
+  @Description(
+      "If an error is encountered with sending logs directly to Cloud Logging, direct logging will not be attempted for this many seconds.")
+  Integer getDirectLoggingCooldownSeconds();
+
+  void setDirectLoggingCooldownSeconds(Integer value);
+
   /**
    * Defines a log level override for a specific class, package, or name.
    *

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

@@ -151,7 +151,6 @@ applyJavaNature(
             exclude "META-INF/LICENSE.txt"
             exclude "about.html"
         })
-
 /******************************************************************************/
 // Configure the worker root project
 
@@ -167,6 +166,11 @@ configurations {
 
 dependencies {
     implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
+    implementation library.java.gax
+    implementation library.java.google_cloud_core
+    implementation library.java.guava
+    permitUnusedDeclared library.java.guava
+    implementation library.java.protobuf_java
 
     // Note that any dependency that is modified here should also be modified within
     // runners/google-cloud-dataflow-java/worker/build.gradle using the rules provided
@@ -193,6 +197,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)