Fix dapr init scheduler (#1428)

* Add E2E to validate scheduler after init.

Signed-off-by: Artur Souza <[email protected]>

* Adds dapr_scheduler verify container

Signed-off-by: joshvanl <[email protected]>

* Assert eventually TCP connect

Signed-off-by: joshvanl <[email protected]>

* Fix eventually t check

Signed-off-by: joshvanl <[email protected]>

* Write etcd-data-dir to custom path with volume

Signed-off-by: joshvanl <[email protected]>

* Adds Helpers to test funcs

Signed-off-by: joshvanl <[email protected]>

* Adds container name to TCP check

Signed-off-by: joshvanl <[email protected]>

* Use rc.3 for scheduler

Signed-off-by: joshvanl <[email protected]>

* Print container logs on failed TCP conn

Signed-off-by: joshvanl <[email protected]>

* Fix params

Signed-off-by: joshvanl <[email protected]>

* Use b

Signed-off-by: joshvanl <[email protected]>

* Addfs `dev` flag to rc init

Signed-off-by: joshvanl <[email protected]>

* Fix version check

Signed-off-by: joshvanl <[email protected]>

* Skip TCP check on slim mode

Signed-off-by: joshvanl <[email protected]>

* Remove debug test code

Signed-off-by: joshvanl <[email protected]>

---------

Signed-off-by: Artur Souza <[email protected]>
Signed-off-by: joshvanl <[email protected]>
Co-authored-by: Artur Souza <[email protected]>

Author: JoshVanL

pkg/standalone/standalone.go

@@ -643,7 +643,7 @@ func runSchedulerService(wg *sync.WaitGroup, errorChan chan<- error, info initIn
 		"--entrypoint", "./scheduler",
 	}
 	if info.schedulerVolume != nil {
-		args = append(args, "--volume", *info.schedulerVolume+":/data-default-dapr-scheduler-server-0")
+		args = append(args, "--volume", *info.schedulerVolume+":/var/run/dapr/scheduler")
 	}
 
 	if info.dockerNetwork != "" {
@@ -664,7 +664,7 @@ func runSchedulerService(wg *sync.WaitGroup, errorChan chan<- error, info initIn
 		)
 	}
 
-	args = append(args, image)
+	args = append(args, image, "--etcd-data-dir=/var/run/dapr/scheduler")
 
 	_, err = utils.RunCmdAndWait(runtimeCmd, args...)
 	if err != nil {

tests/e2e/standalone/init_test.go

@@ -18,12 +18,16 @@ package standalone_test
 
 import (
 	"context"
+	"net"
 	"os"
 	"path/filepath"
 	"runtime"
+	"strconv"
 	"strings"
 	"testing"
+	"time"
 
+	"github.com/Masterminds/semver"
 	"github.com/dapr/cli/tests/e2e/common"
 	"github.com/dapr/cli/tests/e2e/spawn"
 	"github.com/docker/docker/api/types"
@@ -157,6 +161,48 @@ func TestStandaloneInit(t *testing.T) {
 		verifyContainers(t, latestDaprRuntimeVersion)
 		verifyBinaries(t, daprPath, latestDaprRuntimeVersion, latestDaprDashboardVersion)
 		verifyConfigs(t, daprPath)
+
+		placementPort := 50005
+		if runtime.GOOS == "windows" {
+			placementPort = 6050
+		}
+
+		verifyTCPLocalhost(t,  placementPort)
+	})
+
+	t.Run("init version with scheduler", func(t *testing.T) {
+		// Ensure a clean environment
+		must(t, cmdUninstall, "failed to uninstall Dapr")
+
+		args := []string{
+			"--runtime-version", "1.14.0-rc.3",
+			"--dev",
+		}
+		output, err := cmdInit(args...)
+		t.Log(output)
+		require.NoError(t, err, "init failed")
+		assert.Contains(t, output, "Success! Dapr is up and running.")
+
+		homeDir, err := os.UserHomeDir()
+		require.NoError(t, err, "failed to get user home directory")
+
+		daprPath := filepath.Join(homeDir, ".dapr")
+		require.DirExists(t, daprPath, "Directory %s does not exist", daprPath)
+
+		_, latestDaprDashboardVersion := common.GetVersionsFromEnv(t, true)
+		verifyContainers(t, "1.14.0-rc.3")
+		verifyBinaries(t, daprPath, "1.14.0-rc.3", latestDaprDashboardVersion)
+		verifyConfigs(t, daprPath)
+
+		placementPort := 50005
+		schedulerPort := 50006
+		if runtime.GOOS == "windows" {
+			placementPort = 6050
+			schedulerPort = 6060
+		}
+
+		verifyTCPLocalhost(t,  placementPort)
+		verifyTCPLocalhost(t,  schedulerPort)
 	})
 
 	t.Run("init without runtime-version flag with mariner images", func(t *testing.T) {
@@ -187,10 +233,11 @@ func TestStandaloneInit(t *testing.T) {
 // Note, in case of slim installation, the containers are not installed and
 // this test is automatically skipped.
 func verifyContainers(t *testing.T, daprRuntimeVersion string) {
+	t.Helper()
+
 	t.Run("verifyContainers", func(t *testing.T) {
 		if isSlimMode() {
-			t.Log("Skipping container verification because of slim installation")
-			return
+			t.Skip("Skipping container verification because of slim installation")
 		}
 
 		cli, err := dockerClient.NewClientWithOpts(dockerClient.FromEnv)
@@ -205,6 +252,12 @@ func verifyContainers(t *testing.T, daprRuntimeVersion string) {
 			"dapr_redis":     "",
 		}
 
+		v, err := semver.NewVersion(daprRuntimeVersion)
+		require.NoError(t, err)
+		if v.Major() >= 1 && v.Minor() >= 14 {
+			daprContainers["dapr_scheduler"] = daprRuntimeVersion
+		}
+
 		for _, container := range containers {
 			t.Logf("Found container: %v %s %s\n", container.Names, container.Image, container.State)
 			if container.State != "running" {
@@ -233,6 +286,8 @@ func verifyContainers(t *testing.T, daprRuntimeVersion string) {
 
 // verifyBinaries ensures that the correct binaries are present in the correct path.
 func verifyBinaries(t *testing.T, daprPath, runtimeVersion, dashboardVersion string) {
+	t.Helper()
+
 	binPath := filepath.Join(daprPath, "bin")
 	require.DirExists(t, binPath, "Directory %s does not exist", binPath)
 
@@ -247,6 +302,8 @@ func verifyBinaries(t *testing.T, daprPath, runtimeVersion, dashboardVersion str
 
 	for bin, version := range binaries {
 		t.Run("verifyBinaries/"+bin, func(t *testing.T) {
+			t.Helper()
+
 			file := filepath.Join(binPath, bin)
 			if runtime.GOOS == "windows" {
 				file += ".exe"
@@ -265,6 +322,8 @@ func verifyBinaries(t *testing.T, daprPath, runtimeVersion, dashboardVersion str
 // verifyConfigs ensures that the Dapr configuration and component YAMLs
 // are present in the correct path and have the correct values.
 func verifyConfigs(t *testing.T, daprPath string) {
+	t.Helper()
+
 	configSpec := map[interface{}]interface{}{}
 	// tracing is not enabled in slim mode by default.
 	if !isSlimMode() {
@@ -353,3 +412,22 @@ func verifyConfigs(t *testing.T, daprPath string) {
 		})
 	}
 }
+
+// verifyTCPLocalhost verifies a given localhost TCP port is being listened to.
+func verifyTCPLocalhost(t *testing.T, port int) {
+	t.Helper()
+
+	if isSlimMode() {
+		t.Skip("Skipping container verification because of slim installation")
+	}
+
+	// Check that the server is up and can accept connections.
+	endpoint := "127.0.0.1:" + strconv.Itoa(port)
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		conn, err := net.Dial("tcp", endpoint)
+		//nolint:testifylint
+		if assert.NoError(c, err) {
+			conn.Close()
+		}
+	}, time.Second*10, time.Millisecond*10)
+}