gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-20 03:11:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add leaders to get nodes command

Browse Source
This commit is contained in:
Julian Sommer
2026-03-24 11:50:02 +10:00
parent 0fbf4537c0
commit 638828f636
5 changed files with 286 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/spinifex/cmd/get.go
+17 -1
View File
@@ -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,
cmd/spinifex/cmd/get_test.go
+49
View File
@@ -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)
})
}
}
spinifex/daemon/daemon_handlers.go
+103
View File
@@ -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) {
spinifex/daemon/roles_test.go
+113
View File
@@ -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)
}
spinifex/types/node.go
+4
View File
@@ -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.