diff --git a/cmd/ccr.go b/cmd/ccr.go new file mode 100644 index 0000000..053f030 --- /dev/null +++ b/cmd/ccr.go @@ -0,0 +1,103 @@ +/* +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 Ccr(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "ccr", + Aliases: []string{"replication", "rep"}, + Usage: "manage cross cluster replication", + + Commands: []*cli.Command{ + CcrStatus(conf), + CcrShardPause(conf), + CcrShardResume(conf), + CcrFollower(conf), + }, + } +} + +func CcrStatus(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "status", + Aliases: []string{"st"}, + Usage: "cross cluster replication status (yaml config with 2 clusters required)", + + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "exclude", + Usage: "regexp of indicies to exclude", + Destination: &conf.Exclude, + Aliases: []string{"e"}, + }, + }, + + Action: func(ctx context.Context, cmd *cli.Command) error { + leader := cmd.Args().Get(0) + follower := cmd.Args().Get(1) + + if leader == "" || follower == "" { + return errors.New("no leader and follower aliases specified") + } + + _, hasLeader := conf.Clusters[leader] + _, hasFollower := conf.Clusters[follower] + + if !hasLeader || !hasFollower { + return errors.New("either leader or follower alias not configured") + } + + if err := es.CcrStatus(conf, leader, follower); err != nil { + return err + } + + return nil + }, + } +} + +func CcrShardPause(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "pause", + Usage: "pause shard allocation", + + Action: func(ctx context.Context, cmd *cli.Command) error { + return es.CcrShardPause(conf) + }, + } +} + +func CcrShardResume(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "resume", + Usage: "resume shard allocation", + + Action: func(ctx context.Context, cmd *cli.Command) error { + return es.CcrShardResume(conf) + }, + } +} diff --git a/cmd/ccr_follower.go b/cmd/ccr_follower.go new file mode 100644 index 0000000..5cae8e0 --- /dev/null +++ b/cmd/ccr_follower.go @@ -0,0 +1,108 @@ +/* +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 CcrFollower(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "follower", + Aliases: []string{"f"}, + Usage: "manage ccr follower indices", + + Commands: []*cli.Command{ + CcrFollowerShow(conf), + CcrFollowerAdd(conf), + CcrFollowerDelete(conf), + }, + } +} + +func CcrFollowerAdd(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "add", + Aliases: []string{"+"}, + Usage: "add ccr follower index", + UsageText: "add [options] ", + + Flags: []cli.Flag{ + &cli.BoolFlag{ + Name: "wait", + Usage: "wait for active shards", + Destination: &conf.Wait, + Aliases: []string{"w"}, + }, + }, + + Action: func(ctx context.Context, cmd *cli.Command) error { + args := cmd.Args() + + if args.Len() != 1 { + return errors.New("missing arguments: ") + } + + return es.CcrFollowerAdd(conf, cmd.Args().Get(0)) + }, + } +} + +func CcrFollowerDelete(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "delete", + Aliases: []string{"-"}, + Usage: "delete ccr follower index", + UsageText: "delete ", + + Action: func(ctx context.Context, cmd *cli.Command) error { + args := cmd.Args() + + if args.Len() != 1 { + return errors.New("missing arguments: ") + } + + // just an alias to index delete + return es.IndexDelete(conf, cmd.Args().Get(0)) + }, + } +} + +func CcrFollowerShow(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "show", + Aliases: []string{"sh"}, + Usage: "show ccr follower index details", + UsageText: "show ", + + Action: func(ctx context.Context, cmd *cli.Command) error { + args := cmd.Args() + + if args.Len() != 1 { + return errors.New("missing arguments: ") + } + + return es.CcrFollowerShow(conf, cmd.Args().Get(0)) + }, + } +} diff --git a/cmd/cluster.go b/cmd/cluster.go index 40f897d..fdb8264 100644 --- a/cmd/cluster.go +++ b/cmd/cluster.go @@ -18,7 +18,6 @@ package cmd import ( "context" - "errors" "codeberg.org/scip/esctl/pkg/cfg" "codeberg.org/scip/esctl/pkg/es" @@ -26,10 +25,6 @@ import ( "github.com/urfave/cli/v3" ) -const ( - SETTINGS = `https://www.elastic.co/docs/reference/elasticsearch/configuration-reference` -) - func Cluster(conf *cfg.Config) *cli.Command { return &cli.Command{ Name: "cluster", @@ -37,7 +32,6 @@ func Cluster(conf *cfg.Config) *cli.Command { Usage: "manage cluster[s]", Commands: []*cli.Command{ - ClusterCompare(conf), ClusterStatus(conf), ClusterList(conf), ClusterSettings(conf), @@ -52,11 +46,7 @@ func ClusterList(conf *cfg.Config) *cli.Command { Aliases: []string{"ls"}, Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.ClusterList(conf); err != nil { - return err - } - - return nil + return es.ClusterList(conf) }, } } @@ -77,138 +67,7 @@ func ClusterStatus(conf *cfg.Config) *cli.Command { }, Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.ClusterStatus(conf); err != nil { - return err - } - - return nil - }, - } -} - -func ClusterCompare(conf *cfg.Config) *cli.Command { - return &cli.Command{ - Name: "compare", - Aliases: []string{"c"}, - Usage: "compare cluster[s] (yaml config with 2 clusters required)", - - Flags: []cli.Flag{ - &cli.StringFlag{ - Name: "exclude", - Usage: "regexp of indicies to exclude", - Destination: &conf.Exclude, - Aliases: []string{"e"}, - }, - }, - - Action: func(ctx context.Context, cmd *cli.Command) error { - leader := cmd.Args().Get(0) - follower := cmd.Args().Get(1) - - if leader == "" || follower == "" { - return errors.New("no leader and follower aliases specified") - } - - _, hasLeader := conf.Clusters[leader] - _, hasFollower := conf.Clusters[follower] - - if !hasLeader || !hasFollower { - return errors.New("either leader or follower alias not configured") - } - - if err := es.ClusterCompare(conf, leader, follower); err != nil { - return err - } - - return nil - }, - } -} - -func ClusterSettings(conf *cfg.Config) *cli.Command { - return &cli.Command{ - Name: "settings", - Usage: "cluster settings management", - Aliases: []string{"config"}, - - Commands: []*cli.Command{ - ClusterSettingsList(conf), - ClusterSettingsSet(conf), - }, - } -} - -func ClusterSettingsList(conf *cfg.Config) *cli.Command { - return &cli.Command{ - Name: "list", - Usage: "show cluster settings", - Aliases: []string{"ls", "get"}, - - Flags: []cli.Flag{ - &cli.BoolFlag{ - Name: "persistent", - Usage: "only show persistent setting[s] (default)", - Destination: &conf.Persistent, - Aliases: []string{"p"}, - }, - &cli.BoolFlag{ - Name: "transient", - Usage: "only show transient setting[s]", - Destination: &conf.Transient, - Aliases: []string{"t"}, - }, - &cli.BoolFlag{ - Name: "default", - Usage: "only show default setting[s]", - Destination: &conf.Default, - Aliases: []string{"D"}, - }, - }, - - Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.ClusterSettingsList(conf); err != nil { - return err - } - - return nil - }, - } -} - -func ClusterSettingsSet(conf *cfg.Config) *cli.Command { - return &cli.Command{ - Name: "set", - Usage: "set|update cluster settings", - Aliases: []string{"set", "update"}, - UsageText: "set [options] setting:value [setting:value ...]\n\nReference: " + SETTINGS, - - Flags: []cli.Flag{ - &cli.BoolFlag{ - Name: "persistent", - Usage: "add persistent setting[s] (default)", - Destination: &conf.Persistent, - Aliases: []string{"p"}, - }, - &cli.BoolFlag{ - Name: "transient", - Usage: "add transient setting[s]", - Destination: &conf.Transient, - Aliases: []string{"t"}, - }, - }, - - Action: func(ctx context.Context, cmd *cli.Command) error { - args := cmd.Args() - - if args.Len() == 0 { - return errors.New("at least one setting must be specified (format: setting:value)") - } - - if err := es.ClusterSettingsSet(conf, cmd.Args()); err != nil { - return err - } - - return nil + return es.ClusterStatus(conf) }, } } diff --git a/cmd/cluster_settings.go b/cmd/cluster_settings.go new file mode 100644 index 0000000..f7d8bd9 --- /dev/null +++ b/cmd/cluster_settings.go @@ -0,0 +1,119 @@ +/* +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" +) + +const ( + SETTINGS = `https://www.elastic.co/docs/reference/elasticsearch/configuration-reference` +) + +func ClusterSettings(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "settings", + Usage: "cluster settings management", + Aliases: []string{"config"}, + + Commands: []*cli.Command{ + ClusterSettingsList(conf), + ClusterSettingsSet(conf), + }, + } +} + +func ClusterSettingsList(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "list", + Usage: "show cluster settings", + Aliases: []string{"ls", "get"}, + + Flags: []cli.Flag{ + &cli.BoolFlag{ + Name: "persistent", + Usage: "only show persistent setting[s] (default)", + Destination: &conf.Persistent, + Aliases: []string{"p"}, + }, + &cli.BoolFlag{ + Name: "transient", + Usage: "only show transient setting[s]", + Destination: &conf.Transient, + Aliases: []string{"t"}, + }, + &cli.BoolFlag{ + Name: "default", + Usage: "only show default setting[s]", + Destination: &conf.Default, + Aliases: []string{"D"}, + }, + }, + + Action: func(ctx context.Context, cmd *cli.Command) error { + if err := es.ClusterSettingsList(conf); err != nil { + return err + } + + return nil + }, + } +} + +func ClusterSettingsSet(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "set", + Usage: "set|update cluster settings", + Aliases: []string{"set", "update"}, + UsageText: "set [options] setting:value [setting:value ...]\n\nReference: " + SETTINGS, + + Flags: []cli.Flag{ + &cli.BoolFlag{ + Name: "persistent", + Usage: "add persistent setting[s] (default)", + Destination: &conf.Persistent, + Aliases: []string{"p"}, + }, + &cli.BoolFlag{ + Name: "transient", + Usage: "add transient setting[s]", + Destination: &conf.Transient, + Aliases: []string{"t"}, + }, + }, + + Action: func(ctx context.Context, cmd *cli.Command) error { + args := cmd.Args() + + if args.Len() == 0 { + return errors.New("at least one setting must be specified (format: setting:value)") + } + + if err := es.ClusterSettingsSet(conf, cmd.Args()); err != nil { + return err + } + + return nil + }, + } +} diff --git a/cmd/doc.go b/cmd/doc.go new file mode 100644 index 0000000..3ad3833 --- /dev/null +++ b/cmd/doc.go @@ -0,0 +1,58 @@ +/* +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 Doc(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "doc", + Usage: "manage documents", + + Commands: []*cli.Command{ + DocAdd(conf), + //Delete(conf), + }, + } +} + +func DocAdd(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "add", + Aliases: []string{"+"}, + Usage: "add JSON document index", + UsageText: "add [options] ''", + + Action: func(ctx context.Context, cmd *cli.Command) error { + args := cmd.Args() + + if args.Len() != 2 { + return errors.New("missing arguments: ") + } + + return es.DocAdd(conf, cmd.Args().Get(0), cmd.Args().Get(1)) + }, + } +} diff --git a/cmd/index.go b/cmd/index.go index 6a6d688..4bd7814 100644 --- a/cmd/index.go +++ b/cmd/index.go @@ -69,11 +69,7 @@ func IndexList(conf *cfg.Config) *cli.Command { }, Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.IndexList(conf); err != nil { - return err - } - - return nil + return es.IndexList(conf) }, } } @@ -85,11 +81,8 @@ func IndexShow(conf *cfg.Config) *cli.Command { Usage: "show details about an index", Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.IndexShow(conf, cmd.Args().Get(0)); err != nil { - return err - } + return es.IndexShow(conf, cmd.Args().Get(0)) - return nil }, // FIXME: doesn't work at all @@ -140,11 +133,14 @@ func IndexCreate(conf *cfg.Config) *cli.Command { }, Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.IndexCreate(conf, cmd.Args().Get(0)); err != nil { - return err + args := cmd.Args() + if args.Len() == 0 { + return fmt.Errorf("no index specified") } - return nil + mappings := args.Slice()[1:] + + return es.IndexCreate(conf, args.Get(0), mappings) }, } } @@ -156,11 +152,7 @@ func IndexDelete(conf *cfg.Config) *cli.Command { Usage: "delete an index", Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.IndexDelete(conf, cmd.Args().Get(0)); err != nil { - return err - } - - return nil + return es.IndexDelete(conf, cmd.Args().Get(0)) }, } } diff --git a/cmd/node.go b/cmd/node.go index 73c823c..cf04748 100644 --- a/cmd/node.go +++ b/cmd/node.go @@ -45,11 +45,7 @@ func NodeList(conf *cfg.Config) *cli.Command { Usage: "list nodes", Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.NodeList(conf); err != nil { - return err - } - - return nil + return es.NodeList(conf) }, } } @@ -62,10 +58,7 @@ func NodeShow(conf *cfg.Config) *cli.Command { UsageText: "show [options] ", Action: func(ctx context.Context, cmd *cli.Command) error { - // if err := es.NodeShow(conf, cmd.Args().Get(0)); err != nil { - // return err - // } - + // return es.NodeShow(conf, cmd.Args().Get(0)) return nil }, } diff --git a/cmd/root.go b/cmd/root.go index 5097d02..34635fe 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -76,7 +76,9 @@ func Main() int { Index(conf), Snapshot(conf), Cluster(conf), + Ccr(conf), Node(conf), + Doc(conf), }, Before: func(ctx context.Context, cmd *cli.Command) (context.Context, error) { diff --git a/cmd/search.go b/cmd/search.go index 76b8453..b193393 100644 --- a/cmd/search.go +++ b/cmd/search.go @@ -18,6 +18,7 @@ package cmd import ( "context" + "errors" "codeberg.org/scip/esctl/pkg/cfg" "codeberg.org/scip/esctl/pkg/es" @@ -30,7 +31,7 @@ func Search(conf *cfg.Config) *cli.Command { Name: "search", Aliases: []string{"/"}, Usage: "search within an index", - UsageText: "search [options] ", + UsageText: "search [options] ...", Flags: []cli.Flag{ &cli.StringFlag{ @@ -63,11 +64,12 @@ func Search(conf *cfg.Config) *cli.Command { }, Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.Search(conf, cmd.Args().Get(0)); err != nil { - return err + args := cmd.Args() + if args.Len() == 0 { + return errors.New("at least one query must be specified (format: field=pattern)") } - return nil + return es.Search(conf, args.Slice()) }, } } diff --git a/cmd/snapshot.go b/cmd/snapshot.go index db3c039..ee5070a 100644 --- a/cmd/snapshot.go +++ b/cmd/snapshot.go @@ -54,11 +54,7 @@ func SnapshotList(conf *cfg.Config) *cli.Command { }, Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.SnapshotList(conf); err != nil { - return err - } - - return nil + return es.SnapshotList(conf) }, } } @@ -71,11 +67,7 @@ func SnapshotShow(conf *cfg.Config) *cli.Command { UsageText: "show [options] ", Action: func(ctx context.Context, cmd *cli.Command) error { - if err := es.SnapshotShow(conf, cmd.Args().Get(0)); err != nil { - return err - } - - return nil + return es.SnapshotShow(conf, cmd.Args().Get(0)) }, } } diff --git a/pkg/cfg/config.go b/pkg/cfg/config.go index 062caf6..cc67fa2 100644 --- a/pkg/cfg/config.go +++ b/pkg/cfg/config.go @@ -31,7 +31,7 @@ import ( ) const ( - Version string = `v0.0.7` + Version string = `v0.0.8` ) type Cluster struct { diff --git a/pkg/es/ccr.go b/pkg/es/ccr.go new file mode 100644 index 0000000..308d4a3 --- /dev/null +++ b/pkg/es/ccr.go @@ -0,0 +1,87 @@ +/* +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" + + "codeberg.org/scip/esctl/pkg/cfg" + "github.com/elastic/go-elasticsearch/v9/typedapi/types" +) + +const ( + shard_pause = `cluster.routing.allocation.enable` +) + +func CcrShardPause(conf *cfg.Config) error { + return ClusterSettingsSetSingle(conf, shard_pause, "none") +} + +func CcrShardResume(conf *cfg.Config) error { + return ClusterSettingsSetSingle(conf, shard_pause, "all") +} + +func CcrStatus(conf *cfg.Config, leader, follower string) error { + if !checkClusterIsLeader(conf, leader) { + if !checkClusterIsLeader(conf, follower) { + return errors.New("leader/follower attribution is invalid, both clusters are followers") + } + + // reverse attribution + f := follower + follower = leader + leader = f + slog.Debug("leader/follower attribution is invalid, reversing", "leader", leader, "follower", follower) + } + + indices := ClusterIndices{} + + for _, alias := range []string{leader, follower} { + cat := conf.Clusters[alias].ES.Cat.Indices(). + // we need to add custom request headers, required for older ES instances + Header("content-type", "application/json"). + Header("accept", "application/json") + + res, err := cat.Do(context.Background()) + if err != nil { + return fmt.Errorf("failed to get indicies on %s: %s", alias, err) + } + + indices[alias] = make(map[string]*types.IndicesRecord, len(res)) + + for _, index := range res { + indices[alias][*index.Index] = &index + } + } + + if !checkClusterStatus(conf, leader, follower) { + return errors.New("one of the two clusters is in a failed state") + } + + findIlmErrors(conf, leader, follower) + + if findIndicesOnlyOnLeader(conf, indices, leader, follower) && + findOrphanedIndices(conf, indices, leader, follower) && + findFailedFollowerIndices(conf, indices, follower) { + fmt.Println("everything's hunky-dory.") + } + + return nil +} diff --git a/pkg/es/ccr_follower.go b/pkg/es/ccr_follower.go new file mode 100644 index 0000000..3ef2ebe --- /dev/null +++ b/pkg/es/ccr_follower.go @@ -0,0 +1,106 @@ +/* +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" + "fmt" + "log/slog" + + "codeberg.org/scip/esctl/pkg/cfg" +) + +func CcrFollowerAdd(conf *cfg.Config, index string) error { + res, err := conf.DefaultCluster.ES.Cluster.RemoteInfo(). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + return fmt.Errorf("failed to retrieve follower info: %s", err) + } + + remote := "" + for name := range res { + remote = name + break + } + + if remote == "" { + return fmt.Errorf("cluster doesn't have a follower: %s", err) + } + + create := conf.DefaultCluster.ES.Ccr.Follow(index). + LeaderIndex(index). + RemoteCluster(remote). + Header("content-type", "application/json"). + Header("accept", "application/json") + + if conf.Wait { + create.WaitForActiveShards("all") + } + + _, err = create.Do(context.Background()) + + if err != nil { + return fmt.Errorf("failed to create follower index: %s", err) + } + + return nil +} + +func CcrFollowerShow(conf *cfg.Config, index string) error { + res, err := conf.DefaultCluster.ES.Ccr.FollowStats(index). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + return fmt.Errorf("failed to retrieve follower index info: %s", err) + } + + slog.Debug("ES result", "follower stats", res.Indices) + + if len(res.Indices) == 0 { + return fmt.Errorf("cluster did not return any follower stats for index %s", index) + } + + if len(res.Indices[0].Shards) == 0 { + return fmt.Errorf("cluster did not return any shard stats on follower index %s", index) + } + + follower := res.Indices[0].Shards[0] + + table := NewTable(2, 9) + table.Addheaders("field", "value") + + table.entries = [][]string{ + {"name", index}, + {"remote_cluster", follower.RemoteCluster}, + {"leader_checkpoint", fmt.Sprintf("%d", follower.LeaderGlobalCheckpoint)}, + {"follower_checkpoint", fmt.Sprintf("%d", follower.FollowerGlobalCheckpoint)}, + {"bytes_read", fmt.Sprintf("%d", follower.BytesRead)}, + {"failed_read_requests", fmt.Sprintf("%d", follower.FailedReadRequests)}, + {"failed_write_requests", fmt.Sprintf("%d", follower.FailedWriteRequests)}, + {"successful_read_requests", fmt.Sprintf("%d", follower.SuccessfulReadRequests)}, + {"successful_write_requests", fmt.Sprintf("%d", follower.SuccessfulWriteRequests)}, + } + + if err := table.PrintMarkdown(); err != nil { + return err + } + + return nil +} diff --git a/pkg/es/cluster.go b/pkg/es/cluster.go index f9a0550..062944d 100644 --- a/pkg/es/cluster.go +++ b/pkg/es/cluster.go @@ -18,7 +18,6 @@ package es import ( "context" - "errors" "fmt" "log/slog" "slices" @@ -47,46 +46,6 @@ type apiResponse struct { which int } -func ClusterCompare(conf *cfg.Config, leader, follower string) error { - if !checkClusterFollower(conf, leader) { - return errors.New("leader/follower attribution is invalid, reverse cluster attribution and retry") - } - - indices := ClusterIndices{} - - for _, alias := range []string{leader, follower} { - cat := conf.Clusters[alias].ES.Cat.Indices(). - // we need to add custom request headers, required for older ES instances - Header("content-type", "application/json"). - Header("accept", "application/json") - - res, err := cat.Do(context.Background()) - if err != nil { - return fmt.Errorf("failed to get indicies on %s: %s", alias, err) - } - - indices[alias] = make(map[string]*types.IndicesRecord, len(res)) - - for _, index := range res { - indices[alias][*index.Index] = &index - } - } - - if !checkClusterStatus(conf, leader, follower) { - return errors.New("one of the two clusters is in a failed state") - } - - findIlmErrors(conf, leader, follower) - - if findIndicesOnlyOnLeader(conf, indices, leader, follower) && - findOrphanedIndices(conf, indices, leader, follower) && - findFailedFollowerIndices(conf, indices, follower) { - fmt.Println("everything's hunky-dory.") - } - - return nil -} - func ClusterList(conf *cfg.Config) error { table := NewTable(3, len(conf.Clusters)) @@ -102,8 +61,6 @@ func ClusterList(conf *cfg.Config) error { slices.Sort(names) - idx = 0 - for idx, name := range names { current := name == "default" || name == conf.CurrentCluster cluster := conf.Clusters[name] @@ -118,7 +75,6 @@ func ClusterList(conf *cfg.Config) error { } table.entries[idx] = []string{name, cluster.Uri, fmt.Sprintf("%t", current)} - idx++ } if err := table.PrintMarkdown(); err != nil { diff --git a/pkg/es/cluster_settings.go b/pkg/es/cluster_settings.go index f09e01a..84fa050 100644 --- a/pkg/es/cluster_settings.go +++ b/pkg/es/cluster_settings.go @@ -139,3 +139,22 @@ func ClusterSettingsSet(conf *cfg.Config, args cli.Args) error { return nil } + +func ClusterSettingsSetSingle(conf *cfg.Config, setting, value string) error { + message, err := json.Marshal(value) + if err != nil { + return fmt.Errorf("failed to marshall persistent value <%v> to valid JSON: %s", value, err) + } + + _, err = conf.Clusters[conf.CurrentCluster].ES.Cluster.PutSettings(). + AddPersistent(setting, message). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + + if err != nil { + return fmt.Errorf("failed to set %s: %s", setting, err) + } + + return nil +} diff --git a/pkg/es/cluster_util.go b/pkg/es/cluster_util.go index 1feca49..f1025d3 100644 --- a/pkg/es/cluster_util.go +++ b/pkg/es/cluster_util.go @@ -34,7 +34,7 @@ const ( DefaultExclude = `(part|monitoring|.internal|metrics-endpoint)` ) -func checkClusterFollower(conf *cfg.Config, leader string) bool { +func checkClusterIsLeader(conf *cfg.Config, leader string) bool { stats, err := conf.Clusters[leader].ES.Ccr.Stats(). Header("content-type", "application/json"). Header("accept", "application/json"). diff --git a/pkg/es/doc.go b/pkg/es/doc.go new file mode 100644 index 0000000..66905f0 --- /dev/null +++ b/pkg/es/doc.go @@ -0,0 +1,55 @@ +/* +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" + "encoding/json" + "fmt" + "time" + + "codeberg.org/scip/esctl/pkg/cfg" +) + +// create index with: +// +// esctl index create foo [id:keyword user:text age:integer] +// +// then add a doc: +// +// esctl doc add foo2 '{"id":"d8d8d","user":"scip"}' +func DocAdd(conf *cfg.Config, index, jsondoc string) error { + data := map[string]any{} + + err := json.Unmarshal([]byte(jsondoc), &data) + if err != nil { + return fmt.Errorf("supplied document was not valid JSON: %s", err) + } + + now := fmt.Sprintf("%d", time.Now().Unix()) + + _, err = conf.DefaultCluster.ES.Create(index, now). + Document(data). + Header("content-type", "application/json"). + Header("accept", "application/json"). + Do(context.Background()) + if err != nil { + return fmt.Errorf("failed to create new doc in index %s: %s", index, err) + } + + return nil +} diff --git a/pkg/es/index.go b/pkg/es/index.go index fd83a8e..4adc6dc 100644 --- a/pkg/es/index.go +++ b/pkg/es/index.go @@ -21,6 +21,7 @@ import ( "fmt" "log/slog" "strconv" + "strings" "time" "codeberg.org/scip/esctl/pkg/cfg" @@ -134,11 +135,7 @@ func IndexShow(conf *cfg.Config, index string) error { return nil } -func IndexCreate(conf *cfg.Config, index string) error { - if index == "" { - return fmt.Errorf("no index specified") - } - +func IndexCreate(conf *cfg.Config, index string, mappings []string) error { settings := esdsl.NewIndexSettings() create := conf.DefaultCluster.ES.Indices.Create(index). @@ -156,6 +153,30 @@ func IndexCreate(conf *cfg.Config, index string) error { settings = settings.NumberOfReplicas(strconv.Itoa(conf.Replicas)) } + if len(mappings) > 0 { + maps := esdsl.NewTypeMapping() + + for _, mapping := range mappings { + parts := strings.Split(mapping, ":") + if len(parts) != 2 { + return fmt.Errorf("invalid mapping %s, expect (type: integer, text, date, keyword)", mapping) + } + + switch parts[1] { + case "text": + maps.AddProperty(parts[0], esdsl.NewTextProperty()) + case "integer": + maps.AddProperty(parts[0], esdsl.NewIntegerNumberProperty()) + case "date": + maps.AddProperty(parts[0], esdsl.NewDateProperty()) + case "keyword": + maps.AddProperty(parts[0], esdsl.NewKeywordProperty()) + } + } + + create.Mappings(maps) + } + _, err := create.Settings(settings). Do(context.Background()) diff --git a/pkg/es/search.go b/pkg/es/search.go index 13298ab..a417408 100644 --- a/pkg/es/search.go +++ b/pkg/es/search.go @@ -34,9 +34,17 @@ Execute an ES search. q is the actual search query given as arg to the 'search' cmd additional filters can be given as -F key=value */ -func Search(conf *cfg.Config, q string) error { - query := esdsl.NewBoolQuery(). - Must(esdsl.NewMatchQuery("message", q)) +func Search(conf *cfg.Config, queries []string) error { + query := esdsl.NewBoolQuery() + + for _, q := range queries { + parts := strings.Split(q, "=") + if len(parts) != 2 { + return fmt.Errorf("search queries must be in the form field=pattern") + } + + query.Must(esdsl.NewMatchQuery(parts[0], parts[1])) + } filters := make([]types.QueryVariant, len(conf.Filter))