mirror of
https://codeberg.org/scip/esctl.git
synced 2026-08-24 13:14:18 +02:00
399 lines
8.6 KiB
Go
399 lines
8.6 KiB
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).WithSize(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
|
|
}
|