feature(sync): Implement syncing across replicas in a Deployment,ReplicaSet,StatefulSet

cmd/sync.go

@@ -38,6 +38,7 @@ type SyncCmd struct {
 	ImageSelector string
 	Container     string
 	Pod           string
+	SyncReplicas  bool
 	Pick          bool
 	Wait          bool
 	Polling       bool
@@ -85,6 +86,7 @@ devspace sync --path=.:/app --pod=my-pod --container=my-container
 
 	syncCmd.Flags().StringVarP(&cmd.Container, "container", "c", "", "Container name within pod where to sync to")
 	syncCmd.Flags().StringVar(&cmd.Pod, "pod", "", "Pod to sync to")
+	syncCmd.Flags().BoolVar(&cmd.SyncReplicas, "sync-replicas", false, "Sync all replicas in the selected deployment")
 	syncCmd.Flags().StringVarP(&cmd.LabelSelector, "label-selector", "l", "", "Comma separated key=value selector list (e.g. release=test)")
 	syncCmd.Flags().StringVar(&cmd.ImageSelector, "image-selector", "", "The image to search a pod for (e.g. nginx, nginx:latest, ${runtime.images.app}, nginx:${runtime.images.app.tag})")
 	syncCmd.Flags().BoolVar(&cmd.Pick, "pick", true, "Select a pod")
@@ -270,6 +272,12 @@ func (cmd *SyncCmd) Run(f factory.Factory) error {
 	if err != nil {
 		return errors.Wrap(err, "apply flags to sync config")
 	}
+	if syncConfig.syncConfig.SyncReplicas {
+		cmd.SyncReplicas = true
+	}
+	if cmd.SyncReplicas {
+		options = options.WithPick(false)
+	}
 
 	// Start sync
 	options = options.WithSkipInitContainers(true)
@@ -313,6 +321,9 @@ func (cmd *SyncCmd) applyFlagsToSyncConfig(syncConfig *latest.SyncConfig, option
 	if cmd.DownloadOnly {
 		syncConfig.DisableUpload = cmd.DownloadOnly
 	}
+	if cmd.SyncReplicas {
+		syncConfig.SyncReplicas = true
+	}
 
 	// if selection is specified through flags, we don't want to use the loaded
 	// sync config selection from the devspace.yaml.

e2e/framework/helper.go

@@ -11,6 +11,7 @@ import (
 	"github.com/loft-sh/devspace/e2e/kube"
 	"github.com/onsi/gomega"
 	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 )
@@ -169,6 +170,52 @@ func ExpectRemoteContainerFileContents(labelSelector, container string, namespac
 	ExpectNoErrorWithOffset(1, err)
 }
 
+func ExpectRemoteFileContentsOnAllPods(labelSelector, containerName, namespace, filePath, contents string, wantCount int) {
+	kubeClient, err := kube.NewKubeHelper()
+	ExpectNoErrorWithOffset(1, err)
+
+	want := strings.TrimSpace(contents)
+	err = wait.PollUntilContextTimeout(context.TODO(), time.Second, time.Minute*3, true, func(ctx context.Context) (done bool, err error) {
+		pods, err := kubeClient.RawClient().CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: labelSelector})
+		if err != nil {
+			return false, nil
+		}
+		readyPods := make([]corev1.Pod, 0, len(pods.Items))
+		for _, p := range pods.Items {
+			if p.Status.Phase != corev1.PodRunning {
+				continue
+			}
+			if !podContainerReady(&p, containerName) {
+				continue
+			}
+			readyPods = append(readyPods, p)
+		}
+		if len(readyPods) < wantCount {
+			return false, nil
+		}
+		for i := range readyPods {
+			out, err := kubeClient.ExecInPod(ctx, &readyPods[i], containerName, []string{"cat", filePath})
+			if err != nil {
+				return false, nil
+			}
+			if strings.TrimSpace(out) != want {
+				return false, nil
+			}
+		}
+		return true, nil
+	})
+	ExpectNoErrorWithOffset(1, err)
+}
+
+func podContainerReady(pod *corev1.Pod, containerName string) bool {
+	for _, cs := range pod.Status.ContainerStatuses {
+		if cs.Name == containerName {
+			return cs.Ready
+		}
+	}
+	return false
+}
+
 func ExpectLocalFileContentsImmediately(filePath string, contents string) {
 	out, err := os.ReadFile(filePath)
 	ExpectNoError(err)

e2e/kube/kube.go

@@ -98,3 +98,12 @@ func (k *KubeHelper) DeleteNamespace(name string) error {
 	}
 	return nil
 }
+
+// ExecInPod runs a command in an existing pod container (used when multiple pods match a selector).
+func (k *KubeHelper) ExecInPod(ctx context.Context, pod *corev1.Pod, containerName string, command []string) (string, error) {
+	stdout, stderr, err := k.client.ExecBuffered(ctx, pod, containerName, command, nil)
+	if err != nil {
+		return "", fmt.Errorf("exec error: %v %s", err, string(stderr))
+	}
+	return string(stdout), nil
+}

e2e/tests/sync/sync.go

@@ -846,4 +846,55 @@ var _ = DevSpaceDescribe("sync", func() {
 		// wait for the command to finish
 		waitGroup.Wait()
 	})
+
+	ginkgo.It("should sync to all replicas when syncReplicas is enabled", func() {
+		tempDir, err := framework.CopyToTempDir("tests/sync/testdata/sync-replicas")
+		framework.ExpectNoError(err)
+		defer framework.CleanupTempDir(initialDir, tempDir)
+
+		ns, err := kubeClient.CreateNamespace("sync")
+		framework.ExpectNoError(err)
+		defer func() {
+			err := kubeClient.DeleteNamespace(ns)
+			framework.ExpectNoError(err)
+		}()
+
+		cancelCtx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		devCmd := &cmd.RunPipelineCmd{
+			GlobalFlags: &flags.GlobalFlags{
+				NoWarn:    true,
+				Namespace: ns,
+			},
+			Pipeline: "dev",
+			Ctx:      cancelCtx,
+		}
+
+		waitGroup := sync.WaitGroup{}
+		waitGroup.Add(1)
+		go func() {
+			defer ginkgo.GinkgoRecover()
+			defer waitGroup.Done()
+			err = devCmd.RunDefault(f)
+			framework.ExpectNoError(err)
+		}()
+
+		const (
+			podLabel     = "app.kubernetes.io/name=sync-replicas"
+			container    = "app"
+			replicaCount = 2
+		)
+
+		framework.ExpectRemoteFileContentsOnAllPods(podLabel, container, ns, "/app/file1.txt", "Hello World", replicaCount)
+
+		payload := randutil.GenerateRandomString(5000)
+		err = os.WriteFile(filepath.Join(tempDir, "replica-upload.txt"), []byte(payload), 0666)
+		framework.ExpectNoError(err)
+
+		framework.ExpectRemoteFileContentsOnAllPods(podLabel, container, ns, "/app/replica-upload.txt", payload, replicaCount)
+
+		cancel()
+		waitGroup.Wait()
+	})
 })

e2e/tests/sync/testdata/sync-replicas/devspace.yaml

@@ -0,0 +1,19 @@
+version: v2beta1
+vars:
+  IMAGE: alpine:3.20
+deployments:
+  test:
+    kubectl:
+      manifests:
+        - k8s/deployment.yaml
+pipelines:
+  deploy: |-
+    run_dependencies --all
+    create_deployments --all
+dev:
+  test:
+    imageSelector: ${IMAGE}
+    sync:
+      - path: ./:/app
+        waitInitialSync: true
+        syncReplicas: true

e2e/tests/sync/testdata/sync-replicas/file1.txt

@@ -0,0 +1 @@
+Hello World

e2e/tests/sync/testdata/sync-replicas/k8s/deployment.yaml

@@ -0,0 +1,20 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: sync-replicas
+  labels:
+    app.kubernetes.io/name: sync-replicas
+spec:
+  replicas: 2
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: sync-replicas
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: sync-replicas
+    spec:
+      containers:
+        - name: app
+          image: alpine:3.20
+          command: ["tail", "-f", "/dev/null"]

pkg/devspace/config/versions/latest/schema.go

@@ -1151,6 +1151,8 @@ type SyncConfig struct {
 	DisableDownload bool `yaml:"disableDownload,omitempty" json:"disableDownload,omitempty" jsonschema_extras:"group=one_direction,group_name=One-Directional Sync"`
 	// DisableUpload will disable uploading completely
 	DisableUpload bool `yaml:"disableUpload,omitempty" json:"disableUpload,omitempty" jsonschema_extras:"group=one_direction"`
+	// SyncReplicas enables sync for all replicas in the selected deployment.
+	SyncReplicas bool `yaml:"syncReplicas,omitempty" json:"syncReplicas,omitempty"`
 
 	// BandwidthLimits can be used to limit the amount of bytes that are transferred by DevSpace with this
 	// sync configuration

pkg/devspace/services/podreplace/builder.go

@@ -13,6 +13,7 @@ import (
 	devspacecontext "github.com/loft-sh/devspace/pkg/devspace/context"
 	"github.com/loft-sh/devspace/pkg/devspace/kubectl/selector"
 	"github.com/loft-sh/devspace/pkg/util/hash"
+	"github.com/loft-sh/devspace/pkg/util/ptr"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	appsv1 "k8s.io/api/apps/v1"
@@ -156,6 +157,7 @@ func buildDeployment(ctx devspacecontext.Context, name string, target runtime.Ob
 	}
 
 	deployment.Spec.Template = *podTemplate
+	deployment.Spec.Replicas = desiredReplacementReplicas(target, devPod)
 	if deployment.Spec.Selector == nil {
 		deployment.Spec.Selector = &metav1.LabelSelector{}
 	}
@@ -177,6 +179,45 @@ func buildDeployment(ctx devspacecontext.Context, name string, target runtime.Ob
 	return deployment, nil
 }
 
+func desiredReplacementReplicas(target runtime.Object, devPod *latest.DevPod) *int32 {
+	// Preserve historical behavior unless syncReplicas is explicitly enabled.
+	if !hasSyncReplicas(devPod) {
+		return ptr.Int32(1)
+	}
+
+	switch t := target.(type) {
+	case *appsv1.ReplicaSet:
+		if t.Spec.Replicas != nil {
+			return ptr.Int32(*t.Spec.Replicas)
+		}
+	case *appsv1.Deployment:
+		if t.Spec.Replicas != nil {
+			return ptr.Int32(*t.Spec.Replicas)
+		}
+	case *appsv1.StatefulSet:
+		if t.Spec.Replicas != nil {
+			return ptr.Int32(*t.Spec.Replicas)
+		}
+	}
+
+	return ptr.Int32(1)
+}
+
+func hasSyncReplicas(devPod *latest.DevPod) bool {
+	hasSyncReplicas := false
+	loader.EachDevContainer(devPod, func(devContainer *latest.DevContainer) bool {
+		for _, s := range devContainer.Sync {
+			if s.SyncReplicas {
+				hasSyncReplicas = true
+				return false
+			}
+		}
+		return true
+	})
+
+	return hasSyncReplicas
+}
+
 func modifyDevContainer(ctx devspacecontext.Context, devPod *latest.DevPod, devContainer *latest.DevContainer, podTemplate *corev1.PodTemplateSpec) error {
 	err := replaceImage(ctx, devPod, devContainer, podTemplate)
 	if err != nil {

pkg/devspace/services/podreplace/replace.go

@@ -158,7 +158,7 @@ func updateNeeded(ctx devspacecontext.Context, deployment *appsv1.Deployment, de
 
 	// update deployment
 	originalDeployment := deployment.DeepCopy()
-	deployment.Spec.Replicas = ptr.Int32(1)
+	deployment.Spec.Replicas = newDeployment.Spec.Replicas
 	deployment.Spec.Selector = newDeployment.Spec.Selector
 	deployment.Spec.Template = newDeployment.Spec.Template
 	deployment.Annotations = newDeployment.Annotations