add node show + node clients (#61)

pkg/cfg/cluster.go

@@ -104,7 +104,6 @@ func (cluster *Cluster) ES() *elasticsearch.TypedClient {
 
 // add authentication to es client, if not yet done
 func (cluster *Cluster) CheckAuth() error {
-
 	if cluster.Pass == "" && cluster.User != "" && cluster.Token == "" && cluster.Default {
 		// no token - user is set, but no password.
 

pkg/es/api.go

@@ -172,7 +172,17 @@ func CallAPI(conf *cfg.Config, verb, path, data string) ([]byte, error) {
 
 	req.Header.Add("Content-Type", "application/json")
 	req.Header.Add("accept", "application/json")
-	req.Header.Add("Authorization", "Basic "+encodeAuth(conf.DefaultCluster.User, conf.DefaultCluster.Pass))
+
+	// make sure we have got all we need
+	if err := conf.DefaultCluster.CheckAuth(); err != nil {
+		return nil, err
+	}
+
+	if conf.DefaultCluster.Token != "" {
+		req.Header.Add("Authorization", "APIKey "+conf.DefaultCluster.Token)
+	} else {
+		req.Header.Add("Authorization", "Basic "+encodeAuth(conf.DefaultCluster.User, conf.DefaultCluster.Pass))
+	}
 
 	// actually execute the request
 	resp, err := client.Do(req)

pkg/es/node.go

@@ -20,9 +20,12 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"strings"
+	"time"
 
 	"codeberg.org/scip/esctl/pkg/cfg"
 	"codeberg.org/scip/esctl/pkg/printer"
+	"github.com/dustin/go-humanize"
 )
 
 func NodeList(conf *cfg.Config) error {
@@ -56,3 +59,145 @@ func NodeList(conf *cfg.Config) error {
 
 	return nil
 }
+
+func NodeNames(conf *cfg.Config) ([]string, error) {
+	nodes, err := conf.DefaultCluster.ES().Cat.Nodes().
+		Do(context.Background())
+	if err != nil {
+		return nil, fmt.Errorf("failed to get nodes: %s", esErrorString(err))
+	}
+
+	slog.Debug("ES result", "nodes", nodes)
+
+	nodelist := make([]string, len(nodes))
+
+	for idx, node := range nodes {
+		nodelist[idx] = *node.Name
+	}
+
+	return nodelist, err
+}
+
+// FIXME: adding the settings metric leads to json unmarshall error:
+// https://github.com/elastic/go-elasticsearch/issues/1524
+func NodeShow(conf *cfg.Config, nodename string) error {
+	res, err := conf.DefaultCluster.ES().Nodes.Info().
+		NodeId(nodename).
+		Metric("os, jvm, thread_pool, remote_cluster_server").
+		Do(context.Background())
+	if err != nil {
+		return fmt.Errorf("failed to get node info: %s", esErrorString(err))
+	}
+
+	slog.Debug("ES result", "node", res)
+
+	stats, err := conf.DefaultCluster.ES().Nodes.Stats().
+		NodeId(nodename).
+		Do(context.Background())
+	if err != nil {
+		return fmt.Errorf("failed to get node stats: %s", esErrorString(err))
+	}
+
+	slog.Debug("ES result", "stat", stats)
+
+	for id, info := range res.Nodes {
+		stat := stats.Nodes[id]
+
+		table := printer.NewTable(conf, 2, 0)
+		table.Addheaders(nodename+" property", "value")
+
+		roles := make([]string, len(info.Roles))
+		for idx, role := range info.Roles {
+			roles[idx] = role.Name
+		}
+
+		k8snode := info.Attributes["k8s_node_name"]
+
+		table.Entries = [][]string{
+			{"Id", id},
+			{"Name", nodename},
+			{"Kubernetes node", k8snode},
+			{"Ip address", info.Ip},
+			{"Node rank", *stat.AdaptiveSelection[id].Rank},
+			{"JVM", info.Jvm.VmName + " " + info.Jvm.Version},
+			{"JVM Started", time.UnixMilli(info.Jvm.StartTimeInMillis).String()},
+			{"OS", info.Os.PrettyName + " " + info.Os.Version},
+			{"Node roles", strings.Join(roles, ",")},
+			{"Node version", info.Version},
+			{"HTTP clients", fmt.Sprintf("%d", *stat.Http.CurrentOpen)},
+			{"CPUs", fmt.Sprintf("%d", *info.Os.AllocatedProcessors)},
+			{"Load 15m/5m/1m", fmt.Sprintf("%.2f/%.2f/%.2f",
+				stat.Os.Cpu.LoadAverage["15m"],
+				stat.Os.Cpu.LoadAverage["5m"],
+				stat.Os.Cpu.LoadAverage["1m"],
+			)},
+			{"Open FD's", fmt.Sprintf("%d", *stat.Process.OpenFileDescriptors)},
+			{"Response time avg", fmt.Sprintf("%dns", *stat.AdaptiveSelection[id].AvgResponseTimeNs)},
+			{"Memory usage (used/avail)",
+				humanize.Bytes(uint64(*stat.Os.Mem.UsedInBytes)) + " / " + humanize.Bytes(uint64(*stat.Os.Mem.TotalInBytes))},
+		}
+
+		if len(stat.Fs.Data) > 0 {
+			fs := stat.Fs.Data[0]
+			table.Entries = append(table.Entries, [][]string{
+				{"Storage usage (used/avail)",
+					humanize.Bytes(uint64(*fs.AvailableInBytes)) + " / " + humanize.Bytes(uint64(*fs.TotalInBytes))},
+				{"Storage mount", *fs.Mount},
+			}...)
+		}
+
+		if err := table.Print(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func NodeClients(conf *cfg.Config, nodename string) error {
+	stats, err := conf.DefaultCluster.ES().Nodes.Stats().
+		NodeId(nodename).
+		Do(context.Background())
+	if err != nil {
+		return fmt.Errorf("failed to get node stats: %s", esErrorString(err))
+	}
+
+	slog.Debug("ES result", "stat", stats)
+
+	table := printer.NewTable(conf, 5, 0)
+	table.Addheaders("agent", "id", "when", "from host", "url")
+
+	for _, stat := range stats.Nodes {
+		for _, client := range stat.Http.Clients {
+			if client.ClosedTimeMillis == nil {
+				agent := ""
+				if client.Agent != nil {
+					agent = *client.Agent
+				}
+
+				uri := ""
+				if client.LastUri != nil {
+					uri = *client.LastUri
+
+					if !conf.All {
+						parts := strings.Split(uri, "?")
+						uri = parts[0]
+					}
+				}
+
+				table.AddRow(
+					agent,
+					fmt.Sprintf("%d", *client.Id),
+					time.UnixMilli(*client.LastRequestTimeMillis).String(),
+					*client.RemoteAddress,
+					uri,
+				)
+			}
+		}
+
+		break
+	}
+
+	table.Sort()
+	return table.Print()
+}