Repository: GoogleContainerTools/skaffold
Pull request #7172 (2022-03-09 18:36:08)

feat: add metrics related to cross-platform build
Add metrics related to cross-platform builds:
- Count of artifacts defining target platforms.
- Platforms specified via CLI flag `--platform`.
- Resolved build target platforms.
- Platform kind for the kubernetes cluster nodes.


Patch:
diff --git a/cmd/skaffold/app/cmd/flags.go b/cmd/skaffold/app/cmd/flags.go
index 38adfe044..7052e7dc9 100644
--- a/cmd/skaffold/app/cmd/flags.go
+++ b/cmd/skaffold/app/cmd/flags.go
@@ -606,6 +606,14 @@ var flagRegistry = []Flag{
 		FlagAddMethod: "BoolVar",
 		DefinedOn:     []string{"deploy"},
 	},
+	{
+		Name:          "resource-selector-rules-file",
+		Usage:         "Path to JSON file specifying the deny list of yaml objects for skaffold to NOT transform with 'image' and 'label' field replacements.  NOTE: this list is additive to skaffold's default denylist and denylist has priority over allowlist",
+		Value:         &opts.TransformRulesFile,
+		DefValue:      "",
+		FlagAddMethod: "StringVar",
+		DefinedOn:     []string{"dev", "render", "run", "debug", "deploy"},
+	},
 }
 
 func methodNameByType(v reflect.Value) string {
diff --git a/docs/content/en/docs/references/cli/_index.md b/docs/content/en/docs/references/cli/_index.md
index 3bd41d1b2..01cd22f0f 100644
--- a/docs/content/en/docs/references/cli/_index.md
+++ b/docs/content/en/docs/references/cli/_index.md
@@ -422,6 +422,7 @@ Options:
       --propagate-profiles=true: Setting '--propagate-profiles=false' disables propagating profiles set by the '--profile' flag across config dependencies. This mean that only profiles defined directly in the target 'skaffold.yaml' file are activated.
       --protocols=[]: Priority sorted order of debugger protocols to support.
       --remote-cache-dir='': Specify the location of the git repositories cache (default $HOME/.skaffold/repos)
+      --resource-selector-rules-file='': Path to JSON file specifying the deny list of yaml objects for skaffold to NOT transform with 'image' and 'label' field replacements.  NOTE: this list is additive to skaffold's default denylist and denylist has priority over allowlist
       --rpc-http-port=: tcp port to expose the Skaffold API over HTTP REST
       --rpc-port=: tcp port to expose the Skaffold API over gRPC
       --skip-tests=false: Whether to skip the tests after building
@@ -478,6 +479,7 @@ Env vars:
 * `SKAFFOLD_PROPAGATE_PROFILES` (same as `--propagate-profiles`)
 * `SKAFFOLD_PROTOCOLS` (same as `--protocols`)
 * `SKAFFOLD_REMOTE_CACHE_DIR` (same as `--remote-cache-dir`)
+* `SKAFFOLD_RESOURCE_SELECTOR_RULES_FILE` (same as `--resource-selector-rules-file`)
 * `SKAFFOLD_RPC_HTTP_PORT` (same as `--rpc-http-port`)
 * `SKAFFOLD_RPC_PORT` (same as `--rpc-port`)
 * `SKAFFOLD_SKIP_TESTS` (same as `--skip-tests`)
@@ -588,6 +590,7 @@ Options:
       --profile-auto-activation=true: Set to false to disable profile auto activation
       --propagate-profiles=true: Setting '--propagate-profiles=false' disables propagating profiles set by the '--profile' flag across config dependencies. This mean that only profiles defined directly in the target 'skaffold.yaml' file are activated.
       --remote-cache-dir='': Specify the location of the git repositories cache (default $HOME/.skaffold/repos)
+      --resource-selector-rules-file='': Path to JSON file specifying the deny list of yaml objects for skaffold to NOT transform with 'image' and 'label' field replacements.  NOTE: this list is additive to skaffold's default denylist and denylist has priority over allowlist
       --rpc-http-port=: tcp port to expose the Skaffold API over HTTP REST
       --rpc-port=: tcp port to expose the Skaffold API over gRPC
       --skip-render=false: Don't render the manifests, just deploy them
@@ -632,6 +635,7 @@ Env vars:
 * `SKAFFOLD_PROFILE_AUTO_ACTIVATION` (same as `--profile-auto-activation`)
 * `SKAFFOLD_PROPAGATE_PROFILES` (same as `--propagate-profiles`)
 * `SKAFFOLD_REMOTE_CACHE_DIR` (same as `--remote-cache-dir`)
+* `SKAFFOLD_RESOURCE_SELECTOR_RULES_FILE` (same as `--resource-selector-rules-file`)
 * `SKAFFOLD_RPC_HTTP_PORT` (same as `--rpc-http-port`)
 * `SKAFFOLD_RPC_PORT` (same as `--rpc-port`)
 * `SKAFFOLD_SKIP_RENDER` (same as `--skip-render`)
@@ -684,6 +688,7 @@ Options:
       --profile-auto-activation=true: Set to false to disable profile auto activation
       --propagate-profiles=true: Setting '--propagate-profiles=false' disables propagating profiles set by the '--profile' flag across config dependencies. This mean that only profiles defined directly in the target 'skaffold.yaml' file are activated.
       --remote-cache-dir='': Specify the location of the git repositories cache (default $HOME/.skaffold/repos)
+      --resource-selector-rules-file='': Path to JSON file specifying the deny list of yaml objects for skaffold to NOT transform with 'image' and 'label' field replacements.  NOTE: this list is additive to skaffold's default denylist and denylist has priority over allowlist
       --rpc-http-port=: tcp port to expose the Skaffold API over HTTP REST
       --rpc-port=: tcp port to expose the Skaffold API over gRPC
       --skip-tests=false: Whether to skip the tests after building
@@ -740,6 +745,7 @@ Env vars:
 * `SKAFFOLD_PROFILE_AUTO_ACTIVATION` (same as `--profile-auto-activation`)
 * `SKAFFOLD_PROPAGATE_PROFILES` (same as `--propagate-profiles`)
 * `SKAFFOLD_REMOTE_CACHE_DIR` (same as `--remote-cache-dir`)
+* `SKAFFOLD_RESOURCE_SELECTOR_RULES_FILE` (same as `--resource-selector-rules-file`)
 * `SKAFFOLD_RPC_HTTP_PORT` (same as `--rpc-http-port`)
 * `SKAFFOLD_RPC_PORT` (same as `--rpc-port`)
 * `SKAFFOLD_SKIP_TESTS` (same as `--skip-tests`)
@@ -934,6 +940,7 @@ Options:
       --profile-auto-activation=true: Set to false to disable profile auto activation
       --propagate-profiles=true: Setting '--propagate-profiles=false' disables propagating profiles set by the '--profile' flag across config dependencies. This mean that only profiles defined directly in the target 'skaffold.yaml' file are activated.
       --remote-cache-dir='': Specify the location of the git repositories cache (default $HOME/.skaffold/repos)
+      --resource-selector-rules-file='': Path to JSON file specifying the deny list of yaml objects for skaffold to NOT transform with 'image' and 'label' field replacements.  NOTE: this list is additive to skaffold's default denylist and denylist has priority over allowlist
       --sync-remote-cache='always': Controls how Skaffold manages the remote config cache (see `remote-cache-dir`). One of `always` (default), `missing`, or `never`. `always` syncs remote repositories to latest on access. `missing` only clones remote repositories if they do not exist locally. `never` means the user takes responsibility for updating remote repositories.
       --wait-for-connection=false: Blocks ending execution of skaffold until the /v2/events gRPC/HTTP endpoint is hit
 
@@ -962,6 +969,7 @@ Env vars:
 * `SKAFFOLD_PROFILE_AUTO_ACTIVATION` (same as `--profile-auto-activation`)
 * `SKAFFOLD_PROPAGATE_PROFILES` (same as `--propagate-profiles`)
 * `SKAFFOLD_REMOTE_CACHE_DIR` (same as `--remote-cache-dir`)
+* `SKAFFOLD_RESOURCE_SELECTOR_RULES_FILE` (same as `--resource-selector-rules-file`)
 * `SKAFFOLD_SYNC_REMOTE_CACHE` (same as `--sync-remote-cache`)
 * `SKAFFOLD_WAIT_FOR_CONNECTION` (same as `--wait-for-connection`)
 
@@ -1009,6 +1017,7 @@ Options:
       --profile-auto-activation=true: Set to false to disable profile auto activation
       --propagate-profiles=true: Setting '--propagate-profiles=false' disables propagating profiles set by the '--profile' flag across config dependencies. This mean that only profiles defined directly in the target 'skaffold.yaml' file are activated.
       --remote-cache-dir='': Specify the location of the git repositories cache (default $HOME/.skaffold/repos)
+      --resource-selector-rules-file='': Path to JSON file specifying the deny list of yaml objects for skaffold to NOT transform with 'image' and 'label' field replacements.  NOTE: this list is additive to skaffold's default denylist and denylist has priority over allowlist
       --rpc-http-port=: tcp port to expose the Skaffold API over HTTP REST
       --rpc-port=: tcp port to expose the Skaffold API over gRPC
       --skip-tests=false: Whether to skip the tests after building
@@ -1060,6 +1069,7 @@ Env vars:
 * `SKAFFOLD_PROFILE_AUTO_ACTIVATION` (same as `--profile-auto-activation`)
 * `SKAFFOLD_PROPAGATE_PROFILES` (same as `--propagate-profiles`)
 * `SKAFFOLD_REMOTE_CACHE_DIR` (same as `--remote-cache-dir`)
+* `SKAFFOLD_RESOURCE_SELECTOR_RULES_FILE` (same as `--resource-selector-rules-file`)
 * `SKAFFOLD_RPC_HTTP_PORT` (same as `--rpc-http-port`)
 * `SKAFFOLD_RPC_PORT` (same as `--rpc-port`)
 * `SKAFFOLD_SKIP_TESTS` (same as `--skip-tests`)
diff --git a/docs/content/en/schemas/v2beta28.json b/docs/content/en/schemas/v2beta28.json
index d04c42f08..1f3b065f0 100755
--- a/docs/content/en/schemas/v2beta28.json
+++ b/docs/content/en/schemas/v2beta28.json
@@ -3301,6 +3301,11 @@
           "description": "describes user defined resources to port-forward.",
           "x-intellij-html-description": "describes user defined resources to port-forward."
         },
+        "resourceSelector": {
+          "$ref": "#/definitions/ResourceSelectorConfig",
+          "description": "describes user defined filters describing how skaffold should treat objects/fields during rendering.",
+          "x-intellij-html-description": "describes user defined filters describing how skaffold should treat objects/fields during rendering."
+        },
         "test": {
           "items": {
             "$ref": "#/definitions/TestCase"
@@ -3317,7 +3322,8 @@
         "build",
         "test",
         "deploy",
-        "portForward"
+        "portForward",
+        "resourceSelector"
       ],
       "additionalProperties": false,
       "type": "object",
@@ -3355,9 +3361,14 @@
     },
     "ResourceFilter": {
       "required": [
-        "type"
+        "groupKind"
       ],
       "properties": {
+        "groupKind": {
+          "type": "string",
+          "description": "compact format of a resource type.",
+          "x-intellij-html-description": "compact format of a resource type."
+        },
         "image": {
           "items": {
             "type": "string"
@@ -3375,15 +3386,10 @@
           "description": "an optional slide of JSON-path-like paths of where to add a labels block if missing.",
           "x-intellij-html-description": "an optional slide of JSON-path-like paths of where to add a labels block if missing.",
           "default": "[]"
-        },
-        "type": {
-          "type": "string",
-          "description": "compact format of a resource type.",
-          "x-intellij-html-description": "compact format of a resource type."
         }
       },
       "preferredOrder": [
-        "type",
+        "groupKind",
         "image",
         "labels"
       ],
@@ -3460,6 +3466,34 @@
       "description": "describes the resource requirements for the kaniko pod.",
       "x-intellij-html-description": "describes the resource requirements for the kaniko pod."
     },
+    "ResourceSelectorConfig": {
+      "properties": {
+        "allow": {
+          "items": {
+            "$ref": "#/definitions/ResourceFilter"
+          },
+          "type": "array",
+          "description": "configures an allowlist for transforming manifests.",
+          "x-intellij-html-description": "configures an allowlist for transforming manifests."
+        },
+        "deny": {
+          "items": {
+            "$ref": "#/definitions/ResourceFilter"
+          },
+          "type": "array",
+          "description": "configures an allowlist for transforming manifests.",
+          "x-intellij-html-description": "configures an allowlist for transforming manifests."
+        }
+      },
+      "preferredOrder": [
+        "allow",
+        "deny"
+      ],
+      "additionalProperties": false,
+      "type": "object",
+      "description": "contains all the configuration needed by the deploy steps.",
+      "x-intellij-html-description": "contains all the configuration needed by the deploy steps."
+    },
     "ResourceType": {
       "type": "string",
       "description": "describes the Kubernetes resource types used for port forwarding.",
@@ -3526,6 +3560,11 @@
           "description": "describes a list of other required configs for the current config.",
           "x-intellij-html-description": "describes a list of other required configs for the current config."
         },
+        "resourceSelector": {
+          "$ref": "#/definitions/ResourceSelectorConfig",
+          "description": "describes user defined filters describing how skaffold should treat objects/fields during rendering.",
+          "x-intellij-html-description": "describes user defined filters describing how skaffold should treat objects/fields during rendering."
+        },
         "test": {
           "items": {
             "$ref": "#/definitions/TestCase"
@@ -3544,6 +3583,7 @@
         "test",
         "deploy",
         "portForward",
+        "resourceSelector",
         "profiles"
       ],
       "additionalProperties": false,
diff --git a/pkg/skaffold/config/options.go b/pkg/skaffold/config/options.go
index 45d3b5b5d..d955a8a93 100644
--- a/pkg/skaffold/config/options.go
+++ b/pkg/skaffold/config/options.go
@@ -46,12 +46,13 @@ type SkaffoldOptions struct {
 	// TODO(https://github.com/GoogleContainerTools/skaffold/issues/3668):
 	// remove minikubeProfile from here and instead detect it by matching the
 	// kubecontext API Server to minikube profiles
-	MinikubeProfile string
-	Namespace       string
-	RenderOutput    string
-	RepoCacheDir    string
-	Trigger         string
-	User            string
+	MinikubeProfile    string
+	Namespace          string
+	RenderOutput       string
+	RepoCacheDir       string
+	Trigger            string
+	User               string
+	TransformRulesFile string
 
 	ConfigurationFilter []string
 	CustomLabels        []string
diff --git a/pkg/skaffold/deploy/kpt/kpt.go b/pkg/skaffold/deploy/kpt/kpt.go
index e4741f301..f2a6b2fc7 100644
--- a/pkg/skaffold/deploy/kpt/kpt.go
+++ b/pkg/skaffold/deploy/kpt/kpt.go
@@ -30,6 +30,7 @@ import (
 
 	"golang.org/x/mod/semver"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	apimachinery "k8s.io/apimachinery/pkg/runtime/schema"
 	k8syaml "sigs.k8s.io/yaml"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/access"
@@ -100,6 +101,9 @@ type Deployer struct {
 	namespace          string
 
 	namespaces *[]string
+
+	transformableAllowlist map[apimachinery.GroupKind]latestV1.ResourceFilter
+	transformableDenylist  map[apimachinery.GroupKind]latestV1.ResourceFilter
 }
 
 type Config interface {
@@ -110,7 +114,7 @@ type Config interface {
 }
 
 // NewDeployer generates a new Deployer object contains the kptDeploy schema.
-func NewDeployer(cfg Config, labeller *label.DefaultLabeller, d *latestV1.KptDeploy) *Deployer {
+func NewDeployer(cfg Config, labeller *label.DefaultLabeller, d *latestV1.KptDeploy) (*Deployer, error) {
 	podSelector := kubernetes.NewImageList()
 	kubectl := pkgkubectl.NewCLI(cfg, cfg.GetKubeNamespace())
 	namespaces, err := deployutil.GetAllPodNamespaces(cfg.GetNamespace(), cfg.GetPipelines())
@@ -118,24 +122,30 @@ func NewDeployer(cfg Config, labeller *label.DefaultLabeller, d *latestV1.KptDep
 		olog.Entry(context.TODO()).Warn("unable to parse namespaces - deploy might not work correctly!")
 	}
 	logger := component.NewLogger(cfg, kubectl, podSelector, &namespaces)
-	return &Deployer{
-		KptDeploy:          d,
-		podSelector:        podSelector,
-		namespaces:         &namespaces,
-		accessor:           component.NewAccessor(cfg, cfg.GetKubeContext(), kubectl, podSelector, labeller, &namespaces),
-		debugger:           component.NewDebugger(cfg.Mode(), podSelector, &namespaces, cfg.GetKubeContext()),
-		imageLoader:        component.NewImageLoader(cfg, kubectl),
-		logger:             logger,
-		statusMonitor:      component.NewMonitor(cfg, cfg.GetKubeContext(), labeller, &namespaces),
-		syncer:             component.NewSyncer(kubectl, &namespaces, logger.GetFormatter()),
-		insecureRegistries: cfg.GetInsecureRegistries(),
-		labels:             labeller.Labels(),
-		globalConfig:       cfg.GlobalConfig(),
-		hasKustomization:   hasKustomization,
-		kubeContext:        cfg.GetKubeContext(),
-		kubeConfig:         cfg.GetKubeConfig(),
-		namespace:          cfg.GetKubeNamespace(),
+	transformableAllowlist, transformableDenylist, err := deployutil.ConsolidateTransformConfiguration(cfg)
+	if err != nil {
+		return nil, err
 	}
+	return &Deployer{
+		KptDeploy:              d,
+		podSelector:            podSelector,
+		namespaces:             &namespaces,
+		accessor:               component.NewAccessor(cfg, cfg.GetKubeContext(), kubectl, podSelector, labeller, &namespaces),
+		debugger:               component.NewDebugger(cfg.Mode(), podSelector, &namespaces, cfg.GetKubeContext()),
+		imageLoader:            component.NewImageLoader(cfg, kubectl),
+		logger:                 logger,
+		statusMonitor:          component.NewMonitor(cfg, cfg.GetKubeContext(), labeller, &namespaces),
+		syncer:                 component.NewSyncer(kubectl, &namespaces, logger.GetFormatter()),
+		insecureRegistries:     cfg.GetInsecureRegistries(),
+		labels:                 labeller.Labels(),
+		globalConfig:           cfg.GlobalConfig(),
+		hasKustomization:       hasKustomization,
+		kubeContext:            cfg.GetKubeContext(),
+		kubeConfig:             cfg.GetKubeConfig(),
+		namespace:              cfg.GetKubeNamespace(),
+		transformableAllowlist: transformableAllowlist,
+		transformableDenylist:  transformableDenylist,
+	}, nil
 }
 
 func (k *Deployer) trackNamespaces(namespaces []string) {
@@ -475,12 +485,12 @@ func (k *Deployer) renderManifests(ctx context.Context, builds []graph.Artifact)
 		return nil, fmt.Errorf("excluding kpt functions from manifests: %w", err)
 	}
 	if k.originalImages == nil {
-		k.originalImages, err = manifests.GetImages()
+		k.originalImages, err = manifests.GetImages(manifest.NewResourceSelectorImages(k.transformableAllowlist, k.transformableDenylist))
 		if err != nil {
 			return nil, err
 		}
 	}
-	manifests, err = manifests.ReplaceImages(ctx, builds)
+	manifests, err = manifests.ReplaceImages(ctx, builds, manifest.NewResourceSelectorImages(k.transformableAllowlist, k.transformableDenylist))
 	if err != nil {
 		return nil, fmt.Errorf("replacing images in manifests: %w", err)
 	}
@@ -489,7 +499,7 @@ func (k *Deployer) renderManifests(ctx context.Context, builds []graph.Artifact)
 		return nil, err
 	}
 
-	return manifests.SetLabels(k.labels)
+	return manifests.SetLabels(k.labels, manifest.NewResourceSelectorLabels(k.transformableAllowlist, k.transformableDenylist))
 }
 
 func sink(ctx context.Context, buf []byte, sinkDir string) error {
diff --git a/pkg/skaffold/deploy/kubectl/kubectl.go b/pkg/skaffold/deploy/kubectl/kubectl.go
index a5c264c91..f4664c005 100644
--- a/pkg/skaffold/deploy/kubectl/kubectl.go
+++ b/pkg/skaffold/deploy/kubectl/kubectl.go
@@ -27,6 +27,7 @@ import (
 
 	"github.com/segmentio/textio"
 	"go.opentelemetry.io/otel/trace"
+	apimachinery "k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/access"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
@@ -80,6 +81,9 @@ type Deployer struct {
 	skipRender         bool
 
 	namespaces *[]string
+
+	transformableAllowlist map[apimachinery.GroupKind]latestV1.ResourceFilter
+	transformableDenylist  map[apimachinery.GroupKind]latestV1.ResourceFilter
 }
 
 // NewDeployer returns a new Deployer for a DeployConfig filled
@@ -101,27 +105,32 @@ func NewDeployer(cfg Config, labeller *label.DefaultLabeller, d *latestV1.Kubect
 		olog.Entry(context.TODO()).Warn("unable to parse namespaces - deploy might not work correctly!")
 	}
 	logger := component.NewLogger(cfg, kubectl.CLI, podSelector, &namespaces)
-
+	transformableAllowlist, transformableDenylist, err := deployutil.ConsolidateTransformConfiguration(cfg)
+	if err != nil {
+		return nil, err
+	}
 	return &Deployer{
-		KubectlDeploy:      d,
-		podSelector:        podSelector,
-		namespaces:         &namespaces,
-		accessor:           component.NewAccessor(cfg, cfg.GetKubeContext(), kubectl.CLI, podSelector, labeller, &namespaces),
-		debugger:           component.NewDebugger(cfg.Mode(), podSelector, &namespaces, cfg.GetKubeContext()),
-		imageLoader:        component.NewImageLoader(cfg, kubectl.CLI),
-		logger:             logger,
-		statusMonitor:      component.NewMonitor(cfg, cfg.GetKubeContext(), labeller, &namespaces),
-		syncer:             component.NewSyncer(kubectl.CLI, &namespaces, logger.GetFormatter()),
-		hookRunner:         hooks.NewDeployRunner(kubectl.CLI, d.LifecycleHooks, &namespaces, logger.GetFormatter(), hooks.NewDeployEnvOpts(labeller.GetRunID(), kubectl.KubeContext, namespaces)),
-		workingDir:         cfg.GetWorkingDir(),
-		globalConfig:       cfg.GlobalConfig(),
-		defaultRepo:        cfg.DefaultRepo(),
-		multiLevelRepo:     cfg.MultiLevelRepo(),
-		kubectl:            kubectl,
-		insecureRegistries: cfg.GetInsecureRegistries(),
-		skipRender:         cfg.SkipRender(),
-		labeller:           labeller,
-		hydratedManifests:  cfg.HydratedManifests(),
+		KubectlDeploy:          d,
+		podSelector:            podSelector,
+		namespaces:             &namespaces,
+		accessor:               component.NewAccessor(cfg, cfg.GetKubeContext(), kubectl.CLI, podSelector, labeller, &namespaces),
+		debugger:               component.NewDebugger(cfg.Mode(), podSelector, &namespaces, cfg.GetKubeContext()),
+		imageLoader:            component.NewImageLoader(cfg, kubectl.CLI),
+		logger:                 logger,
+		statusMonitor:          component.NewMonitor(cfg, cfg.GetKubeContext(), labeller, &namespaces),
+		syncer:                 component.NewSyncer(kubectl.CLI, &namespaces, logger.GetFormatter()),
+		hookRunner:             hooks.NewDeployRunner(kubectl.CLI, d.LifecycleHooks, &namespaces, logger.GetFormatter(), hooks.NewDeployEnvOpts(labeller.GetRunID(), kubectl.KubeContext, namespaces)),
+		workingDir:             cfg.GetWorkingDir(),
+		globalConfig:           cfg.GlobalConfig(),
+		defaultRepo:            cfg.DefaultRepo(),
+		multiLevelRepo:         cfg.MultiLevelRepo(),
+		kubectl:                kubectl,
+		insecureRegistries:     cfg.GetInsecureRegistries(),
+		skipRender:             cfg.SkipRender(),
+		labeller:               labeller,
+		hydratedManifests:      cfg.HydratedManifests(),
+		transformableAllowlist: transformableAllowlist,
+		transformableDenylist:  transformableDenylist,
 	}, nil
 }
 
@@ -188,7 +197,7 @@ func (k *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 			endTrace(instrumentation.TraceEndError(err))
 			return err
 		}
-		manifests, err = manifests.SetLabels(k.labeller.Labels())
+		manifests, err = manifests.SetLabels(k.labeller.Labels(), manifest.NewResourceSelectorLabels(k.transformableAllowlist, k.transformableDenylist))
 		endTrace()
 	case k.skipRender:
 		childCtx, endTrace = instrumentation.StartTrace(ctx, "Deploy_readManifests")
@@ -197,7 +206,7 @@ func (k *Deployer) Deploy(ctx context.Context, out io.Writer, builds []graph.Art
 			endTrace(instrumentation.TraceEndError(err))
 			return err
 		}
-		manifests, err = manifests.SetLabels(k.labeller.Labels())
+		manifests, err = manifests.SetLabels(k.labeller.Labels(), manifest.NewResourceSelectorLabels(k.transformableAllowlist, k.transformableDenylist))
 		endTrace()
 	default:
 		childCtx, endTrace = instrumentation.StartTrace(ctx, "Deploy_renderManifests")
@@ -434,7 +443,9 @@ func (k *Deployer) renderManifests(ctx context.Context, out io.Writer, builds []
 	originalManifests := append(localManifests, remoteManifests...)
 
 	if len(k.originalImages) == 0 {
-		k.originalImages, err = originalManifests.GetImages()
+		// TODO(aaron-prindle) maybe use different resoureselector?
+		k.originalImages, err = originalManifests.GetImages(manifest.NewResourceSelectorImages(k.transformableAllowlist, k.transformableDenylist))
+		// k.originalImages, err = originalManifests.GetImages(k.transformableAllowlist, k.transformableDenylist)
 		if err != nil {
 			return nil, err
 		}
@@ -457,13 +468,13 @@ func (k *Deployer) renderManifests(ctx context.Context, out io.Writer, builds []
 		}
 	}
 	if len(remoteManifests) > 0 {
-		remoteManifests, err = remoteManifests.ReplaceRemoteManifestImages(ctx, builds)
+		remoteManifests, err = remoteManifests.ReplaceRemoteManifestImages(ctx, builds, manifest.NewResourceSelectorImages(k.transformableAllowlist, k.transformableDenylist))
 		if err != nil {
 			return nil, err
 		}
 	}
 	if len(localManifests) > 0 {
-		localManifests, err = localManifests.ReplaceImages(ctx, builds)
+		localManifests, err = localManifests.ReplaceImages(ctx, builds, manifest.NewResourceSelectorImages(k.transformableAllowlist, k.transformableDenylist))
 		if err != nil {
 			return nil, err
 		}
@@ -475,7 +486,7 @@ func (k *Deployer) renderManifests(ctx context.Context, out io.Writer, builds []
 		return nil, err
 	}
 
-	return modifiedManifests.SetLabels(k.labeller.Labels())
+	return modifiedManifests.SetLabels(k.labeller.Labels(), manifest.NewResourceSelectorLabels(k.transformableAllowlist, k.transformableDenylist))
 }
 
 // Cleanup deletes what was deployed by calling Deploy.
@@ -506,7 +517,7 @@ func (k *Deployer) Cleanup(ctx context.Context, out io.Writer, dryRun bool) erro
 			rm = append(rm, manifest)
 		}
 
-		upd, err := rm.ReplaceRemoteManifestImages(ctx, k.originalImages)
+		upd, err := rm.ReplaceRemoteManifestImages(ctx, k.originalImages, manifest.NewResourceSelectorImages(k.transformableAllowlist, k.transformableDenylist))
 		if err != nil {
 			return err
 		}
diff --git a/pkg/skaffold/deploy/kustomize/kustomize.go b/pkg/skaffold/deploy/kustomize/kustomize.go
index 2df6aa447..0bf3bc66e 100644
--- a/pkg/skaffold/deploy/kustomize/kustomize.go
+++ b/pkg/skaffold/deploy/kustomize/kustomize.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/segmentio/textio"
 	yamlv3 "gopkg.in/yaml.v3"
+	apimachinery "k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/access"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
@@ -127,6 +128,9 @@ type Deployer struct {
 	useKubectlKustomize bool
 
 	namespaces *[]string
+
+	transformableAllowlist map[apimachinery.GroupKind]latestV1.ResourceFilter
+	transformableDenylist  map[apimachinery.GroupKind]latestV1.ResourceFilter
 }
 
 func NewDeployer(cfg kubectl.Config, labeller *label.DefaultLabeller, d *latestV1.KustomizeDeploy) (*Deployer, error) {
@@ -149,22 +153,28 @@ func NewDeployer(cfg kubectl.Config, labeller *label.DefaultLabeller, d *latestV
 		olog.Entry(context.TODO()).Warn("unable to parse namespaces - deploy might not work correctly!")
 	}
 	logger := component.NewLogger(cfg, kubectl.CLI, podSelector, &namespaces)
+	transformableAllowlist, transformableDenylist, err := deployutil.ConsolidateTransformConfiguration(cfg)
+	if err != nil {
+		return nil, err
+	}
 	return &Deployer{
-		KustomizeDeploy:     d,
-		podSelector:         podSelector,
-		namespaces:          &namespaces,
-		accessor:            component.NewAccessor(cfg, cfg.GetKubeContext(), kubectl.CLI, podSelector, labeller, &namespaces),
-		debugger:            component.NewDebugger(cfg.Mode(), podSelector, &namespaces, cfg.GetKubeContext()),
-		hookRunner:          hooks.NewDeployRunner(kubectl.CLI, d.LifecycleHooks, &namespaces, logger.GetFormatter(), hooks.NewDeployEnvOpts(labeller.GetRunID(), kubectl.KubeContext, namespaces)),
-		imageLoader:         component.NewImageLoader(cfg, kubectl.CLI),
-		logger:              logger,
-		statusMonitor:       component.NewMonitor(cfg, cfg.GetKubeContext(), labeller, &namespaces),
-		syncer:              component.NewSyncer(kubectl.CLI, &namespaces, logger.GetFormatter()),
-		kubectl:             kubectl,
-		insecureRegistries:  cfg.GetInsecureRegistries(),
-		globalConfig:        cfg.GlobalConfig(),
-		labels:              labeller.Labels(),
-		useKubectlKustomize: useKubectlKustomize,
+		KustomizeDeploy:        d,
+		podSelector:            podSelector,
+		namespaces:             &namespaces,
+		accessor:               component.NewAccessor(cfg, cfg.GetKubeContext(), kubectl.CLI, podSelector, labeller, &namespaces),
+		debugger:               component.NewDebugger(cfg.Mode(), podSelector, &namespaces, cfg.GetKubeContext()),
+		hookRunner:             hooks.NewDeployRunner(kubectl.CLI, d.LifecycleHooks, &namespaces, logger.GetFormatter(), hooks.NewDeployEnvOpts(labeller.GetRunID(), kubectl.KubeContext, namespaces)),
+		imageLoader:            component.NewImageLoader(cfg, kubectl.CLI),
+		logger:                 logger,
+		statusMonitor:          component.NewMonitor(cfg, cfg.GetKubeContext(), labeller, &namespaces),
+		syncer:                 component.NewSyncer(kubectl.CLI, &namespaces, logger.GetFormatter()),
+		kubectl:                kubectl,
+		insecureRegistries:     cfg.GetInsecureRegistries(),
+		globalConfig:           cfg.GlobalConfig(),
+		labels:                 labeller.Labels(),
+		useKubectlKustomize:    useKubectlKustomize,
+		transformableAllowlist: transformableAllowlist,
+		transformableDenylist:  transformableDenylist,
 	}, nil
 }
 
@@ -325,13 +335,13 @@ func (k *Deployer) renderManifests(ctx context.Context, out io.Writer, builds []
 	}
 
 	if len(k.originalImages) == 0 {
-		k.originalImages, err = manifests.GetImages()
+		k.originalImages, err = manifests.GetImages(manifest.NewResourceSelectorImages(k.transformableAllowlist, k.transformableDenylist))
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	manifests, err = manifests.ReplaceImages(ctx, builds)
+	manifests, err = manifests.ReplaceImages(ctx, builds, manifest.NewResourceSelectorImages(k.transformableAllowlist, k.transformableDenylist))
 	if err != nil {
 		return nil, err
 	}
@@ -340,7 +350,7 @@ func (k *Deployer) renderManifests(ctx context.Context, out io.Writer, builds []
 		return nil, err
 	}
 
-	return manifests.SetLabels(k.labels)
+	return manifests.SetLabels(k.labels, manifest.NewResourceSelectorLabels(k.transformableAllowlist, k.transformableDenylist))
 }
 
 // Cleanup deletes what was deployed by calling Deploy.
diff --git a/pkg/skaffold/deploy/types/types.go b/pkg/skaffold/deploy/types/types.go
index e4f7a3c10..63545396f 100644
--- a/pkg/skaffold/deploy/types/types.go
+++ b/pkg/skaffold/deploy/types/types.go
@@ -34,7 +34,9 @@ type Config interface {
 	DefaultRepo() *string
 	MultiLevelRepo() *bool
 	SkipRender() bool
-	TransformableAllowList() []latestV1.ResourceFilter
+	TransformAllowList() []latestV1.ResourceFilter
+	TransformDenyList() []latestV1.ResourceFilter
+	TransformRulesFile() string
 }
 
 // Artifact contains all information about a completed deployment
diff --git a/pkg/skaffold/deploy/util/util.go b/pkg/skaffold/deploy/util/util.go
index c6b521f43..a7ef02b7b 100644
--- a/pkg/skaffold/deploy/util/util.go
+++ b/pkg/skaffold/deploy/util/util.go
@@ -18,6 +18,8 @@ package util
 
 import (
 	"fmt"
+	"io/ioutil"
+	"strings"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/discovery"
@@ -25,10 +27,14 @@ import (
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/deploy/types"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/docker"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/manifest"
+	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/util/stringset"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/yaml"
 )
 
 // ApplyDefaultRepo applies the default repo to a given image tag.
@@ -98,3 +104,92 @@ func GroupVersionResource(disco discovery.DiscoveryInterface, gvk schema.GroupVe
 
 	return false, schema.GroupVersionResource{}, fmt.Errorf("could not find resource for %s", gvk.String())
 }
+
+func ConsolidateTransformConfiguration(cfg types.Config) (map[schema.GroupKind]latestV1.ResourceFilter, map[schema.GroupKind]latestV1.ResourceFilter, error) {
+	// TODO(aaron-prindle) currently this also modifies the flag & config to support a JSON path syntax for input.
+	// this should be done elsewhere eventually
+
+	transformableAllowlist := map[schema.GroupKind]latestV1.ResourceFilter{}
+	transformableDenylist := map[schema.GroupKind]latestV1.ResourceFilter{}
+	// add default values
+	for _, rf := range manifest.TransformAllowlist {
+		groupKind := schema.ParseGroupKind(rf.GroupKind)
+		transformableAllowlist[groupKind] = convertJSONPathIndex(rf)
+	}
+	for _, rf := range manifest.TransformDenylist {
+		groupKind := schema.ParseGroupKind(rf.GroupKind)
+		transformableDenylist[groupKind] = convertJSONPathIndex(rf)
+	}
+
+	// add user schema values, override defaults
+	for _, rf := range cfg.TransformAllowList() {
+		groupKind := schema.ParseGroupKind(rf.GroupKind)
+		transformableAllowlist[groupKind] = convertJSONPathIndex(rf)
+		delete(transformableDenylist, groupKind)
+	}
+	for _, rf := range cfg.TransformDenyList() {
+		groupKind := schema.ParseGroupKind(rf.GroupKind)
+		transformableDenylist[groupKind] = convertJSONPathIndex(rf)
+		delete(transformableAllowlist, groupKind)
+	}
+
+	// add user flag values, override user schema values and defaults
+	// TODO(aaron-prindle) see if workdir needs to be considered in this read
+	if cfg.TransformRulesFile() != "" {
+		transformRulesFromFile, err := ioutil.ReadFile(cfg.TransformRulesFile())
+		if err != nil {
+			return nil, nil, err
+		}
+		rsc := latestV1.ResourceSelectorConfig{}
+		err = yaml.Unmarshal(transformRulesFromFile, &rsc)
+		if err != nil {
+			return nil, nil, err
+		}
+		for _, rf := range rsc.Allow {
+			groupKind := schema.ParseGroupKind(rf.GroupKind)
+			transformableAllowlist[groupKind] = convertJSONPathIndex(rf)
+			delete(transformableDenylist, groupKind)
+		}
+
+		for _, rf := range rsc.Deny {
+			groupKind := schema.ParseGroupKind(rf.GroupKind)
+			transformableDenylist[groupKind] = convertJSONPathIndex(rf)
+			delete(transformableAllowlist, groupKind)
+		}
+	}
+
+	return transformableAllowlist, transformableDenylist, nil
+}
+
+func convertJSONPathIndex(rf latestV1.ResourceFilter) latestV1.ResourceFilter {
+	nrf := latestV1.ResourceFilter{}
+	nrf.GroupKind = rf.GroupKind
+
+	if len(rf.Labels) > 0 {
+		nlabels := []string{}
+		for _, str := range rf.Labels {
+			if str == ".*" {
+				nlabels = append(nlabels, str)
+				continue
+			}
+			nstr := strings.ReplaceAll(str, ".*", "")
+			nlabels = append(nlabels, nstr)
+		}
+		nrf.Labels = nlabels
+	}
+
+	if len(rf.Image) > 0 {
+		nimage := []string{}
+		for _, str := range rf.Image {
+			if str == ".*" {
+				nimage = append(nimage, str)
+				continue
+			}
+			nstr := strings.ReplaceAll(str, ".*", "")
+			nimage = append(nimage, nstr)
+		}
+		nrf.Image = nimage
+	}
+
+	return nrf
+}
diff --git a/pkg/skaffold/instrumentation/export.go b/pkg/skaffold/instrumentation/export.go
index c7696ad40..d9c313478 100644
--- a/pkg/skaffold/instrumentation/export.go
+++ b/pkg/skaffold/instrumentation/export.go
@@ -271,11 +271,30 @@ func builderMetrics(ctx context.Context, meter skaffoldMeter, m metric.Meter, la
 	builderCounter := metric.Must(m).NewInt64ValueRecorder("builders", metric.WithDescription("Builders used"))
 	artifactCounter := metric.Must(m).NewInt64ValueRecorder("artifacts", metric.WithDescription("Number of artifacts used"))
 	dependenciesCounter := metric.Must(m).NewInt64ValueRecorder("artifact-dependencies", metric.WithDescription("Number of artifacts with dependencies"))
+	platformsCounter := metric.Must(m).NewInt64ValueRecorder("artifact-with-platforms", metric.WithDescription("Number of artifacts with target platforms specified"))
 	for builder, count := range meter.Builders {
 		bLabel := attribute.String("builder", builder)
 		builderCounter.Record(ctx, 1, append(labels, bLabel)...)
 		artifactCounter.Record(ctx, int64(count), append(labels, bLabel)...)
 		dependenciesCounter.Record(ctx, int64(meter.BuildDependencies[builder]), append(labels, bLabel)...)
+		platformsCounter.Record(ctx, int64(meter.BuildWithPlatforms[builder]), append(labels, bLabel)...)
+	}
+
+	if len(meter.ResolvedBuildTargetPlatforms) > 0 {
+		platforms := metric.Must(m).NewInt64ValueRecorder("build-platforms", metric.WithDescription("The resolved build target platforms for each run"))
+		for _, buildPlatform := range meter.ResolvedBuildTargetPlatforms {
+			platforms.Record(ctx, 1, append(labels, attribute.String("os_arch", buildPlatform))...)
+		}
+	}
+
+	if len(meter.CliBuildTargetPlatforms) > 0 {
+		platforms := metric.Must(m).NewInt64ValueRecorder("cli-platforms", metric.WithDescription("The build target platforms specified via CLI flag --platform"))
+		platforms.Record(ctx, 1, append(labels, attribute.String("os_arch", meter.CliBuildTargetPlatforms))...)
+	}
+
+	if len(meter.DeployNodePlatforms) > 0 {
+		platforms := metric.Must(m).NewInt64ValueRecorder("node-platforms", metric.WithDescription("The kubernetes cluster node platforms"))
+		platforms.Record(ctx, 1, append(labels, attribute.String("os_arch", meter.DeployNodePlatforms))...)
 	}
 }
 
diff --git a/pkg/skaffold/instrumentation/meter.go b/pkg/skaffold/instrumentation/meter.go
index 3827881e8..2a114161c 100644
--- a/pkg/skaffold/instrumentation/meter.go
+++ b/pkg/skaffold/instrumentation/meter.go
@@ -39,18 +39,19 @@ const (
 
 var (
 	meter = skaffoldMeter{
-		OS:                runtime.GOOS,
-		Arch:              runtime.GOARCH,
-		EnumFlags:         map[string]string{},
-		Builders:          map[string]int{},
-		BuildDependencies: map[string]int{},
-		SyncType:          map[string]bool{},
-		Hooks:             map[HookPhase]int{},
-		DevIterations:     []devIteration{},
-		StartTime:         time.Now(),
-		Version:           version.Get().Version,
-		ExitCode:          0,
-		ErrorCode:         proto.StatusCode_OK,
+		OS:                 runtime.GOOS,
+		Arch:               runtime.GOARCH,
+		EnumFlags:          map[string]string{},
+		Builders:           map[string]int{},
+		BuildDependencies:  map[string]int{},
+		BuildWithPlatforms: map[string]int{},
+		SyncType:           map[string]bool{},
+		Hooks:              map[HookPhase]int{},
+		DevIterations:      []devIteration{},
+		StartTime:          time.Now(),
+		Version:            version.Get().Version,
+		ExitCode:           0,
+		ErrorCode:          proto.StatusCode_OK,
 	}
 	MeteredCommands     = stringset.New()
 	doesBuild           = stringset.New()
@@ -93,6 +94,9 @@ func InitMeterFromConfig(configs []*latestV1.SkaffoldConfig, user, deployCtx str
 			if len(artifact.Dependencies) > 0 {
 				meter.BuildDependencies[yamltags.GetYamlTag(artifact.ArtifactType)]++
 			}
+			if len(config.Pipeline.Build.Platforms) > 0 || len(artifact.Platforms) > 0 {
+				meter.BuildWithPlatforms[yamltags.GetYamlTag(artifact.ArtifactType)]++
+			}
 			meter.Hooks[HookPhases.PreBuild] += len(artifact.LifecycleHooks.PreHooks)
 			meter.Hooks[HookPhases.PostBuild] += len(artifact.LifecycleHooks.PostHooks)
 
@@ -145,6 +149,23 @@ func AddFlag(flag *flag.Flag) {
 	}
 }
 
+func AddResolvedBuildTargetPlatforms(platforms string) {
+	platforms = strings.ReplaceAll(platforms, ",", ";")
+	if !stringslice.Contains(meter.ResolvedBuildTargetPlatforms, platforms) {
+		meter.ResolvedBuildTargetPlatforms = append(meter.ResolvedBuildTargetPlatforms, platforms)
+	}
+}
+
+func AddCliBuildTargetPlatforms(platforms string) {
+	platforms = strings.ReplaceAll(platforms, ",", ";")
+	meter.CliBuildTargetPlatforms = platforms
+}
+
+func AddDeployNodePlatforms(platforms string) {
+	platforms = strings.ReplaceAll(platforms, ",", ";")
+	meter.DeployNodePlatforms = platforms
+}
+
 func getClusterType(deployCtx string) string {
 	if strings.HasPrefix(deployCtx, "gke_") {
 		return gke
diff --git a/pkg/skaffold/instrumentation/types.go b/pkg/skaffold/instrumentation/types.go
index 80bbc6b37..507066ccf 100644
--- a/pkg/skaffold/instrumentation/types.go
+++ b/pkg/skaffold/instrumentation/types.go
@@ -64,6 +64,9 @@ type skaffoldMeter struct {
 	// Builders Enum values for all the builders used to build the artifacts built.
 	Builders map[string]int
 
+	// BuildWithPlatforms Enum values for all the builders with target platform constraints.
+	BuildWithPlatforms map[string]int
+
 	// BuildDependencies Enum values for all the builders using build dependencies.
 	BuildDependencies map[string]int
 
@@ -92,6 +95,15 @@ type skaffoldMeter struct {
 
 	// ClusterType reports if user cluster is a GKE cluster or not.
 	ClusterType string
+
+	// ResolvedBuildTargetPlatforms represents the set of resolved build target platforms for each pipeline
+	ResolvedBuildTargetPlatforms []string
+
+	// CliBuildTargetPlatforms represents the build target platforms specified via command line flag `--platform`
+	CliBuildTargetPlatforms string
+
+	// DeployNodePlatforms represents the set of kubernetes cluster node platforms
+	DeployNodePlatforms string
 }
 
 // devIteration describes how an iteration and started and if an error happened.
diff --git a/pkg/skaffold/kubernetes/manifest/images.go b/pkg/skaffold/kubernetes/manifest/images.go
index 3ccd7954d..2a4d2044d 100644
--- a/pkg/skaffold/kubernetes/manifest/images.go
+++ b/pkg/skaffold/kubernetes/manifest/images.go
@@ -20,16 +20,85 @@ import (
 	"context"
 	"strconv"
 
+	apimachinery "k8s.io/apimachinery/pkg/runtime/schema"
+
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/docker"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/instrumentation"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/output/log"
+	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 )
 
+type ResourceSelectorImages struct {
+	allowlist map[apimachinery.GroupKind]latestV1.ResourceFilter
+	denylist  map[apimachinery.GroupKind]latestV1.ResourceFilter
+}
+
+func NewResourceSelectorImages(allowlist map[apimachinery.GroupKind]latestV1.ResourceFilter, denylist map[apimachinery.GroupKind]latestV1.ResourceFilter) *ResourceSelectorImages {
+	return &ResourceSelectorImages{
+		allowlist: allowlist,
+		denylist:  denylist,
+	}
+}
+
+func (rsi *ResourceSelectorImages) allowByGroupKind(gk apimachinery.GroupKind) bool {
+	if _, allowed := rsi.allowlist[gk]; allowed {
+		// TODO(aaron-prindle) see if it makes sense to make this only use the allowlist...
+		if rf, disallowed := rsi.denylist[gk]; disallowed {
+			for _, s := range rf.Labels {
+				if s == ".*" {
+					return false
+				}
+			}
+			for _, s := range rf.Image {
+				if s == ".*" {
+					return false
+				}
+			}
+		}
+		return true
+	}
+	return false
+}
+
+func (rsi *ResourceSelectorImages) allowByNavpath(gk apimachinery.GroupKind, navpath string, k string) (string, bool) {
+	for _, w := range ConfigConnectorResourceSelector {
+		if w.Matches(gk.Group, gk.Kind) {
+			return "", true
+		}
+	}
+
+	if rf, ok := rsi.denylist[gk]; ok {
+		for _, denypath := range rf.Image {
+			if denypath == ".*" {
+				return "", false
+			}
+			if navpath == denypath {
+				return "", false
+			}
+		}
+	}
+
+	if rf, ok := rsi.allowlist[gk]; ok {
+		for _, allowpath := range rf.Image {
+			if allowpath == ".*" {
+				if k != "image" {
+					return "", false
+				}
+				return "", true
+			}
+			if navpath == allowpath {
+				return "", true
+			}
+		}
+	}
+	return "", false
+}
+
 // GetImages gathers a map of base image names to the image with its tag
-func (l *ManifestList) GetImages() ([]graph.Artifact, error) {
+func (l *ManifestList) GetImages(rs ResourceSelector) ([]graph.Artifact, error) {
 	s := &imageSaver{}
-	_, err := l.Visit(s)
+	_, err := l.Visit(s, rs)
 	return s.Images, parseImagesInManifestErr(err)
 }
 
@@ -37,7 +106,7 @@ type imageSaver struct {
 	Images []graph.Artifact
 }
 
-func (is *imageSaver) Visit(navpath string, o map[string]interface{}, k string, v interface{}) bool {
+func (is *imageSaver) Visit(gk apimachinery.GroupKind, navpath string, o map[string]interface{}, k string, v interface{}, rs ResourceSelector) bool {
 	if k != "image" {
 		return true
 	}
@@ -61,17 +130,17 @@ func (is *imageSaver) Visit(navpath string, o map[string]interface{}, k string,
 
 // ReplaceImages replaces image names in a list of manifests.
 // It doesn't replace images that are referenced by digest.
-func (l *ManifestList) ReplaceImages(ctx context.Context, builds []graph.Artifact) (ManifestList, error) {
-	return l.replaceImages(ctx, builds, selectLocalManifestImages)
+func (l *ManifestList) ReplaceImages(ctx context.Context, builds []graph.Artifact, rs ResourceSelector) (ManifestList, error) {
+	return l.replaceImages(ctx, builds, selectLocalManifestImages, rs)
 }
 
 // ReplaceRemoteManifestImages replaces all image names in a list containing remote manifests.
 // This will even override images referenced by digest or with a different repository
-func (l *ManifestList) ReplaceRemoteManifestImages(ctx context.Context, builds []graph.Artifact) (ManifestList, error) {
-	return l.replaceImages(ctx, builds, selectRemoteManifestImages)
+func (l *ManifestList) ReplaceRemoteManifestImages(ctx context.Context, builds []graph.Artifact, rs ResourceSelector) (ManifestList, error) {
+	return l.replaceImages(ctx, builds, selectRemoteManifestImages, rs)
 }
 
-func (l *ManifestList) replaceImages(ctx context.Context, builds []graph.Artifact, selector imageSelector) (ManifestList, error) {
+func (l *ManifestList) replaceImages(ctx context.Context, builds []graph.Artifact, selector imageSelector, rs ResourceSelector) (ManifestList, error) {
 	_, endTrace := instrumentation.StartTrace(ctx, "ReplaceImages", map[string]string{
 		"manifestEntries":   strconv.Itoa(len(*l)),
 		"numImagesReplaced": strconv.Itoa(len(builds)),
@@ -80,7 +149,7 @@ func (l *ManifestList) replaceImages(ctx context.Context, builds []graph.Artifac
 
 	replacer := newImageReplacer(builds, selector)
 
-	updated, err := l.Visit(replacer)
+	updated, err := l.Visit(replacer, rs)
 	if err != nil {
 		endTrace(instrumentation.TraceEndError(err))
 		return nil, replaceImageErr(err)
@@ -112,8 +181,8 @@ func newImageReplacer(builds []graph.Artifact, selector imageSelector) *imageRep
 	}
 }
 
-func (r *imageReplacer) Visit(navpath string, o map[string]interface{}, k string, v interface{}) bool {
-	if k != "image" {
+func (r *imageReplacer) Visit(gk apimachinery.GroupKind, navpath string, o map[string]interface{}, k string, v interface{}, rs ResourceSelector) bool {
+	if _, ok := rs.allowByNavpath(gk, navpath, k); !ok {
 		return true
 	}
 
@@ -121,6 +190,7 @@ func (r *imageReplacer) Visit(navpath string, o map[string]interface{}, k string
 	if !ok {
 		return true
 	}
+
 	parsed, err := docker.ParseReference(image)
 	if err != nil {
 		log.Entry(context.TODO()).Debugf("Couldn't parse image [%s]: %s", image, err.Error())
diff --git a/pkg/skaffold/kubernetes/manifest/labels.go b/pkg/skaffold/kubernetes/manifest/labels.go
index 28076a365..dec25cfd4 100644
--- a/pkg/skaffold/kubernetes/manifest/labels.go
+++ b/pkg/skaffold/kubernetes/manifest/labels.go
@@ -18,18 +18,100 @@ package manifest
 
 import (
 	"context"
+	"strings"
+
+	apimachinery "k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/output/log"
+	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 )
 
+type ResourceSelectorLabels struct {
+	allowlist map[apimachinery.GroupKind]latestV1.ResourceFilter
+	denylist  map[apimachinery.GroupKind]latestV1.ResourceFilter
+}
+
+func NewResourceSelectorLabels(allowlist map[apimachinery.GroupKind]latestV1.ResourceFilter, denylist map[apimachinery.GroupKind]latestV1.ResourceFilter) *ResourceSelectorLabels {
+	return &ResourceSelectorLabels{
+		allowlist: allowlist,
+		denylist:  denylist,
+	}
+}
+
+func (rsi *ResourceSelectorLabels) allowByGroupKind(gk apimachinery.GroupKind) bool {
+	if _, allowed := rsi.allowlist[gk]; allowed {
+		// TODO(aaron-prindle) see if it makes sense to make this only use the allowlist...
+		if rf, disallowed := rsi.denylist[gk]; disallowed {
+			for _, s := range rf.Labels {
+				if s == ".*" {
+					return false
+				}
+			}
+			for _, s := range rf.Image {
+				if s == ".*" {
+					return false
+				}
+			}
+		}
+		return true
+	}
+	return false
+}
+
+func (rsi *ResourceSelectorLabels) allowByNavpath(gk apimachinery.GroupKind, navpath string, k string) (string, bool) {
+	for _, w := range ConfigConnectorResourceSelector {
+		if w.Matches(gk.Group, gk.Kind) {
+			return "labels", true
+		}
+	}
+
+	if rf, ok := rsi.denylist[gk]; ok {
+		for _, denypath := range rf.Labels {
+			if denypath == ".*" {
+				return "", false
+			}
+			// truncate the last part of the labels path and see if this matches
+			lastDot := strings.LastIndex(denypath, ".")
+			if lastDot == -1 {
+				// no dot exists in denypath, denypath is invalid
+				break
+			}
+			if navpath == denypath[:lastDot] {
+				return "", false
+			}
+		}
+	}
+
+	if rf, ok := rsi.allowlist[gk]; ok {
+		for _, allowpath := range rf.Labels {
+			if allowpath == ".*" {
+				if k != "metadata" {
+					return "", false
+				}
+				return "labels", true
+			}
+			// truncate the last part of the labels path and see if this matches
+			lastDot := strings.LastIndex(allowpath, ".")
+			if lastDot == -1 {
+				// no dot exists in allowpath, allowpath is invalid
+				continue
+			}
+			if navpath == allowpath[:lastDot] {
+				return allowpath[lastDot+1:], true
+			}
+		}
+	}
+	return "", false
+}
+
 // SetLabels add labels to a list of Kubernetes manifests.
-func (l *ManifestList) SetLabels(labels map[string]string) (ManifestList, error) {
+func (l *ManifestList) SetLabels(labels map[string]string, rs ResourceSelector) (ManifestList, error) {
 	if len(labels) == 0 {
 		return *l, nil
 	}
 
 	replacer := newLabelsSetter(labels)
-	updated, err := l.Visit(replacer)
+	updated, err := l.Visit(replacer, rs)
 	if err != nil {
 		return nil, labelSettingErr(err)
 	}
@@ -49,37 +131,34 @@ func newLabelsSetter(labels map[string]string) *labelsSetter {
 	}
 }
 
-func (r *labelsSetter) Visit(navpath string, o map[string]interface{}, k string, v interface{}) bool {
-	if k != "metadata" {
+func (r *labelsSetter) Visit(gk apimachinery.GroupKind, navpath string, o map[string]interface{}, k string, v interface{}, rs ResourceSelector) bool {
+	labelsField, ok := rs.allowByNavpath(gk, navpath, k)
+	if !ok {
 		return true
 	}
 
 	if len(r.labels) == 0 {
 		return false
 	}
-
 	metadata, ok := v.(map[string]interface{})
 	if !ok {
 		return true
 	}
 
-	l, present := metadata["labels"]
+	l, present := metadata[labelsField]
 	if !present {
-		metadata["labels"] = r.labels
+		metadata[labelsField] = r.labels
 		return false
 	}
-
 	labels, ok := l.(map[string]interface{})
 	if !ok {
 		return true
 	}
-
 	for k, v := range r.labels {
 		// Don't overwrite existing labels
 		if _, present := labels[k]; !present {
 			labels[k] = v
 		}
 	}
-
 	return false
 }
diff --git a/pkg/skaffold/kubernetes/manifest/namespaces.go b/pkg/skaffold/kubernetes/manifest/namespaces.go
index fe7d3a66c..026685a10 100644
--- a/pkg/skaffold/kubernetes/manifest/namespaces.go
+++ b/pkg/skaffold/kubernetes/manifest/namespaces.go
@@ -20,13 +20,19 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
 // CollectNamespaces returns all the namespaces in the manifests.
 func (l *ManifestList) CollectNamespaces() ([]string, error) {
 	replacer := newNamespaceCollector()
 
-	if _, err := l.Visit(replacer); err != nil {
+	// TODO(aaron-prindle) make sure this is ok?
+	rs := &ResourceSelectorImages{}
+	if _, err := l.Visit(replacer, rs); err != nil {
+		// if _, err := l.Visit(replacer, make(map[schema.GroupKind]latestV1.ResourceFilter), make(map[schema.GroupKind]latestV1.ResourceFilter)); err != nil {
+		// TODO(aaron-prindle) verify this doesn't need to support allow/deny list, also see if 'nil' is better option for unused inputs
 		return nil, fmt.Errorf("collecting namespaces: %w", err)
 	}
 
@@ -48,7 +54,7 @@ func newNamespaceCollector() *namespaceCollector {
 	}
 }
 
-func (r *namespaceCollector) Visit(navpath string, o map[string]interface{}, k string, v interface{}) bool {
+func (r *namespaceCollector) Visit(gk schema.GroupKind, navpath string, o map[string]interface{}, k string, v interface{}, rs ResourceSelector) bool {
 	if k != "metadata" {
 		return true
 	}
diff --git a/pkg/skaffold/kubernetes/manifest/visitor.go b/pkg/skaffold/kubernetes/manifest/visitor.go
index c9db7e8a5..0ae639bd8 100644
--- a/pkg/skaffold/kubernetes/manifest/visitor.go
+++ b/pkg/skaffold/kubernetes/manifest/visitor.go
@@ -18,43 +18,129 @@ package manifest
 
 import (
 	"fmt"
-	"path"
 
 	apimachinery "k8s.io/apimachinery/pkg/runtime/schema"
 
+	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/yaml"
 )
 
-// transformableAllowlist is the set of kinds that can be transformed by Skaffold.
-var transformableAllowlist = map[apimachinery.GroupKind]bool{
-	{Group: "", Kind: "Pod"}:                                true,
-	{Group: "apps", Kind: "DaemonSet"}:                      true,
-	{Group: "apps", Kind: "Deployment"}:                     true, // v1beta1, v1beta2: deprecated in K8s 1.9, removed in 1.16
-	{Group: "apps", Kind: "ReplicaSet"}:                     true,
-	{Group: "apps", Kind: "StatefulSet"}:                    true,
-	{Group: "batch", Kind: "CronJob"}:                       true,
-	{Group: "batch", Kind: "Job"}:                           true,
-	{Group: "extensions", Kind: "DaemonSet"}:                true, // v1beta1: deprecated in K8s 1.9, removed in 1.16
-	{Group: "extensions", Kind: "Deployment"}:               true, // v1beta1: deprecated in K8s 1.9, removed in 1.16
-	{Group: "extensions", Kind: "ReplicaSet"}:               true, // v1beta1: deprecated in K8s 1.9, removed in 1.16
-	{Group: "serving.knative.dev", Kind: "Service"}:         true,
-	{Group: "agones.dev", Kind: "Fleet"}:                    true,
-	{Group: "agones.dev", Kind: "GameServer"}:               true,
-	{Group: "argoproj.io", Kind: "Rollout"}:                 true,
-	{Group: "argoproj.io", Kind: "ClusterWorkflowTemplate"}: true,
-	{Group: "argoproj.io", Kind: "Workflow"}:                true,
-	{Group: "argoproj.io", Kind: "WorkflowTemplate"}:        true,
+type ResourceSelector interface {
+	allowByGroupKind(apimachinery.GroupKind) bool
+	allowByNavpath(apimachinery.GroupKind, string, string) (string, bool)
+}
+
+// TransformAllowlist is the default allowlist of kinds that can be transformed by Skaffold.
+var TransformAllowlist = map[apimachinery.GroupKind]latestV1.ResourceFilter{
+	{Group: "", Kind: "Pod"}: {
+		GroupKind: "Pod",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "", Kind: "Service"}: {
+		GroupKind: "Service",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "apps", Kind: "DaemonSet"}: {
+		GroupKind: "DaemonSet.apps",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "apps", Kind: "Deployment"}: {
+		GroupKind: "Deployment.apps",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "apps", Kind: "ReplicaSet"}: {
+		GroupKind: "ReplicaSet.apps",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "apps", Kind: "StatefulSet"}: {
+		GroupKind: "StatefulSet.apps",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "batch", Kind: "CronJob"}: {
+		GroupKind: "CronJob.batch",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "batch", Kind: "Job"}: {
+		GroupKind: "Job.batch",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "extensions", Kind: "DaemonSet"}: {
+		GroupKind: "DaemonSet.extensions",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "extensions", Kind: "Deployment"}: {
+		GroupKind: "Deployment.extensions",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "extensions", Kind: "ReplicaSet"}: {
+		GroupKind: "ReplicaSet.extensions",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "serving.knative.dev", Kind: "Service"}: {
+		GroupKind: "Service.serving.knative.dev",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "agones.dev", Kind: "Fleet"}: {
+		GroupKind: "Fleet.agones.dev",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "agones.dev", Kind: "GameServer"}: {
+		GroupKind: "GameServer.agones.dev",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "argoproj.io", Kind: "Rollout"}: {
+		GroupKind: "Rollout.argoproj.io",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "argoproj.io", Kind: "ClusterWorkflowTemplate"}: {
+		GroupKind: "ClusterWorkflowTemplate.argoproj.io",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "argoproj.io", Kind: "Workflow"}: {
+		GroupKind: "Workflow.argoproj.io",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+	{Group: "argoproj.io", Kind: "WorkflowTemplate"}: {
+		GroupKind: "WorkflowTemplate.argoproj.io",
+		Image:     []string{".*"},
+		Labels:    []string{".*"},
+	},
+}
+
+// TransformDenylist is the default denylist on the set of kinds that can be transformed by Skaffold.
+var TransformDenylist = map[apimachinery.GroupKind]latestV1.ResourceFilter{
+	{Group: "apps", Kind: "StatefulSet"}: {
+		GroupKind: "StatefulSet.apps",
+		Labels:    []string{".spec.volumeClaimTemplates.metadata.labels"},
+	},
 }
 
 // FieldVisitor represents the aggregation/transformation that should be performed on each traversed field.
 type FieldVisitor interface {
 	// Visit is called for each transformable key contained in the object and may apply transformations/aggregations on it.
 	// It should return true to allow recursive traversal or false when the entry was transformed.
-	Visit(path string, object map[string]interface{}, key string, value interface{}) bool
+	Visit(gk apimachinery.GroupKind, navpath string, object map[string]interface{}, key string, value interface{}, rs ResourceSelector) bool
 }
 
 // Visit recursively visits all transformable object fields within the manifests and lets the visitor apply transformations/aggregations on them.
-func (l *ManifestList) Visit(visitor FieldVisitor) (ManifestList, error) {
+func (l *ManifestList) Visit(visitor FieldVisitor, rs ResourceSelector) (ManifestList, error) {
 	var updated ManifestList
 
 	for _, manifest := range *l {
@@ -67,7 +153,7 @@ func (l *ManifestList) Visit(visitor FieldVisitor) (ManifestList, error) {
 			continue
 		}
 
-		traverseManifestFields(m, visitor)
+		traverseManifestFields(m, visitor, rs)
 
 		updatedManifest, err := yaml.Marshal(m)
 		if err != nil {
@@ -81,14 +167,30 @@ func (l *ManifestList) Visit(visitor FieldVisitor) (ManifestList, error) {
 }
 
 // traverseManifest traverses all transformable fields contained within the manifest.
-func traverseManifestFields(manifest map[string]interface{}, visitor FieldVisitor) {
-	if shouldTransformManifest(manifest) {
+func traverseManifestFields(manifest map[string]interface{}, visitor FieldVisitor, rs ResourceSelector) {
+	var groupKind apimachinery.GroupKind
+	var apiVersion string
+	if value, ok := manifest["apiVersion"].(string); ok {
+		apiVersion = value
+	}
+	var kind string
+	if value, ok := manifest["kind"].(string); ok {
+		kind = value
+	}
+
+	gvk := apimachinery.FromAPIVersionAndKind(apiVersion, kind)
+	groupKind = apimachinery.GroupKind{
+		Group: gvk.Group,
+		Kind:  gvk.Kind,
+	}
+
+	if shouldTransformManifest(manifest, rs) {
 		visitor = &recursiveVisitorDecorator{visitor}
 	}
-	visitFields("/", manifest, visitor)
+	visitFields(groupKind, "", manifest, visitor, rs)
 }
 
-func shouldTransformManifest(manifest map[string]interface{}) bool {
+func shouldTransformManifest(manifest map[string]interface{}, rs ResourceSelector) bool {
 	var apiVersion string
 	switch value := manifest["apiVersion"].(type) {
 	case string:
@@ -111,14 +213,16 @@ func shouldTransformManifest(manifest map[string]interface{}) bool {
 		Kind:  gvk.Kind,
 	}
 
-	if result, found := transformableAllowlist[groupKind]; found {
-		return result
+	if rs.allowByGroupKind(groupKind) {
+		return true
 	}
+
 	for _, w := range ConfigConnectorResourceSelector {
 		if w.Matches(gvk.Group, gvk.Kind) {
 			return true
 		}
 	}
+
 	return false
 }
 
@@ -127,29 +231,25 @@ type recursiveVisitorDecorator struct {
 	delegate FieldVisitor
 }
 
-func (d *recursiveVisitorDecorator) Visit(path string, o map[string]interface{}, k string, v interface{}) bool {
-	if d.delegate.Visit(path, o, k, v) {
-		visitFields(path, v, d)
+func (d *recursiveVisitorDecorator) Visit(gk apimachinery.GroupKind, navpath string, o map[string]interface{}, k string, v interface{}, rs ResourceSelector) bool {
+	if d.delegate.Visit(gk, navpath, o, k, v, rs) {
+		visitFields(gk, navpath, v, d, rs)
 	}
 	return false
 }
 
 // visitFields traverses all fields and calls the visitor for each.
-// navpath: a '/' delimited path representing the fields navigated to this point
-func visitFields(navpath string, o interface{}, visitor FieldVisitor) {
+// navpath: a '.' delimited path representing the fields navigated to this point
+func visitFields(gk apimachinery.GroupKind, navpath string, o interface{}, visitor FieldVisitor, rs ResourceSelector) {
 	switch entries := o.(type) {
 	case []interface{}:
 		for _, v := range entries {
 			// this case covers lists so we don't update the navpath
-			visitFields(navpath, v, visitor)
+			visitFields(gk, navpath, v, visitor, rs)
 		}
 	case map[string]interface{}:
 		for k, v := range entries {
-			// TODO(6416) temporary fix for StatefulSet + PVC use case, need to do something similar to the proposal in #6236 for full fix
-			if navpath == "/spec/volumeClaimTemplates" {
-				continue
-			}
-			visitor.Visit(path.Join(navpath, k), entries, k, v)
+			visitor.Visit(gk, navpath+"."+k, entries, k, v, rs)
 		}
 	}
 }
diff --git a/pkg/skaffold/platform/resolver.go b/pkg/skaffold/platform/resolver.go
index 9c6e9a054..344354cdf 100644
--- a/pkg/skaffold/platform/resolver.go
+++ b/pkg/skaffold/platform/resolver.go
@@ -25,6 +25,7 @@ import (
 	coreV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/config"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/instrumentation"
 	kubernetesclient "github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/client"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/output/log"
 	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
@@ -58,6 +59,7 @@ func NewResolver(ctx context.Context, pipelines []latestV1.Pipeline, cliPlatform
 		return r, fmt.Errorf("failed to parse platforms: %w", err)
 	}
 	log.Entry(ctx).Debugf("CLI platforms provided: %q", fromCli)
+	instrumentation.AddCliBuildTargetPlatforms(fromCli.String())
 
 	if runMode == config.RunModes.Dev || runMode == config.RunModes.Debug || runMode == config.RunModes.Run {
 		fromClusterNodes, err = getClusterPlatforms(ctx, kubeContext, runMode == config.RunModes.Dev || runMode == config.RunModes.Debug)
@@ -67,7 +69,7 @@ func NewResolver(ctx context.Context, pipelines []latestV1.Pipeline, cliPlatform
 		}
 	}
 	log.Entry(ctx).Debugf("platforms detected from active kubernetes cluster nodes: %q", fromClusterNodes)
-
+	instrumentation.AddDeployNodePlatforms(fromClusterNodes.String())
 	for _, pipeline := range pipelines {
 		platforms := fromCli
 		if platforms.IsEmpty() {
@@ -85,6 +87,7 @@ func NewResolver(ctx context.Context, pipelines []latestV1.Pipeline, cliPlatform
 				return r, fmt.Errorf("build target platforms %q do not match active kubernetes cluster node platforms %q", platforms, fromClusterNodes)
 			}
 		}
+		instrumentation.AddResolvedBuildTargetPlatforms(platforms.String())
 		for _, artifact := range pipeline.Build.Artifacts {
 			pl := platforms
 			constraints, err := Parse(artifact.Platforms)
diff --git a/pkg/skaffold/render/renderer/renderer.go b/pkg/skaffold/render/renderer/renderer.go
index d25e33163..60284bd0b 100644
--- a/pkg/skaffold/render/renderer/renderer.go
+++ b/pkg/skaffold/render/renderer/renderer.go
@@ -139,11 +139,13 @@ func (r *SkaffoldRenderer) Render(ctx context.Context, out io.Writer, builds []g
 	if err != nil {
 		return err
 	}
-	manifests, err = manifests.ReplaceImages(ctx, builds)
+	// TODO(aaron-prindle) wire proper transform allow/deny list args when going to V2
+	manifests, err = manifests.ReplaceImages(ctx, builds, manifest.NewResourceSelectorImages(manifest.TransformAllowlist, manifest.TransformDenylist))
 	if err != nil {
 		return err
 	}
-	manifests.SetLabels(r.labels)
+	// TODO(aaron-prindle) wire proper transform allow/deny list args when going to V2
+	manifests.SetLabels(r.labels, manifest.NewResourceSelectorLabels(manifest.TransformAllowlist, manifest.TransformDenylist))
 
 	// cache the dry manifests to the temp directory. manifests.yaml will be truncated if already exists.
 	dryConfigPath := filepath.Join(r.hydrationDir, dryFileName)
diff --git a/pkg/skaffold/runner/deployer.go b/pkg/skaffold/runner/deployer.go
index 7210d7791..435c85571 100644
--- a/pkg/skaffold/runner/deployer.go
+++ b/pkg/skaffold/runner/deployer.go
@@ -135,8 +135,11 @@ func GetDeployer(ctx context.Context, runCtx *runcontext.RunContext, labeller *l
 		}
 
 		if d.KptDeploy != nil {
-			deployer := kpt.NewDeployer(dCtx, labeller, d.KptDeploy)
-			deployers = append(deployers, deployer)
+			k, err := kpt.NewDeployer(dCtx, labeller, d.KptDeploy)
+			if err != nil {
+				return nil, err
+			}
+			deployers = append(deployers, k)
 		}
 
 		if d.KubectlDeploy != nil {
diff --git a/pkg/skaffold/runner/runcontext/context.go b/pkg/skaffold/runner/runcontext/context.go
index 1f88caf11..02d59dec3 100644
--- a/pkg/skaffold/runner/runcontext/context.go
+++ b/pkg/skaffold/runner/runcontext/context.go
@@ -107,15 +107,26 @@ func (ps Pipelines) TestCases() []*latestV1.TestCase {
 	return tests
 }
 
-// TransformableAllowList returns combined allowlist from pipelines
-func (ps Pipelines) TransformableAllowList() []latestV1.ResourceFilter {
-	var allowList []latestV1.ResourceFilter
+// TransformAllowList returns combined allowlist from pipelines
+func (ps Pipelines) TransformAllowList() []latestV1.ResourceFilter {
+	var allowlist []latestV1.ResourceFilter
 	for _, p := range ps.pipelines {
-		if p.Deploy.TransformableAllowList != nil {
-			allowList = append(allowList, p.Deploy.TransformableAllowList...)
+		if p.ResourceSelector.Allow != nil {
+			allowlist = append(allowlist, p.ResourceSelector.Allow...)
 		}
 	}
-	return allowList
+	return allowlist
+}
+
+// TransformDenyList returns combined denylist from pipelines
+func (ps Pipelines) TransformDenyList() []latestV1.ResourceFilter {
+	var denylist []latestV1.ResourceFilter
+	for _, p := range ps.pipelines {
+		if p.ResourceSelector.Deny != nil {
+			denylist = append(denylist, p.ResourceSelector.Deny...)
+		}
+	}
+	return denylist
 }
 
 func (ps Pipelines) StatusCheckDeadlineSeconds() int {
@@ -166,8 +177,12 @@ func (rc *RunContext) IsTestPhaseActive() bool {
 	return !rc.SkipTests() && len(rc.TestCases()) != 0
 }
 
-func (rc *RunContext) TransformableAllowList() []latestV1.ResourceFilter {
-	return rc.Pipelines.TransformableAllowList()
+func (rc *RunContext) TransformAllowList() []latestV1.ResourceFilter {
+	return rc.Pipelines.TransformAllowList()
+}
+
+func (rc *RunContext) TransformDenyList() []latestV1.ResourceFilter {
+	return rc.Pipelines.TransformDenyList()
 }
 
 // AddSkaffoldLabels tells the Runner whether to add skaffold-specific labels.
@@ -227,6 +242,7 @@ func (rc *RunContext) GetRunID() string                              { return rc
 func (rc *RunContext) RPCPort() *int                                 { return rc.Opts.RPCPort.Value() }
 func (rc *RunContext) RPCHTTPPort() *int                             { return rc.Opts.RPCHTTPPort.Value() }
 func (rc *RunContext) PushImages() config.BoolOrUndefined            { return rc.Opts.PushImages }
+func (rc *RunContext) TransformRulesFile() string                    { return rc.Opts.TransformRulesFile }
 func (rc *RunContext) JSONParseConfig() latestV1.JSONParseConfig {
 	return rc.DefaultPipeline().Deploy.Logs.JSONParse
 }


Test patch:
diff --git a/pkg/skaffold/deploy/deploy_problems_test.go b/pkg/skaffold/deploy/deploy_problems_test.go
index 2ea3548ad..f7d555cc8 100644
--- a/pkg/skaffold/deploy/deploy_problems_test.go
+++ b/pkg/skaffold/deploy/deploy_problems_test.go
@@ -108,18 +108,20 @@ type mockConfig struct {
 	kubeContext string
 }
 
-func (m mockConfig) MinikubeProfile() string                           { return m.minikube }
-func (m mockConfig) GetPipelines() []latestV1.Pipeline                 { return []latestV1.Pipeline{} }
-func (m mockConfig) GetWorkingDir() string                             { return "" }
-func (m mockConfig) GetNamespace() string                              { return "" }
-func (m mockConfig) GlobalConfig() string                              { return "" }
-func (m mockConfig) ConfigurationFile() string                         { return "" }
-func (m mockConfig) DefaultRepo() *string                              { return &m.minikube }
-func (m mockConfig) MultiLevelRepo() *bool                             { return nil }
-func (m mockConfig) SkipRender() bool                                  { return true }
-func (m mockConfig) Prune() bool                                       { return true }
-func (m mockConfig) ContainerDebugging() bool                          { return false }
-func (m mockConfig) GetKubeContext() string                            { return m.kubeContext }
-func (m mockConfig) GetInsecureRegistries() map[string]bool            { return map[string]bool{} }
-func (m mockConfig) Mode() config.RunMode                              { return config.RunModes.Dev }
-func (m mockConfig) TransformableAllowList() []latestV1.ResourceFilter { return nil }
+func (m mockConfig) MinikubeProfile() string                       { return m.minikube }
+func (m mockConfig) GetPipelines() []latestV1.Pipeline             { return []latestV1.Pipeline{} }
+func (m mockConfig) GetWorkingDir() string                         { return "" }
+func (m mockConfig) GetNamespace() string                          { return "" }
+func (m mockConfig) GlobalConfig() string                          { return "" }
+func (m mockConfig) ConfigurationFile() string                     { return "" }
+func (m mockConfig) DefaultRepo() *string                          { return &m.minikube }
+func (m mockConfig) MultiLevelRepo() *bool                         { return nil }
+func (m mockConfig) SkipRender() bool                              { return true }
+func (m mockConfig) Prune() bool                                   { return true }
+func (m mockConfig) ContainerDebugging() bool                      { return false }
+func (m mockConfig) GetKubeContext() string                        { return m.kubeContext }
+func (m mockConfig) GetInsecureRegistries() map[string]bool        { return map[string]bool{} }
+func (m mockConfig) Mode() config.RunMode                          { return config.RunModes.Dev }
+func (m mockConfig) TransformAllowList() []latestV1.ResourceFilter { return nil }
+func (m mockConfig) TransformDenyList() []latestV1.ResourceFilter  { return nil }
+func (m mockConfig) TransformRulesFile() string                    { return "" }
diff --git a/pkg/skaffold/deploy/kpt/kpt_test.go b/pkg/skaffold/deploy/kpt/kpt_test.go
index 0c185280d..8d9ee193c 100644
--- a/pkg/skaffold/deploy/kpt/kpt_test.go
+++ b/pkg/skaffold/deploy/kpt/kpt_test.go
@@ -234,7 +234,11 @@ func TestKpt_Deploy(t *testing.T) {
 			t.Override(&client.Client, deployutil.MockK8sClient)
 			t.NewTempDir().Chdir()
 
-			k := NewDeployer(&kptConfig{}, &label.DefaultLabeller{}, &test.kpt)
+			var err error
+			k, err := NewDeployer(&kptConfig{}, &label.DefaultLabeller{}, &test.kpt)
+			if err != nil {
+				t.Fatalf("unexpected error occurred attempting to create new kpt deployer")
+			}
 			if test.hasKustomization != nil {
 				k.hasKustomization = test.hasKustomization
 			}
@@ -245,7 +249,7 @@ func TestKpt_Deploy(t *testing.T) {
 				t.CheckNoError(os.Mkdir(k.Live.Apply.Dir, 0755))
 			}
 
-			err := k.Deploy(context.Background(), ioutil.Discard, test.builds)
+			err = k.Deploy(context.Background(), ioutil.Discard, test.builds)
 			t.CheckError(test.shouldErr, err)
 		})
 	}
@@ -377,7 +381,10 @@ func TestKpt_Dependencies(t *testing.T) {
 			tmpDir.WriteFiles(test.createFiles)
 			tmpDir.WriteFiles(test.kustomizations)
 
-			k := NewDeployer(&kptConfig{}, &label.DefaultLabeller{}, &test.kpt)
+			k, err := NewDeployer(&kptConfig{}, &label.DefaultLabeller{}, &test.kpt)
+			if err != nil {
+				t.Fatalf("unexpected error occurred in NewDeployer: %v", err)
+			}
 
 			res, err := k.Dependencies()
 
@@ -428,7 +435,7 @@ func TestKpt_Cleanup(t *testing.T) {
 				t.CheckNoError(os.Mkdir(test.applyDir, 0755))
 			}
 
-			k := NewDeployer(&kptConfig{
+			k, err := NewDeployer(&kptConfig{
 				workingDir: ".",
 			}, &label.DefaultLabeller{}, &latestV1.KptDeploy{
 				Live: latestV1.KptLive{
@@ -437,8 +444,11 @@ func TestKpt_Cleanup(t *testing.T) {
 					},
 				},
 			})
+			if err != nil {
+				t.Fatalf("unexpected error occurred in NewDeployer: %v", err)
+			}
 
-			err := k.Cleanup(context.Background(), ioutil.Discard, false)
+			err = k.Cleanup(context.Background(), ioutil.Discard, false)
 
 			t.CheckError(test.shouldErr, err)
 		})
@@ -792,13 +802,17 @@ spec:
 
 			labeller := label.NewLabeller(false, test.labels, "")
 
-			k := NewDeployer(&kptConfig{workingDir: "."}, labeller, &test.kpt)
+			k, err := NewDeployer(&kptConfig{workingDir: "."}, labeller, &test.kpt)
+			if err != nil {
+				t.Fatalf("unexpected error occurred in NewDeployer: %v", err)
+			}
+
 			if test.hasKustomization != nil {
 				k.hasKustomization = test.hasKustomization
 			}
 
 			var b bytes.Buffer
-			err := k.Render(context.Background(), &b, test.builds, true, "")
+			err = k.Render(context.Background(), &b, test.builds, true, "")
 
 			t.CheckErrorAndDeepEqual(test.shouldErr, err, test.expected, b.String())
 		})
@@ -867,11 +881,14 @@ func TestKpt_GetApplyDir(t *testing.T) {
 				tmpDir.Touch(".kpt-hydrated/inventory-template.yaml")
 			}
 
-			k := NewDeployer(&kptConfig{
+			k, err := NewDeployer(&kptConfig{
 				workingDir: ".",
 			}, &label.DefaultLabeller{}, &latestV1.KptDeploy{
 				Live: test.live,
 			})
+			if err != nil {
+				t.Fatalf("unexpected error occurred in NewDeployer: %v", err)
+			}
 
 			applyDir, err := k.getApplyDir(context.Background())
 
@@ -1028,7 +1045,11 @@ spec:
 	}
 	for _, test := range tests {
 		testutil.Run(t, test.description, func(t *testutil.T) {
-			k := NewDeployer(&kptConfig{}, &label.DefaultLabeller{}, nil)
+			k, err := NewDeployer(&kptConfig{}, &label.DefaultLabeller{}, nil)
+			if err != nil {
+				t.Fatalf("unexpected error occurred in NewDeployer: %v", err)
+			}
+
 			actualManifest, err := k.excludeKptFn(test.manifests)
 			t.CheckErrorAndDeepEqual(false, err, test.expected.String(), actualManifest.String())
 		})
@@ -1175,7 +1196,7 @@ func TestNonEmptyKubeconfig(t *testing.T) {
 	testutil.Run(t, "", func(t *testutil.T) {
 		t.Override(&util.DefaultExecCommand, commands)
 		t.Override(&client.Client, deployutil.MockK8sClient)
-		k := NewDeployer(&kptConfig{config: "testConfigPath"}, &label.DefaultLabeller{}, &latestV1.KptDeploy{
+		k, err := NewDeployer(&kptConfig{config: "testConfigPath"}, &label.DefaultLabeller{}, &latestV1.KptDeploy{
 			Dir: ".",
 			Live: latestV1.KptLive{
 				Apply: latestV1.KptApplyInventory{
@@ -1183,9 +1204,13 @@ func TestNonEmptyKubeconfig(t *testing.T) {
 				},
 			},
 		})
+		if err != nil {
+			t.Fatalf("unexpected error occurred in NewDeployer: %v", err)
+		}
+
 		t.CheckNoError(os.Mkdir(k.Live.Apply.Dir, 0755))
 		defer os.RemoveAll(k.Live.Apply.Dir)
-		err := k.Deploy(context.Background(), ioutil.Discard, []graph.Artifact{})
+		err = k.Deploy(context.Background(), ioutil.Discard, []graph.Artifact{})
 		t.CheckNoError(err)
 	})
 }
diff --git a/pkg/skaffold/deploy/util/util_test.go b/pkg/skaffold/deploy/util/util_test.go
index b9aca0b7e..dec3f8c7b 100644
--- a/pkg/skaffold/deploy/util/util_test.go
+++ b/pkg/skaffold/deploy/util/util_test.go
@@ -20,9 +20,14 @@ import (
 	"testing"
 
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/graph"
 	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/kubernetes/manifest"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/runner/runcontext"
+	latestV1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
+	"github.com/GoogleContainerTools/skaffold/pkg/skaffold/yaml"
 	"github.com/GoogleContainerTools/skaffold/testutil"
 )
 
@@ -180,3 +185,128 @@ func TestAddTagsToPodSelector(t *testing.T) {
 		})
 	}
 }
+
+func TestConsolidateTransformConfiguration(t *testing.T) {
+	tests := []struct {
+		description           string
+		shouldErr             bool
+		allowSchemaTransforms []latestV1.ResourceFilter
+		denySchemaTransforms  []latestV1.ResourceFilter
+		flagTransforms        latestV1.ResourceSelectorConfig
+		expected              func(map[schema.GroupKind]latestV1.ResourceFilter, map[schema.GroupKind]latestV1.ResourceFilter) (map[schema.GroupKind]latestV1.ResourceFilter, map[schema.GroupKind]latestV1.ResourceFilter)
+	}{
+		{
+			description: "verify schema transform configuration outprioritizes default hardcoded transform configuration",
+			denySchemaTransforms: []latestV1.ResourceFilter{
+				{
+					GroupKind: "Deployment.apps",
+				},
+			},
+			expected: func(allow map[schema.GroupKind]latestV1.ResourceFilter, deny map[schema.GroupKind]latestV1.ResourceFilter) (map[schema.GroupKind]latestV1.ResourceFilter, map[schema.GroupKind]latestV1.ResourceFilter) {
+				// Deployment.apps removed from hardcoded allowlist
+				delete(allow, schema.GroupKind{Group: "apps", Kind: "Deployment"})
+				// Deployment.apps added to denylist
+				deny[schema.GroupKind{Group: "apps", Kind: "Deployment"}] = latestV1.ResourceFilter{GroupKind: "Deployment.apps"}
+				return allow, deny
+			},
+		},
+		{
+			description: "verify flag transform configuration outprioritizes schema transform configuration",
+			flagTransforms: latestV1.ResourceSelectorConfig{
+				Allow: []latestV1.ResourceFilter{
+					{
+						GroupKind: "Test.skaffold.dev",
+					},
+				},
+			},
+			denySchemaTransforms: []latestV1.ResourceFilter{
+				{
+					GroupKind: "Test.skaffold.dev",
+				},
+			},
+			expected: func(allow map[schema.GroupKind]latestV1.ResourceFilter, deny map[schema.GroupKind]latestV1.ResourceFilter) (map[schema.GroupKind]latestV1.ResourceFilter, map[schema.GroupKind]latestV1.ResourceFilter) {
+				// Test.skaffold.dev added to allowlist as flag config outprioritizes schema config
+				allow[schema.GroupKind{Group: "skaffold.dev", Kind: "Test"}] = latestV1.ResourceFilter{GroupKind: "Test.skaffold.dev"}
+				return allow, deny
+			},
+		},
+		{
+			description: "verify denylist outprioritizes allowlist transform configuration (for same config input source)",
+			flagTransforms: latestV1.ResourceSelectorConfig{
+				Allow: []latestV1.ResourceFilter{
+					{
+						GroupKind: "Test.skaffold.dev",
+					},
+				},
+				Deny: []latestV1.ResourceFilter{
+					{
+						GroupKind: "Test.skaffold.dev",
+					},
+				},
+			},
+			expected: func(allow map[schema.GroupKind]latestV1.ResourceFilter, deny map[schema.GroupKind]latestV1.ResourceFilter) (map[schema.GroupKind]latestV1.ResourceFilter, map[schema.GroupKind]latestV1.ResourceFilter) {
+				// Test.skaffold.dev added to denylist as deny config outprioritizes allow config for same priority config source (both flag config)
+				deny[schema.GroupKind{Group: "skaffold.dev", Kind: "Test"}] = latestV1.ResourceFilter{GroupKind: "Test.skaffold.dev"}
+				return allow, deny
+			},
+		},
+	}
+	for _, test := range tests {
+		testutil.Run(t, test.description, func(t *testutil.T) {
+			// convert flagTransform struct to yaml
+			buf, err := yaml.Marshal(test.flagTransforms)
+			if err != nil {
+				t.Fatalf("error marshalling flagTransforms test inputs: %v", err)
+			}
+
+			// denybuf, err := yaml.Marshal(test.denyFlagTransforms)
+			// if err != nil {
+			// t.Fatalf("error marshalling denyFlagTransforms test inputs: %v", err)
+			// }
+
+			flagTransformYAMLFile := t.TempFile("TestConsolidateTransformConfiguration", buf)
+
+			cfg := &mockDeployConfig{
+				transformAllowList: test.allowSchemaTransforms,
+				transformDenyList:  test.denySchemaTransforms,
+				transformRulesFile: flagTransformYAMLFile,
+			}
+			allowlist, denylist, err := ConsolidateTransformConfiguration(cfg)
+			t.CheckError(test.shouldErr, err)
+
+			copyAllow := map[schema.GroupKind]latestV1.ResourceFilter{}
+			for k, v := range manifest.TransformAllowlist {
+				copyAllow[k] = v
+			}
+
+			copyDeny := map[schema.GroupKind]latestV1.ResourceFilter{}
+			for k, v := range manifest.TransformDenylist {
+				copyDeny[k] = v
+			}
+			expectedAllowlist, expectedDenyList := test.expected(copyAllow, copyDeny)
+			t.CheckDeepEqual(expectedAllowlist, allowlist)
+			t.CheckDeepEqual(expectedDenyList, denylist)
+		})
+	}
+}
+
+type mockDeployConfig struct {
+	runcontext.RunContext // Embedded to provide the default values.
+	transformAllowList    []latestV1.ResourceFilter
+	transformDenyList     []latestV1.ResourceFilter
+	transformRulesFile    string
+}
+
+func (c *mockDeployConfig) ForceDeploy() bool                                     { return false }
+func (c *mockDeployConfig) GetKubeConfig() string                                 { return "" }
+func (c *mockDeployConfig) GetKubeContext() string                                { return "" }
+func (c *mockDeployConfig) GetKubeNamespace() string                              { return "" }
+func (c *mockDeployConfig) ConfigurationFile() string                             { return "" }
+func (c *mockDeployConfig) PortForwardResources() []*latestV1.PortForwardResource { return nil }
+func (c *mockDeployConfig) TransformAllowList() []latestV1.ResourceFilter {
+	return c.transformAllowList
+}
+func (c *mockDeployConfig) TransformDenyList() []latestV1.ResourceFilter {
+	return c.transformDenyList
+}
+func (c *mockDeployConfig) TransformRulesFile() string { return c.transformRulesFile }
diff --git a/pkg/skaffold/instrumentation/export_test.go b/pkg/skaffold/instrumentation/export_test.go
index 7fff45bea..3a32eca37 100644
--- a/pkg/skaffold/instrumentation/export_test.go
+++ b/pkg/skaffold/instrumentation/export_test.go
@@ -46,52 +46,63 @@ var testKey = `{
 func TestExportMetrics(t *testing.T) {
 	startTime, _ := time.Parse(time.ANSIC, "Mon Jan 2 15:04:05 -0700 MST 2006")
 	buildMeter := skaffoldMeter{
-		Command:           "build",
-		BuildArtifacts:    5,
-		Version:           "vTest.0",
-		Arch:              "test arch",
-		OS:                "test os",
-		ConfigCount:       1,
-		Deployers:         []string{"test kubectl"},
-		Builders:          map[string]int{"docker": 1, "buildpacks": 1},
-		BuildDependencies: map[string]int{"docker": 1},
-		EnumFlags:         map[string]string{"test": "test_value"},
-		StartTime:         startTime,
-		Duration:          time.Minute,
+		Command:                      "build",
+		BuildArtifacts:               5,
+		Version:                      "vTest.0",
+		Arch:                         "test arch",
+		OS:                           "test os",
+		ConfigCount:                  1,
+		Deployers:                    []string{"test kubectl"},
+		Builders:                     map[string]int{"docker": 1, "buildpacks": 1},
+		BuildWithPlatforms:           map[string]int{"docker": 1},
+		BuildDependencies:            map[string]int{"docker": 1},
+		EnumFlags:                    map[string]string{"test": "test_value"},
+		DeployNodePlatforms:          "linux/amd64",
+		CliBuildTargetPlatforms:      "linux/amd64;linux/arm64",
+		ResolvedBuildTargetPlatforms: []string{"linux/amd64"},
+		StartTime:                    startTime,
+		Duration:                     time.Minute,
 	}
 	devMeter := skaffoldMeter{
-		Command:           "dev",
-		Version:           "vTest.1",
-		Arch:              "test arch 2",
-		OS:                "test os 2",
-		ConfigCount:       2,
-		PlatformType:      "test platform 1:test platform 2",
-		Deployers:         []string{"test helm", "test kpt"},
-		SyncType:          map[string]bool{"manual": true},
-		EnumFlags:         map[string]string{"test_run": "test_run_value"},
-		Builders:          map[string]int{"kustomize": 3, "buildpacks": 2},
-		BuildDependencies: map[string]int{"docker": 1},
-		HelmReleasesCount: 2,
-		DevIterations:     []devIteration{{"sync", 0}, {"build", 400}, {"build", 0}, {"sync", 200}, {"deploy", 0}},
-		ErrorCode:         proto.StatusCode_BUILD_CANCELLED,
-		StartTime:         startTime.Add(time.Hour * 24 * 30),
-		Duration:          time.Minute * 2,
+		Command:                      "dev",
+		Version:                      "vTest.1",
+		Arch:                         "test arch 2",
+		OS:                           "test os 2",
+		ConfigCount:                  2,
+		PlatformType:                 "test platform 1:test platform 2",
+		Deployers:                    []string{"test helm", "test kpt"},
+		SyncType:                     map[string]bool{"manual": true},
+		EnumFlags:                    map[string]string{"test_run": "test_run_value"},
+		Builders:                     map[string]int{"kustomize": 3, "buildpacks": 2},
+		BuildWithPlatforms:           map[string]int{"kustomize": 3},
+		BuildDependencies:            map[string]int{"docker": 1},
+		DeployNodePlatforms:          "linux/amd64",
+		CliBuildTargetPlatforms:      "linux/amd64;linux/arm64",
+		ResolvedBuildTargetPlatforms: []string{"linux/amd64"},
+		HelmReleasesCount:            2,
+		DevIterations:                []devIteration{{"sync", 0}, {"build", 400}, {"build", 0}, {"sync", 200}, {"deploy", 0}},
+		ErrorCode:                    proto.StatusCode_BUILD_CANCELLED,
+		StartTime:                    startTime.Add(time.Hour * 24 * 30),
+		Duration:                     time.Minute * 2,
 	}
 	debugMeter := skaffoldMeter{
-		Command:       "debug",
-		Version:       "vTest.2",
-		Arch:          "test arch 1",
-		OS:            "test os 2",
-		ConfigCount:   2,
-		PlatformType:  "test platform",
-		Deployers:     []string{"test helm", "test kpt"},
-		SyncType:      map[string]bool{"manual": true, "sync": true},
-		EnumFlags:     map[string]string{"test_run": "test_run_value"},
-		Builders:      map[string]int{"jib": 3, "buildpacks": 2},
-		DevIterations: []devIteration{{"build", 104}, {"build", 0}, {"sync", 0}, {"deploy", 1014}},
-		ErrorCode:     proto.StatusCode_BUILD_CANCELLED,
-		StartTime:     startTime.Add(time.Hour * 24 * 10),
-		Duration:      time.Minute * 4,
+		Command:                      "debug",
+		Version:                      "vTest.2",
+		Arch:                         "test arch 1",
+		OS:                           "test os 2",
+		ConfigCount:                  2,
+		PlatformType:                 "test platform",
+		Deployers:                    []string{"test helm", "test kpt"},
+		SyncType:                     map[string]bool{"manual": true, "sync": true},
+		EnumFlags:                    map[string]string{"test_run": "test_run_value"},
+		Builders:                     map[string]int{"jib": 3, "buildpacks": 2},
+		DeployNodePlatforms:          "linux/amd64",
+		CliBuildTargetPlatforms:      "linux/amd64;linux/arm64",
+		ResolvedBuildTargetPlatforms: []string{"linux/amd64"},
+		DevIterations:                []devIteration{{"build", 104}, {"build", 0}, {"sync", 0}, {"deploy", 1014}},
+		ErrorCode:                    proto.StatusCode_BUILD_CANCELLED,
+		StartTime:                    startTime.Add(time.Hour * 24 * 10),
+		Duration:                     time.Minute * 4,
 	}
 	metersBytes, _ := json.Marshal([]skaffoldMeter{buildMeter, devMeter, debugMeter})
 	fs := &testutil.FakeFileSystem{
@@ -389,12 +400,16 @@ func checkOutput(t *testutil.T, meters []skaffoldMeter, b []byte) {
 	commandCount := make(map[interface{}]int)
 	errorCount := make(map[interface{}]int)
 	builders := make(map[interface{}]int)
+	buildersWithPlatforms := make(map[interface{}]int)
 	buildDeps := make(map[interface{}]int)
 	helmReleases := make(map[interface{}]int)
 	devIterations := make(map[interface{}]int)
 	deployers := make(map[interface{}]int)
 	enumFlags := make(map[interface{}]int)
 	platform := make(map[interface{}]int)
+	buildPlatforms := make(map[interface{}]int)
+	cliPlatforms := make(map[interface{}]int)
+	nodePlatforms := make(map[interface{}]int)
 
 	testMaps := []map[interface{}]int{
 		platform, osCount, versionCount, archCount, durationCount, commandCount, errorCount, builders, devIterations, deployers}
@@ -417,6 +432,9 @@ func checkOutput(t *testutil.T, meters []skaffoldMeter, b []byte) {
 			for k, v := range meter.Builders {
 				builders[k] += v
 			}
+			for k, v := range meter.BuildWithPlatforms {
+				buildersWithPlatforms[k] += v
+			}
 			for k, v := range meter.BuildDependencies {
 				buildDeps[k] += v
 			}
@@ -452,6 +470,8 @@ func checkOutput(t *testutil.T, meters []skaffoldMeter, b []byte) {
 			durationCount[fmt.Sprintf("%s:%f", l.Labels["command"], l.value().(float64))]--
 		case "artifacts":
 			builders[l.Labels["builder"]] -= int(l.value().(float64)) - 1
+		case "artifact-with-platforms":
+			buildersWithPlatforms[l.Labels["builder"]] -= int(l.value().(float64)) - 1
 		case "artifact-dependencies":
 			buildDeps[l.Labels["builder"]] -= int(l.value().(float64)) - 1
 		case "builders":
@@ -468,6 +488,12 @@ func checkOutput(t *testutil.T, meters []skaffoldMeter, b []byte) {
 			enumFlags[l.Labels["flag_name"]+":"+l.Labels["value"]]--
 		case "helmReleases":
 			helmReleases[l.Labels["helmReleases"]]++
+		case "build-platforms":
+			buildPlatforms[l.Labels["build-platforms"]]++
+		case "node-platforms":
+			nodePlatforms[l.Labels["node-platforms"]]++
+		case "cli-platforms":
+			cliPlatforms[l.Labels["cli-platforms"]]++
 		default:
 			switch {
 			case MeteredCommands.Contains(l.Name):
diff --git a/pkg/skaffold/kubernetes/manifest/images_test.go b/pkg/skaffold/kubernetes/manifest/images_test.go
index 97d963d58..7ca1cf5dc 100644
--- a/pkg/skaffold/kubernetes/manifest/images_test.go
+++ b/pkg/skaffold/kubernetes/manifest/images_test.go
@@ -55,7 +55,7 @@ spec:
 		},
 	}
 
-	actual, err := manifests.GetImages()
+	actual, err := manifests.GetImages(NewResourceSelectorImages(TransformAllowlist, TransformDenylist))
 	testutil.CheckErrorAndDeepEqual(t, false, err, expectedImages, actual)
 }
 
@@ -121,7 +121,7 @@ spec:
 		fakeWarner := &warnings.Collect{}
 		t.Override(&warnings.Printf, fakeWarner.Warnf)
 
-		resultManifest, err := manifests.ReplaceRemoteManifestImages(context.TODO(), builds)
+		resultManifest, err := manifests.ReplaceRemoteManifestImages(context.TODO(), builds, NewResourceSelectorImages(TransformAllowlist, TransformDenylist))
 
 		t.CheckNoError(err)
 		t.CheckDeepEqual(expected.String(), resultManifest.String())
@@ -196,7 +196,7 @@ spec:
 		fakeWarner := &warnings.Collect{}
 		t.Override(&warnings.Printf, fakeWarner.Warnf)
 
-		resultManifest, err := manifests.ReplaceImages(context.TODO(), builds)
+		resultManifest, err := manifests.ReplaceImages(context.TODO(), builds, NewResourceSelectorImages(TransformAllowlist, TransformDenylist))
 
 		t.CheckNoError(err)
 		t.CheckDeepEqual(expected.String(), resultManifest.String())
@@ -207,7 +207,7 @@ func TestReplaceEmptyManifest(t *testing.T) {
 	manifests := ManifestList{[]byte(""), []byte("  ")}
 	expected := ManifestList{}
 
-	resultManifest, err := manifests.ReplaceImages(context.TODO(), nil)
+	resultManifest, err := manifests.ReplaceImages(context.TODO(), nil, NewResourceSelectorImages(TransformAllowlist, TransformDenylist))
 
 	testutil.CheckErrorAndDeepEqual(t, false, err, expected.String(), resultManifest.String())
 }
@@ -215,7 +215,7 @@ func TestReplaceEmptyManifest(t *testing.T) {
 func TestReplaceInvalidManifest(t *testing.T) {
 	manifests := ManifestList{[]byte("INVALID")}
 
-	_, err := manifests.ReplaceImages(context.TODO(), nil)
+	_, err := manifests.ReplaceImages(context.TODO(), nil, NewResourceSelectorImages(TransformAllowlist, TransformDenylist))
 
 	testutil.CheckError(t, true, err)
 }
@@ -228,7 +228,7 @@ image:
 - value2
 `)}
 
-	output, err := manifests.ReplaceImages(context.TODO(), nil)
+	output, err := manifests.ReplaceImages(context.TODO(), nil, NewResourceSelectorImages(TransformAllowlist, TransformDenylist))
 
 	testutil.CheckErrorAndDeepEqual(t, false, err, manifests.String(), output.String())
 }
diff --git a/pkg/skaffold/kubernetes/manifest/labels_test.go b/pkg/skaffold/kubernetes/manifest/labels_test.go
index 803f762e0..1dc19ca11 100644
--- a/pkg/skaffold/kubernetes/manifest/labels_test.go
+++ b/pkg/skaffold/kubernetes/manifest/labels_test.go
@@ -19,6 +19,9 @@ package manifest
 import (
 	"testing"
 
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	v1 "github.com/GoogleContainerTools/skaffold/pkg/skaffold/schema/latest/v1"
 	"github.com/GoogleContainerTools/skaffold/testutil"
 )
 
@@ -51,7 +54,7 @@ spec:
 	resultManifest, err := manifests.SetLabels(map[string]string{
 		"key1": "value1",
 		"key2": "value2",
-	})
+	}, NewResourceSelectorLabels(TransformAllowlist, TransformDenylist))
 
 	testutil.CheckErrorAndDeepEqual(t, false, err, expected.String(), resultManifest.String())
 }
@@ -89,7 +92,7 @@ spec:
 		"key0": "should-be-ignored",
 		"key1": "value1",
 		"key2": "value2",
-	})
+	}, NewResourceSelectorLabels(TransformAllowlist, TransformDenylist))
 
 	testutil.CheckErrorAndDeepEqual(t, false, err, expected.String(), resultManifest.String())
 }
@@ -117,7 +120,7 @@ spec:
     name: example
 `)}
 
-	resultManifest, err := manifests.SetLabels(nil)
+	resultManifest, err := manifests.SetLabels(nil, NewResourceSelectorLabels(TransformAllowlist, TransformDenylist))
 
 	testutil.CheckErrorAndDeepEqual(t, false, err, expected.String(), resultManifest.String())
 }
@@ -136,7 +139,7 @@ metadata:
   name: getting-started
 `)}
 
-	resultManifest, err := manifests.SetLabels(map[string]string{"key0": "value0"})
+	resultManifest, err := manifests.SetLabels(map[string]string{"key0": "value0"}, NewResourceSelectorLabels(TransformAllowlist, TransformDenylist))
 
 	testutil.CheckErrorAndDeepEqual(t, false, err, manifests.String(), resultManifest.String())
 }
@@ -184,7 +187,13 @@ spec:
 	resultManifest, err := manifests.SetLabels(map[string]string{
 		"key0": "value0",
 		"key1": "value1",
-	})
+	}, NewResourceSelectorLabels(map[schema.GroupKind]v1.ResourceFilter{
+		{Group: "apiextensions.k8s.io", Kind: "CustomResourceDefinition"}: {
+			GroupKind: "CustomResourceDefinition.apiextensions.k8s.io",
+			Image:     []string{".*"},
+			Labels:    []string{".metadata.labels"},
+		},
+	}, nil))
 
 	testutil.CheckErrorAndDeepEqual(t, false, err, expected.String(), resultManifest.String())
 }
diff --git a/pkg/skaffold/kubernetes/manifest/visitor_test.go b/pkg/skaffold/kubernetes/manifest/visitor_test.go
index ae89145ae..45a622069 100644
--- a/pkg/skaffold/kubernetes/manifest/visitor_test.go
+++ b/pkg/skaffold/kubernetes/manifest/visitor_test.go
@@ -21,6 +21,8 @@ import (
 	"regexp"
 	"testing"
 
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
 	"github.com/GoogleContainerTools/skaffold/testutil"
 )
 
@@ -30,7 +32,7 @@ type mockVisitor struct {
 	replaceWith interface{}
 }
 
-func (m *mockVisitor) Visit(navpath string, o map[string]interface{}, k string, v interface{}) bool {
+func (m *mockVisitor) Visit(gk schema.GroupKind, navpath string, o map[string]interface{}, k string, v interface{}, rs ResourceSelector) bool {
 	s := fmt.Sprintf("%+v", v)
 	if len(s) > 4 {
 		s = s[:4] + "..."
@@ -249,7 +251,7 @@ spec:
 	for _, test := range tests {
 		testutil.Run(t, test.description, func(t *testutil.T) {
 			visitor := &mockVisitor{map[string]int{}, test.pivotKey, test.replaceWith}
-			actual, err := test.manifests.Visit(visitor)
+			actual, err := test.manifests.Visit(visitor, NewResourceSelectorImages(TransformAllowlist, TransformDenylist))
 			expectedVisits := map[string]int{}
 			for _, visit := range test.expected {
 				expectedVisits[visit]++
@@ -369,7 +371,7 @@ func TestShouldTransformManifest(t *testing.T) {
 	}
 	for _, test := range tests {
 		testutil.Run(t, fmt.Sprintf("%v", test.manifest), func(t *testutil.T) {
-			result := shouldTransformManifest(test.manifest)
+			result := shouldTransformManifest(test.manifest, NewResourceSelectorImages(TransformAllowlist, TransformDenylist))
 			t.CheckDeepEqual(test.expected, result)
 		})
 	}
diff --git a/pkg/skaffold/runner/v1/runner_test.go b/pkg/skaffold/runner/v1/runner_test.go
index 178842704..77fe66d74 100644
--- a/pkg/skaffold/runner/v1/runner_test.go
+++ b/pkg/skaffold/runner/v1/runner_test.go
@@ -437,8 +437,12 @@ func TestNewForConfig(t *testing.T) {
 					DeployType: latestV1.DeployType{
 						KubectlDeploy: &latestV1.KubectlDeploy{},
 					},
-					TransformableAllowList: []latestV1.ResourceFilter{
-						{Type: "example.com/Application"},
+				},
+				ResourceSelector: latestV1.ResourceSelectorConfig{
+					Allow: []latestV1.ResourceFilter{
+						{
+							GroupKind: "example.com/Application",
+						},
 					},
 				},
 			},
@@ -487,9 +491,9 @@ func TestNewForConfig(t *testing.T) {
 				},
 			}
 			// Test transformableAllowList
-			filters := runCtx.TransformableAllowList()
-			if test.pipeline.Deploy.TransformableAllowList != nil {
-				t.CheckDeepEqual(test.pipeline.Deploy.TransformableAllowList, filters)
+			filters := runCtx.TransformAllowList()
+			if test.pipeline.ResourceSelector.Allow != nil {
+				t.CheckDeepEqual(test.pipeline.ResourceSelector.Allow, filters)
 			} else {
 				t.CheckEmpty(filters)
 			}
diff --git a/pkg/skaffold/schema/latest/v1/config.go b/pkg/skaffold/schema/latest/v1/config.go
index 47822ec8c..aaad55661 100644
--- a/pkg/skaffold/schema/latest/v1/config.go
+++ b/pkg/skaffold/schema/latest/v1/config.go
@@ -73,6 +73,9 @@ type Pipeline struct {
 
 	// PortForward describes user defined resources to port-forward.
 	PortForward []*PortForwardResource `yaml:"portForward,omitempty"`
+
+	// ResourceSelector describes user defined filters describing how skaffold should treat objects/fields during rendering.
+	ResourceSelector ResourceSelectorConfig `yaml:"resourceSelector,omitempty"`
 }
 
 // GitInfo contains information on the origin of skaffold configurations cloned from a git repository.
@@ -145,6 +148,14 @@ type PortForwardResource struct {
 	LocalPort int `yaml:"localPort,omitempty"`
 }
 
+// ResourceSelectorConfig contains all the configuration needed by the deploy steps.
+type ResourceSelectorConfig struct {
+	// Allow configures an allowlist for transforming manifests.
+	Allow []ResourceFilter `yaml:"allow,omitempty"`
+	// Deny configures an allowlist for transforming manifests.
+	Deny []ResourceFilter `yaml:"deny,omitempty"`
+}
+
 // BuildConfig contains all the configuration for the build steps.
 type BuildConfig struct {
 	// Artifacts lists the images you're going to be building.
@@ -526,9 +537,6 @@ type DeployConfig struct {
 
 	// Logs configures how container logs are printed as a result of a deployment.
 	Logs LogsConfig `yaml:"logs,omitempty"`
-
-	// TransformableAllowList configures an allowlist for transforming manifests.
-	TransformableAllowList []ResourceFilter `yaml:"-"`
 }
 
 // DeployType contains the specific implementation and parameters needed
@@ -1504,8 +1512,8 @@ type NamedContainerHook struct {
 
 // ResourceFilter contains definition to filter which resource to transform.
 type ResourceFilter struct {
-	// Type is the compact format of a resource type.
-	Type string `yaml:"type" yamltags:"required"`
+	// GroupKind is the compact format of a resource type.
+	GroupKind string `yaml:"groupKind" yamltags:"required"`
 	// Image is an optional slice of JSON-path-like paths of where to rewrite images.
 	Image []string `yaml:"image,omitempty"`
 	// Labels is an optional slide of JSON-path-like paths of where to add a labels block if missing.