feat: add DAG run resource limits (#2179)

Author: yohamta0

api/v1/api.gen.go

@@ -11065,6 +11065,8 @@ components:
           x-deprecated-reason: "For concurrency control, configure global queues in config.yaml instead"
         runConfig:
           $ref: "#/components/schemas/RunConfig"
+        resources:
+          $ref: "#/components/schemas/DAGResources"
       required:
         - name
 
@@ -11396,6 +11398,8 @@ components:
             type: string
         runConfig:
           $ref: "#/components/schemas/RunConfig"
+        resources:
+          $ref: "#/components/schemas/DAGResources"
       required:
         - name
 
@@ -11530,6 +11534,26 @@ components:
         - disableParamEdit
         - disableRunIdEdit
 
+    DAGResources:
+      type: object
+      additionalProperties: false
+      description: "Resource limits requested for a DAG run"
+      properties:
+        limits:
+          $ref: "#/components/schemas/DAGResourceLimits"
+
+    DAGResourceLimits:
+      type: object
+      additionalProperties: false
+      description: "CPU and memory limits requested for a DAG run"
+      properties:
+        cpu:
+          type: string
+          description: "CPU limit as cores (for example, \"2\" or \"0.5\") or millicores (for example, \"500m\")"
+        memory:
+          type: string
+          description: "Memory limit in bytes or with a unit suffix (for example, \"512Mi\", \"1Gi\", or \"2G\")"
+
     DAGArtifactsConfig:
       type: object
       description: "Configuration for DAG run artifact storage"

api/v1/api.yaml

@@ -758,6 +758,10 @@
       "$ref": "#/definitions/container",
       "description": "Default container configuration for all steps in the DAG. Steps can override this configuration with their own container settings."
     },
+    "resources": {
+      "$ref": "#/definitions/resources",
+      "description": "CPU and memory limits requested for this DAG run. Dagu attempts to enforce these limits where supported and warns while continuing when enforcement is unavailable."
+    },
     "registry_auths": {
       "oneOf": [
         {
@@ -926,6 +930,44 @@
     }
   },
   "definitions": {
+    "resources": {
+      "type": "object",
+      "additionalProperties": false,
+      "properties": {
+        "limits": {
+          "$ref": "#/definitions/resourceLimits"
+        }
+      },
+      "description": "Resource limits requested for a DAG run."
+    },
+    "resourceLimits": {
+      "type": "object",
+      "additionalProperties": false,
+      "properties": {
+        "cpu": {
+          "oneOf": [
+            {
+              "type": "string",
+              "pattern": "^0*[1-9][0-9]*m$"
+            },
+            {
+              "type": "string",
+              "pattern": "^[0-9]+(\\.[0-9]{1,3})?$",
+              "not": {
+                "pattern": "^0+(\\.0{1,3})?$"
+              }
+            }
+          ],
+          "description": "CPU limit as cores (e.g., \"2\", \"0.5\") or millicores (e.g., \"500m\")."
+        },
+        "memory": {
+          "type": "string",
+          "pattern": "^([1-9][0-9]*(\\.[0-9]+)?|0\\.[0-9]*[1-9][0-9]*)([KkMmGgTtPpEe]i?[Bb]?|[Bb])?$",
+          "description": "Memory limit in bytes or with a unit suffix (e.g., \"512Mi\", \"1Gi\", \"2G\")."
+        }
+      },
+      "description": "CPU and memory limits requested for a DAG run."
+    },
     "stepRetryPolicy": {
       "type": "object",
       "additionalProperties": false,

internal/cmn/schema/dag.schema.json

@@ -1084,6 +1084,106 @@ steps:
 	require.Contains(t, err.Error(), "retry_policy")
 }
 
+func TestDAGSchemaResources(t *testing.T) {
+	t.Parallel()
+
+	resolved := mustResolveDAGSchema(t)
+
+	tests := []struct {
+		name    string
+		spec    string
+		wantErr string
+	}{
+		{
+			name: "ValidLimits",
+			spec: `
+name: limited-dag
+resources:
+  limits:
+    cpu: "500m"
+    memory: "512Mi"
+steps:
+  - run: echo hi
+`,
+		},
+		{
+			name: "RejectsInvalidCPU",
+			spec: `
+name: limited-dag
+resources:
+  limits:
+    cpu: nope
+steps:
+  - run: echo hi
+`,
+			wantErr: "resources",
+		},
+		{
+			name: "RejectsSubMilliCPU",
+			spec: `
+name: limited-dag
+resources:
+  limits:
+    cpu: "0.0005"
+steps:
+  - run: echo hi
+`,
+			wantErr: "resources",
+		},
+		{
+			name: "RejectsFractionalMillicores",
+			spec: `
+name: limited-dag
+resources:
+  limits:
+    cpu: "0.5m"
+steps:
+  - run: echo hi
+`,
+			wantErr: "resources",
+		},
+		{
+			name: "RejectsInvalidMemory",
+			spec: `
+name: limited-dag
+resources:
+  limits:
+    memory: nope
+steps:
+  - run: echo hi
+`,
+			wantErr: "resources",
+		},
+		{
+			name: "RejectsUnknownLimitField",
+			spec: `
+name: limited-dag
+resources:
+  limits:
+    gpu: "1"
+steps:
+  - run: echo hi
+`,
+			wantErr: "resources",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			doc := mustParseYAMLDocument(t, tt.spec)
+			err := resolved.Validate(doc)
+			if tt.wantErr == "" {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			require.Contains(t, err.Error(), tt.wantErr)
+		})
+	}
+}
+
 func TestDAGSchemaStepRetryPolicyRejectsUnknownField(t *testing.T) {
 	t.Parallel()
 

internal/cmn/schema/dag_schema_test.go

@@ -218,6 +218,8 @@ type DAG struct {
 	Container *Container `json:"container,omitempty"`
 	// RunConfig contains configuration for controlling user interactions during DAG runs.
 	RunConfig *RunConfig `json:"runConfig,omitempty"`
+	// Resources contains CPU and memory limits requested for this DAG run.
+	Resources *Resources `json:"resources,omitempty"`
 	// Webhook contains DAG-level webhook trigger behavior configuration.
 	Webhook *WebhookConfig `json:"webhook,omitempty"`
 	// RegistryAuths maps registry hostnames to authentication configs.
@@ -375,6 +377,9 @@ func (d *DAG) Clone() *DAG {
 		artifactsCopy := *d.Artifacts
 		clone.Artifacts = &artifactsCopy
 	}
+	if d.Resources != nil {
+		clone.Resources = d.Resources.Clone()
+	}
 	if d.Webhook != nil {
 		webhookCopy := *d.Webhook
 		webhookCopy.ForwardHeaders = append([]string(nil), d.Webhook.ForwardHeaders...)