Add BuildSubtask emission for v2 API (GoogleContainerTools#5710)

* add BuildSubtask emission

* add func to get iteration from handler

pkg/skaffold/build/scheduler.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/color"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/event"
+	eventV2 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/event/v2"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/tag"
@@ -88,23 +89,27 @@ func (s *scheduler) build(ctx context.Context, tags tag.ImageTags, i int) error
 	defer release()
 
 	event.BuildInProgress(a.ImageName)
+	eventV2.BuildInProgress(i, a.ImageName)
 
 	w, closeFn, err := s.logger.GetWriter()
 	if err != nil {
 		event.BuildFailed(a.ImageName, err)
+		eventV2.BuildFailed(i, a.ImageName, err)
 		return err
 	}
 	defer closeFn()
 
 	finalTag, err := performBuild(ctx, w, tags, a, s.artifactBuilder)
 	if err != nil {
 		event.BuildFailed(a.ImageName, err)
+		eventV2.BuildFailed(i, a.ImageName, err)
 		return err
 	}
 
 	s.results.Record(a, finalTag)
 	n.markComplete()
 	event.BuildComplete(a.ImageName)
+	eventV2.BuildSucceeded(i, a.ImageName)
 	return nil
 }
 

pkg/skaffold/event/v2/event.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"strconv"
 	"sync"
 
 	//nolint:golint,staticcheck
@@ -69,6 +70,7 @@ type eventHandler struct {
 	logLock  sync.Mutex
 	cfg      event.Config
 
+	iteration int
 	state     proto.State
 	stateLock sync.Mutex
 	eventChan chan *proto.Event
@@ -81,6 +83,10 @@ type listener struct {
 	closed   bool
 }
 
+func GetIteration() int {
+	return handler.iteration
+}
+
 func GetState() (*proto.State, error) {
 	state := handler.getState()
 	return &state, nil
@@ -243,6 +249,12 @@ func emptyStatusCheckState() *proto.StatusCheckState {
 	}
 }
 
+// InitializeState instantiates the global state of the skaffold runner, as well as the event log.
+func InitializeState(cfg event.Config) {
+	handler.cfg = cfg
+	handler.setState(emptyState(cfg))
+}
+
 func AutoTriggerDiff(phase constants.Phase, val bool) (bool, error) {
 	switch phase {
 	case constants.Build:
@@ -257,6 +269,10 @@ func AutoTriggerDiff(phase constants.Phase, val bool) (bool, error) {
 }
 
 func TaskInProgress(task constants.Phase, iteration int) {
+	if task == constants.DevLoop {
+		handler.iteration = iteration
+	}
+
 	handler.handleTaskEvent(&proto.TaskEvent{
 		Id:        fmt.Sprintf("%s-%d", task, iteration),
 		Task:      string(task),
@@ -285,6 +301,34 @@ func TaskSucceeded(task constants.Phase, iteration int) {
 	})
 }
 
+func BuildInProgress(id int, artifact string) {
+	handler.handleBuildSubtaskEvent(&proto.BuildSubtaskEvent{
+		Id:       strconv.Itoa(id),
+		TaskId:   fmt.Sprintf("%s-%d", constants.Build, handler.iteration),
+		Artifact: artifact,
+		Status:   InProgress,
+	})
+}
+
+func BuildFailed(id int, artifact string, err error) {
+	handler.handleBuildSubtaskEvent(&proto.BuildSubtaskEvent{
+		Id:            strconv.Itoa(id),
+		TaskId:        fmt.Sprintf("%s-%d", constants.Build, handler.iteration),
+		Artifact:      artifact,
+		Status:        Failed,
+		ActionableErr: sErrors.ActionableErrV2(handler.cfg, constants.Build, err),
+	})
+}
+
+func BuildSucceeded(id int, artifact string) {
+	handler.handleBuildSubtaskEvent(&proto.BuildSubtaskEvent{
+		Id:       strconv.Itoa(id),
+		TaskId:   fmt.Sprintf("%s-%d", constants.Build, handler.iteration),
+		Artifact: artifact,
+		Status:   Succeeded,
+	})
+}
+
 func (ev *eventHandler) setState(state proto.State) {
 	ev.stateLock.Lock()
 	ev.state = state
@@ -311,6 +355,14 @@ func (ev *eventHandler) handleTaskEvent(e *proto.TaskEvent) {
 	})
 }
 
+func (ev *eventHandler) handleBuildSubtaskEvent(e *proto.BuildSubtaskEvent) {
+	ev.handle(&proto.Event{
+		EventType: &proto.Event_BuildSubtaskEvent{
+			BuildSubtaskEvent: e,
+		},
+	})
+}
+
 func (ev *eventHandler) handleExec(event *proto.Event) {
 	switch e := event.GetEventType().(type) {
 	case *proto.Event_BuildSubtaskEvent:

pkg/skaffold/runner/new.go

@@ -34,6 +34,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/label"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/status"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/event"
+	eventV2 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/event/v2"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/filemon"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	pkgkubectl "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
@@ -52,6 +53,7 @@ import (
 func NewForConfig(runCtx *runcontext.RunContext) (*SkaffoldRunner, error) {
 	event.InitializeState(runCtx)
 	event.LogMetaEvent()
+	eventV2.InitializeState(runCtx)
 	kubectlCLI := pkgkubectl.NewCLI(runCtx, "")
 
 	tagger, err := tag.NewTaggerMux(runCtx)

testutil/event/config.go

@@ -18,6 +18,7 @@ package event
 
 import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/event"
+	eventV2 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/event/v2"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest"
 )
 
@@ -26,6 +27,7 @@ func InitializeState(pipes []latest.Pipeline) {
 		pipes: pipes,
 	}
 	event.InitializeState(cfg)
+	eventV2.InitializeState(cfg)
 }
 
 type config struct {