diff --git a/cmd/datastream.go b/cmd/datastream.go index 7705114..cbc2fdf 100644 --- a/cmd/datastream.go +++ b/cmd/datastream.go @@ -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 ", + + 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) + }, + } +} diff --git a/cmd/index_alias.go b/cmd/index_alias.go index ac5d506..5f4098f 100644 --- a/cmd/index_alias.go +++ b/cmd/index_alias.go @@ -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 ", + + 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", diff --git a/cmd/index_template.go b/cmd/index_template.go index 363caf7..d7e0c14 100644 --- a/cmd/index_template.go +++ b/cmd/index_template.go @@ -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,8 +177,10 @@ https://www.elastic.co/docs/reference/elasticsearch/index-settings return fmt.Errorf("no name specified") } - if !slices.Contains([]string{"standard", "timeseries", "logsdb", "lookup"}, conf.Mode) { - return errors.New("mode must be one of: standard, timeseries, logsdb or lookup") + 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:] diff --git a/pkg/cfg/config.go b/pkg/cfg/config.go index bb80b87..2b9f86c 100644 --- a/pkg/cfg/config.go +++ b/pkg/cfg/config.go @@ -48,12 +48,14 @@ type Cluster struct { } type Config struct { - ConfigFile string // -c - CurrentCluster string // -C - Debug bool // -d - Output string // -o - Clusters map[string]*Cluster - DefaultCluster *Cluster + ConfigFile string // -c + CurrentCluster string // -C + Debug bool // -d + Output string // -o + 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,30 +74,39 @@ 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 - Path string // search+doc sh: -p - Subhelp bool // search+doc sh: -H - Tail bool // search: -f [tail] - Or bool // search: -O - Range string // search: -r - 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 + From, To, MaxItems int // search: flags + Filter []string // search: -F + Path string // search+doc sh: -p + Subhelp bool // search+doc sh: -H + Tail bool // search: -f [tail] + Or bool // search: -O + Range string // search: -r + 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 + + 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 { diff --git a/pkg/es/datastream.go b/pkg/es/datastream.go index 42cf097..0d5677e 100644 --- a/pkg/es/datastream.go +++ b/pkg/es/datastream.go @@ -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() +} diff --git a/pkg/es/index_alias.go b/pkg/es/index_alias.go index e29476f..0f25085 100644 --- a/pkg/es/index_alias.go +++ b/pkg/es/index_alias.go @@ -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() +} diff --git a/pkg/es/index_template.go b/pkg/es/index_template.go index f25b471..d50baca 100644 --- a/pkg/es/index_template.go +++ b/pkg/es/index_template.go @@ -127,26 +127,28 @@ func IndexTemplateShow(conf *cfg.Config, tplname string) error { table = printer.NewTable(conf, 2, 0) table.Addheaders("index field mapping", "type") - for name, field := range tpl.IndexTemplate.Template.Mappings.Properties { - typeval := "" + if tpl.IndexTemplate.Template.Mappings != nil { + for name, field := range tpl.IndexTemplate.Template.Mappings.Properties { + typeval := "" - switch val := field.(type) { - case *types.IntegerNumberProperty: - typeval = val.Type - case *types.KeywordProperty: - typeval = val.Type - case *types.DateProperty: - typeval = val.Type + switch val := field.(type) { + case *types.IntegerNumberProperty: + typeval = val.Type + case *types.KeywordProperty: + typeval = val.Type + case *types.DateProperty: + typeval = val.Type + } + + table.Entries = append(table.Entries, []string{ + name, typeval, + }) } - table.Entries = append(table.Entries, []string{ - name, typeval, - }) - } - - fmt.Println() - if err := table.Print(); err != nil { - return err + fmt.Println() + 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() diff --git a/pkg/es/rollover.go b/pkg/es/rollover.go new file mode 100644 index 0000000..014845e --- /dev/null +++ b/pkg/es/rollover.go @@ -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 . +*/ +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()) +}