From ec1ce56b2d17dce743169773b47710cfadbac878 Mon Sep 17 00:00:00 2001 From: "T. von Dein" Date: Wed, 10 Jun 2026 15:43:14 +0200 Subject: [PATCH] add datastream support, refactor custom completion (#32) --- cmd/ccr.go | 4 +- cmd/ccr_follower.go | 14 +-- cmd/completion.go | 71 +++++-------- cmd/datastream.go | 140 +++++++++++++++++++++++++ cmd/index.go | 10 +- cmd/index_alias.go | 4 +- cmd/roles.go | 2 +- cmd/root.go | 15 +++ pkg/cfg/config.go | 3 +- pkg/es/datastream.go | 239 +++++++++++++++++++++++++++++++++++++++++++ pkg/es/index.go | 6 ++ 11 files changed, 445 insertions(+), 63 deletions(-) create mode 100644 cmd/datastream.go create mode 100644 pkg/es/datastream.go diff --git a/cmd/ccr.go b/cmd/ccr.go index 8e19893..4eb536c 100644 --- a/cmd/ccr.go +++ b/cmd/ccr.go @@ -59,7 +59,7 @@ func CcrStatus(conf *cfg.Config) *cli.Command { }, ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeCluster(cmd) + complete(cmd, Ccluster) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -115,7 +115,7 @@ func CcrRemoteInfo(conf *cfg.Config) *cli.Command { UsageText: "info [options] []", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { diff --git a/cmd/ccr_follower.go b/cmd/ccr_follower.go index 20bb02b..ea1c811 100644 --- a/cmd/ccr_follower.go +++ b/cmd/ccr_follower.go @@ -60,7 +60,7 @@ func CcrFollowerRenew(conf *cfg.Config) *cli.Command { }, ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -82,7 +82,7 @@ func CcrFollowerResume(conf *cfg.Config) *cli.Command { UsageText: "resume [options] ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -104,7 +104,7 @@ func CcrFollowerPause(conf *cfg.Config) *cli.Command { UsageText: "pause [options] ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -126,7 +126,7 @@ func CcrFollowerUnfollow(conf *cfg.Config) *cli.Command { UsageText: "unfollow [options] ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -158,7 +158,7 @@ func CcrFollowerAdd(conf *cfg.Config) *cli.Command { }, ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -181,7 +181,7 @@ func CcrFollowerDelete(conf *cfg.Config) *cli.Command { UsageText: "delete ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -205,7 +205,7 @@ func CcrFollowerShow(conf *cfg.Config) *cli.Command { UsageText: "show ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { diff --git a/cmd/completion.go b/cmd/completion.go index 8fb4459..107f278 100644 --- a/cmd/completion.go +++ b/cmd/completion.go @@ -25,7 +25,14 @@ import ( "github.com/urfave/cli/v3" ) -func completeIndex(cmd *cli.Command) { +const ( + Cindex = iota + Cdatastream + Crole + Ccluster +) + +func complete(cmd *cli.Command, what int) { if cmd.NArg() > 0 { return } @@ -37,53 +44,27 @@ func completeIndex(cmd *cli.Command) { return } - indices, err := es.IndexNames(conf) + var list []string + var err error + + switch what { + case Cindex: + list, err = es.IndexNames(conf) + case Crole: + list, err = es.RoleNames(conf) + case Ccluster: + for cluster := range conf.Clusters { + list = append(list, cluster) + } + case Cdatastream: + list, err = es.DatastreamNames(conf) + } + if err != nil { return } - for _, index := range indices { - fmt.Println(index) - } - -} - -func completeRole(cmd *cli.Command) { - if cmd.NArg() > 0 { - return - } - - // FIXME: config should load from root.Before(), see https://github.com/urfave/cli/issues/2348 - // workaround: load it directly here - conf := cfg.NewConfig() - if err := conf.Init(); err != nil { - return - } - - roles, err := es.RoleNames(conf) - if err != nil { - return - } - - for _, role := range roles { - fmt.Println(role) - } - -} - -func completeCluster(cmd *cli.Command) { - if cmd.NArg() > 0 { - return - } - - // FIXME: config should load from root.Before(), see https://github.com/urfave/cli/issues/2348 - // workaround: load it directly here - conf := cfg.NewConfig() - if err := conf.Init(); err != nil { - return - } - - for cluster := range conf.Clusters { - fmt.Println(cluster) + for _, item := range list { + fmt.Println(item) } } diff --git a/cmd/datastream.go b/cmd/datastream.go new file mode 100644 index 0000000..7705114 --- /dev/null +++ b/cmd/datastream.go @@ -0,0 +1,140 @@ +/* +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 cmd + +import ( + "context" + "errors" + + "codeberg.org/scip/esctl/pkg/cfg" + "codeberg.org/scip/esctl/pkg/es" + + "github.com/urfave/cli/v3" +) + +func Datastream(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "datastream", + Aliases: []string{"ds"}, + Usage: "manage data streams", + + Commands: []*cli.Command{ + DatastreamList(conf), + DatastreamShow(conf), + DatastreamCreate(conf), + DatastreamDelete(conf), + }, + } +} + +func DatastreamList(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "list", + Aliases: []string{"ls"}, + Usage: "list indicies", + + Flags: []cli.Flag{ + &cli.IntFlag{ + Name: "max", + Usage: "max number of items to process", + Destination: &conf.MaxItems, + Aliases: []string{"m"}, + }, + &cli.BoolFlag{ + Name: "hidden", + Usage: "include hidden data streams", + Destination: &conf.Hidden, + Aliases: []string{"H"}, + }, + &cli.StringSliceFlag{ + Name: "filter", + Usage: "show only data streams matching the filter", + Destination: &conf.Filter, + Aliases: []string{"F"}, + }, + }, + + Action: func(ctx context.Context, cmd *cli.Command) error { + return es.DatastreamList(conf) + }, + } +} + +func DatastreamShow(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "show", + Aliases: []string{"sh"}, + Usage: "show details about an data stream", + + Action: func(ctx context.Context, cmd *cli.Command) error { + index := cmd.Args().Get(0) + if index == "" { + return errors.New("no data stream specified") + } + + return es.DatastreamShow(conf, cmd.Args().Get(0)) + }, + + ShellComplete: func(ctx context.Context, cmd *cli.Command) { + complete(cmd, Cdatastream) + }, + } +} + +func DatastreamCreate(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "create", + Aliases: []string{"+"}, + Usage: "create a new data stream", + UsageText: "create [options] ", + + Action: func(ctx context.Context, cmd *cli.Command) error { + index := cmd.Args().Get(0) + if index == "" { + return errors.New("no data stream specified") + } + + return es.DatastreamCreate(conf, cmd.Args().Get(0)) + }, + + ShellComplete: func(ctx context.Context, cmd *cli.Command) { + complete(cmd, Cdatastream) + }, + } +} + +func DatastreamDelete(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "delete", + Aliases: []string{"rm"}, + Usage: "delete a data stream", + UsageText: "delete [options] ", + + Action: func(ctx context.Context, cmd *cli.Command) error { + index := cmd.Args().Get(0) + if index == "" { + return errors.New("no data stream specified") + } + + return es.DatastreamDelete(conf, cmd.Args().Get(0)) + }, + + ShellComplete: func(ctx context.Context, cmd *cli.Command) { + complete(cmd, Cdatastream) + }, + } +} diff --git a/cmd/index.go b/cmd/index.go index 0064dfd..8b40687 100644 --- a/cmd/index.go +++ b/cmd/index.go @@ -134,7 +134,7 @@ func IndexShow(conf *cfg.Config) *cli.Command { }, ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, } } @@ -189,7 +189,7 @@ func IndexDelete(conf *cfg.Config) *cli.Command { UsageText: "delete ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -210,7 +210,7 @@ func IndexClose(conf *cfg.Config) *cli.Command { UsageText: "close ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -231,7 +231,7 @@ func IndexModify(conf *cfg.Config) *cli.Command { UsageText: "modify ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Flags: []cli.Flag{ @@ -282,7 +282,7 @@ func IndexFields(conf *cfg.Config) *cli.Command { }, ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { diff --git a/cmd/index_alias.go b/cmd/index_alias.go index 2153cf5..ac5d506 100644 --- a/cmd/index_alias.go +++ b/cmd/index_alias.go @@ -51,7 +51,7 @@ func IndexAliasCreate(conf *cfg.Config) *cli.Command { UsageText: "create ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { @@ -75,7 +75,7 @@ func IndexAliasDelete(conf *cfg.Config) *cli.Command { UsageText: "delete ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeIndex(cmd) + complete(cmd, Cindex) }, Action: func(ctx context.Context, cmd *cli.Command) error { diff --git a/cmd/roles.go b/cmd/roles.go index 68a0959..8d4d480 100644 --- a/cmd/roles.go +++ b/cmd/roles.go @@ -68,7 +68,7 @@ func RoleShow(conf *cfg.Config) *cli.Command { UsageText: "show [options] ", ShellComplete: func(ctx context.Context, cmd *cli.Command) { - completeRole(cmd) + complete(cmd, Crole) }, Action: func(ctx context.Context, cmd *cli.Command) error { diff --git a/cmd/root.go b/cmd/root.go index 3dc28d0..56f4dbc 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -90,6 +90,7 @@ func Main() int { Cluster(conf), Ccr(conf), Index(conf), + Datastream(conf), Shard(conf), Snapshot(conf), Node(conf), @@ -146,6 +147,20 @@ friends.#(first%"D*").last >> "Murphy" friends.#(first!%"D*").last >> "Craig" friends.#(nets.#(=="fb"))#.first >> ["Dale","Roger"] +To extract more than one field, use: + +{"ns":source.namespace_name,"time":source.@timestamp} >> + +{ + "ns": "3f80316965c64405-275a75782e984bcc8d82", + "time": "2026-06-10T04:23:38.914080361+00:00" +} +{ + "ns": "3f80316965c64405-f0e1dcef668e45bb9afd", + "time": "2026-06-10T09:39:00.554892542+00:00" +} + + Documentation: https://github.com/tidwall/gjson/blob/master/SYNTAX.md` return &cli.Command{ diff --git a/pkg/cfg/config.go b/pkg/cfg/config.go index a4905e8..29228bb 100644 --- a/pkg/cfg/config.go +++ b/pkg/cfg/config.go @@ -34,7 +34,7 @@ import ( ) const ( - Version string = `v0.0.18` + Version string = `v0.0.19` ) var ( @@ -83,6 +83,7 @@ type Config struct { NotDeployed bool // role diff: -n Undefined bool // role diff: -u Diff bool // role diff: -D + Hidden bool // ds ls: -H } func NewConfig() *Config { diff --git a/pkg/es/datastream.go b/pkg/es/datastream.go new file mode 100644 index 0000000..42cf097 --- /dev/null +++ b/pkg/es/datastream.go @@ -0,0 +1,239 @@ +/* +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" + "errors" + "fmt" + "log/slog" + "regexp" + + "codeberg.org/scip/esctl/pkg/cfg" + "codeberg.org/scip/esctl/pkg/printer" + "github.com/dustin/go-humanize" + "github.com/elastic/go-elasticsearch/v9/typedapi/types" +) + +func DatastreamNames(conf *cfg.Config) ([]string, error) { + res, err := conf.DefaultCluster.ES.Indices.GetDataStream(). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + return nil, fmt.Errorf("failed to get data streams: %s", esErrorString(err)) + } + + dss := make([]string, len(res.DataStreams)) + for idx, ds := range res.DataStreams { + dss[idx] = ds.Name + } + + return dss, nil +} + +func DatastreamList(conf *cfg.Config) error { + res, err := conf.DefaultCluster.ES.Indices.GetDataStream(). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + return fmt.Errorf("failed to get data streams: %s", esErrorString(err)) + } + + slog.Debug("ES result", "data streams", res) + + list := filterDatastreams(conf, res.DataStreams) + + size := len(list) + + if conf.MaxItems > 0 { + if size > conf.MaxItems { + size = conf.MaxItems + } + } + + table := printer.NewTable(conf, 7, size) + table.Addheaders("name", "ilm policy", "hidden", "system", "replicated", "generation", "timestamp field") + + for idx, ds := range list { + name := printer.Colorize(conf, ds.Status.String(), ds.Name) + policy := "" + if ds.IlmPolicy != nil { + policy = *ds.IlmPolicy + } + + table.Entries[idx] = []string{ + name, + policy, + fmt.Sprintf("%t", ds.Hidden), + fmt.Sprintf("%t", *ds.System), + fmt.Sprintf("%t", *ds.Replicated), + fmt.Sprintf("%d", ds.Generation), + ds.TimestampField.Name, + } + + if idx == size-1 { + break + } + } + + table.Sort() + if err := table.Print(); err != nil { + return err + } + + return nil +} + +func filterDatastreams(conf *cfg.Config, list []types.DataStream) []types.DataStream { + selected := []types.DataStream{} + + for _, ds := range list { + if !conf.Hidden && ds.Hidden { + continue + } + + selected = append(selected, ds) + } + + if len(conf.Filter) == 0 { + return selected + } + + // we support just one filter here, for now + filter := *regexp.MustCompile(conf.Filter[0]) + newlist := []types.DataStream{} + + for _, ds := range selected { + if filter.MatchString(ds.Name) { + newlist = append(newlist, ds) + } + } + + return newlist +} + +func DatastreamShow(conf *cfg.Config, dsname string) error { + res, err := conf.DefaultCluster.ES.Indices.GetDataStream(). + Name(dsname). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + return fmt.Errorf("failed to get data stream: %s", esErrorString(err)) + } + + slog.Debug("ES result", "data stream", res) + + if len(res.DataStreams) == 0 { + return errors.New("no data stream found with that name") + } + + stats, err := conf.DefaultCluster.ES.Indices.DataStreamsStats(). + Name(dsname). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + return fmt.Errorf("failed to get data stream stats: %s", esErrorString(err)) + } + + table := printer.NewTable(conf, 11, 2) + table.Addheaders("data stream property", "value") + + ds := res.DataStreams[0] + + name := printer.Colorize(conf, ds.Status.String(), ds.Name) + policy := "" + if ds.IlmPolicy != nil { + policy = *ds.IlmPolicy + } + + table.Entries = [][]string{ + {"name", name}, + {"ilm policy", policy}, + {"hidden", fmt.Sprintf("%t", ds.Hidden)}, + {"system", fmt.Sprintf("%t", *ds.System)}, + {"replicated", fmt.Sprintf("%t", *ds.Replicated)}, + {"generation", fmt.Sprintf("%d", ds.Generation)}, + {"timestamp field", ds.TimestampField.Name}, + {"backing indices", fmt.Sprintf("%d", stats.BackingIndices)}, + {"size", humanize.Bytes(uint64(stats.TotalStoreSizeBytes))}, + {"shards-failed", fmt.Sprintf("%d", stats.Shards_.Failed)}, + {"shards-successful", fmt.Sprintf("%d", stats.Shards_.Successful)}, + } + + if err := table.Print(); err != nil { + return err + } + + table = printer.NewTable(conf, 5, len(ds.Indices)) + table.Addheaders("backing index name", "uuid", "prefer ilm", "ilm policy", "managed by") + + for idx, index := range ds.Indices { + policy := "" + if ds.IlmPolicy != nil { + policy = *ds.IlmPolicy + } + + table.Entries[idx] = []string{ + index.IndexName, + index.IndexUuid, + fmt.Sprintf("%t", *index.PreferIlm), + policy, + index.ManagedBy.Name, + } + + if idx < len(ds.Indices) { + fmt.Println() + } + } + + table.Sort() + if err := table.Print(); err != nil { + return err + } + + return nil +} + +func DatastreamCreate(conf *cfg.Config, dsname string) error { + _, err := conf.DefaultCluster.ES.Indices.CreateDataStream(dsname). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + + if err != nil { + return fmt.Errorf("failed to create datastream: %s", esErrorString(err)) + } + + return nil +} + +func DatastreamDelete(conf *cfg.Config, dsname string) error { + _, err := conf.DefaultCluster.ES.Indices.DeleteDataStream(dsname). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + + if err != nil { + return fmt.Errorf("failed to delete datastream: %s", esErrorString(err)) + } + + return nil +} diff --git a/pkg/es/index.go b/pkg/es/index.go index a1d329b..0aace00 100644 --- a/pkg/es/index.go +++ b/pkg/es/index.go @@ -59,6 +59,12 @@ func filterIndices(conf *cfg.Config, list indices.Response) indices.Response { if !conf.Partials && strings.HasPrefix(*index.Index, "partial-") { continue } + + if strings.HasPrefix(*index.Index, ".ds-") { + // ignore data stream backing indicies + continue + } + selectedlist = append(selectedlist, index) }