Send success for pods for a deployment when its rolled out successful…

…ly (#6534)

* Send success for pods if deployment is successfule

* fix test

* fix lint

pkg/skaffold/kubernetes/status/resource/deployment.go

@@ -30,6 +30,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/output/log"
 	"github.com/GoogleContainerTools/skaffold/proto/v1"
+	protoV2 "github.com/GoogleContainerTools/skaffold/proto/v2"
 )
 
 const (
@@ -132,12 +133,24 @@ func (d *Deployment) CheckStatus(ctx context.Context, cfg kubectl.Config) {
 
 	details := d.cleanupStatus(string(b))
 
-	ae := parseKubectlRolloutError(details, err)
-	if ae.ErrCode == proto.StatusCode_STATUSCHECK_KUBECTL_PID_KILLED {
-		ae.Message = fmt.Sprintf("received Ctrl-C or deployments could not stabilize within %v: %v", d.deadline, err)
-	}
-
+	ae := parseKubectlRolloutError(details, d.deadline, err)
 	d.UpdateStatus(ae)
+	// send event update in check status.
+	event.ResourceStatusCheckEventCompleted(d.String(), ae)
+	eventV2.ResourceStatusCheckEventCompleted(d.String(), sErrors.V2fromV1(ae))
+	// if deployment is successfully rolled out, send pod success event to make sure
+	// all pod are marked as success in V2
+	// See https://github.com/GoogleCloudPlatform/cloud-code-vscode-internal/issues/5277
+	if ae.ErrCode == proto.StatusCode_STATUSCHECK_SUCCESS {
+		for _, pod := range d.pods {
+			eventV2.ResourceStatusCheckEventCompletedMessage(
+				pod.String(),
+				fmt.Sprintf("%s %s: running.\n", tabHeader, pod.String()),
+				protoV2.ActionableErr{ErrCode: proto.StatusCode_STATUSCHECK_SUCCESS},
+			)
+		}
+		return
+	}
 	if err := d.fetchPods(ctx); err != nil {
 		log.Entry(ctx).Debugf("pod statuses could not be fetched this time due to %s", err)
 	}
@@ -233,7 +246,7 @@ func (d *Deployment) cleanupStatus(msg string) string {
 // $kubectl logs testPod  -f
 // 2020/06/18 17:28:31 service is running
 // Killed: 9
-func parseKubectlRolloutError(details string, err error) proto.ActionableErr {
+func parseKubectlRolloutError(details string, deadline time.Duration, err error) proto.ActionableErr {
 	switch {
 	case err == nil && strings.Contains(details, rollOutSuccess):
 		return proto.ActionableErr{
@@ -253,7 +266,7 @@ func parseKubectlRolloutError(details string, err error) proto.ActionableErr {
 	case strings.Contains(err.Error(), killedErrMsg):
 		return proto.ActionableErr{
 			ErrCode: proto.StatusCode_STATUSCHECK_KUBECTL_PID_KILLED,
-			Message: msgKubectlKilled,
+			Message: fmt.Sprintf("received Ctrl-C or deployments could not stabilize within %v: %s", deadline, msgKubectlKilled),
 		}
 	default:
 		return proto.ActionableErr{

pkg/skaffold/kubernetes/status/resource/deployment_test.go

@@ -23,12 +23,15 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/diag/validator"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
+	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util"
 	"github.com/GoogleContainerTools/skaffold/proto/v1"
 	"github.com/GoogleContainerTools/skaffold/testutil"
+	testEvent "github.com/GoogleContainerTools/skaffold/testutil/event"
 )
 
 func TestDeploymentCheckStatus(t *testing.T) {
@@ -100,6 +103,7 @@ func TestDeploymentCheckStatus(t *testing.T) {
 	for _, test := range tests {
 		testutil.Run(t, test.description, func(t *testutil.T) {
 			t.Override(&util.DefaultExecCommand, test.commands)
+			testEvent.InitializeState([]latestV1.Pipeline{{}})
 
 			r := NewDeployment("graph", "test", 0)
 			r.CheckStatus(context.Background(), &statusConfig{})
@@ -140,7 +144,7 @@ func TestParseKubectlError(t *testing.T) {
 			err:         errors.New("signal: killed"),
 			expectedAe: proto.ActionableErr{
 				ErrCode: proto.StatusCode_STATUSCHECK_KUBECTL_PID_KILLED,
-				Message: msgKubectlKilled,
+				Message: "received Ctrl-C or deployments could not stabilize within 10s: kubectl rollout status command interrupted\n",
 			},
 		},
 		{
@@ -162,7 +166,7 @@ func TestParseKubectlError(t *testing.T) {
 	}
 	for _, test := range tests {
 		testutil.Run(t, test.description, func(t *testutil.T) {
-			ae := parseKubectlRolloutError(test.details, test.err)
+			ae := parseKubectlRolloutError(test.details, 10*time.Second, test.err)
 			t.CheckDeepEqual(test.expectedAe, ae)
 		})
 	}

pkg/skaffold/kubernetes/status/status_check.go

@@ -288,8 +288,6 @@ func (s *Monitor) printStatusCheckSummary(out io.Writer, r *resource.Deployment,
 		// another deployment failed
 		return
 	}
-	event.ResourceStatusCheckEventCompleted(r.String(), ae)
-	eventV2.ResourceStatusCheckEventCompleted(r.String(), sErrors.V2fromV1(ae))
 	out, _ = output.WithEventContext(context.Background(), out, constants.Deploy, r.String())
 	status := fmt.Sprintf("%s %s", tabHeader, r)
 	if ae.ErrCode != proto.StatusCode_STATUSCHECK_SUCCESS {

pkg/skaffold/kubernetes/status/status_check_test.go

@@ -596,14 +596,6 @@ func TestPollDeployment(t *testing.T) {
 					"Pending",
 					proto.ActionableErr{ErrCode: proto.StatusCode_STATUSCHECK_NODE_DISK_PRESSURE},
 					[]string{"err"})},
-				// pod recovered
-				{validator.NewResource(
-					"test",
-					"pod",
-					"dep-pod",
-					"Running",
-					proto.ActionableErr{ErrCode: proto.StatusCode_STATUSCHECK_SUCCESS},
-					nil)},
 			},
 			expected: proto.StatusCode_STATUSCHECK_SUCCESS,
 		},