/* 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" "encoding/json" "errors" "fmt" "log/slog" "codeberg.org/scip/esctl/pkg/cfg" "github.com/elastic/go-elasticsearch/v9/typedapi/types" "github.com/urfave/cli/v3" ) 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 ClusterList(conf *cfg.Config) error { table := NewTable(3, len(conf.Clusters)) table.Addheaders("cluster", "uri", "default") idx := 0 for name, cluster := range conf.Clusters { current := name == "default" || name == conf.CurrentCluster _, err := cluster.ES.Cluster.Health(). Header("content-type", "application/json"). Header("accept", "application/json"). Do(context.Background()) if err == nil { name = Colorize("green", name) } table.entries[idx] = []string{name, cluster.Uri, fmt.Sprintf("%t", current)} idx++ } table.Sort() table.PrintMarkdown() return nil } func ClusterStatus(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 { return fmt.Errorf("failed to get cluster health: %s", err) } slog.Debug("ES result", "cluster health", res) 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) } table := NewTable(2, 5) table.Addheaders(cluster, "status") table.entries = [][]string{ {"Cluster Name", Colorize(*&res.Status.Name, res.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)}, } table.PrintMarkdown() } return nil } func ClusterSettingsList(conf *cfg.Config) error { res, err := conf.DefaultCluster.ES.Cluster.GetSettings(). Header("content-type", "application/json"). Header("accept", "application/json"). Do(context.Background()) if err != nil { return fmt.Errorf("failed to get cluster settings: %s", err) } table := NewTable(2, 0) table.Addheaders("setting", "value") entries := [][]string{} settingshash := res.Persistent // == map[string]json.RawMessage switch { case conf.Transient: settingshash = res.Transient case conf.Default: settingshash = res.Defaults } for topic, val := range settingshash { data := map[string]map[string]any{} err := json.Unmarshal(val, &data) if err != nil { return fmt.Errorf("failed to unmarshall setting for topic %s: %s", topic, err) } for key, settings := range data { for setting, value := range settings { entries = append(entries, []string{ fmt.Sprintf("%s.%s.%s", topic, key, setting), fmt.Sprintf("%v", value), }) } } } table.entries = entries table.Sort() table.PrintMarkdown() return nil } func ClusterSettingsSet(conf *cfg.Config, args cli.Args) error { put := conf.Clusters[conf.CurrentCluster].ES.Cluster.PutSettings() for _, arg := range args.Slice() { setting, value := splitArg(arg) switch { case conf.Transient: message, err := json.Marshal(value) if err != nil { return fmt.Errorf("failed to marshall transient value <%v> to valid JSON: %s", value, err) } put.AddTransient(setting, message) case conf.Persistent: fallthrough default: message, err := json.Marshal(value) if err != nil { return fmt.Errorf("failed to marshall persistent value <%v> to valid JSON: %s", value, err) } put.AddPersistent(setting, message) } } _, err := put. Header("content-type", "application/json"). Header("accept", "application/json"). Do(context.Background()) if err != nil { return fmt.Errorf("failed to set settings: %s", err) } return nil }