From 31b8cd5effb23451aa46bffe6257c19eed53876f Mon Sep 17 00:00:00 2001 From: Cassandra Coyle Date: Mon, 24 Aug 2026 12:01:24 -0500 Subject: [PATCH 1/2] scheduler placement changes Signed-off-by: Cassandra Coyle --- cmd/init.go | 3 ++ cmd/renew_certificate.go | 7 ++++ pkg/standalone/standalone.go | 54 +++++++++++++++++++++++++++++-- pkg/standalone/standalone_test.go | 21 ++++++++++++ 4 files changed, 82 insertions(+), 3 deletions(-) diff --git a/cmd/init.go b/cmd/init.go index 528a1bfe36..11afa246aa 100644 --- a/cmd/init.go +++ b/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)") diff --git a/cmd/renew_certificate.go b/cmd/renew_certificate.go index fc9e12a619..7ea1cb67e6 100644 --- a/cmd/renew_certificate.go +++ b/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,6 +188,12 @@ 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 { diff --git a/pkg/standalone/standalone.go b/pkg/standalone/standalone.go index 5f505ca62e..cfceb07a80 100644 --- a/pkg/standalone/standalone.go +++ b/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,26 @@ 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) { + 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 +299,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 +369,7 @@ func Init(opts InitOptions) error { imageVariant: imageVariant, schedulerVolume: schedulerVolume, schedulerOverrideBroadcastHostPort: schedulerOverrideBroadcastHostPort, + schedulerPlacement: opts.SchedulerPlacement, redisStack: redisStack, } for _, step := range initSteps { @@ -362,7 +398,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 +409,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 +582,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 +780,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 +910,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 } diff --git a/pkg/standalone/standalone_test.go b/pkg/standalone/standalone_test.go index 7600547c4c..4f6099eacc 100644 --- a/pkg/standalone/standalone_test.go +++ b/pkg/standalone/standalone_test.go @@ -386,6 +386,27 @@ 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}, + } + 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 From ace7f627bc325e4e3bd1cf4234aaf95c4742ca78 Mon Sep 17 00:00:00 2001 From: Cassandra Coyle Date: Mon, 24 Aug 2026 13:41:40 -0500 Subject: [PATCH 2/2] PR feedback Signed-off-by: Cassandra Coyle --- cmd/renew_certificate.go | 4 ++-- pkg/standalone/standalone.go | 4 ++++ pkg/standalone/standalone_test.go | 2 ++ 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/cmd/renew_certificate.go b/cmd/renew_certificate.go index 7ea1cb67e6..50a9e69370 100644 --- a/cmd/renew_certificate.go +++ b/cmd/renew_certificate.go @@ -197,12 +197,12 @@ func restartControlPlaneService() error { 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) diff --git a/pkg/standalone/standalone.go b/pkg/standalone/standalone.go index cfceb07a80..77d8d48d12 100644 --- a/pkg/standalone/standalone.go +++ b/pkg/standalone/standalone.go @@ -190,6 +190,10 @@ func isBinaryInstallationRequired(binaryFilePrefix, binInstallDir string) (bool, // 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 diff --git a/pkg/standalone/standalone_test.go b/pkg/standalone/standalone_test.go index 4f6099eacc..e74ce90a5e 100644 --- a/pkg/standalone/standalone_test.go +++ b/pkg/standalone/standalone_test.go @@ -397,6 +397,8 @@ func TestIsSchedulerPlacementIncluded(t *testing.T) { {"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) {