diff --git a/examples/k3d/k3d_test.go b/examples/k3d/k3d_test.go
new file mode 100644
index 00000000..0a8df20c
--- /dev/null
+++ b/examples/k3d/k3d_test.go
@@ -0,0 +1,76 @@
+/*
+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"
+	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"
+	"testing"
+	"time"
+)
+
+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*1))
+			if err != nil {
+				t.Fatal(err)
+			}
+			return context.WithValue(ctx, "test-deployment", &dep)
+		}).
+		Feature()
+
+	testEnv.Test(t, deploymentFeature)
+}
diff --git a/examples/k3d/main_test.go b/examples/k3d/main_test.go
new file mode 100644
index 00000000..b00b26b2
--- /dev/null
+++ b/examples/k3d/main_test.go
@@ -0,0 +1,48 @@
+/*
+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
+
+func TestMain(m *testing.M) {
+	testEnv = env.New()
+	clusterName := envconf.RandomName("k3d-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..46a60c24 100644
--- a/pkg/envfuncs/provider_funcs.go
+++ b/pkg/envfuncs/provider_funcs.go
@@ -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
@@ -121,7 +132,7 @@ 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))
 		if clusterVal == 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,7 +155,7 @@ 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))
 		if clusterVal == 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)
 		}
 
diff --git a/support/k3d/k3d.go b/support/k3d/k3d.go
new file mode 100644
index 00000000..75ba78b4
--- /dev/null
+++ b/support/k3d/k3d.go
@@ -0,0 +1,294 @@
+/*
+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"
+	"os"
+	"strings"
+
+	"k8s.io/client-go/rest"
+
+	"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"
+
+	log "k8s.io/klog/v2"
+)
+
+var k3dVersion = "v5.7.1"
+
+type Cluster struct {
+	path           string
+	name           string
+	kubeConfigFile string
+	version        string
+	image          string
+	rc             *rest.Config
+	args           []string
+}
+
+var _ support.E2EClusterProviderWithImageLoader = &Cluster{}
+
+func WithArgs(args ...string) support.ClusterOpts {
+	return func(c support.E2EClusterProvider) {
+		k, ok := c.(*Cluster)
+		if ok {
+			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)
+	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 create process output due to an error")
+		}
+		return "", fmt.Errorf("k3d: failed to create cluster %q: %s: %s: %s", c.name, p.Err(), p.Result(), string(outBytes))
+	}
+	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 doesn't implement a WaitForControlPlane handler. The --wait argument passed to the `kwokctl` should take care of this already")
+	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...)
+}
diff --git a/support/kind/kind.go b/support/kind/kind.go
index 4ca3d2a1..a691fdb4 100644
--- a/support/kind/kind.go
+++ b/support/kind/kind.go
@@ -261,7 +261,7 @@ func (k *Cluster) findOrInstallKind() error {
 	return err
 }
 
-func (k *Cluster) LoadImage(ctx context.Context, image string) error {
+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())
@@ -269,7 +269,7 @@ func (k *Cluster) LoadImage(ctx context.Context, image string) error {
 	return nil
 }
 
-func (k *Cluster) LoadImageArchive(ctx context.Context, imageArchive string) error {
+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())
diff --git a/support/types.go b/support/types.go
index 25fcae26..2c0d1376 100644
--- a/support/types.go
+++ b/support/types.go
@@ -94,12 +94,12 @@ type E2EClusterProviderWithImageLoader interface {
 	// 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
+	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) error
+	LoadImageArchive(ctx context.Context, archivePath string, args ...string) error
 }