Fix #6182: Add Google Batch LogsPolicy PATH option for GCS bucket log storage

- Add logsBucket configuration option to BatchConfig
- Support both CLOUD_LOGGING (default) and PATH log destinations
- Add validation for GCS bucket paths (must start with gs://)
- Include comprehensive tests for new functionality
- Maintain backward compatibility

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>
Signed-off-by: David Glazer <[email protected]>

Author: 2 people

plugins/nf-google/src/main/nextflow/cloud/google/batch/GoogleBatchTaskHandler.groovy

@@ -447,14 +447,29 @@ class GoogleBatchTaskHandler extends TaskHandler implements FusionAwareTask {
         return Job.newBuilder()
             .addTaskGroups(taskGroup)
             .setAllocationPolicy(allocationPolicy)
-            .setLogsPolicy(
-                LogsPolicy.newBuilder()
-                    .setDestination(LogsPolicy.Destination.CLOUD_LOGGING)
-            )
+            .setLogsPolicy(createLogsPolicy())
             .putAllLabels(task.config.getResourceLabels())
             .build()
     }
 
+    /**
+     * Create the LogsPolicy based on configuration
+     * @return LogsPolicy configured for either PATH (GCS bucket) or CLOUD_LOGGING
+     */
+    protected LogsPolicy createLogsPolicy() {
+        final logsBucket = executor.batchConfig.logsBucket
+        if( logsBucket ) {
+            return LogsPolicy.newBuilder()
+                .setDestination(LogsPolicy.Destination.PATH)
+                .setLogsPath(logsBucket)
+                .build()
+        } else {
+            return LogsPolicy.newBuilder()
+                .setDestination(LogsPolicy.Destination.CLOUD_LOGGING)
+                .build()
+        }
+    }
+
     /**
      * @return Retrieve the submitted task state
      */

plugins/nf-google/src/main/nextflow/cloud/google/batch/client/BatchConfig.groovy

@@ -85,6 +85,12 @@ class BatchConfig implements ConfigScope {
     """)
     final boolean installGpuDrivers
 
+    @ConfigOption
+    @Description("""
+        The Google Cloud Storage bucket path where job logs should be stored, e.g. `gs://my-logs-bucket/logs`. When specified, Google Batch will write job logs to this location instead of Cloud Logging. The bucket must be accessible and writable by the service account.
+    """)
+    final String logsBucket
+
     @ConfigOption
     @Description("""
         Max number of execution attempts of a job interrupted by a Compute Engine Spot reclaim event (default: `0`).
@@ -142,6 +148,7 @@ class BatchConfig implements ConfigScope {
         cpuPlatform = opts.cpuPlatform
         gcsfuseOptions = opts.gcsfuseOptions as List<String> ?: DEFAULT_GCSFUSE_OPTS
         installGpuDrivers = opts.installGpuDrivers as boolean
+        logsBucket = validateLogsBucket(opts.logsBucket)
         maxSpotAttempts = opts.maxSpotAttempts != null ? opts.maxSpotAttempts as int : DEFAULT_MAX_SPOT_ATTEMPTS
         network = opts.network
         networkTags = opts.networkTags as List<String> ?: Collections.emptyList()
@@ -155,4 +162,17 @@ class BatchConfig implements ConfigScope {
 
     BatchRetryConfig getRetryConfig() { retry }
 
+    private static String validateLogsBucket(String bucket) {
+        if( !bucket )
+            return null
+
+        if( !bucket.startsWith('gs://') )
+            throw new IllegalArgumentException("Logs bucket path must start with 'gs://' - provided: $bucket")
+
+        if( bucket.length() <= 5 || bucket == 'gs://' )
+            throw new IllegalArgumentException("Invalid logs bucket path - provided: $bucket")
+
+        return bucket
+    }
+
 }

plugins/nf-google/src/test/nextflow/cloud/google/batch/GoogleBatchTaskHandlerTest.groovy

@@ -659,6 +659,52 @@ class GoogleBatchTaskHandlerTest extends Specification {
         "SUCCEEDED"  | JobStatus.State.FAILED     | makeTaskStatus(TaskStatus.State.SUCCEEDED, "") // get from task status
     }
 
+    def 'should create submit request with logs bucket PATH policy' () {
+        given:
+        def WORK_DIR = CloudStorageFileSystem.forBucket('foo').getPath('/scratch')
+        def CONTAINER_IMAGE = 'ubuntu:22.1'
+        def LOGS_BUCKET = 'gs://my-logs-bucket/logs'
+
+        def session = Mock(Session) {
+            getBucketDir() >> CloudStorageFileSystem.forBucket('foo').getPath('/')
+            getConfig() >> [google: [project: 'test-project', batch: [logsBucket: LOGS_BUCKET]]]
+        }
+
+        def exec = Mock(GoogleBatchExecutor) {
+            getSession() >> session
+            getBatchConfig() >> new BatchConfig([logsBucket: LOGS_BUCKET])
+            isFusionEnabled() >> false
+        }
+
+        def bean = new TaskBean(workDir: WORK_DIR, inputFiles: [:])
+        def task = Mock(TaskRun) {
+            toTaskBean() >> bean
+            getHashLog() >> 'abcd1234'
+            getWorkDir() >> WORK_DIR
+            getContainer() >> CONTAINER_IMAGE
+            getConfig() >> Mock(TaskConfig) {
+                getCpus() >> 2
+                getResourceLabels() >> [:]
+            }
+        }
+
+        def mounts = ['/mnt/disks/foo/scratch:/mnt/disks/foo/scratch:rw']
+        def volumes = [ GCS_VOL ]
+        def launcher = new GoogleBatchLauncherSpecMock('bash .command.run', mounts, volumes)
+
+        def handler = Spy(new GoogleBatchTaskHandler(task, exec))
+
+        when:
+        def req = handler.newSubmitRequest(task, launcher)
+        then:
+        handler.fusionEnabled() >> false
+        handler.findBestMachineType(_, false) >> null
+
+        and:
+        req.getLogsPolicy().getDestination().toString() == 'PATH'
+        req.getLogsPolicy().getLogsPath() == LOGS_BUCKET
+    }
+
     def makeTask(String name, TaskStatus.State state){
         Task.newBuilder().setName(name)
             .setStatus(TaskStatus.newBuilder().setState(state).build())

plugins/nf-google/src/test/nextflow/cloud/google/batch/client/BatchConfigTest.groovy

@@ -67,4 +67,38 @@ class BatchConfigTest extends Specification {
         config.bootDiskSize == MemoryUnit.of('100GB')
     }
 
+    @Requires({System.getenv('GOOGLE_APPLICATION_CREDENTIALS')})
+    def 'should validate logs bucket config' () {
+        when:
+        def config = new BatchConfig([logsBucket: 'gs://my-logs-bucket/logs'])
+        then:
+        config.logsBucket == 'gs://my-logs-bucket/logs'
+
+        when:
+        config = new BatchConfig([:])
+        then:
+        config.logsBucket == null
+    }
+
+    @Requires({System.getenv('GOOGLE_APPLICATION_CREDENTIALS')})
+    def 'should reject invalid logs bucket paths' () {
+        when:
+        new BatchConfig([logsBucket: 'invalid-bucket'])
+        then:
+        def e = thrown(IllegalArgumentException)
+        e.message.contains("Logs bucket path must start with 'gs://'")
+
+        when:
+        new BatchConfig([logsBucket: 'gs://'])
+        then:
+        e = thrown(IllegalArgumentException)
+        e.message.contains("Invalid logs bucket path")
+
+        when:
+        new BatchConfig([logsBucket: 's3://bucket'])
+        then:
+        e = thrown(IllegalArgumentException)
+        e.message.contains("Logs bucket path must start with 'gs://'")
+    }
+
 }