mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-20 03:11:46 +00:00
feat: nats TLS (#144)
This commit is contained in:
Julian Sommer
@@ -83,7 +83,7 @@ jobs:
|
||||
SSH_KEY=$(echo "$INVENTORY" | jq -r '.ssh_key_path')
|
||||
SSH_KEY="${SSH_KEY/#\~/$HOME}"
|
||||
ssh -o StrictHostKeyChecking=no -i "$SSH_KEY" tf-user@${WAN_IP} \
|
||||
'cd ~/e2e-tests && ./run-e2e.sh && ./run-cert-e2e.sh && ./run-lb-e2e.sh'
|
||||
"cd ~/e2e-tests && ./run-e2e.sh && ./run-cert-e2e.sh && ENDPOINT=https://${WAN_IP}:9999 ./run-lb-e2e.sh"
|
||||
|
||||
- name: Teardown
|
||||
if: always()
|
||||
|
||||
@@ -22,7 +22,7 @@ Environment=SPINIFEX_AWSGW_TLS_KEY=/etc/spinifex/server.key
|
||||
NoNewPrivileges=yes
|
||||
ProtectSystem=strict
|
||||
PrivateDevices=yes
|
||||
ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/master.key /etc/spinifex/server.pem /etc/spinifex/server.key
|
||||
ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/master.key /etc/spinifex/server.pem /etc/spinifex/server.key /etc/spinifex/ca.pem
|
||||
ReadWritePaths=/var/lib/spinifex/awsgw /var/log/spinifex
|
||||
|
||||
# Hardening
|
||||
|
||||
@@ -23,7 +23,7 @@ PrivateDevices=no
|
||||
DevicePolicy=closed
|
||||
DeviceAllow=/dev/kvm rw
|
||||
DeviceAllow=/dev/net/tun rw
|
||||
ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/server.pem /etc/spinifex/server.key
|
||||
ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/server.pem /etc/spinifex/server.key /etc/spinifex/ca.pem
|
||||
ReadWritePaths=/var/lib/spinifex/spinifex /var/log/spinifex /run/spinifex /run/spinifex/nbd
|
||||
|
||||
# Hardening
|
||||
|
||||
@@ -17,7 +17,7 @@ Environment=SPINIFEX_CONFIG_PATH=/etc/spinifex/nats/nats.conf
|
||||
NoNewPrivileges=yes
|
||||
ProtectSystem=strict
|
||||
PrivateDevices=yes
|
||||
ReadOnlyPaths=/etc/spinifex/nats
|
||||
ReadOnlyPaths=/etc/spinifex/nats /etc/spinifex/server.pem /etc/spinifex/server.key /etc/spinifex/ca.pem
|
||||
ReadWritePaths=/var/lib/spinifex/nats /var/log/spinifex
|
||||
|
||||
# Hardening
|
||||
|
||||
@@ -22,7 +22,7 @@ Environment=SPINIFEX_PREDASTORE_TLS_KEY=/etc/spinifex/server.key
|
||||
NoNewPrivileges=yes
|
||||
ProtectSystem=strict
|
||||
PrivateDevices=yes
|
||||
ReadOnlyPaths=/etc/spinifex/predastore /etc/spinifex/spinifex.toml /etc/spinifex/master.key /etc/spinifex/server.pem /etc/spinifex/server.key
|
||||
ReadOnlyPaths=/etc/spinifex/predastore /etc/spinifex/spinifex.toml /etc/spinifex/master.key /etc/spinifex/server.pem /etc/spinifex/server.key /etc/spinifex/ca.pem
|
||||
ReadWritePaths=/var/lib/spinifex/predastore /var/log/spinifex
|
||||
|
||||
# Hardening
|
||||
|
||||
@@ -21,7 +21,7 @@ Environment=SPINIFEX_BASE_DIR=/var/lib/spinifex/viperblock/
|
||||
NoNewPrivileges=yes
|
||||
ProtectSystem=strict
|
||||
PrivateDevices=yes
|
||||
ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/systemd.env
|
||||
ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/systemd.env /etc/spinifex/ca.pem
|
||||
ReadWritePaths=/var/lib/spinifex/viperblock /var/log/spinifex /run/spinifex/nbd
|
||||
|
||||
# Hardening
|
||||
|
||||
@@ -19,7 +19,7 @@ Environment=SPINIFEX_BASE_DIR=/var/lib/spinifex/vpcd/
|
||||
# Filesystem access — NoNewPrivileges omitted (vpcd uses sudo for ovs-vsctl, ovn-nbctl, systemctl)
|
||||
ProtectSystem=full
|
||||
PrivateDevices=yes
|
||||
ReadOnlyPaths=/etc/spinifex/spinifex.toml
|
||||
ReadOnlyPaths=/etc/spinifex/spinifex.toml /etc/spinifex/ca.pem
|
||||
ReadWritePaths=/var/lib/spinifex/vpcd /var/log/spinifex
|
||||
|
||||
# Hardening
|
||||
|
||||
@@ -239,7 +239,7 @@ func init() {
|
||||
adminInitCmd.Flags().Int("nodes", 3, "Number of nodes to expect for cluster")
|
||||
adminInitCmd.Flags().String("host", "", "Leader node to join (if not specified, tries multicast discovery)")
|
||||
adminInitCmd.Flags().Int("port", 4432, "Port to bind cluster services on")
|
||||
adminInitCmd.Flags().String("bind", "0.0.0.0", "IP address to bind services to (e.g., 10.11.12.1 for multi-node)")
|
||||
adminInitCmd.Flags().String("bind", "127.0.0.1", "IP address to bind services to (e.g., 10.11.12.1 for multi-node)")
|
||||
adminInitCmd.Flags().String("cluster-bind", "", "IP address to bind NATS cluster services to (e.g., 10.11.12.1 for multi-node)")
|
||||
adminInitCmd.Flags().String("cluster-routes", "", "NATS cluster hosts for routing specify multiple with comma (e.g., 10.11.12.1:4248,10.11.12.2:4248 for multi-node)")
|
||||
adminInitCmd.Flags().String("predastore-nodes", "", "Comma-separated IPs for multi-node Predastore cluster (e.g., 10.11.12.1,10.11.12.2,10.11.12.3). Requires >= 3 nodes.")
|
||||
@@ -864,7 +864,7 @@ func runAdminInit(cmd *cobra.Command, args []string) {
|
||||
spxRoot = filepath.Clean(spxRoot)
|
||||
|
||||
// Determine if this is a multi-node formation
|
||||
isMultiNode := nodes >= 2 && bindIP != "0.0.0.0"
|
||||
isMultiNode := nodes >= 2 && bindIP != "0.0.0.0" && bindIP != "127.0.0.1"
|
||||
|
||||
if isMultiNode {
|
||||
// Build cluster-wide network config for propagation to joining nodes
|
||||
|
||||
@@ -90,7 +90,7 @@ func loadConfigAndConnect() (*config.ClusterConfig, *nats.Conn, error) {
|
||||
}
|
||||
|
||||
nodeConfig := cfg.Nodes[cfg.Node]
|
||||
nc, err := utils.ConnectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token)
|
||||
nc, err := utils.ConnectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token, nodeConfig.NATS.CACert)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to connect to NATS: %w", err)
|
||||
}
|
||||
|
||||
@@ -288,6 +288,7 @@ var viperblockStartCmd = &cobra.Command{
|
||||
service, err := service.New("viperblock", &viperblockd.Config{
|
||||
NatsHost: nodeConfig.NATS.Host,
|
||||
NatsToken: nodeConfig.NATS.ACL.Token,
|
||||
NatsCACert: nodeConfig.NATS.CACert,
|
||||
PluginPath: pluginPath,
|
||||
S3Host: nodeConfig.Predastore.Host,
|
||||
Bucket: nodeConfig.Predastore.Bucket,
|
||||
@@ -743,6 +744,7 @@ var vpcdStartCmd = &cobra.Command{
|
||||
svc, err := service.New("vpcd", &vpcd.Config{
|
||||
NatsHost: nodeConfig.NATS.Host,
|
||||
NatsToken: nodeConfig.NATS.ACL.Token,
|
||||
NatsCACert: nodeConfig.NATS.CACert,
|
||||
OVNNBAddr: nodeConfig.VPCD.OVNNBAddr,
|
||||
OVNSBAddr: nodeConfig.VPCD.OVNSBAddr,
|
||||
BaseDir: nodeConfig.BaseDir,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# spinifex-config-version: 2
|
||||
# spinifex-config-version: 3
|
||||
# NATS Server Configuration
|
||||
# Generated by Spinifex admin init
|
||||
|
||||
@@ -9,12 +9,25 @@ listen: {{.BindIP}}:4222
|
||||
listen: 0.0.0.0:4222
|
||||
{{- end }}
|
||||
|
||||
tls {
|
||||
cert_file: "{{.ConfigDir}}/server.pem"
|
||||
key_file: "{{.ConfigDir}}/server.key"
|
||||
ca_file: "{{.ConfigDir}}/ca.pem"
|
||||
}
|
||||
|
||||
{{- if and .ClusterBindIP .ClusterRoutes }}
|
||||
|
||||
cluster {
|
||||
name: {{ .ClusterName }}
|
||||
listen: {{.ClusterBindIP}}:4248
|
||||
|
||||
tls {
|
||||
cert_file: "{{.ConfigDir}}/server.pem"
|
||||
key_file: "{{.ConfigDir}}/server.key"
|
||||
ca_file: "{{.ConfigDir}}/ca.pem"
|
||||
verify: true
|
||||
}
|
||||
|
||||
routes = [
|
||||
{{- range .ClusterRoutes }}
|
||||
"nats-route://{{$.NatsToken}}@{{ . }}",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Predastore S3-compatible storage configuration file in TOML format.
|
||||
# Generated for multi-node cluster
|
||||
version = "2"
|
||||
version = "3"
|
||||
region = "{{.Region}}"
|
||||
|
||||
host = "0.0.0.0"
|
||||
@@ -65,5 +65,6 @@ burst = 200
|
||||
[iam]
|
||||
nats_url = "nats://{{.BindIP}}:4222"
|
||||
nats_token = "{{.NatsToken}}"
|
||||
nats_ca_cert = "{{.ConfigDir}}/ca.pem"
|
||||
master_key_path = "{{.ConfigDir}}/master.key"
|
||||
access_keys_bucket = "spinifex-iam-access-keys"
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Predastore S3-compatible storage configuration file in TOML format.
|
||||
# MULTI NODE used for testing
|
||||
version = "2"
|
||||
version = "3"
|
||||
region = "{{.Region}}"
|
||||
|
||||
# Optionally configure port and host, or specify via daemon CLI options
|
||||
@@ -99,5 +99,6 @@ burst = 200
|
||||
[iam]
|
||||
nats_url = "nats://{{if eq .BindIP "0.0.0.0"}}localhost{{else}}{{.BindIP}}{{end}}:4222"
|
||||
nats_token = "{{.NatsToken}}"
|
||||
nats_ca_cert = "{{.ConfigDir}}/ca.pem"
|
||||
master_key_path = "{{.ConfigDir}}/master.key"
|
||||
access_keys_bucket = "spinifex-iam-access-keys"
|
||||
@@ -1,5 +1,5 @@
|
||||
# Domain configuration file in TOML format.
|
||||
version = "1.0"
|
||||
version = "2"
|
||||
epoch = 1
|
||||
node = "{{.Node}}"
|
||||
{{- if .ExternalMode}}
|
||||
@@ -69,6 +69,7 @@ config = "awsgw/awsgw.toml"
|
||||
|
||||
[nodes.{{.Node}}.nats]
|
||||
host = "{{.BindIP}}:4222"
|
||||
cacert = "{{.ConfigDir}}/ca.pem"
|
||||
|
||||
[nodes.{{.Node}}.nats.acl]
|
||||
token = "{{.NatsToken}}"
|
||||
|
||||
@@ -13,7 +13,12 @@ if [[ ! -s "$PROFILE" ]]; then
|
||||
exit 1
|
||||
fi
|
||||
|
||||
TOTAL=$(go tool cover -func="$PROFILE" | tail -1 | awk '{print $NF}' | tr -d '%')
|
||||
# Filter out packages excluded from coverage requirements (e.g. migrations)
|
||||
FILTERED=$(mktemp)
|
||||
trap 'rm -f "$FILTERED"' EXIT
|
||||
awk 'NR==1 || !/\/migrate\//' "$PROFILE" > "$FILTERED"
|
||||
|
||||
TOTAL=$(go tool cover -func="$FILTERED" | tail -1 | awk '{print $NF}' | tr -d '%')
|
||||
|
||||
if [[ -z "$TOTAL" ]]; then
|
||||
echo "ERROR: No coverage data generated — tests may have failed to compile"
|
||||
|
||||
@@ -66,7 +66,7 @@ trap 'rm -rf "$TMPDIR"' EXIT
|
||||
# Only non-test .go files that were added or modified.
|
||||
# -M enables rename detection so file splits/moves only count truly changed lines.
|
||||
# -C enables copy detection for the same reason (file split = partial copy).
|
||||
git diff "${BASE_REF}" HEAD --unified=0 -M -C --diff-filter=AM -- '*.go' ':!*_test.go' | awk '
|
||||
git diff "${BASE_REF}" HEAD --unified=0 -M -C --diff-filter=AM -- '*.go' ':!*_test.go' ':!spinifex/migrate/*' | awk '
|
||||
/^\+\+\+ / {
|
||||
file = substr($2, 3)
|
||||
next
|
||||
|
||||
+26
-12
@@ -271,19 +271,33 @@ if has_service "nats"; then
|
||||
# Wait for NATS JetStream to be ready before Predastore starts.
|
||||
# Predastore lazily opens IAM KV buckets — it starts with config-only auth
|
||||
# and activates IAM auth once the spinifex daemon creates the KV buckets.
|
||||
echo "🔍 Waiting for NATS JetStream..."
|
||||
NATS_JS_READY=false
|
||||
for i in $(seq 1 15); do
|
||||
# Send a NATS INFO probe — JetStream-enabled servers include "jetstream" in the INFO response
|
||||
if echo "" | nc -w 1 "$NATS_CHECK_HOST" 4222 2>/dev/null | grep -q "jetstream"; then
|
||||
echo " ✅ NATS JetStream is ready"
|
||||
NATS_JS_READY=true
|
||||
break
|
||||
# Wait for NATS JetStream to be ready before Predastore starts.
|
||||
# Predastore lazily opens IAM KV buckets — it starts with config-only auth
|
||||
# and activates IAM auth once the spinifex daemon creates the KV buckets.
|
||||
# Use the monitoring HTTP endpoint (plaintext nc probe doesn't work with TLS).
|
||||
# Monitoring is only on the primary node; secondary nodes skip this check.
|
||||
NATS_HAS_MONITORING=false
|
||||
if [ -f "$CONFIG_DIR/nats/nats.conf" ] && grep -q '^http:' "$CONFIG_DIR/nats/nats.conf"; then
|
||||
NATS_HAS_MONITORING=true
|
||||
fi
|
||||
|
||||
if [ "$NATS_HAS_MONITORING" = "true" ]; then
|
||||
echo "🔍 Waiting for NATS JetStream..."
|
||||
NATS_JS_READY=false
|
||||
for i in $(seq 1 15); do
|
||||
if curl -sf "http://127.0.0.1:8222/jsz" > /dev/null 2>&1; then
|
||||
echo " ✅ NATS JetStream is ready"
|
||||
NATS_JS_READY=true
|
||||
break
|
||||
fi
|
||||
sleep 1
|
||||
done
|
||||
if [ "$NATS_JS_READY" = "false" ]; then
|
||||
echo " ⚠️ NATS JetStream not confirmed ready — Predastore may fall back to config-only auth"
|
||||
fi
|
||||
sleep 1
|
||||
done
|
||||
if [ "$NATS_JS_READY" = "false" ]; then
|
||||
echo " ⚠️ NATS JetStream not confirmed ready — Predastore may fall back to config-only auth"
|
||||
else
|
||||
echo "🔍 NATS monitoring not configured — skipping JetStream readiness check"
|
||||
sleep 2
|
||||
fi
|
||||
else
|
||||
echo "1️⃣ Skipping NATS (not a local service)"
|
||||
|
||||
@@ -678,7 +678,7 @@ func GenerateSignedCertWithDNS(certPath, keyPath, caCertPath, caKeyPath string,
|
||||
NotBefore: notBefore,
|
||||
NotAfter: notAfter,
|
||||
KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
|
||||
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
|
||||
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
|
||||
BasicConstraintsValid: true,
|
||||
DNSNames: dnsNames,
|
||||
IPAddresses: ipAddresses,
|
||||
@@ -740,7 +740,7 @@ func GenerateSelfSignedCert(certPath, keyPath string) error {
|
||||
NotBefore: notBefore,
|
||||
NotAfter: notAfter,
|
||||
KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
|
||||
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
|
||||
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth},
|
||||
BasicConstraintsValid: true,
|
||||
DNSNames: []string{"localhost"},
|
||||
IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")},
|
||||
|
||||
@@ -117,9 +117,10 @@ type DaemonConfig struct {
|
||||
|
||||
// NATSConfig holds the NATS configuration
|
||||
type NATSConfig struct {
|
||||
Host string `json:"Host" mapstructure:"host"`
|
||||
ACL NATSACL `json:"ACL" mapstructure:"acl"`
|
||||
Sub NATSSub `json:"Sub" mapstructure:"sub"`
|
||||
Host string `json:"Host" mapstructure:"host"`
|
||||
CACert string `json:"CACert" mapstructure:"cacert"`
|
||||
ACL NATSACL `json:"ACL" mapstructure:"acl"`
|
||||
Sub NATSSub `json:"Sub" mapstructure:"sub"`
|
||||
}
|
||||
|
||||
// NATSACL holds the NATS ACL configuration
|
||||
|
||||
@@ -851,7 +851,7 @@ func (d *Daemon) connectNATS() error {
|
||||
start := time.Now()
|
||||
|
||||
for {
|
||||
nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token)
|
||||
nc, err := utils.ConnectNATS(d.config.NATS.Host, d.config.NATS.ACL.Token, d.config.NATS.CACert)
|
||||
if err == nil {
|
||||
d.natsConn = nc
|
||||
if time.Since(start) > time.Second {
|
||||
@@ -860,6 +860,11 @@ func (d *Daemon) connectNATS() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// TLS configuration errors are permanent — retrying will not help.
|
||||
if strings.Contains(err.Error(), "CA cert") {
|
||||
return fmt.Errorf("NATS TLS configuration error: %w", err)
|
||||
}
|
||||
|
||||
elapsed := time.Since(start)
|
||||
if elapsed >= maxWait {
|
||||
return fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err)
|
||||
|
||||
@@ -0,0 +1,148 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func init() {
|
||||
// nats.conf: version 2 → 3 — add TLS blocks for client and cluster connections.
|
||||
DefaultRegistry.RegisterConfig("nats.conf", ConfigMigration{
|
||||
FromVersion: 2,
|
||||
ToVersion: 3,
|
||||
Description: "Add TLS for client connections and mutual TLS for cluster routes",
|
||||
Run: func(ctx ConfigContext) error {
|
||||
path := filepath.Join(ctx.ConfigDir, "nats", "nats.conf")
|
||||
content, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
text := string(content)
|
||||
|
||||
// Skip if TLS is already configured.
|
||||
if strings.Contains(text, "tls {") {
|
||||
ctx.Logger.Info("nats.conf: TLS block already present, skipping")
|
||||
return nil
|
||||
}
|
||||
|
||||
certFile := fmt.Sprintf("%s/server.pem", ctx.ConfigDir)
|
||||
keyFile := fmt.Sprintf("%s/server.key", ctx.ConfigDir)
|
||||
caFile := fmt.Sprintf("%s/ca.pem", ctx.ConfigDir)
|
||||
|
||||
// 1. Inject client TLS block after the top-level listen line (port 4222 only,
|
||||
// not the cluster listen on 4248).
|
||||
clientTLS := fmt.Sprintf("\n\ntls {\n cert_file: \"%s\"\n key_file: \"%s\"\n ca_file: \"%s\"\n}", certFile, keyFile, caFile)
|
||||
|
||||
listenRe := regexp.MustCompile(`(?m)^listen:\s*\S+:4222\s*$`)
|
||||
if !listenRe.MatchString(text) {
|
||||
return fmt.Errorf("nats.conf: client listen line (port 4222) not found")
|
||||
}
|
||||
loc := listenRe.FindStringIndex(text)
|
||||
text = text[:loc[1]] + clientTLS + text[loc[1]:]
|
||||
|
||||
// 2. Inject cluster TLS block inside the cluster {} block, after the listen line.
|
||||
clusterTLS := fmt.Sprintf("\n\n tls {\n cert_file: \"%s\"\n key_file: \"%s\"\n ca_file: \"%s\"\n verify: true\n }", certFile, keyFile, caFile)
|
||||
|
||||
// Match the cluster listen line and insert TLS block after it.
|
||||
clusterListenRe := regexp.MustCompile(`(?m)^(\s*listen:\s*\S+:4248)\s*$`)
|
||||
if clusterListenRe.MatchString(text) {
|
||||
text = clusterListenRe.ReplaceAllStringFunc(text, func(match string) string {
|
||||
return strings.TrimRight(match, "\n") + clusterTLS
|
||||
})
|
||||
}
|
||||
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, []byte(text), info.Mode())
|
||||
},
|
||||
})
|
||||
|
||||
// spinifex.toml: version 1 → 2 — add cacert under [nodes.*.nats].
|
||||
DefaultRegistry.RegisterConfigTarget("spinifex.toml", "spinifex.toml", &TOMLVersionReader{})
|
||||
|
||||
DefaultRegistry.RegisterConfig("spinifex.toml", ConfigMigration{
|
||||
FromVersion: 1,
|
||||
ToVersion: 2,
|
||||
Description: "Add NATS CA cert path for TLS",
|
||||
Run: func(ctx ConfigContext) error {
|
||||
path := filepath.Join(ctx.ConfigDir, "spinifex.toml")
|
||||
content, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
text := string(content)
|
||||
|
||||
// Skip if cacert is already present.
|
||||
if strings.Contains(text, "cacert") {
|
||||
ctx.Logger.Info("spinifex.toml: cacert already present, skipping")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Rewrite 0.0.0.0 → 127.0.0.1 for the NATS host. 0.0.0.0 is a valid
|
||||
// bind address but not a valid TLS connect address (certs have real IPs).
|
||||
text = strings.ReplaceAll(text, `host = "0.0.0.0:4222"`, `host = "127.0.0.1:4222"`)
|
||||
|
||||
// Insert cacert after each host line under [nodes.*.nats].
|
||||
// Pattern: host = "x.x.x.x:4222" followed by a newline.
|
||||
caPath := fmt.Sprintf("%s/ca.pem", ctx.ConfigDir)
|
||||
natsHostRe := regexp.MustCompile(`(?m)^(host\s*=\s*"[^"]+:4222")\s*$`)
|
||||
if !natsHostRe.MatchString(text) {
|
||||
return fmt.Errorf("spinifex.toml: NATS host line (host = \"...:4222\") not found, cannot insert cacert")
|
||||
}
|
||||
text = natsHostRe.ReplaceAllStringFunc(text, func(match string) string {
|
||||
return strings.TrimRight(match, "\n") + fmt.Sprintf("\ncacert = \"%s\"", caPath)
|
||||
})
|
||||
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, []byte(text), info.Mode())
|
||||
},
|
||||
})
|
||||
|
||||
// predastore.toml: version 2 → 3 — add nats_ca_cert under [iam].
|
||||
DefaultRegistry.RegisterConfig("predastore.toml", ConfigMigration{
|
||||
FromVersion: 2,
|
||||
ToVersion: 3,
|
||||
Description: "Add NATS CA cert path for TLS",
|
||||
Run: func(ctx ConfigContext) error {
|
||||
path := filepath.Join(ctx.ConfigDir, "predastore", "predastore.toml")
|
||||
content, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
text := string(content)
|
||||
|
||||
// Skip if nats_ca_cert is already present.
|
||||
if strings.Contains(text, "nats_ca_cert") {
|
||||
ctx.Logger.Info("predastore.toml: nats_ca_cert already present, skipping")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Insert nats_ca_cert after the nats_token line.
|
||||
caPath := fmt.Sprintf("%s/ca.pem", ctx.ConfigDir)
|
||||
natsTokenRe := regexp.MustCompile(`(?m)^(nats_token\s*=\s*"[^"]*")\s*$`)
|
||||
if !natsTokenRe.MatchString(text) {
|
||||
return fmt.Errorf("predastore.toml: nats_token line not found, cannot insert nats_ca_cert")
|
||||
}
|
||||
text = natsTokenRe.ReplaceAllStringFunc(text, func(match string) string {
|
||||
return strings.TrimRight(match, "\n") + fmt.Sprintf("\nnats_ca_cert = \"%s\"", caPath)
|
||||
})
|
||||
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, []byte(text), info.Mode())
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -102,7 +102,7 @@ func launchService(config *config.ClusterConfig) error {
|
||||
|
||||
// Connect to NATS for service communication. On concurrent startup the
|
||||
// local NATS server may not be listening yet, so retry with backoff.
|
||||
natsConn, err := connectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token)
|
||||
natsConn, err := connectNATS(nodeConfig.NATS.Host, nodeConfig.NATS.ACL.Token, nodeConfig.NATS.CACert)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -205,13 +205,13 @@ func launchService(config *config.ClusterConfig) error {
|
||||
|
||||
// connectNATS establishes a connection to NATS with retry/backoff. On concurrent
|
||||
// startup the local NATS server may not be listening yet.
|
||||
func connectNATS(host, token string) (*nats.Conn, error) {
|
||||
func connectNATS(host, token, caCertPath string) (*nats.Conn, error) {
|
||||
const maxWait = 5 * time.Minute
|
||||
retryDelay := 500 * time.Millisecond
|
||||
start := time.Now()
|
||||
|
||||
for {
|
||||
nc, err := utils.ConnectNATS(host, token)
|
||||
nc, err := utils.ConnectNATS(host, token, caCertPath)
|
||||
if err == nil {
|
||||
if time.Since(start) > time.Second {
|
||||
slog.Info("NATS connection established", "elapsed", time.Since(start).Round(time.Second))
|
||||
@@ -219,6 +219,11 @@ func connectNATS(host, token string) (*nats.Conn, error) {
|
||||
return nc, nil
|
||||
}
|
||||
|
||||
// TLS configuration errors are permanent — retrying will not help.
|
||||
if strings.Contains(err.Error(), "CA cert") {
|
||||
return nil, fmt.Errorf("NATS TLS configuration error: %w", err)
|
||||
}
|
||||
|
||||
elapsed := time.Since(start)
|
||||
if elapsed >= maxWait {
|
||||
return nil, fmt.Errorf("NATS connect failed after %s: %w", elapsed.Round(time.Second), err)
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"text/template"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/admin"
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/assert"
|
||||
@@ -34,6 +35,14 @@ func TestRenderedConfig_EnforcesAuth(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
confPath := filepath.Join(tmpDir, "nats.conf")
|
||||
|
||||
// Generate test CA and server certs for TLS.
|
||||
caCertPath := filepath.Join(tmpDir, "ca.pem")
|
||||
caKeyPath := filepath.Join(tmpDir, "ca.key")
|
||||
require.NoError(t, admin.GenerateCACert(caCertPath, caKeyPath))
|
||||
serverCertPath := filepath.Join(tmpDir, "server.pem")
|
||||
serverKeyPath := filepath.Join(tmpDir, "server.key")
|
||||
require.NoError(t, admin.GenerateSignedCert(serverCertPath, serverKeyPath, caCertPath, caKeyPath, "127.0.0.1"))
|
||||
|
||||
// Read and render the production template.
|
||||
raw, err := os.ReadFile(templatePath(t))
|
||||
require.NoError(t, err)
|
||||
@@ -50,6 +59,7 @@ func TestRenderedConfig_EnforcesAuth(t *testing.T) {
|
||||
"NatsToken": token,
|
||||
"DataDir": tmpDir,
|
||||
"LogDir": tmpDir,
|
||||
"ConfigDir": tmpDir,
|
||||
})
|
||||
f.Close()
|
||||
require.NoError(t, err)
|
||||
@@ -70,22 +80,30 @@ func TestRenderedConfig_EnforcesAuth(t *testing.T) {
|
||||
require.True(t, ns.ReadyForConnections(5*time.Second))
|
||||
t.Cleanup(func() { ns.Shutdown() })
|
||||
|
||||
// Build TLS option using the test CA cert for client connections.
|
||||
tlsOpt := nats.RootCAs(caCertPath)
|
||||
|
||||
t.Run("no token rejected", func(t *testing.T) {
|
||||
_, err := nats.Connect(ns.ClientURL(), nats.MaxReconnects(0))
|
||||
_, err := nats.Connect(ns.ClientURL(), tlsOpt, nats.MaxReconnects(0))
|
||||
assert.Error(t, err, "unauthenticated connection should be rejected")
|
||||
})
|
||||
|
||||
t.Run("wrong token rejected", func(t *testing.T) {
|
||||
_, err := nats.Connect(ns.ClientURL(), nats.Token("wrong-token"), nats.MaxReconnects(0))
|
||||
_, err := nats.Connect(ns.ClientURL(), tlsOpt, nats.Token("wrong-token"), nats.MaxReconnects(0))
|
||||
assert.Error(t, err, "wrong token should be rejected")
|
||||
})
|
||||
|
||||
t.Run("correct token accepted", func(t *testing.T) {
|
||||
nc, err := nats.Connect(ns.ClientURL(), nats.Token(token), nats.MaxReconnects(0))
|
||||
nc, err := nats.Connect(ns.ClientURL(), tlsOpt, nats.Token(token), nats.MaxReconnects(0))
|
||||
require.NoError(t, err, "correct token should be accepted")
|
||||
defer nc.Close()
|
||||
assert.True(t, nc.IsConnected())
|
||||
})
|
||||
|
||||
t.Run("plaintext rejected", func(t *testing.T) {
|
||||
_, err := nats.Connect(ns.ClientURL(), nats.Token(token), nats.MaxReconnects(0))
|
||||
assert.Error(t, err, "plaintext connection should be rejected when TLS is enabled")
|
||||
})
|
||||
}
|
||||
|
||||
// TestRenderedConfig_HasMigrationVersionMarker ensures the nats.conf template
|
||||
@@ -98,6 +116,6 @@ func TestRenderedConfig_HasMigrationVersionMarker(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
|
||||
firstLine := strings.SplitN(string(raw), "\n", 2)[0]
|
||||
assert.Equal(t, "# spinifex-config-version: 2", firstLine,
|
||||
assert.Equal(t, "# spinifex-config-version: 3", firstLine,
|
||||
"nats.conf template must start with the current migration version marker")
|
||||
}
|
||||
|
||||
@@ -39,6 +39,7 @@ type Config struct {
|
||||
Debug bool
|
||||
NatsHost string
|
||||
NatsToken string
|
||||
NatsCACert string
|
||||
MountedVolumes []MountedVolume
|
||||
S3Host string
|
||||
Bucket string
|
||||
@@ -169,7 +170,7 @@ func (svc *Service) Reload() (err error) {
|
||||
|
||||
func launchService(cfg *Config) (err error) {
|
||||
// Connect to NATS
|
||||
nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken)
|
||||
nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken, cfg.NatsCACert)
|
||||
if err != nil {
|
||||
slog.Error("Failed to connect to NATS", "err", err)
|
||||
return err
|
||||
|
||||
@@ -44,6 +44,8 @@ type Config struct {
|
||||
NatsHost string
|
||||
// NatsToken is the NATS authentication token.
|
||||
NatsToken string
|
||||
// NatsCACert is the path to the CA certificate for NATS TLS.
|
||||
NatsCACert string
|
||||
// OVNNBAddr is the OVN Northbound DB address (e.g., "tcp:127.0.0.1:6641").
|
||||
OVNNBAddr string
|
||||
// OVNSBAddr is the OVN Southbound DB address (e.g., "tcp:127.0.0.1:6642"), used for monitoring.
|
||||
@@ -272,7 +274,7 @@ func launchService(cfg *Config) error {
|
||||
slog.Info("OVN preflight passed (br-int exists, ovn-controller running)")
|
||||
|
||||
// Connect to NATS
|
||||
nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken)
|
||||
nc, err := utils.ConnectNATS(cfg.NatsHost, cfg.NatsToken, cfg.NatsCACert)
|
||||
if err != nil {
|
||||
slog.Error("Failed to connect to NATS", "err", err)
|
||||
return err
|
||||
|
||||
+19
-1
@@ -1,10 +1,13 @@
|
||||
package utils
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -12,7 +15,8 @@ import (
|
||||
|
||||
// ConnectNATS establishes a connection to a NATS server with standard reconnect
|
||||
// handling and logging. If token is non-empty, token authentication is used.
|
||||
func ConnectNATS(host, token string) (*nats.Conn, error) {
|
||||
// If caCertPath is non-empty, TLS is enabled using the given CA certificate.
|
||||
func ConnectNATS(host, token, caCertPath string) (*nats.Conn, error) {
|
||||
opts := []nats.Option{
|
||||
nats.ReconnectWait(time.Second),
|
||||
nats.MaxReconnects(-1),
|
||||
@@ -28,6 +32,20 @@ func ConnectNATS(host, token string) (*nats.Conn, error) {
|
||||
opts = append(opts, nats.Token(token))
|
||||
}
|
||||
|
||||
if caCertPath != "" {
|
||||
caCert, err := os.ReadFile(caCertPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read CA cert %s: %w", caCertPath, err)
|
||||
}
|
||||
pool := x509.NewCertPool()
|
||||
if !pool.AppendCertsFromPEM(caCert) {
|
||||
return nil, fmt.Errorf("failed to parse CA cert from %s", caCertPath)
|
||||
}
|
||||
opts = append(opts, nats.Secure(&tls.Config{
|
||||
RootCAs: pool,
|
||||
}))
|
||||
}
|
||||
|
||||
nc, err := nats.Connect(host, opts...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("NATS connect failed: %w", err)
|
||||
|
||||
+154
-4
@@ -1,7 +1,18 @@
|
||||
package utils
|
||||
|
||||
import (
|
||||
"crypto/ecdsa"
|
||||
"crypto/elliptic"
|
||||
"crypto/rand"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"crypto/x509/pkix"
|
||||
"encoding/json"
|
||||
"encoding/pem"
|
||||
"math/big"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -21,7 +32,7 @@ func startTestNATSServer(t *testing.T) *server.Server {
|
||||
func TestConnectNATS_Success(t *testing.T) {
|
||||
ns := startTestNATSServer(t)
|
||||
|
||||
nc, err := ConnectNATS(ns.ClientURL(), "")
|
||||
nc, err := ConnectNATS(ns.ClientURL(), "", "")
|
||||
require.NoError(t, err)
|
||||
defer nc.Close()
|
||||
|
||||
@@ -44,7 +55,7 @@ func TestConnectNATS_WithToken(t *testing.T) {
|
||||
t.Cleanup(func() { ns.Shutdown() })
|
||||
|
||||
// With correct token — should succeed
|
||||
nc, err := ConnectNATS(ns.ClientURL(), "test-token-123")
|
||||
nc, err := ConnectNATS(ns.ClientURL(), "test-token-123", "")
|
||||
require.NoError(t, err)
|
||||
defer nc.Close()
|
||||
assert.True(t, nc.IsConnected())
|
||||
@@ -65,17 +76,156 @@ func TestConnectNATS_WrongToken(t *testing.T) {
|
||||
require.True(t, ns.ReadyForConnections(5*time.Second))
|
||||
t.Cleanup(func() { ns.Shutdown() })
|
||||
|
||||
_, err = ConnectNATS(ns.ClientURL(), "wrong-token")
|
||||
_, err = ConnectNATS(ns.ClientURL(), "wrong-token", "")
|
||||
assert.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "NATS connect failed")
|
||||
}
|
||||
|
||||
func TestConnectNATS_BadAddress(t *testing.T) {
|
||||
_, err := ConnectNATS("nats://127.0.0.1:1", "")
|
||||
_, err := ConnectNATS("nats://127.0.0.1:1", "", "")
|
||||
assert.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "NATS connect failed")
|
||||
}
|
||||
|
||||
func TestConnectNATS_MissingCACert(t *testing.T) {
|
||||
_, err := ConnectNATS("nats://127.0.0.1:4222", "", "/nonexistent/ca.pem")
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "read CA cert")
|
||||
assert.Contains(t, err.Error(), "/nonexistent/ca.pem")
|
||||
}
|
||||
|
||||
func TestConnectNATS_MalformedCACert(t *testing.T) {
|
||||
tmp := t.TempDir()
|
||||
badCert := filepath.Join(tmp, "bad-ca.pem")
|
||||
require.NoError(t, os.WriteFile(badCert, []byte("not a PEM certificate"), 0o644))
|
||||
|
||||
_, err := ConnectNATS("nats://127.0.0.1:4222", "", badCert)
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "failed to parse CA cert")
|
||||
}
|
||||
|
||||
// generateTestCA creates an ephemeral CA cert+key and writes PEM files to dir.
|
||||
func generateTestCA(t *testing.T, dir, name string) (certPath, keyPath string) {
|
||||
t.Helper()
|
||||
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
require.NoError(t, err)
|
||||
|
||||
tmpl := &x509.Certificate{
|
||||
SerialNumber: big.NewInt(1),
|
||||
Subject: pkix.Name{CommonName: name},
|
||||
NotBefore: time.Now().Add(-time.Hour),
|
||||
NotAfter: time.Now().Add(time.Hour),
|
||||
IsCA: true,
|
||||
BasicConstraintsValid: true,
|
||||
KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign,
|
||||
}
|
||||
der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key)
|
||||
require.NoError(t, err)
|
||||
|
||||
certPath = filepath.Join(dir, name+".pem")
|
||||
keyPath = filepath.Join(dir, name+".key")
|
||||
require.NoError(t, os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644))
|
||||
keyDER, err := x509.MarshalECPrivateKey(key)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600))
|
||||
return certPath, keyPath
|
||||
}
|
||||
|
||||
// generateTestServerCert creates a server cert signed by the given CA.
|
||||
func generateTestServerCert(t *testing.T, dir, caCertPath, caKeyPath string) (certPath, keyPath string) {
|
||||
t.Helper()
|
||||
caCertPEM, err := os.ReadFile(caCertPath)
|
||||
require.NoError(t, err)
|
||||
block, _ := pem.Decode(caCertPEM)
|
||||
caCert, err := x509.ParseCertificate(block.Bytes)
|
||||
require.NoError(t, err)
|
||||
|
||||
caKeyPEM, err := os.ReadFile(caKeyPath)
|
||||
require.NoError(t, err)
|
||||
keyBlock, _ := pem.Decode(caKeyPEM)
|
||||
caKey, err := x509.ParseECPrivateKey(keyBlock.Bytes)
|
||||
require.NoError(t, err)
|
||||
|
||||
serverKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
require.NoError(t, err)
|
||||
|
||||
tmpl := &x509.Certificate{
|
||||
SerialNumber: big.NewInt(2),
|
||||
Subject: pkix.Name{CommonName: "localhost"},
|
||||
NotBefore: time.Now().Add(-time.Hour),
|
||||
NotAfter: time.Now().Add(time.Hour),
|
||||
KeyUsage: x509.KeyUsageDigitalSignature,
|
||||
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
|
||||
IPAddresses: []net.IP{net.ParseIP("127.0.0.1")},
|
||||
}
|
||||
der, err := x509.CreateCertificate(rand.Reader, tmpl, caCert, &serverKey.PublicKey, caKey)
|
||||
require.NoError(t, err)
|
||||
|
||||
certPath = filepath.Join(dir, "server.pem")
|
||||
keyPath = filepath.Join(dir, "server.key")
|
||||
require.NoError(t, os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644))
|
||||
keyDER, err := x509.MarshalECPrivateKey(serverKey)
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600))
|
||||
return certPath, keyPath
|
||||
}
|
||||
|
||||
// startTLSNATSServer starts a NATS server with TLS using the given cert files.
|
||||
func startTLSNATSServer(t *testing.T, serverCertPath, serverKeyPath, caCertPath string) *server.Server {
|
||||
t.Helper()
|
||||
cert, err := tls.LoadX509KeyPair(serverCertPath, serverKeyPath)
|
||||
require.NoError(t, err)
|
||||
caPEM, err := os.ReadFile(caCertPath)
|
||||
require.NoError(t, err)
|
||||
pool := x509.NewCertPool()
|
||||
require.True(t, pool.AppendCertsFromPEM(caPEM))
|
||||
|
||||
tlsCfg := &tls.Config{
|
||||
Certificates: []tls.Certificate{cert},
|
||||
ClientCAs: pool,
|
||||
}
|
||||
|
||||
opts := &server.Options{
|
||||
Host: "127.0.0.1",
|
||||
Port: -1,
|
||||
NoLog: true,
|
||||
NoSigs: true,
|
||||
TLSConfig: tlsCfg,
|
||||
}
|
||||
|
||||
ns, err := server.NewServer(opts)
|
||||
require.NoError(t, err)
|
||||
go ns.Start()
|
||||
require.True(t, ns.ReadyForConnections(5*time.Second))
|
||||
t.Cleanup(func() { ns.Shutdown() })
|
||||
return ns
|
||||
}
|
||||
|
||||
func TestConnectNATS_TLSSuccess(t *testing.T) {
|
||||
tmp := t.TempDir()
|
||||
caCertPath, caKeyPath := generateTestCA(t, tmp, "ca")
|
||||
serverCertPath, serverKeyPath := generateTestServerCert(t, tmp, caCertPath, caKeyPath)
|
||||
|
||||
ns := startTLSNATSServer(t, serverCertPath, serverKeyPath, caCertPath)
|
||||
|
||||
nc, err := ConnectNATS(ns.ClientURL(), "", caCertPath)
|
||||
require.NoError(t, err)
|
||||
defer nc.Close()
|
||||
assert.True(t, nc.IsConnected())
|
||||
}
|
||||
|
||||
func TestConnectNATS_WrongCA(t *testing.T) {
|
||||
tmp := t.TempDir()
|
||||
caCertPath, caKeyPath := generateTestCA(t, tmp, "ca")
|
||||
serverCertPath, serverKeyPath := generateTestServerCert(t, tmp, caCertPath, caKeyPath)
|
||||
wrongCACertPath, _ := generateTestCA(t, tmp, "wrong-ca")
|
||||
|
||||
ns := startTLSNATSServer(t, serverCertPath, serverKeyPath, caCertPath)
|
||||
|
||||
_, err := ConnectNATS(ns.ClientURL(), "", wrongCACertPath)
|
||||
assert.Error(t, err, "connection with wrong CA should fail")
|
||||
}
|
||||
|
||||
func TestNATSRequest_Success(t *testing.T) {
|
||||
ns := startTestNATSServer(t)
|
||||
|
||||
|
||||
@@ -779,10 +779,9 @@ init_leader_node() {
|
||||
# shellcheck disable=SC2086
|
||||
./bin/spx admin init \
|
||||
--node node1 \
|
||||
--nodes 3 \
|
||||
--bind "${NODE1_IP}" \
|
||||
--cluster-bind "${NODE1_IP}" \
|
||||
--cluster-routes "${NODE1_IP}:${NATS_CLUSTER_PORT}" \
|
||||
--predastore-nodes "${NODE1_IP},${NODE2_IP},${NODE3_IP}" \
|
||||
--port ${CLUSTER_PORT} \
|
||||
--region ap-southeast-2 \
|
||||
--az ap-southeast-2a \
|
||||
@@ -827,12 +826,10 @@ join_follower_node() {
|
||||
# Remove old node directory
|
||||
rm -rf "$data_dir/"
|
||||
|
||||
# Route to node1 (seed node) - other nodes discovered via NATS gossip
|
||||
./bin/spx admin join \
|
||||
--node "node$node_num" \
|
||||
--bind "$node_ip" \
|
||||
--cluster-bind "$node_ip" \
|
||||
--cluster-routes "${NODE1_IP}:${NATS_CLUSTER_PORT}" \
|
||||
--host "${NODE1_IP}:${CLUSTER_PORT}" \
|
||||
--token "$JOIN_TOKEN" \
|
||||
--data-dir "$data_dir/" \
|
||||
|
||||
Reference in New Issue
Block a user