diff --git a/Makefile b/Makefile
index 5bd1b562..18a36463 100644
--- a/Makefile
+++ b/Makefile
@@ -91,7 +91,7 @@ build: generate build_push_docker ## Builds and pushes the docker image to tag d
 	@echo "TIME: $(shell date) end make build"
 
 .PHONY: test
-test: generate go_test ## Run tests (but not internal/teste2e)
+test: generate go_test go_test_k8s_1_28 ## Run tests (but not internal/teste2e)
 	@echo "TIME: $(shell date) end make test"
 
 .PHONY: deploy
@@ -167,6 +167,11 @@ go_test: ctrl_manifests envtest # Run tests (but not internal/teste2e)
 	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) -p path)" \
 		go test ./internal/.../. -coverprofile cover.out -race | tee test_results.txt
 
+.PHONY: go_test_k8s_1_28
+go_test_k8s_1_28: ctrl_manifests envtest # Run tests using older kubernetes version 1.28 (but not internal/teste2e)
+	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use 1.28.x -p path)" \
+		go test ./internal/.../. -coverprofile cover.out -race | tee test_results.txt
+
 ##
 # 3rd Party License Checks
 .PHONY: license_check
diff --git a/internal/controller/authproxyworkload_controller_test.go b/internal/controller/authproxyworkload_controller_test.go
index fdd3c042..9418cfd8 100644
--- a/internal/controller/authproxyworkload_controller_test.go
+++ b/internal/controller/authproxyworkload_controller_test.go
@@ -490,7 +490,7 @@ func reconciler(p *cloudsqlapi.AuthProxyWorkload, cb client.Client, defaultProxy
 	r := &AuthProxyWorkloadReconciler{
 		Client:          cb,
 		recentlyDeleted: &recentlyDeletedCache{},
-		updater:         workload.NewUpdater("cloud-sql-proxy-operator/dev", defaultProxyImage),
+		updater:         workload.NewUpdater("cloud-sql-proxy-operator/dev", defaultProxyImage, false),
 	}
 	req := ctrl.Request{
 		NamespacedName: types.NamespacedName{
diff --git a/internal/controller/pod_controller_test.go b/internal/controller/pod_controller_test.go
index e0122e13..ae31d7d3 100644
--- a/internal/controller/pod_controller_test.go
+++ b/internal/controller/pod_controller_test.go
@@ -151,7 +151,7 @@ func podWebhookController(cb client.Client) (*PodAdmissionWebhook, context.Conte
 	r := &PodAdmissionWebhook{
 		Client:  cb,
 		decoder: d,
-		updater: workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage),
+		updater: workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false),
 	}
 
 	return r, ctx, nil
@@ -293,7 +293,7 @@ func podDeleteControllerForTest(c client.Client) (*podDeleteController, context.
 	r := &podDeleteController{
 		Client:  c,
 		Scheme:  c.Scheme(),
-		updater: workload.NewUpdater("cloud-sql-proxy-operator/dev", "proxy:1.0"),
+		updater: workload.NewUpdater("cloud-sql-proxy-operator/dev", "proxy:1.0", false),
 	}
 	return r, ctx
 }
diff --git a/internal/controller/setup.go b/internal/controller/setup.go
index 65d46a6a..ca184ed3 100644
--- a/internal/controller/setup.go
+++ b/internal/controller/setup.go
@@ -19,6 +19,9 @@ import (
 	"github.com/GoogleCloudPlatform/cloud-sql-proxy-operator/internal/workload"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	utilversion "k8s.io/apimachinery/pkg/util/version"
+	"k8s.io/apimachinery/pkg/version"
+	"k8s.io/client-go/kubernetes"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -37,13 +40,33 @@ func InitScheme(scheme *runtime.Scheme) {
 	//+kubebuilder:scaffold:scheme
 }
 
+func getKubernetesVersion(mgr manager.Manager) (*version.Info, error) {
+	kubeClient, err := kubernetes.NewForConfig(mgr.GetConfig())
+	if err != nil {
+		return nil, err
+	}
+
+	return kubeClient.Discovery().ServerVersion()
+}
+
+func UseSidecar(v *version.Info) bool {
+	return utilversion.MustParseSemantic(v.GitVersion).AtLeast(utilversion.MustParseSemantic("v1.29.0"))
+}
+
 // SetupManagers was moved out of ../main.go here so that it can be invoked
 // from the testintegration tests AND from the actual operator.
 func SetupManagers(mgr manager.Manager, userAgent, defaultProxyImage string) error {
-	u := workload.NewUpdater(userAgent, defaultProxyImage)
+	var err error
+	kubeVersion, err := getKubernetesVersion(mgr)
+	if err != nil {
+		setupLog.Error(err, "unable to get kubernetes version", "controller", "AuthProxyWorkload")
+		return err
+	}
+	setupLog.Info("Kubernetes", "version", kubeVersion.String())
+
+	u := workload.NewUpdater(userAgent, defaultProxyImage, UseSidecar(kubeVersion))
 
 	setupLog.Info("Configuring reconcilers...")
-	var err error
 
 	_, err = NewAuthProxyWorkloadReconciler(mgr, u)
 	if err != nil {
diff --git a/internal/testhelpers/resources.go b/internal/testhelpers/resources.go
index 3ff18b43..bbcc1d27 100644
--- a/internal/testhelpers/resources.go
+++ b/internal/testhelpers/resources.go
@@ -440,7 +440,11 @@ func (cc *TestCaseClient) ExpectPodContainerCount(ctx context.Context, podSelect
 			return fmt.Errorf("got 0 pods, want at least 1 pod")
 		}
 		for _, pod := range pods.Items {
-			got := len(pod.Spec.Containers)
+			// The use len(Containers) + len(InitContainers) as the total number
+			// of containers on the pod. This way the assertion works the same
+			// when it runs with k8s <= 1.28 using regular containers for the proxy
+			// as well as with k8s >= 1.29 using init containers for the proxy.
+			got := len(pod.Spec.Containers) + len(pod.Spec.InitContainers)
 			if got != count {
 				countBadPods++
 			}
@@ -679,7 +683,7 @@ func NewAuthProxyWorkload(key types.NamespacedName) *cloudsqlapi.AuthProxyWorklo
 }
 
 // CreateAuthProxyWorkload creates an AuthProxyWorkload in the kubernetes cluster.
-func (cc *TestCaseClient) CreateAuthProxyWorkload(ctx context.Context, key types.NamespacedName, appLabel string, connectionString string, kind string) (*cloudsqlapi.AuthProxyWorkload, error) {
+func (cc *TestCaseClient) CreateAuthProxyWorkload(ctx context.Context, key types.NamespacedName, appLabel, connectionString, kind string) (*cloudsqlapi.AuthProxyWorkload, error) {
 	p := NewAuthProxyWorkload(key)
 	AddTCPInstance(p, connectionString)
 	cc.ConfigureSelector(p, appLabel, kind)
diff --git a/internal/testintegration/integration_test.go b/internal/testintegration/integration_test.go
index 9dcac68d..f02877a1 100644
--- a/internal/testintegration/integration_test.go
+++ b/internal/testintegration/integration_test.go
@@ -265,7 +265,14 @@ func TestUpdateWorkloadContainerWhenDefaultProxyImageChanges(t *testing.T) {
 	// Check that the pods have the expected default proxy image
 	pods, err := testhelpers.ListPods(ctx, tcc.Client, tcc.Namespace, d.Spec.Selector)
 	for _, p := range pods.Items {
-		if got, want := p.Spec.Containers[1].Image, workload.DefaultProxyImage; got != want {
+		want := workload.DefaultProxyImage
+		var got string
+		if len(p.Spec.Containers) > 1 {
+			got = p.Spec.Containers[1].Image
+		} else {
+			got = p.Spec.InitContainers[0].Image
+		}
+		if got != want {
 			t.Errorf("got %v, want %v image before operator upgrade", got, want)
 		}
 	}
@@ -305,7 +312,15 @@ func TestUpdateWorkloadContainerWhenDefaultProxyImageChanges(t *testing.T) {
 	// Check that the new pods have the new default proxy image
 	pods, err = testhelpers.ListPods(ctx, tcc.Client, tcc.Namespace, d.Spec.Selector)
 	for _, p := range pods.Items {
-		if got, want := p.Spec.Containers[1].Image, newDefault; got != want {
+		want := newDefault
+		var got string
+		if len(p.Spec.Containers) > 1 {
+			got = p.Spec.Containers[1].Image
+		} else {
+			got = p.Spec.InitContainers[0].Image
+		}
+
+		if got != want {
 			t.Errorf("got %v, want %v image before operator upgrade", got, want)
 		}
 	}
diff --git a/internal/workload/podspec_updates.go b/internal/workload/podspec_updates.go
index 5e3e5936..5696e982 100644
--- a/internal/workload/podspec_updates.go
+++ b/internal/workload/podspec_updates.go
@@ -89,12 +89,19 @@ type Updater struct {
 
 	// defaultProxyImage is the current default proxy image for the operator
 	defaultProxyImage string
+
+	// useSidecar specifies whether to use the Kubernetes SidecarContainers feature
+	useSidecar bool
 }
 
 // NewUpdater creates a new instance of Updater with a supplier
-// that loads the default proxy impage from the public docker registry
-func NewUpdater(userAgent string, defaultProxyImage string) *Updater {
-	return &Updater{userAgent: userAgent, defaultProxyImage: defaultProxyImage}
+// that loads the default proxy image from the public docker registry
+func NewUpdater(userAgent, defaultProxyImage string, useSidecar bool) *Updater {
+	return &Updater{
+		userAgent:         userAgent,
+		defaultProxyImage: defaultProxyImage,
+		useSidecar:        useSidecar,
+	}
 }
 
 // ConfigError is an error with extra details about why an AuthProxyWorkload
@@ -513,19 +520,13 @@ func (s *updateState) update(wl *PodWorkload, matches []*cloudsqlapi.AuthProxyWo
 
 	s.initState(matches)
 	podSpec := wl.PodSpec()
-	containers := podSpec.Containers
-
-	var nonAuthProxyContainers []corev1.Container
-	for i := 0; i < len(containers); i++ {
-		if !strings.HasPrefix(containers[i].Name, ContainerPrefix) {
-			nonAuthProxyContainers = append(nonAuthProxyContainers, containers[i])
-		}
-	}
+	allContainers := append(podSpec.Containers, podSpec.InitContainers...)
 
-	for i := 0; i < len(nonAuthProxyContainers); i++ {
-		c := nonAuthProxyContainers[i]
-		for j := 0; j < len(c.Ports); j++ {
-			s.addWorkloadPort(c.Ports[j].ContainerPort)
+	for _, container := range allContainers {
+		if !strings.HasPrefix(container.Name, ContainerPrefix) {
+			for _, port := range container.Ports {
+				s.addWorkloadPort(port.ContainerPort)
+			}
 		}
 	}
 
@@ -541,7 +542,12 @@ func (s *updateState) update(wl *PodWorkload, matches []*cloudsqlapi.AuthProxyWo
 
 		newContainer := corev1.Container{}
 		s.updateContainer(inst, &newContainer)
-		containers = append(containers, newContainer)
+
+		if s.updater.useSidecar {
+			podSpec.InitContainers = append([]corev1.Container{newContainer}, podSpec.InitContainers...)
+		} else {
+			podSpec.Containers = append(podSpec.Containers, newContainer)
+		}
 
 		// Add pod annotation for each instance
 		k, v := s.updater.PodAnnotation(inst)
@@ -550,17 +556,13 @@ func (s *updateState) update(wl *PodWorkload, matches []*cloudsqlapi.AuthProxyWo
 	// Add the envvar containing the proxy quit urls to the workloads
 	s.addQuitEnvVar()
 
-	podSpec.Containers = containers
-
 	if len(ann) != 0 {
 		wl.SetPodTemplateAnnotations(ann)
 	}
 
-	for i := range podSpec.Containers {
-		c := &podSpec.Containers[i]
-		s.updateContainerEnv(c)
-		s.applyContainerVolumes(c)
-	}
+	s.processContainers(&podSpec.Containers)
+	s.processContainers(&podSpec.InitContainers)
+
 	s.applyVolumes(&podSpec)
 
 	// only return ConfigError if there were reported
@@ -574,6 +576,15 @@ func (s *updateState) update(wl *PodWorkload, matches []*cloudsqlapi.AuthProxyWo
 	return nil
 }
 
+// processContainers applies container envs and volumes to all containers
+func (s *updateState) processContainers(containers *[]corev1.Container) {
+	for i := range *containers {
+		c := &(*containers)[i]
+		s.updateContainerEnv(c)
+		s.applyContainerVolumes(c)
+	}
+}
+
 // updateContainer Creates or updates the proxy container in the workload's PodSpec
 func (s *updateState) updateContainer(p *cloudsqlapi.AuthProxyWorkload, c *corev1.Container) {
 	// if the c was fully overridden, just use that c.
@@ -610,8 +621,11 @@ func (s *updateState) updateContainer(p *cloudsqlapi.AuthProxyWorkload, c *corev
 	}
 
 	c.Name = ContainerName(p)
-	c.ImagePullPolicy = "IfNotPresent"
-
+	c.ImagePullPolicy = corev1.PullIfNotPresent
+	if s.updater.useSidecar {
+		policy := corev1.ContainerRestartPolicyAlways
+		c.RestartPolicy = &policy
+	}
 	s.applyContainerSpec(p, c)
 
 	// Build the c
diff --git a/internal/workload/podspec_updates_test.go b/internal/workload/podspec_updates_test.go
index 01bcbf08..811b2da5 100644
--- a/internal/workload/podspec_updates_test.go
+++ b/internal/workload/podspec_updates_test.go
@@ -140,7 +140,7 @@ func TestUpdatePodWorkload(t *testing.T) {
 		wantContainerName       = "csql-default-" + wantsName
 		wantsInstanceName       = "project:server:db"
 		wantsInstanceArg        = fmt.Sprintf("%s?port=%d", wantsInstanceName, wantsPort)
-		u                       = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u                       = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 	var err error
 
@@ -195,7 +195,7 @@ func TestUpdateWorkloadFixedPort(t *testing.T) {
 			"DB_HOST": "127.0.0.1",
 			"DB_PORT": strconv.Itoa(int(wantsPort)),
 		}
-		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 
 	// Create a pod
@@ -263,7 +263,7 @@ func TestWorkloadNoPortSet(t *testing.T) {
 			"DB_PORT": strconv.Itoa(int(wantsPort)),
 		}
 	)
-	u := workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+	u := workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 
 	// Create a pod
 	wl := podWorkload()
@@ -321,7 +321,7 @@ func TestContainerImageChanged(t *testing.T) {
 	var (
 		wantsInstanceName = "project:server:db"
 		wantImage         = "custom-image:latest"
-		u                 = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u                 = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 
 	// Create a pod
@@ -363,7 +363,7 @@ func TestContainerImageEmpty(t *testing.T) {
 	var (
 		wantsInstanceName = "project:server:db"
 		wantImage         = workload.DefaultProxyImage
-		u                 = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u                 = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 	// Create a AuthProxyWorkload that matches the deployment
 
@@ -422,7 +422,7 @@ func TestContainerReplaced(t *testing.T) {
 		wantContainer     = &corev1.Container{
 			Name: "sample", Image: "debian:latest", Command: []string{"/bin/bash"},
 		}
-		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 
 	// Create a pod
@@ -476,7 +476,7 @@ func TestResourcesFromSpec(t *testing.T) {
 			},
 		}
 
-		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 
 	// Create a pod
@@ -757,7 +757,7 @@ func TestProxyCLIArgs(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.desc, func(t *testing.T) {
-			u := workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+			u := workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 
 			// Create a pod
 			wl := &workload.PodWorkload{Pod: &corev1.Pod{
@@ -892,7 +892,7 @@ func TestPodTemplateAnnotations(t *testing.T) {
 			"cloudsql.cloud.google.com/instance2": "2," + workload.DefaultProxyImage,
 		}
 
-		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 
 	// Create a pod
@@ -926,7 +926,7 @@ func TestPodTemplateAnnotations(t *testing.T) {
 
 func TestTelemetryAddsTelemetryContainerPort(t *testing.T) {
 
-	var u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+	var u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 
 	// Create a pod
 	wl := podWorkload()
@@ -994,7 +994,7 @@ func TestTelemetryAddsTelemetryContainerPort(t *testing.T) {
 func TestQuitURLEnvVar(t *testing.T) {
 
 	var (
-		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 
 	// Create a pod
@@ -1035,7 +1035,7 @@ func TestQuitURLEnvVar(t *testing.T) {
 
 func TestPreStopHook(t *testing.T) {
 
-	var u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+	var u = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 
 	// Create a pod
 	wl := podWorkload()
@@ -1123,7 +1123,7 @@ func TestWorkloadUnixVolume(t *testing.T) {
 		wantWorkloadEnv = map[string]string{
 			"DB_SOCKET_PATH": wantsUnixSocketPath,
 		}
-		u = workload.NewUpdater("authproxyworkload/dev", workload.DefaultProxyImage)
+		u = workload.NewUpdater("authproxyworkload/dev", workload.DefaultProxyImage, false)
 	)
 
 	// Create a pod
@@ -1201,7 +1201,7 @@ func TestWorkloadUnixVolume(t *testing.T) {
 func TestUpdater_CheckWorkloadContainers(t *testing.T) {
 	var (
 		wantsInstanceName = "project:server:db"
-		u                 = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage)
+		u                 = workload.NewUpdater("cloud-sql-proxy-operator/dev", workload.DefaultProxyImage, false)
 	)
 
 	// Create a AuthProxyWorkloads to match the pods