diff --git a/pkg/es/cluster.go b/pkg/es/cluster.go index 6990079..4175dbd 100644 --- a/pkg/es/cluster.go +++ b/pkg/es/cluster.go @@ -18,12 +18,14 @@ package es import ( "context" + "errors" "fmt" "log" "log/slog" "regexp" "codeberg.org/scip/esctl/pkg/cfg" + "github.com/elastic/go-elasticsearch/v9/typedapi/cluster/health" "github.com/elastic/go-elasticsearch/v9/typedapi/types" ) @@ -34,6 +36,10 @@ const ( type ClusterIndices map[string]map[string]*types.IndicesRecord 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") + } + indices := ClusterIndices{} for _, alias := range []string{leader, follower} { @@ -44,7 +50,7 @@ func ClusterCompare(conf *cfg.Config, leader, follower string) error { res, err := cat.Do(context.Background()) if err != nil { - return fmt.Errorf("Error getting indicies on %s: %s", alias, err) + return fmt.Errorf("failed to get indicies on %s: %s", alias, err) } indices[alias] = make(map[string]*types.IndicesRecord, len(res)) @@ -54,7 +60,17 @@ func ClusterCompare(conf *cfg.Config, leader, follower string) error { } } - findIndicesOnlyOnMaster(conf, indices, leader, follower) + if !checkClusterStatus(conf, leader, follower) { + return errors.New("One of the two clusters is in a failed state") + } + + findIlmErrors(conf, leader, follower) + + if findIndicesOnlyOnLeader(conf, indices, leader, follower) && + findOrphanedIndices(conf, indices, leader, follower) && + findFailedFollowerIndices(conf, indices, follower) { + fmt.Println("everything's hunky-dory.") + } return nil } @@ -179,3 +195,279 @@ func findIndicesOnlyOnMaster(conf *cfg.Config, indices ClusterIndices, leader, f table.Sort() table.PrintMarkdown() } + +func checkClusterFollower(conf *cfg.Config, leader string) bool { + stats, err := conf.Clusters[leader].ES.Ccr.Stats(). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + fmt.Printf("failed to get ccr stats from %s: %s", leader, err) + return false + } + + if len(stats.AutoFollowStats.AutoFollowedClusters) == 0 { + // is not following anyone + return true + } + + if stats.AutoFollowStats.AutoFollowedClusters[0].ClusterName != "" { + fmt.Println("leader/follower attribution is invalid, reverse cluster attribution and retry") + return false + } + + return true +} + +// checks if both clusters are green +func checkClusterStatus(conf *cfg.Config, leader, follower string) bool { + status := map[string]*health.Response{} + + for _, cluster := range []string{leader, follower} { + st, err := conf.Clusters[leader].ES.Cluster.Health(). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + fmt.Printf("failed to get health from %s: %s", cluster, err) + return false + } + + status[cluster] = st + } + + table := NewTable(3, 5) + + table.Addheaders("setting", "leader:"+leader, "follower:"+follower) + + table.entries = [][]string{ + {"Cluster Name", + Colorize(*&status[leader].Status.Name, status[leader].ClusterName), + Colorize(*&status[follower].Status.Name, status[follower].ClusterName), + }, + {"Active Shards", + fmt.Sprintf("%d", status[leader].ActiveShards), + fmt.Sprintf("%d", status[follower].ActiveShards), + }, + {"Active Primary Shards", + fmt.Sprintf("%d", status[leader].ActivePrimaryShards), + fmt.Sprintf("%d", status[follower].ActivePrimaryShards), + }, + {"Indicies", + fmt.Sprintf("%d", len(status[leader].Indices)), + fmt.Sprintf("%d", len(status[follower].Indices)), + }, + {"Nodes", + fmt.Sprintf("%d", status[leader].NumberOfNodes), + fmt.Sprintf("%d", status[follower].NumberOfNodes), + }, + } + + table.PrintMarkdown() + + if status[leader].Status.Name == "green" && status[follower].Status.Name == "green" { + return true + } + + return false +} + +// finds indices on both clusters which have ilm errors +func findIlmErrors(conf *cfg.Config, leader, follower string) bool { + failed := map[string]map[string]string{} + + for _, cluster := range []string{leader, follower} { + ilm, err := conf.Clusters[cluster].ES.Ilm.ExplainLifecycle("_all"). + OnlyManaged(true). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + fmt.Printf("failed to get ilm status from %s: %s", cluster, err) + return false + } + + failed[cluster] = map[string]string{} + + for name, ilmstate := range ilm.Indices { + count := ilmstate.(*types.LifecycleExplainManaged).FailedStepRetryCount + + if count != nil && *count > 0 { + failed[cluster][name] = fmt.Sprintf("%d", *count) + } + } + } + + if len(failed[leader]) == 0 && len(failed[follower]) == 0 { + return true + } + + for idx, cluster := range []string{leader, follower} { + which := "leader" + if idx > 0 { + which = "follower" + } + + if len(failed[cluster]) > 0 { + idx := 0 + table := NewTable(2, len(failed[cluster])) + table.Addheaders("ilm errors on "+which, "errors") + + for name, count := range failed[cluster] { + table.entries[idx] = []string{name, count} + idx++ + } + + table.Sort() + table.PrintMarkdown() + } + } + + return false +} + +// find unsynchronized indicies only present on leader +func findIndicesOnlyOnLeader(conf *cfg.Config, indices ClusterIndices, leader, follower string) bool { + exclude := regexp.MustCompile(DefaultExclude) + if conf.Exclude != "" { + exclude = regexp.MustCompile(conf.Exclude) + } + + indexOnlyOnLeader := map[string]*types.IndicesRecord{} + for name, index := range indices[leader] { + if exclude.MatchString(name) { + continue + } + + _, followerHasIt := indices[follower][name] + if !followerHasIt { + // fetch index details + res, err := conf.Clusters[leader].ES.Indices.Get(name). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + continue // ignore it then + } + + _, defined := res[name] + if !defined { + // json response map didn't contain the index + continue + } + + isWritable := false + for _, alias := range res[name].Aliases { + if *alias.IsWriteIndex { + isWritable = true + break + } + } + + if isWritable { + // ignore index if associated alias index is writing + continue + } + + indexOnlyOnLeader[name] = index + } + } + + if len(indexOnlyOnLeader) == 0 { + return true + } + + idx := 0 + table := NewTable(3, len(indexOnlyOnLeader)) + table.Addheaders("index only on leader", "size", "docscount") + + for name, index := range indexOnlyOnLeader { + name := Colorize("red", name) + + table.entries[idx] = []string{name, *index.DatasetSize, *index.DocsCount} + idx++ + } + + table.Sort() + table.PrintMarkdown() + + return false +} + +// find indices only present on follower +func findOrphanedIndices(conf *cfg.Config, indices ClusterIndices, leader, follower string) bool { + orphaned := map[string]*types.IndicesRecord{} + + for name, index := range indices[follower] { + _, leaderHasIt := indices[leader][name] + if !leaderHasIt { + // fetch index details + res, err := conf.Clusters[follower].ES.Indices.Get(name). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + continue // ignore it then + } + + _, defined := res[name] + if !defined { + // json response map didn't contain the index + continue + } + + orphaned[name] = index + } + } + + if len(orphaned) == 0 { + return true + } + + idx := 0 + table := NewTable(3, len(orphaned)) + table.Addheaders("orphaned index on follower", "size", "docscount") + + for name, index := range orphaned { + name := Colorize("red", name) + + table.entries[idx] = []string{name, *index.DatasetSize, *index.DocsCount} + idx++ + } + + table.Sort() + table.PrintMarkdown() + + return false +} + +// find red indices on follower +func findFailedFollowerIndices(conf *cfg.Config, indices ClusterIndices, follower string) bool { + red := map[string]*types.IndicesRecord{} + + for name, index := range indices[follower] { + if *index.Health == "red" { + red[name] = index + } + } + + if len(red) == 0 { + return true + } + + idx := 0 + table := NewTable(3, len(red)) + table.Addheaders("red index on follower", "size", "docscount") + + for name, index := range red { + name := Colorize("red", name) + + table.entries[idx] = []string{name, *index.DatasetSize, *index.DocsCount} + idx++ + } + + table.Sort() + table.PrintMarkdown() + + return false +}