From a71be5001a255d27625c2fb0d1e2666f96e38860 Mon Sep 17 00:00:00 2001 From: Thomas von Dein Date: Tue, 5 May 2026 12:07:31 +0200 Subject: [PATCH] add missing utilities.go --- pkg/es/utilities.go | 379 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 379 insertions(+) create mode 100644 pkg/es/utilities.go diff --git a/pkg/es/utilities.go b/pkg/es/utilities.go new file mode 100644 index 0000000..efb5b94 --- /dev/null +++ b/pkg/es/utilities.go @@ -0,0 +1,379 @@ +/* +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" + "github.com/elastic/go-elasticsearch/v9/typedapi/cluster/health" + "github.com/elastic/go-elasticsearch/v9/typedapi/types" +) + +// 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 +} + +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] + } +}