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

type NextPhase struct {
	minage      time.Duration
	minsize     int64
	phase       string
	previousHot bool
}

type PhaseData struct {
	index                   string
	policy                  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, filter string) error {
	phaseData, err := getIlmPhaseData(conf)
	if err != nil {
		return err
	}

	flt := regexp.MustCompile(filter)

	minage := time.Duration(0)
	if conf.Ilm.MinAge != "" {
		minage = parseDuration(conf.Ilm.MinAge)
	}

	headers := []string{"index", "current size", "current age",
		"virtual age", "min age", "min size", "current phase", "next phase"}
	if conf.Verbose {
		headers = append(headers, "ilm policy")
	}

	table := printer.NewTable(conf, len(headers), 0)
	table.Addheaders(headers...)

	for _, phase := range phaseData {
		if filter != "" && !flt.MatchString(phase.index) {
			continue
		}

		if conf.Ilm.MinAge != "" && phase.age < minage {
			continue
		}

		virtualAge := virtualAge(&phase)

		row := []any{
			phase.index,
			printer.Bytes(phase.size),
			formatDuration(phase.age),
			formatDuration(virtualAge),
			formatDuration(phase.minage),
			printer.Bytes(phase.minsize),
			phase.currentPhase,
			phase.nextPhase,
		}

		if conf.Verbose {
			row = append(row, phase.policy)
		}

		table.AddRow(row...)
	}

	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 := max(virtualAge(&phase), 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 := new(sync.WaitGroup{})

	wg.Go(func() {
		getApiData(conf, conf.DefaultCluster.ES(), responses, "indicesbytes")
	})

	wg.Go(func() {
		getApiData(conf, conf.DefaultCluster.ES(), responses, "explain")
	})

	wg.Go(func() {
		getApiData(conf, conf.DefaultCluster.ES(), responses, "policies")
	})

	wg.Wait()

	var (
		ilmdetails  *explainlifecycle.Response
		indicesres  *indices.Response
		ilmpolicies getlifecycle.Response
	)

	for range 3 {
		res := <-responses

		if res.error != nil {
			return nil, res.error
		}

		switch res.which {
		case ResponseExplain:
			ilmdetails = res.explainlifecycle
		case ResponseIndices:
			indicesres = res.indicesbytes
		case ResponseLifecycle:
			ilmpolicies = *res.lifecycle
		}
	}

	list := []PhaseData{}

	indices := make(map[string]types.IndicesRecord)
	for _, index := range *indicesres {
		indices[*index.Index] = index
	}

	for indexName, explain := range ilmdetails.Indices {
		if !conf.Hidden && strings.HasPrefix(indexName, ".") {
			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,
			policy:       *explain.Policy,
			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 := new(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
}