feat(spec): add containerName field to DAG-level container (#1496)

internal/core/container.go

@@ -7,6 +7,8 @@ import (
 
 // Container defines the container configuration for the DAG.
 type Container struct {
+	// Name is the container name to use. If empty, Docker generates a random name.
+	Name string `yaml:"name,omitempty"`
 	// Image is the container image to use.
 	Image string `yaml:"image,omitempty"`
 	// PullPolicy is the policy to pull the image (e.g., "Always", "IfNotPresent").

internal/core/spec/builder.go

@@ -330,6 +330,7 @@ func buildContainer(ctx BuildContext, spec *definition, dag *core.DAG) error {
 	}
 
 	container := core.Container{
+		Name:          strings.TrimSpace(spec.Container.Name),
 		Image:         spec.Container.Image,
 		PullPolicy:    pullPolicy,
 		Env:           envs,

internal/core/spec/builder_test.go

@@ -2836,6 +2836,54 @@ steps:
 		assert.Equal(t, core.PullPolicyAlways, dag.Container.PullPolicy)
 	})
 
+	t.Run("ContainerWithName", func(t *testing.T) {
+		yaml := `
+container:
+  name: my-dag-container
+  image: alpine:latest
+steps:
+  - name: step1
+    command: echo hello
+`
+		ctx := context.Background()
+		dag, err := spec.LoadYAML(ctx, []byte(yaml))
+		require.NoError(t, err)
+		require.NotNil(t, dag.Container)
+		assert.Equal(t, "my-dag-container", dag.Container.Name)
+		assert.Equal(t, "alpine:latest", dag.Container.Image)
+	})
+
+	t.Run("ContainerNameEmpty", func(t *testing.T) {
+		yaml := `
+container:
+  image: alpine:latest
+steps:
+  - name: step1
+    command: echo hello
+`
+		ctx := context.Background()
+		dag, err := spec.LoadYAML(ctx, []byte(yaml))
+		require.NoError(t, err)
+		require.NotNil(t, dag.Container)
+		assert.Equal(t, "", dag.Container.Name)
+	})
+
+	t.Run("ContainerNameTrimmed", func(t *testing.T) {
+		yaml := `
+container:
+  name: "  my-container  "
+  image: alpine:latest
+steps:
+  - name: step1
+    command: echo hello
+`
+		ctx := context.Background()
+		dag, err := spec.LoadYAML(ctx, []byte(yaml))
+		require.NoError(t, err)
+		require.NotNil(t, dag.Container)
+		assert.Equal(t, "my-container", dag.Container.Name)
+	})
+
 	t.Run("ContainerWithAllFields", func(t *testing.T) {
 		yaml := `
 container:

internal/core/spec/definition.go

@@ -216,6 +216,8 @@ type mailOnDef struct {
 
 // containerDef defines the container configuration for the DAG.
 type containerDef struct {
+	// Name is the container name to use. If empty, Docker generates a random name.
+	Name string `yaml:"name,omitempty"`
 	// Image is the container image to use.
 	Image string `yaml:"image,omitempty"`
 	// PullPolicy is the policy to pull the image (e.g., "Always", "IfNotPresent").

internal/runtime/builtin/docker/client.go

@@ -164,6 +164,23 @@ func (c *Client) CreateContainerKeepAlive(ctx context.Context) error {
 		return fmt.Errorf("container already exists. id=%s", c.containerID)
 	}
 
+	// Check if a container with the specified name already exists
+	if name := c.cfg.ContainerName; name != "" {
+		info, err := c.cli.ContainerInspect(ctx, name)
+		if err == nil {
+			// Container exists - fail regardless of state
+			if info.State != nil && info.State.Running {
+				return fmt.Errorf("container with name %q already exists and is running", name)
+			}
+			return fmt.Errorf("container with name %q already exists", name)
+		}
+		// If error is not "not found", it's an unexpected error
+		if !errdefs.IsNotFound(err) {
+			return fmt.Errorf("failed to check existing container %q: %w", name, err)
+		}
+		// Container doesn't exist, proceed to create
+	}
+
 	// Choose startup mode and command
 	var cmd []string
 	mode := c.cfg.Startup

internal/runtime/builtin/docker/client_test.go

@@ -1109,6 +1109,41 @@ func TestLoadConfig(t *testing.T) {
 				ExecOptions: &container.ExecOptions{},
 			},
 		},
+		{
+			name: "ContainerNamePropagation",
+			input: core.Container{
+				Name:  "my-dag-container",
+				Image: "alpine",
+			},
+			expected: &Config{
+				ContainerName: "my-dag-container",
+				Image:         "alpine",
+				AutoRemove:    true,
+				Container: &container.Config{
+					Image: "alpine",
+				},
+				Host:        &container.HostConfig{},
+				Network:     &network.NetworkingConfig{},
+				ExecOptions: &container.ExecOptions{},
+			},
+		},
+		{
+			name: "ContainerNameEmptyWhenNotSpecified",
+			input: core.Container{
+				Image: "alpine",
+			},
+			expected: &Config{
+				ContainerName: "",
+				Image:         "alpine",
+				AutoRemove:    true,
+				Container: &container.Config{
+					Image: "alpine",
+				},
+				Host:        &container.HostConfig{},
+				Network:     &network.NetworkingConfig{},
+				ExecOptions: &container.ExecOptions{},
+			},
+		},
 	}
 
 	for _, tt := range tests {
@@ -1122,6 +1157,7 @@ func TestLoadConfig(t *testing.T) {
 			}
 
 			require.NoError(t, err)
+			assert.Equal(t, tt.expected.ContainerName, result.ContainerName)
 			assert.Equal(t, tt.expected.Image, result.Image)
 			assert.Equal(t, tt.expected.Platform, result.Platform)
 			assert.Equal(t, tt.expected.Pull, result.Pull)

internal/runtime/builtin/docker/config.go

@@ -224,19 +224,20 @@ func LoadConfig(workDir string, ct core.Container, registryAuths map[string]*cor
 	}
 
 	return loadDefaults(&Config{
-		Image:       ct.Image,
-		Platform:    ct.Platform,
-		Pull:        ct.PullPolicy,
-		AutoRemove:  autoRemove,
-		Container:   containerConfig,
-		Host:        hostConfig,
-		Network:     networkConfig,
-		ExecOptions: execOptions,
-		Startup:     strings.ToLower(strings.TrimSpace(string(ct.Startup))),
-		WaitFor:     strings.ToLower(strings.TrimSpace(string(ct.WaitFor))),
-		LogPattern:  ct.LogPattern,
-		StartCmd:    append([]string{}, ct.Command...),
-		AuthManager: authManager,
+		ContainerName: ct.Name,
+		Image:         ct.Image,
+		Platform:      ct.Platform,
+		Pull:          ct.PullPolicy,
+		AutoRemove:    autoRemove,
+		Container:     containerConfig,
+		Host:          hostConfig,
+		Network:       networkConfig,
+		ExecOptions:   execOptions,
+		Startup:       strings.ToLower(strings.TrimSpace(string(ct.Startup))),
+		WaitFor:       strings.ToLower(strings.TrimSpace(string(ct.WaitFor))),
+		LogPattern:    ct.LogPattern,
+		StartCmd:      append([]string{}, ct.Command...),
+		AuthManager:   authManager,
 	}), nil
 }
 

schemas/dag.schema.json

@@ -1135,6 +1135,10 @@
         }
       ],
       "properties": {
+        "name": {
+          "type": "string",
+          "description": "Custom container name. If empty, Docker generates a random name. Must be unique - if a container with this name already exists (running or stopped), the DAG will fail."
+        },
         "image": {
           "type": "string",
           "description": "Container image to use (e.g., 'python:3.11', 'node:20'). Required when using container configuration."