diff --git a/examples/k3d/k3d_test.go b/examples/k3d/k3d_test.go
new file mode 100644
index 00000000..3480e66c
--- /dev/null
+++ b/examples/k3d/k3d_test.go
@@ -0,0 +1,95 @@
+/*
+Copyright 2024 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 k3d
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"sigs.k8s.io/e2e-framework/pkg/stepfuncs"
+	"sigs.k8s.io/e2e-framework/support"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/e2e-framework/klient/wait"
+	"sigs.k8s.io/e2e-framework/klient/wait/conditions"
+	"sigs.k8s.io/e2e-framework/pkg/envconf"
+	"sigs.k8s.io/e2e-framework/pkg/features"
+)
+
+func newDeployment(namespace string, name string, replicaCount int32) *appsv1.Deployment {
+	podSpec := corev1.PodSpec{
+		Containers: []corev1.Container{
+			{
+				Name:  "my-container",
+				Image: "nginx",
+			},
+		},
+	}
+	return &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, Labels: map[string]string{"app": "test-app"}},
+		Spec: appsv1.DeploymentSpec{
+			Replicas: &replicaCount,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{"app": "test-app"},
+			},
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": "test-app"}},
+				Spec:       podSpec,
+			},
+		},
+	}
+}
+
+func TestK3DCluster(t *testing.T) {
+	deploymentFeature := features.New("Should be able to create a new deployment in the k3d cluster").
+		Assess("Create a new deployment", func(ctx context.Context, t *testing.T, c *envconf.Config) context.Context {
+			deployment := newDeployment(c.Namespace(), "test-deployment", 1)
+			if err := c.Client().Resources().Create(ctx, deployment); err != nil {
+				t.Fatal(err)
+			}
+			var dep appsv1.Deployment
+			if err := c.Client().Resources().Get(ctx, "test-deployment", c.Namespace(), &dep); err != nil {
+				t.Fatal(err)
+			}
+			err := wait.For(conditions.New(c.Client().Resources()).DeploymentConditionMatch(&dep, appsv1.DeploymentAvailable, corev1.ConditionTrue), wait.WithTimeout(time.Minute*3))
+			if err != nil {
+				t.Fatal(err)
+			}
+			return context.WithValue(ctx, "test-deployment", &dep)
+		}).
+		Feature()
+
+	nodeAddFeature := features.New("Should be able to add a new node to the k3d cluster").
+		Setup(stepfuncs.PerformNodeOperation(support.AddNode, &support.Node{
+			Name:    fmt.Sprintf("%s-agent", clusterName),
+			Cluster: clusterName,
+			Role:    "agent",
+		})).
+		Assess("Check if the node is added to the cluster", func(ctx context.Context, t *testing.T, c *envconf.Config) context.Context {
+			var node corev1.Node
+			if err := c.Client().Resources().Get(ctx, fmt.Sprintf("k3d-%s-agent-0", clusterName), c.Namespace(), &node); err != nil {
+				t.Fatal(err)
+			}
+			return ctx
+		}).Feature()
+
+	testEnv.Test(t, deploymentFeature, nodeAddFeature)
+}
diff --git a/examples/k3d/main_test.go b/examples/k3d/main_test.go
new file mode 100644
index 00000000..7c585310
--- /dev/null
+++ b/examples/k3d/main_test.go
@@ -0,0 +1,51 @@
+/*
+Copyright 2024 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 k3d
+
+import (
+	"os"
+	"testing"
+
+	"sigs.k8s.io/e2e-framework/pkg/env"
+	"sigs.k8s.io/e2e-framework/pkg/envconf"
+	"sigs.k8s.io/e2e-framework/pkg/envfuncs"
+	"sigs.k8s.io/e2e-framework/support/k3d"
+)
+
+var (
+	testEnv     env.Environment
+	clusterName string
+)
+
+func TestMain(m *testing.M) {
+	testEnv = env.New()
+	clusterName = envconf.RandomName("test", 16)
+	namespace := envconf.RandomName("k3d-ns", 16)
+
+	testEnv.Setup(
+		envfuncs.CreateClusterWithOpts(k3d.NewProvider(), clusterName, k3d.WithImage("rancher/k3s:v1.29.6-k3s1")),
+		envfuncs.CreateNamespace(namespace),
+		envfuncs.LoadImageToCluster(clusterName, "rancher/k3s:v1.29.6-k3s1", "--verbose", "--mode", "direct"),
+	)
+
+	testEnv.Finish(
+		envfuncs.DeleteNamespace(namespace),
+		envfuncs.DestroyCluster(clusterName),
+	)
+
+	os.Exit(testEnv.Run(m))
+}
diff --git a/pkg/envfuncs/provider_funcs.go b/pkg/envfuncs/provider_funcs.go
index 4d2b6ac6..c3fcd770 100644
--- a/pkg/envfuncs/provider_funcs.go
+++ b/pkg/envfuncs/provider_funcs.go
@@ -20,19 +20,19 @@ import (
 	"context"
 	"fmt"
 
+	"sigs.k8s.io/e2e-framework/pkg/utils"
+
 	"sigs.k8s.io/e2e-framework/pkg/env"
 	"sigs.k8s.io/e2e-framework/pkg/envconf"
 	"sigs.k8s.io/e2e-framework/support"
 )
 
-type clusterNameContextKey string
-
 var LoadDockerImageToCluster = LoadImageToCluster
 
 // GetClusterFromContext helps extract the E2EClusterProvider object from the context.
 // This can be used to setup and run tests of multi cluster e2e Prioviders.
 func GetClusterFromContext(ctx context.Context, clusterName string) (support.E2EClusterProvider, bool) {
-	c := ctx.Value(clusterNameContextKey(clusterName))
+	c := ctx.Value(support.ClusterNameContextKey(clusterName))
 	if c == nil {
 		return nil, false
 	}
@@ -47,8 +47,19 @@ func GetClusterFromContext(ctx context.Context, clusterName string) (support.E2E
 // NOTE: the returned function will update its env config with the
 // kubeconfig file for the config client.
 func CreateCluster(p support.E2EClusterProvider, clusterName string) env.Func {
+	return CreateClusterWithOpts(p, clusterName)
+}
+
+// CreateClusterWithOpts returns an env.Func that is used to
+// create an E2E provider cluster that is then injected in the context
+// using the name as a key. This can be provided with additional opts to extend the create
+// workflow of the cluster.
+//
+// NOTE: the returned function will update its env config with the
+// kubeconfig file for the config client.
+func CreateClusterWithOpts(p support.E2EClusterProvider, clusterName string, opts ...support.ClusterOpts) env.Func {
 	return func(ctx context.Context, cfg *envconf.Config) (context.Context, error) {
-		k := p.SetDefaults().WithName(clusterName)
+		k := p.SetDefaults().WithName(clusterName).WithOpts(opts...)
 		kubecfg, err := k.Create(ctx)
 		if err != nil {
 			return ctx, err
@@ -63,7 +74,7 @@ func CreateCluster(p support.E2EClusterProvider, clusterName string) env.Func {
 		}
 
 		// store entire cluster value in ctx for future access using the cluster name
-		return context.WithValue(ctx, clusterNameContextKey(clusterName), k), nil
+		return context.WithValue(ctx, support.ClusterNameContextKey(clusterName), k), nil
 	}
 }
 
@@ -90,7 +101,7 @@ func CreateClusterWithConfig(p support.E2EClusterProvider, clusterName, configFi
 		}
 
 		// store entire cluster value in ctx for future access using the cluster name
-		return context.WithValue(ctx, clusterNameContextKey(clusterName), k), nil
+		return context.WithValue(ctx, support.ClusterNameContextKey(clusterName), k), nil
 	}
 }
 
@@ -100,7 +111,7 @@ func CreateClusterWithConfig(p support.E2EClusterProvider, clusterName, configFi
 // NOTE: this should be used in a Environment.Finish step.
 func DestroyCluster(name string) env.Func {
 	return func(ctx context.Context, cfg *envconf.Config) (context.Context, error) {
-		clusterVal := ctx.Value(clusterNameContextKey(name))
+		clusterVal := ctx.Value(support.ClusterNameContextKey(name))
 		if clusterVal == nil {
 			return ctx, fmt.Errorf("destroy e2e provider cluster func: context cluster is nil")
 		}
@@ -121,9 +132,9 @@ func DestroyCluster(name string) env.Func {
 // LoadImageToCluster returns an EnvFunc that
 // retrieves a previously saved e2e provider Cluster in the context (using the name), and then loads a container image
 // from the host into the cluster.
-func LoadImageToCluster(name, image string) env.Func {
+func LoadImageToCluster(name, image string, args ...string) env.Func {
 	return func(ctx context.Context, cfg *envconf.Config) (context.Context, error) {
-		clusterVal := ctx.Value(clusterNameContextKey(name))
+		clusterVal := ctx.Value(support.ClusterNameContextKey(name))
 		if clusterVal == nil {
 			return ctx, fmt.Errorf("load image func: context cluster is nil")
 		}
@@ -133,7 +144,7 @@ func LoadImageToCluster(name, image string) env.Func {
 			return ctx, fmt.Errorf("load image archive func: cluster provider does not support LoadImage helper")
 		}
 
-		if err := cluster.LoadImage(ctx, image); err != nil {
+		if err := cluster.LoadImage(ctx, image, args...); err != nil {
 			return ctx, fmt.Errorf("load image: %w", err)
 		}
 
@@ -144,9 +155,9 @@ func LoadImageToCluster(name, image string) env.Func {
 // LoadImageArchiveToCluster returns an EnvFunc that
 // retrieves a previously saved e2e provider Cluster in the context (using the name), and then loads a container image TAR archive
 // from the host into the cluster.
-func LoadImageArchiveToCluster(name, imageArchive string) env.Func {
+func LoadImageArchiveToCluster(name, imageArchive string, args ...string) env.Func {
 	return func(ctx context.Context, cfg *envconf.Config) (context.Context, error) {
-		clusterVal := ctx.Value(clusterNameContextKey(name))
+		clusterVal := ctx.Value(support.ClusterNameContextKey(name))
 		if clusterVal == nil {
 			return ctx, fmt.Errorf("load image archive func: context cluster is nil")
 		}
@@ -156,7 +167,7 @@ func LoadImageArchiveToCluster(name, imageArchive string) env.Func {
 			return ctx, fmt.Errorf("load image archive func: cluster provider does not support LoadImageArchive helper")
 		}
 
-		if err := cluster.LoadImageArchive(ctx, imageArchive); err != nil {
+		if err := cluster.LoadImageArchive(ctx, imageArchive, args...); err != nil {
 			return ctx, fmt.Errorf("load image archive: %w", err)
 		}
 
@@ -169,7 +180,7 @@ func LoadImageArchiveToCluster(name, imageArchive string) env.Func {
 // in the provided destination.
 func ExportClusterLogs(name, dest string) env.Func {
 	return func(ctx context.Context, cfg *envconf.Config) (context.Context, error) {
-		clusterVal := ctx.Value(clusterNameContextKey(name))
+		clusterVal := ctx.Value(support.ClusterNameContextKey(name))
 		if clusterVal == nil {
 			return ctx, fmt.Errorf("export e2e provider cluster logs: context cluster is nil")
 		}
@@ -186,3 +197,12 @@ func ExportClusterLogs(name, dest string) env.Func {
 		return ctx, nil
 	}
 }
+
+// PerformNodeOperation returns an EnvFunc that can be used to perform some node lifecycle operations.
+// This can be used to add/remove/start/stop nodes in the cluster.
+func PerformNodeOperation(clusterName string, action support.NodeOperation, node *support.Node, args ...string) env.Func {
+	return func(ctx context.Context, cfg *envconf.Config) (context.Context, error) {
+		err := utils.PerformNodeLifecycleOperation(ctx, action, node, args...)
+		return ctx, err
+	}
+}
diff --git a/pkg/stepfuncs/nodelifecycle_funcs.go b/pkg/stepfuncs/nodelifecycle_funcs.go
new file mode 100644
index 00000000..bf5339d7
--- /dev/null
+++ b/pkg/stepfuncs/nodelifecycle_funcs.go
@@ -0,0 +1,43 @@
+/*
+Copyright 2024 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 stepfuncs
+
+import (
+	"context"
+	"testing"
+
+	"sigs.k8s.io/e2e-framework/pkg/utils"
+
+	"sigs.k8s.io/e2e-framework/pkg/envconf"
+	"sigs.k8s.io/e2e-framework/pkg/types"
+	"sigs.k8s.io/e2e-framework/support"
+)
+
+// PerformNodeOperation returns a step function that performs a node operation on a cluster.
+// This can be integrated as a setup function for a feature in question before the feature
+// is tested.
+func PerformNodeOperation(action support.NodeOperation, node *support.Node, args ...string) types.StepFunc {
+	return func(ctx context.Context, t *testing.T, config *envconf.Config) context.Context {
+		t.Helper()
+
+		err := utils.PerformNodeLifecycleOperation(ctx, action, node, args...)
+		if err != nil {
+			t.Fatalf("failed to perform node operation: %v", err)
+		}
+		return ctx
+	}
+}
diff --git a/pkg/types/types.go b/pkg/types/types.go
index f921d86a..ef640e9c 100644
--- a/pkg/types/types.go
+++ b/pkg/types/types.go
@@ -18,8 +18,11 @@ package types
 
 import (
 	"context"
+	"net"
 	"testing"
 
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/e2e-framework/klient"
 	"sigs.k8s.io/e2e-framework/pkg/envconf"
 	"sigs.k8s.io/e2e-framework/pkg/flags"
 )
@@ -138,3 +141,139 @@ type DescribableFeature interface {
 	// feature.
 	Description() string
 }
+
+type ClusterOpts func(c E2EClusterProvider)
+
+type Node struct {
+	Name    string
+	Role    string
+	Cluster string
+	State   string
+	IP      net.IP
+}
+
+type NodeOperation string
+
+const (
+	AddNode    NodeOperation = "add"
+	RemoveNode NodeOperation = "remove"
+	StartNode  NodeOperation = "start"
+	StopNode   NodeOperation = "stop"
+)
+
+type ClusterNameContextKey string
+
+type E2EClusterProvider interface {
+	// WithName is used to configure the cluster Name that should be used while setting up the cluster. Might
+	// Not apply for all providers.
+	WithName(name string) E2EClusterProvider
+
+	// WithVersion helps you override the default version used while using the cluster provider.
+	// This can be useful in providing a mechanism to the end users where they want to test their
+	// code against a certain specific version of k8s that is not the default one configured
+	// for the provider
+	WithVersion(version string) E2EClusterProvider
+
+	// WithPath heps you customize the executable binary that is used to back the cluster provider.
+	// This is useful in cases where your binary is present in a non standard location output of the
+	// PATH variable and you want to use that instead of framework trying to install one on it's own.
+	WithPath(path string) E2EClusterProvider
+
+	// WithOpts provides a way to customize the options that can be used while setting up the
+	// cluster using the providers such as kind or kwok or anything else. These helpers can be
+	// leveraged to setup arguments or configuration values that can be provided while performing
+	// the cluster bring up
+	WithOpts(opts ...ClusterOpts) E2EClusterProvider
+
+	// Create Provides an interface to start the cluster creation workflow using the selected provider
+	Create(ctx context.Context, args ...string) (string, error)
+
+	// CreateWithConfig is used to provide a mechanism where cluster providers that take an input config
+	// file and then setup the cluster accordingly. This can be used to provide input such as kind config
+	CreateWithConfig(ctx context.Context, configFile string) (string, error)
+
+	// GetKubeconfig provides a way to extract the kubeconfig file associated with the cluster in question
+	// using the cluster provider native way
+	GetKubeconfig() string
+
+	// GetKubectlContext is used to extract the kubectl context to be used while performing the operation
+	GetKubectlContext() string
+
+	// ExportLogs is used to export the cluster logs via the cluster provider native workflow. This
+	// can be used to export logs from the cluster after test failures for example to analyze the test
+	// failures better after the fact.
+	ExportLogs(ctx context.Context, dest string) error
+
+	// Destroy is used to cleanup a cluster brought up as part of the test workflow
+	Destroy(ctx context.Context) error
+
+	// SetDefaults is a handler function invoked after creating an object of type E2EClusterProvider. This method is
+	// invoked as the first step after creating an object in order to make sure the default values for required
+	// attributes are setup accordingly if any.
+	SetDefaults() E2EClusterProvider
+
+	// WaitForControlPlane is a helper function that can be used to indiate the Provider based cluster create workflow
+	// that the control plane is fully up and running. This method is invoked after the Create/CreateWithConfig handlers
+	// and is expected to return an error if the control plane doesn't stabilize. If the provider being implemented
+	// does not have a clear mechanism to identify the Control plane readiness or is not required to wait for the control
+	// plane to be ready, such providers can simply add a no-op workflow for this function call.
+	// Returning an error message from this handler will stop the workflow of e2e-framework as returning an error from this
+	// is considered as  failure to provision a cluster
+	WaitForControlPlane(ctx context.Context, client klient.Client) error
+
+	// KubernetesRestConfig is a helper function that provides an instance of rest.Config which can then be used to
+	// create your own clients if you chose to do so.
+	KubernetesRestConfig() *rest.Config
+}
+
+type E2EClusterProviderWithImageLoader interface {
+	E2EClusterProvider
+
+	// LoadImage is used to load a set of Docker images to the cluster via the cluster provider native workflow
+	// Not every provider will have a mechanism like this/need to do this. So, providers that do not have this support
+	// can just provide a no-op implementation to be compliant with the interface
+	LoadImage(ctx context.Context, image string, args ...string) error
+
+	// LoadImageArchive is used to provide a mechanism where a tar.gz archive containing the docker images used
+	// by the services running on the cluster can be imported and loaded into the cluster prior to the execution of
+	// test if required.
+	// Not every provider will have a mechanism like this/need to do this. So, providers that do not have this support
+	// can just provide a no-op implementation to be compliant with the interface
+	LoadImageArchive(ctx context.Context, archivePath string, args ...string) error
+}
+
+// E2EClusterProviderWithLifeCycle is an interface that extends the E2EClusterProviderWithImageLoader
+// interface to provide a mechanism to add/remove nodes from the cluster as part of the E2E Test workflow.
+//
+// This can be useful while performing the e2e test that revolves around the node lifecycle events.
+// eg: You have a kubernetes controller that acts upon the v1.Node resource of the k8s and you want to
+// test out how the Remove operation impacts your workflow.
+// Or you want to simulate a case where one or more node of your cluster is down and you want to see how
+// your application reacts to such failure events.
+type E2EClusterProviderWithLifeCycle interface {
+	E2EClusterProvider
+
+	// AddNode is used to add a new node to the existing cluster as part of the E2E Test workflow.
+	// Not every provider will have a mechanism to support this. e.g Kind. But k3d has support for this.
+	// This will be implemented as an optional interface depending on the provider in question.
+	AddNode(ctx context.Context, node *Node, args ...string) error
+
+	// RemoveNode can be used to remove a node from an existing cluster as part of the E2E Test workflow.
+	// Not every provider will have a mechanism to support this. e.g Kind. But k3d has support for this.
+	// This will be implemented as an optional interface depending on the provider in question.
+	RemoveNode(ctx context.Context, node *Node, args ...string) error
+
+	// StartNode is used to start a node that was shutdown/powered down as part of the E2E Test workflow.
+	// Not every provider will have a mechanism to support this. e.g Kind. But k3d has support for this.
+	// This will be implemented as an optional interface depending on the provider in question.
+	StartNode(ctx context.Context, node *Node, args ...string) error
+
+	// StopNode can be used to stop an running node from the cluster as part of the E2E test Workflow.
+	// Not every provider will have a mechanism to support this. e.g Kind. But k3d has support for this.
+	// This will be implemented as an optional interface depending on the provider in question.
+	StopNode(ctx context.Context, node *Node, args ...string) error
+
+	// ListNode can be used to fetch the list of nodes in the cluster. This can be used to extract the
+	// List of existing nodes on the cluster and their state before they can be operated on.
+	ListNode(ctx context.Context, args ...string) ([]Node, error)
+}
diff --git a/support/utils/command.go b/pkg/utils/command.go
similarity index 54%
rename from support/utils/command.go
rename to pkg/utils/command.go
index f23aabeb..0d9724b3 100644
--- a/support/utils/command.go
+++ b/pkg/utils/command.go
@@ -17,6 +17,7 @@ limitations under the License.
 package utils
 
 import (
+	"bytes"
 	"fmt"
 	"io"
 
@@ -34,41 +35,57 @@ var commandRunner = gexe.New()
 // be set in the in the invoker to make sure the right path is used for the binaries while invoking
 // rest of the workfow after this helper is triggered.
 func FindOrInstallGoBasedProvider(pPath, provider, module, version string) (string, error) {
-	if commandRunner.Prog().Avail(pPath) != "" {
+	if gexe.ProgAvail(pPath) != "" {
 		log.V(4).InfoS("Found Provider tooling already installed on the machine", "command", pPath)
 		return pPath, nil
 	}
 
+	var stdout, stderr bytes.Buffer
 	installCommand := fmt.Sprintf("go install %s@%s", module, version)
 	log.V(4).InfoS("Installing provider tooling using go install", "command", installCommand)
-	p := commandRunner.RunProc(installCommand)
-	if p.Err() != nil {
-		return "", fmt.Errorf("failed to install %s: %s", pPath, p.Err())
+	p := commandRunner.NewProc(installCommand)
+	p.SetStdout(&stdout)
+	p.SetStderr(&stderr)
+	result := p.Run()
+	if result.Err() != nil {
+		return "", fmt.Errorf("failed to install %s: %s: \n %s", pPath, result.Result(), stderr.String())
 	}
 
-	if !p.IsSuccess() || p.ExitCode() != 0 {
-		return "", fmt.Errorf("failed to install %s: %s", pPath, p.Result())
+	if !result.IsSuccess() || result.ExitCode() != 0 {
+		return "", fmt.Errorf("failed to install %s: %s \n %s", pPath, result.Result(), stderr.String())
 	}
 
-	if providerPath := commandRunner.Prog().Avail(provider); providerPath != "" {
+	log.V(4).InfoS("Installed provider tooling using go install", "command", installCommand, "output", stdout.String())
+
+	if providerPath := gexe.ProgAvail(provider); providerPath != "" {
 		log.V(4).Infof("Installed %s at %s", pPath, providerPath)
 		return provider, nil
 	}
 
-	p = commandRunner.RunProc("ls $GOPATH/bin")
-	if p.Err() != nil {
-		return "", fmt.Errorf("failed to install %s: %s", pPath, p.Err())
+	p = commandRunner.NewProc("ls $GOPATH/bin")
+	stdout.Reset()
+	stderr.Reset()
+	p.SetStdout(&stdout)
+	p.SetStderr(&stderr)
+	result = p.Run()
+	if result.Err() != nil {
+		return "", fmt.Errorf("failed to install %s: %s \n %ss", pPath, result.Result(), stderr.String())
 	}
 
-	p = commandRunner.RunProc("echo $PATH:$GOPATH/bin")
-	if p.Err() != nil {
-		return "", fmt.Errorf("failed to install %s: %s", pPath, p.Err())
+	p = commandRunner.NewProc("echo $PATH:$GOPATH/bin")
+	stdout.Reset()
+	stderr.Reset()
+	p.SetStdout(&stdout)
+	p.SetStderr(&stderr)
+	result = p.Run()
+	if result.Err() != nil {
+		return "", fmt.Errorf("failed to install %s: %s \n %s", pPath, result.Result(), stderr.String())
 	}
 
-	log.V(4).Info(`Setting path to include $GOPATH/bin:`, p.Result())
-	commandRunner.SetEnv("PATH", p.Result())
+	log.V(4).Info(`Setting path to include $GOPATH/bin:`, result.Result())
+	commandRunner.SetEnv("PATH", result.Result())
 
-	if providerPath := commandRunner.Prog().Avail(provider); providerPath != "" {
+	if providerPath := gexe.ProgAvail(provider); providerPath != "" {
 		log.V(4).Infof("Installed %s at %s", pPath, providerPath)
 		return provider, nil
 	}
@@ -92,7 +109,26 @@ func RunCommandWithSeperatedOutput(command string, stdout, stderr io.Writer) err
 	return result.Err()
 }
 
+// RunCommandWithCustomWriter run command and returns an *exec.Proc with information about the executed process.
+// This helps map the STDOUT/STDERR to custom writer to extract data from the output.
+func RunCommandWithCustomWriter(command string, stdout, stderr io.Writer) *exec.Proc {
+	p := commandRunner.NewProc(command)
+	p.SetStdout(stdout)
+	p.SetStderr(stderr)
+	return p.Run()
+}
+
 // FetchCommandOutput run command and returns the combined stderr and stdout output.
 func FetchCommandOutput(command string) string {
 	return commandRunner.Run(command)
 }
+
+// FetchSeperatedCommandOutput run command and returns the command by splitting the stdout and stderr
+// into different buffers and returns the Process with the buffer that can be ready from to extract
+// the data set on the respective buffers
+func FetchSeperatedCommandOutput(command string) (p *exec.Proc, stdout, stderr bytes.Buffer) {
+	p = commandRunner.NewProc(command)
+	p.SetStdout(&stdout)
+	p.SetStderr(&stderr)
+	return p.Run(), stdout, stderr
+}
diff --git a/pkg/utils/nodelifecycle.go b/pkg/utils/nodelifecycle.go
new file mode 100644
index 00000000..f0ae1e34
--- /dev/null
+++ b/pkg/utils/nodelifecycle.go
@@ -0,0 +1,51 @@
+/*
+Copyright 2024 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 utils
+
+import (
+	"context"
+	"fmt"
+
+	"sigs.k8s.io/e2e-framework/pkg/types"
+)
+
+// PerformNodeLifecycleOperation performs a node operation on a cluster. These operations can range from Add/Remove/Start/Stop.
+// This helper is re-used in both node lifecycle handler used as types.StepFunc or env.Func
+func PerformNodeLifecycleOperation(ctx context.Context, action types.NodeOperation, node *types.Node, args ...string) error {
+	clusterVal := ctx.Value(types.ClusterNameContextKey(node.Cluster))
+	if clusterVal == nil {
+		return fmt.Errorf("%s node to cluster: context cluster is nil", action)
+	}
+
+	clusterProvider, ok := clusterVal.(types.E2EClusterProviderWithLifeCycle)
+	if !ok {
+		return fmt.Errorf("cluster provider %s doesn't support node lifecycle operations", node.Cluster)
+	}
+
+	switch action {
+	case types.AddNode:
+		return clusterProvider.AddNode(ctx, node, args...)
+	case types.RemoveNode:
+		return clusterProvider.RemoveNode(ctx, node, args...)
+	case types.StartNode:
+		return clusterProvider.StartNode(ctx, node, args...)
+	case types.StopNode:
+		return clusterProvider.StopNode(ctx, node, args...)
+	default:
+		return fmt.Errorf("unknown node operation: %s", action)
+	}
+}
diff --git a/support/README.md b/support/README.md
new file mode 100644
index 00000000..7768e185
--- /dev/null
+++ b/support/README.md
@@ -0,0 +1,3 @@
+### This entire package and it's sub packages are present only to account for backward compatiblity.
+
+Any new Provider should be added under `third_party/<provider>`
\ No newline at end of file
diff --git a/support/k3d/k3d.go b/support/k3d/k3d.go
new file mode 100644
index 00000000..b14eae87
--- /dev/null
+++ b/support/k3d/k3d.go
@@ -0,0 +1,30 @@
+/*
+Copyright 2024 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 k3d
+
+import (
+	tptk3d "sigs.k8s.io/e2e-framework/third_party/k3d"
+)
+
+type Cluster = tptk3d.Cluster
+
+var (
+	WithArgs    = tptk3d.WithArgs
+	WithImage   = tptk3d.WithImage
+	NewCluster  = tptk3d.NewCluster
+	NewProvider = tptk3d.NewProvider
+)
diff --git a/support/kind/kind.go b/support/kind/kind.go
index 4ca3d2a1..e39e623d 100644
--- a/support/kind/kind.go
+++ b/support/kind/kind.go
@@ -17,293 +17,14 @@ limitations under the License.
 package kind
 
 import (
-	"bytes"
-	"context"
-	"fmt"
-	"io"
-	"os"
-	"strings"
-
-	v1 "k8s.io/api/core/v1"
-	"k8s.io/client-go/rest"
-	log "k8s.io/klog/v2"
-	"sigs.k8s.io/e2e-framework/klient"
-	"sigs.k8s.io/e2e-framework/klient/conf"
-	"sigs.k8s.io/e2e-framework/klient/k8s/resources"
-	"sigs.k8s.io/e2e-framework/klient/wait"
-	"sigs.k8s.io/e2e-framework/klient/wait/conditions"
-	"sigs.k8s.io/e2e-framework/support"
-	"sigs.k8s.io/e2e-framework/support/utils"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	tptkind "sigs.k8s.io/e2e-framework/third_party/kind"
 )
 
-var kindVersion = "v0.17.0"
-
-type Cluster struct {
-	path        string
-	name        string
-	kubecfgFile string
-	version     string
-	image       string
-	rc          *rest.Config
-}
-
-// Enforce Type check always to avoid future breaks
-var _ support.E2EClusterProvider = &Cluster{}
-
-func NewCluster(name string) *Cluster {
-	return &Cluster{name: name}
-}
-
-func NewProvider() support.E2EClusterProvider {
-	return &Cluster{}
-}
-
-func WithImage(image string) support.ClusterOpts {
-	return func(c support.E2EClusterProvider) {
-		k, ok := c.(*Cluster)
-		if ok {
-			k.image = image
-		}
-	}
-}
-
-func WithPath(path string) support.ClusterOpts {
-	return func(c support.E2EClusterProvider) {
-		k, ok := c.(*Cluster)
-		if ok {
-			k.path = path
-		}
-	}
-}
-
-func (k *Cluster) SetDefaults() support.E2EClusterProvider {
-	if k.path == "" {
-		k.path = "kind"
-	}
-	return k
-}
-
-func (k *Cluster) WithName(name string) support.E2EClusterProvider {
-	k.name = name
-	return k
-}
-
-func (k *Cluster) WithPath(path string) support.E2EClusterProvider {
-	k.path = path
-	return k
-}
-
-func (k *Cluster) WithVersion(ver string) support.E2EClusterProvider {
-	k.version = ver
-	return k
-}
-
-func (k *Cluster) WithOpts(opts ...support.ClusterOpts) support.E2EClusterProvider {
-	for _, o := range opts {
-		o(k)
-	}
-	return k
-}
-
-func (k *Cluster) getKubeconfig() (string, error) {
-	kubecfg := fmt.Sprintf("%s-kubecfg", k.name)
-
-	var stdout, stderr bytes.Buffer
-	err := utils.RunCommandWithSeperatedOutput(fmt.Sprintf(`%s get kubeconfig --name %s`, k.path, k.name), &stdout, &stderr)
-	if err != nil {
-		return "", fmt.Errorf("kind get kubeconfig: stderr: %s: %w", stderr.String(), err)
-	}
-	log.V(4).Info("kind get kubeconfig stderr \n", stderr.String())
-
-	file, err := os.CreateTemp("", fmt.Sprintf("kind-cluster-%s", kubecfg))
-	if err != nil {
-		return "", fmt.Errorf("kind kubeconfig file: %w", err)
-	}
-	defer file.Close()
-
-	k.kubecfgFile = file.Name()
-
-	if n, err := io.WriteString(file, stdout.String()); n == 0 || err != nil {
-		return "", fmt.Errorf("kind kubecfg file: bytes copied: %d: %w]", n, err)
-	}
-
-	return file.Name(), nil
-}
-
-func (k *Cluster) clusterExists(name string) (string, bool) {
-	clusters := utils.FetchCommandOutput(fmt.Sprintf("%s get clusters", k.path))
-	for _, c := range strings.Split(clusters, "\n") {
-		if c == name {
-			return clusters, true
-		}
-	}
-	return clusters, false
-}
-
-func (k *Cluster) CreateWithConfig(ctx context.Context, kindConfigFile string) (string, error) {
-	var args []string
-	if kindConfigFile != "" {
-		args = append(args, "--config", kindConfigFile)
-	}
-	return k.Create(ctx, args...)
-}
+type Cluster = tptkind.Cluster
 
-func (k *Cluster) Create(ctx context.Context, args ...string) (string, error) {
-	log.V(4).Info("Creating kind cluster ", k.name)
-	if err := k.findOrInstallKind(); err != nil {
-		return "", err
-	}
-
-	if _, ok := k.clusterExists(k.name); ok {
-		log.V(4).Info("Skipping Kind Cluster.Create: cluster already created: ", k.name)
-		kConfig, err := k.getKubeconfig()
-		if err != nil {
-			return "", err
-		}
-		return kConfig, k.initKubernetesAccessClients()
-	}
-
-	if k.image != "" {
-		args = append(args, "--image", k.image)
-	}
-
-	command := fmt.Sprintf(`%s create cluster --name %s`, k.path, k.name)
-	if len(args) > 0 {
-		command = fmt.Sprintf("%s %s", command, strings.Join(args, " "))
-	}
-	log.V(4).Info("Launching:", command)
-	p := utils.RunCommand(command)
-	if p.Err() != nil {
-		outBytes, err := io.ReadAll(p.Out())
-		if err != nil {
-			log.ErrorS(err, "failed to read data from the kind create process output due to an error")
-		}
-		return "", fmt.Errorf("kind: failed to create cluster %q: %s: %s: %s", k.name, p.Err(), p.Result(), string(outBytes))
-	}
-	clusters, ok := k.clusterExists(k.name)
-	if !ok {
-		return "", fmt.Errorf("kind Cluster.Create: cluster %v still not in 'cluster list' after creation: %v", k.name, clusters)
-	}
-	log.V(4).Info("kind clusters available: ", clusters)
-
-	kConfig, err := k.getKubeconfig()
-	if err != nil {
-		return "", err
-	}
-	return kConfig, k.initKubernetesAccessClients()
-}
-
-func (k *Cluster) initKubernetesAccessClients() error {
-	cfg, err := conf.New(k.kubecfgFile)
-	if err != nil {
-		return err
-	}
-	k.rc = cfg
-	return nil
-}
-
-func (k *Cluster) GetKubeconfig() string {
-	return k.kubecfgFile
-}
-
-func (k *Cluster) GetKubectlContext() string {
-	return fmt.Sprintf("kind-%s", k.name)
-}
-
-// ExportLogs export all cluster logs to the provided path.
-func (k *Cluster) ExportLogs(ctx context.Context, dest string) error {
-	log.V(4).Info("Exporting kind cluster logs to ", dest)
-	if err := k.findOrInstallKind(); err != nil {
-		return err
-	}
-
-	p := utils.RunCommand(fmt.Sprintf(`%s export logs %s --name %s`, k.path, dest, k.name))
-	if p.Err() != nil {
-		return fmt.Errorf("kind: export cluster %v logs failed: %s: %s", k.name, p.Err(), p.Result())
-	}
-
-	return nil
-}
-
-func (k *Cluster) Destroy(ctx context.Context) error {
-	log.V(4).Info("Destroying kind cluster ", k.name)
-	if err := k.findOrInstallKind(); err != nil {
-		return err
-	}
-
-	p := utils.RunCommand(fmt.Sprintf(`%s delete cluster --name %s`, k.path, k.name))
-	if p.Err() != nil {
-		outBytes, err := io.ReadAll(p.Out())
-		if err != nil {
-			log.ErrorS(err, "failed to read data from the kind delete process output due to an error")
-		}
-		return fmt.Errorf("kind: failed to delete cluster %q: %s: %s: %s", k.name, p.Err(), p.Result(), string(outBytes))
-	}
-
-	log.V(4).Info("Removing kubeconfig file ", k.kubecfgFile)
-	if err := os.RemoveAll(k.kubecfgFile); err != nil {
-		return fmt.Errorf("kind: remove kubefconfig %v failed: %w", k.kubecfgFile, err)
-	}
-
-	return nil
-}
-
-func (k *Cluster) findOrInstallKind() error {
-	if k.version != "" {
-		kindVersion = k.version
-	}
-	path, err := utils.FindOrInstallGoBasedProvider(k.path, "kind", "sigs.k8s.io/kind", kindVersion)
-	if path != "" {
-		k.path = path
-	}
-	return err
-}
-
-func (k *Cluster) LoadImage(ctx context.Context, image string) error {
-	p := utils.RunCommand(fmt.Sprintf(`%s load docker-image --name %s %s`, k.path, k.name, image))
-	if p.Err() != nil {
-		return fmt.Errorf("kind: load docker-image %v failed: %s: %s", image, p.Err(), p.Result())
-	}
-	return nil
-}
-
-func (k *Cluster) LoadImageArchive(ctx context.Context, imageArchive string) error {
-	p := utils.RunCommand(fmt.Sprintf(`%s load image-archive --name %s %s`, k.path, k.name, imageArchive))
-	if p.Err() != nil {
-		return fmt.Errorf("kind: load image-archive %v failed: %s: %s", imageArchive, p.Err(), p.Result())
-	}
-	return nil
-}
-
-func (k *Cluster) WaitForControlPlane(ctx context.Context, client klient.Client) error {
-	r, err := resources.New(client.RESTConfig())
-	if err != nil {
-		return err
-	}
-	for _, sl := range []metav1.LabelSelectorRequirement{
-		{Key: "component", Operator: metav1.LabelSelectorOpIn, Values: []string{"etcd", "kube-apiserver", "kube-controller-manager", "kube-scheduler"}},
-		{Key: "k8s-app", Operator: metav1.LabelSelectorOpIn, Values: []string{"kindnet", "kube-dns", "kube-proxy"}},
-	} {
-		selector, err := metav1.LabelSelectorAsSelector(
-			&metav1.LabelSelector{
-				MatchExpressions: []metav1.LabelSelectorRequirement{
-					sl,
-				},
-			},
-		)
-		if err != nil {
-			return err
-		}
-		err = wait.For(conditions.New(r).ResourceListN(&v1.PodList{}, len(sl.Values), resources.WithLabelSelector(selector.String())))
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (k *Cluster) KubernetesRestConfig() *rest.Config {
-	return k.rc
-}
+var (
+	NewCluster  = tptkind.NewCluster
+	NewProvider = tptkind.NewProvider
+	WithImage   = tptkind.WithImage
+	WithPath    = tptkind.WithPath
+)
diff --git a/support/kwok/kwok.go b/support/kwok/kwok.go
index 7ec07ada..692c0215 100644
--- a/support/kwok/kwok.go
+++ b/support/kwok/kwok.go
@@ -17,257 +17,13 @@ limitations under the License.
 package kwok
 
 import (
-	"bytes"
-	"context"
-	"fmt"
-	"io"
-	"os"
-	"path/filepath"
-	"strings"
-	"time"
-
-	"k8s.io/client-go/rest"
-	klog "k8s.io/klog/v2"
-	"sigs.k8s.io/e2e-framework/klient"
-	"sigs.k8s.io/e2e-framework/klient/conf"
-	"sigs.k8s.io/e2e-framework/support"
-	"sigs.k8s.io/e2e-framework/support/utils"
+	tptkwok "sigs.k8s.io/e2e-framework/third_party/kwok"
 )
 
-var kwokVersion = "v0.5.0"
-
-type Cluster struct {
-	name         string
-	path         string
-	kubecfgFile  string
-	version      string
-	waitDuration time.Duration
-	rc           *rest.Config
-}
-
-var _ support.E2EClusterProvider = &Cluster{}
-
-func NewCluster(name string) *Cluster {
-	return &Cluster{name: name, waitDuration: 1 * time.Minute}
-}
-
-func NewProvider() support.E2EClusterProvider {
-	return &Cluster{}
-}
-
-func WithPath(path string) support.ClusterOpts {
-	return func(c support.E2EClusterProvider) {
-		k, ok := c.(*Cluster)
-		if ok {
-			k.path = path
-		}
-	}
-}
-
-func (k *Cluster) findOrInstallKwokCtl() error {
-	if k.version != "" {
-		kwokVersion = k.version
-	}
-	path, err := utils.FindOrInstallGoBasedProvider(k.path, "kwokctl", "sigs.k8s.io/kwok/cmd/kwokctl", kwokVersion)
-	if path != "" {
-		k.path = path
-	}
-	return err
-}
-
-func (k *Cluster) clusterExists(name string) (string, bool) {
-	clusters := utils.FetchCommandOutput(fmt.Sprintf("%s get clusters", k.path))
-	for _, c := range strings.Split(clusters, "\n") {
-		if c == name {
-			return clusters, true
-		}
-	}
-	return clusters, false
-}
-
-func (k *Cluster) getKubeconfig() (string, error) {
-	kubecfg := fmt.Sprintf("%s-kubecfg", k.name)
-
-	var stdout, stderr bytes.Buffer
-	err := utils.RunCommandWithSeperatedOutput(fmt.Sprintf(`%s get kubeconfig --name %s`,
-		k.path, k.name), &stdout, &stderr)
-	if err != nil {
-		return "", fmt.Errorf("kwokctl get kubeconfig: stderr: %s: %w", stderr.String(), err)
-	}
-
-	file, err := os.CreateTemp("", fmt.Sprintf("kwok-cluster-%s", kubecfg))
-	if err != nil {
-		return "", fmt.Errorf("kwok kubeconfig file: %w", err)
-	}
-	defer file.Close()
-
-	k.kubecfgFile = file.Name()
-
-	if n, err := io.WriteString(file, stdout.String()); n == 0 || err != nil {
-		return "", fmt.Errorf("kwok kubecfg file: bytes copied: %d: %w]", n, err)
-	}
-
-	return file.Name(), nil
-}
-
-func (k *Cluster) initKubernetesAccessClients() error {
-	cfg, err := conf.New(k.kubecfgFile)
-	if err != nil {
-		return err
-	}
-	k.rc = cfg
-	return nil
-}
-
-func (k *Cluster) Create(ctx context.Context, args ...string) (string, error) {
-	klog.V(4).Info("Creating a kwok cluster ", k.name)
-	if err := k.findOrInstallKwokCtl(); err != nil {
-		return "", err
-	}
-	if _, ok := k.clusterExists(k.name); ok {
-		klog.V(4).Info("Skipping Kwok Cluster creation. Cluster already created ", k.name)
-		kConfig, err := k.getKubeconfig()
-		if err != nil {
-			return "", err
-		}
-		return kConfig, k.initKubernetesAccessClients()
-	}
-
-	command := fmt.Sprintf(`%s create cluster --name %s --wait %s`, k.path, k.name, k.waitDuration.String())
-	if len(args) > 0 {
-		command = fmt.Sprintf("%s %s", command, strings.Join(args, " "))
-	}
-	klog.V(4).Info("Launching:", command)
-	p := utils.RunCommand(command)
-	if p.Err() != nil {
-		outBytes, err := io.ReadAll(p.Out())
-		if err != nil {
-			klog.ErrorS(err, "failed to read data from the kwok create process output due to an error")
-		}
-		return "", fmt.Errorf("kwok: failed to create cluster %q: %s: %s: %s", k.name, p.Err(), p.Result(), string(outBytes))
-	}
-
-	clusters, ok := k.clusterExists(k.name)
-	if !ok {
-		return "", fmt.Errorf("kwok Cluster.Create: cluster %v still not in 'cluster list' after creation: %v", k.name, clusters)
-	}
-	klog.V(4).Info("kwok cluster available: ", clusters)
+type Cluster = tptkwok.Cluster
 
-	kConfig, err := k.getKubeconfig()
-	if err != nil {
-		return "", err
-	}
-	return kConfig, k.initKubernetesAccessClients()
-}
-
-func (k *Cluster) CreateWithConfig(ctx context.Context, configFile string) (string, error) {
-	if configFile == "" {
-		return k.Create(ctx)
-	}
-
-	return k.Create(ctx, "--config", configFile)
-}
-
-func (k *Cluster) Destroy(ctx context.Context) error {
-	klog.V(4).Info("Destroying kwok cluster ", k.name)
-	if err := k.findOrInstallKwokCtl(); err != nil {
-		return err
-	}
-
-	p := utils.RunCommand(fmt.Sprintf(`%s delete cluster --name %s`, k.path, k.name))
-	if p.Err() != nil {
-		outBytes, err := io.ReadAll(p.Out())
-		if err != nil {
-			klog.ErrorS(err, "failed to read data from the kwok delete process output due to an error")
-		}
-		return fmt.Errorf("kwok: failed to delete cluster %q: %s: %s: %s", k.name, p.Err(), p.Result(), string(outBytes))
-	}
-
-	klog.V(4).Info("Removing kubeconfig file ", k.kubecfgFile)
-	if err := os.RemoveAll(k.kubecfgFile); err != nil {
-		return fmt.Errorf("kwok: remove kubefconfig failed: %w", err)
-	}
-	return nil
-}
-
-func (k *Cluster) ExportLogs(ctx context.Context, dest string) error {
-	if err := k.findOrInstallKwokCtl(); err != nil {
-		return err
-	}
-	// In kwokctl 0.3.0 and above, there is a new kwokctl export logs feature that has been added which can
-	// simplify the workf of exporting the logs for us. Let us check if the CLI has that command and if so
-	// let us use that to export logs. Otherwise, we can fallback to exporting individual items.
-	p := utils.RunCommand(fmt.Sprintf("%s export logs --help", k.path))
-	if p.ExitCode() == 0 {
-		return utils.RunCommand(fmt.Sprintf("%s --name %s export logs %s", k.path, k.name, dest)).Err()
-	}
-
-	// TODO: Get Rid of this if we decide to enforce a min version of the kwokctl at some point
-	for _, component := range []string{"audit", "etcd", "kube-apiserver", "kube-controller-manager", "kube-scheduler", "kwok-controller", "prometheus"} {
-		command := fmt.Sprintf("%s logs %s", k.path, component)
-		p := utils.RunCommand(command)
-		if p.Err() != nil {
-			klog.ErrorS(p.Err(), "ran into an error trying to export the log", "component", component)
-			continue
-		}
-		var stdout bytes.Buffer
-		if _, err := stdout.ReadFrom(p.Out()); err != nil {
-			return fmt.Errorf("kwokctl logs %s stdout bytes: %w", component, err)
-		}
-		file, err := os.Create(filepath.Join(dest, fmt.Sprintf("%s.log", component)))
-		if err != nil {
-			klog.ErrorS(err, "ran into an error trying to create file to export logs", "component", component)
-			continue
-		}
-		if n, err := io.Copy(file, &stdout); n == 0 || err != nil {
-			klog.ErrorS(err, "kwokctl logs %s file: bytes copied: %d: %w]", component, n, err)
-		}
-	}
-	return nil
-}
-
-func (k *Cluster) GetKubectlContext() string {
-	return fmt.Sprintf("kwok-%s", k.name)
-}
-
-func (k *Cluster) GetKubeconfig() string {
-	return k.kubecfgFile
-}
-
-func (k *Cluster) SetDefaults() support.E2EClusterProvider {
-	if k.path == "" {
-		k.path = "kwokctl"
-	}
-	return k
-}
-
-func (k *Cluster) WaitForControlPlane(ctx context.Context, client klient.Client) error {
-	klog.V(4).Info("kwokctl doesn't implement a WaitForControlPlane handler. The --wait argument passed to the `kwokctl` should take care of this already")
-	return nil
-}
-
-func (k *Cluster) WithName(name string) support.E2EClusterProvider {
-	k.name = name
-	return k
-}
-
-func (k *Cluster) WithOpts(opts ...support.ClusterOpts) support.E2EClusterProvider {
-	for _, o := range opts {
-		o(k)
-	}
-	return k
-}
-
-func (k *Cluster) WithPath(path string) support.E2EClusterProvider {
-	k.path = path
-	return k
-}
-
-func (k *Cluster) WithVersion(version string) support.E2EClusterProvider {
-	k.version = version
-	return k
-}
-
-func (k *Cluster) KubernetesRestConfig() *rest.Config {
-	return k.rc
-}
+var (
+	NewCluster  = tptkwok.NewCluster
+	NewProvider = tptkwok.NewProvider
+	WithPath    = tptkwok.WithPath
+)
diff --git a/support/types.go b/support/types.go
index 25fcae26..4c555beb 100644
--- a/support/types.go
+++ b/support/types.go
@@ -17,89 +17,22 @@ limitations under the License.
 package support
 
 import (
-	"context"
-
-	"k8s.io/client-go/rest"
-	"sigs.k8s.io/e2e-framework/klient"
+	"sigs.k8s.io/e2e-framework/pkg/types"
 )
 
-type ClusterOpts func(c E2EClusterProvider)
-
-type E2EClusterProvider interface {
-	// WithName is used to configure the cluster Name that should be used while setting up the cluster. Might
-	// Not apply for all providers.
-	WithName(name string) E2EClusterProvider
-
-	// WithVersion helps you override the default version used while using the cluster provider.
-	// This can be useful in providing a mechanism to the end users where they want to test their
-	// code against a certain specific version of k8s that is not the default one configured
-	// for the provider
-	WithVersion(version string) E2EClusterProvider
-
-	// WithPath heps you customize the executable binary that is used to back the cluster provider.
-	// This is useful in cases where your binary is present in a non standard location output of the
-	// PATH variable and you want to use that instead of framework trying to install one on it's own.
-	WithPath(path string) E2EClusterProvider
-
-	// WithOpts provides a way to customize the options that can be used while setting up the
-	// cluster using the providers such as kind or kwok or anything else. These helpers can be
-	// leveraged to setup arguments or configuration values that can be provided while performing
-	// the cluster bring up
-	WithOpts(opts ...ClusterOpts) E2EClusterProvider
-
-	// Create Provides an interface to start the cluster creation workflow using the selected provider
-	Create(ctx context.Context, args ...string) (string, error)
-
-	// CreateWithConfig is used to provide a mechanism where cluster providers that take an input config
-	// file and then setup the cluster accordingly. This can be used to provide input such as kind config
-	CreateWithConfig(ctx context.Context, configFile string) (string, error)
-
-	// GetKubeconfig provides a way to extract the kubeconfig file associated with the cluster in question
-	// using the cluster provider native way
-	GetKubeconfig() string
-
-	// GetKubectlContext is used to extract the kubectl context to be used while performing the operation
-	GetKubectlContext() string
-
-	// ExportLogs is used to export the cluster logs via the cluster provider native workflow. This
-	// can be used to export logs from the cluster after test failures for example to analyze the test
-	// failures better after the fact.
-	ExportLogs(ctx context.Context, dest string) error
-
-	// Destroy is used to cleanup a cluster brought up as part of the test workflow
-	Destroy(ctx context.Context) error
-
-	// SetDefaults is a handler function invoked after creating an object of type E2EClusterProvider. This method is
-	// invoked as the first step after creating an object in order to make sure the default values for required
-	// attributes are setup accordingly if any.
-	SetDefaults() E2EClusterProvider
-
-	// WaitForControlPlane is a helper function that can be used to indiate the Provider based cluster create workflow
-	// that the control plane is fully up and running. This method is invoked after the Create/CreateWithConfig handlers
-	// and is expected to return an error if the control plane doesn't stabilize. If the provider being implemented
-	// does not have a clear mechanism to identify the Control plane readiness or is not required to wait for the control
-	// plane to be ready, such providers can simply add a no-op workflow for this function call.
-	// Returning an error message from this handler will stop the workflow of e2e-framework as returning an error from this
-	// is considered as  failure to provision a cluster
-	WaitForControlPlane(ctx context.Context, client klient.Client) error
-
-	// KubernetesRestConfig is a helper function that provides an instance of rest.Config which can then be used to
-	// create your own clients if you chose to do so.
-	KubernetesRestConfig() *rest.Config
-}
-
-type E2EClusterProviderWithImageLoader interface {
-	E2EClusterProvider
-
-	// LoadImage is used to load a set of Docker images to the cluster via the cluster provider native workflow
-	// Not every provider will have a mechanism like this/need to do this. So, providers that do not have this support
-	// can just provide a no-op implementation to be compliant with the interface
-	LoadImage(ctx context.Context, image string) error
+type (
+	ClusterOpts                       = types.ClusterOpts
+	Node                              = types.Node
+	NodeOperation                     = types.NodeOperation
+	ClusterNameContextKey             = types.ClusterNameContextKey
+	E2EClusterProvider                = types.E2EClusterProvider
+	E2EClusterProviderWithImageLoader = types.E2EClusterProviderWithImageLoader
+	E2EClusterProviderWithLifeCycle   = types.E2EClusterProviderWithLifeCycle
+)
 
-	// LoadImageArchive is used to provide a mechanism where a tar.gz archive containing the docker images used
-	// by the services running on the cluster can be imported and loaded into the cluster prior to the execution of
-	// test if required.
-	// Not every provider will have a mechanism like this/need to do this. So, providers that do not have this support
-	// can just provide a no-op implementation to be compliant with the interface
-	LoadImageArchive(ctx context.Context, archivePath string) error
-}
+const (
+	AddNode    = types.AddNode
+	RemoveNode = types.RemoveNode
+	StartNode  = types.StartNode
+	StopNode   = types.StopNode
+)
diff --git a/third_party/k3d/k3d.go b/third_party/k3d/k3d.go
new file mode 100644
index 00000000..4a033597
--- /dev/null
+++ b/third_party/k3d/k3d.go
@@ -0,0 +1,400 @@
+/*
+Copyright 2024 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 k3d
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/util/json"
+
+	"k8s.io/client-go/rest"
+
+	"sigs.k8s.io/e2e-framework/klient"
+	"sigs.k8s.io/e2e-framework/klient/conf"
+	"sigs.k8s.io/e2e-framework/pkg/utils"
+	"sigs.k8s.io/e2e-framework/support"
+
+	log "k8s.io/klog/v2"
+)
+
+var k3dVersion = "v5.7.2"
+
+type Cluster struct {
+	path           string
+	name           string
+	kubeConfigFile string
+	version        string
+	image          string
+	rc             *rest.Config
+	args           []string
+}
+
+// k3dNode is a struct containing a subset of values that are part of the k3d node list -o json
+// data. This currently contains only those fields that are of interest to generate the
+// support.Node struct in return for performing node operations.
+type k3dNode struct {
+	Name string `json:"name"`
+	Role string `json:"role"`
+	IP   struct {
+		IP string `json:"IP"`
+	} `json:"IP"`
+	State struct {
+		Running bool   `json:"Running"`
+		Status  string `json:"Status"`
+	} `json:"State"`
+}
+
+var (
+	_ support.E2EClusterProviderWithImageLoader = &Cluster{}
+	_ support.E2EClusterProviderWithLifeCycle   = &Cluster{}
+)
+
+func WithArgs(args ...string) support.ClusterOpts {
+	return func(c support.E2EClusterProvider) {
+		k, ok := c.(*Cluster)
+		if ok {
+			k.args = append(k.args, args...)
+		}
+	}
+}
+
+func WithImage(image string) support.ClusterOpts {
+	return func(c support.E2EClusterProvider) {
+		k, ok := c.(*Cluster)
+		if ok {
+			k.image = image
+		}
+	}
+}
+
+func NewCluster(name string) *Cluster {
+	return &Cluster{name: name}
+}
+
+func NewProvider() support.E2EClusterProvider {
+	return &Cluster{}
+}
+
+func (c *Cluster) findOrInstallK3D() error {
+	if c.version != "" {
+		k3dVersion = c.version
+	}
+	path, err := utils.FindOrInstallGoBasedProvider(c.path, "k3d", "github.com/k3d-io/k3d/v5", k3dVersion)
+	if path != "" {
+		c.path = path
+	}
+	return err
+}
+
+func (c *Cluster) getKubeConfig() (string, error) {
+	kubeCfg := fmt.Sprintf("%s-kubecfg", c.name)
+
+	var stdout, stderr bytes.Buffer
+	err := utils.RunCommandWithSeperatedOutput(fmt.Sprintf("%s kubeconfig get %s", c.path, c.name), &stdout, &stderr)
+	if err != nil {
+		return "", fmt.Errorf("failed to get kubeconfig: %s", stderr.String())
+	}
+	log.V(4).Info("k3d kubeconfig get stderr \n", stderr.String())
+
+	file, err := os.CreateTemp("", fmt.Sprintf("k3d-cluster-%s", kubeCfg))
+	if err != nil {
+		return "", fmt.Errorf("k3d kubeconfig file: %w", err)
+	}
+	defer file.Close()
+
+	c.kubeConfigFile = file.Name()
+
+	if n, err := io.WriteString(file, stdout.String()); n == 0 || err != nil {
+		return "", fmt.Errorf("k3d kubeconfig file: bytes copied: %d: %w]", n, err)
+	}
+
+	return file.Name(), nil
+}
+
+func (c *Cluster) clusterExists(name string) (string, bool) {
+	clusters := utils.FetchCommandOutput(fmt.Sprintf("%s cluster get --no-headers", c.path))
+	for _, c := range strings.Split(clusters, "\n") {
+		if strings.HasPrefix(c, name) {
+			return clusters, true
+		}
+	}
+	return clusters, false
+}
+
+func (c *Cluster) startCluster(name string) error {
+	cmd := fmt.Sprintf("%s cluster start %s", c.path, name)
+	log.V(4).InfoS("Starting k3d cluster", "command", cmd)
+	p := utils.RunCommand(cmd)
+	if p.Err() != nil {
+		return fmt.Errorf("k3d: failed to start cluster %q: %s: %s", name, p.Err(), p.Result())
+	}
+	return nil
+}
+
+func (c *Cluster) initKubernetesAccessClients() error {
+	cfg, err := conf.New(c.kubeConfigFile)
+	if err != nil {
+		return err
+	}
+	c.rc = cfg
+	return nil
+}
+
+func (c *Cluster) WithName(name string) support.E2EClusterProvider {
+	c.name = name
+	return c
+}
+
+func (c *Cluster) WithVersion(version string) support.E2EClusterProvider {
+	c.version = version
+	return c
+}
+
+func (c *Cluster) WithPath(path string) support.E2EClusterProvider {
+	c.path = path
+	return c
+}
+
+func (c *Cluster) WithOpts(opts ...support.ClusterOpts) support.E2EClusterProvider {
+	for _, o := range opts {
+		o(c)
+	}
+	return c
+}
+
+func (c *Cluster) Create(ctx context.Context, args ...string) (string, error) {
+	log.V(4).InfoS("Creating k3d cluster", "name", c.name)
+	if err := c.findOrInstallK3D(); err != nil {
+		return "", fmt.Errorf("failed to find or install k3d: %w", err)
+	}
+
+	if _, ok := c.clusterExists(c.name); ok {
+		// This is being done as an extra step to ensure that in case you have the cluster by the same name, but it is not up.
+		// Starting an already started cluster won't cause any harm. So, we will just start it once before continuing
+		// further down the line and process rest of the workflows
+		if err := c.startCluster(c.name); err != nil {
+			return "", err
+		}
+		log.V(4).InfoS("Skipping k3d cluster creation. Cluster already exists", "name", c.name)
+		kConfig, err := c.getKubeConfig()
+		if err != nil {
+			return "", err
+		}
+		return kConfig, c.initKubernetesAccessClients()
+	}
+
+	if c.image != "" {
+		args = append(args, "--image", c.image)
+	}
+
+	args = append(args, c.args...)
+	cmd := fmt.Sprintf("%s cluster create %s", c.path, c.name)
+
+	if len(args) > 0 {
+		cmd = fmt.Sprintf("%s %s", cmd, strings.Join(args, " "))
+	}
+	log.V(4).InfoS("Launching k3d cluster", "command", cmd)
+
+	var stdout, stderr bytes.Buffer
+
+	p := utils.RunCommandWithCustomWriter(cmd, &stdout, &stderr)
+	if p.Err() != nil {
+		return "", fmt.Errorf("k3d: failed to create cluster %q: %s: %s: %s %s", c.name, p.Err(), p.Result(), stdout.String(), stderr.String())
+	}
+	clusters, ok := c.clusterExists(c.name)
+	if !ok {
+		return "", fmt.Errorf("k3d cluster create: cluster %v still not in 'cluster list' after creation: %v", c.name, clusters)
+	}
+	log.V(4).Info("k3d clusters available: ", clusters)
+
+	kConfig, err := c.getKubeConfig()
+	if err != nil {
+		return "", err
+	}
+	return kConfig, c.initKubernetesAccessClients()
+}
+
+func (c *Cluster) CreateWithConfig(ctx context.Context, configFile string) (string, error) {
+	var args []string
+	if configFile != "" {
+		args = append(args, "--config", configFile)
+	}
+	return c.Create(ctx, args...)
+}
+
+func (c *Cluster) GetKubeconfig() string {
+	return c.kubeConfigFile
+}
+
+func (c *Cluster) GetKubectlContext() string {
+	return fmt.Sprintf("k3d-%s", c.name)
+}
+
+func (c *Cluster) ExportLogs(ctx context.Context, dest string) error {
+	log.Warning("ExportLogs not implemented for k3d. Please use regular kubectl like commands to extract the logs from the cluster")
+	return nil
+}
+
+func (c *Cluster) Destroy(ctx context.Context) error {
+	log.V(4).InfoS("Destroying k3d cluster", "name", c.name)
+	if err := c.findOrInstallK3D(); err != nil {
+		return fmt.Errorf("failed to find or install k3d: %w", err)
+	}
+
+	if _, ok := c.clusterExists(c.name); !ok {
+		log.V(4).InfoS("Skipping k3d cluster destruction. Cluster does not exist", "name", c.name)
+		return nil
+	}
+
+	cmd := fmt.Sprintf("%s cluster delete %s", c.path, c.name)
+	log.V(4).InfoS("Destroying k3d cluster", "command", cmd)
+	p := utils.RunCommand(cmd)
+	if p.Err() != nil {
+		outBytes, err := io.ReadAll(p.Out())
+		if err != nil {
+			log.ErrorS(err, "failed to read data from the k3d cluster delete process output due to an error")
+		}
+		return fmt.Errorf("k3d: failed to delete cluster %q: %s: %s: %s", c.name, p.Err(), p.Result(), string(outBytes))
+	}
+
+	log.V(4).InfoS("Removing kubeconfig file", "configFile", c.kubeConfigFile)
+	if err := os.RemoveAll(c.kubeConfigFile); err != nil {
+		return fmt.Errorf("k3d: failed to remove kubeconfig file %q: %w", c.kubeConfigFile, err)
+	}
+	return nil
+}
+
+func (c *Cluster) SetDefaults() support.E2EClusterProvider {
+	if c.path == "" {
+		c.path = "k3d"
+	}
+	return c
+}
+
+func (c *Cluster) WaitForControlPlane(ctx context.Context, client klient.Client) error {
+	log.V(4).Info(" k3d provider doesn't implement a WaitForControlPlane as the provider automatically wait for the control plane")
+	return nil
+}
+
+func (c *Cluster) KubernetesRestConfig() *rest.Config {
+	return c.rc
+}
+
+func (c *Cluster) LoadImage(ctx context.Context, image string, args ...string) error {
+	log.V(4).InfoS("Performing Image load operation", "cluster", c.name, "image", image, "args", args)
+	p := utils.RunCommand(fmt.Sprintf("%s image import --cluster %s %s %s", c.path, c.name, strings.Join(args, " "), image))
+	if p.Err() != nil {
+		return fmt.Errorf("k3d: load docker-image %v failed: %s: %s", image, p.Err(), p.Result())
+	}
+	return nil
+}
+
+func (c *Cluster) LoadImageArchive(ctx context.Context, imageArchive string, args ...string) error {
+	return c.LoadImage(ctx, imageArchive, args...)
+}
+
+func (c *Cluster) AddNode(ctx context.Context, node *support.Node, args ...string) error {
+	cmd := fmt.Sprintf("%s node create %s --cluster %s", c.path, node.Name, c.name)
+
+	if len(args) > 0 {
+		cmd = fmt.Sprintf("%s %s", cmd, strings.Join(args, " "))
+	}
+	if node.Role != "" {
+		cmd = fmt.Sprintf("%s --role %s", cmd, node.Role)
+	}
+
+	log.V(4).InfoS("Adding node to k3d cluster", "command", cmd)
+	p, stdout, stderr := utils.FetchSeperatedCommandOutput(cmd)
+	if p.Err() != nil || (p.Exited() && p.ExitCode() != 0) {
+		log.ErrorS(p.Err(), "failed to add node to k3d cluster", "stdout", stdout.String(), "stderr", stderr.String())
+		return fmt.Errorf("k3d: failed to add node %q to cluster %q: %s: %s", node.Name, c.name, p.Err(), p.Result())
+	}
+	return nil
+}
+
+func (c *Cluster) RemoveNode(ctx context.Context, node *support.Node, args ...string) error {
+	cmd := fmt.Sprintf("%s node delete %s", c.path, node.Name)
+
+	if len(args) > 0 {
+		cmd = fmt.Sprintf("%s %s", cmd, strings.Join(args, " "))
+	}
+
+	log.V(4).InfoS("Removing node from k3d cluster", "command", cmd)
+	p, stdout, stderr := utils.FetchSeperatedCommandOutput(cmd)
+	if p.Err() != nil || (p.Exited() && p.ExitCode() != 0) {
+		log.ErrorS(p.Err(), "failed to remove node from k3d cluster", "stdout", stdout.String(), "stderr", stderr.String())
+		return fmt.Errorf("k3d: failed to remove node %q from cluster %q: %s: %s", node.Name, c.name, p.Err(), p.Result())
+	}
+	return nil
+}
+
+func (c *Cluster) StartNode(ctx context.Context, node *support.Node, args ...string) error {
+	cmd := fmt.Sprintf("%s node start %s", c.path, node.Name)
+	if len(args) > 0 {
+		cmd = fmt.Sprintf("%s %s", cmd, strings.Join(args, " "))
+	}
+	log.V(4).InfoS("Starting node in k3d cluster", "command", cmd)
+	p, stdout, stderr := utils.FetchSeperatedCommandOutput(cmd)
+	if p.Err() != nil || (p.Exited() && p.ExitCode() != 0) {
+		log.ErrorS(p.Err(), "failed to start node in k3d cluster", "stdout", stdout.String(), "stderr", stderr.String())
+		return fmt.Errorf("k3d: failed to start node %q in cluster %q: %s: %s", node.Name, c.name, p.Err(), p.Result())
+	}
+	return nil
+}
+
+func (c *Cluster) StopNode(ctx context.Context, node *support.Node, args ...string) error {
+	cmd := fmt.Sprintf("%s node stop %s", c.path, node.Name)
+	if len(args) > 0 {
+		cmd = fmt.Sprintf("%s %s", cmd, strings.Join(args, " "))
+	}
+	log.V(4).InfoS("Stopping node in k3d cluster", "command", cmd)
+	p, stdout, stderr := utils.FetchSeperatedCommandOutput(cmd)
+	if p.Err() != nil || (p.Exited() && p.ExitCode() != 0) {
+		log.ErrorS(p.Err(), "failed to stop node in k3d cluster", "stdout", stdout.String(), "stderr", stderr.String())
+		return fmt.Errorf("k3d: failed to stop node %q in cluster %q: %s: %s", node.Name, c.name, p.Err(), p.Result())
+	}
+	return nil
+}
+
+func (c *Cluster) ListNode(ctx context.Context, args ...string) ([]support.Node, error) {
+	cmd := fmt.Sprintf("%s node list -o json", c.path)
+	p := utils.RunCommand(cmd)
+	if p.Err() != nil || (p.Exited() && p.ExitCode() != 0) {
+		return nil, fmt.Errorf("k3d: failed to list nodes: %s: %s", p.Err(), p.Result())
+	}
+	var nodeInfo []k3dNode
+	if err := json.Unmarshal([]byte(p.Result()), &nodeInfo); err != nil {
+		return nil, fmt.Errorf("k3d: failed to unmarshal node list: %s", err)
+	}
+	nodes := make([]support.Node, len(nodeInfo))
+	for _, n := range nodeInfo {
+		nodes = append(nodes, support.Node{
+			Name:    n.Name,
+			Role:    n.Role,
+			IP:      net.ParseIP(n.IP.IP),
+			State:   n.State.Status,
+			Cluster: c.name,
+		})
+	}
+	return nodes, nil
+}
diff --git a/third_party/kind/kind.go b/third_party/kind/kind.go
new file mode 100644
index 00000000..f79c4299
--- /dev/null
+++ b/third_party/kind/kind.go
@@ -0,0 +1,309 @@
+/*
+Copyright 2021 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 kind
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/rest"
+	log "k8s.io/klog/v2"
+	"sigs.k8s.io/e2e-framework/klient"
+	"sigs.k8s.io/e2e-framework/klient/conf"
+	"sigs.k8s.io/e2e-framework/klient/k8s/resources"
+	"sigs.k8s.io/e2e-framework/klient/wait"
+	"sigs.k8s.io/e2e-framework/klient/wait/conditions"
+	"sigs.k8s.io/e2e-framework/pkg/utils"
+	"sigs.k8s.io/e2e-framework/support"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+var kindVersion = "v0.26.0"
+
+type Cluster struct {
+	path        string
+	name        string
+	kubecfgFile string
+	version     string
+	image       string
+	rc          *rest.Config
+}
+
+// Enforce Type check always to avoid future breaks
+var _ support.E2EClusterProvider = &Cluster{}
+
+func NewCluster(name string) *Cluster {
+	return &Cluster{name: name}
+}
+
+func NewProvider() support.E2EClusterProvider {
+	return &Cluster{}
+}
+
+func WithImage(image string) support.ClusterOpts {
+	return func(c support.E2EClusterProvider) {
+		k, ok := c.(*Cluster)
+		if ok {
+			k.image = image
+		}
+	}
+}
+
+func WithPath(path string) support.ClusterOpts {
+	return func(c support.E2EClusterProvider) {
+		k, ok := c.(*Cluster)
+		if ok {
+			k.path = path
+		}
+	}
+}
+
+func (k *Cluster) SetDefaults() support.E2EClusterProvider {
+	if k.path == "" {
+		k.path = "kind"
+	}
+	return k
+}
+
+func (k *Cluster) WithName(name string) support.E2EClusterProvider {
+	k.name = name
+	return k
+}
+
+func (k *Cluster) WithPath(path string) support.E2EClusterProvider {
+	k.path = path
+	return k
+}
+
+func (k *Cluster) WithVersion(ver string) support.E2EClusterProvider {
+	k.version = ver
+	return k
+}
+
+func (k *Cluster) WithOpts(opts ...support.ClusterOpts) support.E2EClusterProvider {
+	for _, o := range opts {
+		o(k)
+	}
+	return k
+}
+
+func (k *Cluster) getKubeconfig() (string, error) {
+	kubecfg := fmt.Sprintf("%s-kubecfg", k.name)
+
+	var stdout, stderr bytes.Buffer
+	err := utils.RunCommandWithSeperatedOutput(fmt.Sprintf(`%s get kubeconfig --name %s`, k.path, k.name), &stdout, &stderr)
+	if err != nil {
+		return "", fmt.Errorf("kind get kubeconfig: stderr: %s: %w", stderr.String(), err)
+	}
+	log.V(4).Info("kind get kubeconfig stderr \n", stderr.String())
+
+	file, err := os.CreateTemp("", fmt.Sprintf("kind-cluster-%s", kubecfg))
+	if err != nil {
+		return "", fmt.Errorf("kind kubeconfig file: %w", err)
+	}
+	defer file.Close()
+
+	k.kubecfgFile = file.Name()
+
+	if n, err := io.WriteString(file, stdout.String()); n == 0 || err != nil {
+		return "", fmt.Errorf("kind kubecfg file: bytes copied: %d: %w]", n, err)
+	}
+
+	return file.Name(), nil
+}
+
+func (k *Cluster) clusterExists(name string) (string, bool) {
+	clusters := utils.FetchCommandOutput(fmt.Sprintf("%s get clusters", k.path))
+	for _, c := range strings.Split(clusters, "\n") {
+		if c == name {
+			return clusters, true
+		}
+	}
+	return clusters, false
+}
+
+func (k *Cluster) CreateWithConfig(ctx context.Context, kindConfigFile string) (string, error) {
+	var args []string
+	if kindConfigFile != "" {
+		args = append(args, "--config", kindConfigFile)
+	}
+	return k.Create(ctx, args...)
+}
+
+func (k *Cluster) Create(ctx context.Context, args ...string) (string, error) {
+	log.V(4).Info("Creating kind cluster ", k.name)
+	if err := k.findOrInstallKind(); err != nil {
+		return "", err
+	}
+
+	if _, ok := k.clusterExists(k.name); ok {
+		log.V(4).Info("Skipping Kind Cluster.Create: cluster already created: ", k.name)
+		kConfig, err := k.getKubeconfig()
+		if err != nil {
+			return "", err
+		}
+		return kConfig, k.initKubernetesAccessClients()
+	}
+
+	if k.image != "" {
+		args = append(args, "--image", k.image)
+	}
+
+	command := fmt.Sprintf(`%s create cluster --name %s`, k.path, k.name)
+	if len(args) > 0 {
+		command = fmt.Sprintf("%s %s", command, strings.Join(args, " "))
+	}
+	log.V(4).Info("Launching:", command)
+	p := utils.RunCommand(command)
+	if p.Err() != nil {
+		outBytes, err := io.ReadAll(p.Out())
+		if err != nil {
+			log.ErrorS(err, "failed to read data from the kind create process output due to an error")
+		}
+		return "", fmt.Errorf("kind: failed to create cluster %q: %s: %s: %s", k.name, p.Err(), p.Result(), string(outBytes))
+	}
+	clusters, ok := k.clusterExists(k.name)
+	if !ok {
+		return "", fmt.Errorf("kind Cluster.Create: cluster %v still not in 'cluster list' after creation: %v", k.name, clusters)
+	}
+	log.V(4).Info("kind clusters available: ", clusters)
+
+	kConfig, err := k.getKubeconfig()
+	if err != nil {
+		return "", err
+	}
+	return kConfig, k.initKubernetesAccessClients()
+}
+
+func (k *Cluster) initKubernetesAccessClients() error {
+	cfg, err := conf.New(k.kubecfgFile)
+	if err != nil {
+		return err
+	}
+	k.rc = cfg
+	return nil
+}
+
+func (k *Cluster) GetKubeconfig() string {
+	return k.kubecfgFile
+}
+
+func (k *Cluster) GetKubectlContext() string {
+	return fmt.Sprintf("kind-%s", k.name)
+}
+
+// ExportLogs export all cluster logs to the provided path.
+func (k *Cluster) ExportLogs(ctx context.Context, dest string) error {
+	log.V(4).Info("Exporting kind cluster logs to ", dest)
+	if err := k.findOrInstallKind(); err != nil {
+		return err
+	}
+
+	p := utils.RunCommand(fmt.Sprintf(`%s export logs %s --name %s`, k.path, dest, k.name))
+	if p.Err() != nil {
+		return fmt.Errorf("kind: export cluster %v logs failed: %s: %s", k.name, p.Err(), p.Result())
+	}
+
+	return nil
+}
+
+func (k *Cluster) Destroy(ctx context.Context) error {
+	log.V(4).Info("Destroying kind cluster ", k.name)
+	if err := k.findOrInstallKind(); err != nil {
+		return err
+	}
+
+	p := utils.RunCommand(fmt.Sprintf(`%s delete cluster --name %s`, k.path, k.name))
+	if p.Err() != nil {
+		outBytes, err := io.ReadAll(p.Out())
+		if err != nil {
+			log.ErrorS(err, "failed to read data from the kind delete process output due to an error")
+		}
+		return fmt.Errorf("kind: failed to delete cluster %q: %s: %s: %s", k.name, p.Err(), p.Result(), string(outBytes))
+	}
+
+	log.V(4).Info("Removing kubeconfig file ", k.kubecfgFile)
+	if err := os.RemoveAll(k.kubecfgFile); err != nil {
+		return fmt.Errorf("kind: remove kubefconfig %v failed: %w", k.kubecfgFile, err)
+	}
+
+	return nil
+}
+
+func (k *Cluster) findOrInstallKind() error {
+	if k.version != "" {
+		kindVersion = k.version
+	}
+	path, err := utils.FindOrInstallGoBasedProvider(k.path, "kind", "sigs.k8s.io/kind", kindVersion)
+	if path != "" {
+		k.path = path
+	}
+	return err
+}
+
+func (k *Cluster) LoadImage(ctx context.Context, image string, args ...string) error {
+	p := utils.RunCommand(fmt.Sprintf(`%s load docker-image --name %s %s`, k.path, k.name, image))
+	if p.Err() != nil {
+		return fmt.Errorf("kind: load docker-image %v failed: %s: %s", image, p.Err(), p.Result())
+	}
+	return nil
+}
+
+func (k *Cluster) LoadImageArchive(ctx context.Context, imageArchive string, args ...string) error {
+	p := utils.RunCommand(fmt.Sprintf(`%s load image-archive --name %s %s`, k.path, k.name, imageArchive))
+	if p.Err() != nil {
+		return fmt.Errorf("kind: load image-archive %v failed: %s: %s", imageArchive, p.Err(), p.Result())
+	}
+	return nil
+}
+
+func (k *Cluster) WaitForControlPlane(ctx context.Context, client klient.Client) error {
+	r, err := resources.New(client.RESTConfig())
+	if err != nil {
+		return err
+	}
+	for _, sl := range []metav1.LabelSelectorRequirement{
+		{Key: "component", Operator: metav1.LabelSelectorOpIn, Values: []string{"etcd", "kube-apiserver", "kube-controller-manager", "kube-scheduler"}},
+		{Key: "k8s-app", Operator: metav1.LabelSelectorOpIn, Values: []string{"kindnet", "kube-dns", "kube-proxy"}},
+	} {
+		selector, err := metav1.LabelSelectorAsSelector(
+			&metav1.LabelSelector{
+				MatchExpressions: []metav1.LabelSelectorRequirement{
+					sl,
+				},
+			},
+		)
+		if err != nil {
+			return err
+		}
+		err = wait.For(conditions.New(r).ResourceListN(&v1.PodList{}, len(sl.Values), resources.WithLabelSelector(selector.String())))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (k *Cluster) KubernetesRestConfig() *rest.Config {
+	return k.rc
+}
diff --git a/third_party/ko/ko.go b/third_party/ko/ko.go
index b6841186..345ff899 100644
--- a/third_party/ko/ko.go
+++ b/third_party/ko/ko.go
@@ -25,7 +25,7 @@ import (
 
 	"github.com/vladimirvivien/gexe"
 	log "k8s.io/klog/v2"
-	"sigs.k8s.io/e2e-framework/support/utils"
+	"sigs.k8s.io/e2e-framework/pkg/utils"
 )
 
 type localImageContextKey string
diff --git a/third_party/kwok/kwok.go b/third_party/kwok/kwok.go
new file mode 100644
index 00000000..fc08e394
--- /dev/null
+++ b/third_party/kwok/kwok.go
@@ -0,0 +1,273 @@
+/*
+Copyright 2023 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 kwok
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"k8s.io/client-go/rest"
+	klog "k8s.io/klog/v2"
+	"sigs.k8s.io/e2e-framework/klient"
+	"sigs.k8s.io/e2e-framework/klient/conf"
+	"sigs.k8s.io/e2e-framework/pkg/utils"
+	"sigs.k8s.io/e2e-framework/support"
+)
+
+var kwokVersion = "v0.5.0"
+
+type Cluster struct {
+	name         string
+	path         string
+	kubecfgFile  string
+	version      string
+	waitDuration time.Duration
+	rc           *rest.Config
+}
+
+var _ support.E2EClusterProvider = &Cluster{}
+
+func NewCluster(name string) *Cluster {
+	return &Cluster{name: name, waitDuration: 1 * time.Minute}
+}
+
+func NewProvider() support.E2EClusterProvider {
+	return &Cluster{}
+}
+
+func WithPath(path string) support.ClusterOpts {
+	return func(c support.E2EClusterProvider) {
+		k, ok := c.(*Cluster)
+		if ok {
+			k.path = path
+		}
+	}
+}
+
+func (k *Cluster) findOrInstallKwokCtl() error {
+	if k.version != "" {
+		kwokVersion = k.version
+	}
+	path, err := utils.FindOrInstallGoBasedProvider(k.path, "kwokctl", "sigs.k8s.io/kwok/cmd/kwokctl", kwokVersion)
+	if path != "" {
+		k.path = path
+	}
+	return err
+}
+
+func (k *Cluster) clusterExists(name string) (string, bool) {
+	clusters := utils.FetchCommandOutput(fmt.Sprintf("%s get clusters", k.path))
+	for _, c := range strings.Split(clusters, "\n") {
+		if c == name {
+			return clusters, true
+		}
+	}
+	return clusters, false
+}
+
+func (k *Cluster) getKubeconfig() (string, error) {
+	kubecfg := fmt.Sprintf("%s-kubecfg", k.name)
+
+	var stdout, stderr bytes.Buffer
+	err := utils.RunCommandWithSeperatedOutput(fmt.Sprintf(`%s get kubeconfig --name %s`,
+		k.path, k.name), &stdout, &stderr)
+	if err != nil {
+		return "", fmt.Errorf("kwokctl get kubeconfig: stderr: %s: %w", stderr.String(), err)
+	}
+
+	file, err := os.CreateTemp("", fmt.Sprintf("kwok-cluster-%s", kubecfg))
+	if err != nil {
+		return "", fmt.Errorf("kwok kubeconfig file: %w", err)
+	}
+	defer file.Close()
+
+	k.kubecfgFile = file.Name()
+
+	if n, err := io.WriteString(file, stdout.String()); n == 0 || err != nil {
+		return "", fmt.Errorf("kwok kubecfg file: bytes copied: %d: %w]", n, err)
+	}
+
+	return file.Name(), nil
+}
+
+func (k *Cluster) initKubernetesAccessClients() error {
+	cfg, err := conf.New(k.kubecfgFile)
+	if err != nil {
+		return err
+	}
+	k.rc = cfg
+	return nil
+}
+
+func (k *Cluster) Create(ctx context.Context, args ...string) (string, error) {
+	klog.V(4).Info("Creating a kwok cluster ", k.name)
+	if err := k.findOrInstallKwokCtl(); err != nil {
+		return "", err
+	}
+	if _, ok := k.clusterExists(k.name); ok {
+		klog.V(4).Info("Skipping Kwok Cluster creation. Cluster already created ", k.name)
+		kConfig, err := k.getKubeconfig()
+		if err != nil {
+			return "", err
+		}
+		return kConfig, k.initKubernetesAccessClients()
+	}
+
+	command := fmt.Sprintf(`%s create cluster --name %s --wait %s`, k.path, k.name, k.waitDuration.String())
+	if len(args) > 0 {
+		command = fmt.Sprintf("%s %s", command, strings.Join(args, " "))
+	}
+	klog.V(4).Info("Launching:", command)
+	p := utils.RunCommand(command)
+	if p.Err() != nil {
+		outBytes, err := io.ReadAll(p.Out())
+		if err != nil {
+			klog.ErrorS(err, "failed to read data from the kwok create process output due to an error")
+		}
+		return "", fmt.Errorf("kwok: failed to create cluster %q: %s: %s: %s", k.name, p.Err(), p.Result(), string(outBytes))
+	}
+
+	clusters, ok := k.clusterExists(k.name)
+	if !ok {
+		return "", fmt.Errorf("kwok Cluster.Create: cluster %v still not in 'cluster list' after creation: %v", k.name, clusters)
+	}
+	klog.V(4).Info("kwok cluster available: ", clusters)
+
+	kConfig, err := k.getKubeconfig()
+	if err != nil {
+		return "", err
+	}
+	return kConfig, k.initKubernetesAccessClients()
+}
+
+func (k *Cluster) CreateWithConfig(ctx context.Context, configFile string) (string, error) {
+	if configFile == "" {
+		return k.Create(ctx)
+	}
+
+	return k.Create(ctx, "--config", configFile)
+}
+
+func (k *Cluster) Destroy(ctx context.Context) error {
+	klog.V(4).Info("Destroying kwok cluster ", k.name)
+	if err := k.findOrInstallKwokCtl(); err != nil {
+		return err
+	}
+
+	p := utils.RunCommand(fmt.Sprintf(`%s delete cluster --name %s`, k.path, k.name))
+	if p.Err() != nil {
+		outBytes, err := io.ReadAll(p.Out())
+		if err != nil {
+			klog.ErrorS(err, "failed to read data from the kwok delete process output due to an error")
+		}
+		return fmt.Errorf("kwok: failed to delete cluster %q: %s: %s: %s", k.name, p.Err(), p.Result(), string(outBytes))
+	}
+
+	klog.V(4).Info("Removing kubeconfig file ", k.kubecfgFile)
+	if err := os.RemoveAll(k.kubecfgFile); err != nil {
+		return fmt.Errorf("kwok: remove kubefconfig failed: %w", err)
+	}
+	return nil
+}
+
+func (k *Cluster) ExportLogs(ctx context.Context, dest string) error {
+	if err := k.findOrInstallKwokCtl(); err != nil {
+		return err
+	}
+	// In kwokctl 0.3.0 and above, there is a new kwokctl export logs feature that has been added which can
+	// simplify the workf of exporting the logs for us. Let us check if the CLI has that command and if so
+	// let us use that to export logs. Otherwise, we can fallback to exporting individual items.
+	p := utils.RunCommand(fmt.Sprintf("%s export logs --help", k.path))
+	if p.ExitCode() == 0 {
+		return utils.RunCommand(fmt.Sprintf("%s --name %s export logs %s", k.path, k.name, dest)).Err()
+	}
+
+	// TODO: Get Rid of this if we decide to enforce a min version of the kwokctl at some point
+	for _, component := range []string{"audit", "etcd", "kube-apiserver", "kube-controller-manager", "kube-scheduler", "kwok-controller", "prometheus"} {
+		command := fmt.Sprintf("%s logs %s", k.path, component)
+		p := utils.RunCommand(command)
+		if p.Err() != nil {
+			klog.ErrorS(p.Err(), "ran into an error trying to export the log", "component", component)
+			continue
+		}
+		var stdout bytes.Buffer
+		if _, err := stdout.ReadFrom(p.Out()); err != nil {
+			return fmt.Errorf("kwokctl logs %s stdout bytes: %w", component, err)
+		}
+		file, err := os.Create(filepath.Join(dest, fmt.Sprintf("%s.log", component)))
+		if err != nil {
+			klog.ErrorS(err, "ran into an error trying to create file to export logs", "component", component)
+			continue
+		}
+		if n, err := io.Copy(file, &stdout); n == 0 || err != nil {
+			klog.ErrorS(err, "kwokctl logs %s file: bytes copied: %d: %w]", component, n, err)
+		}
+	}
+	return nil
+}
+
+func (k *Cluster) GetKubectlContext() string {
+	return fmt.Sprintf("kwok-%s", k.name)
+}
+
+func (k *Cluster) GetKubeconfig() string {
+	return k.kubecfgFile
+}
+
+func (k *Cluster) SetDefaults() support.E2EClusterProvider {
+	if k.path == "" {
+		k.path = "kwokctl"
+	}
+	return k
+}
+
+func (k *Cluster) WaitForControlPlane(ctx context.Context, client klient.Client) error {
+	klog.V(4).Info("kwokctl doesn't implement a WaitForControlPlane handler. The --wait argument passed to the `kwokctl` should take care of this already")
+	return nil
+}
+
+func (k *Cluster) WithName(name string) support.E2EClusterProvider {
+	k.name = name
+	return k
+}
+
+func (k *Cluster) WithOpts(opts ...support.ClusterOpts) support.E2EClusterProvider {
+	for _, o := range opts {
+		o(k)
+	}
+	return k
+}
+
+func (k *Cluster) WithPath(path string) support.E2EClusterProvider {
+	k.path = path
+	return k
+}
+
+func (k *Cluster) WithVersion(version string) support.E2EClusterProvider {
+	k.version = version
+	return k
+}
+
+func (k *Cluster) KubernetesRestConfig() *rest.Config {
+	return k.rc
+}
diff --git a/third_party/vcluster/vcluster.go b/third_party/vcluster/vcluster.go
index 993021e8..56b170dd 100644
--- a/third_party/vcluster/vcluster.go
+++ b/third_party/vcluster/vcluster.go
@@ -35,8 +35,8 @@ import (
 	"sigs.k8s.io/e2e-framework/klient/k8s/resources"
 	"sigs.k8s.io/e2e-framework/klient/wait"
 	"sigs.k8s.io/e2e-framework/klient/wait/conditions"
+	"sigs.k8s.io/e2e-framework/pkg/utils"
 	"sigs.k8s.io/e2e-framework/support"
-	"sigs.k8s.io/e2e-framework/support/utils"
 	"sigs.k8s.io/yaml"
 )
 
@@ -248,8 +248,11 @@ func (c *Cluster) WaitForControlPlane(ctx context.Context, client klient.Client)
 	if err != nil {
 		return err
 	}
+
+	// vcluster seem to name the dns pod with label vcluster-kube-dns.
+	// Wish there was a way to avoid hard coding this slightly better.
 	for _, sl := range []metav1.LabelSelectorRequirement{
-		{Key: "k8s-app", Operator: metav1.LabelSelectorOpIn, Values: []string{"kube-dns"}},
+		{Key: "k8s-app", Operator: metav1.LabelSelectorOpIn, Values: []string{"vcluster-kube-dns"}},
 	} {
 		selector, err := metav1.LabelSelectorAsSelector(
 			&metav1.LabelSelector{