/* 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 . */ 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" ) const ( DefaultExclude = `(part|monitoring|.internal|metrics-endpoint)` ) 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} { cat := conf.Clusters[alias].ES.Cat.Indices(). // we need to add custom request headers, required for older ES instances Header("content-type", "application/json"). Header("accept", "application/json") res, err := cat.Do(context.Background()) if err != nil { return fmt.Errorf("failed to get indicies on %s: %s", alias, err) } indices[alias] = make(map[string]*types.IndicesRecord, len(res)) for _, index := range res { indices[alias][*index.Index] = &index } } 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 } func List(conf *cfg.Config) error { table := NewTable(2, len(conf.Clusters)) table.Addheaders("cluster", "uri") idx := 0 for name, cluster := range conf.Clusters { table.entries[idx] = []string{name, cluster.Uri} idx++ } table.PrintMarkdown() return nil } func Status(conf *cfg.Config) error { clusters := []string{} if conf.All { for key, _ := range conf.Clusters { clusters = append(clusters, key) } } else { clusters = []string{"default"} } for _, cluster := range clusters { es := conf.DefaultCluster.ES if cluster != "default" { 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 { log.Fatalf("Error getting health: %s", err) } slog.Debug("ES result", "cluster health", res) table := NewTable(2, 5) table.Addheaders(cluster, "status") table.entries = [][]string{ {"Cluster Name", Colorize(*&res.Status.Name, res.ClusterName)}, {"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)}, } table.PrintMarkdown() } return nil } // look for indicies only on leader func findIndicesOnlyOnMaster(conf *cfg.Config, indices ClusterIndices, leader, follower string) { 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 } } 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() } 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 }