mirror of
https://codeberg.org/scip/esctl.git
synced 2026-08-24 22:24:21 +02:00
add alias+datastream rollover, and auto rollover on index tpl change (#40)
fixes #37
This commit is contained in:
@@ -37,6 +37,7 @@ func Datastream(conf *cfg.Config) *cli.Command {
|
||||
DatastreamShow(conf),
|
||||
DatastreamCreate(conf),
|
||||
DatastreamDelete(conf),
|
||||
DatastreamRollover(conf),
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -138,3 +139,61 @@ func DatastreamDelete(conf *cfg.Config) *cli.Command {
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func DatastreamRollover(conf *cfg.Config) *cli.Command {
|
||||
return &cli.Command{
|
||||
Name: "rollover",
|
||||
Aliases: []string{"roll"},
|
||||
Usage: "roll over a data stream",
|
||||
UsageText: "rollover <data stream>",
|
||||
|
||||
ShellComplete: func(ctx context.Context, cmd *cli.Command) {
|
||||
complete(cmd, Cindex)
|
||||
},
|
||||
|
||||
Flags: []cli.Flag{
|
||||
&cli.BoolFlag{
|
||||
Name: "wait",
|
||||
Usage: "wait for active shards",
|
||||
Destination: &conf.Wait,
|
||||
Aliases: []string{"w"},
|
||||
},
|
||||
&cli.BoolFlag{
|
||||
Name: "dry-run",
|
||||
Usage: "checks conditions but does not perform a rollover",
|
||||
Destination: &conf.Wait,
|
||||
Aliases: []string{"n"},
|
||||
},
|
||||
&cli.StringFlag{
|
||||
Name: "max-age",
|
||||
Usage: "roll over after max age (eg: 7d, 2m, 8h)",
|
||||
Destination: &conf.MaxAge,
|
||||
},
|
||||
&cli.IntFlag{
|
||||
Name: "max-docs",
|
||||
Usage: "roll over after max docs",
|
||||
Destination: &conf.MaxDocs,
|
||||
},
|
||||
&cli.IntFlag{
|
||||
Name: "max-primary-shard-size",
|
||||
Usage: "roll over when primary shard size reaches size",
|
||||
Destination: &conf.MaxShardSize,
|
||||
},
|
||||
&cli.IntFlag{
|
||||
Name: "max-primary-shard-docs",
|
||||
Usage: "roll over after max docs in primary shard reached",
|
||||
Destination: &conf.MaxShardDocs,
|
||||
},
|
||||
},
|
||||
|
||||
Action: func(ctx context.Context, cmd *cli.Command) error {
|
||||
alias := cmd.Args().Get(0)
|
||||
|
||||
if alias == "" {
|
||||
return errors.New("no data stream specified")
|
||||
}
|
||||
|
||||
return es.DatastreamRollover(conf, alias)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,8 +36,8 @@ func IndexAlias(conf *cfg.Config) *cli.Command {
|
||||
IndexAliasCreate(conf),
|
||||
IndexAliasList(conf),
|
||||
IndexAliasDelete(conf),
|
||||
// FIXME: implement IndexAliasShow + IndexAliasAdd
|
||||
//IndexAliasShow(conf),
|
||||
IndexAliasRollover(conf),
|
||||
// FIXME: implement IndexAliasAdd
|
||||
//IndexAliasAdd(conf), // see https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-indices-update-aliases
|
||||
},
|
||||
}
|
||||
@@ -91,6 +91,64 @@ func IndexAliasDelete(conf *cfg.Config) *cli.Command {
|
||||
}
|
||||
}
|
||||
|
||||
func IndexAliasRollover(conf *cfg.Config) *cli.Command {
|
||||
return &cli.Command{
|
||||
Name: "rollover",
|
||||
Aliases: []string{"roll"},
|
||||
Usage: "roll over an index alias",
|
||||
UsageText: "rollover <alias>",
|
||||
|
||||
ShellComplete: func(ctx context.Context, cmd *cli.Command) {
|
||||
complete(cmd, Cindex)
|
||||
},
|
||||
|
||||
Flags: []cli.Flag{
|
||||
&cli.BoolFlag{
|
||||
Name: "wait",
|
||||
Usage: "wait for active shards",
|
||||
Destination: &conf.Wait,
|
||||
Aliases: []string{"w"},
|
||||
},
|
||||
&cli.BoolFlag{
|
||||
Name: "dry-run",
|
||||
Usage: "checks conditions but does not perform a rollover",
|
||||
Destination: &conf.Wait,
|
||||
Aliases: []string{"n"},
|
||||
},
|
||||
&cli.StringFlag{
|
||||
Name: "max-age",
|
||||
Usage: "roll over after max age (eg: 7d, 2m, 8h)",
|
||||
Destination: &conf.MaxAge,
|
||||
},
|
||||
&cli.IntFlag{
|
||||
Name: "max-docs",
|
||||
Usage: "roll over after max docs",
|
||||
Destination: &conf.MaxDocs,
|
||||
},
|
||||
&cli.IntFlag{
|
||||
Name: "max-primary-shard-size",
|
||||
Usage: "roll over when primary shard size reaches size",
|
||||
Destination: &conf.MaxShardSize,
|
||||
},
|
||||
&cli.IntFlag{
|
||||
Name: "max-primary-shard-docs",
|
||||
Usage: "roll over after max docs in primary shard reached",
|
||||
Destination: &conf.MaxShardDocs,
|
||||
},
|
||||
},
|
||||
|
||||
Action: func(ctx context.Context, cmd *cli.Command) error {
|
||||
alias := cmd.Args().Get(0)
|
||||
|
||||
if alias == "" {
|
||||
return errors.New("no alias specified")
|
||||
}
|
||||
|
||||
return es.IndexAliasRollover(conf, alias)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func IndexAliasList(conf *cfg.Config) *cli.Command {
|
||||
return &cli.Command{
|
||||
Name: "list",
|
||||
|
||||
@@ -113,6 +113,12 @@ https://www.elastic.co/docs/reference/elasticsearch/index-settings
|
||||
Destination: &conf.AutoCreate,
|
||||
Aliases: []string{"a"},
|
||||
},
|
||||
&cli.BoolFlag{
|
||||
Name: "rollover",
|
||||
Usage: "automatically rollover associated aliases",
|
||||
Destination: &conf.Rollover,
|
||||
Aliases: []string{"R"},
|
||||
},
|
||||
&cli.StringFlag{
|
||||
Name: "mode",
|
||||
Usage: "index mode, one of: standard, timeseries, logsdb or lookup",
|
||||
@@ -171,9 +177,11 @@ https://www.elastic.co/docs/reference/elasticsearch/index-settings
|
||||
return fmt.Errorf("no name specified")
|
||||
}
|
||||
|
||||
if conf.Mode != "" {
|
||||
if !slices.Contains([]string{"standard", "timeseries", "logsdb", "lookup"}, conf.Mode) {
|
||||
return errors.New("mode must be one of: standard, timeseries, logsdb or lookup")
|
||||
}
|
||||
}
|
||||
|
||||
mappings := args.Slice()[1:]
|
||||
|
||||
|
||||
@@ -54,6 +54,8 @@ type Config struct {
|
||||
Output string // -o <mode>
|
||||
Clusters map[string]*Cluster
|
||||
DefaultCluster *Cluster
|
||||
HaveJQ bool // determined at runtime by ourselfes
|
||||
|
||||
Index string // index: -i
|
||||
Failed, Partials bool // index: flags
|
||||
Shards, Replicas int // index create+allocation: -s -r
|
||||
@@ -72,6 +74,7 @@ type Config struct {
|
||||
AutoCreate bool // index template create: -a
|
||||
Mode string // index template create: -a
|
||||
Retention string // index template create: -r
|
||||
Rollover bool // index template create: -R
|
||||
|
||||
From, To, MaxItems int // search: flags
|
||||
Filter []string // search: -F
|
||||
@@ -83,19 +86,27 @@ type Config struct {
|
||||
TimestampFormat string // search: --timestamp-format
|
||||
Explain bool // search: -e
|
||||
Validate bool // search: --validate
|
||||
|
||||
SortBy string // sort: -k
|
||||
Ascending bool // sort: -a
|
||||
|
||||
Exclude string // cluster compare: -e (regexp)
|
||||
All, Verbose bool // cluster status: -a -v
|
||||
Persistent, Transient, Default bool // cluster settings set: -p -t -D
|
||||
|
||||
Force bool // ccr follower renew: -f
|
||||
HaveJQ bool // determined at runtime by ourselfes
|
||||
|
||||
DebugHTTP bool // root: --debug-http
|
||||
Separator string // role diff: -s
|
||||
NotDeployed bool // role diff: -n
|
||||
Undefined bool // role diff: -u
|
||||
Diff bool // role diff: -D
|
||||
Hidden bool // ds ls: -H
|
||||
|
||||
// rollover
|
||||
MaxAge string
|
||||
MaxDocs, MaxShardSize, MaxShardDocs int // roll over
|
||||
DryRun bool // rollover: -n
|
||||
}
|
||||
|
||||
func NewConfig() *Config {
|
||||
|
||||
@@ -237,3 +237,23 @@ func DatastreamDelete(conf *cfg.Config, dsname string) error {
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func DatastreamRollover(conf *cfg.Config, ds string) error {
|
||||
res, err := RolloverAlias(conf, ds)
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to rollover data stream: %s", esErrorString(err))
|
||||
}
|
||||
|
||||
table := printer.NewTable(conf, 2, 5)
|
||||
table.Addheaders("rollover response", "value")
|
||||
table.Entries = [][]string{
|
||||
{"acknowledged", fmt.Sprintf("%t", res.Acknowledged)},
|
||||
{"rolled over", fmt.Sprintf("%t", res.RolledOver)},
|
||||
{"shards acknowledged", fmt.Sprintf("%t", res.ShardsAcknowledged)},
|
||||
{"old index", res.OldIndex},
|
||||
{"new index", res.NewIndex},
|
||||
}
|
||||
|
||||
return table.Print()
|
||||
}
|
||||
|
||||
@@ -113,3 +113,23 @@ func IndexAliasDelete(conf *cfg.Config, index, alias string) error {
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func IndexAliasRollover(conf *cfg.Config, alias string) error {
|
||||
res, err := RolloverAlias(conf, alias)
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to rollover index alias: %s", esErrorString(err))
|
||||
}
|
||||
|
||||
table := printer.NewTable(conf, 2, 5)
|
||||
table.Addheaders("rollover response", "value")
|
||||
table.Entries = [][]string{
|
||||
{"acknowledged", fmt.Sprintf("%t", res.Acknowledged)},
|
||||
{"rolled over", fmt.Sprintf("%t", res.RolledOver)},
|
||||
{"shards acknowledged", fmt.Sprintf("%t", res.ShardsAcknowledged)},
|
||||
{"old index", res.OldIndex},
|
||||
{"new index", res.NewIndex},
|
||||
}
|
||||
|
||||
return table.Print()
|
||||
}
|
||||
|
||||
@@ -127,6 +127,7 @@ func IndexTemplateShow(conf *cfg.Config, tplname string) error {
|
||||
table = printer.NewTable(conf, 2, 0)
|
||||
table.Addheaders("index field mapping", "type")
|
||||
|
||||
if tpl.IndexTemplate.Template.Mappings != nil {
|
||||
for name, field := range tpl.IndexTemplate.Template.Mappings.Properties {
|
||||
typeval := ""
|
||||
|
||||
@@ -148,6 +149,7 @@ func IndexTemplateShow(conf *cfg.Config, tplname string) error {
|
||||
if err := table.Print(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -343,6 +345,12 @@ func IndexTemplateModify(conf *cfg.Config, name string, mappings []string) error
|
||||
return fmt.Errorf("failed to modify index template: %s", esErrorString(err))
|
||||
}
|
||||
|
||||
if conf.Rollover {
|
||||
if err := rolloverAliasIndexTemplate(conf, name); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -358,6 +366,72 @@ func IndexTemplateDelete(conf *cfg.Config, name string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Based on given index template find the associated index patterns,
|
||||
// find indices matching those, find their associated aliases and
|
||||
// rollover all we find. Only run when conf.Rollover==true
|
||||
func rolloverAliasIndexTemplate(conf *cfg.Config, name string) error {
|
||||
res, err := conf.DefaultCluster.ES.Indices.GetIndexTemplate().
|
||||
Name(name).
|
||||
Do(context.Background())
|
||||
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get index template: %s", esErrorString(err))
|
||||
}
|
||||
|
||||
slog.Debug("res", "index template", res)
|
||||
|
||||
if len(res.IndexTemplates) != 1 {
|
||||
return errors.New("multiple or no index template matched the pattern")
|
||||
}
|
||||
|
||||
patterns := res.IndexTemplates[0].IndexTemplate.IndexPatterns
|
||||
aliases := map[string]int{}
|
||||
|
||||
// find all aliases matching the patterns
|
||||
for _, pattern := range patterns {
|
||||
res, err := conf.DefaultCluster.ES.Indices.ResolveIndex(pattern).
|
||||
Do(context.Background())
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to resolve index pattern: %s", esErrorString(err))
|
||||
}
|
||||
|
||||
for _, index := range res.Indices {
|
||||
for _, alias := range index.Aliases {
|
||||
aliases[alias] = 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(aliases) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
table := printer.NewTable(conf, 4, 0)
|
||||
table.Addheaders("rollover alias", "status", "ack", "new index")
|
||||
|
||||
// apply rollover to all matching aliases, if any
|
||||
for alias := range aliases {
|
||||
res, err := RolloverAlias(conf, alias)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
table.Entries = append(table.Entries, []string{
|
||||
alias,
|
||||
fmt.Sprintf("%t", res.RolledOver),
|
||||
fmt.Sprintf("%t", res.Acknowledged),
|
||||
res.NewIndex,
|
||||
})
|
||||
|
||||
if err := table.Print(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
}
|
||||
|
||||
func modMappings(mappings []string) (types.TypeMappingVariant, error) {
|
||||
typemaps := esdsl.NewTypeMapping()
|
||||
|
||||
|
||||
65
pkg/es/rollover.go
Normal file
65
pkg/es/rollover.go
Normal file
@@ -0,0 +1,65 @@
|
||||
/*
|
||||
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"
|
||||
|
||||
"codeberg.org/scip/esctl/pkg/cfg"
|
||||
"github.com/elastic/go-elasticsearch/v9/typedapi/esdsl"
|
||||
"github.com/elastic/go-elasticsearch/v9/typedapi/indices/rollover"
|
||||
"github.com/elastic/go-elasticsearch/v9/typedapi/types"
|
||||
)
|
||||
|
||||
func RolloverConditions(conf *cfg.Config) types.RolloverConditionsVariant {
|
||||
cond := esdsl.NewRolloverConditions()
|
||||
|
||||
if conf.MaxAge != "" {
|
||||
cond.MaxAge(esdsl.NewDuration().String(conf.MaxAge))
|
||||
}
|
||||
if conf.MaxDocs > 0 {
|
||||
cond.MaxDocs(int64(conf.MaxDocs))
|
||||
}
|
||||
if conf.MaxShardSize > 0 {
|
||||
bs := esdsl.NewByteSize()
|
||||
bs.Int64(int64(conf.MaxShardSize))
|
||||
cond.MaxPrimaryShardSize(bs)
|
||||
}
|
||||
if conf.MaxShardDocs > 0 {
|
||||
cond.MaxPrimaryShardDocs(int64(conf.MaxShardDocs))
|
||||
}
|
||||
|
||||
return cond
|
||||
}
|
||||
|
||||
func RolloverAlias(conf *cfg.Config, alias string) (*rollover.Response, error) {
|
||||
roll := conf.DefaultCluster.ES.Indices.Rollover(alias)
|
||||
|
||||
if conf.Wait {
|
||||
roll.WaitForActiveShards("all")
|
||||
}
|
||||
|
||||
if conf.DryRun {
|
||||
roll.DryRun(true)
|
||||
}
|
||||
|
||||
cond := RolloverConditions(conf)
|
||||
|
||||
roll.Conditions(cond)
|
||||
|
||||
return roll.Do(context.Background())
|
||||
}
|
||||
Reference in New Issue
Block a user