Upstream concourse/concourse state of the touched files before bb8b7aa7e9

atc/exec/task_config_source.go

@@ -0,0 +1,284 @@
+package exec
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"strings"
+
+	"code.cloudfoundry.org/lager/v3"
+	"code.cloudfoundry.org/lager/v3/lagerctx"
+	"github.com/concourse/concourse/atc"
+	"github.com/concourse/concourse/atc/exec/build"
+	"github.com/concourse/concourse/vars"
+	"github.com/concourse/concourse/worker/baggageclaim"
+	"sigs.k8s.io/yaml"
+)
+
+//counterfeiter:generate . TaskConfigSource
+
+// TaskConfigSource is used to determine a Task step's TaskConfig.
+type TaskConfigSource interface {
+	// FetchConfig returns the TaskConfig, and may have to a task config file out
+	// of the artifact.Repository.
+	FetchConfig(context.Context, lager.Logger, *build.Repository) (atc.TaskConfig, error)
+	Warnings() []string
+}
+
+// StaticConfigSource represents a statically configured TaskConfig.
+type StaticConfigSource struct {
+	Config *atc.TaskConfig
+}
+
+// FetchConfig returns the configuration.
+func (configSource StaticConfigSource) FetchConfig(context.Context, lager.Logger, *build.Repository) (atc.TaskConfig, error) {
+	taskConfig := atc.TaskConfig{}
+	if configSource.Config != nil {
+		taskConfig = *configSource.Config
+	}
+	return taskConfig, nil
+}
+
+func (configSource StaticConfigSource) Warnings() []string {
+	return []string{}
+}
+
+// FileConfigSource represents a dynamically configured TaskConfig, which will
+// be fetched from a specified file in the artifact.Repository.
+type FileConfigSource struct {
+	ConfigPath string
+	Streamer   Streamer
+}
+
+// FetchConfig reads the specified file from the artifact.Repository and loads the
+// TaskConfig contained therein (expecting it to be YAML format).
+//
+// The path must be in the format SOURCE_NAME/FILE/PATH.yml. The SOURCE_NAME
+// will be used to determine the StreamableArtifactSource in the artifact.Repository to
+// stream the file out of.
+//
+// If the source name is missing (i.e. if the path is just "foo.yml"),
+// UnspecifiedArtifactSourceError is returned.
+//
+// If the specified source name cannot be found, UnknownArtifactSourceError is
+// returned.
+//
+// If the task config file is not found, or is invalid YAML, or is an invalid
+// task configuration, the respective errors will be bubbled up.
+func (configSource FileConfigSource) FetchConfig(ctx context.Context, logger lager.Logger, repo *build.Repository) (atc.TaskConfig, error) {
+	segs := strings.SplitN(configSource.ConfigPath, "/", 2)
+	if len(segs) != 2 {
+		return atc.TaskConfig{}, UnspecifiedArtifactSourceError{configSource.ConfigPath}
+	}
+
+	sourceName := build.ArtifactName(segs[0])
+	filePath := segs[1]
+
+	artifact, _, found := repo.ArtifactFor(sourceName)
+	if !found {
+		return atc.TaskConfig{}, UnknownArtifactSourceError{sourceName, configSource.ConfigPath}
+	}
+	stream, err := configSource.Streamer.StreamFile(lagerctx.NewContext(ctx, logger), artifact, filePath)
+	if err != nil {
+		if err == baggageclaim.ErrFileNotFound {
+			return atc.TaskConfig{}, fmt.Errorf("task config '%s/%s' not found", sourceName, filePath)
+		}
+		return atc.TaskConfig{}, err
+	}
+
+	defer stream.Close()
+
+	byteConfig, err := io.ReadAll(stream)
+	if err != nil {
+		return atc.TaskConfig{}, err
+	}
+
+	config, err := atc.NewTaskConfig(byteConfig)
+	if err != nil {
+		return atc.TaskConfig{}, fmt.Errorf("failed to create task config from bytes %s: %s", configSource.ConfigPath, err)
+	}
+
+	return config, nil
+}
+
+func (configSource FileConfigSource) Warnings() []string {
+	return []string{}
+}
+
+// BaseResourceTypeDefaultsApplySource applies base resource type defaults to image_source.
+type BaseResourceTypeDefaultsApplySource struct {
+	ConfigSource  TaskConfigSource
+	ResourceTypes atc.ResourceTypes
+}
+
+func (configSource BaseResourceTypeDefaultsApplySource) FetchConfig(ctx context.Context, logger lager.Logger, repo *build.Repository) (atc.TaskConfig, error) {
+	config, err := configSource.ConfigSource.FetchConfig(ctx, logger, repo)
+	if err != nil {
+		return config, err
+	}
+
+	config.ImageResource.ApplySourceDefaults(configSource.ResourceTypes)
+
+	return config, nil
+}
+
+func (configSource BaseResourceTypeDefaultsApplySource) Warnings() []string {
+	return []string{}
+}
+
+type OverrideContainerLimitsSource struct {
+	ConfigSource TaskConfigSource
+	Limits       *atc.ContainerLimits
+}
+
+// FetchConfig overrides container limits, allowing the user to set container limits required by a task loaded
+// from a file by providing them in static configuration.
+func (configSource *OverrideContainerLimitsSource) FetchConfig(ctx context.Context, logger lager.Logger, source *build.Repository) (atc.TaskConfig, error) {
+	taskConfig, err := configSource.ConfigSource.FetchConfig(ctx, logger, source)
+	if err != nil {
+		return atc.TaskConfig{}, err
+	}
+
+	if configSource.Limits == nil {
+		return taskConfig, nil
+	}
+
+	if taskConfig.Limits == nil {
+		taskConfig.Limits = &atc.ContainerLimits{}
+	}
+
+	if configSource.Limits.CPU != nil {
+		taskConfig.Limits.CPU = configSource.Limits.CPU
+	}
+
+	if configSource.Limits.Memory != nil {
+		taskConfig.Limits.Memory = configSource.Limits.Memory
+	}
+
+	return taskConfig, nil
+}
+
+func (configSource *OverrideContainerLimitsSource) Warnings() []string {
+	return make([]string, 0)
+}
+
+var _ TaskConfigSource = &OverrideContainerLimitsSource{}
+
+// OverrideParamsConfigSource is used to override params in a config source
+type OverrideParamsConfigSource struct {
+	ConfigSource TaskConfigSource
+	Params       atc.TaskEnv
+	WarningList  []string
+}
+
+// FetchConfig overrides parameters, allowing the user to set params required by a task loaded
+// from a file by providing them in static configuration.
+func (configSource *OverrideParamsConfigSource) FetchConfig(ctx context.Context, logger lager.Logger, source *build.Repository) (atc.TaskConfig, error) {
+	taskConfig, err := configSource.ConfigSource.FetchConfig(ctx, logger, source)
+	if err != nil {
+		return atc.TaskConfig{}, err
+	}
+
+	if taskConfig.Params == nil {
+		taskConfig.Params = atc.TaskEnv{}
+	}
+
+	for key, val := range configSource.Params {
+		if _, exists := taskConfig.Params[key]; !exists {
+			configSource.WarningList = append(configSource.WarningList, fmt.Sprintf("%s was defined in pipeline but missing from task file", key))
+		}
+
+		taskConfig.Params[key] = val
+	}
+
+	return taskConfig, nil
+}
+
+func (configSource OverrideParamsConfigSource) Warnings() []string {
+	return configSource.WarningList
+}
+
+// InterpolateTemplateConfigSource represents a config source interpolated by template vars
+type InterpolateTemplateConfigSource struct {
+	ConfigSource  TaskConfigSource
+	Vars          []vars.Variables
+	ExpectAllKeys bool
+}
+
+// FetchConfig returns the interpolated configuration
+func (configSource InterpolateTemplateConfigSource) FetchConfig(ctx context.Context, logger lager.Logger, source *build.Repository) (atc.TaskConfig, error) {
+	taskConfig, err := configSource.ConfigSource.FetchConfig(ctx, logger, source)
+	if err != nil {
+		return atc.TaskConfig{}, err
+	}
+
+	byteConfig, err := yaml.Marshal(taskConfig)
+	if err != nil {
+		return atc.TaskConfig{}, fmt.Errorf("failed to marshal task config: %s", err)
+	}
+
+	// process task config using the provided variables
+	byteConfig, err = vars.NewTemplateResolver(byteConfig, configSource.Vars).Resolve(configSource.ExpectAllKeys)
+	if err != nil {
+		return atc.TaskConfig{}, fmt.Errorf("failed to interpolate task config: %s", err)
+	}
+
+	taskConfig, err = atc.NewTaskConfig(byteConfig)
+	if err != nil {
+		return atc.TaskConfig{}, fmt.Errorf("failed to create task config from bytes: %s", err)
+	}
+
+	return taskConfig, nil
+}
+
+func (configSource InterpolateTemplateConfigSource) Warnings() []string {
+	return []string{}
+}
+
+// ValidatingConfigSource delegates to another ConfigSource, and validates its
+// task config.
+type ValidatingConfigSource struct {
+	ConfigSource TaskConfigSource
+}
+
+// FetchConfig fetches the config using the underlying ConfigSource, and checks
+// that it's valid.
+func (configSource ValidatingConfigSource) FetchConfig(ctx context.Context, logger lager.Logger, source *build.Repository) (atc.TaskConfig, error) {
+	config, err := configSource.ConfigSource.FetchConfig(ctx, logger, source)
+	if err != nil {
+		return atc.TaskConfig{}, err
+	}
+
+	if err := config.Validate(); err != nil {
+		return atc.TaskConfig{}, err
+	}
+
+	return config, nil
+}
+
+func (configSource ValidatingConfigSource) Warnings() []string {
+	return configSource.ConfigSource.Warnings()
+}
+
+// UnknownArtifactSourceError is returned when the artifact.ArtifactName specified by the
+// path does not exist in the artifact.Repository.
+type UnknownArtifactSourceError struct {
+	SourceName build.ArtifactName
+	ConfigPath string
+}
+
+// Error returns a human-friendly error message.
+func (err UnknownArtifactSourceError) Error() string {
+	return fmt.Sprintf("unknown artifact source: '%s' in file path '%s'", err.SourceName, err.ConfigPath)
+}
+
+// UnspecifiedArtifactSourceError is returned when the specified path is of a
+// file in the toplevel directory, and so it does not indicate a SourceName.
+type UnspecifiedArtifactSourceError struct {
+	Path string
+}
+
+// Error returns a human-friendly error message.
+func (err UnspecifiedArtifactSourceError) Error() string {
+	return fmt.Sprintf("config path '%s' does not specify where the file lives", err.Path)
+}

atc/exec/task_config_source_test.go

@@ -0,0 +1,633 @@
+package exec_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"code.cloudfoundry.org/lager/v3/lagertest"
+	"github.com/concourse/concourse/atc"
+	. "github.com/concourse/concourse/atc/exec"
+	"github.com/concourse/concourse/atc/exec/build"
+	"github.com/concourse/concourse/atc/exec/execfakes"
+	"github.com/concourse/concourse/atc/runtime/runtimetest"
+	"github.com/concourse/concourse/vars"
+	"github.com/concourse/concourse/worker/baggageclaim"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/onsi/gomega/gbytes"
+	"sigs.k8s.io/yaml"
+)
+
+var _ = Describe("TaskConfigSource", func() {
+	var (
+		taskConfig atc.TaskConfig
+		taskVars   atc.Params
+		repo       *build.Repository
+		logger     *lagertest.TestLogger
+	)
+
+	BeforeEach(func() {
+		logger = lagertest.NewTestLogger("task-config-source-test")
+		repo = build.NewRepository()
+		taskConfig = atc.TaskConfig{
+			Platform:  "some-platform",
+			RootfsURI: "some-image",
+			ImageResource: &atc.ImageResource{
+				Type: "docker",
+				Source: atc.Source{
+					"a":               "b",
+					"evaluated-value": "((task-variable-name))",
+				},
+				Params: atc.Params{
+					"some":            "params",
+					"evaluated-value": "((task-variable-name))",
+				},
+				Version: atc.Version{"some": "version"},
+			},
+			Params: atc.TaskEnv{
+				"key1": "key1-((task-variable-name))",
+				"key2": "key2-((task-variable-name))",
+			},
+			Run: atc.TaskRunConfig{
+				Path: "ls",
+				Args: []string{"-al", "((task-variable-name))"},
+				Dir:  "some/dir",
+				User: "some-user",
+			},
+			Inputs: []atc.TaskInputConfig{
+				{Name: "some-input", Path: "some-path"},
+			},
+		}
+		taskVars = atc.Params{
+			"task-variable-name": "task-variable-value",
+		}
+	})
+
+	Describe("StaticConfigSource", func() {
+
+		It("fetches task config successfully", func() {
+			configSource := StaticConfigSource{Config: &taskConfig}
+			fetchedConfig, fetchErr := configSource.FetchConfig(context.TODO(), logger, repo)
+			Expect(fetchErr).ToNot(HaveOccurred())
+			Expect(fetchedConfig).To(Equal(taskConfig))
+		})
+
+		It("fetches config of nil task successfully", func() {
+			configSource := StaticConfigSource{Config: nil}
+			fetchedConfig, fetchErr := configSource.FetchConfig(context.TODO(), logger, repo)
+			Expect(fetchErr).ToNot(HaveOccurred())
+			Expect(fetchedConfig).To(Equal(atc.TaskConfig{}))
+		})
+	})
+
+	Describe("FileConfigSource", func() {
+		var (
+			configSource FileConfigSource
+			fakeStreamer *execfakes.FakeStreamer
+			fetchErr     error
+			artifactName string
+		)
+
+		BeforeEach(func() {
+
+			artifactName = "some-artifact-name"
+			fakeStreamer = new(execfakes.FakeStreamer)
+			configSource = FileConfigSource{
+				ConfigPath: artifactName + "/build.yml",
+				Streamer:   fakeStreamer,
+			}
+		})
+
+		JustBeforeEach(func() {
+			_, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+		})
+
+		Context("when the path does not indicate an artifact source", func() {
+			BeforeEach(func() {
+				configSource.ConfigPath = "foo-bar.yml"
+			})
+
+			It("returns an error", func() {
+				Expect(fetchErr).To(Equal(UnspecifiedArtifactSourceError{"foo-bar.yml"}))
+			})
+		})
+
+		Context("when the file's artifact can be found in the repository", func() {
+			var volume *runtimetest.Volume
+
+			BeforeEach(func() {
+				volume = runtimetest.NewVolume("some-volume")
+				repo.RegisterArtifact(build.ArtifactName(artifactName), volume, false)
+			})
+
+			Context("when the artifact provides a proper file", func() {
+				var streamedOut *gbytes.Buffer
+
+				BeforeEach(func() {
+					marshalled, err := yaml.Marshal(taskConfig)
+					Expect(err).NotTo(HaveOccurred())
+
+					streamedOut = gbytes.BufferWithBytes(marshalled)
+					fakeStreamer.StreamFileReturns(streamedOut, nil)
+				})
+
+				It("fetches the file via the correct artifact & path", func() {
+					_, artifact, dest := fakeStreamer.StreamFileArgsForCall(0)
+					Expect(artifact).To(Equal(volume))
+					Expect(dest).To(Equal("build.yml"))
+				})
+
+				It("succeeds", func() {
+					Expect(fetchErr).NotTo(HaveOccurred())
+				})
+
+				It("closes the stream", func() {
+					Expect(streamedOut.Closed()).To(BeTrue())
+				})
+			})
+
+			Context("when the artifact source provides an invalid configuration", func() {
+				var streamedOut *gbytes.Buffer
+
+				BeforeEach(func() {
+					invalidConfig := taskConfig
+					invalidConfig.Platform = ""
+					invalidConfig.Run = atc.TaskRunConfig{}
+
+					marshalled, err := yaml.Marshal(invalidConfig)
+					Expect(err).NotTo(HaveOccurred())
+
+					streamedOut = gbytes.BufferWithBytes(marshalled)
+					fakeStreamer.StreamFileReturns(streamedOut, nil)
+				})
+
+				It("returns an error", func() {
+					Expect(fetchErr).To(HaveOccurred())
+				})
+			})
+
+			Context("when the artifact source provides a malformed file", func() {
+				var streamedOut *gbytes.Buffer
+
+				BeforeEach(func() {
+					streamedOut = gbytes.BufferWithBytes([]byte("bogus"))
+					fakeStreamer.StreamFileReturns(streamedOut, nil)
+				})
+
+				It("fails", func() {
+					Expect(fetchErr).To(HaveOccurred())
+				})
+
+				It("closes the stream", func() {
+					Expect(streamedOut.Closed()).To(BeTrue())
+				})
+			})
+
+			Context("when the artifact source provides a valid file with invalid keys", func() {
+				var streamedOut *gbytes.Buffer
+
+				BeforeEach(func() {
+					streamedOut = gbytes.BufferWithBytes([]byte(`
+platform: beos
+
+intputs: []
+
+run: {path: a/file}
+`))
+					fakeStreamer.StreamFileReturns(streamedOut, nil)
+				})
+
+				It("fails", func() {
+					Expect(fetchErr).To(HaveOccurred())
+				})
+
+				It("closes the stream", func() {
+					Expect(streamedOut.Closed()).To(BeTrue())
+				})
+			})
+
+			Context("when streaming the file out fails", func() {
+				disaster := errors.New("nope")
+
+				BeforeEach(func() {
+					fakeStreamer.StreamFileReturns(nil, disaster)
+				})
+
+				It("returns the error", func() {
+					Expect(fetchErr).To(HaveOccurred())
+				})
+			})
+
+			Context("when the file task is not found", func() {
+				BeforeEach(func() {
+					fakeStreamer.StreamFileReturns(nil, baggageclaim.ErrFileNotFound)
+				})
+
+				It("returns the error", func() {
+					Expect(fetchErr).To(HaveOccurred())
+					Expect(fetchErr.Error()).To(Equal(fmt.Sprintf("task config '%s/build.yml' not found", artifactName)))
+				})
+			})
+		})
+
+		Context("when the file's artifact source cannot be found in the repository", func() {
+			It("returns an UnknownArtifactSourceError", func() {
+				Expect(fetchErr).To(Equal(UnknownArtifactSourceError{SourceName: build.ArtifactName(artifactName), ConfigPath: artifactName + "/build.yml"}))
+			})
+		})
+	})
+
+	Describe("OverrideParamsConfigSource", func() {
+		var (
+			config       atc.TaskConfig
+			configSource TaskConfigSource
+
+			overrideParams atc.TaskEnv
+
+			fetchedConfig atc.TaskConfig
+			fetchErr      error
+		)
+
+		BeforeEach(func() {
+			config = atc.TaskConfig{
+				Platform:  "some-platform",
+				RootfsURI: "some-image",
+				Params:    atc.TaskEnv{"PARAM": "A", "ORIG_PARAM": "D"},
+				Run: atc.TaskRunConfig{
+					Path: "echo",
+					Args: []string{"bananapants"},
+				},
+			}
+
+			overrideParams = atc.TaskEnv{"PARAM": "B", "EXTRA_PARAM": "C"}
+		})
+
+		Context("when there are no params to override", func() {
+			BeforeEach(func() {
+				configSource = &OverrideParamsConfigSource{
+					ConfigSource: StaticConfigSource{Config: &config},
+				}
+			})
+
+			JustBeforeEach(func() {
+				fetchedConfig, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+			})
+
+			It("succeeds", func() {
+				Expect(fetchErr).NotTo(HaveOccurred())
+			})
+
+			It("returns the same config", func() {
+				Expect(fetchedConfig).To(Equal(config))
+			})
+
+			It("returns no warnings", func() {
+				Expect(configSource.Warnings()).To(HaveLen(0))
+			})
+		})
+
+		Context("when override params are specified", func() {
+			BeforeEach(func() {
+				configSource = &OverrideParamsConfigSource{
+					ConfigSource: StaticConfigSource{Config: &config},
+					Params:       overrideParams,
+				}
+			})
+
+			JustBeforeEach(func() {
+				fetchedConfig, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+			})
+
+			It("succeeds", func() {
+				Expect(fetchErr).NotTo(HaveOccurred())
+			})
+
+			It("returns the config with overridden parameters", func() {
+				Expect(fetchedConfig.Params).To(Equal(atc.TaskEnv{
+					"ORIG_PARAM":  "D",
+					"PARAM":       "B",
+					"EXTRA_PARAM": "C",
+				}))
+			})
+
+			It("returns a deprecation warning", func() {
+				Expect(configSource.Warnings()).To(HaveLen(1))
+				Expect(configSource.Warnings()[0]).To(ContainSubstring("EXTRA_PARAM was defined in pipeline but missing from task file"))
+			})
+		})
+	})
+
+	Describe("OverrideContainerLimitsSource", func() {
+		var (
+			config         atc.TaskConfig
+			noLimitsConfig atc.TaskConfig
+			configSource   TaskConfigSource
+
+			overrideLimits atc.ContainerLimits
+
+			fetchedConfig atc.TaskConfig
+			fetchErr      error
+		)
+
+		BeforeEach(func() {
+			config = atc.TaskConfig{
+				Platform:  "some-platform",
+				RootfsURI: "some-image",
+				Limits:    &atc.ContainerLimits{CPU: newCPULimit(1024), Memory: newMemoryLimit(209715200)},
+				Run: atc.TaskRunConfig{
+					Path: "echo",
+					Args: []string{"bananapants"},
+				},
+			}
+
+			noLimitsConfig = atc.TaskConfig{
+				Platform:  "some-platform",
+				RootfsURI: "some-image",
+				Run: atc.TaskRunConfig{
+					Path: "echo",
+					Args: []string{"bananapants"},
+				},
+			}
+
+			overrideLimits = atc.ContainerLimits{CPU: newCPULimit(2048), Memory: newMemoryLimit(209715200)}
+		})
+
+		Context("when there are no container limits to override", func() {
+			BeforeEach(func() {
+				configSource = &OverrideContainerLimitsSource{
+					ConfigSource: StaticConfigSource{Config: &config},
+				}
+			})
+
+			JustBeforeEach(func() {
+				fetchedConfig, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+			})
+
+			It("succeeds", func() {
+				Expect(fetchErr).NotTo(HaveOccurred())
+			})
+
+			It("returns the same config", func() {
+				Expect(fetchedConfig).To(Equal(config))
+			})
+		})
+
+		Context("when override container limits are specified but no configuration container limits", func() {
+			BeforeEach(func() {
+				configSource = &OverrideContainerLimitsSource{
+					ConfigSource: StaticConfigSource{Config: &noLimitsConfig},
+					Limits:       &overrideLimits,
+				}
+			})
+
+			JustBeforeEach(func() {
+				fetchedConfig, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+			})
+
+			It("succeeds", func() {
+				Expect(fetchErr).NotTo(HaveOccurred())
+			})
+
+			It("returns the config with overridden limits", func() {
+				Expect(fetchedConfig).NotTo(BeNil())
+				Expect(*fetchedConfig.Limits).To(Equal(atc.ContainerLimits{
+					CPU:    newCPULimit(2048),
+					Memory: newMemoryLimit(209715200),
+				}))
+			})
+		})
+
+		Context("when override container limits are specified", func() {
+			BeforeEach(func() {
+				configSource = &OverrideContainerLimitsSource{
+					ConfigSource: StaticConfigSource{Config: &config},
+					Limits:       &overrideLimits,
+				}
+			})
+
+			JustBeforeEach(func() {
+				fetchedConfig, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+			})
+
+			It("succeeds", func() {
+				Expect(fetchErr).NotTo(HaveOccurred())
+			})
+
+			It("returns the config with overridden limits", func() {
+				Expect(fetchedConfig).NotTo(BeNil())
+				Expect(*fetchedConfig.Limits).To(Equal(atc.ContainerLimits{
+					CPU:    newCPULimit(2048),
+					Memory: newMemoryLimit(209715200),
+				}))
+			})
+		})
+	})
+
+	Describe("ValidatingConfigSource", func() {
+		var (
+			fakeConfigSource *execfakes.FakeTaskConfigSource
+
+			configSource TaskConfigSource
+
+			fetchedConfig atc.TaskConfig
+			fetchErr      error
+		)
+
+		BeforeEach(func() {
+			fakeConfigSource = new(execfakes.FakeTaskConfigSource)
+
+			configSource = ValidatingConfigSource{fakeConfigSource}
+		})
+
+		JustBeforeEach(func() {
+			fetchedConfig, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+		})
+
+		Context("when the config is valid", func() {
+			config := atc.TaskConfig{
+				Platform:  "some-platform",
+				RootfsURI: "some-image",
+				Params:    atc.TaskEnv{"PARAM": "A"},
+				Run: atc.TaskRunConfig{
+					Path: "echo",
+					Args: []string{"bananapants"},
+				},
+			}
+
+			BeforeEach(func() {
+				fakeConfigSource.FetchConfigReturns(config, nil)
+			})
+
+			It("returns the config and no error", func() {
+				Expect(fetchErr).ToNot(HaveOccurred())
+				Expect(fetchedConfig).To(Equal(config))
+			})
+		})
+
+		Context("when the config is invalid", func() {
+			BeforeEach(func() {
+				fakeConfigSource.FetchConfigReturns(atc.TaskConfig{
+					RootfsURI: "some-image",
+					Params:    atc.TaskEnv{"PARAM": "A"},
+					Run: atc.TaskRunConfig{
+						Args: []string{"bananapants"},
+					},
+				}, nil)
+			})
+
+			It("returns the validation error", func() {
+				Expect(fetchErr).To(HaveOccurred())
+			})
+		})
+
+		Context("when fetching the config fails", func() {
+			disaster := errors.New("nope")
+
+			BeforeEach(func() {
+				fakeConfigSource.FetchConfigReturns(atc.TaskConfig{}, disaster)
+			})
+
+			It("returns the error", func() {
+				Expect(fetchErr).To(Equal(disaster))
+			})
+		})
+	})
+
+	Describe("InterpolateTemplateConfigSource", func() {
+		var (
+			configSource  TaskConfigSource
+			fetchedConfig atc.TaskConfig
+			fetchErr      error
+			expectAllKeys bool
+		)
+
+		JustBeforeEach(func() {
+			configSource = StaticConfigSource{Config: &taskConfig}
+			configSource = InterpolateTemplateConfigSource{
+				ConfigSource:  configSource,
+				Vars:          []vars.Variables{vars.StaticVariables(taskVars)},
+				ExpectAllKeys: expectAllKeys,
+			}
+			fetchedConfig, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+		})
+
+		Context("when expect all keys", func() {
+			BeforeEach(func() {
+				expectAllKeys = true
+			})
+
+			It("fetches task config successfully", func() {
+				Expect(fetchErr).ToNot(HaveOccurred())
+			})
+
+			It("resolves task config parameters successfully", func() {
+				Expect(fetchedConfig.Run.Args).To(Equal([]string{"-al", "task-variable-value"}))
+				Expect(fetchedConfig.Params).To(Equal(atc.TaskEnv{
+					"key1": "key1-task-variable-value",
+					"key2": "key2-task-variable-value",
+				}))
+				Expect(fetchedConfig.ImageResource.Source).To(Equal(atc.Source{
+					"a":               "b",
+					"evaluated-value": "task-variable-value",
+				}))
+			})
+		})
+
+		Context("when not expect all keys", func() {
+			BeforeEach(func() {
+				expectAllKeys = false
+				taskVars = atc.Params{}
+			})
+
+			It("fetches task config successfully", func() {
+				Expect(fetchErr).ToNot(HaveOccurred())
+			})
+
+			It("resolves task config parameters successfully", func() {
+				Expect(fetchedConfig.Run.Args).To(Equal([]string{"-al", "((task-variable-name))"}))
+				Expect(fetchedConfig.Params).To(Equal(atc.TaskEnv{
+					"key1": "key1-((task-variable-name))",
+					"key2": "key2-((task-variable-name))",
+				}))
+				Expect(fetchedConfig.ImageResource.Source).To(Equal(atc.Source{
+					"a":               "b",
+					"evaluated-value": "((task-variable-name))",
+				}))
+			})
+		})
+	})
+
+	Context("BaseResourceTypeDefaultsApplySource", func() {
+		var (
+			configSource  TaskConfigSource
+			resourceTypes atc.ResourceTypes
+			fetchedConfig atc.TaskConfig
+			fetchErr      error
+		)
+
+		JustBeforeEach(func() {
+			configSource = StaticConfigSource{Config: &taskConfig}
+			configSource = BaseResourceTypeDefaultsApplySource{
+				ConfigSource:  configSource,
+				ResourceTypes: resourceTypes,
+			}
+			fetchedConfig, fetchErr = configSource.FetchConfig(context.TODO(), logger, repo)
+		})
+
+		Context("resourceTypes is empty, and no base resource type defaults configured", func() {
+			It("fetchedConfig should be identical to the original", func() {
+				Expect(fetchErr).ToNot(HaveOccurred())
+				Expect(fetchedConfig).To(Equal(taskConfig))
+			})
+		})
+
+		Context("resourceTypes is empty, and base resource type defaults configured", func() {
+			BeforeEach(func() {
+				atc.LoadBaseResourceTypeDefaults(map[string]atc.Source{"docker": atc.Source{"some-key": "some-value"}})
+			})
+			AfterEach(func() {
+				atc.LoadBaseResourceTypeDefaults(map[string]atc.Source{})
+			})
+
+			It("defaults should be added to image source", func() {
+				Expect(fetchErr).ToNot(HaveOccurred())
+				Expect(fetchedConfig.ImageResource.Source).To(Equal(atc.Source{
+					"a":               "b",
+					"evaluated-value": "((task-variable-name))",
+					"some-key":        "some-value",
+				}))
+			})
+		})
+
+		Context("resourceTypes contains image source type", func() {
+			BeforeEach(func() {
+				resourceTypes = atc.ResourceTypes{
+					{
+						Name:     "docker",
+						Defaults: atc.Source{"some-key": "some-value"},
+					},
+				}
+			})
+
+			It("defaults should be added to image source", func() {
+				Expect(fetchErr).ToNot(HaveOccurred())
+				Expect(fetchedConfig.ImageResource.Source).To(Equal(atc.Source{
+					"a":               "b",
+					"evaluated-value": "((task-variable-name))",
+					"some-key":        "some-value",
+				}))
+			})
+		})
+	})
+})
+
+func newCPULimit(cpuLimit uint64) *atc.CPULimit {
+	limit := atc.CPULimit(cpuLimit)
+	return &limit
+}
+
+func newMemoryLimit(memoryLimit uint64) *atc.MemoryLimit {
+	limit := atc.MemoryLimit(memoryLimit)
+	return &limit
+}

atc/exec/task_step.go

@@ -0,0 +1,634 @@
+package exec
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"code.cloudfoundry.org/lager/v3"
+	"code.cloudfoundry.org/lager/v3/lagerctx"
+	"github.com/concourse/concourse/atc"
+	"github.com/concourse/concourse/atc/db"
+	"github.com/concourse/concourse/atc/exec/build"
+	"github.com/concourse/concourse/atc/runtime"
+	"github.com/concourse/concourse/atc/worker"
+	"github.com/concourse/concourse/tracing"
+	"github.com/concourse/concourse/vars"
+	"github.com/concourse/concourse/worker/baggageclaim"
+	"go.opentelemetry.io/otel/trace"
+)
+
+const taskProcessID = "task"
+
+// MissingInputsError is returned when any of the task's required inputs are
+// missing.
+type MissingInputsError struct {
+	Inputs []string
+}
+
+// Error prints a human-friendly message listing the inputs that were missing.
+func (err MissingInputsError) Error() string {
+	return fmt.Sprintf("missing inputs: %s", strings.Join(err.Inputs, ", "))
+}
+
+type MissingTaskImageSourceError struct {
+	SourceName string
+}
+
+func (err MissingTaskImageSourceError) Error() string {
+	return fmt.Sprintf(`missing image artifact source: %s
+
+make sure there's a corresponding 'get' step, or a task that produces it as an output`, err.SourceName)
+}
+
+type TaskImageSourceParametersError struct {
+	Err error
+}
+
+func (err TaskImageSourceParametersError) Error() string {
+	return fmt.Sprintf("failed to evaluate image resource parameters: %s", err.Err)
+}
+
+//counterfeiter:generate . TaskDelegateFactory
+type TaskDelegateFactory interface {
+	TaskDelegate(state RunState) TaskDelegate
+}
+
+//counterfeiter:generate . TaskDelegate
+type TaskDelegate interface {
+	StartSpan(context.Context, string, tracing.Attrs) (context.Context, trace.Span)
+
+	FetchImage(context.Context, atc.ImageResource, atc.ResourceTypes, bool, atc.Tags, bool) (runtime.ImageSpec, error)
+
+	Stdout() io.Writer
+	Stderr() io.Writer
+
+	SetTaskConfig(config atc.TaskConfig)
+
+	Initializing(lager.Logger)
+	Starting(lager.Logger)
+	Finished(lager.Logger, ExitStatus)
+	Errored(lager.Logger, string)
+
+	BeforeSelectWorker(lager.Logger) error
+	WaitingForWorker(lager.Logger)
+	SelectedWorker(lager.Logger, string)
+	StreamingVolume(lager.Logger, string, string, string)
+	WaitingForStreamedVolume(lager.Logger, string, string)
+	BuildStartTime() time.Time
+}
+
+// TaskStep executes a TaskConfig, whose inputs will be fetched from the
+// artifact.Repository and outputs will be added to the artifact.Repository.
+type TaskStep struct {
+	planID              atc.PlanID
+	plan                atc.TaskPlan
+	defaultLimits       atc.ContainerLimits
+	metadata            StepMetadata
+	containerMetadata   db.ContainerMetadata
+	strategy            worker.PlacementStrategy
+	workerPool          Pool
+	streamer            Streamer
+	delegateFactory     TaskDelegateFactory
+	defaultTaskTimeout  time.Duration
+	defaultTaskCacheTTL time.Duration
+}
+
+func NewTaskStep(
+	planID atc.PlanID,
+	plan atc.TaskPlan,
+	defaultLimits atc.ContainerLimits,
+	metadata StepMetadata,
+	containerMetadata db.ContainerMetadata,
+	strategy worker.PlacementStrategy,
+	workerPool Pool,
+	streamer Streamer,
+	delegateFactory TaskDelegateFactory,
+	defaultTaskTimeout time.Duration,
+	defaultTaskCacheTTL time.Duration,
+) Step {
+	return &TaskStep{
+		planID:              planID,
+		plan:                plan,
+		defaultLimits:       defaultLimits,
+		metadata:            metadata,
+		containerMetadata:   containerMetadata,
+		strategy:            strategy,
+		workerPool:          workerPool,
+		streamer:            streamer,
+		delegateFactory:     delegateFactory,
+		defaultTaskTimeout:  defaultTaskTimeout,
+		defaultTaskCacheTTL: defaultTaskCacheTTL,
+	}
+}
+
+// Run will first select the worker based on the TaskConfig's platform and the
+// TaskStep's tags, and prioritize it by availability of volumes for the TaskConfig's
+// inputs. Inputs that did not have volumes available on the worker will be streamed
+// in to the container.
+//
+// If any inputs are not available in the artifact.Repository, MissingInputsError
+// is returned.
+//
+// Once all the inputs are satisfied, the task's script will be executed. If
+// the task is canceled via the context, the script will be interrupted.
+//
+// If the script exits successfully, the outputs specified in the TaskConfig
+// are registered with the artifact.Repository. If no outputs are specified, the
+// task's entire working directory is registered as an StreamableArtifactSource under the
+// name of the task.
+func (step *TaskStep) Run(ctx context.Context, state RunState) (bool, error) {
+	delegate := step.delegateFactory.TaskDelegate(state)
+	ctx, span := delegate.StartSpan(ctx, "task", tracing.Attrs{
+		"name": step.plan.Name,
+	})
+
+	ok, err := step.run(ctx, state, delegate)
+	tracing.End(span, err)
+
+	return ok, err
+}
+
+func (step *TaskStep) run(ctx context.Context, state RunState, delegate TaskDelegate) (bool, error) {
+	logger := lagerctx.FromContext(ctx)
+	logger = logger.Session("task-step", lager.Data{
+		"step-name": step.plan.Name,
+		"job-id":    step.metadata.JobID,
+	})
+
+	var taskConfigSource TaskConfigSource
+	var taskVars []vars.Variables
+
+	if step.plan.ConfigPath != "" {
+		// external task - construct a source which reads it from file, and apply base resource type defaults.
+		taskConfigSource = FileConfigSource{ConfigPath: step.plan.ConfigPath, Streamer: step.streamer}
+
+		// for interpolation - use 'vars' from the pipeline, and then fill remaining with cred variables.
+		// this 2-phase strategy allows to interpolate 'vars' by cred variables.
+		if len(step.plan.Vars) > 0 {
+			taskConfigSource = InterpolateTemplateConfigSource{
+				ConfigSource:  taskConfigSource,
+				Vars:          []vars.Variables{vars.StaticVariables(step.plan.Vars)},
+				ExpectAllKeys: false,
+			}
+		}
+		taskVars = []vars.Variables{state}
+	} else {
+		// embedded task - first we take it
+		taskConfigSource = StaticConfigSource{Config: step.plan.Config}
+
+		// for interpolation - use just cred variables
+		taskVars = []vars.Variables{state}
+	}
+
+	// apply resource type defaults
+	taskConfigSource = BaseResourceTypeDefaultsApplySource{
+		ConfigSource:  taskConfigSource,
+		ResourceTypes: step.plan.ResourceTypes,
+	}
+
+	// override limits
+	taskConfigSource = &OverrideContainerLimitsSource{ConfigSource: taskConfigSource, Limits: step.plan.Limits}
+
+	// override params
+	taskConfigSource = &OverrideParamsConfigSource{ConfigSource: taskConfigSource, Params: step.plan.Params}
+
+	// interpolate template vars
+	taskConfigSource = InterpolateTemplateConfigSource{
+		ConfigSource:  taskConfigSource,
+		Vars:          taskVars,
+		ExpectAllKeys: true,
+	}
+
+	// validate
+	taskConfigSource = ValidatingConfigSource{ConfigSource: taskConfigSource}
+
+	repository := state.ArtifactRepository()
+
+	config, err := taskConfigSource.FetchConfig(ctx, logger, repository)
+
+	for _, warning := range taskConfigSource.Warnings() {
+		fmt.Fprintln(delegate.Stderr(), "[WARNING]", warning)
+	}
+
+	if err != nil {
+		return false, err
+	}
+
+	if config.Limits == nil {
+		config.Limits = &atc.ContainerLimits{}
+	}
+	if config.Limits.CPU == nil {
+		config.Limits.CPU = step.defaultLimits.CPU
+	}
+	if config.Limits.Memory == nil {
+		config.Limits.Memory = step.defaultLimits.Memory
+	}
+
+	delegate.Initializing(logger)
+
+	imageSpec, err := step.imageSpec(ctx, logger, state, delegate, config)
+	if err != nil {
+		return false, err
+	}
+
+	if config.Run.Path == "" {
+		path, args, err := step.getOciEntrypoint(ctx, imageSpec)
+		if err != nil {
+			return false, err
+		}
+
+		config.Run.Path = path
+		// Append any args defined in the Task config to those from the
+		// container image
+		config.Run.Args = append(args, config.Run.Args...)
+	}
+
+	if config.Run.Path == "" {
+		return false, errors.New("No executable defined for task. Specify an executable to run in either the task config (run.path) or as an ENTRYPOINT/CMD in the container image.")
+	}
+
+	delegate.SetTaskConfig(config)
+
+	containerSpec, err := step.containerSpec(logger, state, imageSpec, config, step.containerMetadata)
+	if err != nil {
+		return false, err
+	}
+	tracing.Inject(ctx, &containerSpec)
+
+	owner := db.NewBuildStepContainerOwner(step.metadata.BuildID, step.planID, step.metadata.TeamID)
+
+	err = delegate.BeforeSelectWorker(logger)
+	if err != nil {
+		return false, err
+	}
+
+	worker, err := step.workerPool.FindOrSelectWorker(
+		ctx,
+		owner,
+		containerSpec,
+		step.workerSpec(config),
+		step.strategy,
+		delegate,
+	)
+	if err != nil {
+		return false, err
+	}
+
+	defer func() {
+		step.workerPool.ReleaseWorker(
+			logger,
+			containerSpec,
+			worker,
+			step.strategy,
+		)
+	}()
+
+	ctx, cancel, err := MaybeTimeout(ctx, step.plan.Timeout, step.defaultTaskTimeout)
+	if err != nil {
+		return false, err
+	}
+	defer cancel()
+
+	ctx = lagerctx.NewContext(ctx, logger)
+
+	delegate.SelectedWorker(logger, worker.Name())
+
+	container, volumeMounts, err := worker.FindOrCreateContainer(ctx, owner, step.containerMetadata, containerSpec, delegate)
+	if err != nil {
+		if strings.Contains(err.Error(), "rootfs must not be empty") {
+			return false, errors.Join(
+				err,
+				errors.New(`No image specified for task? Make sure to specify "image" on your task step or "image_resource" in your task config. The image must be in rootfs format.`),
+			)
+		}
+		return false, err
+	}
+
+	delegate.Starting(logger)
+	process, err := attachOrRun(
+		ctx,
+		container,
+		runtime.ProcessSpec{
+			ID:   taskProcessID,
+			Path: config.Run.Path,
+			Args: config.Run.Args,
+			Dir:  resolvePath(step.containerMetadata.WorkingDirectory, config.Run.Dir),
+			User: config.Run.User,
+			// Guardian sets the default TTY window size to width: 80, height: 24,
+			// which creates ANSI control sequences that do not work with other window sizes
+			TTY: &runtime.TTYSpec{
+				WindowSize: runtime.WindowSize{
+					Columns: 500,
+					Rows:    500,
+				},
+			},
+		},
+		runtime.ProcessIO{
+			Stdout: delegate.Stdout(),
+			Stderr: delegate.Stderr(),
+		},
+	)
+	if err != nil {
+		return false, err
+	}
+
+	result, runErr := process.Wait(ctx)
+
+	step.registerOutputs(logger, repository, config, volumeMounts, step.containerMetadata)
+
+	// Do not initialize caches for one-off builds
+	if step.metadata.JobID != 0 {
+		if err := step.registerCaches(ctx, config, volumeMounts, step.containerMetadata); err != nil {
+			return false, err
+		}
+	}
+
+	if runErr != nil {
+		if errors.Is(runErr, context.DeadlineExceeded) {
+			delegate.Errored(logger, TimeoutLogMessage)
+			return false, nil
+		}
+
+		return false, runErr
+	}
+
+	delegate.Finished(logger, ExitStatus(result.ExitStatus))
+	return result.ExitStatus == 0, nil
+}
+
+func attachOrRun(ctx context.Context, container runtime.Container, spec runtime.ProcessSpec, io runtime.ProcessIO) (runtime.Process, error) {
+	process, err := container.Attach(ctx, spec.ID, io)
+	if err == nil {
+		return process, nil
+	}
+	return container.Run(ctx, spec, io)
+}
+
+func (step *TaskStep) imageSpec(ctx context.Context, logger lager.Logger, state RunState, delegate TaskDelegate, config atc.TaskConfig) (runtime.ImageSpec, error) {
+	imageSpec := runtime.ImageSpec{
+		Privileged: bool(step.plan.Privileged),
+	}
+
+	// Determine the source of the container image
+	// a reference to an artifact (get step, task output) ?
+	if step.plan.ImageArtifactName != "" {
+		artifact, _, found := state.ArtifactRepository().ArtifactFor(build.ArtifactName(step.plan.ImageArtifactName))
+		if !found {
+			return runtime.ImageSpec{}, MissingTaskImageSourceError{step.plan.ImageArtifactName}
+		}
+		imageSpec.ImageArtifact = artifact
+
+		//an image_resource
+	} else if config.ImageResource != nil {
+		imageSpec, err := delegate.FetchImage(
+			ctx,
+			*config.ImageResource,
+			step.plan.ResourceTypes,
+			step.plan.Privileged,
+			step.plan.Tags,
+			step.plan.CheckSkipInterval,
+		)
+		return imageSpec, err
+
+		// a rootfs_uri
+	} else if config.RootfsURI != "" {
+		imageSpec.ImageURL = config.RootfsURI
+	}
+
+	return imageSpec, nil
+}
+
+func (step *TaskStep) containerInputs(logger lager.Logger, repository *build.Repository, config atc.TaskConfig, metadata db.ContainerMetadata) ([]runtime.Input, error) {
+	var inputs []runtime.Input
+
+	var missingRequiredInputs []string
+
+	for _, input := range config.Inputs {
+		inputName := input.Name
+		if sourceName, ok := step.plan.InputMapping[inputName]; ok {
+			inputName = sourceName
+		}
+
+		artifact, fromCache, found := repository.ArtifactFor(build.ArtifactName(inputName))
+		if !found {
+			if !input.Optional {
+				missingRequiredInputs = append(missingRequiredInputs, inputName)
+			}
+			continue
+		}
+		inputs = append(inputs, runtime.Input{
+			Artifact:        artifact,
+			DestinationPath: artifactPath(metadata.WorkingDirectory, input.Name, input.Path),
+			FromCache:       fromCache,
+		})
+	}
+
+	if len(missingRequiredInputs) > 0 {
+		return nil, MissingInputsError{missingRequiredInputs}
+	}
+
+	return inputs, nil
+}
+
+func (step *TaskStep) containerSpec(logger lager.Logger, state RunState, imageSpec runtime.ImageSpec, config atc.TaskConfig, metadata db.ContainerMetadata) (runtime.ContainerSpec, error) {
+	env := step.metadata.TaskEnv()
+	env = append(env, config.Params.Env()...)
+
+	containerSpec := runtime.ContainerSpec{
+		TeamID:   step.metadata.TeamID,
+		TeamName: step.metadata.TeamName,
+		JobID:    step.metadata.JobID,
+		StepName: step.plan.Name,
+
+		ImageSpec: imageSpec,
+		Env:       env,
+		Type:      metadata.Type,
+
+		Dir:      metadata.WorkingDirectory,
+		Hermetic: step.plan.Hermetic,
+		User:     config.Run.User,
+	}
+
+	var err error
+	containerSpec.Inputs, err = step.containerInputs(logger, state.ArtifactRepository(), config, metadata)
+	if err != nil {
+		return runtime.ContainerSpec{}, err
+	}
+
+	containerSpec.Caches = make([]string, len(config.Caches))
+	for i, cache := range config.Caches {
+		containerSpec.Caches[i] = cache.Path
+	}
+
+	containerSpec.Outputs = make(runtime.OutputPaths, len(config.Outputs))
+	for _, output := range config.Outputs {
+		containerSpec.Outputs[output.Name] = ensureTrailingSlash(artifactPath(metadata.WorkingDirectory, output.Name, output.Path))
+	}
+
+	if config.Limits != nil {
+		containerSpec.Limits.CPU = (*uint64)(config.Limits.CPU)
+		containerSpec.Limits.Memory = (*uint64)(config.Limits.Memory)
+	}
+
+	return containerSpec, nil
+}
+
+func (step *TaskStep) workerSpec(config atc.TaskConfig) worker.Spec {
+	return worker.Spec{
+		Platform: config.Platform,
+		Tags:     step.plan.Tags,
+		TeamID:   step.metadata.TeamID,
+	}
+}
+
+func (step *TaskStep) registerOutputs(logger lager.Logger, repository *build.Repository, config atc.TaskConfig, volumeMounts []runtime.VolumeMount, metadata db.ContainerMetadata) {
+	logger.Debug("registering-outputs", lager.Data{"outputs": config.Outputs})
+
+	for _, output := range config.Outputs {
+		outputName := output.Name
+		if destinationName, ok := step.plan.OutputMapping[output.Name]; ok {
+			outputName = destinationName
+		}
+
+		outputPath := artifactPath(metadata.WorkingDirectory, output.Name, output.Path)
+
+		for _, mount := range volumeMounts {
+			if filepath.Clean(mount.MountPath) == filepath.Clean(outputPath) {
+				repository.RegisterArtifact(build.ArtifactName(outputName), mount.Volume, false)
+			}
+		}
+	}
+}
+
+func (step *TaskStep) registerCaches(ctx context.Context, config atc.TaskConfig, volumeMounts []runtime.VolumeMount, metadata db.ContainerMetadata) error {
+	logger := lagerctx.FromContext(ctx)
+	for _, cacheConfig := range config.Caches {
+		var ttl time.Duration
+		if cacheConfig.TTL != "" {
+			var err error
+			ttl, err = time.ParseDuration(cacheConfig.TTL)
+			if err != nil {
+				return fmt.Errorf("invalid cache ttl %q: %w", cacheConfig.TTL, err)
+			}
+		}
+
+		if ttl == 0 && step.defaultTaskCacheTTL > 0 {
+			ttl = step.defaultTaskCacheTTL
+		}
+
+		for _, volumeMount := range volumeMounts {
+			mountPath := resolvePath(metadata.WorkingDirectory, cacheConfig.Path)
+			if filepath.Clean(volumeMount.MountPath) == mountPath {
+				logger.Debug("initializing-cache", lager.Data{
+					"cache": cacheConfig.Path,
+				})
+				err := volumeMount.Volume.InitializeTaskCache(
+					ctx,
+					step.metadata.JobID,
+					step.plan.Name,
+					cacheConfig.Path,
+					step.plan.Privileged,
+					ttl,
+				)
+				if err != nil {
+					return err
+				}
+
+				break
+			}
+		}
+	}
+
+	return nil
+}
+
+type ociRunConfig struct {
+	Cmd        []string `json:"cmd"`
+	EntryPoint []string `json:"entrypoint"`
+}
+
+// Try and use OCI image Entrypoint and/or CMD. This relies on the
+// registry-image resource providing this information when it converts OCI
+// images into our rootfs format.
+func (step *TaskStep) getOciEntrypoint(ctx context.Context, imageSpec runtime.ImageSpec) (string, []string, error) {
+	path := ""
+	args := []string{}
+
+	if imageSpec.ImageArtifact == nil {
+		// there's no image artifact for Windows and Darwin tasks
+		return path, args, nil
+	}
+
+	metadataStream, err := step.streamer.StreamFile(ctx, imageSpec.ImageArtifact, "metadata.json")
+	if err != nil {
+		if err == baggageclaim.ErrFileNotFound {
+			return path, args, FileNotFoundError{
+				Name:     imageSpec.ImageArtifact.Handle(),
+				FilePath: "metadata.json",
+			}
+		}
+		return path, args, err
+	}
+
+	defer metadataStream.Close()
+
+	imageMetadata, err := io.ReadAll(metadataStream)
+	if err != nil {
+		return path, args, err
+	}
+
+	imgCmd := ociRunConfig{}
+	err = json.Unmarshal(imageMetadata, &imgCmd)
+	if err != nil {
+		return path, args, fmt.Errorf("error parsing metadata.json from rootfs: %s", err.Error())
+	}
+
+	if len(imgCmd.EntryPoint) >= 1 {
+		path = imgCmd.EntryPoint[0]
+	}
+
+	if len(imgCmd.EntryPoint) > 1 {
+		args = append(args, imgCmd.EntryPoint[1:]...)
+	}
+
+	if path == "" && len(imgCmd.Cmd) >= 1 {
+		path = imgCmd.Cmd[0]
+		if len(imgCmd.Cmd) > 1 {
+			args = append(args, imgCmd.Cmd[1:]...)
+		}
+	} else {
+		args = append(args, imgCmd.Cmd...)
+	}
+
+	return path, args, err
+}
+
+func artifactPath(workingDir string, name string, path string) string {
+	subdir := path
+	if path == "" {
+		subdir = name
+	}
+
+	return resolvePath(workingDir, subdir)
+}
+
+func resolvePath(workingDir string, path string) string {
+	if filepath.IsAbs(path) {
+		return path
+	}
+	path = strings.ReplaceAll(path, "..", "")
+	return filepath.Join(workingDir, path)
+}
+
+func ensureTrailingSlash(path string) string {
+	if strings.HasSuffix(path, "/") {
+		return path
+	}
+	return path + "/"
+}