enhance/clusterstatus (#44)

fixes #35

pkg/es/cluster.go

@@ -27,6 +27,8 @@ import (
 	"codeberg.org/scip/esctl/pkg/cfg"
 	"codeberg.org/scip/esctl/pkg/printer"
 	"github.com/dustin/go-humanize"
+	"github.com/elastic/go-elasticsearch/v9/typedapi/cat/indices"
+	"github.com/elastic/go-elasticsearch/v9/typedapi/cat/tasks"
 	"github.com/elastic/go-elasticsearch/v9/typedapi/ccr/stats"
 	"github.com/elastic/go-elasticsearch/v9/typedapi/cluster/health"
 	clusterstats "github.com/elastic/go-elasticsearch/v9/typedapi/cluster/stats"
@@ -39,17 +41,21 @@ const (
 	ResponseInfo
 	ResponseCcr
 	ResponseStats
+	ResponseIndices
+	ResponseTasks
 )
 
 type ClusterIndices map[string]map[string]*types.IndicesRecord
 
 type apiResponse struct {
-	error  error
-	info   *info.Response
-	health *health.Response
-	ccr    *stats.Response
-	stats  *clusterstats.Response
-	which  int
+	error   error
+	info    *info.Response
+	health  *health.Response
+	ccr     *stats.Response
+	stats   *clusterstats.Response
+	indices *indices.Response
+	tasks   *tasks.Response
+	which   int
 }
 
 func ClusterList(conf *cfg.Config) error {
@@ -94,7 +100,7 @@ func ClusterList(conf *cfg.Config) error {
 // have to do 3 of'em for each cluster. This speeds things up.
 func ClusterStatus(conf *cfg.Config) error {
 	clusters := []string{}
-	gocount := 3
+	gocount := 5
 	if conf.Verbose {
 		gocount++
 	}
@@ -120,6 +126,8 @@ func ClusterStatus(conf *cfg.Config) error {
 		go getClusterData(es, wg, responses, "health")
 		go getClusterData(es, wg, responses, "info")
 		go getClusterData(es, wg, responses, "ccrstats")
+		go getClusterData(es, wg, responses, "indices")
+		go getClusterData(es, wg, responses, "tasks")
 
 		if conf.Verbose {
 			go getClusterData(es, wg, responses, "stats")
@@ -131,6 +139,8 @@ func ClusterStatus(conf *cfg.Config) error {
 		var info *info.Response
 		var ccrstats *stats.Response
 		var clusterstats *clusterstats.Response
+		var indexstats *indices.Response
+		var taskstatus *tasks.Response
 
 		for i := 0; i < gocount; i++ {
 			r := <-responses
@@ -148,11 +158,17 @@ func ClusterStatus(conf *cfg.Config) error {
 				info = r.info
 			case ResponseStats:
 				clusterstats = r.stats
+			case ResponseIndices:
+				indexstats = r.indices
+			case ResponseTasks:
+				taskstatus = r.tasks
 			}
 		}
 
 		slog.Debug("ES result", "cluster health", clusterhealth)
 
+		isleader := len(ccrstats.AutoFollowStats.AutoFollowedClusters) == 0
+
 		ccrfollowing := ""
 		if len(ccrstats.AutoFollowStats.AutoFollowedClusters) > 0 {
 			// is following another cluster
@@ -163,6 +179,22 @@ func ClusterStatus(conf *cfg.Config) error {
 			)
 		}
 
+		// look for red indices, if any
+		redindices := 0
+		for _, index := range *indexstats {
+			if *index.Health == "red" {
+				redindices++
+			}
+		}
+
+		// look for long running tasks
+		longtasks := 0
+		for _, task := range *taskstatus {
+			if strings.Contains(*task.RunningTime, "d") {
+				longtasks++
+			}
+		}
+
 		table := printer.NewTable(conf, 2, 7)
 		table.Addheaders(cluster, "status")
 
@@ -170,10 +202,22 @@ func ClusterStatus(conf *cfg.Config) error {
 			{"Cluster Name", clusterhealth.ClusterName},
 			{"ES Status", printer.Colorize(conf, clusterhealth.Status.Name, clusterhealth.Status.Name)},
 			{"ES Version", info.Version.Int},
+			{"Is Leader", fmt.Sprintf("%t", isleader)},
 			{"Active Shards", fmt.Sprintf("%d", clusterhealth.ActiveShards)},
 			{"Active Primary Shards", fmt.Sprintf("%d", clusterhealth.ActivePrimaryShards)},
+			{"Unassigned Shards", fmt.Sprintf("%d", clusterhealth.UnassignedShards)},
+			{"Unassigned Primary Shards", fmt.Sprintf("%d", clusterhealth.UnassignedPrimaryShards)},
+			{"Pending Tasks", fmt.Sprintf("%d", clusterhealth.NumberOfPendingTasks)},
 			{"Nodes", fmt.Sprintf("%d", clusterhealth.NumberOfNodes)},
-			{"AutoFollow (success/failed indices)", ccrfollowing},
+			{"Red Indices", fmt.Sprintf("%d", redindices)},
+			{"Long Running Tasks", fmt.Sprintf("%d", longtasks)},
+		}
+
+		if !isleader {
+			table.Entries = append(table.Entries, [][]string{
+				{"AutoFollow (success/failed indices)", ccrfollowing},
+				{"Followed Indices", fmt.Sprintf("%d", len(ccrstats.FollowStats.Indices))},
+			}...)
 		}
 
 		if conf.Verbose {
@@ -202,11 +246,8 @@ func gatherClusterStats(conf *cfg.Config, clusterstats *clusterstats.Response, t
 			clusterstats.Nodes.Jvm.Versions[0].VmVersion}, " ")
 	}
 
-	isleader := checkClusterIsLeader(conf, conf.CurrentCluster)
-
 	table.Entries = append(table.Entries, [][]string{
 		{"Indicies", fmt.Sprintf("%d", clusterstats.Indices.Count)},
-		{"Is Leader", fmt.Sprintf("%t", isleader)},
 		{"Docs", fmt.Sprintf("%d", clusterstats.Indices.Docs.Count)},
 		{"Total Size", humanize.Bytes(uint64(clusterstats.Indices.Docs.TotalSizeInBytes))},
 		{"Total Queries", fmt.Sprintf("%d", querycount)},
@@ -217,7 +258,7 @@ func gatherClusterStats(conf *cfg.Config, clusterstats *clusterstats.Response, t
 			humanize.Bytes(uint64(clusterstats.Indices.Store.SizeInBytes)),
 			humanize.Bytes(uint64(*clusterstats.Indices.Store.TotalDataSetSizeInBytes)),
 		)},
-		{"JVM Heap", fmt.Sprintf(
+		{"JVM Heap Memory", fmt.Sprintf(
 			"%s/%s",
 			humanize.Bytes(uint64(clusterstats.Nodes.Jvm.Mem.HeapUsedInBytes)),
 			humanize.Bytes(uint64(clusterstats.Nodes.Jvm.Mem.HeapMaxInBytes)),

pkg/es/cluster_util.go

@@ -390,6 +390,26 @@ func getClusterData(es *elasticsearch.TypedClient, wg *sync.WaitGroup, reschan c
 		ar.stats = res
 		ar.which = ResponseStats
 		arerr = err
+
+	case "indices":
+		res, err := es.Cat.Indices().
+			Header("content-type", "application/json").
+			Header("accept", "application/json").
+			Do(context.Background())
+
+		ar.indices = &res
+		ar.which = ResponseIndices
+		arerr = err
+
+	case "tasks":
+		res, err := es.Cat.Tasks().
+			Header("content-type", "application/json").
+			Header("accept", "application/json").
+			Do(context.Background())
+
+		ar.tasks = &res
+		ar.which = ResponseTasks
+		arerr = err
 	}
 
 	if arerr != nil {