shards, aliases and filtering (#18)

This commit is contained in:
T. von Dein
2026-05-19 13:58:08 +02:00
parent 565d92b491
commit 58d94a4f56
9 changed files with 537 additions and 21 deletions

4
.gitignore vendored
View File

@@ -26,3 +26,7 @@ go.work.sum
.env
esctl
*.sh
*.json
*.log

View File

@@ -18,6 +18,7 @@ package cmd
import (
"context"
"errors"
"fmt"
"codeberg.org/scip/esctl/pkg/cfg"
@@ -39,6 +40,8 @@ func Index(conf *cfg.Config) *cli.Command {
IndexDelete(conf),
IndexClose(conf),
IndexAllocation(conf),
IndexModify(conf),
IndexAlias(conf),
},
}
}
@@ -68,6 +71,12 @@ func IndexList(conf *cfg.Config) *cli.Command {
Destination: &conf.Failed,
Aliases: []string{"r"},
},
&cli.StringSliceFlag{
Name: "filter",
Usage: "show only indicies matching the filter",
Destination: &conf.Filter,
Aliases: []string{"F"},
},
},
Action: func(ctx context.Context, cmd *cli.Command) error {
@@ -98,6 +107,11 @@ func IndexAllocation(conf *cfg.Config) *cli.Command {
},
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
if index == "" {
return errors.New("no index specified")
}
return es.IndexAllocation(conf, cmd.Args().Get(0))
},
}
@@ -110,6 +124,11 @@ func IndexShow(conf *cfg.Config) *cli.Command {
Usage: "show details about an index",
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
if index == "" {
return errors.New("no index specified")
}
return es.IndexShow(conf, cmd.Args().Get(0))
},
@@ -180,6 +199,11 @@ func IndexDelete(conf *cfg.Config) *cli.Command {
Usage: "delete an index",
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
if index == "" {
return errors.New("no index specified")
}
return es.IndexDelete(conf, cmd.Args().Get(0))
},
}
@@ -191,7 +215,38 @@ func IndexClose(conf *cfg.Config) *cli.Command {
Usage: "close an index",
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
if index == "" {
return errors.New("no index specified")
}
return es.IndexClose(conf, cmd.Args().Get(0))
},
}
}
func IndexModify(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "modify",
Usage: "modify an index",
UsageText: "modify <index[,index,...]|_all>",
Flags: []cli.Flag{
&cli.IntFlag{
Name: "replicas",
Usage: "number of replicas",
Destination: &conf.Replicas,
Aliases: []string{"r"},
},
},
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
if index == "" {
return errors.New("no index specified")
}
return es.IndexModify(conf, index)
},
}
}

84
cmd/index_alias.go Normal file
View File

@@ -0,0 +1,84 @@
/*
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 IndexAlias(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "alias",
Aliases: []string{"a"},
Usage: "manage index aliases",
Commands: []*cli.Command{
IndexAliasCreate(conf),
IndexAliasList(conf),
//IndexAliasShow(conf),
//IndexAliasDelete(conf),
//IndexAliasAdd(conf), // see https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-update-aliases
},
}
}
func IndexAliasCreate(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "create",
Aliases: []string{"+"},
Usage: "create an index alias",
UsageText: "create <index> <alias>",
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
alias := cmd.Args().Get(1)
if index == "" || alias == "" {
return errors.New("no index or alias specified")
}
return es.IndexAliasCreate(conf, index, alias)
},
}
}
func IndexAliasList(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "list",
Aliases: []string{"ls"},
Usage: "list index aliases",
Flags: []cli.Flag{
&cli.StringSliceFlag{
Name: "filter",
Usage: "show only aliases for indicies matching the filter",
Destination: &conf.Filter,
Aliases: []string{"F"},
},
},
Action: func(ctx context.Context, cmd *cli.Command) error {
return es.IndexAliasList(conf)
},
}
}

View File

@@ -73,10 +73,11 @@ func Main() int {
Commands: []*cli.Command{
Search(conf),
Index(conf),
Snapshot(conf),
Cluster(conf),
Ccr(conf),
Index(conf),
Shard(conf),
Snapshot(conf),
Node(conf),
Doc(conf),
Repl(conf),

96
cmd/shards.go Normal file
View File

@@ -0,0 +1,96 @@
/*
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 Shard(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "shard",
Aliases: []string{"s"},
Usage: "manage shards",
Commands: []*cli.Command{
ShardList(conf),
ShardShow(conf),
},
}
}
func ShardList(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "list",
Aliases: []string{"ls"},
Usage: "list shards",
Flags: []cli.Flag{
&cli.IntFlag{
Name: "max",
Usage: "max number of items to process",
Destination: &conf.MaxItems,
Aliases: []string{"m"},
},
&cli.BoolFlag{
Name: "reds",
Usage: "include only failed shards",
Destination: &conf.Failed,
Aliases: []string{"r"},
},
&cli.BoolFlag{
Name: "primaries",
Usage: "show only primary shards",
Destination: &conf.Primary,
Aliases: []string{"p"},
},
&cli.BoolFlag{
Name: "verbose",
Usage: "show node name and ip as well",
Destination: &conf.Verbose,
Aliases: []string{"v"},
},
},
Action: func(ctx context.Context, cmd *cli.Command) error {
return es.ShardList(conf)
},
}
}
func ShardShow(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "show",
Aliases: []string{"sh"},
Usage: "show details about a shard",
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
if index == "" {
return errors.New("no index specified")
}
return es.ShardShow(conf, index)
},
}
}

View File

@@ -31,7 +31,7 @@ import (
)
const (
Version string = `v0.0.10`
Version string = `v0.0.11`
)
type Cluster struct {

View File

@@ -20,11 +20,13 @@ import (
"context"
"fmt"
"log/slog"
"regexp"
"strconv"
"strings"
"time"
"codeberg.org/scip/esctl/pkg/cfg"
"github.com/elastic/go-elasticsearch/v9/typedapi/cat/indices"
"github.com/elastic/go-elasticsearch/v9/typedapi/esdsl"
"github.com/elastic/go-elasticsearch/v9/typedapi/types/enums/healthstatus"
)
@@ -48,6 +50,24 @@ func IndexNames(conf *cfg.Config) ([]string, error) {
return indices, nil
}
func filterIndices(conf *cfg.Config, list indices.Response) indices.Response {
if len(conf.Filter) == 0 {
return list
}
// we support just one filter here, for now
filter := *regexp.MustCompile(conf.Filter[0])
newlist := indices.Response{}
for _, index := range list {
if filter.MatchString(*index.Index) {
newlist = append(newlist, index)
}
}
return newlist
}
func IndexList(conf *cfg.Config) error {
cat := conf.DefaultCluster.ES.Cat.Indices().
// we need to add custom request headers, required for older ES instances
@@ -65,7 +85,9 @@ func IndexList(conf *cfg.Config) error {
slog.Debug("ES result", "indicies", res)
size := len(res)
list := filterIndices(conf, res)
size := len(list)
if conf.MaxItems > 0 {
if size > conf.MaxItems {
@@ -76,7 +98,7 @@ func IndexList(conf *cfg.Config) error {
table := NewTable(3, size)
table.Addheaders("name", "size", "docscount")
for idx, index := range res {
for idx, index := range list {
name := Colorize(*index.Health, *index.Index)
table.entries[idx] = []string{name, *index.DatasetSize, *index.DocsCount}
@@ -95,10 +117,6 @@ func IndexList(conf *cfg.Config) error {
}
func IndexShow(conf *cfg.Config, index string) error {
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").
@@ -188,10 +206,6 @@ func IndexCreate(conf *cfg.Config, index string, mappings []string) error {
}
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").
@@ -218,16 +232,11 @@ func IndexClose(conf *cfg.Config, index string) error {
}
func IndexAllocation(conf *cfg.Config, index string) error {
if index == "" {
return fmt.Errorf("no index specified")
}
alloc := conf.DefaultCluster.ES.Cluster.AllocationExplain().
res, err := conf.DefaultCluster.ES.Cluster.AllocationExplain().
Index(index).
Primary(conf.Primary).
Shard(conf.Shards)
res, err := alloc.Header("content-type", "application/json").
Shard(conf.Shards).
Header("content-type", "application/json").
Header("accept", "application/json").
Do(context.Background())
if err != nil {
@@ -265,3 +274,19 @@ func IndexAllocation(conf *cfg.Config, index string) error {
return nil
}
func IndexModify(conf *cfg.Config, index string) error {
settings := esdsl.NewIndexSettings().NumberOfReplicas(strconv.Itoa(conf.Replicas))
_, err := conf.DefaultCluster.ES.Indices.PutSettings().
Indices(index).
Index(settings).
Header("content-type", "application/json").
Header("accept", "application/json").
Do(context.Background())
if err != nil {
return fmt.Errorf("failed to modify index settings: %s", err)
}
return nil
}

101
pkg/es/index_alias.go Normal file
View File

@@ -0,0 +1,101 @@
/*
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"
"regexp"
"strings"
"codeberg.org/scip/esctl/pkg/cfg"
)
// FIXME: add filter support, see IndexCreate mapping
func IndexAliasCreate(conf *cfg.Config, index, alias string) error {
create := conf.DefaultCluster.ES.Indices.PutAlias(index, alias).
Header("content-type", "application/json").
Header("accept", "application/json")
res, err := create.Do(context.Background())
slog.Debug("create alias", "result", res)
if err != nil {
return fmt.Errorf("failed to create index alias: %s", err)
}
return nil
}
func IndexAliasList(conf *cfg.Config) error {
filter := regexp.Regexp{}
if len(conf.Filter) > 0 {
// we support just one filter here, for now
filter = *regexp.MustCompile(conf.Filter[0])
}
create := conf.DefaultCluster.ES.Indices.GetAlias().
Index("_all").
Header("content-type", "application/json").
Header("accept", "application/json")
res, err := create.Do(context.Background())
if err != nil {
return fmt.Errorf("failed to list index aliases: %s", err)
}
slog.Debug("aliases list", "result", res)
aliaslist := map[string][]string{} // index => []aliases
for index, aliases := range res {
if len(conf.Filter) > 0 {
if !filter.MatchString(index) {
continue
}
}
for alias := range aliases.Aliases {
aliaslist[index] = append(aliaslist[index], alias)
}
}
table := NewTable(2, len(aliaslist))
table.Addheaders("index", "alias")
idx := 0
for index, aliases := range aliaslist {
table.entries[idx] = []string{
index,
strings.Join(aliases, ","),
}
idx++
}
table.Sort()
if err := table.PrintMarkdown(); err != nil {
return err
}
return nil
}

150
pkg/es/shard.go Normal file
View File

@@ -0,0 +1,150 @@
/*
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"
"github.com/elastic/go-elasticsearch/v9/typedapi/cat/shards"
)
func colorzizeShard(state, name string) string {
color := "red" // in case of RELOCATING and UNASSIGNED
switch state {
case "STARTED":
color = "green"
case "INITIALIZING":
color = "yellow"
}
return Colorize(color, name)
}
func resolvePrirep(state string) string {
switch state {
case "p":
return "primary"
}
return "replica"
}
func filterShards(conf *cfg.Config, shardlist shards.Response) shards.Response {
filtered := shards.Response{}
size := len(shardlist)
if conf.MaxItems > 0 {
if size > conf.MaxItems {
size = conf.MaxItems
}
}
for idx, shard := range shardlist {
if conf.Failed {
if *shard.State == "STARTED" {
continue
}
}
if conf.Primary {
if *shard.Prirep != "p" {
continue
}
}
if idx == size-1 {
break
}
filtered = append(filtered, shard)
}
return filtered
}
func ShardList(conf *cfg.Config) error {
res, err := conf.DefaultCluster.ES.Cat.Shards().
Header("content-type", "application/json").
Header("accept", "application/json").
Do(context.Background())
if err != nil {
return fmt.Errorf("failed to get shards: %s", err)
}
shardlist := filterShards(conf, res)
slog.Debug("ES result", "shards", res)
return printShards(conf, shardlist)
}
func printShards(conf *cfg.Config, shardlist shards.Response) error {
headers := []string{"index", "shard", "is primary", "store", "dataset", "docs"}
if conf.Verbose {
headers = append(headers, "node", "ip")
}
table := NewTable(len(headers), len(shardlist))
table.Addheaders(headers...)
for idx, shard := range shardlist {
name := colorzizeShard(*shard.State, *shard.Index)
table.entries[idx] = []string{
name,
*shard.Shard,
resolvePrirep(*shard.Prirep),
*shard.Store,
*shard.Dataset,
*shard.Docs,
}
if conf.Verbose {
table.entries[idx] = append(table.entries[idx],
*shard.Node,
*shard.Ip,
)
}
}
table.Sort()
if err := table.PrintMarkdown(); err != nil {
return err
}
return nil
}
func ShardShow(conf *cfg.Config, index string) error {
res, err := conf.DefaultCluster.ES.Cat.Shards().Index(index).
Header("content-type", "application/json").
Header("accept", "application/json").
Do(context.Background())
if err != nil {
return fmt.Errorf("failed to get shards: %s", err)
}
slog.Debug("ES result", "shards", res)
conf.Verbose = true
return printShards(conf, res)
}