diff --git a/cmd/spinifex/cmd/get.go b/cmd/spinifex/cmd/get.go index cb3c72704..d6eef9022 100644 --- a/cmd/spinifex/cmd/get.go +++ b/cmd/spinifex/cmd/get.go @@ -138,6 +138,20 @@ func formatUptime(seconds int64) string { return fmt.Sprintf("%dm", mins) } +func formatRoles(resp types.NodeStatusResponse) string { + var roles []string + if resp.NATSRole != "" { + roles = append(roles, "nats:"+resp.NATSRole) + } + if resp.PredastoreRole != "" { + roles = append(roles, "predastore:"+resp.PredastoreRole) + } + if len(roles) == 0 { + return "-" + } + return strings.Join(roles, ",") +} + func formatMemGB(gb float64) string { if gb >= 1.0 { return fmt.Sprintf("%.1fGi", gb) @@ -172,7 +186,7 @@ func runGetNodes(cmd *cobra.Command, args []string) { // Build table: union of config-known nodes + NATS responders tableData := pterm.TableData{ - {"NAME", "STATUS", "IP", "REGION", "AZ", "UPTIME", "VMs", "SERVICES"}, + {"NAME", "STATUS", "ROLES", "IP", "REGION", "AZ", "UPTIME", "VMs", "SERVICES"}, } // Collect all node names: config + responded (union) @@ -195,6 +209,7 @@ func runGetNodes(cmd *cobra.Command, args []string) { tableData = append(tableData, []string{ resp.Node, resp.Status, + formatRoles(resp), resp.Host, resp.Region, resp.AZ, @@ -206,6 +221,7 @@ func runGetNodes(cmd *cobra.Command, args []string) { tableData = append(tableData, []string{ name, "NotReady", + "-", nodeCfg.Host, nodeCfg.Region, nodeCfg.AZ, diff --git a/cmd/spinifex/cmd/get_test.go b/cmd/spinifex/cmd/get_test.go new file mode 100644 index 000000000..90c049a51 --- /dev/null +++ b/cmd/spinifex/cmd/get_test.go @@ -0,0 +1,49 @@ +package cmd + +import ( + "testing" + + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/stretchr/testify/assert" +) + +func TestFormatRoles(t *testing.T) { + tests := []struct { + name string + resp types.NodeStatusResponse + want string + }{ + { + name: "no roles", + resp: types.NodeStatusResponse{}, + want: "-", + }, + { + name: "nats leader only", + resp: types.NodeStatusResponse{NATSRole: "leader"}, + want: "nats:leader", + }, + { + name: "predastore follower only", + resp: types.NodeStatusResponse{PredastoreRole: "follower"}, + want: "predastore:follower", + }, + { + name: "both roles", + resp: types.NodeStatusResponse{NATSRole: "leader", PredastoreRole: "follower"}, + want: "nats:leader,predastore:follower", + }, + { + name: "both leaders", + resp: types.NodeStatusResponse{NATSRole: "leader", PredastoreRole: "leader"}, + want: "nats:leader,predastore:leader", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := formatRoles(tt.resp) + assert.Equal(t, tt.want, got) + }) + } +} diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index dc3fc382d..424902082 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -1,10 +1,13 @@ package daemon import ( + "crypto/tls" "encoding/json" "fmt" "log/slog" "net" + "net/http" + "sync" "time" "github.com/mulgadc/spinifex/spinifex/awserrors" @@ -175,9 +178,109 @@ func (d *Daemon) handleNodeStatus(msg *nats.Msg) { InstanceTypes: caps, } + // Query service roles concurrently to halve worst-case latency (500ms vs 1s). + var wg sync.WaitGroup + wg.Add(2) + go func() { defer wg.Done(); resp.NATSRole = d.queryNATSRole() }() + go func() { defer wg.Done(); resp.PredastoreRole = d.queryPredastoreRole() }() + wg.Wait() + respondWithJSON(msg, resp) } +const ( + roleLeader = "leader" + roleFollower = "follower" + + natsMonitorPort = 8222 + predastoreDBPort = 6660 +) + +// queryNATSRole queries the local NATS monitoring endpoint to determine this +// node's JetStream meta-leader status. Returns "leader", "follower", or "". +func (d *Daemon) queryNATSRole() string { + if !d.config.HasService("nats") { + return "" + } + url := fmt.Sprintf("http://%s:%d/varz", d.daemonIP(), natsMonitorPort) + return fetchNATSRole(url, roleHTTPClient) +} + +// queryPredastoreRole queries the local Predastore DB status endpoint to +// determine this node's Raft role. Returns "leader", "follower", or "". +func (d *Daemon) queryPredastoreRole() string { + if !d.config.HasService("predastore") { + return "" + } + url := fmt.Sprintf("https://%s:%d/status", d.daemonIP(), predastoreDBPort) + return fetchPredastoreRole(url, roleTLSHTTPClient) +} + +var roleHTTPClient = &http.Client{Timeout: 500 * time.Millisecond} + +var roleTLSHTTPClient = &http.Client{ + Timeout: 500 * time.Millisecond, + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, // local self-signed cert + }, +} + +// fetchNATSRole queries a NATS /varz endpoint and returns "leader", "follower", or "". +func fetchNATSRole(url string, client *http.Client) string { + resp, err := client.Get(url) //nolint:noctx // internal monitoring call + if err != nil { + slog.Debug("Failed to query NATS monitoring", "err", err) + return "" + } + defer resp.Body.Close() + + var varz struct { + ServerName string `json:"server_name"` + JetStream struct { + Meta struct { + Leader string `json:"leader"` + ClusterSize int `json:"cluster_size"` + } `json:"meta"` + } `json:"jetstream"` + } + if err := json.NewDecoder(resp.Body).Decode(&varz); err != nil { + slog.Debug("Failed to decode NATS varz", "err", err) + return "" + } + + // Single node or no meta cluster — this node is the leader by default + if varz.JetStream.Meta.ClusterSize <= 1 { + return roleLeader + } + if varz.JetStream.Meta.Leader == varz.ServerName { + return roleLeader + } + return roleFollower +} + +// fetchPredastoreRole queries a Predastore /status endpoint and returns "leader", "follower", or "". +func fetchPredastoreRole(url string, client *http.Client) string { + resp, err := client.Get(url) //nolint:noctx // internal monitoring call + if err != nil { + slog.Debug("Failed to query Predastore status", "err", err) + return "" + } + defer resp.Body.Close() + + var status struct { + IsLeader bool `json:"is_leader"` + } + if err := json.NewDecoder(resp.Body).Decode(&status); err != nil { + slog.Debug("Failed to decode Predastore status", "err", err) + return "" + } + + if status.IsLeader { + return roleLeader + } + return roleFollower +} + // handleNodeVMs responds with the list of VMs running on this node. // Used by the CLI: spx get vms. func (d *Daemon) handleNodeVMs(msg *nats.Msg) { diff --git a/spinifex/daemon/roles_test.go b/spinifex/daemon/roles_test.go new file mode 100644 index 000000000..3bfc413ab --- /dev/null +++ b/spinifex/daemon/roles_test.go @@ -0,0 +1,113 @@ +package daemon + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestFetchNATSRole_SingleNode(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{ + "server_name": "spinifex-nats-node1", + "jetstream": { + "meta": { + "leader": "", + "cluster_size": 0 + } + } + }`)) + })) + defer srv.Close() + + role := fetchNATSRole(srv.URL, srv.Client()) + assert.Equal(t, "leader", role) +} + +func TestFetchNATSRole_ClusterLeader(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{ + "server_name": "spinifex-nats-node1", + "jetstream": { + "meta": { + "leader": "spinifex-nats-node1", + "cluster_size": 3 + } + } + }`)) + })) + defer srv.Close() + + role := fetchNATSRole(srv.URL, srv.Client()) + assert.Equal(t, "leader", role) +} + +func TestFetchNATSRole_ClusterFollower(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{ + "server_name": "spinifex-nats-node2", + "jetstream": { + "meta": { + "leader": "spinifex-nats-node1", + "cluster_size": 3 + } + } + }`)) + })) + defer srv.Close() + + role := fetchNATSRole(srv.URL, srv.Client()) + assert.Equal(t, "follower", role) +} + +func TestFetchNATSRole_Unreachable(t *testing.T) { + role := fetchNATSRole("http://127.0.0.1:1/varz", &http.Client{}) + assert.Equal(t, "", role) +} + +func TestFetchNATSRole_BadJSON(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`not json`)) + })) + defer srv.Close() + + role := fetchNATSRole(srv.URL, srv.Client()) + assert.Equal(t, "", role) +} + +func TestFetchPredastoreRole_Leader(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"node_id":"1","state":"Leader","is_leader":true}`)) + })) + defer srv.Close() + + role := fetchPredastoreRole(srv.URL, srv.Client()) + assert.Equal(t, "leader", role) +} + +func TestFetchPredastoreRole_Follower(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"node_id":"2","state":"Follower","is_leader":false}`)) + })) + defer srv.Close() + + role := fetchPredastoreRole(srv.URL, srv.Client()) + assert.Equal(t, "follower", role) +} + +func TestFetchPredastoreRole_Unreachable(t *testing.T) { + role := fetchPredastoreRole("https://127.0.0.1:1/status", &http.Client{}) + assert.Equal(t, "", role) +} + +func TestFetchPredastoreRole_BadJSON(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`not json`)) + })) + defer srv.Close() + + role := fetchPredastoreRole(srv.URL, srv.Client()) + assert.Equal(t, "", role) +} diff --git a/spinifex/types/node.go b/spinifex/types/node.go index be1758fb8..61322d16f 100644 --- a/spinifex/types/node.go +++ b/spinifex/types/node.go @@ -20,6 +20,10 @@ type NodeStatusResponse struct { AllocMemGB float64 `json:"alloc_mem_gb"` VMCount int `json:"vm_count"` InstanceTypes []InstanceTypeCap `json:"instance_types"` + + // Leader roles for clustered services (empty string = service not running or not clustered) + NATSRole string `json:"nats_role,omitempty"` // "leader" or "follower" + PredastoreRole string `json:"predastore_role,omitempty"` // "leader" or "follower" } // InstanceTypeCap describes available capacity for one instance type on a node.