add ilm forecast data collection, generalized parallel api calls

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",

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

pkg/cfg/ilm.go

@@ -39,6 +39,8 @@ type Ilm struct {
 
 	DeleteMinAge              string
 	DeleteSearchableSnapshots bool
+
+	FromPhase, Within string // forecast
 }
 
 func (cfg *Ilm) HaveHot() bool {

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 getApiData(es, wg, responses, "health")
-		go getClusterData(es, wg, responses, "info")
+		go getApiData(es, wg, responses, "info")
-		go getClusterData(es, wg, responses, "ccrstats")
+		go getApiData(es, wg, responses, "ccrstats")
-		go getClusterData(es, wg, responses, "indices")
+		go getApiData(es, wg, responses, "indices")
-		go getClusterData(es, wg, responses, "tasks")
+		go getApiData(es, wg, responses, "tasks")
 
 		if conf.Verbose {
-			go getClusterData(es, wg, responses, "stats")
+			go getApiData(es, wg, responses, "stats")
 		}
 
 		wg.Wait()

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.
 //

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

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