enhance cluster status (#11)

pkg/cfg/config.go

@@ -31,7 +31,7 @@ import (
 )
 
 const (
-	Version string = `v0.0.5`
+	Version string = `v0.0.6`
 )
 
 type Cluster struct {

pkg/es/cluster.go

@@ -22,8 +22,12 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"sync"
 
 	"codeberg.org/scip/esctl/pkg/cfg"
+	"github.com/elastic/go-elasticsearch/v9/typedapi/ccr/stats"
+	"github.com/elastic/go-elasticsearch/v9/typedapi/cluster/health"
+	"github.com/elastic/go-elasticsearch/v9/typedapi/core/info"
 	"github.com/elastic/go-elasticsearch/v9/typedapi/types"
 	"github.com/urfave/cli/v3"
 )
@@ -32,8 +36,22 @@ const (
 	DefaultExclude = `(part|monitoring|.internal|metrics-endpoint)`
 )
 
+const (
+	ResponseHealth = iota
+	ResponseInfo
+	ResponseCcr
+)
+
 type ClusterIndices map[string]map[string]*types.IndicesRecord
 
+type apiResponse struct {
+	error  error
+	info   *info.Response
+	health *health.Response
+	ccr    *stats.Response
+	which  int
+}
+
 func ClusterCompare(conf *cfg.Config, leader, follower string) error {
 	if !checkClusterFollower(conf, leader) {
 		return errors.New("leader/follower attribution is invalid, reverse cluster attribution and retry")
@@ -102,6 +120,8 @@ func ClusterList(conf *cfg.Config) error {
 	return nil
 }
 
+// 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 ClusterStatus(conf *cfg.Config) error {
 	clusters := []string{}
 
@@ -119,34 +139,60 @@ func ClusterStatus(conf *cfg.Config) error {
 			es = conf.Clusters[cluster].ES
 		}
 
-		res, err := es.Cluster.Health().
-			Header("content-type", "application/json").
-			Header("accept", "application/json").
-			Do(context.Background())
-		if err != nil {
-			return fmt.Errorf("failed to get cluster health: %s", err)
+		responses := make(chan apiResponse, 3)
+		wg := &sync.WaitGroup{}
+
+		wg.Add(3)
+		go getClusterData(es, wg, responses, "health")
+		go getClusterData(es, wg, responses, "info")
+		go getClusterData(es, wg, responses, "ccrstats")
+
+		wg.Wait()
+
+		var clusterhealth *health.Response
+		var info *info.Response
+		var ccrstats *stats.Response
+
+		for i := 0; i < 3; i++ {
+			r := <-responses
+
+			if r.error != nil {
+				return r.error
 			}
 
-		slog.Debug("ES result", "cluster health", res)
+			switch r.which {
+			case ResponseHealth:
+				clusterhealth = r.health
+			case ResponseCcr:
+				ccrstats = r.ccr
+			case ResponseInfo:
+				info = r.info
+			}
+		}
 
-		info, err := es.Info().
-			Header("content-type", "application/json").
-			Header("accept", "application/json").
-			Do(context.Background())
-		if err != nil {
-			return fmt.Errorf("failed to get cluster info: %s", err)
+		slog.Debug("ES result", "cluster health", clusterhealth)
+
+		ccrfollowing := ""
+		if len(ccrstats.AutoFollowStats.AutoFollowedClusters) > 0 {
+			// is following another cluster
+			ccrfollowing = fmt.Sprintf("%s (%d/%d)",
+				ccrstats.AutoFollowStats.AutoFollowedClusters[0].ClusterName,
+				ccrstats.AutoFollowStats.NumberOfSuccessfulFollowIndices,
+				ccrstats.AutoFollowStats.NumberOfFailedFollowIndices,
+			)
 		}
 
 		table := NewTable(2, 5)
 		table.Addheaders(cluster, "status")
 
 		table.entries = [][]string{
-			{"Cluster Name", Colorize(*&res.Status.Name, res.ClusterName)},
+			{"Cluster Name", Colorize(*&clusterhealth.Status.Name, clusterhealth.ClusterName)},
 			{"ES Version", fmt.Sprintf("%s", info.Version.Int)},
-			{"Active Shards", fmt.Sprintf("%d", res.ActiveShards)},
-			{"Active Primary Shards", fmt.Sprintf("%d", res.ActivePrimaryShards)},
-			{"Indicies", fmt.Sprintf("%d", len(res.Indices))},
-			{"Nodes", fmt.Sprintf("%d", res.NumberOfNodes)},
+			{"Active Shards", fmt.Sprintf("%d", clusterhealth.ActiveShards)},
+			{"Active Primary Shards", fmt.Sprintf("%d", clusterhealth.ActivePrimaryShards)},
+			{"Indicies", fmt.Sprintf("%d", len(clusterhealth.Indices))},
+			{"Nodes", fmt.Sprintf("%d", clusterhealth.NumberOfNodes)},
+			{"AutoFollow (success/failed indices)", ccrfollowing},
 		}
 
 		table.PrintMarkdown()

pkg/es/cluster_util.go

@@ -18,11 +18,14 @@ package es
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"regexp"
 	"strings"
+	"sync"
 
 	"codeberg.org/scip/esctl/pkg/cfg"
+	"github.com/elastic/go-elasticsearch/v9"
 	"github.com/elastic/go-elasticsearch/v9/typedapi/cluster/health"
 	"github.com/elastic/go-elasticsearch/v9/typedapi/types"
 )
@@ -377,3 +380,48 @@ func splitArg(arg string) (string, string) {
 		return parts[0], parts[1]
 	}
 }
+
+func getClusterData(es *elasticsearch.TypedClient, wg *sync.WaitGroup, reschan chan apiResponse, which string) {
+	defer wg.Done()
+
+	ar := apiResponse{}
+	arerr := errors.New("")
+
+	switch which {
+	case "health":
+		res, err := es.Cluster.Health().
+			Header("content-type", "application/json").
+			Header("accept", "application/json").
+			Do(context.Background())
+
+		ar.health = res
+		ar.which = ResponseHealth
+		arerr = err
+
+	case "info":
+		res, err := es.Info().
+			Header("content-type", "application/json").
+			Header("accept", "application/json").
+			Do(context.Background())
+
+		ar.info = res
+		ar.which = ResponseInfo
+		arerr = err
+
+	case "ccrstats":
+		res, err := es.Ccr.Stats().
+			Header("content-type", "application/json").
+			Header("accept", "application/json").
+			Do(context.Background())
+
+		ar.ccr = res
+		ar.which = ResponseCcr
+		arerr = err
+	}
+
+	if arerr != nil {
+		ar.error = fmt.Errorf("failed to get cluster health: %s", arerr)
+	}
+
+	reschan <- ar
+}