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