commit af4510fc7f162db46fb726cdcb150713af7cc4e0 Author: upstream Date: Fri Sep 11 23:23:28 2026 +0200 Upstream concourse/concourse state of the touched files before bb8b7aa7e9 diff --git a/atc/exec/task_config_source.go b/atc/exec/task_config_source.go new file mode 100644 index 0000000..b63de8b --- /dev/null +++ b/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) +} diff --git a/atc/exec/task_config_source_test.go b/atc/exec/task_config_source_test.go new file mode 100644 index 0000000..4ba3c9a --- /dev/null +++ b/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 +} diff --git a/atc/exec/task_step.go b/atc/exec/task_step.go new file mode 100644 index 0000000..3014520 --- /dev/null +++ b/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 + "/" +} diff --git a/atc/exec/task_step_test.go b/atc/exec/task_step_test.go new file mode 100644 index 0000000..b599624 --- /dev/null +++ b/atc/exec/task_step_test.go @@ -0,0 +1,1188 @@ +package exec_test + +import ( + "context" + "errors" + "fmt" + "io" + "strings" + "time" + + "github.com/concourse/concourse/atc" + "github.com/concourse/concourse/atc/db" + "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" + "github.com/concourse/concourse/atc/runtime/runtimetest" + "github.com/concourse/concourse/atc/worker" + "github.com/concourse/concourse/tracing" + "github.com/concourse/concourse/vars" + "github.com/concourse/concourse/worker/baggageclaim" + "github.com/onsi/gomega/gbytes" + "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("TaskStep", func() { + var ( + ctx context.Context + cancel func() + + stdoutBuf *gbytes.Buffer + stderrBuf *gbytes.Buffer + + fakePool *execfakes.FakePool + fakeStreamer *execfakes.FakeStreamer + + fakeDelegate *execfakes.FakeTaskDelegate + + fakeDelegateFactory *execfakes.FakeTaskDelegateFactory + + taskPlan *atc.TaskPlan + + state exec.RunState + repo *build.Repository + + taskStep exec.Step + stepOk bool + stepErr error + + cpuLimit = atc.CPULimit(1024) + memoryLimit = atc.MemoryLimit(1024) + containerMetadata = db.ContainerMetadata{ + WorkingDirectory: "some-artifact-root", + Type: db.ContainerTypeTask, + StepName: "some-step", + } + + stepMetadata = exec.StepMetadata{ + TeamID: 123, + BuildID: 1234, + JobID: 12345, + ExternalURL: "http://foo.bar", + } + + planID = atc.PlanID("42") + + expectedOwner = db.NewBuildStepContainerOwner(stepMetadata.BuildID, planID, stepMetadata.TeamID) + + defaultTaskTimeout time.Duration = 0 + defaultTaskCacheTTL time.Duration = 0 + ) + + BeforeEach(func() { + ctx, cancel = context.WithCancel(context.Background()) + + stdoutBuf = gbytes.NewBuffer() + stderrBuf = gbytes.NewBuffer() + + fakeStreamer = new(execfakes.FakeStreamer) + + fakeDelegate = new(execfakes.FakeTaskDelegate) + fakeDelegate.StdoutReturns(stdoutBuf) + fakeDelegate.StderrReturns(stderrBuf) + + fakeDelegate.StartSpanReturns(ctx, tracing.NoopSpan) + + fakeDelegateFactory = new(execfakes.FakeTaskDelegateFactory) + fakeDelegateFactory.TaskDelegateReturns(fakeDelegate) + + state = exec.NewRunState(noopStepper, vars.StaticVariables{"source-param": "super-secret-source"}) + repo = state.ArtifactRepository() + + taskPlan = &atc.TaskPlan{ + Name: "some-task", + Privileged: false, + ResourceTypes: atc.ResourceTypes{ + { + Name: "custom-resource", + Type: "custom-type", + Source: atc.Source{"some-custom": "((source-param))"}, + Params: atc.Params{"some-custom": "param"}, + }, + }, + } + }) + + JustBeforeEach(func() { + plan := atc.Plan{ + ID: planID, + Task: taskPlan, + } + + // stuff stored on task step still + taskStep = exec.NewTaskStep( + plan.ID, + *plan.Task, + atc.ContainerLimits{}, + stepMetadata, + containerMetadata, + nil, + fakePool, + fakeStreamer, + fakeDelegateFactory, + defaultTaskTimeout, + defaultTaskCacheTTL, + ) + + stepOk, stepErr = taskStep.Run(ctx, state) + }) + + expectWorkerSpecResourceTypeUnset := func() { + Expect(fakePool.FindOrSelectWorkerCallCount()).To(Equal(1)) + _, _, _, workerSpec, _, _ := fakePool.FindOrSelectWorkerArgsForCall(0) + Expect(workerSpec.ResourceType).To(Equal("")) + } + + Context("when the plan has a config", func() { + var chosenWorker *runtimetest.Worker + var chosenContainer *runtimetest.WorkerContainer + + BeforeEach(func() { + taskPlan.Config = &atc.TaskConfig{ + Platform: "some-platform", + Limits: &atc.ContainerLimits{ + CPU: &cpuLimit, + Memory: &memoryLimit, + }, + Params: atc.TaskEnv{ + "SECURE": "secret-task-param", + }, + Run: atc.TaskRunConfig{ + Path: "ls", + Args: []string{"some", "args"}, + }, + } + + chosenWorker = runtimetest.NewWorker("worker"). + WithContainer( + expectedOwner, + runtimetest.NewContainer().WithProcess( + runtime.ProcessSpec{ + ID: "task", + Path: "ls", + Args: []string{"some", "args"}, + Dir: "some-artifact-root", + TTY: &runtime.TTYSpec{ + WindowSize: runtime.WindowSize{ + Columns: 500, + Rows: 500, + }, + }, + }, + runtimetest.ProcessStub{Attachable: true}, + ), + nil, + ) + chosenContainer = chosenWorker.Containers[0] + fakePool = new(execfakes.FakePool) + fakePool.FindOrSelectWorkerReturns(chosenWorker, nil) + }) + + It("Task env includes atc external url", func() { + Expect(chosenContainer.Spec.Env).To(ConsistOf("ATC_EXTERNAL_URL=http://foo.bar", "SECURE=secret-task-param")) + }) + + Context("before running the task", func() { + BeforeEach(func() { + chosenContainer.ProcessDefs[0].Stub.Do = func(_ context.Context, _ *runtimetest.Process) error { + defer GinkgoRecover() + Expect(fakeDelegate.InitializingCallCount()).To(Equal(1)) + + return nil + } + }) + + It("invokes the delegate's Initializing callback", func() { + // validate the process actually ran + Expect(chosenContainer.RunningProcesses()).To(HaveLen(1)) + }) + }) + + Describe("worker selection", func() { + var ctx context.Context + var workerSpec worker.Spec + + JustBeforeEach(func() { + Expect(fakePool.FindOrSelectWorkerCallCount()).To(Equal(1)) + ctx, _, _, workerSpec, _, _ = fakePool.FindOrSelectWorkerArgsForCall(0) + }) + + It("doesn't enforce a timeout", func() { + _, ok := ctx.Deadline() + Expect(ok).To(BeFalse()) + }) + + It("emits a BeforeSelectWorker event", func() { + Expect(fakeDelegate.BeforeSelectWorkerCallCount()).To(Equal(1)) + }) + + It("emits a SelectedWorker event", func() { + Expect(fakeDelegate.SelectedWorkerCallCount()).To(Equal(1)) + _, workerName := fakeDelegate.SelectedWorkerArgsForCall(0) + Expect(workerName).To(Equal("worker")) + }) + + Context("when tags are configured", func() { + BeforeEach(func() { + taskPlan.Tags = atc.Tags{"plan", "tags"} + }) + + It("creates a worker spec with the tags", func() { + Expect(workerSpec.Tags).To(Equal([]string{"plan", "tags"})) + }) + }) + + Context("when selecting a worker fails", func() { + BeforeEach(func() { + fakePool.FindOrSelectWorkerReturns(nil, errors.New("nope")) + }) + + It("returns an err", func() { + Expect(stepErr).To(MatchError(ContainSubstring("nope"))) + }) + }) + }) + + It("sets the config on the TaskDelegate", func() { + Expect(fakeDelegate.SetTaskConfigCallCount()).To(Equal(1)) + actualTaskConfig := fakeDelegate.SetTaskConfigArgsForCall(0) + Expect(actualTaskConfig).To(Equal(*taskPlan.Config)) + }) + + Context("when privileged", func() { + BeforeEach(func() { + taskPlan.Privileged = true + }) + + It("marks the container's image spec as privileged", func() { + Expect(chosenContainer.Spec.ImageSpec.Privileged).To(BeTrue()) + }) + }) + + It("uses the correct container limits", func() { + Expect(atc.CPULimit(*chosenContainer.Spec.Limits.CPU)).To(Equal(atc.CPULimit(1024))) + Expect(atc.MemoryLimit(*chosenContainer.Spec.Limits.Memory)).To(Equal(atc.MemoryLimit(1024))) + }) + + Context("when toplevel limits are set", func() { + BeforeEach(func() { + cpu := atc.CPULimit(2048) + memory := atc.MemoryLimit(2048) + taskPlan.Limits = &atc.ContainerLimits{ + CPU: &cpu, + Memory: &memory, + } + }) + + It("overrides the limits from the config", func() { + Expect(atc.CPULimit(*chosenContainer.Spec.Limits.CPU)).To(Equal(atc.CPULimit(2048))) + Expect(atc.MemoryLimit(*chosenContainer.Spec.Limits.Memory)).To(Equal(atc.MemoryLimit(2048))) + }) + }) + + Context("when hermetic is configured", func() { + BeforeEach(func() { + taskPlan.Hermetic = true + }) + + It("returns the context.Canceled error", func() { + Expect(chosenContainer.Spec.Hermetic).To(Equal(true)) + }) + }) + + Context("when a user is configured", func() { + BeforeEach(func() { + taskPlan.Config.Run.User = "some-user" + }) + + It("sets the user on the container spec", func() { + Expect(chosenContainer.Spec.User).To(Equal("some-user")) + }) + }) + + Context("when a timeout is configured", func() { + BeforeEach(func() { + taskPlan.Timeout = "1ms" + + chosenContainer.ProcessDefs[0].Stub.Do = func(ctx context.Context, _ *runtimetest.Process) error { + select { + case <-ctx.Done(): + return fmt.Errorf("wrapped: %w", ctx.Err()) + case <-time.After(100 * time.Millisecond): + return nil + } + } + }) + + It("fails without error", func() { + Expect(stepOk).To(BeFalse()) + Expect(stepErr).To(BeNil()) + }) + + It("emits an Errored event", func() { + Expect(fakeDelegate.ErroredCallCount()).To(Equal(1)) + _, status := fakeDelegate.ErroredArgsForCall(0) + Expect(status).To(Equal(exec.TimeoutLogMessage)) + }) + + Context("when the timeout is bogus", func() { + BeforeEach(func() { + taskPlan.Timeout = "bogus" + }) + + It("fails miserably", func() { + Expect(stepErr).To(MatchError("parse timeout: time: invalid duration \"bogus\"")) + }) + }) + }) + + Context("when there is default task timeout", func() { + BeforeEach(func() { + defaultTaskTimeout = time.Minute * 30 + }) + + It("enforces it on the task", func() { + t, ok := chosenContainer.ContextOfRun().Deadline() + Expect(ok).To(BeTrue()) + Expect(t).To(BeTemporally("~", time.Now().Add(time.Minute*30), time.Minute)) + }) + }) + + Context("when there is default task timeout and the plan specifies a timeout also", func() { + BeforeEach(func() { + defaultTaskTimeout = time.Minute * 30 + taskPlan.Timeout = "1h" + }) + + It("enforces the plan's timeout on the task step", func() { + t, ok := chosenContainer.ContextOfRun().Deadline() + Expect(ok).To(BeTrue()) + Expect(t).To(BeTemporally("~", time.Now().Add(time.Hour), time.Minute)) + }) + }) + + Context("when rootfs uri is set instead of image resource", func() { + BeforeEach(func() { + taskPlan.Config.RootfsURI = "some-image" + }) + + It("correctly sets up the image spec", func() { + Expect(chosenContainer.Spec.ImageSpec).To(Equal(runtime.ImageSpec{ + ImageURL: "some-image", + Privileged: false, + })) + }) + }) + + Context("when tracing is enabled", func() { + BeforeEach(func() { + defaultTaskTimeout = 0 + exporter := tracetest.NewInMemoryExporter() + tp := trace.NewTracerProvider(trace.WithSyncer(exporter)) + tracing.ConfigureTraceProvider(tp) + + spanCtx, buildSpan := tracing.StartSpan(ctx, "build", nil) + fakeDelegate.StartSpanReturns(spanCtx, buildSpan) + + chosenContainer.ProcessDefs[0].Stub.Do = func(ctx context.Context, _ *runtimetest.Process) error { + defer GinkgoRecover() + // Properly propagates span context + Expect(tracing.FromContext(ctx)).To(Equal(buildSpan)) + return nil + } + }) + + AfterEach(func() { + tracing.Configured = false + }) + + It("populates the TRACEPARENT env var", func() { + Expect(chosenContainer.Spec.Env).To(ContainElement(MatchRegexp(`TRACEPARENT=.+`))) + }) + }) + + Context("when the configuration specifies paths for inputs", func() { + var input1 *runtimetest.Volume + var input2 *runtimetest.Volume + + BeforeEach(func() { + input1 = runtimetest.NewVolume("input1") + input2 = runtimetest.NewVolume("input2") + + taskPlan.Config.Inputs = []atc.TaskInputConfig{ + {Name: "some-input", Path: "some-input-configured-path"}, + {Name: "some-other-input"}, + } + }) + + Context("when all inputs are present", func() { + var input3 *runtimetest.Volume + var input4 *runtimetest.Volume + + BeforeEach(func() { + input3 = runtimetest.NewVolume("input3") + input4 = runtimetest.NewVolume("input4") + + // If or not framCache when RegisterArtifact, that should not impact task execution. + repo.RegisterArtifact("some-input", input1, false) + repo.RegisterArtifact("some-other-input", input2, true) + repo.RegisterArtifact("absolute-input", input3, false) + repo.RegisterArtifact("parent-dirs-input", input4, false) + + taskPlan.Config.Inputs = []atc.TaskInputConfig{ + {Name: "some-input", Path: "some-input-configured-path"}, + {Name: "some-other-input"}, + {Name: "absolute-input", Path: "/absolute-input"}, + {Name: "parent-dirs-input", Path: "../parent-dirs/../input"}, + } + }) + + It("configures the inputs for the containerSpec correctly", func() { + Expect(chosenContainer.Spec.Inputs).To(ConsistOf([]runtime.Input{ + { + Artifact: input1, + DestinationPath: "some-artifact-root/some-input-configured-path", + FromCache: false, + }, + { + Artifact: input2, + DestinationPath: "some-artifact-root/some-other-input", + FromCache: true, + }, + { + Artifact: input3, + DestinationPath: "/absolute-input", + FromCache: false, + }, + { + Artifact: input4, + DestinationPath: "some-artifact-root/parent-dirs/input", + FromCache: false, + }, + })) + Expect(stepErr).ToNot(HaveOccurred()) + }) + }) + + Context("when any of the inputs are missing", func() { + BeforeEach(func() { + repo.RegisterArtifact("some-input", input1, false) + }) + + It("returns a MissingInputsError", func() { + Expect(stepErr).To(BeAssignableToTypeOf(exec.MissingInputsError{})) + Expect(stepErr.(exec.MissingInputsError).Inputs).To(ConsistOf("some-other-input")) + }) + }) + + Context("when only inputs are cache", func() { + BeforeEach(func() { + // If or not framCache when RegisterArtifact, that should not impact task execution. + repo.RegisterArtifact("some-input", input1, true) + repo.RegisterArtifact("some-other-input", input2, true) + }) + + It("configures the inputs for the containerSpec correctly", func() { + Expect(chosenContainer.Spec.Inputs).To(ConsistOf([]runtime.Input{ + { + Artifact: input1, + DestinationPath: "some-artifact-root/some-input-configured-path", + FromCache: true, + }, + { + Artifact: input2, + DestinationPath: "some-artifact-root/some-other-input", + FromCache: true, + }, + })) + Expect(stepErr).ToNot(HaveOccurred()) + }) + }) + }) + + Context("when input is remapped", func() { + var remappedInputArtifact *runtimetest.Volume + + BeforeEach(func() { + remappedInputArtifact = runtimetest.NewVolume("input1") + taskPlan.InputMapping = map[string]string{"remapped-input": "remapped-input-src"} + taskPlan.Config.Inputs = []atc.TaskInputConfig{ + {Name: "remapped-input"}, + } + }) + + Context("when all inputs are present in the in artifact repository", func() { + BeforeEach(func() { + repo.RegisterArtifact("remapped-input-src", remappedInputArtifact, false) + }) + + It("uses remapped input", func() { + Expect(chosenContainer.Spec.Inputs).To(ConsistOf([]runtime.Input{ + { + Artifact: remappedInputArtifact, + DestinationPath: "some-artifact-root/remapped-input", + }, + })) + Expect(stepErr).ToNot(HaveOccurred()) + }) + }) + }) + + Context("when some inputs are optional", func() { + var ( + optionalInputArtifact, optionalInput2Artifact, requiredInputArtifact *runtimetest.Volume + ) + + BeforeEach(func() { + optionalInputArtifact = runtimetest.NewVolume("optional1") + optionalInput2Artifact = runtimetest.NewVolume("optional2") + requiredInputArtifact = runtimetest.NewVolume("required1") + taskPlan.Config.Inputs = []atc.TaskInputConfig{ + {Name: "optional-input", Optional: true}, + {Name: "optional-input-2", Optional: true}, + {Name: "required-input"}, + } + }) + + Context("when an optional input is missing", func() { + BeforeEach(func() { + repo.RegisterArtifact("required-input", requiredInputArtifact, false) + repo.RegisterArtifact("optional-input-2", optionalInput2Artifact, false) + }) + + It("runs successfully without the optional input", func() { + Expect(chosenContainer.Spec.Inputs).To(ConsistOf([]runtime.Input{ + { + Artifact: requiredInputArtifact, + DestinationPath: "some-artifact-root/required-input", + }, + { + Artifact: optionalInput2Artifact, + DestinationPath: "some-artifact-root/optional-input-2", + }, + })) + Expect(stepErr).ToNot(HaveOccurred()) + }) + }) + + Context("when a required input is missing", func() { + BeforeEach(func() { + repo.RegisterArtifact("optional-input", optionalInputArtifact, false) + repo.RegisterArtifact("optional-input-2", optionalInput2Artifact, false) + }) + + It("returns a MissingInputsError", func() { + Expect(stepErr).To(BeAssignableToTypeOf(exec.MissingInputsError{})) + Expect(stepErr.(exec.MissingInputsError).Inputs).To(ConsistOf("required-input")) + }) + }) + }) + + Context("when the configuration specifies paths for caches", func() { + var ( + volume1 *runtimetest.Volume + volume2 *runtimetest.Volume + ) + + BeforeEach(func() { + taskPlan.Config.Caches = []atc.TaskCacheConfig{ + {Path: "some-path-1"}, + {Path: "some-path-2"}, + } + + volume1 = runtimetest.NewVolume("volume1") + volume2 = runtimetest.NewVolume("volume2") + + chosenContainer.Mounts = []runtime.VolumeMount{ + { + Volume: volume1, + MountPath: "some-artifact-root/some-path-1", + }, + { + Volume: volume2, + MountPath: "some-artifact-root/some-path-2", + }, + } + }) + + It("creates the containerSpec with the caches", func() { + Expect(chosenContainer.Spec.Caches).To(ConsistOf("some-path-1", "some-path-2")) + }) + + itRegistersCaches := func(didRegister bool) { + It("registers cache volumes as task caches", func() { + Expect(volume1.TaskCacheInitialized).To(Equal(didRegister)) + Expect(volume2.TaskCacheInitialized).To(Equal(didRegister)) + }) + } + + Context("when task belongs to a job", func() { + BeforeEach(func() { + stepMetadata.JobID = 12 + }) + + Context("when the task succeeds", func() { + itRegistersCaches(true) + }) + + Context("when the task exits nonzero", func() { + BeforeEach(func() { + chosenContainer.ProcessDefs[0].Stub.ExitStatus = 1 + }) + + itRegistersCaches(true) + }) + + Context("when the task errors", func() { + BeforeEach(func() { + chosenContainer.ProcessDefs[0].Stub.Err = "blah" + }) + + itRegistersCaches(true) + }) + }) + + Context("when task does not belong to job (one-off build)", func() { + BeforeEach(func() { + stepMetadata.JobID = 0 + }) + + It("does not error", func() { + Expect(stepErr).ToNot(HaveOccurred()) + }) + + itRegistersCaches(false) + }) + }) + + Context("when task config's cache TTL is not set", func() { + var volume1 *runtimetest.Volume + + BeforeEach(func() { + stepMetadata.JobID = 12 + + taskPlan.Config.Caches = []atc.TaskCacheConfig{ + {Path: "some-path-1"}, + } + + volume1 = runtimetest.NewVolume("volume1") + + chosenContainer.Mounts = []runtime.VolumeMount{ + { + Volume: volume1, + MountPath: "some-artifact-root/some-path-1", + }, + } + }) + + Context("when defaultTaskCacheTTL is set", func() { + BeforeEach(func() { + defaultTaskCacheTTL = 24 * time.Hour + DeferCleanup(func() { + defaultTaskCacheTTL = 0 + }) + }) + + It("uses the defaultTaskCacheTTL", func() { + Expect(volume1.TaskCacheInitialized).To(BeTrue()) + Expect(volume1.TaskCacheTTL).To(Equal(24 * time.Hour)) + }) + }) + + Context("when defaultTaskCacheTTL is not set", func() { + It("uses zero TTL", func() { + Expect(volume1.TaskCacheInitialized).To(BeTrue()) + Expect(volume1.TaskCacheTTL).To(Equal(time.Duration(0))) + }) + }) + }) + + Context("when task config's cache TTL is set", func() { + var volume1 *runtimetest.Volume + + BeforeEach(func() { + stepMetadata.JobID = 12 + + taskPlan.Config.Caches = []atc.TaskCacheConfig{ + {Path: "some-path-1", TTL: "1h"}, + } + + volume1 = runtimetest.NewVolume("volume1") + + chosenContainer.Mounts = []runtime.VolumeMount{ + { + Volume: volume1, + MountPath: "some-artifact-root/some-path-1", + }, + } + }) + + Context("when defaultTaskCacheTTL is also set", func() { + BeforeEach(func() { + defaultTaskCacheTTL = 24 * time.Hour + DeferCleanup(func() { + defaultTaskCacheTTL = 0 + }) + }) + + It("uses task config's cache TTL over defaultTaskCacheTTL", func() { + Expect(volume1.TaskCacheInitialized).To(BeTrue()) + Expect(volume1.TaskCacheTTL).To(Equal(1 * time.Hour)) + }) + }) + + Context("when defaultTaskCacheTTL is not set", func() { + It("uses cacheConfig.TTL", func() { + Expect(volume1.TaskCacheInitialized).To(BeTrue()) + Expect(volume1.TaskCacheTTL).To(Equal(1 * time.Hour)) + }) + }) + }) + + Context("when the configuration specifies paths for outputs", func() { + var outputVolume1, outputVolume2, outputVolume3, + outputVolume4, outputVolume5 *runtimetest.Volume + + BeforeEach(func() { + taskPlan.Config.Outputs = []atc.TaskOutputConfig{ + {Name: "some-output", Path: "some-output-configured-path"}, + {Name: "some-other-output"}, + {Name: "some-trailing-slash-output", Path: "some-output-configured-path-with-trailing-slash/"}, + {Name: "absolute-output", Path: "/absolute/output"}, + {Name: "parent-dir-output", Path: "../parent/dir/../output"}, + } + taskPlan.OutputMapping = map[string]string{ + "some-other-output": "some-remapped-output", + } + + outputVolume1 = runtimetest.NewVolume("output1") + outputVolume2 = runtimetest.NewVolume("output2") + outputVolume3 = runtimetest.NewVolume("output3") + outputVolume4 = runtimetest.NewVolume("output4") + outputVolume5 = runtimetest.NewVolume("output5") + + chosenContainer.Mounts = []runtime.VolumeMount{ + { + Volume: outputVolume1, + MountPath: "some-artifact-root/some-output-configured-path/", + }, + { + Volume: outputVolume2, + MountPath: "some-artifact-root/some-other-output/", + }, + { + Volume: outputVolume3, + MountPath: "some-artifact-root/some-output-configured-path-with-trailing-slash/", + }, + { + Volume: outputVolume4, + MountPath: "/absolute/output/", + }, + { + Volume: outputVolume5, + MountPath: "some-artifact-root/parent/dir/output/", + }, + } + }) + + It("configures them appropriately in the container spec", func() { + Expect(chosenContainer.Spec.Outputs).To(Equal(runtime.OutputPaths{ + "some-output": "some-artifact-root/some-output-configured-path/", + "some-other-output": "some-artifact-root/some-other-output/", + "some-trailing-slash-output": "some-artifact-root/some-output-configured-path-with-trailing-slash/", + "absolute-output": "/absolute/output/", + "parent-dir-output": "some-artifact-root/parent/dir/output/", + })) + }) + + It("registers the outputs in the build repo", func() { + Expect(repo.AsMap()).To(Equal(map[build.ArtifactName]build.ArtifactEntry{ + "some-output": { + Artifact: outputVolume1, + FromCache: false, + }, + "some-remapped-output": { + Artifact: outputVolume2, + FromCache: false, + }, + "some-trailing-slash-output": { + Artifact: outputVolume3, + FromCache: false, + }, + "absolute-output": { + Artifact: outputVolume4, + FromCache: false, + }, + "parent-dir-output": { + Artifact: outputVolume5, + FromCache: false, + }, + })) + }) + }) + + Context("when missing the platform", func() { + BeforeEach(func() { + taskPlan.Config.Platform = "" + }) + + It("returns the error", func() { + Expect(stepErr).To(HaveOccurred()) + }) + + It("is not successful", func() { + Expect(stepOk).To(BeFalse()) + }) + }) + + Context("when missing the path to the executable and ENTRYPOINT/CMD is not defined", func() { + BeforeEach(func() { + taskPlan.Config.Run.Path = "" + }) + + It("returns the error", func() { + Expect(stepErr).To(HaveOccurred()) + Expect(stepErr.Error()).To(ContainSubstring("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.")) + }) + + It("is not successful", func() { + Expect(stepOk).To(BeFalse()) + }) + + Context("when the image's metadata.json is malformed", func() { + BeforeEach(func() { + taskPlan.Config.Run.Path = "" + fakeStreamer.StreamFileReturns(io.NopCloser(strings.NewReader("definitely not json")), nil) + + taskPlan.ImageArtifactName = "some-image-artifact" + imageVolume := runtimetest.NewVolume("image-volume") + repo.RegisterArtifact("some-image-artifact", imageVolume, false) + }) + + It("returns the error", func() { + Expect(stepErr).To(HaveOccurred()) + Expect(stepErr.Error()).To(ContainSubstring("error parsing metadata.json from rootfs")) + }) + + It("is not successful", func() { + Expect(stepOk).To(BeFalse()) + }) + + }) + + Context("when the image's metadata.json is not present in the image artifact", func() { + BeforeEach(func() { + taskPlan.Config.Run.Path = "" + fakeStreamer.StreamFileReturns(nil, baggageclaim.ErrFileNotFound) + + taskPlan.ImageArtifactName = "some-image-artifact" + imageVolume := runtimetest.NewVolume("image-volume") + repo.RegisterArtifact("some-image-artifact", imageVolume, false) + }) + + It("returns the error", func() { + Expect(stepErr).To(HaveOccurred()) + Expect(stepErr.Error()).To(ContainSubstring("file 'metadata.json' not found within artifact 'image-volume'")) + }) + + It("is not successful", func() { + Expect(stepOk).To(BeFalse()) + }) + + }) + }) + + Context("the task container has an ENTRYPOINT/CMD defined", func() { + BeforeEach(func() { + taskPlan.Config.Run.Path = "" + taskPlan.Config.Run.Args = []string{} + rootfsMetadata := `{ + "entrypoint": [ + "/bin/sh", + "-c" + ], + "cmd": [ + "echo hello world" + ] +}` + fakeStreamer.StreamFileReturns(io.NopCloser(strings.NewReader(rootfsMetadata)), nil) + + taskPlan.ImageArtifactName = "some-image-artifact" + imageVolume := runtimetest.NewVolume("image-volume") + repo.RegisterArtifact("some-image-artifact", imageVolume, false) + }) + + It("uses the ENTRYPOINT/CMD and passes it back to the task delegate", func() { + Expect(fakeDelegate.SetTaskConfigCallCount()).To(Equal(1)) + actualTaskConfig := fakeDelegate.SetTaskConfigArgsForCall(0) + Expect(actualTaskConfig.Run.Path).To(Equal("/bin/sh")) + Expect(actualTaskConfig.Run.Args).To(Equal([]string{"-c", "echo hello world"})) + }) + + Context("there are args in the task config", func() { + BeforeEach(func() { + taskPlan.Config.Run.Args = []string{"task", "args"} + }) + + It("they are appended to the ENTRYPOINT/CMD args", func() { + Expect(fakeDelegate.SetTaskConfigCallCount()).To(Equal(1)) + actualTaskConfig := fakeDelegate.SetTaskConfigArgsForCall(0) + Expect(actualTaskConfig.Run.Path).To(Equal("/bin/sh")) + Expect(actualTaskConfig.Run.Args).To(Equal([]string{"-c", "echo hello world", "task", "args"})) + }) + }) + + Context("only ENTRYPOINT is specified", func() { + BeforeEach(func() { + taskPlan.Config.Run.Path = "" + taskPlan.Config.Run.Args = []string{} + rootfsMetadata := `{ "entrypoint": [ "some-program", "some-arg" ] }` + fakeStreamer.StreamFileReturns(io.NopCloser(strings.NewReader(rootfsMetadata)), nil) + }) + + It("is parsed correctly", func() { + Expect(fakeDelegate.SetTaskConfigCallCount()).To(Equal(1)) + actualTaskConfig := fakeDelegate.SetTaskConfigArgsForCall(0) + Expect(actualTaskConfig.Run.Path).To(Equal("some-program")) + Expect(actualTaskConfig.Run.Args).To(Equal([]string{"some-arg"})) + }) + }) + + Context("only CMD is specified", func() { + BeforeEach(func() { + taskPlan.Config.Run.Path = "" + taskPlan.Config.Run.Args = []string{} + rootfsMetadata := `{ "cmd": [ "cmd-program", "cmd-arg" ] }` + fakeStreamer.StreamFileReturns(io.NopCloser(strings.NewReader(rootfsMetadata)), nil) + }) + + It("is parsed correctly", func() { + Expect(fakeDelegate.SetTaskConfigCallCount()).To(Equal(1)) + actualTaskConfig := fakeDelegate.SetTaskConfigArgsForCall(0) + Expect(actualTaskConfig.Run.Path).To(Equal("cmd-program")) + Expect(actualTaskConfig.Run.Args).To(Equal([]string{"cmd-arg"})) + }) + }) + }) + + Context("when an image artifact name is specified", func() { + BeforeEach(func() { + taskPlan.ImageArtifactName = "some-image-artifact" + }) + + Context("when the image artifact is registered in the artifact repo", func() { + var imageVolume *runtimetest.Volume + + BeforeEach(func() { + imageVolume = runtimetest.NewVolume("image-volume") + repo.RegisterArtifact("some-image-artifact", imageVolume, false) + }) + + It("configures it in the containerSpec's ImageSpec", func() { + Expect(chosenContainer.Spec.ImageSpec).To(Equal(runtime.ImageSpec{ + ImageArtifact: imageVolume, + })) + + expectWorkerSpecResourceTypeUnset() + }) + + Describe("when task config specifies image and/or image resource as well as image artifact", func() { + Context("when streaming the metadata from the worker succeeds", func() { + JustBeforeEach(func() { + Expect(stepErr).ToNot(HaveOccurred()) + }) + + Context("when the task config also specifies a rootfs_uri", func() { + BeforeEach(func() { + taskPlan.Config.RootfsURI = "some-image" + }) + + It("still uses the image artifact", func() { + Expect(chosenContainer.Spec.ImageSpec).To(Equal(runtime.ImageSpec{ + ImageArtifact: imageVolume, + })) + expectWorkerSpecResourceTypeUnset() + }) + }) + + Context("when the task config also specifies image_resource", func() { + BeforeEach(func() { + taskPlan.Config.ImageResource = &atc.ImageResource{ + Type: "docker", + Source: atc.Source{"some": "super-secret-source"}, + Params: atc.Params{"some": "params"}, + Version: atc.Version{"some": "version"}, + } + }) + + It("still uses the image artifact", func() { + Expect(chosenContainer.Spec.ImageSpec).To(Equal(runtime.ImageSpec{ + ImageArtifact: imageVolume, + })) + expectWorkerSpecResourceTypeUnset() + }) + }) + }) + }) + }) + + Context("when the image artifact is NOT registered in the artifact repo", func() { + It("returns a MissingTaskImageSourceError", func() { + Expect(stepErr).To(Equal(exec.MissingTaskImageSourceError{"some-image-artifact"})) + }) + + It("is not successful", func() { + Expect(stepOk).To(BeFalse()) + }) + }) + }) + + Context("when the image_resource is specified (even if rootfs_uri is configured)", func() { + var fetchedImageSpec runtime.ImageSpec + + BeforeEach(func() { + taskPlan.Config.RootfsURI = "some-image" + taskPlan.Config.ImageResource = &atc.ImageResource{ + Type: "docker", + Source: atc.Source{"some": "super-secret-source"}, + Params: atc.Params{"some": "params"}, + } + taskPlan.Tags = atc.Tags{"some", "tags"} + + fetchedImageSpec = runtime.ImageSpec{ + ImageArtifact: runtimetest.NewVolume("some-volume"), + } + + fakeDelegate.FetchImageReturns(fetchedImageSpec, nil) + }) + + It("succeeds", func() { + Expect(stepErr).ToNot(HaveOccurred()) + Expect(stepOk).To(BeTrue()) + }) + + It("fetches the image", func() { + Expect(fakeDelegate.FetchImageCallCount()).To(Equal(1)) + _, imageResource, types, privileged, tags, skipInterval := fakeDelegate.FetchImageArgsForCall(0) + Expect(imageResource).To(Equal(atc.ImageResource{ + Type: "docker", + Source: atc.Source{"some": "super-secret-source"}, + Params: atc.Params{"some": "params"}, + })) + Expect(types).To(Equal(taskPlan.ResourceTypes)) + Expect(privileged).To(BeFalse()) + Expect(tags).To(Equal(atc.Tags{"some", "tags"})) + Expect(skipInterval).To(Equal(false)) + }) + + It("creates the specs with the fetched image", func() { + Expect(chosenContainer.Spec.ImageSpec).To(Equal(fetchedImageSpec)) + }) + + Context("when privileged", func() { + BeforeEach(func() { + taskPlan.Privileged = true + }) + + It("fetches a privileged image", func() { + Expect(fakeDelegate.FetchImageCallCount()).To(Equal(1)) + _, _, _, privileged, _, _ := fakeDelegate.FetchImageArgsForCall(0) + Expect(privileged).To(BeTrue()) + }) + }) + + Context("when check skip interval is true", func() { + BeforeEach(func() { + taskPlan.CheckSkipInterval = true + }) + + It("fetches an image with forced check", func() { + Expect(fakeDelegate.FetchImageCallCount()).To(Equal(1)) + _, _, _, _, _, skipInterval := fakeDelegate.FetchImageArgsForCall(0) + Expect(skipInterval).To(BeTrue()) + }) + }) + }) + + Context("when a run dir and user are specified", func() { + BeforeEach(func() { + taskPlan.Config.Run.Dir = "/some/dir" + taskPlan.Config.Run.User = "some-user" + + chosenContainer.ProcessDefs[0].Spec.Dir = "/some/dir" + chosenContainer.ProcessDefs[0].Spec.User = "some-user" + }) + + It("runs with the specified dir and user", func() { + Expect(stepErr).ToNot(HaveOccurred()) + Expect(chosenContainer.RunningProcesses()).To(HaveLen(1)) + }) + }) + + Context("when running the task exits with a non-zero status", func() { + BeforeEach(func() { + chosenContainer.ProcessDefs[0].Stub.ExitStatus = 1 + }) + + It("doesn't error", func() { + Expect(stepErr).ToNot(HaveOccurred()) + }) + + It("is not successful", func() { + Expect(stepOk).To(BeFalse()) + }) + + It("finishes the step", func() { + Expect(fakeDelegate.FinishedCallCount()).To(Equal(1)) + _, status := fakeDelegate.FinishedArgsForCall(0) + Expect(status).To(Equal(exec.ExitStatus(1))) + }) + }) + + Context("when running the task fails", func() { + BeforeEach(func() { + chosenContainer.ProcessDefs[0].Stub.Err = "failed to run the task" + }) + + It("returns the error", func() { + Expect(stepErr).To(HaveOccurred()) + }) + + It("is not successful", func() { + Expect(stepOk).To(BeFalse()) + }) + }) + + Context("when the task step is interrupted", func() { + BeforeEach(func() { + cancel() + chosenContainer.ProcessDefs[0].Stub.Do = func(ctx context.Context, _ *runtimetest.Process) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(1 * time.Second): + Fail("didn't return context error") + panic("unreachable") + } + } + }) + + It("returns the context.Canceled error", func() { + Expect(stepErr).To(Equal(context.Canceled)) + }) + + It("is not successful", func() { + Expect(stepOk).To(BeFalse()) + }) + + It("doesn't register a artifact", func() { + Expect(repo.AsMap()).To(BeEmpty()) + }) + }) + }) +})