Merge pull request GoogleContainerTools#723 from dgageot/labels

Remove duplication in code handling labels

cmd/skaffold/app/cmd/deploy.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/build"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/docker"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/label"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 )
@@ -73,11 +72,6 @@ func runDeploy(out io.Writer, filename string) error {
 		})
 	}
 
-	dRes, err := r.Deploy(ctx, deployOut, builds)
-	if err != nil {
-		return errors.Wrap(err, "deploy step")
-	}
-	label.LabelDeployResults(r.Labels(), dRes)
-
-	return nil
+	_, err = r.Deploy(ctx, deployOut, builds)
+	return err
 }

pkg/skaffold/build/tag/tag.go

@@ -18,8 +18,9 @@ package tag
 
 // Tagger is an interface for tag strategies to be implemented against
 type Tagger interface {
-	GenerateFullyQualifiedImageName(workingDir string, tagOpts *Options) (string, error)
 	Labels() map[string]string
+
+	GenerateFullyQualifiedImageName(workingDir string, tagOpts *Options) (string, error)
 }
 
 type Options struct {

pkg/skaffold/deploy/deploy.go

@@ -21,9 +21,8 @@ import (
 	"fmt"
 	"io"
 
-	"k8s.io/apimachinery/pkg/runtime"
-
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/build"
+	"k8s.io/apimachinery/pkg/runtime"
 )
 
 // Artifact contains all information about a completed deployment
@@ -35,7 +34,6 @@ type Artifact struct {
 // Deployer is the Deploy API of skaffold and responsible for deploying
 // the build results to a Kubernetes cluster
 type Deployer interface {
-	// Labels returns a list of labels to be attached to each deployed Kubernetes object
 	Labels() map[string]string
 
 	// Deploy should ensure that the build results are deployed to the Kubernetes

pkg/skaffold/runner/label/label.go → pkg/skaffold/deploy/labels.go

@@ -14,19 +14,18 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package label
+package deploy
 
 import (
+	"context"
 	"encoding/json"
+	"io"
 	"time"
 
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/build"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/constants"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
 	"github.com/sirupsen/logrus"
-
-	clientgo "k8s.io/client-go/kubernetes"
-
 	appsv1 "k8s.io/api/apps/v1"
 	appsv1beta1 "k8s.io/api/apps/v1beta1"
 	appsv1beta2 "k8s.io/api/apps/v1beta2"
@@ -36,8 +35,51 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	patch "k8s.io/apimachinery/pkg/util/strategicpatch"
+	clientgo "k8s.io/client-go/kubernetes"
 )
 
+// Labeller can give key/value labels to set on deployed resources.
+type Labeller interface {
+	Labels() map[string]string
+}
+
+type withLabels struct {
+	Deployer
+
+	labellers []Labeller
+}
+
+// WithLabels creates a deployer that sets labels on deployed resources.
+func WithLabels(d Deployer, labellers ...Labeller) Deployer {
+	return &withLabels{
+		Deployer:  d,
+		labellers: labellers,
+	}
+}
+
+func (w *withLabels) Deploy(ctx context.Context, out io.Writer, artifacts []build.Artifact) ([]Artifact, error) {
+	dRes, err := w.Deployer.Deploy(ctx, out, artifacts)
+
+	labelDeployResults(merge(w.labellers...), dRes)
+
+	return dRes, err
+}
+
+// merge merges the labels from multiple sources.
+func merge(sources ...Labeller) map[string]string {
+	merged := make(map[string]string)
+
+	for _, src := range sources {
+		if src != nil {
+			for k, v := range src.Labels() {
+				merged[k] = v
+			}
+		}
+	}
+
+	return merged
+}
+
 type objectType int
 
 // List of API Objects supported by the Skaffold Labeler
@@ -205,14 +247,14 @@ var objectMetas = map[objectType]objectMeta{
 const tries int = 3
 const sleeptime time.Duration = 300 * time.Millisecond
 
-//nolint
-func LabelDeployResults(labels map[string]string, results []deploy.Artifact) {
+func labelDeployResults(labels map[string]string, results []Artifact) {
 	// use the kubectl client to update all k8s objects with a skaffold watermark
 	client, err := kubernetes.Client()
 	if err != nil {
 		logrus.Warnf("error retrieving kubernetes client: %s", err.Error())
 		return
 	}
+
 	for _, res := range results {
 		err = nil
 		for i := 0; i < tries; i++ {
@@ -247,7 +289,7 @@ func retrieveNamespace(ns string, m metav1.ObjectMeta) string {
 }
 
 // TODO(nkubala): change this to use the client-go dynamic client or something equally clean
-func updateRuntimeObject(client clientgo.Interface, labels map[string]string, res deploy.Artifact) error {
+func updateRuntimeObject(client clientgo.Interface, labels map[string]string, res Artifact) error {
 	for k, v := range constants.Labels.DefaultLabels {
 		labels[k] = v
 	}

pkg/skaffold/runner/runner.go

@@ -29,7 +29,6 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
 	kubectx "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/context"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/label"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/v1alpha2"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/watch"
 	"github.com/pkg/errors"
@@ -45,35 +44,9 @@ type SkaffoldRunner struct {
 	watch.WatcherFactory
 	build.DependencyMapFactory
 
-	opts   *config.SkaffoldOptions
 	builds []build.Artifact
 }
 
-func (r *SkaffoldRunner) Labels() map[string]string {
-	labels := map[string]string{}
-	if r.opts != nil {
-		for k, v := range r.opts.Labels() {
-			labels[k] = v
-		}
-	}
-	if r.Builder != nil {
-		for k, v := range r.Builder.Labels() {
-			labels[k] = v
-		}
-	}
-	if r.Deployer != nil {
-		for k, v := range r.Deployer.Labels() {
-			labels[k] = v
-		}
-	}
-	if r.Tagger != nil {
-		for k, v := range r.Tagger.Labels() {
-			labels[k] = v
-		}
-	}
-	return labels
-}
-
 // NewForConfig returns a new SkaffoldRunner for a SkaffoldConfig
 func NewForConfig(opts *config.SkaffoldOptions, cfg *config.SkaffoldConfig) (*SkaffoldRunner, error) {
 	kubeContext, err := kubectx.CurrentContext()
@@ -82,6 +55,11 @@ func NewForConfig(opts *config.SkaffoldOptions, cfg *config.SkaffoldConfig) (*Sk
 	}
 	logrus.Infof("Using kubectl context: %s", kubeContext)
 
+	tagger, err := getTagger(cfg.Build.TagPolicy, opts.CustomTag)
+	if err != nil {
+		return nil, errors.Wrap(err, "parsing skaffold tag config")
+	}
+
 	builder, err := getBuilder(&cfg.Build, kubeContext)
 	if err != nil {
 		return nil, errors.Wrap(err, "parsing skaffold build config")
@@ -92,23 +70,18 @@ func NewForConfig(opts *config.SkaffoldOptions, cfg *config.SkaffoldConfig) (*Sk
 		return nil, errors.Wrap(err, "parsing skaffold deploy config")
 	}
 
+	deployer = deploy.WithLabels(deployer, opts, builder, deployer, tagger)
 	builder, deployer = WithTimings(builder, deployer)
 	if opts.Notification {
 		deployer = WithNotification(deployer)
 	}
 
-	tagger, err := getTagger(cfg.Build.TagPolicy, opts.CustomTag)
-	if err != nil {
-		return nil, errors.Wrap(err, "parsing skaffold tag config")
-	}
-
 	return &SkaffoldRunner{
 		Builder:              builder,
 		Deployer:             deployer,
 		Tagger:               tagger,
 		WatcherFactory:       watch.NewWatcher,
 		DependencyMapFactory: build.NewDependencyMap,
-		opts:                 opts,
 	}, nil
 }
 
@@ -183,11 +156,10 @@ func (r *SkaffoldRunner) Run(ctx context.Context, out io.Writer, artifacts []*v1
 		return errors.Wrap(err, "build step")
 	}
 
-	dRes, err := r.Deploy(ctx, out, bRes)
+	_, err = r.Deploy(ctx, out, bRes)
 	if err != nil {
 		return errors.Wrap(err, "deploy step")
 	}
-	label.LabelDeployResults(r.Labels(), dRes)
 
 	return nil
 }
@@ -220,45 +192,26 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*v1
 	colorPicker := kubernetes.NewColorPicker(artifacts)
 	logger := kubernetes.NewLogAggregator(out, imageList, colorPicker)
 
-	onChange := func(changedPaths []string) error {
+	onDeployChange := func(changedPaths []string) error {
 		logger.Mute()
-		defer logger.Unmute()
 
-		changedArtifacts := depMap.ArtifactsForPaths(changedPaths)
-
-		bRes, err := r.Builder.Build(ctx, out, r.Tagger, changedArtifacts)
-		if err != nil {
-			if r.builds == nil {
-				return errors.Wrap(err, "exiting dev mode because the first build failed")
-			}
+		_, err := r.Deploy(ctx, out, r.builds)
 
-			logrus.Warnln("Skipping Deploy due to build error:", err)
-			return nil
-		}
-
-		// Update which images are logged.
-		for _, build := range bRes {
-			imageList.Add(build.Tag)
-		}
-
-		// Make sure all artifacts are redeployed. Not only those that were just rebuilt.
-		r.builds = mergeWithPreviousBuilds(bRes, r.builds)
-
-		dRes, err := r.Deploy(ctx, out, r.builds)
-		label.LabelDeployResults(r.Labels(), dRes)
+		logger.Unmute()
 		return err
 	}
 
-	onDeployChange := func(changedPaths []string) error {
+	onArtifactChange := func(changedPaths []string) error {
 		logger.Mute()
-		defer logger.Unmute()
 
-		dRes, err := r.Deploy(ctx, out, r.builds)
-		label.LabelDeployResults(r.Labels(), dRes)
+		changedArtifacts := depMap.ArtifactsForPaths(changedPaths)
+		_, err = r.buildAndDeploy(ctx, out, changedArtifacts, imageList)
+
+		logger.Unmute()
 		return err
 	}
 
-	if err := onChange(depMap.Paths()); err != nil {
+	if err := onArtifactChange(depMap.Paths()); err != nil {
 		return nil, errors.Wrap(err, "first build")
 	}
 
@@ -270,7 +223,7 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*v1
 	// Watch files and rebuild
 	g, watchCtx := errgroup.WithContext(ctx)
 	g.Go(func() error {
-		return watcher.Start(watchCtx, out, onChange)
+		return watcher.Start(watchCtx, out, onArtifactChange)
 	})
 	g.Go(func() error {
 		return deployWatcher.Start(watchCtx, ioutil.Discard, onDeployChange)
@@ -279,6 +232,29 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*v1
 	return r.builds, g.Wait()
 }
 
+// buildAndDeploy builds a subset of the artifacts and deploys everything.
+func (r *SkaffoldRunner) buildAndDeploy(ctx context.Context, out io.Writer, artifacts []*v1alpha2.Artifact, images *kubernetes.ImageList) ([]deploy.Artifact, error) {
+	bRes, err := r.Builder.Build(ctx, out, r.Tagger, artifacts)
+	if err != nil {
+		if r.builds == nil {
+			return nil, errors.Wrap(err, "exiting dev mode because the first build failed")
+		}
+
+		logrus.Warnln("Skipping Deploy due to build error:", err)
+		return nil, nil
+	}
+
+	// Update which images are logged.
+	for _, build := range bRes {
+		images.Add(build.Tag)
+	}
+
+	// Make sure all artifacts are redeployed. Not only those that were just rebuilt.
+	r.builds = mergeWithPreviousBuilds(bRes, r.builds)
+
+	return r.Deploy(ctx, out, r.builds)
+}
+
 func mergeWithPreviousBuilds(builds, previous []build.Artifact) []build.Artifact {
 	updatedBuilds := map[string]bool{}
 	for _, build := range builds {