esctl/pkg/es/ilm_forecast.go

/*
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])`)
	IlmPhaseOrder = []string{"hot", "warm", "cold", "frozen", "delete"}
)

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 IlmForecastList(conf *cfg.Config) error {
	phaseData, err := getIlmPhaseData(conf)
	if err != nil {
		return err
	}

	table := printer.NewTable(conf, 8, 0)
	table.Addheaders("index", "current size", "current age", "virtual age", "min age", "min size", "current phase", "next phase")

	for _, phase := range phaseData {
		virtualAge := virtualAge(&phase)

		table.AddRow(
			phase.index,
			humanize.Bytes(uint64(phase.size)),
			phase.age.String(),
			virtualAge.String(),

			phase.minage.String(),
			humanize.Bytes(uint64(phase.minsize)),

			phase.currentPhase,
			phase.nextPhase,
		)
	}

	table.Sort()
	return table.Print()
}

// Loop  over   all  current  index   phases  of  the   current  phase
// (conf.Ilm.FromPhase). Calculate virtualAge  if phase.minsize >0. If
// the actual age is smaller than the virtual age, use this as a base,
// otherwise use virtual  age. Then look if the  rollover would happen
// within the phase.minage window  (conf.Ilm.Within + current age) and
// add the size.
func IlmForecastShow(conf *cfg.Config) error {
	phaseData, err := getIlmPhaseData(conf)
	if err != nil {
		return err
	}

	within := parseDuration(conf.Ilm.Within)
	var toBeFreed int64 = 0

	for _, phase := range phaseData {
		age := virtualAge(&phase)

		if age < phase.age {
			age = phase.age
		}

		if age+within >= phase.minage {
			toBeFreed += phase.size
		}
	}

	fmt.Printf("%s bytes of data in %s phase will be rolled within %s to the next phase\n",
		humanize.Bytes(uint64(toBeFreed)),
		conf.Ilm.FromPhase,
		within)

	return nil
}

func virtualAge(phase *PhaseData) time.Duration {
	if phase.minsize == 0 {
		return time.Duration(0)
	}

	return phase.minage * time.Duration(100*phase.size/phase.minsize) / 100
}

// Retrieve all index, ilm-explain and ilm-policies in parallel
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.indicesbytes
		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 registerPhases(policy types.IlmPolicy, currentPhase string) (map[string]*types.Phase, int) {
	// 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 IlmPhaseOrder {
		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
		}
	}

	return phases, start
}

func findNextPhase(policy types.IlmPolicy, currentPhase string) *NextPhase {
	// phase list to determine which comes next
	phases, start := registerPhases(policy, currentPhase)

	nextPhase := &NextPhase{}

	// finally determine which phase comes next
	// exception: hot, where we look for rollover rules
	for idx, phase := range IlmPhaseOrder {
		if idx < start {
			continue
		}

		if phase == "hot" {
			// if we're starting here, do not look for the next phase
			maxage := parseDuration(phases[phase].Actions.Rollover.MaxAge.(string))

			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
}