Compare commits

..

6 Commits
0.0.2 ... 0.0.4

Author SHA1 Message Date
T. von Dein
a169d23b90 add index create,delete, enhance index show, catch empty index arg 2026-04-29 13:44:21 +02:00
T. von Dein
f89a09d1d2 add support for a standard config in ~/.config/esctl/config.yaml (#7) 2026-04-29 10:26:53 +02:00
T. von Dein
01853c3299 add more cluster comparision funcs (#6) 2026-04-29 09:56:31 +02:00
d6a96ee61f upd help 2026-04-28 13:46:28 +02:00
T. von Dein
f27f9157fb add "cluster ls", put "status" into "cluster status" (#5) 2026-04-28 13:43:13 +02:00
T. von Dein
3cbc67567b add "cluster compare" command and add 1st check (unsync indices) (#2) 2026-04-27 13:05:30 +02:00
12 changed files with 850 additions and 111 deletions

View File

@@ -12,17 +12,21 @@ USAGE:
esctl [global options] [command [command options]]
VERSION:
v0.0.1
v0.0.3
COMMANDS:
health show ES health
search, / search within an index
help, h Shows a list of commands or help for one command
search, / search within an index
index, i manage indicies
snapshot, snap manage snapshots
cluster, c manage cluster[s]
help, h Shows a list of commands or help for one command
GLOBAL OPTIONS:
--debug, -d enable debugging [$ES_DEBUG]
--help, -h show help
--version, -v print the version
--debug, -d enable debugging [$ES_DEBUG]
--config string, -c string config file [$ES_CONFIG]
--cluster string, -C string cluster alias to work with
--help, -h show help
--version, -v print the version
```
Configure `esctl` with environment variables:
@@ -31,6 +35,27 @@ Configure `esctl` with environment variables:
- `ES_USER`: username
- `ES_PASS`: password
Or create a config file such as this:
```yaml
clusters:
default:
uri: https://es.foo.bar:9200/
user: elastic
pass: 123456
other:
uri: https://myes.foo:9200/
user: elastic
pass: asdasdasd
```
and specify it with `-c configfile`. You may also put clusters into a
default config file in `~/.config/esctl/config.yaml`. In this case you
can omit `-c ...`.
If you want to work on a specific cluster, specify its name with the
global `-C` option.
## Introduction
FIXME

View File

@@ -1,2 +1,5 @@
- [Go client docs](https://www.elastic.co/docs/reference/elasticsearch/clients/go/typed-api)
- [ES API docs](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-get)
- Fix index names custom completion
- add cluster default <name> which would add a flag to the config, so that no -C is needed subsequently

121
cmd/cluster.go Normal file
View File

@@ -0,0 +1,121 @@
/*
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 cmd
import (
"context"
"errors"
"codeberg.org/scip/esctl/pkg/cfg"
"codeberg.org/scip/esctl/pkg/es"
"github.com/urfave/cli/v3"
)
func Cluster(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "cluster",
Aliases: []string{"c"},
Usage: "manage cluster[s]",
Commands: []*cli.Command{
Compare(conf),
Status(conf),
List(conf),
},
}
}
func List(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "list",
Usage: "list configured clusters",
Aliases: []string{"ls"},
Action: func(ctx context.Context, cmd *cli.Command) error {
if err := es.List(conf); err != nil {
return err
}
return nil
},
}
}
func Status(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "status",
Usage: "show cluster status",
Aliases: []string{"s"},
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "all",
Usage: "show status of all clusters",
Destination: &conf.All,
Aliases: []string{"a"},
},
},
Action: func(ctx context.Context, cmd *cli.Command) error {
if err := es.Status(conf); err != nil {
return err
}
return nil
},
}
}
func Compare(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "compare",
Aliases: []string{"c"},
Usage: "compare cluster[s] (yaml config with 2 clusters required)",
Flags: []cli.Flag{
&cli.StringFlag{
Name: "exclude",
Usage: "regexp of indicies to exclude",
Destination: &conf.Exclude,
Aliases: []string{"e"},
},
},
Action: func(ctx context.Context, cmd *cli.Command) error {
leader := cmd.Args().Get(0)
follower := cmd.Args().Get(1)
if leader == "" || follower == "" {
return errors.New("no leader and follower aliases specified")
}
_, hasLeader := conf.Clusters[leader]
_, hasFollower := conf.Clusters[follower]
if !hasLeader || !hasFollower {
return errors.New("either leader or follower alias not configured")
}
if err := es.ClusterCompare(conf, leader, follower); err != nil {
return err
}
return nil
},
}
}

View File

@@ -1,42 +0,0 @@
/*
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 cmd
import (
"context"
"codeberg.org/scip/esctl/pkg/cfg"
"codeberg.org/scip/esctl/pkg/es"
"github.com/urfave/cli/v3"
)
func Status(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "status",
Usage: "show ES status",
Aliases: []string{"s"},
Action: func(ctx context.Context, cmd *cli.Command) error {
if err := es.Health(conf); err != nil {
return err
}
return nil
},
}
}

View File

@@ -18,6 +18,7 @@ package cmd
import (
"context"
"fmt"
"codeberg.org/scip/esctl/pkg/cfg"
"codeberg.org/scip/esctl/pkg/es"
@@ -34,6 +35,8 @@ func Index(conf *cfg.Config) *cli.Command {
Commands: []*cli.Command{
IndexList(conf),
IndexShow(conf),
IndexCreate(conf),
IndexDelete(conf),
},
}
}
@@ -88,5 +91,76 @@ func IndexShow(conf *cfg.Config) *cli.Command {
return nil
},
// FIXME: doesn't work at all
// FIXME: also it would ONLY work if the user uses env vars, -C would not be
// there when the completion output is being generated
ShellComplete: func(ctx context.Context, cmd *cli.Command) {
if cmd.NArg() > 0 {
return
}
indices, err := es.IndexNames(conf)
if err != nil {
return
}
for _, index := range indices {
fmt.Println(index)
}
},
}
}
func IndexCreate(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "create",
Aliases: []string{"+"},
Usage: "create a new index",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: "wait",
Usage: "wait for active shards",
Destination: &conf.Wait,
Aliases: []string{"w"},
},
&cli.IntFlag{
Name: "shards",
Usage: "number of shards to create",
Destination: &conf.Shards,
Aliases: []string{"s"},
},
&cli.IntFlag{
Name: "replicas",
Usage: "number of replicas to create",
Destination: &conf.Replicas,
Aliases: []string{"r"},
},
},
Action: func(ctx context.Context, cmd *cli.Command) error {
if err := es.IndexCreate(conf, cmd.Args().Get(0)); err != nil {
return err
}
return nil
},
}
}
func IndexDelete(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "delete",
Aliases: []string{"rm"},
Usage: "delete an index",
Action: func(ctx context.Context, cmd *cli.Command) error {
if err := es.IndexDelete(conf, cmd.Args().Get(0)); err != nil {
return err
}
return nil
},
}
}

View File

@@ -72,15 +72,20 @@ func Main() int {
},
Commands: []*cli.Command{
Status(conf),
Search(conf),
Index(conf),
Snapshot(conf),
Cluster(conf),
},
Before: func(ctx context.Context, cmd *cli.Command) (context.Context, error) {
if err := conf.Init(); err != nil {
Finish(err)
if len(os.Args) > 1 {
return nil, err
} else {
fmt.Println(cmd.UsageText)
return nil, nil
}
}
log.Init(conf)

1
go.mod
View File

@@ -40,6 +40,7 @@ require (
go.opentelemetry.io/otel v1.35.0 // indirect
go.opentelemetry.io/otel/metric v1.35.0 // indirect
go.opentelemetry.io/otel/trace v1.35.0 // indirect
golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f // indirect
golang.org/x/sys v0.42.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)

2
go.sum
View File

@@ -54,6 +54,8 @@ go.opentelemetry.io/otel/metric v1.35.0 h1:0znxYu2SNyuMSQT4Y9WDWej0VpcsxkuklLa4/
go.opentelemetry.io/otel/metric v1.35.0/go.mod h1:nKVFgxBZ2fReX6IlyW28MgZojkoAkJGaE8CpgeAU3oE=
go.opentelemetry.io/otel/trace v1.35.0 h1:dPpEfJu1sDIqruz7BHFG3c7528f6ddfSWfFDVt/xgMs=
go.opentelemetry.io/otel/trace v1.35.0/go.mod h1:WUk7DtFp1Aw2MkvqGdwiXYDZZNvA/1J8o6xRXLrIkyc=
golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f h1:W3F4c+6OLc6H2lb//N1q4WpJkhzJCK5J6kUi1NTVXfM=
golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f/go.mod h1:J1xhfL/vlindoeF/aINzNzt2Bket5bjo9sdOYzOsU80=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=

View File

@@ -30,7 +30,7 @@ import (
)
const (
Version string = `v0.0.2`
Version string = `v0.0.4`
)
type Cluster struct {
@@ -44,10 +44,14 @@ type Config struct {
Debug bool // -d
Clusters map[string]*Cluster
DefaultCluster *Cluster
From, To, MaxItems int
Index string
Filter []string
Failed, Partials bool
Index string // index: -i
Failed, Partials bool // index: flags
Shards, Replicas int // index create: -s -r
Wait bool // index create: -w
From, To, MaxItems int // search: flags
Filter []string // search: -F
Exclude string // cluster compare: -e (regexp)
All bool // cluster status: -a
}
func NewConfig() *Config {
@@ -55,11 +59,17 @@ func NewConfig() *Config {
}
func (conf *Config) Init() error {
if conf.ConfigFile != "" {
DefaultConfig := os.Getenv("HOME") + "/.config/esctl/config.yaml"
switch {
case fileExists(DefaultConfig):
conf.ConfigFile = DefaultConfig
fallthrough
case conf.ConfigFile != "":
if err := conf.LoadConfig(); err != nil {
return err
}
} else {
default:
if err := conf.LoadEnv(); err != nil {
return err
}
@@ -163,3 +173,14 @@ func (conf *Config) SetupES() error {
return nil
}
func fileExists(filename string) bool {
info, err := os.Stat(filename)
if err != nil {
// return false on any error
return false
}
return !info.IsDir()
}

View File

@@ -1,50 +0,0 @@
/*
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"
"log/slog"
"codeberg.org/scip/esctl/pkg/cfg"
)
func Health(conf *cfg.Config) error {
res, err := conf.DefaultCluster.ES.Cluster.Health().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.headers = []string{bold("SETTING"), bold("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
}

474
pkg/es/cluster.go Normal file
View File

@@ -0,0 +1,474 @@
/*
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"
"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.Sort()
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
}

View File

@@ -20,13 +20,44 @@ import (
"context"
"fmt"
"log/slog"
"strconv"
"time"
"codeberg.org/scip/esctl/pkg/cfg"
"github.com/elastic/go-elasticsearch/v9/typedapi/esdsl"
"github.com/elastic/go-elasticsearch/v9/typedapi/types/enums/healthstatus"
)
type Settings struct {
wait_for_active_shards string
number_of_shards int
number_of_replicas int
}
// used for completion
func IndexNames(conf *cfg.Config) ([]string, error) {
res, err := conf.DefaultCluster.ES.Cat.Indices().
Header("content-type", "application/json").
Header("accept", "application/json").
Do(context.Background())
if err != nil {
return nil, fmt.Errorf("Error getting indicies: %s", err)
}
indices := make([]string, len(res))
for idx, index := range res {
indices[idx] = *index.Index
}
return indices, nil
}
func IndexList(conf *cfg.Config) error {
cat := conf.DefaultCluster.ES.Cat.Indices()
cat := conf.DefaultCluster.ES.Cat.Indices().
// we need to add custom request headers, required for older ES instances
Header("content-type", "application/json").
Header("accept", "application/json")
if conf.Failed {
cat = cat.Health(healthstatus.Red)
@@ -67,14 +98,88 @@ func IndexList(conf *cfg.Config) error {
}
func IndexShow(conf *cfg.Config, index string) error {
res, err := conf.DefaultCluster.ES.Indices.Get(index).Do(context.Background())
if index == "" {
return fmt.Errorf("no index specified")
}
res, err := conf.DefaultCluster.ES.Indices.Get(index).
// we need to add custom request headers, required for older ES instances
Header("content-type", "application/json").
Header("accept", "application/json").
Do(context.Background())
if err != nil {
return fmt.Errorf("failed to get index: %s", err)
}
slog.Debug("ES result", "index", res)
// FIXME: add table printer like SnapshotShow
table := NewTable(2, 5)
table.Addheaders("field", "value")
ts, err := strconv.ParseInt(res[index].Settings.Index.CreationDate.(string), 10, 64)
if err != nil {
ts = 0
}
created := time.Unix(ts/1000, 0)
table.entries = [][]string{
{"name", index},
{"replicas", *res[index].Settings.Index.NumberOfReplicas},
{"shards", *res[index].Settings.Index.NumberOfShards},
{"created", created.Format("2006-01-02 15:04:05")},
{"uuid", *res[index].Settings.Index.Uuid},
}
table.PrintMarkdown()
return nil
}
func IndexCreate(conf *cfg.Config, index string) error {
if index == "" {
return fmt.Errorf("no index specified")
}
settings := esdsl.NewIndexSettings()
create := conf.DefaultCluster.ES.Indices.Create(index).
Header("content-type", "application/json").
Header("accept", "application/json")
if conf.Wait {
create.WaitForActiveShards("all")
}
if conf.Shards > 0 {
settings = settings.NumberOfShards(strconv.Itoa(conf.Shards))
}
if conf.Replicas > 0 {
settings = settings.NumberOfReplicas(strconv.Itoa(conf.Replicas))
}
_, err := create.Settings(settings).
Do(context.Background())
if err != nil {
return fmt.Errorf("failed to create index: %s", err)
}
return nil
}
func IndexDelete(conf *cfg.Config, index string) error {
if index == "" {
return fmt.Errorf("no index specified")
}
_, err := conf.DefaultCluster.ES.Indices.Delete(index).
Header("content-type", "application/json").
Header("accept", "application/json").
Do(context.Background())
if err != nil {
return fmt.Errorf("failed to delete index: %s", err)
}
return nil
}