Sched core implementation

VERSION

@@ -1 +1 @@
-26.01.1-edge
+26.02.0-edge

docs/executor.md

@@ -414,6 +414,81 @@ Resource requests and other job characteristics can be controlled via the follow
 - {ref}`process-queue`
 - {ref}`process-time`
 
+(seqera-executor)=
+
+## Seqera
+
+:::{versionadded} 26.04.0
+:::
+
+:::{warning}
+*Preview feature: may change in a future release.*
+:::
+
+The `seqera` executor allows you to run your pipeline using the [Seqera](https://seqera.io) cloud infrastructure. It enables the seamless execution of Nextflow pipelines by offloading process executions to the Seqera scheduler service.
+
+The pipeline processes must specify the Docker image to use by defining the `container` directive, either in the pipeline script or the `nextflow.config` file. Additionally, an S3 bucket must be used as the pipeline work directory.
+
+To enable this executor, set `process.executor = 'seqera'` in the `nextflow.config` file.
+
+Resource requests and other job characteristics can be controlled via the following process directives:
+
+- {ref}`process-arch`
+- {ref}`process-container`
+- {ref}`process-containerOptions`
+- {ref}`process-cpus`
+- {ref}`process-disk`
+- {ref}`process-memory`
+- {ref}`process-time`
+
+### Disk support
+
+When the {ref}`process-disk` directive is specified, the Seqera executor provisions storage for the task container. There are two disk allocation strategies:
+
+- **task** (default): A dedicated EBS volume is created for each task at launch time. This provides isolated, high-performance storage with configurable volume type, IOPS, throughput, and encryption.
+
+- **node**: Uses the instance storage attached at the cluster level. This is shared across tasks running on the same node and does not support EBS-specific options.
+
+#### Task allocation (EBS volumes)
+
+By default, a gp3 volume with 325 MiB/s throughput is used (Fusion recommended settings). You can customize the EBS volume configuration:
+
+```groovy
+seqera {
+    executor {
+        machineRequirement {
+            diskAllocation = 'task'    // Per-task EBS volume (default)
+            diskType = 'ebs/io1'       // Use provisioned IOPS SSD
+            diskIops = 10000           // Required for io1/io2
+            diskThroughputMiBps = 500  // Throughput for gp3 volumes
+            diskEncrypted = true       // Enable KMS encryption
+        }
+    }
+}
+```
+
+Supported volume types: `ebs/gp3` (default), `ebs/gp2`, `ebs/io1`, `ebs/io2`, `ebs/st1`, `ebs/sc1`.
+
+#### Node allocation (instance storage)
+
+To use instance storage instead of per-task EBS volumes:
+
+```groovy
+seqera {
+    executor {
+        machineRequirement {
+            diskAllocation = 'node'    // Use instance storage
+        }
+    }
+}
+```
+
+:::{note}
+When using `node` allocation, the EBS-specific options (`diskType`, `diskIops`, `diskThroughputMiBps`, `diskEncrypted`) are not applicable and will cause an error if specified.
+:::
+
+See the {ref}`seqera scope <config-seqera>` for the available configuration options.
+
 (slurm-executor)=
 
 ## SLURM

docs/reference/config.md

@@ -1487,6 +1487,82 @@ The following settings are available:
 `singularity.runOptions`
 : Specify extra command line options supported by `singularity exec`.
 
+(config-seqera)=
+
+## `seqera`
+
+:::{versionadded} 26.04.0
+:::
+
+:::{warning}
+*Preview feature: may change in a future release.*
+:::
+
+The `seqera` scope allows you to configure the interactions with Seqera services.
+
+### `executor`
+
+The `seqera.executor` scope configures the Seqera scheduler service for the {ref}`seqera-executor`.
+
+The following settings are available:
+
+`seqera.executor.endpoint`
+: The Seqera scheduler service endpoint URL (required).
+
+`seqera.executor.region`
+: The AWS region for task execution (default: `'eu-central-1'`).
+
+`seqera.executor.autoLabels`
+: When `true`, automatically adds workflow metadata labels to the session with the `nextflow.io/` prefix (default: `false`). The following labels are added: `projectName`, `userName`, `runName`, `sessionId`, `resume`, `revision`, `commitId`, `repository`, `manifestName`, `runtimeVersion`. A `seqera.io/runId` label is also added, computed as a SipHash of the session ID and run name.
+
+`seqera.executor.labels`
+: Custom labels to apply to AWS resources for cost tracking and resource organization. Labels are propagated to ECS tasks, capacity providers, and EC2 instances. When used together with `autoLabels`, user-defined labels take precedence over auto-generated labels.
+
+`seqera.executor.machineRequirement.arch`
+: The CPU architecture for task execution, e.g. `'x86_64'` or `'arm64'`.
+
+`seqera.executor.machineRequirement.provisioning`
+: The instance provisioning mode. Can be `'spot'`, `'ondemand'`, or `'spotFirst'`.
+
+`seqera.executor.machineRequirement.maxSpotAttempts`
+: The maximum number of spot retry attempts before falling back to on-demand. Only used when `provisioning` is `'spot'` or `'spotFirst'`.
+
+`seqera.executor.machineRequirement.machineFamilies`
+: List of acceptable EC2 instance families, e.g. `['m5', 'c5', 'r5']`.
+
+`seqera.executor.machineRequirement.diskAllocation`
+: The disk allocation strategy. Can be `'task'` (default) for per-task EBS volumes, or `'node'` for per-node instance storage. When using `'node'` allocation, EBS-specific options (`diskType`, `diskIops`, `diskThroughputMiBps`, `diskEncrypted`) are not applicable.
+
+`seqera.executor.machineRequirement.diskType`
+: The EBS volume type for task scratch disk. Supported types: `'ebs/gp3'` (default), `'ebs/gp2'`, `'ebs/io1'`, `'ebs/io2'`, `'ebs/st1'`, `'ebs/sc1'`. Only applicable when `diskAllocation` is `'task'`.
+
+`seqera.executor.machineRequirement.diskThroughputMiBps`
+: The throughput in MiB/s for gp3 volumes (125-1000). Default: `325` (Fusion recommended). Only applicable when `diskAllocation` is `'task'`.
+
+`seqera.executor.machineRequirement.diskIops`
+: The IOPS for io1/io2/gp3 volumes. Required for io1/io2 volume types. Only applicable when `diskAllocation` is `'task'`.
+
+`seqera.executor.machineRequirement.diskEncrypted`
+: Enable KMS encryption for the EBS volume (default: `false`). Only applicable when `diskAllocation` is `'task'`.
+
+`seqera.executor.taskEnvironment`
+: Custom environment variables to apply to all tasks submitted by the Seqera executor. These are merged with the Fusion environment variables, with Fusion variables taking precedence. For example: `taskEnvironment = [MY_VAR: 'value']`.
+
+`seqera.executor.retryPolicy.delay`
+: The initial delay when a failing HTTP request is retried (default: `'450ms'`).
+
+`seqera.executor.retryPolicy.maxDelay`
+: The maximum delay when a failing HTTP request is retried (default: `'90s'`).
+
+`seqera.executor.retryPolicy.maxAttempts`
+: The maximum number of retry attempts (default: `10`).
+
+`seqera.executor.retryPolicy.jitter`
+: The jitter factor for randomizing retry delays (default: `0.25`).
+
+`seqera.executor.retryPolicy.multiplier`
+: The multiplier for exponential backoff (default: `2.0`).
+
 (config-spack)=
 
 ## `spack`

modules/nextflow/src/main/groovy/nextflow/executor/res/DiskResource.groovy

@@ -22,9 +22,10 @@ import groovy.transform.ToString
 import nextflow.util.MemoryUnit
 
 /**
- * Models disk resource request
- * 
+ * Models disk resource request with support for cloud-specific options.
+ *
  * @author Ben Sherman <bentshermann@gmail.com>
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
  */
 @ToString(includeNames = true, includePackage = false)
 @CompileStatic
@@ -33,6 +34,11 @@ class DiskResource {
 
     final MemoryUnit request
     final String type
+    final Integer iops
+    final Integer throughput
+    final Boolean encrypted
+    final String filesystem
+    final String mountPath
 
     DiskResource( value ) {
         this(request: value)
@@ -43,10 +49,28 @@ class DiskResource {
 
         if( opts.type )
             this.type = opts.type as String
+        if( opts.iops )
+            this.iops = opts.iops as Integer
+        if( opts.throughput )
+            this.throughput = opts.throughput as Integer
+        if( opts.encrypted != null )
+            this.encrypted = opts.encrypted as Boolean
+        if( opts.filesystem )
+            this.filesystem = opts.filesystem as String
+        if( opts.mountPath )
+            this.mountPath = opts.mountPath as String
     }
 
     DiskResource withRequest(MemoryUnit value) {
-        return new DiskResource(request: value, type: this.type)
+        return new DiskResource(
+            request: value,
+            type: this.type,
+            iops: this.iops,
+            throughput: this.throughput,
+            encrypted: this.encrypted,
+            filesystem: this.filesystem,
+            mountPath: this.mountPath
+        )
     }
 
     private static MemoryUnit toMemoryUnit( value ) {

modules/nextflow/src/main/groovy/nextflow/script/PlatformMetadata.groovy

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2013-2025, Seqera Labs
+ *
+ * Licensed 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 nextflow.script
+
+import groovy.transform.CompileStatic
+import groovy.transform.EqualsAndHashCode
+import groovy.transform.ToString
+
+/**
+ * Models Seqera Platform metadata for Nextflow execution
+ *
+ * @author Paolo Di Tommaso <paolo.ditommaso@gmail.com>
+ */
+@CompileStatic
+@ToString(includeNames = true, includePackage = false)
+@EqualsAndHashCode
+class PlatformMetadata {
+
+    /**
+     * Volatile because it is written by TowerClient.onFlowCreate on the main thread
+     * and read by SeqeraExecutor.createRun on the executor thread.
+     */
+    volatile String workflowId
+
+    /**
+     * The Platform watch URL for the current workflow execution.
+     * Set by TowerClient.onFlowBegin and read by SeqeraExecutor.createRun.
+     */
+    volatile String workflowUrl
+
+    PlatformMetadata() {}
+
+    PlatformMetadata(String workflowId) {
+        this.workflowId = workflowId
+    }
+}

modules/nextflow/src/main/groovy/nextflow/script/WorkflowMetadata.groovy

@@ -34,7 +34,6 @@ import nextflow.exception.WorkflowScriptErrorException
 import nextflow.trace.WorkflowStats
 import nextflow.util.Duration
 import nextflow.util.TestOnly
-import org.codehaus.groovy.runtime.InvokerHelper
 /**
  * Models workflow metadata properties and notification handler
  *
@@ -218,6 +217,12 @@ class WorkflowMetadata {
      */
     FusionMetadata fusion
 
+    /**
+     * Metadata specific to Seqera Platform, including:
+     * <li>workflowId: the Platform-assigned workflow identifier
+     */
+    PlatformMetadata platform
+
     /**
      * The list of files that concurred to create the config object
      */
@@ -497,4 +502,14 @@ class WorkflowMetadata {
         session.statsObserver.getStats()
     }
 
+    PlatformMetadata getPlatform() {
+        if( platform!=null )
+            return platform
+        synchronized (this) {
+            if( platform!=null )
+                return platform
+            platform = new PlatformMetadata()
+        }
+        return platform
+    }
 }

modules/nextflow/src/main/groovy/nextflow/trace/TraceRecord.groovy

@@ -124,6 +124,7 @@ class TraceRecord implements Serializable {
     transient private CloudMachineInfo machineInfo
     transient private ContainerMeta containerMeta
     transient private Integer numSpotInterruptions
+    transient private String logStreamId
 
     /**
      * Convert the given value to a string
@@ -622,6 +623,14 @@ class TraceRecord implements Serializable {
         this.numSpotInterruptions = numSpotInterruptions
     }
 
+    String getLogStreamId() {
+        return logStreamId
+    }
+
+    void setLogStreamId(String logStreamId) {
+        this.logStreamId = logStreamId
+    }
+
     ContainerMeta getContainerMeta() {
         return containerMeta
     }

modules/nextflow/src/main/resources/META-INF/plugins-info.txt

@@ -5,5 +5,6 @@ nf-codecommit@0.5.0
 nf-console@1.3.0
 nf-google@1.26.0
 nf-k8s@1.5.0
+nf-seqera@0.6.0
 nf-tower@1.20.0
 nf-wave@1.18.0