add ziti run quickstart cluster

This commit is contained in:
dovholuknf
2026-06-18 17:15:41 -04:00
parent c2e6c8ee3d
commit e2d328985a
9 changed files with 1165 additions and 27 deletions
+18
View File
@@ -3,6 +3,7 @@
## What's New
* [Cluster Quorum Recover](#cluster_quorum_recovery) - A mechanism for recovering clusters that have irrevocably lost the ability to form a quorum
* [Quickstart Cluster](#quickstart-cluster) - `ziti run quickstart cluster` brings up a multi-node HA cluster in a single command for testing and development
* [Fully Connected Controller Mesh](#fully-connected-controller-mesh) - Controllers now proactively keep the cluster mesh fully connected
* [Config Type Target Field](#config-type-target-field) - Config types now have a target field indicating whether they apply to services, routers or other entities
* [Wildcard OIDC Issuers](#wildcard-oidc-issuers) - Controllers with a wildcard server-certificate SAN can serve OIDC for explicitly allow-listed hostnames
@@ -32,6 +33,23 @@ controller and add new peers normally with `ziti ops cluster add`.
* Allow hosting-side crypto material to be generated on per connection basis (instead of per terminator)
## Quickstart Cluster
`ziti run quickstart cluster` stands up a multi-node HA controller cluster with a single command, for
testing, learning, and local development. It launches one quickstart child process per node (default 3,
minimum 3, configurable with `--size`), initializes the first
node, joins the rest, and prints a banner once the whole cluster is online listing each node's controller
address, router address, process id, and per-node log file.
Each node runs as its own operating-system process, so you can stop, restart, or attach a debugger to any
single node to explore HA behavior without disturbing the others. The banner prints the exact
`ziti run quickstart` command needed to start each node by hand.
Lifecycle mirrors the single-node quickstart. Pass `--home` for a persistent cluster you can stop and start
again: restarting against an existing `--home` rejoins the existing cluster rather than re-initializing it,
and the nodes start together to re-form a quorum. Omit `--home` to run from a temporary directory that is
removed on a clean shutdown. Pressing Ctrl-C stops every node.
## Fully Connected Controller Mesh
In an HA cluster, controllers form a mesh of channel connections that raft uses to
@@ -0,0 +1,40 @@
//go:build cli_tests && !windows
/*
Copyright NetFoundry Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cli_tests
import (
"os/exec"
"syscall"
)
// newProcessGroupAttr leaves the child in this test's process group. SIGINT is
// delivered to the parent pid directly, so a new group is not needed.
func newProcessGroupAttr() *syscall.SysProcAttr {
return nil
}
// gracefulStop sends SIGINT to the parent, which the quickstart cluster parent
// handles as its stop signal and relays to its children.
func gracefulStop(cmd *exec.Cmd) error {
return cmd.Process.Signal(syscall.SIGINT)
}
// forceKillTree kills the parent process.
func forceKillTree(cmd *exec.Cmd) {
_ = cmd.Process.Kill()
}
@@ -0,0 +1,44 @@
//go:build cli_tests && windows
/*
Copyright NetFoundry Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cli_tests
import (
"os/exec"
"strconv"
"syscall"
"golang.org/x/sys/windows"
)
// newProcessGroupAttr starts the child in its own process group so it can be
// targeted by a CTRL_BREAK console event while still sharing this test's console.
func newProcessGroupAttr() *syscall.SysProcAttr {
return &syscall.SysProcAttr{CreationFlags: windows.CREATE_NEW_PROCESS_GROUP}
}
// gracefulStop delivers CTRL_BREAK to the child's process group. The Go runtime
// maps it to SIGINT, which the quickstart cluster parent handles as its stop
// signal. It fails if this process has no console attached.
func gracefulStop(cmd *exec.Cmd) error {
return windows.GenerateConsoleCtrlEvent(windows.CTRL_BREAK_EVENT, uint32(cmd.Process.Pid))
}
// forceKillTree kills the child and its descendants.
func forceKillTree(cmd *exec.Cmd) {
_ = exec.Command("taskkill", "/T", "/F", "/PID", strconv.Itoa(cmd.Process.Pid)).Run()
}
+211
View File
@@ -0,0 +1,211 @@
//go:build cli_tests
/*
Copyright NetFoundry Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package cli_tests
import (
"context"
"crypto/x509"
"fmt"
"net"
"os"
"os/exec"
"path/filepath"
"regexp"
"strconv"
"testing"
"time"
"github.com/openziti/edge-api/rest_util"
"github.com/stretchr/testify/require"
)
// Test_Quickstart_Cluster brings up a 3-node cluster via `ziti run quickstart cluster` then:
// - asserts every node's controller comes up and accepts an admin login. Only
// node 1 initializes the admin, so a successful login on nodes 2 and 3 proves
// raft replicated the admin identity and the cluster formed.
// - delivers a graceful stop to the parent (SIGINT on POSIX, a CTRL_BREAK
// console event on Windows) and asserts the auto-created temp home is removed.
// If the stop signal cannot be delivered (e.g. no console attached), it
// force-kills and skips the shutdown assertions.
func Test_Quickstart_Cluster(t *testing.T) {
zitiPath := os.Getenv("ZITI_CLI_TEST_ZITI_BIN")
if zitiPath == "" {
t.Skip("ZITI_CLI_TEST_ZITI_BIN not set")
}
if _, statErr := os.Stat(zitiPath); statErr != nil {
t.Fatalf("ziti binary not found at %s: %v", zitiPath, statErr)
}
const size = 3
// One contiguous block split in half so the ctrl range (base..base+size-1)
// and router range (base+size..base+2*size-1) never overlap.
base := findConsecutivePorts(t, size*2)
ctrlBase := base
routerBase := base + size
cfgDir := filepath.Join(t.TempDir(), "cli-config")
logPath := filepath.Join(t.TempDir(), "cluster.log")
logFile, err := os.Create(logPath)
require.NoError(t, err)
defer func() { _ = logFile.Close() }()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// No --home on purpose: the cluster creates a temp dir and removes it on a
// clean shutdown. Its path is read from the process output.
args := []string{
"run", "quickstart", "cluster",
"--size", strconv.Itoa(size),
"--ctrl-address", "localhost",
"--router-address", "localhost",
fmt.Sprintf("--ctrl-port=%d", ctrlBase),
fmt.Sprintf("--router-port=%d", routerBase),
}
t.Logf("starting: %s %v", zitiPath, args)
clusterCmd := exec.CommandContext(ctx, zitiPath, args...)
clusterCmd.Env = append(os.Environ(), "PFXLOG_NO_JSON=true", "ZITI_CONFIG_DIR="+cfgDir)
clusterCmd.SysProcAttr = newProcessGroupAttr()
clusterCmd.Stdout = logFile
clusterCmd.Stderr = logFile
require.NoError(t, clusterCmd.Start())
defer func() {
if clusterCmd.Process == nil {
return
}
// Backstop in case the test returns before the clean shutdown below.
if gracefulStop(clusterCmd) != nil {
forceKillTree(clusterCmd)
}
waited := make(chan struct{})
go func() {
_, _ = clusterCmd.Process.Wait()
close(waited)
}()
select {
case <-waited:
case <-time.After(30 * time.Second):
}
}()
// Discover the auto-created temp home from the process output.
reHome := regexp.MustCompile(`temporary --home '([^']+)'`)
var tempHome string
deadline := time.Now().Add(30 * time.Second)
for time.Now().Before(deadline) && tempHome == "" {
data, _ := os.ReadFile(logPath)
if m := reHome.FindStringSubmatch(string(data)); m != nil {
tempHome = m[1]
} else {
time.Sleep(250 * time.Millisecond)
}
}
require.NotEmpty(t, tempHome, "did not observe an auto-created temp home in cluster output (see %s)", logPath)
t.Logf("cluster temp home: %s", tempHome)
// Every node must come up and accept an admin login. Nodes 2 and 3 never run
// init, so a successful admin login there proves cluster replication.
for i := 0; i < size; i++ {
ctrlUrl := fmt.Sprintf("https://localhost:%d", int(ctrlBase)+i)
require.NoErrorf(t, waitClusterNodeReady(ctrlUrl, "admin", "admin", 180*time.Second),
"node %d (%s) never became ready; see %s", i+1, ctrlUrl, logPath)
t.Logf("node %d ready and admin login succeeded at %s", i+1, ctrlUrl)
}
// Clean shutdown: deliver a graceful stop to the parent. It relays to the
// children, then removes the temp home.
if stopErr := gracefulStop(clusterCmd); stopErr != nil {
forceKillTree(clusterCmd)
_, _ = clusterCmd.Process.Wait()
t.Skipf("graceful stop signal not deliverable in this environment (%v); verified 3-node bring-up, skipping shutdown assertions", stopErr)
}
exited := make(chan error, 1)
go func() { exited <- clusterCmd.Wait() }()
select {
case <-exited:
case <-time.After(90 * time.Second):
t.Fatalf("cluster did not exit within 90s of the stop signal; see %s", logPath)
}
_, statErr := os.Stat(tempHome)
require.Truef(t, os.IsNotExist(statErr),
"temp home %s should have been removed after a clean shutdown (stat err: %v)", tempHome, statErr)
}
// waitClusterNodeReady polls until the controller at ctrlUrl serves its CA bundle
// and accepts an admin UPDB login, or the timeout elapses.
func waitClusterNodeReady(ctrlUrl, user, pass string, timeout time.Duration) error {
deadline := time.Now().Add(timeout)
var lastErr error
for time.Now().Before(deadline) {
caCerts, err := rest_util.GetControllerWellKnownCas(ctrlUrl)
if err != nil {
lastErr = err
time.Sleep(time.Second)
continue
}
pool := x509.NewCertPool()
for _, ca := range caCerts {
pool.AddCert(ca)
}
if _, err := rest_util.NewEdgeManagementClientWithUpdb(user, pass, ctrlUrl, pool); err != nil {
lastErr = err
time.Sleep(time.Second)
continue
}
return nil
}
return fmt.Errorf("not ready within %s: %w", timeout, lastErr)
}
// findConsecutivePorts returns a base port p such that p..p+n-1 are all bindable.
// Note: there is an inherent TOCTOU race between closing these listeners here and
// the child processes binding them. This is an accepted limitation of testing
// with external processes. Under heavy concurrent load a bind can still lose.
func findConsecutivePorts(t *testing.T, n int) uint16 {
t.Helper()
for attempt := 0; attempt < 100; attempt++ {
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
continue
}
base := l.Addr().(*net.TCPAddr).Port
_ = l.Close()
if base == 0 || base+n-1 > 65535 {
continue
}
held := make([]net.Listener, 0, n)
ok := true
for i := 0; i < n; i++ {
li, e := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", base+i))
if e != nil {
ok = false
break
}
held = append(held, li)
}
for _, li := range held {
_ = li.Close()
}
if ok {
return uint16(base)
}
}
t.Fatalf("could not find %d consecutive free ports", n)
return 0
}
+28 -27
View File
@@ -125,6 +125,7 @@ func NewQuickStartCmd(out io.Writer, errOut io.Writer, context context.Context)
addCommonQuickstartFlags(cmd, options)
addQuickstartHaFlags(cmd, options)
cmd.AddCommand(NewQuickStartJoinClusterCmd(out, errOut, context))
cmd.AddCommand(NewQuickStartClusterCmd(out, errOut, context))
return cmd
}
@@ -316,7 +317,9 @@ func (o *QuickstartOpts) run(ctx context.Context) error {
p := common.NewOptionsProvider(o.out, o.errOut)
fmt.Println("waiting three seconds for controller to become ready...")
if !o.joinCommand {
if o.AlreadyInitialized {
logrus.Infof("instance %s already initialized; skipping cluster init/join and rejoining the existing cluster", o.InstanceID)
} else if !o.joinCommand {
maxRetries := 5
for attempt := 1; attempt <= maxRetries; attempt++ {
fmt.Printf("initializing controller at port: %d\n", o.ControllerPort)
@@ -346,35 +349,32 @@ func (o *QuickstartOpts) run(ctx context.Context) error {
}
}
} else {
agentJoinCmd := agentcli.NewAgentClusterAdd(p)
// Joining can fail transiently while the target elects a leader, so retry until it succeeds or the deadline elapses.
o.waitForLeader()
joinDeadline := time.Now().Add(90 * time.Second)
attempt := 0
for {
attempt++
agentJoinCmd := agentcli.NewAgentClusterAdd(p)
agentJoinCmd.SetArgs([]string{
o.ClusterMember,
fmt.Sprintf("--pid=%d", os.Getpid()),
fmt.Sprintf("--voter=%t", !o.nonVoter),
"--timeout=30s",
})
args := []string{
o.ClusterMember,
fmt.Sprintf("--pid=%d", os.Getpid()),
fmt.Sprintf("--voter=%t", !o.nonVoter),
"--timeout=30s",
}
agentJoinCmd.SetArgs(args)
addChan := make(chan error, 1)
addTimeout := time.Second * 30
go func() {
o.waitForLeader()
addChan <- agentJoinCmd.Execute()
}()
select {
case agentJoinErr := <-addChan:
if agentJoinErr != nil {
joinErr := agentJoinCmd.Execute()
if joinErr == nil {
logrus.Infof("add command successful after %d attempt(s). continuing...", attempt)
break
}
if time.Now().After(joinDeadline) {
o.cleanupHome()
cancel()
return fmt.Errorf("failed to join cluster: %w", agentJoinErr)
return fmt.Errorf("failed to join cluster after %d attempt(s): %w", attempt, joinErr)
}
logrus.Info("Add command successful. continuing...")
case <-time.After(addTimeout):
o.cleanupHome()
cancel()
return fmt.Errorf("timed out adding to cluster")
logrus.Warnf("join attempt %d failed: %v, retrying", attempt, joinErr)
time.Sleep(2 * time.Second)
}
}
@@ -387,6 +387,7 @@ func (o *QuickstartOpts) run(ctx context.Context) error {
o.runRouter(erConfigFile)
ch := make(chan os.Signal, 1)
// os.Interrupt also catches a relayed Windows CTRL_BREAK which the Go runtime maps to SIGINT
signal.Notify(ch, os.Interrupt, syscall.SIGQUIT, syscall.SIGINT, syscall.SIGTERM)
if !o.Routerless {
@@ -419,7 +420,7 @@ func (o *QuickstartOpts) run(ctx context.Context) error {
cont = "`"
}
fmt.Println("Quickly add another member to this cluster using: ")
fmt.Printf(" ziti edge quickstart join %s\n", cont)
fmt.Printf(" ziti run quickstart join %s\n", cont)
fmt.Printf(" --ctrl-port %d %s\n", o.ControllerPort+1, cont)
fmt.Printf(" --router-port %d %s\n", o.RouterPort+1, cont)
fmt.Printf(" --home \"%s\" %s\n", o.Home, cont)
+518
View File
@@ -0,0 +1,518 @@
/*
Copyright NetFoundry Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package run
import (
"bytes"
"context"
"fmt"
"io"
"os"
"os/exec"
"os/signal"
"path/filepath"
"strconv"
"strings"
"sync"
"syscall"
"time"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/ziti/v2/ziti/cmd/helpers"
"github.com/openziti/ziti/v2/ziti/constants"
"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
)
// QuickstartClusterOpts drives `quickstart cluster`: it launches one
// `ziti run quickstart` child process per node and joins them into one HA cluster.
// The parent owns the children's lifecycle. Ctrl-C stops every node, and a
// parent-created temp home is removed on a clean exit.
type QuickstartClusterOpts struct {
Home string
Username string
Password string
ControllerAddress string
RouterAddress string
CtrlPort uint16
RouterPort uint16
TrustDomain string
Size int
ShutdownGrace time.Duration
out io.Writer
errOut io.Writer
verbose bool
cleanOnExit bool
}
func NewQuickStartClusterCmd(out io.Writer, errOut io.Writer, ctx context.Context) *cobra.Command {
options := &QuickstartClusterOpts{}
defaultCtrlPort, _ := strconv.ParseInt(constants.DefaultCtrlEdgeAdvertisedPort, 10, 16)
defaultRouterPort, _ := strconv.ParseInt(constants.DefaultZitiEdgeRouterPort, 10, 16)
cmd := &cobra.Command{
Use: "cluster",
Short: "runs a multi-node OpenZiti cluster, each node a quickstart, in child processes",
Long: "runs a multi-node OpenZiti cluster by launching one quickstart child process per node and joining " +
"them into a single raft cluster, suitable for testing and development. Pressing Ctrl-C stops every " +
"node. If --home is omitted a temporary directory is created and removed on exit.",
RunE: func(cmd *cobra.Command, args []string) error {
options.out = out
options.errOut = errOut
return options.run(ctx)
},
}
cmd.Flags().StringVar(&options.Home, "home", "", "permanent directory to use, or a temp dir (removed on exit) if omitted")
cmd.Flags().StringVarP(&options.Username, "username", "u", "", "admin username, default: admin")
cmd.Flags().StringVarP(&options.Password, "password", "p", "", "admin password, default: admin")
cmd.Flags().StringVar(&options.ControllerAddress, "ctrl-address", "", "advertised controller address used by every node. current: "+helpers.GetCtrlEdgeAdvertisedAddress())
cmd.Flags().StringVar(&options.RouterAddress, "router-address", "", "advertised router address used by every node")
cmd.Flags().Uint16Var(&options.CtrlPort, "ctrl-port", uint16(defaultCtrlPort), "base controller port (node index N listens on base+N)")
cmd.Flags().Uint16Var(&options.RouterPort, "router-port", uint16(defaultRouterPort), "base router port (node index N listens on base+N)")
cmd.Flags().StringVar(&options.TrustDomain, "trust-domain", "quickstart", "trust domain used in SPIFFE ids")
cmd.Flags().IntVar(&options.Size, "size", 3, "number of nodes in the cluster (minimum 3)")
cmd.Flags().DurationVar(&options.ShutdownGrace, "shutdown-grace", 30*time.Second, "max time to wait for nodes to shut down cleanly before the parent gives up waiting")
cmd.Flags().BoolVar(&options.verbose, "verbose", false, "show additional output")
return cmd
}
type nodeExit struct {
idx int
err error
}
func (o *QuickstartClusterOpts) run(ctx context.Context) error {
if o.verbose {
pfxlog.GlobalInit(logrus.DebugLevel, pfxlog.DefaultOptions().Color())
}
if o.Size < 3 {
return fmt.Errorf("--size must be at least 3 (an HA cluster needs at least 3 nodes to tolerate a failure)")
}
if o.Username == "" {
o.Username = "admin"
}
if o.Password == "" {
o.Password = "admin"
}
if strings.TrimSpace(o.TrustDomain) == "" {
o.TrustDomain = "quickstart"
}
if err := o.resolveHome(); err != nil {
return err
}
ctrlAddr := o.ControllerAddress
if ctrlAddr == "" {
ctrlAddr = helpers.GetCtrlEdgeAdvertisedAddress()
}
self, err := os.Executable()
if err != nil {
return fmt.Errorf("could not determine path to this executable: %w", err)
}
// Children run in their own process group (see configureChildProcAttr). The
// parent catches the shutdown signal and relays a clean stop to each child
// (CTRL_BREAK on Windows, SIGINT on POSIX). Nothing force-kills a child. If a
// node will not stop, the temp home is left in place rather than removed under
// a live process.
sigCtx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM)
defer stop()
var children []*exec.Cmd
var logFiles []*os.File
var monitors sync.WaitGroup
exitCh := make(chan nodeExit, o.Size)
cleanup := func() {
for _, c := range children {
relayStop(c)
}
done := make(chan struct{})
go func() {
monitors.Wait()
close(done)
}()
cleanExit := true
select {
case <-done:
case <-time.After(o.ShutdownGrace):
cleanExit = false
logrus.Warnf("not all nodes stopped within %s, leaving the environment in place rather than deleting it under a running node", o.ShutdownGrace)
}
for _, f := range logFiles {
_ = f.Close()
}
switch {
case !o.cleanOnExit:
fmt.Println("environment left intact at: " + o.Home)
case !cleanExit:
fmt.Println("temp directory NOT removed because nodes are still running: " + o.Home)
default:
fmt.Println("removing temp directory at: " + o.Home)
_ = os.RemoveAll(o.Home)
}
}
readyChans := make([]chan struct{}, o.Size)
startNode := func(idx int) error {
cmd := exec.Command(self, o.childArgs(idx, ctrlAddr)...)
// Give each node its own ziti CLI config/session dir. The CLI otherwise
// shares one per-user location and the nodes' concurrent logins collide.
instDir := filepath.Join(o.Home, fmt.Sprintf("instance-%d", idx+1))
cliConfigDir := filepath.Join(instDir, "ziti-cli")
cmd.Env = append(os.Environ(), "ZITI_CONFIG_DIR="+cliConfigDir)
prefix := fmt.Sprintf("[instance-%d] ", idx+1)
ready := make(chan struct{})
readyChans[idx] = ready
// Mirror each node's output to a per-node log file as well as the merged,
// prefixed console stream. The console writer also detects the readiness
// marker. The file gets the raw, unprefixed output.
stdout := io.Writer(newReadyWriter(o.out, prefix, nodeReadyMarker, func() { close(ready) }))
stderr := io.Writer(newPrefixWriter(o.errOut, prefix))
// The instance dir holds the node's db, pki, and CLI config dir, so it
// must exist before the node starts.
if mkErr := os.MkdirAll(instDir, 0o755); mkErr != nil {
return fmt.Errorf("could not create instance directory %s for node %d: %w", instDir, idx+1, mkErr)
}
if logFile, ferr := os.OpenFile(o.nodeLogPath(idx), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644); ferr == nil {
logFiles = append(logFiles, logFile)
stdout = io.MultiWriter(stdout, logFile)
stderr = io.MultiWriter(stderr, logFile)
} else {
logrus.Warnf("could not open log file for node %d: %v", idx+1, ferr)
}
cmd.Stdout = stdout
cmd.Stderr = stderr
configureChildProcAttr(cmd)
if startErr := cmd.Start(); startErr != nil {
return fmt.Errorf("failed to start node %d: %w", idx+1, startErr)
}
children = append(children, cmd)
monitors.Add(1)
go func() {
defer monitors.Done()
exitCh <- nodeExit{idx: idx, err: cmd.Wait()}
}()
return nil
}
if o.isFullRestart() {
// Full restart: every node is already a raft member, so they start
// together to re-form a quorum (a lone node can't elect a leader).
logrus.Infof("existing cluster home detected at %s, starting all %d nodes together to re-form quorum", o.Home, o.Size)
for i := 0; i < o.Size; i++ {
fmt.Printf("starting cluster node %d of %d...\n", i+1, o.Size)
if startErr := startNode(i); startErr != nil {
cleanup()
return startErr
}
}
for i := 0; i < o.Size; i++ {
if waitErr := o.waitForNode(sigCtx, i, readyChans[i], exitCh); waitErr != nil {
cleanup()
return waitErr
}
if sigCtx.Err() != nil {
cleanup()
return nil
}
}
} else {
// First bring-up, or growing an existing single node into a cluster: node 1
// must be leader before the rest join, or they hit CLUSTER_NO_LEADER. Node 1
// initializes if fresh, or restarts as leader if it exists, then the rest join.
fmt.Printf("starting cluster node 1 of %d...\n", o.Size)
if startErr := startNode(0); startErr != nil {
cleanup()
return startErr
}
if waitErr := o.waitForNode(sigCtx, 0, readyChans[0], exitCh); waitErr != nil {
cleanup()
return waitErr
}
if sigCtx.Err() != nil {
cleanup()
return nil
}
// Start the remaining nodes, each joining node 1, serialized on readiness.
for i := 1; i < o.Size; i++ {
fmt.Printf("starting cluster node %d of %d...\n", i+1, o.Size)
if startErr := startNode(i); startErr != nil {
cleanup()
return startErr
}
if waitErr := o.waitForNode(sigCtx, i, readyChans[i], exitCh); waitErr != nil {
cleanup()
return waitErr
}
if sigCtx.Err() != nil {
cleanup()
return nil
}
}
}
// Let the nodes flush their startup output for a few seconds so the banner
// is not scrolled away by trailing log lines.
select {
case <-sigCtx.Done():
cleanup()
return nil
case <-time.After(3 * time.Second):
}
o.printDetails(ctrlAddr, children)
select {
case <-sigCtx.Done():
fmt.Println("\nshutdown signal received, stopping cluster nodes...")
case ne := <-exitCh:
if ne.err != nil {
fmt.Printf("\ncluster node %d exited unexpectedly (%v), stopping remaining nodes...\n", ne.idx+1, ne.err)
} else {
fmt.Printf("\ncluster node %d exited, stopping remaining nodes...\n", ne.idx+1)
}
}
cleanup()
return nil
}
// childArgs builds the argv for node idx, re-invoking this binary as a quickstart
// node. Node 0 initializes the cluster, later nodes join node 0.
func (o *QuickstartClusterOpts) childArgs(idx int, ctrlAddr string) []string {
ctrlPort := o.CtrlPort + uint16(idx)
routerPort := o.RouterPort + uint16(idx)
args := []string{"run", "quickstart"}
if idx > 0 {
args = append(args, "join")
}
args = append(args,
"--home", o.Home,
"--instance-id", fmt.Sprintf("instance-%d", idx+1),
"--ctrl-port", strconv.Itoa(int(ctrlPort)),
"--router-port", strconv.Itoa(int(routerPort)),
"--trust-domain", o.TrustDomain,
"--username", o.Username,
"--password", o.Password,
)
if o.ControllerAddress != "" {
args = append(args, "--ctrl-address", o.ControllerAddress)
}
if o.RouterAddress != "" {
args = append(args, "--router-address", o.RouterAddress)
}
if o.verbose {
args = append(args, "--verbose")
}
if idx > 0 {
args = append(args, "--cluster-member", fmt.Sprintf("tls:%s:%d", ctrlAddr, o.CtrlPort))
}
return args
}
// waitForNode blocks until node idx signals it is fully up (its ready channel is
// closed), the shutdown signal fires, the node exits early, or the timeout
// elapses.
func (o *QuickstartClusterOpts) waitForNode(ctx context.Context, idx int, ready <-chan struct{}, exitCh chan nodeExit) error {
timeout := 180 * time.Second
select {
case <-ready:
logrus.Infof("cluster node %d is up", idx+1)
return nil
case ne := <-exitCh:
// A node died during bring-up. Abort and let cleanup() stop the rest.
return fmt.Errorf("node %d exited before becoming ready: %v", ne.idx+1, ne.err)
case <-ctx.Done():
return nil
case <-time.After(timeout):
return fmt.Errorf("timed out after %s waiting for node %d to become ready", timeout, idx+1)
}
}
func (o *QuickstartClusterOpts) resolveHome() error {
if o.Home == "" {
tmpDir, err := os.MkdirTemp("", "quickstart-cluster")
if err != nil {
return fmt.Errorf("could not create temp directory: %w", err)
}
o.Home = tmpDir
o.cleanOnExit = true
logrus.Infof("temporary --home '%s' will be removed on exit", o.Home)
return nil
}
// Expand a leading ~ only (bare, or before a path separator). A ~ elsewhere
// is a literal character.
if o.Home == "~" || strings.HasPrefix(o.Home, "~/") || strings.HasPrefix(o.Home, `~\`) {
home, err := os.UserHomeDir()
if err != nil {
return fmt.Errorf("could not find user's home directory: %w", err)
}
o.Home = filepath.Join(home, strings.TrimLeft(o.Home[1:], `/\`))
}
logrus.Infof("permanent --home '%s' will not be removed on exit", o.Home)
return nil
}
func (o *QuickstartClusterOpts) printDetails(ctrlAddr string, children []*exec.Cmd) {
fmt.Println("=======================================================================================")
fmt.Printf("cluster of %d nodes started.\n", o.Size)
for i := 0; i < o.Size; i++ {
pid := 0
if i < len(children) && children[i].Process != nil {
pid = children[i].Process.Pid
}
fmt.Printf(" node %d controller: %s:%d router: %s:%d pid: %d\n",
i+1, ctrlAddr, o.CtrlPort+uint16(i), o.routerAddrOrDefault(), o.RouterPort+uint16(i), pid)
}
fmt.Println(" home directory : " + o.Home)
fmt.Println(" configured trust domain: " + o.TrustDomain)
fmt.Println(" per-node logs:")
for i := 0; i < o.Size; i++ {
fmt.Printf(" node %d: %s\n", i+1, o.nodeLogPath(i))
}
exe := "ziti"
if self, err := os.Executable(); err == nil {
exe = self
}
// PowerShell continues lines with a backtick, POSIX shells with a backslash.
cont := "\\"
if os.Getenv("PSModulePath") != "" {
cont = "`"
}
home := o.Home
if strings.ContainsAny(home, " \t") {
home = `"` + home + `"`
}
fmt.Println()
fmt.Println(" to run a node individually (e.g. in separate terminals, or to restart one):")
for i := 0; i < o.Size; i++ {
fmt.Printf(" node %d:\n", i+1)
fmt.Printf(" %s run quickstart %s\n", exe, cont)
fmt.Printf(" --home %s %s\n", home, cont)
fmt.Printf(" --instance-id instance-%d %s\n", i+1, cont)
fmt.Printf(" --ctrl-port %d %s\n", o.CtrlPort+uint16(i), cont)
fmt.Printf(" --router-port %d\n", o.RouterPort+uint16(i))
}
fmt.Println()
fmt.Println(" press Ctrl-C here to stop all nodes.")
fmt.Println("=======================================================================================")
}
// nodeLogPath is the per-node log file each node's output is mirrored to.
func (o *QuickstartClusterOpts) nodeLogPath(idx int) string {
return filepath.Join(o.Home, fmt.Sprintf("instance-%d", idx+1), "quickstart.log")
}
// isFullRestart reports whether every node's data dir already exists in --home.
// If so, all nodes are existing raft members and start together to re-form
// quorum. If only some exist (growing a single node into a cluster, or a node
// that lost its data), node 1 comes up as leader first and the rest join.
func (o *QuickstartClusterOpts) isFullRestart() bool {
if o.Home == "" {
return false
}
for i := 0; i < o.Size; i++ {
// mirrors the child's own "already initialized" check (<home>/<instance>/db)
if _, err := os.Stat(filepath.Join(o.Home, fmt.Sprintf("instance-%d", i+1), "db")); err != nil {
return false
}
}
return true
}
func (o *QuickstartClusterOpts) routerAddrOrDefault() string {
if o.RouterAddress != "" {
return o.RouterAddress
}
return helpers.GetRouterAdvertisedAddress()
}
var prefixWriterMu sync.Mutex
// nodeReadyMarker is a line a quickstart node prints only once it is fully up:
// leader elected or joined, edge router enrolled, and router running. The parent
// gates the next node's start on seeing it.
const nodeReadyMarker = "Quickly add another member"
// prefixWriter prefixes each complete line written to it, so interleaved output
// from multiple child nodes stays attributable. A package-level mutex keeps lines
// from different writers from interleaving mid-line. If sentinel is non-empty, the
// first line containing it fires onSentinel exactly once.
type prefixWriter struct {
w io.Writer
prefix string
buf bytes.Buffer
sentinel string
onSentinel func()
once sync.Once
}
func newPrefixWriter(w io.Writer, prefix string) *prefixWriter {
return &prefixWriter{w: w, prefix: prefix}
}
func newReadyWriter(w io.Writer, prefix, sentinel string, onSentinel func()) *prefixWriter {
return &prefixWriter{w: w, prefix: prefix, sentinel: sentinel, onSentinel: onSentinel}
}
func (p *prefixWriter) Write(b []byte) (int, error) {
n, sawSentinel, err := p.writeLocked(b)
// Fire the readiness callback outside the lock: it runs external code (it
// closes a channel).
if sawSentinel {
p.once.Do(p.onSentinel)
}
return n, err
}
func (p *prefixWriter) writeLocked(b []byte) (int, bool, error) {
prefixWriterMu.Lock()
defer prefixWriterMu.Unlock()
n := len(b)
p.buf.Write(b)
sawSentinel := false
for {
line, err := p.buf.ReadBytes('\n')
if err != nil {
// no full line yet, keep the partial for next write
p.buf.Reset()
p.buf.Write(line)
break
}
if _, werr := io.WriteString(p.w, p.prefix); werr != nil {
return n, sawSentinel, werr
}
if _, werr := p.w.Write(line); werr != nil {
return n, sawSentinel, werr
}
if p.sentinel != "" && p.onSentinel != nil && bytes.Contains(line, []byte(p.sentinel)) {
sawSentinel = true
}
}
return n, sawSentinel, nil
}
@@ -0,0 +1,41 @@
//go:build !windows
/*
Copyright NetFoundry Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package run
import (
"os/exec"
"syscall"
)
// configureChildProcAttr puts each child in its own process group, so the
// terminal's Ctrl-C reaches the parent only and the parent controls shutdown.
func configureChildProcAttr(cmd *exec.Cmd) {
if cmd.SysProcAttr == nil {
cmd.SysProcAttr = &syscall.SysProcAttr{}
}
cmd.SysProcAttr.Setpgid = true
}
// relayStop signals a child to shut down gracefully. The quickstart node handles
// SIGINT as its shutdown trigger.
func relayStop(cmd *exec.Cmd) {
if cmd.Process != nil {
_ = cmd.Process.Signal(syscall.SIGINT)
}
}
@@ -0,0 +1,45 @@
//go:build windows
/*
Copyright NetFoundry Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package run
import (
"os/exec"
"syscall"
"golang.org/x/sys/windows"
)
// configureChildProcAttr starts each child in a new process group, so it is
// addressable for GenerateConsoleCtrlEvent and the console's Ctrl-C does not
// reach it directly. The parent controls shutdown via relayStop.
func configureChildProcAttr(cmd *exec.Cmd) {
if cmd.SysProcAttr == nil {
cmd.SysProcAttr = &syscall.SysProcAttr{}
}
cmd.SysProcAttr.CreationFlags |= windows.CREATE_NEW_PROCESS_GROUP
}
// relayStop sends CTRL_BREAK to the child's process group. The Go runtime maps
// CTRL_BREAK to SIGINT, which the quickstart node handles as its shutdown
// trigger. The child's pid is its process group id (CREATE_NEW_PROCESS_GROUP).
func relayStop(cmd *exec.Cmd) {
if cmd.Process != nil {
_ = windows.GenerateConsoleCtrlEvent(windows.CTRL_BREAK_EVENT, uint32(cmd.Process.Pid))
}
}
+220
View File
@@ -0,0 +1,220 @@
/*
Copyright NetFoundry Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package run
import (
"bytes"
"os"
"strings"
"testing"
)
// argValue returns the element following the first occurrence of flag, or "" if
// flag is absent or has no following element. childArgs emits flags and values
// as separate argv elements (e.g. "--ctrl-port", "1280").
func argValue(args []string, flag string) string {
for i, a := range args {
if a == flag && i+1 < len(args) {
return args[i+1]
}
}
return ""
}
func hasArg(args []string, flag string) bool {
for _, a := range args {
if a == flag {
return true
}
}
return false
}
func assertArg(t *testing.T, args []string, flag, want string) {
t.Helper()
if got := argValue(args, flag); got != want {
t.Errorf("%s = %q, want %q (args: %v)", flag, got, want, args)
}
}
func TestClusterChildArgs_InitNode(t *testing.T) {
o := &QuickstartClusterOpts{
Home: "/tmp/h", Username: "admin", Password: "secret",
CtrlPort: 1280, RouterPort: 3022, TrustDomain: "quickstart", Size: 3,
}
args := o.childArgs(0, "ctrl.example")
if len(args) < 2 || args[0] != "run" || args[1] != "quickstart" {
t.Fatalf("expected 'run quickstart' prefix, got %v", args)
}
if hasArg(args, "join") {
t.Errorf("node 0 must not be a join: %v", args)
}
if hasArg(args, "--cluster-member") {
t.Errorf("node 0 must have no --cluster-member: %v", args)
}
assertArg(t, args, "--instance-id", "instance-1")
assertArg(t, args, "--ctrl-port", "1280")
assertArg(t, args, "--router-port", "3022")
assertArg(t, args, "--trust-domain", "quickstart")
assertArg(t, args, "--username", "admin")
assertArg(t, args, "--password", "secret")
assertArg(t, args, "--home", "/tmp/h")
}
func TestClusterChildArgs_JoinNodePortsAndMember(t *testing.T) {
o := &QuickstartClusterOpts{
Home: "/tmp/h", Username: "admin", Password: "admin",
CtrlPort: 1280, RouterPort: 3022, TrustDomain: "qs", Size: 3,
}
// third node (index 2)
args := o.childArgs(2, "ctrl.example")
if !hasArg(args, "join") {
t.Errorf("node 2 must be a join: %v", args)
}
assertArg(t, args, "--instance-id", "instance-3")
// ports are base + index
assertArg(t, args, "--ctrl-port", "1282")
assertArg(t, args, "--router-port", "3024")
// every join targets node 0's controller (the BASE ctrl port), not its own
assertArg(t, args, "--cluster-member", "tls:ctrl.example:1280")
}
func TestClusterChildArgs_OptionalFlags(t *testing.T) {
with := &QuickstartClusterOpts{
Home: "/h", CtrlPort: 1280, RouterPort: 3022,
ControllerAddress: "cadr", RouterAddress: "radr", verbose: true,
}
a := with.childArgs(0, "cadr")
assertArg(t, a, "--ctrl-address", "cadr")
assertArg(t, a, "--router-address", "radr")
if !hasArg(a, "--verbose") {
t.Errorf("expected --verbose when set: %v", a)
}
without := &QuickstartClusterOpts{Home: "/h", CtrlPort: 1280, RouterPort: 3022}
b := without.childArgs(0, "x")
if hasArg(b, "--ctrl-address") || hasArg(b, "--router-address") {
t.Errorf("address flags must be omitted when unset: %v", b)
}
if hasArg(b, "--verbose") {
t.Errorf("--verbose must be omitted when unset: %v", b)
}
}
func TestPrefixWriter_PrefixesAndBuffersPartialLines(t *testing.T) {
var buf bytes.Buffer
w := newPrefixWriter(&buf, "[n1] ")
// a partial line should be held until its newline arrives
_, _ = w.Write([]byte("hello\nwor"))
if got := buf.String(); got != "[n1] hello\n" {
t.Fatalf("after partial write got %q", got)
}
_, _ = w.Write([]byte("ld\n"))
if got, want := buf.String(), "[n1] hello\n[n1] world\n"; got != want {
t.Fatalf("got %q want %q", got, want)
}
}
func TestReadyWriter_FiresOnceOnSentinel(t *testing.T) {
var buf bytes.Buffer
ready := make(chan struct{})
calls := 0
w := newReadyWriter(&buf, "[n1] ", "READY", func() {
calls++
close(ready)
})
_, _ = w.Write([]byte("starting up\n"))
select {
case <-ready:
t.Fatal("sentinel fired before its line was written")
default:
}
// two matching lines in one write, onSentinel must fire exactly once
_, _ = w.Write([]byte("now READY to serve\nstill READY\n"))
select {
case <-ready:
default:
t.Fatal("expected ready to be signaled after sentinel line")
}
// further matches must not re-invoke (a second close() would panic)
_, _ = w.Write([]byte("READY yet again\n"))
if calls != 1 {
t.Errorf("onSentinel called %d times, want 1", calls)
}
}
func TestResolveHome_TempCreatedAndMarkedForCleanup(t *testing.T) {
o := &QuickstartClusterOpts{}
if err := o.resolveHome(); err != nil {
t.Fatal(err)
}
defer func() {
if o.Home != "" {
_ = os.RemoveAll(o.Home)
}
}()
if o.Home == "" {
t.Fatal("expected a temp home to be created")
}
if !o.cleanOnExit {
t.Error("temp home must be marked cleanOnExit")
}
if _, err := os.Stat(o.Home); err != nil {
t.Errorf("temp home should exist: %v", err)
}
}
func TestResolveHome_ExplicitHomeNotCleaned(t *testing.T) {
dir := t.TempDir()
o := &QuickstartClusterOpts{Home: dir}
if err := o.resolveHome(); err != nil {
t.Fatal(err)
}
if o.cleanOnExit {
t.Error("explicit --home must not be marked cleanOnExit")
}
if o.Home != dir {
t.Errorf("explicit --home changed: got %q want %q", o.Home, dir)
}
}
func TestResolveHome_TildeExpanded(t *testing.T) {
home, err := os.UserHomeDir()
if err != nil {
t.Skip("no user home dir available")
}
o := &QuickstartClusterOpts{Home: "~/some-sub"}
if err := o.resolveHome(); err != nil {
t.Fatal(err)
}
if strings.Contains(o.Home, "~") {
t.Errorf("tilde not expanded: %q", o.Home)
}
if !strings.HasPrefix(o.Home, home) {
t.Errorf("expanded home %q should start with %q", o.Home, home)
}
if o.cleanOnExit {
t.Error("explicit ~ home must not be marked cleanOnExit")
}
}