esctl/pkg/es/node.go

/*
Copyright © 2026 Thomas von Dein

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package es

import (
	"context"
	"fmt"
	"log/slog"
	"strconv"
	"strings"
	"time"

	"codeberg.org/scip/esctl/pkg/cfg"
	"codeberg.org/scip/esctl/pkg/printer"
)

func NodeList(conf *cfg.Config) error {
	// get nodes
	nodes, err := conf.DefaultCluster.ES().Cat.Nodes().Do(context.Background())
	if err != nil {
		return fmt.Errorf("failed to get nodes: %w", esErrorString(err))
	}

	slog.Debug("ES result", "nodes", nodes)

	table := printer.NewTableEmpty(conf).WithHeaders(
		"name", "ip", "load1m", "load5m", "load15m", "ram %", "heap %")

	for _, node := range nodes {
		table.AddRow(
			*node.Name,
			*node.Ip,
			*node.Load1M,
			*node.Load5M,
			*node.Load15M,
			node.RamPercent,
			node.HeapPercent,
		)
	}

	table.Sort()

	if err := table.Print(); err != nil {
		return err
	}

	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: %w", 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: %w", 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: %w", 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"]
		rank := "none"

		adsel, exists := stat.AdaptiveSelection[id]
		if exists {
			rank = *adsel.Rank
		}

		table.Entries = [][]any{
			{"Id", id},
			{"Name", nodename},
			{"Kubernetes node", k8snode},
			{"Ip address", info.Ip},
			{"Node rank", rank},
			{"JVM", info.Jvm.VmName + " " + info.Jvm.Version},
			{"JVM Started", time.UnixMilli(info.Jvm.StartTimeInMillis)},
			{"OS", info.Os.PrettyName + " " + info.Os.Version},
			{"Node roles", roles},
			{"Node version", info.Version},
			// FIXME: not implemented upstream
			// see: https://github.com/elastic/go-elasticsearch/issues/1526
			// {"Allocated shards", stat.Allocations.XXX},
			{"HTTP clients", *stat.Http.CurrentOpen},
			{"CPUs", *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", *stat.Process.OpenFileDescriptors},
			{"HTTP sesssions current/total", fmt.Sprintf("%d/%d",
				*stat.Http.CurrentOpen,
				*stat.Http.TotalOpened,
			)},
			{"Traffic rx/tx",
				printer.ByteString(*stat.Transport.RxSizeInBytes) + " / " + printer.ByteString(*stat.Transport.TxSizeInBytes)},
			{"Response time avg", time.Duration(*stat.AdaptiveSelection[id].AvgResponseTimeNs)},
			{"Memory usage (used/avail)",
				printer.ByteString(*stat.Os.Mem.UsedInBytes) + " / " + printer.ByteString(*stat.Os.Mem.TotalInBytes)},
			{"Search queries current/total", fmt.Sprintf("%d/%d",
				stat.Indices.Search.QueryCurrent,
				stat.Indices.Search.QueryTotal,
			)},
			{"Search efficiency", stat.Indices.Search.QueryTimeInMillis / stat.Indices.Search.QueryTotal},
			{"Docs count", stat.Indices.Docs.Count},
			{"Merges current/total", fmt.Sprintf("%d/%d",
				stat.Indices.Merges.Current,
				stat.Indices.Merges.Total,
			)},
			{"Merge docs count current/total", fmt.Sprintf("%d/%d",
				stat.Indices.Merges.CurrentDocs,
				stat.Indices.Merges.TotalDocs,
			)},
			{"Merge size current/total", fmt.Sprintf("%s/%s",
				printer.ByteString(stat.Indices.Merges.CurrentSizeInBytes),
				printer.ByteString(stat.Indices.Merges.TotalSizeInBytes),
			)},
			{"CircuitBreaker trip count", *stat.Breakers["fielddata"].Tripped},
		}

		if len(stat.Fs.Data) > 0 {
			fs := stat.Fs.Data[0]
			table.AddRow("Storage usage (used/avail)",
				printer.ByteString(*fs.AvailableInBytes)+" / "+printer.ByteString(*fs.TotalInBytes))
			table.AddRow("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: %w", 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,
					strconv.FormatInt(*client.Id, 10),
					time.UnixMilli(*client.LastRequestTimeMillis).String(),
					*client.RemoteAddress,
					uri,
				)
			}
		}

		break
	}

	table.Sort()

	return table.Print()
}

func NodeUsage(conf *cfg.Config, nodeid string) error {
	usage := conf.DefaultCluster.ES().Nodes.Usage()

	if nodeid != "" {
		usage.NodeId(nodeid)
	}

	stats, err := usage.Do(context.Background())
	if err != nil {
		return fmt.Errorf("failed to get node usage: %w", esErrorString(err))
	}

	slog.Debug("ES result", "usage", stats)

	table := printer.NewTableEmpty(conf).WithHeaders(
		"node",
		"bulk",
		"doc get",
		"doc mget",
		"doc update",
		"index doc",
		"index stats",
		"search",
		"msearch",
		"open pit",
	)

	for id, actions := range stats.Nodes {
		node, err := getNodeName(conf, id)
		if err != nil {
			return err
		}

		stat := actions.RestActions

		table.AddRow(
			node,
			stat["bulk_action"],
			stat["document_get_action"],
			stat["document_mget_action"],
			stat["document_update_action"],
			stat["document_index_action"],
			stat["indices_stats_action"],
			stat["search_action"],
			stat["msearch_action"],
			stat["open_point_in_time"],
		)
	}

	return table.Print()
}

func getNodeName(conf *cfg.Config, id string) (string, error) {
	res, err := conf.DefaultCluster.ES().Nodes.Info().
		Metric("os").
		Do(context.Background())
	if err != nil {
		return "", fmt.Errorf("failed to get node info: %w", esErrorString(err))
	}

	for nodeid, node := range res.Nodes {
		if id == nodeid {
			return node.Name, nil
		}
	}

	return "", nil
}