From 5f60696b01cfbb2fc7c49a739e05d10bbf94d7b8 Mon Sep 17 00:00:00 2001 From: Thomas von Dein Date: Thu, 25 Jun 2026 11:54:06 +0200 Subject: [PATCH] add ilm forecast data collection, generalized parallel api calls --- cmd/ilm.go | 61 +------ cmd/ilm_forecast.go | 71 +++++++++ pkg/cfg/ilm.go | 2 + pkg/es/cluster.go | 32 +--- pkg/es/cluster_util.go | 66 -------- pkg/es/ilm_forecast.go | 353 +++++++++++++++++++++++++++++++++++++++++ pkg/es/parallel.go | 152 ++++++++++++++++++ 7 files changed, 585 insertions(+), 152 deletions(-) create mode 100644 cmd/ilm_forecast.go create mode 100644 pkg/es/ilm_forecast.go create mode 100644 pkg/es/parallel.go diff --git a/cmd/ilm.go b/cmd/ilm.go index f1f4c3c..42e49fb 100644 --- a/cmd/ilm.go +++ b/cmd/ilm.go @@ -37,6 +37,7 @@ func Ilm(conf *cfg.Config) *cli.Command { IlmList(conf), IlmShow(conf), IlmCreate(conf), + IlmForecast(conf), }, } } @@ -99,66 +100,6 @@ func IlmShow(conf *cfg.Config) *cli.Command { } } -/* - * -{ - "policy": { - "phases": { - "hot": { - "min_age": "0ms", - "actions": { - "rollover": { - "max_primary_shard_size": "50gb", - "max_age": "1d", - "max_docs": 100000000 - }, - "set_priority": { - "priority": 100 - } - } - }, - "warm": { - "min_age": "7d", - "actions": { - "migrate": {}, - "shrink": { - "number_of_shards": 1 - }, - "forcemerge": { - "max_num_segments": 1 - }, - "set_priority": { - "priority": 50 - } - } - }, - "cold": { - "min_age": "30d", - "actions": { - "migrate": {}, - "searchable_snapshot": { - "snapshot_repository": "s3-logs-archive", - "force_merge_index": true - }, - "set_priority": { - "priority": 0 - } - } - }, - "delete": { - "min_age": "90d", - "actions": { - "delete": { - "delete_searchable_snapshot": true - } - } - } - } - } -} - -*/ - func IlmCreate(conf *cfg.Config) *cli.Command { return &cli.Command{ Name: "create", diff --git a/cmd/ilm_forecast.go b/cmd/ilm_forecast.go new file mode 100644 index 0000000..12e2f22 --- /dev/null +++ b/cmd/ilm_forecast.go @@ -0,0 +1,71 @@ +/* +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" + "slices" + + "codeberg.org/scip/esctl/pkg/cfg" + "codeberg.org/scip/esctl/pkg/es" + + "github.com/urfave/cli/v3" +) + +func IlmForecast(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "forecast", + Usage: "calculate index phase movements", + + Commands: []*cli.Command{ + IlmForecastList(conf), + }, + } +} + +func IlmForecastList(conf *cfg.Config) *cli.Command { + return &cli.Command{ + Name: "list", + Aliases: []string{"ls"}, + Usage: "list index rollover config", + + Flags: []cli.Flag{ + &cli.StringFlag{ + Name: "from-phase", + Usage: "which ilm phase to forecast from", + Destination: &conf.Ilm.FromPhase, + Aliases: []string{"f"}, + }, + &cli.StringFlag{ + Name: "within", + Usage: "duration withing which to forecast", + Destination: &conf.Ilm.Within, + Aliases: []string{"w"}, + }, + }, + + Action: func(ctx context.Context, cmd *cli.Command) error { + valid := []string{"hot", "warm", "cold", "frozen"} + if !slices.Contains(valid, conf.Ilm.FromPhase) { + return errors.New("invalid from phase, allowed: hot, warm, cold, frozen") + } + + return es.IlmForecastList(conf) + }, + } +} diff --git a/pkg/cfg/ilm.go b/pkg/cfg/ilm.go index d6cbd7e..1bdc079 100644 --- a/pkg/cfg/ilm.go +++ b/pkg/cfg/ilm.go @@ -39,6 +39,8 @@ type Ilm struct { DeleteMinAge string DeleteSearchableSnapshots bool + + FromPhase, Within string // forecast } func (cfg *Ilm) HaveHot() bool { diff --git a/pkg/es/cluster.go b/pkg/es/cluster.go index a33e1fe..66ba216 100644 --- a/pkg/es/cluster.go +++ b/pkg/es/cluster.go @@ -35,28 +35,8 @@ import ( "github.com/elastic/go-elasticsearch/v9/typedapi/types" ) -const ( - ResponseHealth = iota - ResponseInfo - ResponseCcr - ResponseStats - ResponseIndices - ResponseTasks -) - type ClusterIndices map[string]map[string]*types.IndicesRecord -type apiResponse struct { - error error - info *info.Response - health *health.Response - ccr *stats.Response - stats *clusterstats.Response - indices *indices.Response - tasks *tasks.Response - which int -} - func ClusterList(conf *cfg.Config) error { table := printer.NewTable(conf, 4, len(conf.Clusters)) @@ -123,14 +103,14 @@ func ClusterStatus(conf *cfg.Config) error { wg := &sync.WaitGroup{} wg.Add(gocount) - go getClusterData(es, wg, responses, "health") - go getClusterData(es, wg, responses, "info") - go getClusterData(es, wg, responses, "ccrstats") - go getClusterData(es, wg, responses, "indices") - go getClusterData(es, wg, responses, "tasks") + go getApiData(es, wg, responses, "health") + go getApiData(es, wg, responses, "info") + go getApiData(es, wg, responses, "ccrstats") + go getApiData(es, wg, responses, "indices") + go getApiData(es, wg, responses, "tasks") if conf.Verbose { - go getClusterData(es, wg, responses, "stats") + go getApiData(es, wg, responses, "stats") } wg.Wait() diff --git a/pkg/es/cluster_util.go b/pkg/es/cluster_util.go index d19eb24..57053d0 100644 --- a/pkg/es/cluster_util.go +++ b/pkg/es/cluster_util.go @@ -18,16 +18,13 @@ package es import ( "context" - "errors" "fmt" "regexp" "strconv" "strings" - "sync" "codeberg.org/scip/esctl/pkg/cfg" "codeberg.org/scip/esctl/pkg/printer" - "github.com/elastic/go-elasticsearch/v9" "github.com/elastic/go-elasticsearch/v9/typedapi/cluster/health" "github.com/elastic/go-elasticsearch/v9/typedapi/types" ) @@ -334,69 +331,6 @@ func splitArg(arg string) (string, string) { } } -func getClusterData(es *elasticsearch.TypedClient, wg *sync.WaitGroup, reschan chan apiResponse, which string) { - defer wg.Done() - - ar := apiResponse{} - arerr := errors.New("") - - switch which { - case "health": - res, err := es.Cluster.Health(). - Do(context.Background()) - - ar.health = res - ar.which = ResponseHealth - arerr = err - - case "info": - res, err := es.Info(). - Do(context.Background()) - - ar.info = res - ar.which = ResponseInfo - arerr = err - - case "ccrstats": - res, err := es.Ccr.Stats(). - Do(context.Background()) - - ar.ccr = res - ar.which = ResponseCcr - arerr = err - - case "stats": - res, err := es.Cluster.Stats(). - Do(context.Background()) - - ar.stats = res - ar.which = ResponseStats - arerr = err - - case "indices": - res, err := es.Cat.Indices(). - Do(context.Background()) - - ar.indices = &res - ar.which = ResponseIndices - arerr = err - - case "tasks": - res, err := es.Cat.Tasks(). - Do(context.Background()) - - ar.tasks = &res - ar.which = ResponseTasks - arerr = err - } - - if arerr != nil { - ar.error = fmt.Errorf("failed to get cluster health: %s", arerr) - } - - reschan <- ar -} - // recursively traverse the raw settings hash and build a flat map // consisting of the translated path and its value. // diff --git a/pkg/es/ilm_forecast.go b/pkg/es/ilm_forecast.go new file mode 100644 index 0000000..f90f52f --- /dev/null +++ b/pkg/es/ilm_forecast.go @@ -0,0 +1,353 @@ +/* +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 ( + "fmt" + "regexp" + "strconv" + "strings" + "sync" + "time" + + "codeberg.org/scip/esctl/pkg/cfg" + "codeberg.org/scip/esctl/pkg/printer" + "github.com/dustin/go-humanize" + "github.com/elastic/go-elasticsearch/v9/typedapi/cat/indices" + "github.com/elastic/go-elasticsearch/v9/typedapi/ilm/explainlifecycle" + "github.com/elastic/go-elasticsearch/v9/typedapi/ilm/getlifecycle" + "github.com/elastic/go-elasticsearch/v9/typedapi/types" +) + +var ( + matchDuration = regexp.MustCompile(`(\d+)([dhms])`) +) + +type NextPhase struct { + minage time.Duration + minsize int64 + phase string + previousHot bool +} + +type PhaseData struct { + index string + size int64 + age time.Duration + minage time.Duration + minsize int64 + currentPhase, nextPhase string +} + +/* + ilm stages: + +types.Lifecycle{ + + Policy: types.IlmPolicy{ + Phases: types.Phases{ + Hot: &types.Phase{ + Actions: &types.IlmActions{ + Rollover: &types.RolloverAction{ + MaxAge: "7d", + MaxPrimaryShardSize: "25gb", + }, + }, + MinAge: "0ms", + }, + Warm: &types.Phase{ + MinAge: "0d", + }, + Frozen: &types.Phase{ + MinAge: "14d", + }, + Delete: &types.Phase{ + MinAge: "60d", + }, + }, + }, + } + + hot for 7 days or 25Gig, then Rollover => warm 14 days => frozen 60 days => delete +*/ +func getIlmPhaseData(conf *cfg.Config) ([]PhaseData, error) { + responses := make(chan apiResponse, 3) + wg := &sync.WaitGroup{} + wg.Add(3) + + go getApiData(conf.DefaultCluster.ES(), wg, responses, "indicesbytes") + go getApiData(conf.DefaultCluster.ES(), wg, responses, "explain") + go getApiData(conf.DefaultCluster.ES(), wg, responses, "policies") + + wg.Wait() + + var ilmdetails *explainlifecycle.Response + var indicesres *indices.Response + var ilmpolicies getlifecycle.Response + + for i := 0; i < 3; i++ { + r := <-responses + + if r.error != nil { + return nil, r.error + } + + switch r.which { + case ResponseExplain: + ilmdetails = r.explainlifecycle + case ResponseIndices: + indicesres = r.indices + case ResponseLifecycle: + ilmpolicies = *r.lifecycle + } + } + + list := []PhaseData{} + + indices := make(map[string]types.IndicesRecord) + for _, index := range *indicesres { + indices[*index.Index] = index + } + + for indexName, explain := range ilmdetails.Indices { + if strings.HasPrefix(indexName, ".ds-") || + strings.HasPrefix(indexName, ".monitoring-") || + strings.HasPrefix(indexName, ".internal") { + continue + } + + // avoid types.LifecycleExplainUnmanaged casting error + switch explain.(type) { + case *types.LifecycleExplainManaged: + default: + continue + } + + explain := explain.(*types.LifecycleExplainManaged) + + if explain.Phase == nil { + continue + } + + phase := *explain.Phase + + if phase != conf.Ilm.FromPhase { + continue + } + + size, err := strconv.ParseInt(*indices[indexName].DatasetSize, 10, 64) + if err != nil { + return nil, fmt.Errorf("failed to convert index.DatasetSize to int: %w", err) + } + + age := time.Duration(*explain.AgeInMillis * int64(time.Millisecond)) + + policy := ilmpolicies[*explain.Policy].Policy + + nextPhase := findNextPhase(policy, phase) + if nextPhase == nil { + continue + } + + list = append(list, PhaseData{ + index: indexName, + size: size, + age: age, + minage: nextPhase.minage, + minsize: nextPhase.minsize, + currentPhase: conf.Ilm.FromPhase, + nextPhase: nextPhase.phase, + }) + } + + return list, nil +} + +func IlmForecastList(conf *cfg.Config) error { + phaseData, err := getIlmPhaseData(conf) + if err != nil { + return err + } + + table := printer.NewTable(conf, 7, 0) + table.Addheaders("index", "current size", "current age", "min age", "min size", "current phase", "next phase") + + for _, phase := range phaseData { + table.AddRow( + phase.index, + humanize.Bytes(uint64(phase.size)), + fmt.Sprintf("%s", phase.age), + + fmt.Sprintf("%s", phase.minage), + humanize.Bytes(uint64(phase.minsize)), + + phase.currentPhase, + phase.nextPhase, + ) + } + + table.Sort() + return table.Print() +} + +func findNextPhase(policy types.IlmPolicy, currentPhase string) *NextPhase { + // phase list to determine which comes next + order := []string{"hot", "warm", "cold", "frozen", "delete"} + + // register phases configured in the policy + phases := map[string]*types.Phase{ + "hot": nil, + "warm": nil, + "cold": nil, + "frozen": nil, + "delete": nil, + } + + // looking for a defined previous phase + start := 0 + + // register phases + for idx, phase := range order { + switch phase { + case "hot": + if policy.Phases.Hot != nil { + phases[phase] = policy.Phases.Hot + } + + case "warm": + if policy.Phases.Warm != nil { + phases[phase] = policy.Phases.Warm + } + + case "cold": + if policy.Phases.Cold != nil { + phases[phase] = policy.Phases.Cold + } + + case "frozen": + if policy.Phases.Frozen != nil { + phases[phase] = policy.Phases.Frozen + } + + case "delete": + if policy.Phases.Delete != nil { + phases[phase] = policy.Phases.Delete + } + } + + if phase == currentPhase { + start = idx + } + } + + nextPhase := &NextPhase{} + + // finally determine which phase comes next + // exception: hot, where we look for rollover rules + for idx, phase := range order { + if idx < start { + continue + } + + if phase == "hot" { + // if we're starting here, do not look for the next phase + maxage, err := time.ParseDuration(phases[phase].Actions.Rollover.MaxAge.(string)) + if err != err { + maxage = 0 + } + + var maxsize int64 = 0 + + switch val := phases[phase].Actions.Rollover.MaxPrimaryShardSize.(type) { + case int64: + maxsize = val + case string: + size, err := humanize.ParseBytes(val) + if err != nil { + panic(err) + } + + maxsize = int64(size) + } + + nextPhase.minage = maxage + nextPhase.minsize = maxsize + nextPhase.previousHot = true + + continue + } + + if phase == currentPhase { + continue // use the next one + } + + if phases[phase] == nil { + continue // skip it, since empty + } + + nextPhase.phase = phase + + if nextPhase.previousHot { + return nextPhase + } + + minage := parseDuration(phases[phase].MinAge.(string)) + + nextPhase.minage = minage + nextPhase.minsize = 0 + + return nextPhase + } + + return nil +} + +/* +We could use time.ParseDuration(), but this doesn't support days. + +We could also use github.com/xhit/go-str2duration/v2, which does +the job, but it's just another dependency, just for this little +gem. And we don't need a time.Time value. And int is good enough +for duration comparison. + +Convert a duration into an integer. Valid time units are "s", "m", "h" and "d". + + via https://codeberg.org/scip/tablizer/src/branch/main/lib/sort.go#L113 +*/ +func parseDuration(duration string) time.Duration { + + seconds := 0 + + for _, match := range matchDuration.FindAllStringSubmatch(duration, -1) { + if len(match) == 3 { + durationvalue, _ := strconv.Atoi(match[1]) + + switch match[2][0] { + case 'd': + seconds += durationvalue * 86400 + case 'h': + seconds += durationvalue * 3600 + case 'm': + seconds += durationvalue * 60 + case 's': + seconds += durationvalue + } + } + } + + return time.Duration(seconds) * time.Second +} diff --git a/pkg/es/parallel.go b/pkg/es/parallel.go new file mode 100644 index 0000000..1b0b7a5 --- /dev/null +++ b/pkg/es/parallel.go @@ -0,0 +1,152 @@ +/* +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" + "sync" + + "github.com/elastic/go-elasticsearch/v9" + "github.com/elastic/go-elasticsearch/v9/typedapi/cat/indices" + "github.com/elastic/go-elasticsearch/v9/typedapi/cat/tasks" + "github.com/elastic/go-elasticsearch/v9/typedapi/ccr/stats" + "github.com/elastic/go-elasticsearch/v9/typedapi/cluster/health" + clusterstats "github.com/elastic/go-elasticsearch/v9/typedapi/cluster/stats" + "github.com/elastic/go-elasticsearch/v9/typedapi/core/info" + "github.com/elastic/go-elasticsearch/v9/typedapi/ilm/explainlifecycle" + "github.com/elastic/go-elasticsearch/v9/typedapi/ilm/getlifecycle" + "github.com/elastic/go-elasticsearch/v9/typedapi/types/enums/bytes" +) + +const ( + ResponseHealth = iota + ResponseInfo + ResponseCcr + ResponseStats + ResponseIndices + ResponseTasks + ResponseExplain + ResponseLifecycle +) + +type apiResponse struct { + error error + info *info.Response + health *health.Response + ccr *stats.Response + stats *clusterstats.Response + indices *indices.Response + indicesbytes *indices.Response + tasks *tasks.Response + lifecycle *getlifecycle.Response + explainlifecycle *explainlifecycle.Response + which int +} + +func getApiData(es *elasticsearch.TypedClient, wg *sync.WaitGroup, + reschan chan apiResponse, which string) { + defer wg.Done() + + ar := apiResponse{} + arerr := errors.New("") + + switch which { + case "health": + res, err := es.Cluster.Health(). + Do(context.Background()) + + ar.health = res + ar.which = ResponseHealth + arerr = err + + case "info": + res, err := es.Info(). + Do(context.Background()) + + ar.info = res + ar.which = ResponseInfo + arerr = err + + case "ccrstats": + res, err := es.Ccr.Stats(). + Do(context.Background()) + + ar.ccr = res + ar.which = ResponseCcr + arerr = err + + case "stats": + res, err := es.Cluster.Stats(). + Do(context.Background()) + + ar.stats = res + ar.which = ResponseStats + arerr = err + + case "indices": + res, err := es.Cat.Indices(). + Do(context.Background()) + + ar.indices = &res + ar.which = ResponseIndices + arerr = err + + case "indicesbytes": + res, err := es.Cat. + Indices(). + Bytes(bytes.Bytes{Name: "b"}). + Do(context.Background()) + + ar.indices = &res + ar.which = ResponseIndices + arerr = err + + case "tasks": + res, err := es.Cat.Tasks(). + Do(context.Background()) + + ar.tasks = &res + ar.which = ResponseTasks + arerr = err + + case "explain": + res, err := es.Ilm. + ExplainLifecycle("_all"). + Do(context.Background()) + + ar.explainlifecycle = res + ar.which = ResponseExplain + arerr = err + + case "policies": + res, err := es.Ilm. + GetLifecycle(). + Do(context.Background()) + + ar.lifecycle = &res + ar.which = ResponseLifecycle + arerr = err + } + + if arerr != nil { + ar.error = fmt.Errorf("failed to get data from API: %s", arerr) + } + + reschan <- ar +}