Move cluster image loading logic into Deployer (#6124)

* Move cluster image loading logic into Deployer

* Load the correct set of images for each deployer

* Pass loader.Config as parameter to loader instead of provider

* add comments for tracked image vars

pkg/skaffold/deploy/deploy.go

@@ -23,18 +23,20 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/access"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/debug"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/loader"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/log"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/status"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/sync"
 )
 
 // NoopComponentProvider is for tests
 var NoopComponentProvider = ComponentProvider{
-	Accessor: &access.NoopProvider{},
-	Debugger: &debug.NoopProvider{},
-	Logger:   &log.NoopProvider{},
-	Monitor:  &status.NoopProvider{},
-	Syncer:   &sync.NoopProvider{},
+	Accessor:    &access.NoopProvider{},
+	Debugger:    &debug.NoopProvider{},
+	ImageLoader: &loader.NoopProvider{},
+	Logger:      &log.NoopProvider{},
+	Monitor:     &status.NoopProvider{},
+	Syncer:      &sync.NoopProvider{},
 }
 
 // Deployer is the Deploy API of skaffold and responsible for deploying
@@ -70,16 +72,20 @@ type Deployer interface {
 	// TrackBuildArtifacts registers build artifacts to be tracked by a Deployer
 	TrackBuildArtifacts([]graph.Artifact)
 
+	// RegisterLocalImages tracks all local images to be loaded by the Deployer
+	RegisterLocalImages([]graph.Artifact)
+
 	// GetStatusMonitor returns a Deployer's implementation of a StatusMonitor
 	GetStatusMonitor() status.Monitor
 }
 
 // ComponentProvider serves as a clean way to send three providers
 // as params to the Deployer constructors
 type ComponentProvider struct {
-	Accessor access.Provider
-	Debugger debug.Provider
-	Logger   log.Provider
-	Monitor  status.Provider
-	Syncer   sync.Provider
+	Accessor    access.Provider
+	Debugger    debug.Provider
+	ImageLoader loader.Provider
+	Logger      log.Provider
+	Monitor     status.Provider
+	Syncer      sync.Provider
 }

pkg/skaffold/deploy/deploy_mux.go

@@ -93,6 +93,12 @@ func (m DeployerMux) GetSyncer() sync.Syncer {
 	return syncers
 }
 
+func (m DeployerMux) RegisterLocalImages(images []graph.Artifact) {
+	for _, deployer := range m.deployers {
+		deployer.RegisterLocalImages(images)
+	}
+}
+
 func (m DeployerMux) Deploy(ctx context.Context, w io.Writer, as []graph.Artifact) ([]string, error) {
 	seenNamespaces := util.NewStringSet()
 

pkg/skaffold/deploy/deploy_mux_test.go

@@ -69,6 +69,8 @@ func (m *MockDeployer) GetSyncer() sync.Syncer {
 	return &sync.NoopSyncer{}
 }
 
+func (m *MockDeployer) RegisterLocalImages(_ []graph.Artifact) {}
+
 func (m *MockDeployer) TrackBuildArtifacts(_ []graph.Artifact) {}
 
 func (m *MockDeployer) Dependencies() ([]string, error) {

pkg/skaffold/deploy/helm/deploy.go

@@ -48,9 +48,11 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/instrumentation"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
+	kloader "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/loader"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/manifest"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/portforward"
 	kstatus "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/status"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/loader"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/log"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/output"
 	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
@@ -86,12 +88,14 @@ type Deployer struct {
 
 	accessor      access.Accessor
 	debugger      debug.Debugger
+	imageLoader   loader.ImageLoader
 	logger        log.Logger
 	statusMonitor status.Monitor
 	syncer        sync.Syncer
 
 	podSelector    *kubernetes.ImageList
-	originalImages []graph.Artifact
+	originalImages []graph.Artifact // the set of images defined in ArtifactOverrides
+	localImages    []graph.Artifact // the set of images marked as "local" by the Runner
 
 	kubeContext string
 	kubeConfig  string
@@ -113,6 +117,7 @@ type Deployer struct {
 type Config interface {
 	kubectl.Config
 	kstatus.Config
+	kloader.Config
 	portforward.Config
 	IsMultiConfig() bool
 }
@@ -144,6 +149,7 @@ func NewDeployer(cfg Config, labels map[string]string, provider deploy.Component
 		podSelector:    podSelector,
 		accessor:       provider.Accessor.GetKubernetesAccessor(cfg, podSelector),
 		debugger:       provider.Debugger.GetKubernetesDebugger(podSelector),
+		imageLoader:    provider.ImageLoader.GetKubernetesImageLoader(cfg),
 		logger:         provider.Logger.GetKubernetesLogger(podSelector),
 		statusMonitor:  provider.Monitor.GetKubernetesMonitor(cfg),
 		syncer:         provider.Syncer.GetKubernetesSyncer(podSelector),
@@ -180,6 +186,10 @@ func (h *Deployer) GetSyncer() sync.Syncer {
 	return h.syncer
 }
 
+func (h *Deployer) RegisterLocalImages(images []graph.Artifact) {
+	h.localImages = images
+}
+
 func (h *Deployer) TrackBuildArtifacts(artifacts []graph.Artifact) {
 	deployutil.AddTagsToPodSelector(artifacts, h.originalImages, h.podSelector)
 	h.logger.RegisterArtifacts(artifacts)
@@ -192,6 +202,13 @@ func (h *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 	})
 	defer endTrace()
 
+	childCtx, endTrace := instrumentation.StartTrace(ctx, "Deploy_LoadImages")
+	if err := h.imageLoader.LoadImages(childCtx, out, h.localImages, h.originalImages, builds); err != nil {
+		endTrace(instrumentation.TraceEndError(err))
+		return nil, err
+	}
+	endTrace()
+
 	logrus.Infof("Deploying with helm v%s ...", h.bV)
 
 	var dRes []types.Artifact

pkg/skaffold/deploy/kpt/kpt.go

@@ -43,9 +43,11 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/instrumentation"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
+	kloader "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/loader"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/manifest"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/portforward"
 	kstatus "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/status"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/loader"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/log"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/output"
 	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
@@ -77,11 +79,13 @@ type Deployer struct {
 	accessor      access.Accessor
 	logger        log.Logger
 	debugger      debug.Debugger
+	imageLoader   loader.ImageLoader
 	statusMonitor status.Monitor
 	syncer        sync.Syncer
 
 	podSelector    *kubernetes.ImageList
-	originalImages []graph.Artifact
+	originalImages []graph.Artifact // the set of images parsed from the Deployer's manifest set
+	localImages    []graph.Artifact // the set of images marked as "local" by the Runner
 
 	insecureRegistries map[string]bool
 	labels             map[string]string
@@ -96,6 +100,7 @@ type Config interface {
 	kubectl.Config
 	kstatus.Config
 	portforward.Config
+	kloader.Config
 }
 
 // NewDeployer generates a new Deployer object contains the kptDeploy schema.
@@ -106,6 +111,7 @@ func NewDeployer(cfg Config, labels map[string]string, provider deploy.Component
 		podSelector:        podSelector,
 		accessor:           provider.Accessor.GetKubernetesAccessor(cfg, podSelector),
 		debugger:           provider.Debugger.GetKubernetesDebugger(podSelector),
+		imageLoader:        provider.ImageLoader.GetKubernetesImageLoader(cfg),
 		logger:             provider.Logger.GetKubernetesLogger(podSelector),
 		statusMonitor:      provider.Monitor.GetKubernetesMonitor(cfg),
 		syncer:             provider.Syncer.GetKubernetesSyncer(podSelector),
@@ -139,6 +145,10 @@ func (k *Deployer) GetSyncer() sync.Syncer {
 	return k.syncer
 }
 
+func (k *Deployer) RegisterLocalImages(images []graph.Artifact) {
+	k.localImages = images
+}
+
 func (k *Deployer) TrackBuildArtifacts(artifacts []graph.Artifact) {
 	deployutil.AddTagsToPodSelector(artifacts, k.originalImages, k.podSelector)
 	k.logger.RegisterArtifacts(artifacts)
@@ -210,7 +220,13 @@ func (k *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 	}
 	endTrace()
 
-	childCtx, endTrace := instrumentation.StartTrace(ctx, "Deploy_renderManifests")
+	childCtx, endTrace := instrumentation.StartTrace(ctx, "Deploy_loadImages")
+	if err := k.imageLoader.LoadImages(childCtx, out, k.localImages, k.originalImages, builds); err != nil {
+		endTrace(instrumentation.TraceEndError(err))
+		return nil, err
+	}
+
+	_, endTrace = instrumentation.StartTrace(ctx, "Deploy_renderManifests")
 	manifests, err := k.renderManifests(childCtx, builds)
 	if err != nil {
 		endTrace(instrumentation.TraceEndError(err))

pkg/skaffold/deploy/kubectl/cli.go

@@ -31,6 +31,7 @@ import (
 	deploy "github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/types"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/instrumentation"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubectl"
+	kloader "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/loader"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/manifest"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/portforward"
 	kstatus "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/status"
@@ -50,6 +51,7 @@ type CLI struct {
 type Config interface {
 	kubectl.Config
 	kstatus.Config
+	kloader.Config
 	portforward.Config
 	deploy.Config
 	ForceDeploy() bool

pkg/skaffold/deploy/kubectl/kubectl.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/segmentio/textio"
 	"github.com/sirupsen/logrus"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/access"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
@@ -39,6 +40,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/instrumentation"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/manifest"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/loader"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/log"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/output"
 	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
@@ -52,12 +54,14 @@ type Deployer struct {
 	*latestV1.KubectlDeploy
 
 	accessor      access.Accessor
+	imageLoader   loader.ImageLoader
 	logger        log.Logger
 	debugger      debug.Debugger
 	statusMonitor status.Monitor
 	syncer        sync.Syncer
 
-	originalImages     []graph.Artifact
+	originalImages     []graph.Artifact // the set of images marked as "local" by the Runner
+	localImages        []graph.Artifact // the set of images parsed from the Deployer's manifest set
 	podSelector        *kubernetes.ImageList
 	hydratedManifests  []string
 	workingDir         string
@@ -89,6 +93,7 @@ func NewDeployer(cfg Config, labels map[string]string, provider deploy.Component
 		podSelector:        podSelector,
 		accessor:           provider.Accessor.GetKubernetesAccessor(cfg, podSelector),
 		debugger:           provider.Debugger.GetKubernetesDebugger(podSelector),
+		imageLoader:        provider.ImageLoader.GetKubernetesImageLoader(cfg),
 		logger:             provider.Logger.GetKubernetesLogger(podSelector),
 		statusMonitor:      provider.Monitor.GetKubernetesMonitor(cfg),
 		syncer:             provider.Syncer.GetKubernetesSyncer(podSelector),
@@ -123,6 +128,10 @@ func (k *Deployer) GetSyncer() sync.Syncer {
 	return k.syncer
 }
 
+func (k *Deployer) RegisterLocalImages(images []graph.Artifact) {
+	k.localImages = images
+}
+
 func (k *Deployer) TrackBuildArtifacts(artifacts []graph.Artifact) {
 	deployutil.AddTagsToPodSelector(artifacts, k.originalImages, k.podSelector)
 	k.logger.RegisterArtifacts(artifacts)
@@ -134,6 +143,8 @@ func (k *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 	var (
 		manifests manifest.ManifestList
 		err       error
+		childCtx  context.Context
+		endTrace  func(...trace.SpanOption)
 	)
 	instrumentation.AddAttributesToCurrentSpanFromContext(ctx, map[string]string{
 		"DeployerType": "kubectl",
@@ -143,7 +154,7 @@ func (k *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 	// also, manually set the labels to ensure the runID is added
 	switch {
 	case len(k.hydratedManifests) > 0:
-		_, endTrace := instrumentation.StartTrace(ctx, "Deploy_createManifestList")
+		_, endTrace = instrumentation.StartTrace(ctx, "Deploy_createManifestList")
 		manifests, err = createManifestList(k.hydratedManifests)
 		if err != nil {
 			endTrace(instrumentation.TraceEndError(err))
@@ -152,11 +163,11 @@ func (k *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 		manifests, err = manifests.SetLabels(k.labels)
 		endTrace()
 	case k.skipRender:
-		childCtx, endTrace := instrumentation.StartTrace(ctx, "Deploy_readManifests")
+		childCtx, endTrace = instrumentation.StartTrace(ctx, "Deploy_readManifests")
 		manifests, err = k.readManifests(childCtx, false)
 		endTrace()
 	default:
-		childCtx, endTrace := instrumentation.StartTrace(ctx, "Deploy_renderManifests")
+		childCtx, endTrace = instrumentation.StartTrace(ctx, "Deploy_renderManifests")
 		manifests, err = k.renderManifests(childCtx, out, builds, false)
 		endTrace()
 	}
@@ -168,15 +179,24 @@ func (k *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 	if len(manifests) == 0 {
 		return nil, nil
 	}
-	_, endTrace := instrumentation.StartTrace(ctx, "Deploy_CollectNamespaces")
+	endTrace()
+
+	_, endTrace = instrumentation.StartTrace(ctx, "Deploy_LoadImages")
+	if err := k.imageLoader.LoadImages(childCtx, out, k.localImages, k.originalImages, builds); err != nil {
+		endTrace(instrumentation.TraceEndError(err))
+		return nil, err
+	}
+	endTrace()
+
+	_, endTrace = instrumentation.StartTrace(ctx, "Deploy_CollectNamespaces")
 	namespaces, err := manifests.CollectNamespaces()
 	if err != nil {
 		event.DeployInfoEvent(fmt.Errorf("could not fetch deployed resource namespace. "+
 			"This might cause port-forward and deploy health-check to fail: %w", err))
 	}
 	endTrace()
 
-	childCtx, endTrace := instrumentation.StartTrace(ctx, "Deploy_WaitForDeletions")
+	childCtx, endTrace = instrumentation.StartTrace(ctx, "Deploy_WaitForDeletions")
 	if err := k.kubectl.WaitForDeletions(childCtx, textio.NewPrefixWriter(out, " - "), manifests); err != nil {
 		endTrace(instrumentation.TraceEndError(err))
 		return nil, err

pkg/skaffold/deploy/kustomize/kustomize.go

@@ -39,6 +39,7 @@ import (
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/instrumentation"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/manifest"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/loader"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/log"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/output"
 	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
@@ -104,12 +105,14 @@ type Deployer struct {
 
 	accessor      access.Accessor
 	logger        log.Logger
+	imageLoader   loader.ImageLoader
 	debugger      debug.Debugger
 	statusMonitor status.Monitor
 	syncer        sync.Syncer
 
 	podSelector    *kubernetes.ImageList
-	originalImages []graph.Artifact
+	originalImages []graph.Artifact // the set of images parsed from the Deployer's manifest set
+	localImages    []graph.Artifact // the set of images marked as "local" by the Runner
 
 	kubectl             kubectl.CLI
 	insecureRegistries  map[string]bool
@@ -138,6 +141,7 @@ func NewDeployer(cfg kubectl.Config, labels map[string]string, provider deploy.C
 		podSelector:         podSelector,
 		accessor:            provider.Accessor.GetKubernetesAccessor(cfg, podSelector),
 		debugger:            provider.Debugger.GetKubernetesDebugger(podSelector),
+		imageLoader:         provider.ImageLoader.GetKubernetesImageLoader(cfg),
 		logger:              provider.Logger.GetKubernetesLogger(podSelector),
 		statusMonitor:       provider.Monitor.GetKubernetesMonitor(cfg),
 		syncer:              provider.Syncer.GetKubernetesSyncer(podSelector),
@@ -169,6 +173,10 @@ func (k *Deployer) GetSyncer() sync.Syncer {
 	return k.syncer
 }
 
+func (k *Deployer) RegisterLocalImages(images []graph.Artifact) {
+	k.localImages = images
+}
+
 func (k *Deployer) TrackBuildArtifacts(artifacts []graph.Artifact) {
 	deployutil.AddTagsToPodSelector(artifacts, k.originalImages, k.podSelector)
 	k.logger.RegisterArtifacts(artifacts)
@@ -210,6 +218,13 @@ func (k *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 	}
 	endTrace()
 
+	childCtx, endTrace = instrumentation.StartTrace(ctx, "Deploy_LoadImages")
+	if err := k.imageLoader.LoadImages(childCtx, out, k.localImages, k.originalImages, builds); err != nil {
+		endTrace(instrumentation.TraceEndError(err))
+		return nil, err
+	}
+	endTrace()
+
 	_, endTrace = instrumentation.StartTrace(ctx, "Deploy_CollectNamespaces")
 	namespaces, err := manifests.CollectNamespaces()
 	if err != nil {