Revert "Revert "Merge pull request GoogleContainerTools#2896 from tej…

…al29/move_poll_deployment""

This reverts commit cf878dc.

pkg/skaffold/deploy/resource.go

@@ -0,0 +1,39 @@
+/*
+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 deploy
+
+import (
+	"context"
+	"time"
+
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
+)
+
+type Resource interface {
+
+	// UpdateStatus updates the resource status
+	UpdateStatus(string, error)
+
+	// IsStatusCheckComplete returns if the resource status check is complele
+	IsStatusCheckComplete() bool
+
+	// Deadline returns the deadline for the resource
+	Deadline() time.Duration
+
+	// CheckStatus checks resource status
+	CheckStatus(context.Context, *runcontext.RunContext)
+}

pkg/skaffold/deploy/resource/deployment.go

@@ -17,12 +17,26 @@ limitations under the License.
 package resource
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"strings"
 	"time"
+
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
 )
 
 const (
-	deploymentType = "deployment"
+	deploymentType   = "deployment"
+	rollOutSuccess   = "successfully rolled out"
+	connectionErrMsg = "Unable to connect to the server"
+	killedErrMsg     = "signal: killed"
+)
+
+var (
+	errKubectlKilled     = errors.New("kubectl rollout status command killed")
+	ErrKubectlConnection = errors.New("kubectl connection error")
 )
 
 type Deployment struct {
@@ -54,17 +68,16 @@ func (d *Deployment) UpdateStatus(details string, err error) {
 	updated := newStatus(details, err)
 	if !d.status.Equal(updated) {
 		d.status = updated
+		if strings.Contains(details, rollOutSuccess) || isErrAndNotRetryAble(err) {
+			d.done = true
+		}
 	}
 }
 
-func (d *Deployment) IsDone() bool {
+func (d *Deployment) IsStatusCheckComplete() bool {
 	return d.done
 }
 
-func (d *Deployment) MarkDone() {
-	d.done = true
-}
-
 func (d *Deployment) ReportSinceLastUpdated() string {
 	if d.status.reported {
 		return ""
@@ -73,6 +86,13 @@ func (d *Deployment) ReportSinceLastUpdated() string {
 	return fmt.Sprintf("%s %s", d, d.status)
 }
 
+func (d *Deployment) CheckStatus(ctx context.Context, runCtx *runcontext.RunContext) {
+	cli := kubectl.NewFromRunContext(runCtx)
+	b, err := cli.RunOut(ctx, "rollout", "status", "deployment", d.name, "--namespace", d.namespace, "--watch=false")
+	err = parseKubectlRolloutError(err)
+	d.UpdateStatus(string(b), err)
+}
+
 func NewDeployment(name string, ns string, deadline time.Duration) *Deployment {
 	return &Deployment{
 		name:      name,
@@ -82,3 +102,23 @@ func NewDeployment(name string, ns string, deadline time.Duration) *Deployment {
 		status:    newStatus("", nil),
 	}
 }
+
+func parseKubectlRolloutError(err error) error {
+	if err == nil {
+		return err
+	}
+	if strings.Contains(err.Error(), connectionErrMsg) {
+		return ErrKubectlConnection
+	}
+	if strings.Contains(err.Error(), killedErrMsg) {
+		return errKubectlKilled
+	}
+	return err
+}
+
+func isErrAndNotRetryAble(err error) bool {
+	if err == nil {
+		return false
+	}
+	return err != ErrKubectlConnection
+}

pkg/skaffold/deploy/resource/deployment_test.go

@@ -17,8 +17,11 @@ limitations under the License.
 package resource
 
 import (
+	"context"
 	"testing"
 
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 	"github.com/GoogleContainerTools/skaffold/testutil"
 	"github.com/pkg/errors"
 )
@@ -82,3 +85,158 @@ func TestReportSinceLastUpdatedMultipleTimes(t *testing.T) {
 		})
 	}
 }
+
+func TestDeploymentCheckStatus(t *testing.T) {
+	rolloutCmd := "kubectl --context kubecontext rollout status deployment dep --namespace test --watch=false"
+	tests := []struct {
+		description     string
+		commands        util.Command
+		expectedErr     string
+		expectedDetails string
+		complete        bool
+	}{
+		{
+			description: "rollout status success",
+			commands: testutil.CmdRunOut(
+				rolloutCmd,
+				"deployment dep successfully rolled out",
+			),
+			expectedDetails: "deployment dep successfully rolled out",
+			complete:        true,
+		},
+		{
+			description: "resource not complete",
+			commands: testutil.CmdRunOut(
+				rolloutCmd,
+				"Waiting for replicas to be available",
+			),
+			expectedDetails: "Waiting for replicas to be available",
+		},
+		{
+			description: "no output",
+			commands: testutil.CmdRunOut(
+				rolloutCmd,
+				"",
+			),
+		},
+		{
+			description: "rollout status error",
+			commands: testutil.CmdRunOutErr(
+				rolloutCmd,
+				"",
+				errors.New("error"),
+			),
+			expectedErr: "error",
+			complete:    true,
+		},
+		{
+			description: "rollout kubectl client connection error",
+			commands: testutil.CmdRunOutErr(
+				rolloutCmd,
+				"",
+				errors.New("Unable to connect to the server"),
+			),
+			expectedErr: ErrKubectlConnection.Error(),
+		},
+	}
+
+	for _, test := range tests {
+		testutil.Run(t, test.description, func(t *testutil.T) {
+			t.Override(&util.DefaultExecCommand, test.commands)
+			r := Deployment{namespace: "test", name: "dep"}
+			runCtx := &runcontext.RunContext{
+				KubeContext: "kubecontext",
+			}
+
+			r.CheckStatus(context.Background(), runCtx)
+			t.CheckDeepEqual(test.complete, r.IsStatusCheckComplete())
+			if test.expectedErr != "" {
+				t.CheckErrorContains(test.expectedErr, r.Status().Error())
+			} else {
+				t.CheckDeepEqual(r.status.details, test.expectedDetails)
+			}
+		})
+	}
+}
+
+func TestParseKubectlError(t *testing.T) {
+	tests := []struct {
+		description string
+		err         error
+		expected    string
+		shouldErr   bool
+	}{
+		{
+			description: "rollout status connection error",
+			err:         errors.New("Unable to connect to the server"),
+			expected:    ErrKubectlConnection.Error(),
+			shouldErr:   true,
+		},
+		{
+			description: "rollout status kubectl command killed",
+			err:         errors.New("signal: killed"),
+			expected:    errKubectlKilled.Error(),
+			shouldErr:   true,
+		},
+		{
+			description: "rollout status random error",
+			err:         errors.New("deployment test not found"),
+			expected:    "deployment test not found",
+			shouldErr:   true,
+		},
+		{
+			description: "rollout status nil error",
+		},
+	}
+	for _, test := range tests {
+		testutil.Run(t, test.description, func(t *testutil.T) {
+			actual := parseKubectlRolloutError(test.err)
+			t.CheckError(test.shouldErr, actual)
+			if test.shouldErr {
+				t.CheckErrorContains(test.expected, actual)
+			}
+		})
+	}
+}
+
+func TestIsErrAndNotRetriable(t *testing.T) {
+	tests := []struct {
+		description string
+		err         error
+		expected    bool
+	}{
+		{
+			description: "rollout status connection error",
+			err:         ErrKubectlConnection,
+		},
+		{
+			description: "rollout status kubectl command killed",
+			err:         errKubectlKilled,
+			expected:    true,
+		},
+		{
+			description: "rollout status random error",
+			err:         errors.New("deployment test not found"),
+			expected:    true,
+		},
+		{
+			description: "rollout status parent context cancelled",
+			err:         context.Canceled,
+			expected:    true,
+		},
+		{
+			description: "rollout status parent conetct timed out",
+			err:         context.DeadlineExceeded,
+			expected:    true,
+		},
+		{
+			description: "rollout status nil error",
+		},
+	}
+	for _, test := range tests {
+		testutil.Run(t, test.description, func(t *testutil.T) {
+			actual := isErrAndNotRetryAble(test.err)
+			t.CheckDeepEqual(test.expected, actual)
+		})
+	}
+}

pkg/skaffold/deploy/status_check.go

@@ -32,7 +32,6 @@ import (
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/color"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/resource"
-	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
 	pkgkubernetes "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
 )
@@ -45,9 +44,6 @@ var (
 
 	// report resource status for pending resources 0.5 second.
 	reportStatusTime = 500 * time.Millisecond
-
-	// For testing
-	executeRolloutStatus = getRollOutStatus
 )
 
 const (
@@ -79,7 +75,7 @@ func StatusCheck(ctx context.Context, defaultLabeller *DefaultLabeller, runCtx *
 		wg.Add(1)
 		go func(d *resource.Deployment) {
 			defer wg.Done()
-			pollDeploymentRolloutStatus(ctx, kubectl.NewFromRunContext(runCtx), d)
+			pollResourceStatus(ctx, runCtx, d)
 			pending := c.markProcessed()
 			printStatusCheckSummary(out, d, pending, c.total)
 		}(d)
@@ -117,24 +113,20 @@ func getDeployments(client kubernetes.Interface, ns string, l *DefaultLabeller,
 	return deployments, nil
 }
 
-func pollDeploymentRolloutStatus(ctx context.Context, k *kubectl.CLI, d *resource.Deployment) {
+func pollResourceStatus(ctx context.Context, runCtx *runcontext.RunContext, r Resource) {
 	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)
-	logrus.Debugf("checking rollout status %s", d.String())
+	timeoutContext, cancel := context.WithTimeout(ctx, r.Deadline()+pollDuration)
+	logrus.Debugf("checking status %s", r)
 	defer cancel()
 	for {
 		select {
 		case <-timeoutContext.Done():
-			err := errors.Wrap(timeoutContext.Err(), fmt.Sprintf("deployment rollout status could not be fetched within %v", d.Deadline()))
-			d.UpdateStatus(err.Error(), err)
-			d.MarkDone()
+			r.UpdateStatus(timeoutContext.Err().Error(), timeoutContext.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") {
-				d.MarkDone()
+			r.CheckStatus(timeoutContext, runCtx)
+			if r.IsStatusCheckComplete() {
 				return
 			}
 		}
@@ -154,11 +146,6 @@ func getSkaffoldDeployStatus(deployments []*resource.Deployment) error {
 	return fmt.Errorf("following deployments are not stable:\n%s", strings.Join(errorStrings, "\n"))
 }
 
-func getRollOutStatus(ctx context.Context, k *kubectl.CLI, dName string) (string, error) {
-	b, err := k.RunOut(ctx, "rollout", "status", "deployment", dName, "--watch=false")
-	return string(b), err
-}
-
 func getDeadline(d int) time.Duration {
 	if d > 0 {
 		return time.Duration(d) * time.Second
@@ -201,7 +188,7 @@ func printResourceStatus(ctx context.Context, out io.Writer, deps []*resource.De
 func printStatus(deps []*resource.Deployment, out io.Writer) bool {
 	allResourcesCheckComplete := true
 	for _, d := range deps {
-		if d.IsDone() {
+		if d.IsStatusCheckComplete() {
 			continue
 		}
 		allResourcesCheckComplete = false