Add GCS bucket mounting logic for LogsPolicy PATH option

- Add helper methods to extract bucket names and convert GCS paths to mount paths
- Ensure logs bucket is mounted as Volume in GoogleBatchScriptLauncher
- Update LogsPolicy to use container mount paths instead of GCS paths
- Add comprehensive tests for bucket mounting and path conversion
- Addresses reviewer feedback about missing bucket mounting requirements

This ensures Google Batch can write logs to the specified GCS bucket by
properly mounting it before referencing it in the LogsPolicy PATH.

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

Co-Authored-By: Claude <[email protected]>
Signed-off-by: Emma Rogge <[email protected]>

Author: 2 people

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

@@ -186,6 +186,13 @@ class GoogleBatchScriptLauncher extends BashWrapperBuilder implements GoogleBatc
 
     GoogleBatchScriptLauncher withConfig(GoogleOpts config) {
         this.config = config
+        // Add logs bucket to mounted volumes if configured
+        if( config?.batch?.logsBucket ) {
+            final logsBucketName = config.batch.extractBucketName(config.batch.logsBucket)
+            if( logsBucketName ) {
+                buckets.add(logsBucketName)
+            }
+        }
         return this
     }
 

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

@@ -459,9 +459,10 @@ class GoogleBatchTaskHandler extends TaskHandler implements FusionAwareTask {
     protected LogsPolicy createLogsPolicy() {
         final logsBucket = executor.batchConfig.logsBucket
         if( logsBucket ) {
+            final containerPath = executor.batchConfig.convertGcsPathToMountPath(logsBucket)
             return LogsPolicy.newBuilder()
                 .setDestination(LogsPolicy.Destination.PATH)
-                .setLogsPath(logsBucket)
+                .setLogsPath(containerPath)
                 .build()
         } else {
             return LogsPolicy.newBuilder()

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

@@ -175,4 +175,31 @@ class BatchConfig implements ConfigScope {
         return bucket
     }
 
+    /**
+     * Extract the bucket name from a GCS path
+     * @param gcsPath GCS path like "gs://bucket-name/path/to/logs"
+     * @return bucket name like "bucket-name"
+     */
+    static String extractBucketName(String gcsPath) {
+        if( !gcsPath || !gcsPath.startsWith('gs://') )
+            return null
+
+        final pathWithoutProtocol = gcsPath.substring(5) // Remove "gs://"
+        final slashIndex = pathWithoutProtocol.indexOf('/')
+        return slashIndex > 0 ? pathWithoutProtocol.substring(0, slashIndex) : pathWithoutProtocol
+    }
+
+    /**
+     * Convert a GCS path to container mount path
+     * @param gcsPath GCS path like "gs://bucket-name/path/to/logs"
+     * @return container path like "/mnt/disks/bucket-name/path/to/logs"
+     */
+    static String convertGcsPathToMountPath(String gcsPath) {
+        if( !gcsPath || !gcsPath.startsWith('gs://') )
+            return gcsPath
+
+        final pathWithoutProtocol = gcsPath.substring(5) // Remove "gs://"
+        return "/mnt/disks/${pathWithoutProtocol}"
+    }
+
 }

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

@@ -702,7 +702,14 @@ class GoogleBatchTaskHandlerTest extends Specification {
 
         and:
         req.getLogsPolicy().getDestination().toString() == 'PATH'
-        req.getLogsPolicy().getLogsPath() == LOGS_BUCKET
+        req.getLogsPolicy().getLogsPath() == '/mnt/disks/my-logs-bucket/logs'
+        and:
+        def taskGroup = req.getTaskGroups(0)
+        def volumes = taskGroup.getTaskSpec().getVolumesList()
+        volumes.size() >= 2  // At least work dir volume and logs bucket volume
+        def logsBucketVolume = volumes.find { it.getGcs().getRemotePath() == 'my-logs-bucket' }
+        logsBucketVolume != null
+        logsBucketVolume.getMountPath() == '/mnt/disks/my-logs-bucket'
     }
 
     def makeTask(String name, TaskStatus.State state){

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

@@ -101,4 +101,23 @@ class BatchConfigTest extends Specification {
         e.message.contains("Logs bucket path must start with 'gs://'")
     }
 
+    def 'should extract bucket name from GCS path' () {
+        expect:
+        BatchConfig.extractBucketName('gs://my-bucket') == 'my-bucket'
+        BatchConfig.extractBucketName('gs://my-bucket/logs') == 'my-bucket'
+        BatchConfig.extractBucketName('gs://my-bucket/path/to/logs') == 'my-bucket'
+        BatchConfig.extractBucketName('gs://') == ''
+        BatchConfig.extractBucketName('invalid-path') == null
+        BatchConfig.extractBucketName(null) == null
+    }
+
+    def 'should convert GCS path to mount path' () {
+        expect:
+        BatchConfig.convertGcsPathToMountPath('gs://my-bucket') == '/mnt/disks/my-bucket'
+        BatchConfig.convertGcsPathToMountPath('gs://my-bucket/logs') == '/mnt/disks/my-bucket/logs'
+        BatchConfig.convertGcsPathToMountPath('gs://my-bucket/path/to/logs') == '/mnt/disks/my-bucket/path/to/logs'
+        BatchConfig.convertGcsPathToMountPath('invalid-path') == 'invalid-path'
+        BatchConfig.convertGcsPathToMountPath(null) == null
+    }
+
 }