diff --git a/cmd/clusterctl/client/alias.go b/cmd/clusterctl/client/alias.go
index 37502e1387df..4941140e01b3 100644
--- a/cmd/clusterctl/client/alias.go
+++ b/cmd/clusterctl/client/alias.go
@@ -20,6 +20,7 @@ import (
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/repository"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 )
 
 // Alias creates local aliases for types defined in the low-level libraries.
@@ -42,3 +43,7 @@ type UpgradePlan cluster.UpgradePlan
 
 // Kubeconfig is a type that specifies inputs related to the actual kubeconfig.
 type Kubeconfig cluster.Kubeconfig
+
+// Processor defines the methods necessary for creating a specific yaml
+// processor.
+type Processor yaml.Processor
diff --git a/cmd/clusterctl/client/client.go b/cmd/clusterctl/client/client.go
index 6b1d1e4fd473..7b0d9999ec36 100644
--- a/cmd/clusterctl/client/client.go
+++ b/cmd/clusterctl/client/client.go
@@ -64,8 +64,21 @@ type clusterctlClient struct {
 	clusterClientFactory    ClusterClientFactory
 }
 
-type RepositoryClientFactory func(config.Provider) (repository.Client, error)
-type ClusterClientFactory func(Kubeconfig) (cluster.Client, error)
+// RepositoryClientFactoryInput represents the inputs required by the
+// RepositoryClientFactory
+type RepositoryClientFactoryInput struct {
+	provider  Provider
+	processor Processor
+}
+type RepositoryClientFactory func(RepositoryClientFactoryInput) (repository.Client, error)
+
+// ClusterClientFactoryInput reporesents the inputs required by the
+// ClusterClientFactory
+type ClusterClientFactoryInput struct {
+	kubeconfig Kubeconfig
+	processor  Processor
+}
+type ClusterClientFactory func(ClusterClientFactoryInput) (cluster.Client, error)
 
 // Ensure clusterctlClient implements Client.
 var _ Client = &clusterctlClient{}
@@ -130,17 +143,25 @@ func newClusterctlClient(path string, options ...Option) (*clusterctlClient, err
 	return client, nil
 }
 
-// defaultClusterFactory is a ClusterClientFactory func the uses the default client provided by the cluster low level library.
-func defaultClusterFactory(configClient config.Client) func(kubeconfig Kubeconfig) (cluster.Client, error) {
-	return func(kubeconfig Kubeconfig) (cluster.Client, error) {
-		// Kubeconfig is a type alias to cluster.Kubeconfig
-		return cluster.New(cluster.Kubeconfig(kubeconfig), configClient), nil
+// defaultRepositoryFactory is a RepositoryClientFactory func the uses the default client provided by the repository low level library.
+func defaultRepositoryFactory(configClient config.Client) RepositoryClientFactory {
+	return func(input RepositoryClientFactoryInput) (repository.Client, error) {
+		return repository.New(
+			input.provider,
+			configClient,
+			repository.InjectYamlProcessor(input.processor),
+		)
 	}
 }
 
-// defaultRepositoryFactory is a RepositoryClientFactory func the uses the default client provided by the repository low level library.
-func defaultRepositoryFactory(configClient config.Client) func(providerConfig config.Provider) (repository.Client, error) {
-	return func(providerConfig config.Provider) (repository.Client, error) {
-		return repository.New(providerConfig, configClient)
+// defaultClusterFactory is a ClusterClientFactory func the uses the default client provided by the cluster low level library.
+func defaultClusterFactory(configClient config.Client) ClusterClientFactory {
+	return func(input ClusterClientFactoryInput) (cluster.Client, error) {
+		return cluster.New(
+			// Kubeconfig is a type alias to cluster.Kubeconfig
+			cluster.Kubeconfig(input.kubeconfig),
+			configClient,
+			cluster.InjectYamlProcessor(input.processor),
+		), nil
 	}
 }
diff --git a/cmd/clusterctl/client/client_test.go b/cmd/clusterctl/client/client_test.go
index cde4ca3e0c9a..ba2cfed67bcf 100644
--- a/cmd/clusterctl/client/client_test.go
+++ b/cmd/clusterctl/client/client_test.go
@@ -29,6 +29,7 @@ import (
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/cluster"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/repository"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/scheme"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
@@ -118,11 +119,11 @@ func newFakeClient(configClient config.Client) *fakeClient {
 		fake.configClient = newFakeConfig()
 	}
 
-	var clusterClientFactory = func(i Kubeconfig) (cluster.Client, error) {
+	var clusterClientFactory = func(i ClusterClientFactoryInput) (cluster.Client, error) {
 		// converting the client.Kubeconfig to cluster.Kubeconfig alias
-		k := cluster.Kubeconfig(i)
+		k := cluster.Kubeconfig(i.kubeconfig)
 		if _, ok := fake.clusters[k]; !ok {
-			return nil, errors.Errorf("Cluster for kubeconfig %q and/or context %q does not exists.", i.Path, i.Context)
+			return nil, errors.Errorf("Cluster for kubeconfig %q and/or context %q does not exist.", i.kubeconfig.Path, i.kubeconfig.Context)
 		}
 		return fake.clusters[k], nil
 	}
@@ -130,11 +131,11 @@ func newFakeClient(configClient config.Client) *fakeClient {
 	fake.internalClient, _ = newClusterctlClient("fake-config",
 		InjectConfig(fake.configClient),
 		InjectClusterClientFactory(clusterClientFactory),
-		InjectRepositoryFactory(func(provider config.Provider) (repository.Client, error) {
-			if _, ok := fake.repositories[provider.ManifestLabel()]; !ok {
-				return nil, errors.Errorf("Repository for kubeconfig %q does not exists.", provider.ManifestLabel())
+		InjectRepositoryFactory(func(input RepositoryClientFactoryInput) (repository.Client, error) {
+			if _, ok := fake.repositories[input.provider.ManifestLabel()]; !ok {
+				return nil, errors.Errorf("Repository for kubeconfig %q does not exist.", input.provider.ManifestLabel())
 			}
-			return fake.repositories[provider.ManifestLabel()], nil
+			return fake.repositories[input.provider.ManifestLabel()], nil
 		}),
 	)
 
@@ -324,6 +325,7 @@ func newFakeRepository(provider config.Provider, configClient config.Client) *fa
 		Provider:       provider,
 		configClient:   configClient,
 		fakeRepository: fakeRepository,
+		processor:      yaml.NewSimpleProcessor(),
 	}
 }
 
@@ -331,6 +333,7 @@ type fakeRepositoryClient struct {
 	config.Provider
 	configClient   config.Client
 	fakeRepository *test.FakeRepository
+	processor      yaml.Processor
 }
 
 var _ repository.Client = &fakeRepositoryClient{}
@@ -349,6 +352,7 @@ func (f fakeRepositoryClient) Components() repository.ComponentsClient {
 		provider:       f.Provider,
 		fakeRepository: f.fakeRepository,
 		configClient:   f.configClient,
+		processor:      f.processor,
 	}
 }
 
@@ -358,6 +362,7 @@ func (f fakeRepositoryClient) Templates(version string) repository.TemplateClien
 		version:               version,
 		fakeRepository:        f.fakeRepository,
 		configVariablesClient: f.configClient.Variables(),
+		processor:             f.processor,
 	}
 }
 
@@ -399,6 +404,7 @@ type fakeTemplateClient struct {
 	version               string
 	fakeRepository        *test.FakeRepository
 	configVariablesClient config.VariablesClient
+	processor             yaml.Processor
 }
 
 func (f *fakeTemplateClient) Get(flavor, targetNamespace string, listVariablesOnly bool) (repository.Template, error) {
@@ -412,7 +418,13 @@ func (f *fakeTemplateClient) Get(flavor, targetNamespace string, listVariablesOn
 	if err != nil {
 		return nil, err
 	}
-	return repository.NewTemplate(content, f.configVariablesClient, targetNamespace, listVariablesOnly)
+	return repository.NewTemplate(repository.TemplateInput{
+		RawArtifact:           content,
+		ConfigVariablesClient: f.configVariablesClient,
+		Processor:             f.processor,
+		TargetNamespace:       targetNamespace,
+		ListVariablesOnly:     listVariablesOnly,
+	})
 }
 
 // fakeMetadataClient provides a super simple MetadataClient (e.g. without support for local overrides/embedded metadata)
@@ -441,6 +453,7 @@ type fakeComponentClient struct {
 	provider       config.Provider
 	fakeRepository *test.FakeRepository
 	configClient   config.Client
+	processor      yaml.Processor
 }
 
 func (f *fakeComponentClient) Get(options repository.ComponentsOptions) (repository.Components, error) {
@@ -454,5 +467,13 @@ func (f *fakeComponentClient) Get(options repository.ComponentsOptions) (reposit
 		return nil, err
 	}
 
-	return repository.NewComponents(f.provider, f.configClient, content, options)
+	return repository.NewComponents(
+		repository.ComponentsInput{
+			Provider:     f.provider,
+			ConfigClient: f.configClient,
+			Processor:    f.processor,
+			RawYaml:      content,
+			Options:      options,
+		},
+	)
 }
diff --git a/cmd/clusterctl/client/cluster/client.go b/cmd/clusterctl/client/cluster/client.go
index eddd052a7cdb..e2e9ddcb9055 100644
--- a/cmd/clusterctl/client/cluster/client.go
+++ b/cmd/clusterctl/client/cluster/client.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/repository"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	logf "sigs.k8s.io/cluster-api/cmd/clusterctl/log"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -97,6 +98,7 @@ type clusterClient struct {
 	proxy                   Proxy
 	repositoryClientFactory RepositoryClientFactory
 	pollImmediateWaiter     PollImmediateWaiter
+	processor               yaml.Processor
 }
 
 type RepositoryClientFactory func(provider config.Provider, configClient config.Client, options ...repository.Option) (repository.Client, error)
@@ -137,7 +139,7 @@ func (c *clusterClient) ProviderUpgrader() ProviderUpgrader {
 }
 
 func (c *clusterClient) Template() TemplateClient {
-	return newTemplateClient(c.proxy, c.configClient)
+	return newTemplateClient(TemplateClientInput{c.proxy, c.configClient, c.processor})
 }
 
 // Option is a configuration option supplied to New
@@ -165,6 +167,17 @@ func InjectPollImmediateWaiter(pollImmediateWaiter PollImmediateWaiter) Option {
 	}
 }
 
+// InjectYamlProcessor allows you to override the yaml processor that the
+// cluster client uses. By default, the SimpleProcessor is used. This is
+// true even if a nil processor is injected.
+func InjectYamlProcessor(p yaml.Processor) Option {
+	return func(c *clusterClient) {
+		if p != nil {
+			c.processor = p
+		}
+	}
+}
+
 // New returns a cluster.Client.
 func New(kubeconfig Kubeconfig, configClient config.Client, options ...Option) Client {
 	return newClusterClient(kubeconfig, configClient, options...)
@@ -174,6 +187,7 @@ func newClusterClient(kubeconfig Kubeconfig, configClient config.Client, options
 	client := &clusterClient{
 		configClient: configClient,
 		kubeconfig:   kubeconfig,
+		processor:    yaml.NewSimpleProcessor(),
 	}
 	for _, o := range options {
 		o(client)
diff --git a/cmd/clusterctl/client/cluster/client_test.go b/cmd/clusterctl/client/cluster/client_test.go
new file mode 100644
index 000000000000..f60c71ead385
--- /dev/null
+++ b/cmd/clusterctl/client/cluster/client_test.go
@@ -0,0 +1,71 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cluster
+
+import (
+	"testing"
+
+	. "github.com/onsi/gomega"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
+)
+
+func Test_newClusterClient_YamlProcessor(t *testing.T) {
+
+	tests := []struct {
+		name   string
+		opts   []Option
+		assert func(*WithT, yaml.Processor)
+	}{
+		{
+			name: "it creates a cluster client with simple yaml processor by default",
+			assert: func(g *WithT, p yaml.Processor) {
+				_, ok := (p).(*yaml.SimpleProcessor)
+				g.Expect(ok).To(BeTrue())
+			},
+		},
+		{
+			name: "it creates a cluster client with specified yaml processor",
+			opts: []Option{InjectYamlProcessor(test.NewFakeProcessor())},
+			assert: func(g *WithT, p yaml.Processor) {
+				_, ok := (p).(*yaml.SimpleProcessor)
+				g.Expect(ok).To(BeFalse())
+				_, ok = (p).(*test.FakeProcessor)
+				g.Expect(ok).To(BeTrue())
+			},
+		},
+		{
+			name: "it creates a cluster client with simple yaml processor even if injected with nil processor",
+			opts: []Option{InjectYamlProcessor(nil)},
+			assert: func(g *WithT, p yaml.Processor) {
+				g.Expect(p).ToNot(BeNil())
+				_, ok := (p).(*yaml.SimpleProcessor)
+				g.Expect(ok).To(BeTrue())
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			client := newClusterClient(Kubeconfig{}, &fakeConfigClient{}, tt.opts...)
+			g.Expect(client).ToNot(BeNil())
+			tt.assert(g, client.processor)
+		})
+	}
+}
diff --git a/cmd/clusterctl/client/cluster/template.go b/cmd/clusterctl/client/cluster/template.go
index a2d597f5a99e..6d155b0fc2e6 100644
--- a/cmd/clusterctl/client/cluster/template.go
+++ b/cmd/clusterctl/client/cluster/template.go
@@ -31,6 +31,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/repository"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -48,17 +49,25 @@ type templateClient struct {
 	proxy               Proxy
 	configClient        config.Client
 	gitHubClientFactory func(configVariablesClient config.VariablesClient) (*github.Client, error)
+	processor           yaml.Processor
 }
 
 // ensure templateClient implements TemplateClient.
 var _ TemplateClient = &templateClient{}
 
+type TemplateClientInput struct {
+	proxy        Proxy
+	configClient config.Client
+	processor    yaml.Processor
+}
+
 // newTemplateClient returns a templateClient.
-func newTemplateClient(proxy Proxy, configClient config.Client) *templateClient {
+func newTemplateClient(input TemplateClientInput) *templateClient {
 	return &templateClient{
-		proxy:               proxy,
-		configClient:        configClient,
+		proxy:               input.proxy,
+		configClient:        input.configClient,
 		gitHubClientFactory: getGitHubClient,
+		processor:           input.processor,
 	}
 }
 
@@ -90,7 +99,13 @@ func (t *templateClient) GetFromConfigMap(configMapNamespace, configMapName, con
 		return nil, errors.Errorf("the ConfigMap %s/%s does not have the %q data key", configMapNamespace, configMapName, configMapDataKey)
 	}
 
-	return repository.NewTemplate([]byte(data), t.configClient.Variables(), targetNamespace, listVariablesOnly)
+	return repository.NewTemplate(repository.TemplateInput{
+		RawArtifact:           []byte(data),
+		ConfigVariablesClient: t.configClient.Variables(),
+		Processor:             t.processor,
+		TargetNamespace:       targetNamespace,
+		ListVariablesOnly:     listVariablesOnly,
+	})
 }
 
 func (t *templateClient) GetFromURL(templateURL, targetNamespace string, listVariablesOnly bool) (repository.Template, error) {
@@ -103,7 +118,13 @@ func (t *templateClient) GetFromURL(templateURL, targetNamespace string, listVar
 		return nil, errors.Wrapf(err, "invalid GetFromURL operation")
 	}
 
-	return repository.NewTemplate(content, t.configClient.Variables(), targetNamespace, listVariablesOnly)
+	return repository.NewTemplate(repository.TemplateInput{
+		RawArtifact:           content,
+		ConfigVariablesClient: t.configClient.Variables(),
+		Processor:             t.processor,
+		TargetNamespace:       targetNamespace,
+		ListVariablesOnly:     listVariablesOnly,
+	})
 }
 
 func (t *templateClient) getURLContent(templateURL string) ([]byte, error) {
diff --git a/cmd/clusterctl/client/cluster/template_test.go b/cmd/clusterctl/client/cluster/template_test.go
index 3b140e10e9ee..c71d2ea0e0f7 100644
--- a/cmd/clusterctl/client/cluster/template_test.go
+++ b/cmd/clusterctl/client/cluster/template_test.go
@@ -33,14 +33,15 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/repository"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
 
-var template = "apiVersion: cluster.x-k8s.io/v1alpha3\n" +
-	"kind: Cluster\n" +
-	"---\n" +
-	"apiVersion: cluster.x-k8s.io/v1alpha3\n" +
-	"kind: Machine"
+var template = `apiVersion: cluster.x-k8s.io/v1alpha3
+kind: Cluster
+---
+apiVersion: cluster.x-k8s.io/v1alpha3
+kind: Machine`
 
 func Test_templateClient_GetFromConfigMap(t *testing.T) {
 	g := NewWithT(t)
@@ -133,19 +134,22 @@ func Test_templateClient_GetFromConfigMap(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 
-			tc := &templateClient{
-				proxy:        tt.fields.proxy,
-				configClient: tt.fields.configClient,
-			}
+			processor := yaml.NewSimpleProcessor()
+			tc := newTemplateClient(TemplateClientInput{tt.fields.proxy, tt.fields.configClient, processor})
 			got, err := tc.GetFromConfigMap(tt.args.configMapNamespace, tt.args.configMapName, tt.args.configMapDataKey, tt.args.targetNamespace, tt.args.listVariablesOnly)
 			if tt.wantErr {
 				g.Expect(err).To(HaveOccurred())
 				return
 			}
-
 			g.Expect(err).NotTo(HaveOccurred())
 
-			wantTemplate, err := repository.NewTemplate([]byte(tt.want), configClient.Variables(), tt.args.targetNamespace, tt.args.listVariablesOnly)
+			wantTemplate, err := repository.NewTemplate(repository.TemplateInput{
+				RawArtifact:           []byte(tt.want),
+				ConfigVariablesClient: configClient.Variables(),
+				Processor:             processor,
+				TargetNamespace:       tt.args.targetNamespace,
+				ListVariablesOnly:     tt.args.listVariablesOnly,
+			})
 			g.Expect(err).NotTo(HaveOccurred())
 			g.Expect(got).To(Equal(wantTemplate))
 		})
@@ -286,7 +290,7 @@ func Test_templateClient_GetFromURL(t *testing.T) {
 	configClient, err := config.New("", config.InjectReader(test.NewFakeReader()))
 	g.Expect(err).NotTo(HaveOccurred())
 
-	client, mux, teardown := test.NewFakeGitHub()
+	fakeGithubClient, mux, teardown := test.NewFakeGitHub()
 	defer teardown()
 
 	mux.HandleFunc("/repos/kubernetes-sigs/cluster-api/contents/config/default/cluster-template.yaml", func(w http.ResponseWriter, r *http.Request) {
@@ -340,12 +344,14 @@ func Test_templateClient_GetFromURL(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 
-			c := &templateClient{
-				configClient: configClient,
-				gitHubClientFactory: func(configVariablesClient config.VariablesClient) (*github.Client, error) {
-					return client, nil
-				},
+			gitHubClientFactory := func(configVariablesClient config.VariablesClient) (*github.Client, error) {
+				return fakeGithubClient, nil
 			}
+			processor := yaml.NewSimpleProcessor()
+			c := newTemplateClient(TemplateClientInput{nil, configClient, processor})
+			// override the github client factory
+			c.gitHubClientFactory = gitHubClientFactory
+
 			got, err := c.GetFromURL(tt.args.templateURL, tt.args.targetNamespace, tt.args.listVariablesOnly)
 			if tt.wantErr {
 				g.Expect(err).To(HaveOccurred())
@@ -354,7 +360,13 @@ func Test_templateClient_GetFromURL(t *testing.T) {
 
 			g.Expect(err).NotTo(HaveOccurred())
 
-			wantTemplate, err := repository.NewTemplate([]byte(tt.want), configClient.Variables(), tt.args.targetNamespace, tt.args.listVariablesOnly)
+			wantTemplate, err := repository.NewTemplate(repository.TemplateInput{
+				RawArtifact:           []byte(tt.want),
+				ConfigVariablesClient: configClient.Variables(),
+				Processor:             processor,
+				TargetNamespace:       tt.args.targetNamespace,
+				ListVariablesOnly:     tt.args.listVariablesOnly,
+			})
 			g.Expect(err).NotTo(HaveOccurred())
 			g.Expect(got).To(Equal(wantTemplate))
 		})
diff --git a/cmd/clusterctl/client/common.go b/cmd/clusterctl/client/common.go
index 13ac5c857aff..8ef02287f716 100644
--- a/cmd/clusterctl/client/common.go
+++ b/cmd/clusterctl/client/common.go
@@ -45,8 +45,9 @@ func (c *clusterctlClient) getComponentsByName(provider string, providerType clu
 	// Get a client for the provider repository and read the provider components;
 	// during the process, provider components will be processed performing variable substitution, customization of target
 	// and watching namespace etc.
-
-	repositoryClientFactory, err := c.repositoryClientFactory(providerConfig)
+	// Currently we are not supporting custom yaml processors for the provider
+	// components. So we revert to using the default SimpleYamlProcessor.
+	repositoryClientFactory, err := c.repositoryClientFactory(RepositoryClientFactoryInput{provider: providerConfig})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/clusterctl/client/config.go b/cmd/clusterctl/client/config.go
index 14838d80cdc3..392658368113 100644
--- a/cmd/clusterctl/client/config.go
+++ b/cmd/clusterctl/client/config.go
@@ -97,6 +97,10 @@ type GetClusterTemplateOptions struct {
 	// ListVariablesOnly sets the GetClusterTemplate method to return the list of variables expected by the template
 	// without executing any further processing.
 	ListVariablesOnly bool
+
+	// YamlProcessor defines the yaml processor to use for the cluster
+	// template processing. If not defined, SimpleProcessor will be used.
+	YamlProcessor Processor
 }
 
 // numSources return the number of template sources currently set on a GetClusterTemplateOptions.
@@ -162,7 +166,7 @@ func (c *clusterctlClient) GetClusterTemplate(options GetClusterTemplateOptions)
 	}
 
 	// Gets  the client for the current management cluster
-	cluster, err := c.clusterClientFactory(options.Kubeconfig)
+	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{options.Kubeconfig, options.YamlProcessor})
 	if err != nil {
 		return nil, err
 	}
@@ -186,7 +190,7 @@ func (c *clusterctlClient) GetClusterTemplate(options GetClusterTemplateOptions)
 
 	// Gets the workload cluster template from the selected source
 	if options.ProviderRepositorySource != nil {
-		return c.getTemplateFromRepository(cluster, *options.ProviderRepositorySource, options.TargetNamespace, options.ListVariablesOnly)
+		return c.getTemplateFromRepository(cluster, options)
 	}
 	if options.ConfigMapSource != nil {
 		return c.getTemplateFromConfigMap(cluster, *options.ConfigMapSource, options.TargetNamespace, options.ListVariablesOnly)
@@ -199,7 +203,12 @@ func (c *clusterctlClient) GetClusterTemplate(options GetClusterTemplateOptions)
 }
 
 // getTemplateFromRepository returns a workload cluster template from a provider repository.
-func (c *clusterctlClient) getTemplateFromRepository(cluster cluster.Client, source ProviderRepositorySourceOptions, targetNamespace string, listVariablesOnly bool) (Template, error) {
+func (c *clusterctlClient) getTemplateFromRepository(cluster cluster.Client, options GetClusterTemplateOptions) (Template, error) {
+	source := *options.ProviderRepositorySource
+	targetNamespace := options.TargetNamespace
+	listVariablesOnly := options.ListVariablesOnly
+	processor := options.YamlProcessor
+
 	// If the option specifying the name of the infrastructure provider to get templates from is empty, try to detect it.
 	provider := source.InfrastructureProvider
 	ensureCustomResourceDefinitions := false
@@ -253,7 +262,7 @@ func (c *clusterctlClient) getTemplateFromRepository(cluster cluster.Client, sou
 		return nil, err
 	}
 
-	repo, err := c.repositoryClientFactory(providerConfig)
+	repo, err := c.repositoryClientFactory(RepositoryClientFactoryInput{provider: providerConfig, processor: processor})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/clusterctl/client/config/client.go b/cmd/clusterctl/client/config/client.go
index 6b344f95bade..bb84b6fd12c7 100644
--- a/cmd/clusterctl/client/config/client.go
+++ b/cmd/clusterctl/client/config/client.go
@@ -18,7 +18,6 @@ package config
 
 import (
 	"github.com/pkg/errors"
-	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
 
 // Client is used to interact with the clusterctl configurations.
@@ -105,6 +104,3 @@ type Reader interface {
 	// UnmarshalKey reads a configuration value and unmarshals it into the provided value object.
 	UnmarshalKey(key string, value interface{}) error
 }
-
-// Ensures FakeReader implements the Reader interface.
-var _ Reader = &test.FakeReader{}
diff --git a/cmd/clusterctl/client/config/variables_client.go b/cmd/clusterctl/client/config/variables_client.go
index 5958bd622e41..380e066b8f95 100644
--- a/cmd/clusterctl/client/config/variables_client.go
+++ b/cmd/clusterctl/client/config/variables_client.go
@@ -16,8 +16,6 @@ limitations under the License.
 
 package config
 
-import "sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
-
 const (
 	// GitHubTokenVariable defines a variable hosting the GitHub access token
 	GitHubTokenVariable = "github-token"
@@ -35,9 +33,6 @@ type VariablesClient interface {
 	Set(key, values string)
 }
 
-// Ensures the FakeVariableClient implements VariablesClient
-var _ VariablesClient = &test.FakeVariableClient{}
-
 // variablesClient implements VariablesClient.
 type variablesClient struct {
 	reader Reader
diff --git a/cmd/clusterctl/client/config/variables_client_test.go b/cmd/clusterctl/client/config/variables_client_test.go
index 94bde37ec540..99a982dd7430 100644
--- a/cmd/clusterctl/client/config/variables_client_test.go
+++ b/cmd/clusterctl/client/config/variables_client_test.go
@@ -24,6 +24,12 @@ import (
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
 
+// Ensures FakeReader implements the Reader interface.
+var _ Reader = &test.FakeReader{}
+
+// Ensures the FakeVariableClient implements VariablesClient
+var _ VariablesClient = &test.FakeVariableClient{}
+
 func Test_variables_Get(t *testing.T) {
 	reader := test.NewFakeReader().WithVar("foo", "bar")
 
diff --git a/cmd/clusterctl/client/delete.go b/cmd/clusterctl/client/delete.go
index fcf8886c14a1..95115ba8c618 100644
--- a/cmd/clusterctl/client/delete.go
+++ b/cmd/clusterctl/client/delete.go
@@ -61,7 +61,7 @@ type DeleteOptions struct {
 }
 
 func (c *clusterctlClient) Delete(options DeleteOptions) error {
-	clusterClient, err := c.clusterClientFactory(options.Kubeconfig)
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{kubeconfig: options.Kubeconfig})
 	if err != nil {
 		return err
 	}
diff --git a/cmd/clusterctl/client/init.go b/cmd/clusterctl/client/init.go
index 1f3bad7b14d3..124bbc7319a2 100644
--- a/cmd/clusterctl/client/init.go
+++ b/cmd/clusterctl/client/init.go
@@ -71,7 +71,7 @@ func (c *clusterctlClient) Init(options InitOptions) ([]Components, error) {
 	log := logf.Log
 
 	// gets access to the management cluster
-	cluster, err := c.clusterClientFactory(options.Kubeconfig)
+	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{kubeconfig: options.Kubeconfig})
 	if err != nil {
 		return nil, err
 	}
@@ -136,7 +136,7 @@ func (c *clusterctlClient) Init(options InitOptions) ([]Components, error) {
 // Init returns the list of images required for init.
 func (c *clusterctlClient) InitImages(options InitOptions) ([]string, error) {
 	// gets access to the management cluster
-	cluster, err := c.clusterClientFactory(options.Kubeconfig)
+	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{kubeconfig: options.Kubeconfig})
 	if err != nil {
 		return nil, err
 	}
diff --git a/cmd/clusterctl/client/move.go b/cmd/clusterctl/client/move.go
index efc26e982171..d07c4ecfdfe0 100644
--- a/cmd/clusterctl/client/move.go
+++ b/cmd/clusterctl/client/move.go
@@ -33,7 +33,7 @@ type MoveOptions struct {
 
 func (c *clusterctlClient) Move(options MoveOptions) error {
 	// Get the client for interacting with the source management cluster.
-	fromCluster, err := c.clusterClientFactory(options.FromKubeconfig)
+	fromCluster, err := c.clusterClientFactory(ClusterClientFactoryInput{kubeconfig: options.FromKubeconfig})
 	if err != nil {
 		return err
 	}
@@ -44,7 +44,7 @@ func (c *clusterctlClient) Move(options MoveOptions) error {
 	}
 
 	// Get the client for interacting with the target management cluster.
-	toCluster, err := c.clusterClientFactory(options.ToKubeconfig)
+	toCluster, err := c.clusterClientFactory(ClusterClientFactoryInput{kubeconfig: options.ToKubeconfig})
 	if err != nil {
 		return err
 	}
diff --git a/cmd/clusterctl/client/repository/client.go b/cmd/clusterctl/client/repository/client.go
index 80d7104f1e6f..877b41c5fab8 100644
--- a/cmd/clusterctl/client/repository/client.go
+++ b/cmd/clusterctl/client/repository/client.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/pkg/errors"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
 
@@ -50,6 +51,7 @@ type repositoryClient struct {
 	config.Provider
 	configClient config.Client
 	repository   Repository
+	processor    yaml.Processor
 }
 
 // ensure repositoryClient implements Client.
@@ -64,7 +66,7 @@ func (c *repositoryClient) Components() ComponentsClient {
 }
 
 func (c *repositoryClient) Templates(version string) TemplateClient {
-	return newTemplateClient(c.Provider, version, c.repository, c.configClient.Variables())
+	return newTemplateClient(TemplateClientInput{version, c.Provider, c.repository, c.configClient.Variables(), c.processor})
 }
 
 func (c *repositoryClient) Metadata(version string) MetadataClient {
@@ -83,6 +85,17 @@ func InjectRepository(repository Repository) Option {
 	}
 }
 
+// InjectYamlProcessor allows you to override the yaml processor that the
+// repository client uses. By default, the SimpleProcessor is used. This is
+// true even if a nil processor is injected.
+func InjectYamlProcessor(p yaml.Processor) Option {
+	return func(c *repositoryClient) {
+		if p != nil {
+			c.processor = p
+		}
+	}
+}
+
 // New returns a Client.
 func New(provider config.Provider, configClient config.Client, options ...Option) (Client, error) {
 	return newRepositoryClient(provider, configClient, options...)
@@ -92,6 +105,7 @@ func newRepositoryClient(provider config.Provider, configClient config.Client, o
 	client := &repositoryClient{
 		Provider:     provider,
 		configClient: configClient,
+		processor:    yaml.NewSimpleProcessor(),
 	}
 	for _, o := range options {
 		o(client)
diff --git a/cmd/clusterctl/client/repository/client_test.go b/cmd/clusterctl/client/repository/client_test.go
index c7ff8f33c790..aa1e9d7b707f 100644
--- a/cmd/clusterctl/client/repository/client_test.go
+++ b/cmd/clusterctl/client/repository/client_test.go
@@ -24,6 +24,7 @@ import (
 
 	clusterctlv1 "sigs.k8s.io/cluster-api/cmd/clusterctl/api/v1alpha3"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
 
@@ -71,3 +72,57 @@ func Test_newRepositoryClient_LocalFileSystemRepository(t *testing.T) {
 		})
 	}
 }
+
+func Test_newRepositoryClient_YamlProcesor(t *testing.T) {
+	tests := []struct {
+		name   string
+		opts   []Option
+		assert func(*WithT, yaml.Processor)
+	}{
+		{
+			name: "it creates a repository client with simple yaml processor by default",
+			assert: func(g *WithT, p yaml.Processor) {
+				_, ok := (p).(*yaml.SimpleProcessor)
+				g.Expect(ok).To(BeTrue())
+			},
+		},
+		{
+			name: "it creates a repository client with specified yaml processor",
+			opts: []Option{InjectYamlProcessor(test.NewFakeProcessor())},
+			assert: func(g *WithT, p yaml.Processor) {
+				_, ok := (p).(*yaml.SimpleProcessor)
+				g.Expect(ok).To(BeFalse())
+				_, ok = (p).(*test.FakeProcessor)
+				g.Expect(ok).To(BeTrue())
+			},
+		},
+		{
+			name: "it creates a repository with simple yaml processor even if injected with nil processor",
+			opts: []Option{InjectYamlProcessor(nil)},
+			assert: func(g *WithT, p yaml.Processor) {
+				g.Expect(p).ToNot(BeNil())
+				_, ok := (p).(*yaml.SimpleProcessor)
+				g.Expect(ok).To(BeTrue())
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+			configProvider := config.NewProvider("fakeProvider", "", clusterctlv1.CoreProviderType)
+			configClient, err := config.New("", config.InjectReader(test.NewFakeReader()))
+			g.Expect(err).NotTo(HaveOccurred())
+
+			tt.opts = append(tt.opts, InjectRepository(test.NewFakeRepository()))
+
+			repoClient, err := newRepositoryClient(
+				configProvider,
+				configClient,
+				tt.opts...,
+			)
+			g.Expect(err).NotTo(HaveOccurred())
+			tt.assert(g, repoClient.processor)
+		})
+	}
+}
diff --git a/cmd/clusterctl/client/repository/components.go b/cmd/clusterctl/client/repository/components.go
index 32afee0c0629..494a990403c9 100644
--- a/cmd/clusterctl/client/repository/components.go
+++ b/cmd/clusterctl/client/repository/components.go
@@ -18,8 +18,6 @@ package repository
 
 import (
 	"fmt"
-	"regexp"
-	"sort"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -27,10 +25,10 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/util/sets"
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
 	clusterctlv1 "sigs.k8s.io/cluster-api/cmd/clusterctl/api/v1alpha3"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/scheme"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/util"
 	utilyaml "sigs.k8s.io/cluster-api/util/yaml"
@@ -52,9 +50,6 @@ const (
 	namespaceArgPrefix      = "--namespace="
 )
 
-// variableRegEx defines the regexp used for searching variables inside a YAML
-var variableRegEx = regexp.MustCompile(`\${\s*([A-Z0-9_]+)\s*}`)
-
 // Components wraps a YAML file that defines the provider components
 // to be installed in a management cluster (CRD, Controller, RBAC etc.)
 // It is important to notice that clusterctl applies a set of processing steps to the “raw” component YAML read
@@ -175,7 +170,8 @@ func (c *components) Yaml() ([]byte, error) {
 	return utilyaml.FromUnstructured(objs)
 }
 
-// ComponentsOptions is the inputs needed by the NewComponents
+// ComponentsOptions represents specific inputs that are passed in to
+// clusterctl library. These are user specified inputs.
 type ComponentsOptions struct {
 	Version           string
 	TargetNamespace   string
@@ -184,6 +180,15 @@ type ComponentsOptions struct {
 	SkipVariables bool
 }
 
+// ComponentsInput represents all the inputs required by NewComponents
+type ComponentsInput struct {
+	Provider     config.Provider
+	ConfigClient config.Client
+	Processor    yaml.Processor
+	RawYaml      []byte
+	Options      ComponentsOptions
+}
+
 // NewComponents returns a new objects embedding a component YAML file
 //
 // It is important to notice that clusterctl applies a set of processing steps to the “raw” component YAML read
@@ -194,25 +199,30 @@ type ComponentsOptions struct {
 // 4. Ensure all the ClusterRoleBinding which are referencing namespaced objects have the name prefixed with the namespace name
 // 5. Set the watching namespace for the provider controller
 // 6. Adds labels to all the components in order to allow easy identification of the provider objects
-func NewComponents(provider config.Provider, configClient config.Client, rawyaml []byte, options ComponentsOptions) (*components, error) {
-	// Inspect the yaml read from the repository for variables.
-	variables := inspectVariables(rawyaml)
+func NewComponents(input ComponentsInput) (*components, error) {
 
-	// Replace variables with corresponding values read from the config
-	yaml, err := replaceVariables(rawyaml, variables, configClient.Variables(), options.SkipVariables)
+	variables, err := input.Processor.GetVariables(input.RawYaml)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to perform variable substitution")
+		return nil, err
+	}
+
+	processedYaml := input.RawYaml
+	if !input.Options.SkipVariables {
+		processedYaml, err = input.Processor.Process(input.RawYaml, input.ConfigClient.Variables().Get)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to perform variable substitution")
+		}
 	}
 
 	// Transform the yaml in a list of objects, so following transformation can work on typed objects (instead of working on a string/slice of bytes)
-	objs, err := utilyaml.ToUnstructured(yaml)
+	objs, err := utilyaml.ToUnstructured(processedYaml)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to parse yaml")
 	}
 
 	// Apply image overrides, if defined
 	objs, err = util.FixImages(objs, func(image string) (string, error) {
-		return configClient.ImageMeta().AlterImage(provider.ManifestLabel(), image)
+		return input.ConfigClient.ImageMeta().AlterImage(input.Provider.ManifestLabel(), image)
 	})
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to apply image overrides")
@@ -243,24 +253,24 @@ func NewComponents(provider config.Provider, configClient config.Client, rawyaml
 	// if targetNamespace is not specified, then defaultTargetNamespace is used. In case both targetNamespace and defaultTargetNamespace
 	// are empty, an error is returned
 
-	if options.TargetNamespace == "" {
-		options.TargetNamespace = defaultTargetNamespace
+	if input.Options.TargetNamespace == "" {
+		input.Options.TargetNamespace = defaultTargetNamespace
 	}
 
-	if options.TargetNamespace == "" {
+	if input.Options.TargetNamespace == "" {
 		return nil, errors.New("target namespace can't be defaulted. Please specify a target namespace")
 	}
 
 	// add a Namespace object if missing (ensure the targetNamespace will be created)
-	instanceObjs = addNamespaceIfMissing(instanceObjs, options.TargetNamespace)
+	instanceObjs = addNamespaceIfMissing(instanceObjs, input.Options.TargetNamespace)
 
 	// fix Namespace name in all the objects
-	instanceObjs = fixTargetNamespace(instanceObjs, options.TargetNamespace)
+	instanceObjs = fixTargetNamespace(instanceObjs, input.Options.TargetNamespace)
 
 	// ensures all the ClusterRole and ClusterRoleBinding have the name prefixed with the namespace name and that
 	// all the clusterRole/clusterRoleBinding namespaced subjects refers to targetNamespace
 	// Nb. Making all the RBAC rules "namespaced" is required for supporting multi-tenancy
-	instanceObjs, err = fixRBAC(instanceObjs, options.TargetNamespace)
+	instanceObjs, err = fixRBAC(instanceObjs, input.Options.TargetNamespace)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to fix ClusterRoleBinding names")
 	}
@@ -273,16 +283,16 @@ func NewComponents(provider config.Provider, configClient config.Client, rawyaml
 	}
 
 	// if the requested watchingNamespace is different from the defaultWatchingNamespace, fix it
-	if defaultWatchingNamespace != options.WatchingNamespace {
-		instanceObjs, err = fixWatchNamespace(instanceObjs, options.WatchingNamespace)
+	if defaultWatchingNamespace != input.Options.WatchingNamespace {
+		instanceObjs, err = fixWatchNamespace(instanceObjs, input.Options.WatchingNamespace)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to set watching namespace")
 		}
 	}
 
 	// Add common labels to both the obj groups.
-	instanceObjs = addCommonLabels(instanceObjs, provider)
-	sharedObjs = addCommonLabels(sharedObjs, provider)
+	instanceObjs = addCommonLabels(instanceObjs, input.Provider)
+	sharedObjs = addCommonLabels(sharedObjs, input.Provider)
 
 	// Add an identifying label to shared components so next invocation of init, clusterctl delete and clusterctl upgrade can act accordingly.
 	// Additionally, the capi-webhook-system namespace gets detached from any provider, so we prevent that deleting
@@ -290,12 +300,12 @@ func NewComponents(provider config.Provider, configClient config.Client, rawyaml
 	sharedObjs = fixSharedLabels(sharedObjs)
 
 	return &components{
-		Provider:          provider,
-		version:           options.Version,
+		Provider:          input.Provider,
+		version:           input.Options.Version,
 		variables:         variables,
 		images:            images,
-		targetNamespace:   options.TargetNamespace,
-		watchingNamespace: options.WatchingNamespace,
+		targetNamespace:   input.Options.TargetNamespace,
+		watchingNamespace: input.Options.WatchingNamespace,
 		instanceObjs:      instanceObjs,
 		sharedObjs:        sharedObjs,
 	}, nil
@@ -333,41 +343,6 @@ func splitInstanceAndSharedResources(objs []unstructured.Unstructured) (instance
 	return
 }
 
-func inspectVariables(data []byte) []string {
-	variables := sets.NewString()
-	match := variableRegEx.FindAllStringSubmatch(string(data), -1)
-
-	for _, m := range match {
-		submatch := m[1]
-		if !variables.Has(submatch) {
-			variables.Insert(submatch)
-		}
-	}
-
-	ret := variables.List()
-	sort.Strings(ret)
-	return ret
-}
-
-func replaceVariables(yaml []byte, variables []string, configVariablesClient config.VariablesClient, skipVariables bool) ([]byte, error) {
-	tmp := string(yaml)
-	var missingVariables []string
-	for _, key := range variables {
-		val, err := configVariablesClient.Get(key)
-		if err != nil {
-			missingVariables = append(missingVariables, key)
-			continue
-		}
-		exp := regexp.MustCompile(`\$\{\s*` + regexp.QuoteMeta(key) + `\s*\}`)
-		tmp = exp.ReplaceAllLiteralString(tmp, val)
-	}
-	if !skipVariables && len(missingVariables) > 0 {
-		return nil, errors.Errorf("value for variables [%s] is not set. Please set the value using os environment variables or the clusterctl config file", strings.Join(missingVariables, ", "))
-	}
-
-	return []byte(tmp), nil
-}
-
 // inspectTargetNamespace identifies the name of the namespace object contained in the components YAML, if any.
 // In case more than one Namespace object is identified, an error is returned.
 func inspectTargetNamespace(objs []unstructured.Unstructured) (string, error) {
diff --git a/cmd/clusterctl/client/repository/components_client.go b/cmd/clusterctl/client/repository/components_client.go
index dde67d6dcb42..52ac571d0d16 100644
--- a/cmd/clusterctl/client/repository/components_client.go
+++ b/cmd/clusterctl/client/repository/components_client.go
@@ -19,6 +19,7 @@ package repository
 import (
 	"github.com/pkg/errors"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	logf "sigs.k8s.io/cluster-api/cmd/clusterctl/log"
 )
 
@@ -33,6 +34,7 @@ type componentsClient struct {
 	provider     config.Provider
 	repository   Repository
 	configClient config.Client
+	processor    yaml.Processor
 }
 
 // ensure componentsClient implements ComponentsClient.
@@ -44,6 +46,7 @@ func newComponentsClient(provider config.Provider, repository Repository, config
 		provider:     provider,
 		repository:   repository,
 		configClient: configClient,
+		processor:    yaml.NewSimpleProcessor(),
 	}
 }
 
@@ -80,5 +83,5 @@ func (f *componentsClient) Get(options ComponentsOptions) (Components, error) {
 		log.Info("Using", "Override", path, "Provider", f.provider.ManifestLabel(), "Version", options.Version)
 	}
 
-	return NewComponents(f.provider, f.configClient, file, options)
+	return NewComponents(ComponentsInput{f.provider, f.configClient, f.processor, file, options})
 }
diff --git a/cmd/clusterctl/client/repository/components_client_test.go b/cmd/clusterctl/client/repository/components_client_test.go
index ced53933a272..6aa9bd529d89 100644
--- a/cmd/clusterctl/client/repository/components_client_test.go
+++ b/cmd/clusterctl/client/repository/components_client_test.go
@@ -21,10 +21,12 @@ import (
 	"testing"
 
 	. "github.com/onsi/gomega"
+	"github.com/pkg/errors"
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
 	clusterctlv1 "sigs.k8s.io/cluster-api/cmd/clusterctl/api/v1alpha3"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 	utilyaml "sigs.k8s.io/cluster-api/util/yaml"
 )
@@ -69,11 +71,13 @@ func Test_componentsClient_Get(t *testing.T) {
 	type fields struct {
 		provider   config.Provider
 		repository Repository
+		processor  yaml.Processor
 	}
 	type args struct {
 		version           string
 		targetNamespace   string
 		watchingNamespace string
+		skipVariables     bool
 	}
 	type want struct {
 		provider          config.Provider
@@ -90,7 +94,7 @@ func Test_componentsClient_Get(t *testing.T) {
 		wantErr bool
 	}{
 		{
-			name: "Pass",
+			name: "successfully gets the components",
 			fields: fields{
 				provider: p1,
 				repository: test.NewFakeRepository().
@@ -112,6 +116,30 @@ func Test_componentsClient_Get(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "successfully gets the components even with SkipVariables defined",
+			fields: fields{
+				provider: p1,
+				repository: test.NewFakeRepository().
+					WithPaths("root", "components.yaml").
+					WithDefaultVersion("v1.0.0").
+					WithFile("v1.0.0", "components.yaml", utilyaml.JoinYaml(namespaceYaml, controllerYaml, configMapYaml)),
+			},
+			args: args{
+				version:           "v1.0.0",
+				targetNamespace:   "",
+				watchingNamespace: "",
+				skipVariables:     true,
+			},
+			want: want{
+				provider:          p1,
+				version:           "v1.0.0",      // version detected
+				targetNamespace:   namespaceName, // default targetNamespace detected
+				watchingNamespace: "",
+				variables:         []string{variableName}, // variable detected
+			},
+			wantErr: false,
+		},
 		{
 			name: "targetNamespace overrides default targetNamespace",
 			fields: fields{
@@ -228,6 +256,41 @@ func Test_componentsClient_Get(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: "Fails if yaml processor cannot get Variables",
+			fields: fields{
+				provider: p1,
+				repository: test.NewFakeRepository().
+					WithPaths("root", "components.yaml").
+					WithDefaultVersion("v1.0.0").
+					WithFile("v1.0.0", "components.yaml", utilyaml.JoinYaml(namespaceYaml, controllerYaml, configMapYaml)),
+				processor: test.NewFakeProcessor().WithGetVariablesErr(errors.New("cannot get vars")),
+			},
+			args: args{
+				version:           "v1.0.0",
+				targetNamespace:   "default",
+				watchingNamespace: "",
+			},
+			wantErr: true,
+		},
+		{
+			name: "Fails if yaml processor cannot process the raw yaml",
+			fields: fields{
+				provider: p1,
+				repository: test.NewFakeRepository().
+					WithPaths("root", "components.yaml").
+					WithDefaultVersion("v1.0.0").
+					WithFile("v1.0.0", "components.yaml", utilyaml.JoinYaml(namespaceYaml, controllerYaml, configMapYaml)),
+
+				processor: test.NewFakeProcessor().WithProcessErr(errors.New("cannot process")),
+			},
+			args: args{
+				version:           "v1.0.0",
+				targetNamespace:   "default",
+				watchingNamespace: "",
+			},
+			wantErr: true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -237,8 +300,12 @@ func Test_componentsClient_Get(t *testing.T) {
 				Version:           tt.args.version,
 				TargetNamespace:   tt.args.targetNamespace,
 				WatchingNamespace: tt.args.watchingNamespace,
+				SkipVariables:     tt.args.skipVariables,
 			}
 			f := newComponentsClient(tt.fields.provider, tt.fields.repository, configClient)
+			if tt.fields.processor != nil {
+				f.processor = tt.fields.processor
+			}
 			got, err := f.Get(options)
 			if tt.wantErr {
 				gs.Expect(err).To(HaveOccurred())
@@ -259,10 +326,18 @@ func Test_componentsClient_Get(t *testing.T) {
 				return
 			}
 
-			if len(tt.want.variables) > 0 {
+			if !tt.args.skipVariables && len(tt.want.variables) > 0 {
 				gs.Expect(yaml).To(ContainSubstring(variableValue))
 			}
 
+			// Verify that when SkipVariables is set we have all the variables
+			// in the template without the values processed.
+			if tt.args.skipVariables {
+				for _, v := range tt.want.variables {
+					gs.Expect(yaml).To(ContainSubstring(v))
+				}
+			}
+
 			for _, o := range got.InstanceObjs() {
 				for _, v := range []string{clusterctlv1.ClusterctlLabelName, clusterv1.ProviderLabelName} {
 					gs.Expect(o.GetLabels()).To(HaveKey(v))
diff --git a/cmd/clusterctl/client/repository/components_test.go b/cmd/clusterctl/client/repository/components_test.go
index 44fb5268fc3a..6826e4385dba 100644
--- a/cmd/clusterctl/client/repository/components_test.go
+++ b/cmd/clusterctl/client/repository/components_test.go
@@ -26,138 +26,8 @@ import (
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1alpha3"
 	clusterctlv1 "sigs.k8s.io/cluster-api/cmd/clusterctl/api/v1alpha3"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
-	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
 
-func Test_inspectVariables(t *testing.T) {
-	type args struct {
-		data string
-	}
-	tests := []struct {
-		name string
-		args args
-		want []string
-	}{
-		{
-			name: "variable with different spacing around the name",
-			args: args{
-				data: "yaml with ${A} ${ B} ${ C} ${ D }",
-			},
-			want: []string{"A", "B", "C", "D"},
-		},
-		{
-			name: "variables used in many places are grouped",
-			args: args{
-				data: "yaml with ${A} ${A} ${A}",
-			},
-			want: []string{"A"},
-		},
-		{
-			name: "variables in multiline texts are processed",
-			args: args{
-				data: "yaml with ${A}\n${B}\n${C}",
-			},
-			want: []string{"A", "B", "C"},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			g := NewWithT(t)
-
-			g.Expect(inspectVariables([]byte(tt.args.data))).To(Equal(tt.want))
-		})
-	}
-}
-
-func Test_replaceVariables(t *testing.T) {
-	type args struct {
-		yaml                  []byte
-		variables             []string
-		configVariablesClient config.VariablesClient
-		skipVariables         bool
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    []byte
-		wantErr bool
-	}{
-		{
-			name: "pass and replaces variables",
-			args: args{
-				yaml:      []byte("foo ${ BAR }"),
-				variables: []string{"BAR"},
-				configVariablesClient: test.NewFakeVariableClient().
-					WithVar("BAR", "bar"),
-				skipVariables: false,
-			},
-			want:    []byte("foo bar"),
-			wantErr: false,
-		},
-		{
-			name: "pass and replaces variables when variable name contains regex metacharacters",
-			args: args{
-				yaml:      []byte("foo ${ BA$R }"),
-				variables: []string{"BA$R"},
-				configVariablesClient: test.NewFakeVariableClient().
-					WithVar("BA$R", "bar"),
-				skipVariables: false,
-			},
-			want:    []byte("foo bar"),
-			wantErr: false,
-		},
-		{
-			name: "pass and replaces variables when variable value contains regex metacharacters",
-			args: args{
-				yaml:      []byte("foo ${ BAR }"),
-				variables: []string{"BAR"},
-				configVariablesClient: test.NewFakeVariableClient().
-					WithVar("BAR", "ba$r"),
-				skipVariables: false,
-			},
-			want:    []byte("foo ba$r"),
-			wantErr: false,
-		},
-		{
-			name: "fails for missing variables and not skip variables",
-			args: args{
-				yaml:                  []byte("foo ${ BAR } ${ BAZ }"),
-				variables:             []string{"BAR", "BAZ"},
-				configVariablesClient: test.NewFakeVariableClient(),
-				skipVariables:         false,
-			},
-			want:    nil,
-			wantErr: true,
-		},
-		{
-			name: "pass when missing variables and skip variables",
-			args: args{
-				yaml:      []byte("foo ${ BAR } ${ BAZ }"),
-				variables: []string{"BAR", "BAZ"},
-				configVariablesClient: test.NewFakeVariableClient().
-					WithVar("BAR", "bar"),
-				skipVariables: true,
-			},
-			want:    []byte("foo bar ${ BAZ }"),
-			wantErr: false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			g := NewWithT(t)
-
-			got, err := replaceVariables(tt.args.yaml, tt.args.variables, tt.args.configVariablesClient, tt.args.skipVariables)
-			if tt.wantErr {
-				g.Expect(err).To(HaveOccurred())
-				return
-			}
-			g.Expect(err).NotTo(HaveOccurred())
-
-			g.Expect(got).To(Equal(tt.want))
-		})
-	}
-}
-
 func Test_inspectTargetNamespace(t *testing.T) {
 	type args struct {
 		objs []unstructured.Unstructured
diff --git a/cmd/clusterctl/client/repository/template.go b/cmd/clusterctl/client/repository/template.go
index 19ddc6142367..e8b7ecf7b088 100644
--- a/cmd/clusterctl/client/repository/template.go
+++ b/cmd/clusterctl/client/repository/template.go
@@ -20,6 +20,7 @@ import (
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	utilyaml "sigs.k8s.io/cluster-api/util/yaml"
 )
 
@@ -69,24 +70,35 @@ func (t *template) Yaml() ([]byte, error) {
 	return utilyaml.FromUnstructured(t.objs)
 }
 
+type TemplateInput struct {
+	RawArtifact           []byte
+	ConfigVariablesClient config.VariablesClient
+	Processor             yaml.Processor
+	TargetNamespace       string
+	ListVariablesOnly     bool
+}
+
 // NewTemplate returns a new objects embedding a cluster template YAML file.
-func NewTemplate(rawYaml []byte, configVariablesClient config.VariablesClient, targetNamespace string, listVariablesOnly bool) (*template, error) {
-	// Inspect variables and replace with values from the configuration.
-	variables := inspectVariables(rawYaml)
-	if listVariablesOnly {
+func NewTemplate(input TemplateInput) (*template, error) {
+	variables, err := input.Processor.GetVariables(input.RawArtifact)
+	if err != nil {
+		return nil, err
+	}
+
+	if input.ListVariablesOnly {
 		return &template{
 			variables:       variables,
-			targetNamespace: targetNamespace,
+			targetNamespace: input.TargetNamespace,
 		}, nil
 	}
 
-	yaml, err := replaceVariables(rawYaml, variables, configVariablesClient, false)
+	processedYaml, err := input.Processor.Process(input.RawArtifact, input.ConfigVariablesClient.Get)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to perform variable substitution")
+		return nil, err
 	}
 
 	// Transform the yaml in a list of objects, so following transformation can work on typed objects (instead of working on a string/slice of bytes).
-	objs, err := utilyaml.ToUnstructured(yaml)
+	objs, err := utilyaml.ToUnstructured(processedYaml)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to parse yaml")
 	}
@@ -94,11 +106,11 @@ func NewTemplate(rawYaml []byte, configVariablesClient config.VariablesClient, t
 	// Ensures all the template components are deployed in the target namespace (applies only to namespaced objects)
 	// This is required in order to ensure a cluster and all the related objects are in a single namespace, that is a requirement for
 	// the clusterctl move operation (and also for many controller reconciliation loops).
-	objs = fixTargetNamespace(objs, targetNamespace)
+	objs = fixTargetNamespace(objs, input.TargetNamespace)
 
 	return &template{
 		variables:       variables,
-		targetNamespace: targetNamespace,
+		targetNamespace: input.TargetNamespace,
 		objs:            objs,
 	}, nil
 }
diff --git a/cmd/clusterctl/client/repository/template_client.go b/cmd/clusterctl/client/repository/template_client.go
index 64dc4cd7dd4c..e896a1ffbf31 100644
--- a/cmd/clusterctl/client/repository/template_client.go
+++ b/cmd/clusterctl/client/repository/template_client.go
@@ -17,10 +17,9 @@ limitations under the License.
 package repository
 
 import (
-	"fmt"
-
 	"github.com/pkg/errors"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	logf "sigs.k8s.io/cluster-api/cmd/clusterctl/log"
 )
 
@@ -36,18 +35,29 @@ type templateClient struct {
 	version               string
 	repository            Repository
 	configVariablesClient config.VariablesClient
+	processor             yaml.Processor
+}
+
+type TemplateClientInput struct {
+	version               string
+	provider              config.Provider
+	repository            Repository
+	configVariablesClient config.VariablesClient
+	processor             yaml.Processor
 }
 
 // Ensure templateClient implements the TemplateClient interface.
 var _ TemplateClient = &templateClient{}
 
-// newTemplateClient returns a templateClient.
-func newTemplateClient(provider config.Provider, version string, repository Repository, configVariablesClient config.VariablesClient) *templateClient {
+// newTemplateClient returns a templateClient. It uses the SimpleYamlProcessor
+// by default
+func newTemplateClient(input TemplateClientInput) *templateClient {
 	return &templateClient{
-		provider:              provider,
-		version:               version,
-		repository:            repository,
-		configVariablesClient: configVariablesClient,
+		provider:              input.provider,
+		version:               input.version,
+		repository:            input.repository,
+		configVariablesClient: input.configVariablesClient,
+		processor:             input.processor,
 	}
 }
 
@@ -61,19 +71,11 @@ func (c *templateClient) Get(flavor, targetNamespace string, listVariablesOnly b
 		return nil, errors.New("invalid arguments: please provide a targetNamespace")
 	}
 
-	// we are always reading templateClient for a well know version, that usually is
-	// the version of the provider installed in the management cluster.
 	version := c.version
-
-	// building template name according with the naming convention
-	name := "cluster-template"
-	if flavor != "" {
-		name = fmt.Sprintf("%s-%s", name, flavor)
-	}
-	name = fmt.Sprintf("%s.yaml", name)
+	name := c.processor.GetTemplateName(version, flavor)
 
 	// read the component YAML, reading the local override file if it exists, otherwise read from the provider repository
-	rawYaml, err := getLocalOverride(&newOverrideInput{
+	rawArtifact, err := getLocalOverride(&newOverrideInput{
 		configVariablesClient: c.configVariablesClient,
 		provider:              c.provider,
 		version:               version,
@@ -83,9 +85,9 @@ func (c *templateClient) Get(flavor, targetNamespace string, listVariablesOnly b
 		return nil, err
 	}
 
-	if rawYaml == nil {
+	if rawArtifact == nil {
 		log.V(5).Info("Fetching", "File", name, "Provider", c.provider.ManifestLabel(), "Version", version)
-		rawYaml, err = c.repository.GetFile(version, name)
+		rawArtifact, err = c.repository.GetFile(version, name)
 		if err != nil {
 			return nil, errors.Wrapf(err, "failed to read %q from provider's repository %q", name, c.provider.ManifestLabel())
 		}
@@ -93,5 +95,5 @@ func (c *templateClient) Get(flavor, targetNamespace string, listVariablesOnly b
 		log.V(1).Info("Using", "Override", name, "Provider", c.provider.ManifestLabel(), "Version", version)
 	}
 
-	return NewTemplate(rawYaml, c.configVariablesClient, targetNamespace, listVariablesOnly)
+	return NewTemplate(TemplateInput{rawArtifact, c.configVariablesClient, c.processor, targetNamespace, listVariablesOnly})
 }
diff --git a/cmd/clusterctl/client/repository/template_client_test.go b/cmd/clusterctl/client/repository/template_client_test.go
index 4a7b15f57f58..f2d696e1224a 100644
--- a/cmd/clusterctl/client/repository/template_client_test.go
+++ b/cmd/clusterctl/client/repository/template_client_test.go
@@ -21,9 +21,11 @@ import (
 	"testing"
 
 	. "github.com/onsi/gomega"
+	"github.com/pkg/errors"
 
 	clusterctlv1 "sigs.k8s.io/cluster-api/cmd/clusterctl/api/v1alpha3"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
 
@@ -35,6 +37,7 @@ func Test_templates_Get(t *testing.T) {
 		provider              config.Provider
 		repository            Repository
 		configVariablesClient config.VariablesClient
+		processor             yaml.Processor
 	}
 	type args struct {
 		flavor            string
@@ -62,6 +65,7 @@ func Test_templates_Get(t *testing.T) {
 					WithDefaultVersion("v1.0").
 					WithFile("v1.0", "cluster-template.yaml", templateMapYaml),
 				configVariablesClient: test.NewFakeVariableClient().WithVar(variableName, variableValue),
+				processor:             yaml.NewSimpleProcessor(),
 			},
 			args: args{
 				flavor:            "",
@@ -84,6 +88,7 @@ func Test_templates_Get(t *testing.T) {
 					WithDefaultVersion("v1.0").
 					WithFile("v1.0", "cluster-template-prod.yaml", templateMapYaml),
 				configVariablesClient: test.NewFakeVariableClient().WithVar(variableName, variableValue),
+				processor:             yaml.NewSimpleProcessor(),
 			},
 			args: args{
 				flavor:            "prod",
@@ -105,6 +110,7 @@ func Test_templates_Get(t *testing.T) {
 					WithPaths("root", "").
 					WithDefaultVersion("v1.0"),
 				configVariablesClient: test.NewFakeVariableClient().WithVar(variableName, variableValue),
+				processor:             yaml.NewSimpleProcessor(),
 			},
 			args: args{
 				flavor:            "",
@@ -123,6 +129,7 @@ func Test_templates_Get(t *testing.T) {
 					WithDefaultVersion("v1.0").
 					WithFile("v1.0", "cluster-template.yaml", templateMapYaml),
 				configVariablesClient: test.NewFakeVariableClient(),
+				processor:             yaml.NewSimpleProcessor(),
 			},
 			args: args{
 				flavor:            "",
@@ -141,6 +148,7 @@ func Test_templates_Get(t *testing.T) {
 					WithDefaultVersion("v1.0").
 					WithFile("v1.0", "cluster-template.yaml", templateMapYaml),
 				configVariablesClient: test.NewFakeVariableClient(),
+				processor:             yaml.NewSimpleProcessor(),
 			},
 			args: args{
 				flavor:            "",
@@ -153,12 +161,38 @@ func Test_templates_Get(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "returns error if processor is unable to get variables",
+			fields: fields{
+				version:  "v1.0",
+				provider: p1,
+				repository: test.NewFakeRepository().
+					WithPaths("root", "").
+					WithDefaultVersion("v1.0").
+					WithFile("v1.0", "cluster-template.yaml", templateMapYaml),
+				configVariablesClient: test.NewFakeVariableClient().WithVar(variableName, variableValue),
+				processor:             test.NewFakeProcessor().WithGetVariablesErr(errors.New("cannot get vars")).WithTemplateName("cluster-template.yaml"),
+			},
+			args: args{
+				targetNamespace:   "ns1",
+				listVariablesOnly: true,
+			},
+			wantErr: true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 
-			f := newTemplateClient(tt.fields.provider, tt.fields.version, tt.fields.repository, tt.fields.configVariablesClient)
+			f := newTemplateClient(
+				TemplateClientInput{
+					version:               tt.fields.version,
+					provider:              tt.fields.provider,
+					repository:            tt.fields.repository,
+					configVariablesClient: tt.fields.configVariablesClient,
+					processor:             tt.fields.processor,
+				},
+			)
 			got, err := f.Get(tt.args.flavor, tt.args.targetNamespace, tt.args.listVariablesOnly)
 			if tt.wantErr {
 				g.Expect(err).To(HaveOccurred())
diff --git a/cmd/clusterctl/client/repository/template_test.go b/cmd/clusterctl/client/repository/template_test.go
index faaa4f04d8ea..9ce045c6bb81 100644
--- a/cmd/clusterctl/client/repository/template_test.go
+++ b/cmd/clusterctl/client/repository/template_test.go
@@ -23,6 +23,7 @@ import (
 	. "github.com/onsi/gomega"
 
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	yaml "sigs.k8s.io/cluster-api/cmd/clusterctl/client/yamlprocessor"
 	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
 )
 
@@ -39,6 +40,7 @@ func Test_newTemplate(t *testing.T) {
 	type args struct {
 		rawYaml               []byte
 		configVariablesClient config.VariablesClient
+		processor             yaml.Processor
 		targetNamespace       string
 		listVariablesOnly     bool
 	}
@@ -57,6 +59,7 @@ func Test_newTemplate(t *testing.T) {
 			args: args{
 				rawYaml:               templateMapYaml,
 				configVariablesClient: test.NewFakeVariableClient().WithVar(variableName, variableValue),
+				processor:             yaml.NewSimpleProcessor(),
 				targetNamespace:       "ns1",
 				listVariablesOnly:     false,
 			},
@@ -71,6 +74,7 @@ func Test_newTemplate(t *testing.T) {
 			args: args{
 				rawYaml:               templateMapYaml,
 				configVariablesClient: test.NewFakeVariableClient(),
+				processor:             yaml.NewSimpleProcessor(),
 				targetNamespace:       "ns1",
 				listVariablesOnly:     true,
 			},
@@ -85,7 +89,13 @@ func Test_newTemplate(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 
-			got, err := NewTemplate(tt.args.rawYaml, tt.args.configVariablesClient, tt.args.targetNamespace, tt.args.listVariablesOnly)
+			got, err := NewTemplate(TemplateInput{
+				RawArtifact:           tt.args.rawYaml,
+				ConfigVariablesClient: tt.args.configVariablesClient,
+				Processor:             tt.args.processor,
+				TargetNamespace:       tt.args.targetNamespace,
+				ListVariablesOnly:     tt.args.listVariablesOnly,
+			})
 			if tt.wantErr {
 				g.Expect(err).To(HaveOccurred())
 				return
diff --git a/cmd/clusterctl/client/upgrade.go b/cmd/clusterctl/client/upgrade.go
index 81bbe8021b41..f1aee7b1de37 100644
--- a/cmd/clusterctl/client/upgrade.go
+++ b/cmd/clusterctl/client/upgrade.go
@@ -33,7 +33,7 @@ type PlanUpgradeOptions struct {
 
 func (c *clusterctlClient) PlanUpgrade(options PlanUpgradeOptions) ([]UpgradePlan, error) {
 	// Get the client for interacting with the management cluster.
-	cluster, err := c.clusterClientFactory(options.Kubeconfig)
+	cluster, err := c.clusterClientFactory(ClusterClientFactoryInput{kubeconfig: options.Kubeconfig})
 	if err != nil {
 		return nil, err
 	}
@@ -89,7 +89,7 @@ type ApplyUpgradeOptions struct {
 
 func (c *clusterctlClient) ApplyUpgrade(options ApplyUpgradeOptions) error {
 	// Get the client for interacting with the management cluster.
-	clusterClient, err := c.clusterClientFactory(options.Kubeconfig)
+	clusterClient, err := c.clusterClientFactory(ClusterClientFactoryInput{kubeconfig: options.Kubeconfig})
 	if err != nil {
 		return err
 	}
diff --git a/cmd/clusterctl/client/yamlprocessor/processor.go b/cmd/clusterctl/client/yamlprocessor/processor.go
new file mode 100644
index 000000000000..5ff060396feb
--- /dev/null
+++ b/cmd/clusterctl/client/yamlprocessor/processor.go
@@ -0,0 +1,32 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package yamlprocessor
+
+// Processor defines the methods necessary for creating a specific yaml
+// processor.
+type Processor interface {
+	// GetTemplateName returns the name of the template that needs to be
+	// retrieved from the source.
+	GetTemplateName(version, flavor string) string
+
+	// GetVariables parses the template blob of bytes and provides a
+	// list of variables that the template requires.
+	GetVariables([]byte) ([]string, error)
+
+	// Process processes the template blob of bytes and will return the final
+	// yaml with values retrieved from the values getter
+	Process([]byte, func(string) (string, error)) ([]byte, error)
+}
diff --git a/cmd/clusterctl/client/yamlprocessor/simple_processor.go b/cmd/clusterctl/client/yamlprocessor/simple_processor.go
new file mode 100644
index 000000000000..773e121a556b
--- /dev/null
+++ b/cmd/clusterctl/client/yamlprocessor/simple_processor.go
@@ -0,0 +1,100 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package yamlprocessor
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+// SimpleProcessor is a yaml processor that does simple variable substitution.
+// It implements the YamlProcessor interface. The variables are defined in
+// the following format ${variable_name}
+type SimpleProcessor struct{}
+
+var _ Processor = &SimpleProcessor{}
+
+func NewSimpleProcessor() *SimpleProcessor {
+	return &SimpleProcessor{}
+}
+
+// GetTemplateName returns the name of the template that the simple processor
+// uses. It follows the cluster template naming convention of
+// "cluster-template<-flavor>.yaml".
+func (tp *SimpleProcessor) GetTemplateName(_, flavor string) string {
+	name := "cluster-template"
+	if flavor != "" {
+		name = fmt.Sprintf("%s-%s", name, flavor)
+	}
+	name = fmt.Sprintf("%s.yaml", name)
+
+	return name
+}
+
+// GetVariables returns a list of the variables specified in the yaml
+// manifest. The format of the variables being parsed is ${VAR}
+func (tp *SimpleProcessor) GetVariables(rawArtifact []byte) ([]string, error) {
+	return inspectVariables(rawArtifact), nil
+}
+
+// Process returns the final yaml with all the variables replaced with their
+// respective values. If there are variables without corresponding values, it
+// will return the yaml along with an error.
+func (tp *SimpleProcessor) Process(rawArtifact []byte, variablesGetter func(string) (string, error)) ([]byte, error) {
+	// Inspect the yaml read from the repository for variables.
+	variables := inspectVariables(rawArtifact)
+
+	// Replace variables with corresponding values read from the config
+	tmp := string(rawArtifact)
+	var err error
+	var missingVariables []string
+	for _, key := range variables {
+		val, err := variablesGetter(key)
+		if err != nil {
+			missingVariables = append(missingVariables, key)
+			continue
+		}
+		exp := regexp.MustCompile(`\$\{\s*` + regexp.QuoteMeta(key) + `\s*\}`)
+		tmp = exp.ReplaceAllLiteralString(tmp, val)
+	}
+	if len(missingVariables) > 0 {
+		err = errors.Errorf("value for variables [%s] is not set. Please set the value using os environment variables or the clusterctl config file", strings.Join(missingVariables, ", "))
+	}
+
+	return []byte(tmp), err
+}
+
+// variableRegEx defines the regexp used for searching variables inside a YAML
+var variableRegEx = regexp.MustCompile(`\${\s*([A-Z0-9_$]+)\s*}`)
+
+func inspectVariables(data []byte) []string {
+	variables := sets.NewString()
+	match := variableRegEx.FindAllStringSubmatch(string(data), -1)
+
+	for _, m := range match {
+		submatch := m[1]
+		if !variables.Has(submatch) {
+			variables.Insert(submatch)
+		}
+	}
+
+	return variables.List()
+}
diff --git a/cmd/clusterctl/client/yamlprocessor/simple_processor_test.go b/cmd/clusterctl/client/yamlprocessor/simple_processor_test.go
new file mode 100644
index 000000000000..6cfc724a653c
--- /dev/null
+++ b/cmd/clusterctl/client/yamlprocessor/simple_processor_test.go
@@ -0,0 +1,156 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package yamlprocessor
+
+import (
+	"testing"
+
+	. "github.com/onsi/gomega"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/client/config"
+	"sigs.k8s.io/cluster-api/cmd/clusterctl/internal/test"
+)
+
+func TestSimpleProcessor_GetTemplateName(t *testing.T) {
+	g := NewWithT(t)
+	p := NewSimpleProcessor()
+	g.Expect(p.GetTemplateName("some-version", "some-flavor")).To(Equal("cluster-template-some-flavor.yaml"))
+	g.Expect(p.GetTemplateName("", "")).To(Equal("cluster-template.yaml"))
+}
+
+func TestSimpleProcessor_GetVariables(t *testing.T) {
+	type args struct {
+		data string
+	}
+	tests := []struct {
+		name string
+		args args
+		want []string
+	}{
+		{
+			name: "variable with different spacing around the name",
+			args: args{
+				data: "yaml with ${A} ${ B} ${ C} ${ D }",
+			},
+			want: []string{"A", "B", "C", "D"},
+		},
+		{
+			name: "variables used in many places are grouped",
+			args: args{
+				data: "yaml with ${A} ${A} ${A}",
+			},
+			want: []string{"A"},
+		},
+		{
+			name: "variables in multiline texts are processed",
+			args: args{
+				data: "yaml with ${A}\n${B}\n${C}",
+			},
+			want: []string{"A", "B", "C"},
+		},
+		{
+			name: "variables are sorted",
+			args: args{
+				data: "yaml with ${C}\n${B}\n${A}",
+			},
+			want: []string{"A", "B", "C"},
+		},
+		{
+			name: "variables with regex metacharacters",
+			args: args{
+				data: "yaml with ${BA$R}\n${FOO}",
+			},
+			want: []string{"BA$R", "FOO"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+			p := NewSimpleProcessor()
+
+			g.Expect(p.GetVariables([]byte(tt.args.data))).To(Equal(tt.want))
+		})
+	}
+}
+
+func TestSimpleProcessor_Process(t *testing.T) {
+	type args struct {
+		yaml                  []byte
+		configVariablesClient config.VariablesClient
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    []byte
+		wantErr bool
+	}{
+		{
+			name: "pass and replaces variables",
+			args: args{
+				yaml: []byte("foo ${ BAR }"),
+				configVariablesClient: test.NewFakeVariableClient().
+					WithVar("BAR", "bar"),
+			},
+			want:    []byte("foo bar"),
+			wantErr: false,
+		},
+		{
+			name: "pass and replaces variables when variable name contains regex metacharacters",
+			args: args{
+				yaml: []byte("foo ${ BA$R }"),
+				configVariablesClient: test.NewFakeVariableClient().
+					WithVar("BA$R", "bar"),
+			},
+			want:    []byte("foo bar"),
+			wantErr: false,
+		},
+		{
+			name: "pass and replaces variables when variable value contains regex metacharacters",
+			args: args{
+				yaml: []byte("foo ${ BAR }"),
+				configVariablesClient: test.NewFakeVariableClient().
+					WithVar("BAR", "ba$r"),
+			},
+			want:    []byte("foo ba$r"),
+			wantErr: false,
+		},
+		{
+			name: "returns error when missing values for template variables",
+			args: args{
+				yaml:                  []byte("foo ${ BAR } ${ BAZ }"),
+				configVariablesClient: test.NewFakeVariableClient(),
+			},
+			want:    nil,
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+			p := NewSimpleProcessor()
+
+			got, err := p.Process(tt.args.yaml, tt.args.configVariablesClient.Get)
+			if tt.wantErr {
+				g.Expect(err).To(HaveOccurred())
+				return
+			}
+			g.Expect(err).NotTo(HaveOccurred())
+
+			g.Expect(got).To(Equal(tt.want))
+		})
+	}
+}
diff --git a/cmd/clusterctl/internal/test/fake_processor.go b/cmd/clusterctl/internal/test/fake_processor.go
new file mode 100644
index 000000000000..63d1960ea623
--- /dev/null
+++ b/cmd/clusterctl/internal/test/fake_processor.go
@@ -0,0 +1,54 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package test
+
+type FakeProcessor struct {
+	errGetVariables error
+	errProcess      error
+	artifactName    string
+}
+
+func NewFakeProcessor() *FakeProcessor {
+	return &FakeProcessor{}
+}
+
+func (fp *FakeProcessor) WithTemplateName(n string) *FakeProcessor {
+	fp.artifactName = n
+	return fp
+}
+
+func (fp *FakeProcessor) WithGetVariablesErr(e error) *FakeProcessor {
+	fp.errGetVariables = e
+	return fp
+}
+
+func (fp *FakeProcessor) WithProcessErr(e error) *FakeProcessor {
+	fp.errProcess = e
+	return fp
+}
+
+func (fp *FakeProcessor) GetTemplateName(version, flavor string) string {
+	return fp.artifactName
+}
+
+func (fp *FakeProcessor) GetVariables(raw []byte) ([]string, error) {
+	return nil, fp.errGetVariables
+}
+
+func (fp *FakeProcessor) Process(raw []byte, variablesGetter func(string) (string, error)) ([]byte, error) {
+	return nil, fp.errProcess
+}