enhance(k8s): use the buildCtx/logCtx to create informerCtx

cmd/vela-worker/exec.go

@@ -129,7 +129,7 @@ func (w *Worker) exec(index int) error {
 		return nil
 	}
 
-	// log streaming uses buildCtx so that it is not subject to the timeout.
+	// log/event streaming uses buildCtx so that it is not subject to the timeout.
 	go func() {
 		logger.Info("streaming build logs")
 		// execute the build with the executor

executor/linux/build.go

@@ -510,6 +510,18 @@ func (c *client) StreamBuild(ctx context.Context) error {
 		c.Logger.Info("all stream functions have returned")
 	}()
 
+	// allow the runtime to do log/event streaming setup at build-level
+	streams.Go(func() error {
+		// If needed, the runtime should handle synchronizing with
+		// AssembleBuild which runs concurrently with StreamBuild.
+		err := c.Runtime.StreamBuild(streamCtx, c.pipeline)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+
 	for {
 		select {
 		case req := <-c.streamRequests:

executor/local/build.go

@@ -363,6 +363,18 @@ func (c *client) StreamBuild(ctx context.Context) error {
 		fmt.Fprintln(os.Stdout, "all stream functions have returned")
 	}()
 
+	// allow the runtime to do log/event streaming setup at build-level
+	streams.Go(func() error {
+		// If needed, the runtime should handle synchronizing with
+		// AssembleBuild which runs concurrently with StreamBuild.
+		err := c.Runtime.StreamBuild(streamCtx, c.pipeline)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+
 	for {
 		select {
 		case req := <-c.streamRequests:

runtime/docker/build.go

@@ -26,6 +26,14 @@ func (c *client) SetupBuild(ctx context.Context, b *pipeline.Build) error {
 	return nil
 }
 
+// StreamBuild initializes log/event streaming for build.
+// This is a no-op for docker.
+func (c *client) StreamBuild(ctx context.Context, b *pipeline.Build) error {
+	c.Logger.Tracef("no-op: streaming build %s", b.ID)
+
+	return nil
+}
+
 // AssembleBuild finalizes pipeline build setup.
 // This is a no-op for docker.
 func (c *client) AssembleBuild(ctx context.Context, b *pipeline.Build) error {

runtime/engine.go

@@ -29,6 +29,10 @@ type Engine interface {
 	// SetupBuild defines a function that
 	// prepares the pipeline build.
 	SetupBuild(context.Context, *pipeline.Build) error
+	// StreamBuild defines a function that initializes
+	// log/event streaming if the runtime needs it.
+	// StreamBuild and AssembleBuild run concurrently.
+	StreamBuild(context.Context, *pipeline.Build) error
 	// AssembleBuild defines a function that
 	// finalizes pipeline build setup.
 	AssembleBuild(context.Context, *pipeline.Build) error

runtime/kubernetes/build.go

@@ -138,6 +138,25 @@ func (c *client) SetupBuild(ctx context.Context, b *pipeline.Build) error {
 	return nil
 }
 
+// StreamBuild initializes log/event streaming for build.
+func (c *client) StreamBuild(ctx context.Context, b *pipeline.Build) error {
+	c.Logger.Tracef("streaming build %s", b.ID)
+
+	select {
+	case <-ctx.Done():
+		// bail out, as build timed out or was canceled.
+		return nil
+	case <-c.PodTracker.Ready:
+		// AssembleBuild signaled that the PodTracker is ready.
+		break
+	}
+
+	// Populate the PodTracker caches before creating the pipeline pod
+	c.PodTracker.Start(ctx)
+
+	return nil
+}
+
 // AssembleBuild finalizes the pipeline build setup.
 // This creates the pod in kubernetes for the pipeline build.
 // After creation, image is the only container field we can edit in kubernetes.
@@ -196,9 +215,10 @@ func (c *client) AssembleBuild(ctx context.Context, b *pipeline.Build) error {
 	// setup containerTeachers now that all containers are defined.
 	c.PodTracker.TrackContainers(c.Pod.Spec.Containers)
 
-	// Populate the PodTracker caches before creating the pipeline pod
-	c.PodTracker.Start(ctx)
+	// send signal to StreamBuild now that PodTracker is ready to be started.
+	close(c.PodTracker.Ready)
 
+	// wait for the PodTracker caches to populate before creating the pipeline pod.
 	if ok := cache.WaitForCacheSync(ctx.Done(), c.PodTracker.PodSynced); !ok {
 		return fmt.Errorf("failed to wait for caches to sync")
 	}

runtime/kubernetes/build_test.go

@@ -432,6 +432,15 @@ func TestKubernetes_AssembleBuild(t *testing.T) {
 				t.Errorf("unable to create runtime engine: %v", err)
 			}
 
+			// StreamBuild and AssembleBuild coordinate their work, so, emulate
+			// executor.StreamBuild which calls runtime.StreamBuild concurrently.
+			go func() {
+				err := _engine.StreamBuild(context.Background(), test.pipeline)
+				if err != nil {
+					t.Errorf("unable to start PodTracker via StreamBuild")
+				}
+			}()
+
 			err = _engine.AssembleBuild(context.Background(), test.pipeline)
 
 			if test.failure {

runtime/kubernetes/pod_tracker.go

@@ -58,6 +58,9 @@ type podTracker struct {
 
 	// Containers maps the container name to a containerTracker
 	Containers map[string]*containerTracker
+
+	// Ready signals when the PodTracker is done with setup and ready to Start.
+	Ready chan struct{}
 }
 
 // HandlePodAdd is an AddFunc for cache.ResourceEventHandlerFuncs for Pods.
@@ -219,6 +222,7 @@ func newPodTracker(log *logrus.Entry, clientset kubernetes.Interface, pod *v1.Po
 		podInformer:     podInformer,
 		PodLister:       podInformer.Lister(),
 		PodSynced:       podInformer.Informer().HasSynced,
+		Ready:           make(chan struct{}),
 	}
 
 	// register event handler funcs in podInformer