mirror of
https://codeberg.org/scip/esctl.git
synced 2026-08-24 06:34:18 +02:00
Add nodes and cluster settings support (#9)
This commit is contained in:
T. von Dein
@@ -18,15 +18,14 @@ package es
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"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"
|
||||
"github.com/urfave/cli/v3"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -75,7 +74,7 @@ func ClusterCompare(conf *cfg.Config, leader, follower string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func List(conf *cfg.Config) error {
|
||||
func ClusterList(conf *cfg.Config) error {
|
||||
table := NewTable(2, len(conf.Clusters))
|
||||
table.Addheaders("cluster", "uri")
|
||||
|
||||
@@ -91,7 +90,7 @@ func List(conf *cfg.Config) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func Status(conf *cfg.Config) error {
|
||||
func ClusterStatus(conf *cfg.Config) error {
|
||||
clusters := []string{}
|
||||
|
||||
if conf.All {
|
||||
@@ -113,7 +112,7 @@ func Status(conf *cfg.Config) error {
|
||||
Header("accept", "application/json").
|
||||
Do(context.Background())
|
||||
if err != nil {
|
||||
log.Fatalf("Error getting health: %s", err)
|
||||
return fmt.Errorf("failed to getcluster health: %s", err)
|
||||
}
|
||||
|
||||
slog.Debug("ES result", "cluster health", res)
|
||||
@@ -135,340 +134,79 @@ func Status(conf *cfg.Config) error {
|
||||
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().
|
||||
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 {
|
||||
fmt.Printf("failed to get ccr stats from %s: %s", leader, err)
|
||||
return false
|
||||
return fmt.Errorf("failed to get cluster settings: %s", err)
|
||||
}
|
||||
|
||||
if len(stats.AutoFollowStats.AutoFollowedClusters) == 0 {
|
||||
// is not following anyone
|
||||
return true
|
||||
}
|
||||
table := NewTable(2, 0)
|
||||
table.Addheaders("setting", "value")
|
||||
entries := [][]string{}
|
||||
|
||||
if stats.AutoFollowStats.AutoFollowedClusters[0].ClusterName != "" {
|
||||
fmt.Println("leader/follower attribution is invalid, reverse cluster attribution and retry")
|
||||
return false
|
||||
}
|
||||
for topic, val := range res.Persistent {
|
||||
data := map[string]map[string]any{}
|
||||
|
||||
return true
|
||||
}
|
||||
err := json.Unmarshal(val, &data)
|
||||
|
||||
// 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
|
||||
return fmt.Errorf("failed to unmarshall setting for topic %s: %s", topic, err)
|
||||
}
|
||||
|
||||
status[cluster] = st
|
||||
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 := 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.entries = entries
|
||||
|
||||
table.Sort()
|
||||
table.PrintMarkdown()
|
||||
|
||||
if status[leader].Status.Name == "green" && status[follower].Status.Name == "green" {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
return nil
|
||||
}
|
||||
|
||||
// finds indices on both clusters which have ilm errors
|
||||
func findIlmErrors(conf *cfg.Config, leader, follower string) bool {
|
||||
failed := map[string]map[string]string{}
|
||||
func ClusterSettingsSet(conf *cfg.Config, args cli.Args) error {
|
||||
put := conf.Clusters[conf.CurrentCluster].ES.Cluster.PutSettings()
|
||||
|
||||
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
|
||||
}
|
||||
for _, arg := range args.Slice() {
|
||||
setting, value := splitArg(arg)
|
||||
|
||||
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())
|
||||
switch {
|
||||
case conf.Transient:
|
||||
message, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
continue // ignore it then
|
||||
return fmt.Errorf("failed to marshall transient value <%v> to valid JSON: %s", value, err)
|
||||
}
|
||||
|
||||
_, 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())
|
||||
put.AddTransient(setting, message)
|
||||
case conf.Persistent:
|
||||
fallthrough
|
||||
default:
|
||||
message, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
continue // ignore it then
|
||||
return fmt.Errorf("failed to marshall persistent value <%v> to valid JSON: %s", value, err)
|
||||
}
|
||||
|
||||
_, defined := res[name]
|
||||
if !defined {
|
||||
// json response map didn't contain the index
|
||||
continue
|
||||
}
|
||||
|
||||
orphaned[name] = index
|
||||
put.AddPersistent(setting, message)
|
||||
}
|
||||
}
|
||||
|
||||
if len(orphaned) == 0 {
|
||||
return true
|
||||
_, 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)
|
||||
}
|
||||
|
||||
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
|
||||
return nil
|
||||
}
|
||||
|
||||
55
pkg/es/node.go
Normal file
55
pkg/es/node.go
Normal file
@@ -0,0 +1,55 @@
|
||||
/*
|
||||
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 <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
package es
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
|
||||
"codeberg.org/scip/esctl/pkg/cfg"
|
||||
)
|
||||
|
||||
func NodeList(conf *cfg.Config) error {
|
||||
// get nodes
|
||||
nodes, err := conf.DefaultCluster.ES.Cat.Nodes().Do(context.Background())
|
||||
if err != nil {
|
||||
return fmt.Errorf("Error getting nodes: %s", err)
|
||||
}
|
||||
|
||||
slog.Debug("ES result", "nodes", nodes)
|
||||
|
||||
table := NewTable(7, len(nodes))
|
||||
table.Addheaders("name", "ip", "load1m", "load5m", "load15m", "ram %", "heap %")
|
||||
|
||||
for idx, node := range nodes {
|
||||
table.entries[idx] = []string{
|
||||
*node.Name,
|
||||
*node.Ip,
|
||||
*node.Load1M,
|
||||
*node.Load5M,
|
||||
*node.Load15M,
|
||||
node.RamPercent.(string),
|
||||
node.HeapPercent.(string),
|
||||
}
|
||||
}
|
||||
|
||||
table.Sort()
|
||||
table.PrintMarkdown()
|
||||
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user