esctl/pkg/es/cluster.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 (
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"

	"codeberg.org/scip/esctl/pkg/cfg"
	"codeberg.org/scip/esctl/pkg/printer"
	"github.com/dustin/go-humanize"
	clusterstats "github.com/elastic/go-elasticsearch/v9/typedapi/cluster/stats"
	"github.com/elastic/go-elasticsearch/v9/typedapi/types"
)

type ClusterIndices map[string]map[string]*types.IndicesRecord

type clusterReachable struct {
	reachable bool
	err       error
}

func ClusterList(conf *cfg.Config) error {
	var (
		mu sync.Mutex
		wg sync.WaitGroup
	)

	reachable := make(map[string]clusterReachable, len(conf.Clusters))

	// check endpoints in parallel to speed things up
	for name, cluster := range conf.Clusters {
		wg.Go(func() {
			online, err := cluster.IsReachable()

			mu.Lock()
			reachable[name] = clusterReachable{reachable: online, err: err}
			mu.Unlock()
		})
	}

	wg.Wait()

	table := printer.NewTable(conf, 5, len(conf.Clusters))
	table.Addheaders("cluster", "uri", "reachable", "current", "error")

	idx := 0

	for name, cluster := range conf.Clusters {
		reachableStr := "no"
		current := "no"
		errmsg := ""

		if reachable[name].reachable {
			reachableStr = printer.Colorize(conf, "green", "reachable")
		}

		if cluster.Default {
			current = printer.Colorize(conf, "green", "yes")

			if !reachable[name].reachable {
				reachableStr = printer.Colorize(conf, "red", "no")
				errmsg = reachable[name].err.Error()
			}
		}

		table.Entries[idx] = []any{name, cluster.Uri, reachableStr, current, errmsg}
		idx++
	}

	table.Sort()

	return table.Print()
}

// We're using goroutines  here to parallelize API  requests, since we
// have to do 3 of'em for each cluster. This speeds things up.
func getClusterStatus(conf *cfg.Config) (*apiResponse, error) {
	gocount := 6
	if conf.Verbose {
		gocount++
	}

	es := conf.DefaultCluster.ES()

	responses := make(chan apiResponse, gocount)
	wg := &sync.WaitGroup{}

	wg.Add(gocount)
	go getApiData(conf, es, wg, responses, "health")
	go getApiData(conf, es, wg, responses, "healthreport")
	go getApiData(conf, es, wg, responses, "info")
	go getApiData(conf, es, wg, responses, "ccr")
	go getApiData(conf, es, wg, responses, "indices")
	go getApiData(conf, es, wg, responses, "tasks")

	if conf.Verbose {
		go getApiData(conf, es, wg, responses, "stats")
	}

	wg.Wait()

	all := apiResponse{}

	var err error

	for range gocount {
		res := <-responses

		err = errors.Join(err, res.error)

		switch res.which {
		case ResponseHealth:
			all.health = res.health
		case ResponseCcr:
			all.ccr = res.ccr
		case ResponseInfo:
			all.info = res.info
		case ResponseStats:
			all.stats = res.stats
		case ResponseIndices:
			all.indices = res.indices
		case ResponseTasks:
			all.tasks = res.tasks
		case ResponseHealthReport:
			all.healthreport = res.healthreport
		}
	}

	return &all, err
}

func ClusterStatus(conf *cfg.Config) error {
	res, err := getClusterStatus(conf)
	if err != nil && !strings.Contains(err.Error(), "current license is non-compliant") {
		return err
	}

	slog.Debug("ES result", "cluster health", res.health)

	var (
		isleader     bool
		ccrfollowing string
	)

	if res.ccr != nil {
		isleader = len(res.ccr.AutoFollowStats.AutoFollowedClusters) == 0

		if len(res.ccr.AutoFollowStats.AutoFollowedClusters) > 0 {
			// is following another cluster
			ccrfollowing = fmt.Sprintf("%s (%d/%d)",
				res.ccr.AutoFollowStats.AutoFollowedClusters[0].ClusterName,
				res.ccr.AutoFollowStats.NumberOfSuccessfulFollowIndices,
				res.ccr.AutoFollowStats.NumberOfFailedFollowIndices,
			)
		}
	}

	// look for red indices, if any
	failedIndices := 0
	greenIndices := 0

	for _, index := range *res.indices {
		switch {
		case *index.Health != "green":
			failedIndices++
		case !strings.HasPrefix(*index.Index, "."):
			// don't count internal indices
			greenIndices++
		}
	}

	// look for long running tasks
	longtasks := 0

	for _, task := range *res.tasks {
		if strings.Contains(*task.RunningTime, "d") {
			longtasks++
		}
	}

	table := printer.NewTable(conf, 2, 7)
	table.Addheaders(conf.DefaultCluster.Name, "status")

	table.Entries = [][]any{
		{"Cluster Name", res.health.ClusterName},
		{"ES Status", printer.Colorize(conf, res.health.Status.Name, res.health.Status.Name)},
		{"ES Version", res.info.Version.Int},
		{"Is Leader", isleader},
		{"Active Shards", res.health.ActiveShards},
		{"Active Primary Shards", res.health.ActivePrimaryShards},
		{"Unassigned Shards", res.health.UnassignedShards},
		{"Unassigned Primary Shards", res.health.UnassignedPrimaryShards},
		{"Pending Tasks", res.health.NumberOfPendingTasks},
		{"Nodes", res.health.NumberOfNodes},
		{"Green Indices", greenIndices},
		{"Failed Indices", failedIndices},
		{"Long Running Tasks", longtasks},
	}

	if !isleader && res.ccr != nil {
		table.Entries = append(table.Entries, [][]any{
			{"AutoFollow (success/failed indices)", ccrfollowing},
			{"Followed Indices", len(res.ccr.FollowStats.Indices)},
		}...)
	}

	if conf.Verbose {
		table = gatherClusterStats(res.stats, table)
	}

	if res.health.Status.Name != "green" {
		for name, indicator := range res.healthreport.Indicators {
			if indicator.Status != "green" {
				table.Entries = append(table.Entries, []any{
					printer.Colorize(conf, indicator.Status, "Bad health "+name), indicator.Symptom,
				})

				for _, diag := range indicator.Diagnosis {
					table.Entries = append(table.Entries, []any{" -> cause", diag.Cause})

					for resource, items := range diag.AffectedResources {
						table.Entries = append(table.Entries, []any{"    -> affected " + resource, strings.Join(items, ",")})
					}

					if diag.Action != "" {
						table.AddRow("        -> suggested action to fix", diag.Action)
					}
				}
			}
		}
	}

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

	return nil
}

func gatherClusterStats(clusterstats *clusterstats.Response, table *printer.Table) *printer.Table {
	var (
		querycount int64
		vmversion  string
	)

	for _, count := range clusterstats.Indices.Search.Queries {
		querycount += count
	}

	if len(clusterstats.Nodes.Jvm.Versions) > 0 {
		vmversion = strings.Join([]string{
			clusterstats.Nodes.Jvm.Versions[0].VmName,
			clusterstats.Nodes.Jvm.Versions[0].VmVersion}, " ")
	}

	table.Entries = append(table.Entries, [][]any{
		{"Indicies", clusterstats.Indices.Count},
		{"Docs", clusterstats.Indices.Docs.Count},
		{"Total Size", printer.Bytes(clusterstats.Indices.Docs.TotalSizeInBytes)},
		{"Total Queries", "%d", querycount},
		{"Shards Primaries", clusterstats.Indices.Shards.Primaries},
		{"Shards Total", clusterstats.Indices.Shards.Total},
		{"Storage", fmt.Sprintf(
			"%s/%s",
			humanize.Bytes(uint64(clusterstats.Indices.Store.SizeInBytes)),
			humanize.Bytes(uint64(*clusterstats.Indices.Store.TotalDataSetSizeInBytes)),
		)},
		{"JVM Heap Memory", fmt.Sprintf(
			"%s/%s",
			humanize.Bytes(uint64(clusterstats.Nodes.Jvm.Mem.HeapUsedInBytes)),
			humanize.Bytes(uint64(clusterstats.Nodes.Jvm.Mem.HeapMaxInBytes)),
		)},
		{"JVM Threads", clusterstats.Nodes.Jvm.Threads},
		{"JVM Version", vmversion},
		{"CPUs", clusterstats.Nodes.Os.AllocatedProcessors},
		{"CPU Usage", fmt.Sprintf("%d%%", clusterstats.Nodes.Process.Cpu.Percent)},
		{"Open FDs", clusterstats.Nodes.Process.OpenFileDescriptors.Avg},
	}...)

	return table
}