From 5e82a3706b93b25c518f658b6672d8eed4a573bf Mon Sep 17 00:00:00 2001 From: "T. von Dein" Date: Thu, 7 May 2026 12:54:27 +0200 Subject: [PATCH] enhance cluster status (#11) --- pkg/cfg/config.go | 2 +- pkg/es/cluster.go | 82 ++++++++++++++++++------ pkg/es/{utilities.go => cluster_util.go} | 48 ++++++++++++++ 3 files changed, 113 insertions(+), 19 deletions(-) rename pkg/es/{utilities.go => cluster_util.go} (89%) diff --git a/pkg/cfg/config.go b/pkg/cfg/config.go index e666b66..edb9876 100644 --- a/pkg/cfg/config.go +++ b/pkg/cfg/config.go @@ -31,7 +31,7 @@ import ( ) const ( - Version string = `v0.0.5` + Version string = `v0.0.6` ) type Cluster struct { diff --git a/pkg/es/cluster.go b/pkg/es/cluster.go index a0c94f7..3a827fa 100644 --- a/pkg/es/cluster.go +++ b/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 + } + + switch r.which { + case ResponseHealth: + clusterhealth = r.health + case ResponseCcr: + ccrstats = r.ccr + case ResponseInfo: + info = r.info + } } - slog.Debug("ES result", "cluster health", res) + slog.Debug("ES result", "cluster health", clusterhealth) - 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) + 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() diff --git a/pkg/es/utilities.go b/pkg/es/cluster_util.go similarity index 89% rename from pkg/es/utilities.go rename to pkg/es/cluster_util.go index efb5b94..5de0a86 100644 --- a/pkg/es/utilities.go +++ b/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 +}