Add Resource.Status object and remove sync.Map

pkg/skaffold/deploy/resource/deployment.go

@@ -30,6 +30,7 @@ type Deployment struct {
 	namespace string
 	rType     string
 	deadline  time.Duration
+	status    *Status
 }
 
 func (d *Deployment) String() string {
@@ -44,11 +45,27 @@ func (d *Deployment) Deadline() time.Duration {
 	return d.deadline
 }
 
+func (d *Deployment) Status() *Status {
+	return d.status
+}
+
+func (d *Deployment) UpdateStatus(details string, err error) {
+	d.status.err = err
+	d.status.details = details
+}
+
 func NewDeployment(name string, ns string, deadline time.Duration) *Deployment {
 	return &Deployment{
 		name:      name,
 		namespace: ns,
 		rType:     deploymentType,
 		deadline:  deadline,
+		status:    NewStatus("", nil),
 	}
 }
+
+// For testing
+func (d *Deployment) WithStatus(details string, err error) *Deployment {
+	d.UpdateStatus(details, err)
+	return d
+}

pkg/skaffold/deploy/resource/status.go

@@ -0,0 +1,40 @@
+/*
+Copyright 2019 The Skaffold Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resource
+
+type Status struct {
+	err     error
+	details string
+}
+
+func (rs *Status) Error() error {
+	return rs.err
+}
+
+func (rs *Status) String() string {
+	if rs.err != nil {
+		return rs.err.Error()
+	}
+	return rs.details
+}
+
+func NewStatus(msg string, err error) *Status {
+	return &Status{
+		details: msg,
+		err:     err,
+	}
+}

pkg/skaffold/deploy/status_check.go

@@ -70,25 +70,22 @@ func StatusCheck(ctx context.Context, defaultLabeller *DefaultLabeller, runCtx *
 	}
 
 	wg := sync.WaitGroup{}
-	// Its safe to use sync.Map without locks here as each subroutine adds a different key to the map.
-	syncMap := &sync.Map{}
 
 	c := newCounter(len(deployments))
 
 	for _, d := range deployments {
 		wg.Add(1)
 		go func(d *resource.Deployment) {
 			defer wg.Done()
-			err := pollDeploymentRolloutStatus(ctx, kubectl.NewFromRunContext(runCtx), d)
-			syncMap.Store(d.String(), err)
+			pollDeploymentRolloutStatus(ctx, kubectl.NewFromRunContext(runCtx), d)
 			pending := c.markProcessed()
 			printStatusCheckSummary(d, pending, c.total, err, out)
 		}(d)
 	}
 
 	// Wait for all deployment status to be fetched
 	wg.Wait()
-	return getSkaffoldDeployStatus(syncMap)
+	return getSkaffoldDeployStatus(deployments)
 }
 
 func getDeployments(client kubernetes.Interface, ns string, l *DefaultLabeller, deadlineDuration time.Duration) ([]*resource.Deployment, error) {
@@ -113,7 +110,7 @@ func getDeployments(client kubernetes.Interface, ns string, l *DefaultLabeller,
 	return deployments, nil
 }
 
-func pollDeploymentRolloutStatus(ctx context.Context, k *kubectl.CLI, d *resource.Deployment) error {
+func pollDeploymentRolloutStatus(ctx context.Context, k *kubectl.CLI, d *resource.Deployment) {
 	pollDuration := time.Duration(defaultPollPeriodInMilliseconds) * time.Millisecond
 	// Add poll duration to account for one last attempt after progressDeadlineSeconds.
 	timeoutContext, cancel := context.WithTimeout(ctx, d.Deadline()+pollDuration)
@@ -123,25 +120,25 @@ func pollDeploymentRolloutStatus(ctx context.Context, k *kubectl.CLI, d *resourc
 		select {
 		case <-timeoutContext.Done():
 			err := errors.Wrap(timeoutContext.Err(), fmt.Sprintf("deployment rollout status could not be fetched within %v", d.Deadline()))
-			return err
+			d.UpdateStatus(err.Error(), err)
+			return
 		case <-time.After(pollDuration):
 			status, err := executeRolloutStatus(timeoutContext, k, d.Name())
+			d.UpdateStatus(status, err)
 			if err != nil || strings.Contains(status, "successfully rolled out") {
-				return err
+				return
 			}
 		}
 	}
 }
 
-func getSkaffoldDeployStatus(m *sync.Map) error {
+func getSkaffoldDeployStatus(deployments []*resource.Deployment) error {
 	var errorStrings []string
-	m.Range(func(k, v interface{}) bool {
-		if t, ok := v.(error); ok {
-			errorStrings = append(errorStrings, fmt.Sprintf("deployment %s failed due to %s", k, t.Error()))
+	for _, d := range deployments {
+		if err := d.Status().Error(); err != nil {
+			errorStrings = append(errorStrings, fmt.Sprintf("deployment %s failed due to %s", d, err.Error()))
 		}
-		return true
-	})
-
+	}
 	if len(errorStrings) == 0 {
 		return nil
 	}

pkg/skaffold/deploy/status_check_test.go

@@ -20,8 +20,6 @@ import (
 	"bytes"
 	"context"
 	"errors"
-	"fmt"
-	"sync"
 	"testing"
 	"time"
 
@@ -198,7 +196,7 @@ func TestGetDeployments(t *testing.T) {
 			client := fakekubeclientset.NewSimpleClientset(objs...)
 			actual, err := getDeployments(client, "test", labeller, time.Duration(200)*time.Second)
 			t.CheckErrorAndDeepEqual(test.shouldErr, err, &test.expected, &actual,
-				cmp.AllowUnexported(resource.Deployment{}))
+				cmp.AllowUnexported(resource.Deployment{}, resource.Status{}))
 		})
 	}
 }
@@ -243,55 +241,58 @@ func TestPollDeploymentRolloutStatus(t *testing.T) {
 
 			cli := &kubectl.CLI{KubeContext: testKubeContext, Namespace: "test"}
 			d := resource.NewDeployment("dep", "test", time.Duration(test.duration)*time.Millisecond)
-			err := pollDeploymentRolloutStatus(context.Background(), cli, d)
-			t.CheckError(test.shouldErr, err)
+			pollDeploymentRolloutStatus(context.Background(), cli, d)
+			t.CheckError(test.shouldErr, d.Status().Error())
 		})
 	}
 }
 
 func TestGetDeployStatus(t *testing.T) {
 	tests := []struct {
 		description    string
-		deps           map[string]interface{}
+		deps           []*resource.Deployment
 		expectedErrMsg []string
 		shouldErr      bool
 	}{
 		{
 			description: "one error",
-			deps: map[string]interface{}{
-				"dep1": "SUCCESS",
-				"dep2": fmt.Errorf("could not return within default timeout"),
+			deps: []*resource.Deployment{
+				resource.NewDeployment("dep1", "test", time.Second).
+					WithStatus("success", nil),
+				resource.NewDeployment("dep2", "test", time.Second).
+					WithStatus("error", errors.New("could not return within default timeout")),
 			},
-			expectedErrMsg: []string{"deployment dep2 failed due to could not return within default timeout"},
+			expectedErrMsg: []string{"dep2 failed due to could not return within default timeout"},
 			shouldErr:      true,
 		},
 		{
 			description: "no error",
-			deps: map[string]interface{}{
-				"dep1": "SUCCESS",
-				"dep2": "RUNNING",
+			deps: []*resource.Deployment{
+				resource.NewDeployment("dep1", "test", time.Second).
+					WithStatus("success", nil),
+				resource.NewDeployment("dep2", "test", time.Second).
+					WithStatus("running", nil),
 			},
 		},
 		{
 			description: "multiple errors",
-			deps: map[string]interface{}{
-				"dep1": "SUCCESS",
-				"dep2": fmt.Errorf("could not return within default timeout"),
-				"dep3": fmt.Errorf("ERROR"),
+			deps: []*resource.Deployment{
+				resource.NewDeployment("dep1", "test", time.Second).
+					WithStatus("success", nil),
+				resource.NewDeployment("dep2", "test", time.Second).
+					WithStatus("error", errors.New("could not return within default timeout")),
+				resource.NewDeployment("dep3", "test", time.Second).
+					WithStatus("error", errors.New("ERROR")),
 			},
-			expectedErrMsg: []string{"deployment dep2 failed due to could not return within default timeout",
-				"deployment dep3 failed due to ERROR"},
+			expectedErrMsg: []string{"dep2 failed due to could not return within default timeout",
+				"dep3 failed due to ERROR"},
 			shouldErr: true,
 		},
 	}
 
 	for _, test := range tests {
 		testutil.Run(t, test.description, func(t *testutil.T) {
-			syncMap := &sync.Map{}
-			for k, v := range test.deps {
-				syncMap.Store(k, v)
-			}
-			err := getSkaffoldDeployStatus(syncMap)
+			err := getSkaffoldDeployStatus(test.deps)
 			t.CheckError(test.shouldErr, err)
 			for _, msg := range test.expectedErrMsg {
 				t.CheckErrorContains(msg, err)