add datastream support, refactor custom completion (#32)

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 <http://www.gnu.org/licenses/>.
+*/
+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
+}

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)
 	}