/* 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" "fmt" "regexp" "strings" "codeberg.org/scip/esctl/pkg/cfg" "codeberg.org/scip/esctl/pkg/printer" "github.com/elastic/go-elasticsearch/v9/typedapi/cluster/health" "github.com/elastic/go-elasticsearch/v9/typedapi/types" ) const ( DefaultExclude = `(part|monitoring|.internal|metrics-endpoint)` ) func checkClusterIsLeader(conf *cfg.Config, leader string) bool { stats, err := conf.Clusters[leader].ES().Ccr.Stats(). Do(context.Background()) if err != nil { fmt.Printf("failed to get ccr stats from %s: %s", leader, esErrorString(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(). Do(context.Background()) if err != nil { fmt.Printf("failed to get health from %s: %s", cluster, esErrorString(err)) return false } status[cluster] = st } table := printer.NewTable(conf, 3, 6) table.Addheaders("setting", "leader:"+leader, "follower:"+follower) table.Entries = [][]any{ {"Cluster Name", status[leader].ClusterName, status[follower].ClusterName}, {"Cluster Status", printer.Colorize(conf, status[leader].Status.Name, status[leader].Status.Name), printer.Colorize(conf, status[follower].Status.Name, status[follower].Status.Name), }, {"Active Shards", status[leader].ActiveShards, status[follower].ActiveShards, }, {"Active Primary Shards", status[leader].ActivePrimaryShards, status[follower].ActivePrimaryShards, }, {"Indicies", len(status[leader].Indices), len(status[follower].Indices), }, {"Nodes", status[leader].NumberOfNodes, status[follower].NumberOfNodes, }, } if err := table.Print(); err != nil { fmt.Println(err) return false } fmt.Println() 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]any{} for _, cluster := range []string{leader, follower} { ilm, err := conf.Clusters[cluster].ES().Ilm.ExplainLifecycle("_all"). OnlyManaged(true). Do(context.Background()) if err != nil { fmt.Printf("failed to get ilm status from %s: %s", cluster, esErrorString(err)) return false } failed[cluster] = map[string]any{} for name, ilmstate := range ilm.Indices { count := ilmstate.(*types.LifecycleExplainManaged).FailedStepRetryCount if count != nil && *count > 0 { failed[cluster][name] = *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 := printer.NewTable(conf, 2, len(failed[cluster])) table.Addheaders("ilm errors on "+which, "errors") for name, count := range failed[cluster] { table.Entries[idx] = []any{name, count} idx++ } table.Sort() if err := table.Print(); err != nil { fmt.Println(err) return false } } } 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). 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 := printer.NewTable(conf, 3, len(indexOnlyOnLeader)) table.Addheaders("index only on leader", "size", "docscount") for name, index := range indexOnlyOnLeader { name := printer.Colorize(conf, "red", name) table.Entries[idx] = []any{name, *index.DatasetSize, *index.DocsCount} idx++ } table.Sort() if err := table.Print(); err != nil { fmt.Println(err) return false } 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). 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 := printer.NewTable(conf, 3, len(orphaned)) table.Addheaders("orphaned index on follower", "size", "docscount") for name, index := range orphaned { name := printer.Colorize(conf, "red", name) table.Entries[idx] = []any{name, *index.DatasetSize, *index.DocsCount} idx++ } table.Sort() if err := table.Print(); err != nil { fmt.Println(err) return false } 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 := printer.NewTable(conf, 3, len(red)) table.Addheaders("red index on follower", "size", "docscount") for name, index := range red { name := printer.Colorize(conf, "red", name) table.Entries[idx] = []any{name, *index.DatasetSize, *index.DocsCount} idx++ } table.Sort() if err := table.Print(); err != nil { fmt.Println(err) return false } return false } func splitArg(arg string) (string, string) { parts := strings.Split(arg, ":") switch len(parts) { case 0: fallthrough case 1: return arg, "" default: return parts[0], parts[1] } }