Placement -> Scheduler CLI Changes

cmd/init.go

@@ -47,6 +47,7 @@ var (
 	imageVariant                       string
 	schedulerVolume                    string
 	schedulerOverrideBroadcastHostPort string
+	schedulerPlacement                 bool
 	redisStack                         bool
 )
 
@@ -195,6 +196,7 @@ dapr init --redis-stack
 				DaprInstallPath:                    runtime.GetDaprRuntimePath(),
 				SchedulerVolume:                    &schedulerVolume,
 				SchedulerOverrideBroadcastHostPort: schedulerHostPort,
+				SchedulerPlacement:                 schedulerPlacement,
 				RedisStack:                         redisStack,
 			})
 			if err != nil {
@@ -245,6 +247,7 @@ func init() {
 	InitCmd.Flags().StringVarP(&imageVariant, "image-variant", "", "", "The image variant to use for the Dapr runtime, for example: mariner")
 	InitCmd.Flags().StringVarP(&schedulerVolume, "scheduler-volume", "", "dapr_scheduler", "Self-hosted only. Specify a volume for the scheduler service data directory.")
 	InitCmd.Flags().StringVarP(&schedulerOverrideBroadcastHostPort, "scheduler-override-broadcast-host-port", "", "", "Self-hosted only. Specify the scheduler broadcast host and port, for example: 192.168.42.42:50006. If not specified, it uses localhost:50006 (6060 for Windows).")
+	InitCmd.Flags().BoolVarP(&schedulerPlacement, "scheduler-placement", "", false, "Self-hosted only. Serve actor placement from the scheduler service instead of running the placement service. Requires Dapr 1.19 or later.")
 	InitCmd.Flags().BoolVarP(&redisStack, "redis-stack", "", false, "Self-hosted only. Use redis-stack-server image instead of standard Redis for RediSearch support")
 	InitCmd.Flags().BoolP("help", "h", false, "Print this help message")
 	InitCmd.Flags().StringArrayVar(&values, "set", []string{}, "set values on the command line (can specify multiple or separate values with commas: key1=val1,key2=val2)")

cmd/renew_certificate.go

@@ -175,6 +175,7 @@ func restartControlPlaneService() error {
 		"deploy/dapr-sidecar-injector",
 		"deploy/dapr-operator",
 		"statefulsets/dapr-placement-server",
+		"statefulsets/dapr-scheduler-server",
 	}
 	namespace, err := kubernetes.GetDaprNamespace()
 	if err != nil {
@@ -187,15 +188,21 @@ func restartControlPlaneService() error {
 	for i, name := range controlPlaneServices {
 		go func(i int, name string) {
 			defer wg.Done()
+			// Not every service is deployed: the placement statefulset is
+			// absent when the scheduler serves actor placement.
+			if _, err := utils.RunCmdAndWait("kubectl", "get", "-n", namespace, name); err != nil {
+				print.InfoStatusEvent(os.Stdout, fmt.Sprintf("%s is not deployed, skipping restart", name))
+				return
+			}
 			print.InfoStatusEvent(os.Stdout, fmt.Sprintf("Restarting %s..", name))
 			_, err := utils.RunCmdAndWait("kubectl", "rollout", "restart", "-n", namespace, name)
 			if err != nil {
-				errs[i] = fmt.Errorf("error in restarting deployment %s. Error is %w", name, err)
+				errs[i] = fmt.Errorf("error in restarting %s. Error is %w", name, err)
 				return
 			}
 			_, err = utils.RunCmdAndWait("kubectl", "rollout", "status", "-n", namespace, name)
 			if err != nil {
-				errs[i] = fmt.Errorf("error in checking status for deployment %s. Error is %w", name, err)
+				errs[i] = fmt.Errorf("error in checking status for %s. Error is %w", name, err)
 				return
 			}
 		}(i, name)

pkg/standalone/standalone.go

@@ -27,6 +27,7 @@ import (
 	"path"
 	path_filepath "path/filepath"
 	"runtime"
+	"slices"
 	"strings"
 	"sync"
 	"time"
@@ -92,6 +93,8 @@ const (
 	schedulerEtcdPort   = 2379
 
 	daprVersionsWithScheduler = ">= 1.14.x"
+
+	daprVersionsWithSchedulerPlacement = ">= 1.19.x"
 )
 
 var (
@@ -145,6 +148,7 @@ type initInfo struct {
 	imageVariant                       string
 	schedulerVolume                    *string
 	schedulerOverrideBroadcastHostPort *string
+	schedulerPlacement                 bool
 	redisStack                         bool
 }
 
@@ -160,6 +164,7 @@ type InitOptions struct {
 	DaprInstallPath                    string
 	SchedulerVolume                    *string
 	SchedulerOverrideBroadcastHostPort *string
+	SchedulerPlacement                 bool
 	RedisStack                         bool
 }
 
@@ -182,6 +187,30 @@ func isBinaryInstallationRequired(binaryFilePrefix, binInstallDir string) (bool,
 	return true, nil
 }
 
+// isSchedulerPlacementIncluded returns true if the scheduler can serve actor
+// placement in a given version of Dapr.
+func isSchedulerPlacementIncluded(runtimeVersion string) (bool, error) {
+	if runtimeVersion == "edge" || runtimeVersion == "dev" {
+		return true, nil
+	}
+
+	c, err := semver.NewConstraint(daprVersionsWithSchedulerPlacement)
+	if err != nil {
+		return false, err
+	}
+
+	v, err := semver.NewVersion(runtimeVersion)
+	if err != nil {
+		return false, err
+	}
+
+	vNoPrerelease, err := v.SetPrerelease("")
+	if err != nil {
+		return false, err
+	}
+	return c.Check(&vNoPrerelease), nil
+}
+
 // isSchedulerIncluded returns true if scheduler is included a given version for Dapr.
 func isSchedulerIncluded(runtimeVersion string) (bool, error) {
 	c, err := semver.NewConstraint(daprVersionsWithScheduler)
@@ -274,6 +303,16 @@ func Init(opts InitOptions) error {
 
 	// After this point runtimeVersion will not be latest string but rather actual version.
 
+	if opts.SchedulerPlacement {
+		ok, serr := isSchedulerPlacementIncluded(runtimeVersion)
+		if serr != nil {
+			return serr
+		}
+		if !ok {
+			return fmt.Errorf("--scheduler-placement requires Dapr %s, got %s", daprVersionsWithSchedulerPlacement, runtimeVersion)
+		}
+	}
+
 	print.InfoStatusEvent(os.Stdout, "Installing runtime version %s", runtimeVersion)
 
 	installDir, err := GetDaprRuntimePath(daprInstallPath)
@@ -334,6 +373,7 @@ func Init(opts InitOptions) error {
 		imageVariant:                       imageVariant,
 		schedulerVolume:                    schedulerVolume,
 		schedulerOverrideBroadcastHostPort: schedulerOverrideBroadcastHostPort,
+		schedulerPlacement:                 opts.SchedulerPlacement,
 		redisStack:                         redisStack,
 	}
 	for _, step := range initSteps {
@@ -362,7 +402,9 @@ func Init(opts InitOptions) error {
 	print.InfoStatusEvent(os.Stdout, "%s binary has been installed to %s.", daprRuntimeFilePrefix, daprBinDir)
 	if slimMode {
 		// Print info on placement binary only on slim install.
-		print.InfoStatusEvent(os.Stdout, "%s binary has been installed to %s.", placementServiceFilePrefix, daprBinDir)
+		if !info.schedulerPlacement {
+			print.InfoStatusEvent(os.Stdout, "%s binary has been installed to %s.", placementServiceFilePrefix, daprBinDir)
+		}
 		print.InfoStatusEvent(os.Stdout, "%s binary has been installed to %s.", schedulerServiceFilePrefix, daprBinDir)
 	} else {
 		runtimeCmd := utils.GetContainerRuntimeCmd(info.containerRuntime)
@@ -371,6 +413,12 @@ func Init(opts InitOptions) error {
 		if isAirGapInit {
 			dockerContainerNames = []string{DaprPlacementContainerName}
 		}
+		if info.schedulerPlacement {
+			// The scheduler serves placement, so no placement container runs.
+			dockerContainerNames = slices.DeleteFunc(dockerContainerNames, func(name string) bool {
+				return name == DaprPlacementContainerName
+			})
+		}
 		hasScheduler, err := isSchedulerIncluded(info.runtimeVersion)
 		if err == nil && hasScheduler {
 			dockerContainerNames = append(dockerContainerNames, DaprSchedulerContainerName)
@@ -538,7 +586,7 @@ func redisImageInfo(redisStack bool, imageRegistryURL string, imageRegistryName
 func runPlacementService(wg *sync.WaitGroup, errorChan chan<- error, info initInfo) {
 	defer wg.Done()
 
-	if info.slimMode {
+	if info.slimMode || info.schedulerPlacement {
 		return
 	}
 
@@ -736,6 +784,10 @@ func runSchedulerService(wg *sync.WaitGroup, errorChan chan<- error, info initIn
 		args = append(args, "--etcd-client-listen-address=0.0.0.0")
 	}
 
+	if info.schedulerPlacement {
+		args = append(args, "--placement-enabled=true")
+	}
+
 	// On non-elevated Windows with WSL2 installed, verify the scheduler ports
 	// are free before attempting the container start, but only when the
 	// scheduler is publishing host ports. WSL2 commonly holds :2379 (etcd)
@@ -862,7 +914,7 @@ func installDaprRuntime(wg *sync.WaitGroup, errorChan chan<- error, info initInf
 func installPlacement(wg *sync.WaitGroup, errorChan chan<- error, info initInfo) {
 	defer wg.Done()
 
-	if !info.slimMode {
+	if !info.slimMode || info.schedulerPlacement {
 		return
 	}
 

pkg/standalone/standalone_test.go

@@ -386,6 +386,29 @@ func TestInitLogActualContainerRuntimeName(t *testing.T) {
 	}
 }
 
+func TestIsSchedulerPlacementIncluded(t *testing.T) {
+	scenarios := []struct {
+		version    string
+		isIncluded bool
+	}{
+		{"1.15.0", false},
+		{"1.17.0", false},
+		{"1.18.1", false},
+		{"1.19.0-rc.1", true},
+		{"1.19.0", true},
+		{"1.20.0", true},
+		{"edge", true},
+		{"dev", true},
+	}
+	for _, scenario := range scenarios {
+		t.Run("isSchedulerPlacementIncludedIn"+scenario.version, func(t *testing.T) {
+			included, err := isSchedulerPlacementIncluded(scenario.version)
+			assert.NoError(t, err)
+			assert.Equal(t, scenario.isIncluded, included)
+		})
+	}
+}
+
 func TestIsSchedulerIncluded(t *testing.T) {
 	scenarios := []struct {
 		version    string