ci-operator: allow users to depend on images in steps

Users may not expect to find the pull spec for an image created at some
other point in the pipeline fed to their step via an environment
variable that they specify. The ci-operator will ensure that their step
only runs once the image is available.

Signed-off-by: Steve Kuznetsov <skuznets@redhat.com>

Author: stevekuznetsov

pkg/api/config.go

@@ -469,6 +469,7 @@ func validateLiteralTestStepCommon(fieldRoot string, step LiteralTestStep, seen
 	if err := validateParameters(fieldRoot, step.Environment, env); err != nil {
 		ret = append(ret, err)
 	}
+	ret = append(ret, validateDependencies(fieldRoot, step.Dependencies)...)
 	return
 }
 
@@ -551,6 +552,26 @@ func validateParameters(fieldRoot string, params []StepParameter, env TestEnviro
 	return nil
 }
 
+func validateDependencies(fieldRoot string, dependencies []StepDependency) []error {
+	var errs []error
+	env := sets.NewString()
+	for i, dependency := range dependencies {
+		if dependency.Name == "" {
+			errs = append(errs, fmt.Errorf("%s.dependencies[%d].name must be set", fieldRoot, i))
+		} else if strings.Contains(dependency.Name, ":") && len(strings.Split(dependency.Name, ":")) != 2 {
+			errs = append(errs, fmt.Errorf("%s.dependencies[%d].name must take the `tag` or `stream:tag` form, not %q", fieldRoot, i, dependency.Name))
+		}
+		if dependency.Env == "" {
+			errs = append(errs, fmt.Errorf("%s.dependencies[%d].env must be set", fieldRoot, i))
+		} else if env.Has(dependency.Env) {
+			errs = append(errs, fmt.Errorf("%s.dependencies[%d].env targets an environment variable that is already set by another dependency", fieldRoot, i))
+		} else {
+			env.Insert(dependency.Env)
+		}
+	}
+	return errs
+}
+
 func validateReleaseBuildConfiguration(input *ReleaseBuildConfiguration, org, repo string) []error {
 	var validationErrors []error
 

pkg/api/config_test.go

@@ -1487,3 +1487,48 @@ func errListMessagesEqual(a, b []error) bool {
 	}
 	return true
 }
+
+func TestValidateDependencies(t *testing.T) {
+	var testCases = []struct {
+		name   string
+		input  []StepDependency
+		output []error
+	}{
+		{
+			name:  "no dependencies",
+			input: nil,
+		},
+		{
+			name: "valid dependencies",
+			input: []StepDependency{
+				{Name: "src", Env: "SOURCE"},
+				{Name: "stable:installer", Env: "INSTALLER"},
+			},
+		},
+		{
+			name: "invalid dependencies",
+			input: []StepDependency{
+				{Name: "", Env: ""},
+				{Name: "src", Env: "SOURCE"},
+				{Name: "src", Env: "SOURCE"},
+				{Name: "src:lol:oops", Env: "WHOA"},
+			},
+			output: []error{
+				errors.New("root.dependencies[0].name must be set"),
+				errors.New("root.dependencies[0].env must be set"),
+				errors.New("root.dependencies[2].env targets an environment variable that is already set by another dependency"),
+				errors.New("root.dependencies[3].name must take the `tag` or `stream:tag` form, not \"src:lol:oops\""),
+			},
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			if actual, expected := validateDependencies("root", testCase.input), testCase.output; !reflect.DeepEqual(actual, expected) {
+				t.Errorf("%s: got incorrect errors: %s", testCase.name, cmp.Diff(actual, expected, cmp.Comparer(func(x, y error) bool {
+					return x.Error() == y.Error()
+				})))
+			}
+		})
+	}
+}

pkg/api/graph.go

@@ -202,6 +202,12 @@ func IsReleaseStream(stream string) bool {
 	return strings.HasPrefix(stream, StableImageStream)
 }
 
+// IsReleasePayloadStream determines if the ImageStream holds
+// release paylaod images.
+func IsReleasePayloadStream(stream string) bool {
+	return stream == ReleaseImageStream
+}
+
 type StepNode struct {
 	Step     Step
 	Children []*StepNode

pkg/api/types.go

@@ -563,6 +563,9 @@ type LiteralTestStep struct {
 	Credentials []CredentialReference `json:"credentials,omitempty"`
 	// Environment lists parameters that should be set by the test.
 	Environment []StepParameter `json:"env,omitempty"`
+	// Dependencies lists images which must be available before the test runs
+	// and the environment variables which are used to expose their pull specs.
+	Dependencies []StepDependency `json:"dependencies,omitempty"`
 	// OptionalOnSuccess defines if this step should be skipped as long
 	// as all `pre` and `test` steps were successful and AllowSkipOnSuccess
 	// flag is set to true in MultiStageTestConfiguration. This option is
@@ -590,6 +593,15 @@ type CredentialReference struct {
 	MountPath string `json:"mount_path"`
 }
 
+// StepDependency defines a dependency on an image and the environment variable
+// used to expose the image's pull spec to the step.
+type StepDependency struct {
+	// Name is the tag or stream:tag that this dependency references
+	Name string `json:"name"`
+	// Env is the environment variable that the image's pull spec is exposed with
+	Env string `json:"env"`
+}
+
 // FromImageTag returns the internal name for the image tag that will be used
 // for this step, if one is configured.
 func (s *LiteralTestStep) FromImageTag() (PipelineImageStreamTagReference, bool) {

pkg/defaults/defaults.go

@@ -224,7 +224,7 @@ func FromConfig(
 			step = release.ImportReleaseStep(resolveConfig.Name, value, false, config.Resources, podClient, imageClient, saGetter, rbacClient, artifactDir, jobSpec)
 		} else if testStep := rawStep.TestStepConfiguration; testStep != nil {
 			if test := testStep.MultiStageTestConfigurationLiteral; test != nil {
-				step = steps.MultiStageTestStep(*testStep, config, params, podClient, secretGetter, saGetter, rbacClient, artifactDir, jobSpec)
+				step = steps.MultiStageTestStep(*testStep, config, params, podClient, secretGetter, saGetter, rbacClient, imageClient, artifactDir, jobSpec)
 				if test.ClusterProfile != "" {
 					step = steps.LeaseStep(leaseClient, test.ClusterProfile.LeaseType(), step, jobSpec.Namespace, namespaceClient)
 				}

pkg/steps/multi_stage.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"strings"
 
+	imageclientset "github.com/openshift/client-go/image/clientset/versioned/typed/image/v1"
 	coreapi "k8s.io/api/core/v1"
 	rbacapi "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -53,6 +54,7 @@ type multiStageTestStep struct {
 	secretClient       coreclientset.SecretsGetter
 	saClient           coreclientset.ServiceAccountsGetter
 	rbacClient         rbacclientset.RbacV1Interface
+	isClient           imageclientset.ImageStreamsGetter
 	artifactDir        string
 	jobSpec            *api.JobSpec
 	pre, test, post    []api.LiteralTestStep
@@ -68,10 +70,11 @@ func MultiStageTestStep(
 	secretClient coreclientset.SecretsGetter,
 	saClient coreclientset.ServiceAccountsGetter,
 	rbacClient rbacclientset.RbacV1Interface,
+	isClient imageclientset.ImageStreamsGetter,
 	artifactDir string,
 	jobSpec *api.JobSpec,
 ) api.Step {
-	return newMultiStageTestStep(testConfig, config, params, podClient, secretClient, saClient, rbacClient, artifactDir, jobSpec)
+	return newMultiStageTestStep(testConfig, config, params, podClient, secretClient, saClient, rbacClient, isClient, artifactDir, jobSpec)
 }
 
 func newMultiStageTestStep(
@@ -82,6 +85,7 @@ func newMultiStageTestStep(
 	secretClient coreclientset.SecretsGetter,
 	saClient coreclientset.ServiceAccountsGetter,
 	rbacClient rbacclientset.RbacV1Interface,
+	isClient imageclientset.ImageStreamsGetter,
 	artifactDir string,
 	jobSpec *api.JobSpec,
 ) *multiStageTestStep {
@@ -99,6 +103,7 @@ func newMultiStageTestStep(
 		secretClient:       secretClient,
 		saClient:           saClient,
 		rbacClient:         rbacClient,
+		isClient:           isClient,
 		artifactDir:        artifactDir,
 		jobSpec:            jobSpec,
 		pre:                ms.Pre,
@@ -179,6 +184,13 @@ func (s *multiStageTestStep) Requires() (ret []api.StepLink) {
 		if link, ok := step.FromImageTag(); ok {
 			internalLinks[link] = struct{}{}
 		}
+
+		for _, dependency := range step.Dependencies {
+			imageStream, name := s.parts(dependency)
+			if link, ok := utils.LinkForImage(imageStream, name); ok {
+				ret = append(ret, link)
+			}
+		}
 	}
 	for link := range internalLinks {
 		ret = append(ret, api.InternalImageLink(link))
@@ -197,6 +209,17 @@ func (s *multiStageTestStep) Requires() (ret []api.StepLink) {
 	return
 }
 
+// parts returns the imageStream and tag name from a user-provided
+// reference to an image in the test namespace
+func (s multiStageTestStep) parts(dependency api.StepDependency) (string, string) {
+	if !strings.Contains(dependency.Name, ":") {
+		return s.imageStreamFor(dependency.Name), dependency.Name
+	} else {
+		parts := strings.Split(dependency.Name, ":")
+		return parts[0], parts[1]
+	}
+}
+
 func (s *multiStageTestStep) Creates() []api.StepLink { return nil }
 func (s *multiStageTestStep) Provides() api.ParameterMap {
 	return nil
@@ -321,6 +344,21 @@ func (s *multiStageTestStep) runSteps(
 	return utilerrors.NewAggregate(errs)
 }
 
+// imageStreamFor guesses at the ImageStream that will hold a tag.
+// We use this to decipher the user's intent when they provide a
+// naked tag in configuration; we support such behavior in order to
+// allow users a simpler workflow for the most common cases, like
+// referring to `pipeline:src`. If they refer to an ambiguous image,
+// however, they will get bad behavior and will need to specify an
+// ImageStrem as well, for instance release-initial:installer.
+func (s *multiStageTestStep) imageStreamFor(image string) string {
+	if s.config.IsPipelineImage(image) || s.config.BuildsImage(image) {
+		return api.PipelineImageStream
+	} else {
+		return api.StableImageStream
+	}
+}
+
 func (s *multiStageTestStep) generatePods(steps []api.LiteralTestStep, env []coreapi.EnvVar,
 	hasPrevErrs bool) ([]coreapi.Pod, error) {
 	var ret []coreapi.Pod
@@ -335,11 +373,7 @@ func (s *multiStageTestStep) generatePods(steps []api.LiteralTestStep, env []cor
 		if link, ok := step.FromImageTag(); ok {
 			image = fmt.Sprintf("%s:%s", api.PipelineImageStream, link)
 		} else {
-			if s.config.IsPipelineImage(image) || s.config.BuildsImage(image) {
-				image = fmt.Sprintf("%s:%s", api.PipelineImageStream, image)
-			} else {
-				image = fmt.Sprintf("%s:%s", api.StableImageStream, image)
-			}
+			image = fmt.Sprintf("%s:%s", s.imageStreamFor(image), image)
 		}
 		resources, err := resourcesFor(step.Resources)
 		if err != nil {
@@ -365,6 +399,12 @@ func (s *multiStageTestStep) generatePods(steps []api.LiteralTestStep, env []cor
 		}...)
 		container.Env = append(container.Env, env...)
 		container.Env = append(container.Env, s.generateParams(step.Environment)...)
+		depEnv, depErrs := s.envForDependencies(step)
+		if len(depErrs) != 0 {
+			errs = append(errs, depErrs...)
+			continue
+		}
+		container.Env = append(container.Env, depEnv...)
 		if owner := s.jobSpec.Owner(); owner != nil {
 			pod.OwnerReferences = append(pod.OwnerReferences, *owner)
 		}
@@ -381,6 +421,23 @@ func (s *multiStageTestStep) generatePods(steps []api.LiteralTestStep, env []cor
 	return ret, utilerrors.NewAggregate(errs)
 }
 
+func (s *multiStageTestStep) envForDependencies(step api.LiteralTestStep) ([]coreapi.EnvVar, []error) {
+	var env []coreapi.EnvVar
+	var errs []error
+	for _, dependency := range step.Dependencies {
+		imageStream, name := s.parts(dependency)
+		ref, err := utils.ImageDigestFor(s.isClient, s.jobSpec.Namespace, imageStream, name)()
+		if err != nil {
+			errs = append(errs, fmt.Errorf("could not determine image pull spec for image %s on step %s", dependency.Name, step.As))
+			continue
+		}
+		env = append(env, coreapi.EnvVar{
+			Name: dependency.Env, Value: ref,
+		})
+	}
+	return env, errs
+}
+
 func addSecretWrapper(pod *coreapi.Pod) {
 	volume := "secret-wrapper"
 	dir := "/tmp/secret-wrapper"

pkg/steps/multi_stage_test.go

@@ -2,14 +2,15 @@ package steps
 
 import (
 	"context"
-	"github.com/google/go-cmp/cmp"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"reflect"
 	"testing"
 	"time"
 
+	"github.com/google/go-cmp/cmp"
+
 	coreapi "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -34,7 +35,7 @@ func TestRequires(t *testing.T) {
 		steps  api.MultiStageTestConfigurationLiteral
 		req    []api.StepLink
 	}{{
-		name: "step has a cluster profile and requires a release image, should not have StableImagesLink",
+		name: "step has a cluster profile and requires a release image, should not have ReleaseImagesLink",
 		steps: api.MultiStageTestConfigurationLiteral{
 			ClusterProfile: api.ClusterProfileAWS,
 			Test:           []api.LiteralTestStep{{From: "from-release"}},
@@ -44,7 +45,7 @@ func TestRequires(t *testing.T) {
 			api.ImagesReadyLink(),
 		},
 	}, {
-		name: "step needs release images, should have StableImagesLink",
+		name: "step needs release images, should have ReleaseImagesLink",
 		steps: api.MultiStageTestConfigurationLiteral{
 			Test: []api.LiteralTestStep{{From: "from-release"}},
 		},
@@ -75,7 +76,7 @@ func TestRequires(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			step := MultiStageTestStep(api.TestStepConfiguration{
 				MultiStageTestConfigurationLiteral: &tc.steps,
-			}, &tc.config, api.NewDeferredParameters(), nil, nil, nil, nil, "", nil)
+			}, &tc.config, api.NewDeferredParameters(), nil, nil, nil, nil, nil, "", nil)
 			ret := step.Requires()
 			if len(ret) == len(tc.req) {
 				matches := true
@@ -127,7 +128,7 @@ func TestGeneratePods(t *testing.T) {
 		},
 	}
 	jobSpec.SetNamespace("namespace")
-	step := newMultiStageTestStep(config.Tests[0], &config, nil, nil, nil, nil, nil, "artifact_dir", &jobSpec)
+	step := newMultiStageTestStep(config.Tests[0], &config, nil, nil, nil, nil, nil, nil, "artifact_dir", &jobSpec)
 	env := []coreapi.EnvVar{
 		{Name: "RELEASE_IMAGE_INITIAL", Value: "release:initial"},
 		{Name: "RELEASE_IMAGE_LATEST", Value: "release:latest"},
@@ -197,7 +198,7 @@ func TestGeneratePodsEnvironment(t *testing.T) {
 					Test:        test,
 					Environment: tc.env,
 				},
-			}, &api.ReleaseBuildConfiguration{}, nil, nil, nil, nil, nil, "", &jobSpec)
+			}, &api.ReleaseBuildConfiguration{}, nil, nil, nil, nil, nil, nil, "", &jobSpec)
 			pods, err := step.(*multiStageTestStep).generatePods(test, nil, false)
 			if err != nil {
 				t.Fatal(err)
@@ -322,7 +323,7 @@ func TestRun(t *testing.T) {
 					Post:               []api.LiteralTestStep{{As: "post0"}, {As: "post1", OptionalOnSuccess: &yes}},
 					AllowSkipOnSuccess: &yes,
 				},
-			}, &api.ReleaseBuildConfiguration{}, nil, &fakePodClient{NewPodClient(client, nil, nil)}, client, client, fakecs.RbacV1(), "", &jobSpec)
+			}, &api.ReleaseBuildConfiguration{}, nil, &fakePodClient{NewPodClient(client, nil, nil)}, client, client, fakecs.RbacV1(), nil, "", &jobSpec)
 			if err := step.Run(context.Background()); tc.failures == nil && err != nil {
 				t.Error(err)
 				return
@@ -378,7 +379,7 @@ func TestArtifacts(t *testing.T) {
 				{As: "test1", ArtifactDir: "/path/to/artifacts"},
 			},
 		},
-	}, &api.ReleaseBuildConfiguration{}, nil, &fakePodClient{NewPodClient(client, nil, nil)}, client, client, fakecs.RbacV1(), tmp, &jobSpec)
+	}, &api.ReleaseBuildConfiguration{}, nil, &fakePodClient{NewPodClient(client, nil, nil)}, client, client, fakecs.RbacV1(), nil, tmp, &jobSpec)
 	if err := step.Run(context.Background()); err != nil {
 		t.Fatal(err)
 	}
@@ -455,7 +456,7 @@ func TestJUnit(t *testing.T) {
 					Test: []api.LiteralTestStep{{As: "test0"}, {As: "test1"}},
 					Post: []api.LiteralTestStep{{As: "post0"}, {As: "post1"}},
 				},
-			}, &api.ReleaseBuildConfiguration{}, nil, &fakePodClient{NewPodClient(client, nil, nil)}, client, client, fakecs.RbacV1(), "/dev/null", &jobSpec)
+			}, &api.ReleaseBuildConfiguration{}, nil, &fakePodClient{NewPodClient(client, nil, nil)}, client, client, fakecs.RbacV1(), nil, "/dev/null", &jobSpec)
 			if err := step.Run(context.Background()); tc.failures == nil && err != nil {
 				t.Error(err)
 				return

pkg/steps/utils/env.go

@@ -154,3 +154,25 @@ func ReleaseNameFrom(envVar string) string {
 	name, _ := imageFromEnv(api.ReleaseImageStream, envVar)
 	return name
 }
+
+// LinkForImage determines what dependent link is required
+// for the user's image dependency
+func LinkForImage(imageStream, tag string) (api.StepLink, bool) {
+	switch {
+	case imageStream == api.PipelineImageStream:
+		// the user needs an image we're building
+		return api.InternalImageLink(api.PipelineImageStreamTagReference(tag)), true
+	case api.IsReleaseStream(imageStream):
+		// the user needs a tag that's a component of some release;
+		// we cant' rely on a specific tag, as they are implicit in
+		// the import process and won't be present in the build graph,
+		// so we wait for the whole import to succeed
+		return api.ReleaseImagesLink(api.ReleaseNameFrom(imageStream)), true
+	case api.IsReleasePayloadStream(imageStream):
+		// the user needs a release payload
+		return api.ReleasePayloadImageLink(tag), true
+	default:
+		// we have no idea what the user's configured
+		return nil, false
+	}
+}