mirror of
https://codeberg.org/scip/esctl.git
synced 2026-08-24 08:04:17 +02:00
531 lines
13 KiB
Go
531 lines
13 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 (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"maps"
|
|
"slices"
|
|
"strings"
|
|
|
|
"codeberg.org/scip/esctl/pkg/cfg"
|
|
"codeberg.org/scip/esctl/pkg/printer"
|
|
"github.com/alecthomas/repr"
|
|
"github.com/charmbracelet/lipgloss"
|
|
"github.com/elastic/go-elasticsearch/v9/typedapi/esdsl"
|
|
"github.com/elastic/go-elasticsearch/v9/typedapi/ilm/putlifecycle"
|
|
"github.com/elastic/go-elasticsearch/v9/typedapi/types"
|
|
)
|
|
|
|
func IlmRetry(conf *cfg.Config, index string) error {
|
|
_, err := conf.DefaultCluster.ES().Ilm.Retry(index).
|
|
Do(context.Background())
|
|
if err != nil {
|
|
return fmt.Errorf("failed to retry ilm: %w", esErrorString(err))
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func IlmStatus(conf *cfg.Config) error {
|
|
res, err := conf.DefaultCluster.ES().Ilm.GetStatus().
|
|
Do(context.Background())
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get ilm status: %w", esErrorString(err))
|
|
}
|
|
|
|
fmt.Println(res.OperationMode.Name)
|
|
|
|
return nil
|
|
}
|
|
|
|
func IlmNames(conf *cfg.Config) ([]string, error) {
|
|
res, err := conf.DefaultCluster.ES().Ilm.GetLifecycle().
|
|
Do(context.Background())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get ilm policies: %w", esErrorString(err))
|
|
}
|
|
|
|
names := slices.Collect(maps.Keys(res))
|
|
|
|
return names, nil
|
|
}
|
|
|
|
func IlmList(conf *cfg.Config, pattern string) error {
|
|
ilm := conf.DefaultCluster.ES().Ilm.GetLifecycle()
|
|
|
|
if pattern != "" {
|
|
ilm.FilterPath(pattern)
|
|
}
|
|
|
|
res, err := ilm.Do(context.Background())
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get ilm policies: %w", esErrorString(err))
|
|
}
|
|
|
|
if conf.Debug {
|
|
repr.Println(res)
|
|
}
|
|
|
|
table := printer.NewTable(conf).WithSize(5, 0)
|
|
table.Addheaders("ilm policy", "hot", "warm", "frozen", "delete")
|
|
|
|
for name, ilm := range res {
|
|
table.AddRow(name,
|
|
ilmPhaseString(ilm.Policy.Phases.Hot, true),
|
|
ilmPhaseString(ilm.Policy.Phases.Warm, true),
|
|
ilmPhaseString(ilm.Policy.Phases.Frozen, true),
|
|
ilmPhaseString(ilm.Policy.Phases.Delete, true),
|
|
)
|
|
}
|
|
|
|
table.Sort()
|
|
|
|
return table.Print()
|
|
}
|
|
|
|
func IlmShow(conf *cfg.Config, policy string) error {
|
|
res, err := conf.DefaultCluster.ES().Ilm.GetLifecycle().
|
|
Policy(policy).
|
|
Do(context.Background())
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get ilm status: %w", esErrorString(err))
|
|
}
|
|
|
|
if conf.Debug {
|
|
repr.Println(res)
|
|
}
|
|
|
|
ilm, exists := res[policy]
|
|
if !exists {
|
|
return errors.New("no ilm policy retrieved")
|
|
}
|
|
|
|
if conf.Ilm.Tree {
|
|
return IlmShowTree(conf, ilm.Policy)
|
|
}
|
|
|
|
table := printer.NewTable(conf).WithSize(2, 5)
|
|
table.Addheaders("ilm policy setting", "value")
|
|
|
|
table.Entries = [][]any{
|
|
{"policy", policy},
|
|
{"hot phase", ilmPhaseString(ilm.Policy.Phases.Hot, false)},
|
|
{"warm phase", ilmPhaseString(ilm.Policy.Phases.Warm, false)},
|
|
{"frozen phase", ilmPhaseString(ilm.Policy.Phases.Frozen, false)},
|
|
{"delete phase", ilmPhaseString(ilm.Policy.Phases.Delete, false)},
|
|
}
|
|
|
|
return table.Print()
|
|
}
|
|
|
|
// Visualize phases or a ILM policy, taking into account that the
|
|
// minAge for the current phases is set in the next phase
|
|
func IlmShowTree(conf *cfg.Config, ilm types.IlmPolicy) error {
|
|
indent := ""
|
|
|
|
table := printer.NewTable(conf).WithSize(4, 0)
|
|
table.Addheaders("phase", "min age", "min size", "snapshot repo")
|
|
|
|
for _, phase := range IlmPhaseOrder {
|
|
if phase == "hot" {
|
|
fmt.Printf("%s%s phase:\n%s rollover after %s\n",
|
|
indent, phase,
|
|
indent, formatDuration(parseDuration(ilm.Phases.Hot.Actions.Rollover.MaxAge.(string))),
|
|
)
|
|
|
|
if ilm.Phases.Hot.Actions.Rollover.MaxPrimaryShardSize != "" {
|
|
fmt.Printf("%s rollover when storage > %s\n",
|
|
indent, ilm.Phases.Hot.Actions.Rollover.MaxPrimaryShardSize)
|
|
}
|
|
|
|
if ilm.Phases.Hot.Actions.Rollover.MaxDocs != nil {
|
|
fmt.Printf("%s rollover docs > %d\n",
|
|
indent, *ilm.Phases.Hot.Actions.Rollover.MaxDocs)
|
|
}
|
|
|
|
indent += " "
|
|
|
|
continue
|
|
}
|
|
|
|
current := getIlmCurrentPhase(ilm, phase)
|
|
|
|
if current == nil {
|
|
continue
|
|
}
|
|
|
|
next := findNextPhase(ilm, phase)
|
|
|
|
fmt.Printf("%s%s phase:\n", indent, phase)
|
|
|
|
if next != nil {
|
|
fmt.Printf("%s rollover after %s\n",
|
|
indent, formatDuration(next.minage),
|
|
)
|
|
|
|
if current.Actions.SearchableSnapshot != nil {
|
|
fmt.Printf("%s roll to snapshot repo: %s\n",
|
|
indent, current.Actions.SearchableSnapshot.SnapshotRepository)
|
|
}
|
|
|
|
if current.Actions.Forcemerge != nil {
|
|
fmt.Printf("%s force merge segments: %d\n",
|
|
indent, current.Actions.Forcemerge.MaxNumSegments)
|
|
}
|
|
|
|
if current.Actions.Shrink != nil {
|
|
fmt.Printf("%s shrink shards: %d\n",
|
|
indent, *current.Actions.Shrink.NumberOfShards)
|
|
}
|
|
|
|
if current.Actions.SetPriority != nil {
|
|
fmt.Printf("%s priority: %d\n",
|
|
indent, *current.Actions.SetPriority.Priority)
|
|
}
|
|
|
|
if current.Actions.Delete != nil && current.Actions.Delete.DeleteSearchableSnapshot != nil {
|
|
fmt.Printf("%s delete searchable snapshots: %t\n",
|
|
indent, *current.Actions.Delete.DeleteSearchableSnapshot)
|
|
}
|
|
}
|
|
|
|
if phase == "delete" {
|
|
fmt.Printf("%s delete immediately\n", indent)
|
|
}
|
|
|
|
indent += " "
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func getIlmCurrentPhase(policy types.IlmPolicy, currentPhase string) *types.Phase {
|
|
switch currentPhase {
|
|
case "hot":
|
|
return policy.Phases.Hot
|
|
case "warm":
|
|
return policy.Phases.Warm
|
|
case "cold":
|
|
return policy.Phases.Cold
|
|
case "frozen":
|
|
return policy.Phases.Frozen
|
|
default:
|
|
return policy.Phases.Delete
|
|
}
|
|
}
|
|
|
|
func ilmPhaseString(phase *types.Phase, short bool) string {
|
|
if phase == nil {
|
|
return ""
|
|
}
|
|
|
|
out := []string{fmt.Sprintf("move-after:%s", phase.MinAge)}
|
|
|
|
if short {
|
|
return strings.Join(out, ",")
|
|
}
|
|
|
|
if phase.Actions.Rollover != nil {
|
|
if phase.Actions.Rollover.MaxAge != "" {
|
|
out = append(out, fmt.Sprintf("rollover:%s", phase.Actions.Rollover.MaxAge))
|
|
if phase.Actions.Rollover.MaxPrimaryShardSize != nil {
|
|
out = append(out, fmt.Sprintf("shardsize:%s", phase.Actions.Rollover.MaxPrimaryShardSize))
|
|
}
|
|
}
|
|
}
|
|
|
|
if phase.Actions.Delete != nil {
|
|
out = append(out, fmt.Sprintf("delete-searchable-snapshot:%t", *phase.Actions.Delete.DeleteSearchableSnapshot))
|
|
}
|
|
|
|
if phase.Actions.SearchableSnapshot != nil {
|
|
out = append(out, "searchable-snapshot:"+phase.Actions.SearchableSnapshot.SnapshotRepository)
|
|
}
|
|
|
|
return strings.Join(out, ",")
|
|
}
|
|
|
|
func IlmExplain(conf *cfg.Config, index string) error {
|
|
res, err := conf.DefaultCluster.ES().Ilm.ExplainLifecycle(index).
|
|
Do(context.Background())
|
|
if err != nil {
|
|
return fmt.Errorf("failed to get ilm state: %w", esErrorString(err))
|
|
}
|
|
|
|
slog.Debug("ilm status", "ilm", res)
|
|
|
|
explain, exists := res.Indices[index]
|
|
if !exists {
|
|
return errors.New("no ilm status retrieved")
|
|
}
|
|
|
|
ilm := explain.(*types.LifecycleExplainManaged)
|
|
|
|
table := printer.NewTable(conf).WithSize(2, 0)
|
|
table.Addheaders("ilm status field", "value")
|
|
|
|
info := ""
|
|
|
|
if len(ilm.StepInfo["reason"]) > 0 {
|
|
err = json.Unmarshal(ilm.StepInfo["reason"], &info)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to unmarshal step info: %w", err)
|
|
}
|
|
}
|
|
|
|
table.Entries = [][]any{
|
|
{"index", ilm.Index},
|
|
{"ilm policy", *ilm.Policy},
|
|
{"action", *ilm.Action},
|
|
{"age", ilm.Age},
|
|
{"managed", ilm.Managed},
|
|
{"phase", *ilm.Phase},
|
|
{"phase execution", ilmPhaseString(ilm.PhaseExecution.PhaseDefinition, false)},
|
|
{"step", *ilm.Step},
|
|
}
|
|
|
|
if ilm.FailedStep != nil {
|
|
table.AddRow("failed step", *ilm.FailedStep)
|
|
table.AddRow("failed step retry count", ilm.FailedStepRetryCount)
|
|
}
|
|
|
|
if err := table.Print(); err != nil {
|
|
return err
|
|
}
|
|
|
|
if info != "" {
|
|
var bold = lipgloss.NewStyle().Bold(true)
|
|
|
|
fmt.Printf("\n%s:\n%s\n",
|
|
bold.Render("Step Reason"),
|
|
info)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func IlmCreate(conf *cfg.Config, policyname string) error {
|
|
var policy *types.IlmPolicy = nil
|
|
|
|
res, err := conf.DefaultCluster.ES().Ilm.GetLifecycle().
|
|
Policy(policyname).
|
|
Do(context.Background())
|
|
if err == nil {
|
|
if conf.Debug {
|
|
repr.Println(res)
|
|
}
|
|
|
|
ilm, exists := res[policyname]
|
|
if !exists {
|
|
return errors.New("no ilm policy retrieved")
|
|
}
|
|
|
|
policy = &ilm.Policy
|
|
}
|
|
|
|
ilm := conf.DefaultCluster.ES().Ilm.PutLifecycle(policyname)
|
|
cfg := conf.Ilm
|
|
|
|
var phases types.PhasesVariant = esdsl.NewPhases()
|
|
|
|
if cfg.HaveHot() {
|
|
hot := types.Phase{}
|
|
|
|
var actions types.IlmActionsVariant = esdsl.NewIlmActions()
|
|
|
|
rollover := new(types.RolloverAction{})
|
|
haveroll := false
|
|
|
|
if policy != nil {
|
|
// update
|
|
actions = policy.Phases.Hot.Actions
|
|
|
|
if policy.Phases.Hot.Actions.Rollover != nil {
|
|
rollover = policy.Phases.Hot.Actions.Rollover
|
|
haveroll = true
|
|
}
|
|
}
|
|
|
|
if cfg.HotMinAge != "" {
|
|
hot.MinAge = cfg.HotMinAge
|
|
}
|
|
|
|
if cfg.HotRolloverMaxAge != "" {
|
|
rollover.MaxAge = cfg.HotRolloverMaxAge
|
|
haveroll = true
|
|
}
|
|
|
|
if cfg.HotRolloverMaxDocs != 0 {
|
|
rollover.MaxDocs = &cfg.HotRolloverMaxDocs
|
|
haveroll = true
|
|
}
|
|
|
|
if cfg.HotRolloverMaxPrimaryShardSize != "" {
|
|
rollover.MaxPrimaryShardSize = &cfg.HotRolloverMaxPrimaryShardSize
|
|
}
|
|
|
|
if haveroll {
|
|
actions.IlmActionsCaster().Rollover = rollover
|
|
}
|
|
|
|
hot.Actions = actions.IlmActionsCaster()
|
|
phases.PhasesCaster().Hot = &hot
|
|
} else if policy != nil {
|
|
// update
|
|
phases.PhasesCaster().Hot = policy.Phases.Hot
|
|
}
|
|
|
|
if cfg.HaveWarm() {
|
|
warm := types.Phase{}
|
|
|
|
var actions types.IlmActionsVariant = esdsl.NewIlmActions()
|
|
|
|
if policy != nil {
|
|
// update
|
|
actions = policy.Phases.Warm.Actions
|
|
}
|
|
|
|
if cfg.WarmForceMerge != 0 {
|
|
actions.IlmActionsCaster().Forcemerge = &types.ForceMergeAction{MaxNumSegments: cfg.WarmForceMerge}
|
|
}
|
|
|
|
if cfg.WarmMinAge != "" {
|
|
warm.MinAge = cfg.WarmMinAge
|
|
}
|
|
|
|
if cfg.WarmPriority != 0 {
|
|
actions.IlmActionsCaster().SetPriority = &types.SetPriorityAction{Priority: &cfg.WarmPriority}
|
|
}
|
|
|
|
if cfg.WarmShrinkShards != 0 {
|
|
actions.IlmActionsCaster().Shrink = &types.ShrinkAction{NumberOfShards: &cfg.WarmShrinkShards}
|
|
}
|
|
|
|
warm.Actions = actions.IlmActionsCaster()
|
|
phases.PhasesCaster().Warm = &warm
|
|
} else if policy != nil && policy.Phases.Warm != nil {
|
|
// update
|
|
phases.PhasesCaster().Warm = policy.Phases.Warm
|
|
}
|
|
|
|
if cfg.HaveCold() {
|
|
cold := types.Phase{}
|
|
|
|
var actions types.IlmActionsVariant = esdsl.NewIlmActions()
|
|
|
|
if policy != nil {
|
|
// update
|
|
actions = policy.Phases.Cold.Actions
|
|
}
|
|
|
|
if cfg.ColdForceMerge != 0 {
|
|
actions.IlmActionsCaster().Forcemerge = &types.ForceMergeAction{MaxNumSegments: cfg.ColdForceMerge}
|
|
}
|
|
|
|
if cfg.ColdMinAge != "" {
|
|
cold.MinAge = cfg.ColdMinAge
|
|
}
|
|
|
|
if cfg.ColdPriority != 0 {
|
|
actions.IlmActionsCaster().SetPriority = &types.SetPriorityAction{Priority: &cfg.ColdPriority}
|
|
}
|
|
|
|
if cfg.ColdSearchableSnapshotRepo != "" {
|
|
actions.IlmActionsCaster().SearchableSnapshot =
|
|
&types.SearchableSnapshotAction{SnapshotRepository: cfg.ColdSearchableSnapshotRepo}
|
|
}
|
|
|
|
cold.Actions = actions.IlmActionsCaster()
|
|
phases.PhasesCaster().Cold = &cold
|
|
} else if policy != nil && policy.Phases.Cold != nil {
|
|
// update
|
|
phases.PhasesCaster().Cold = policy.Phases.Cold
|
|
}
|
|
|
|
if cfg.HaveFrozen() {
|
|
froze := types.Phase{}
|
|
|
|
var actions types.IlmActionsVariant = esdsl.NewIlmActions()
|
|
|
|
if policy != nil {
|
|
// update
|
|
actions = policy.Phases.Frozen.Actions
|
|
}
|
|
|
|
if cfg.FrozenMinAge != "" {
|
|
froze.MinAge = cfg.FrozenMinAge
|
|
}
|
|
|
|
if cfg.FrozenSearchableSnapshotRepo != "" {
|
|
actions.IlmActionsCaster().SearchableSnapshot =
|
|
&types.SearchableSnapshotAction{SnapshotRepository: cfg.FrozenSearchableSnapshotRepo}
|
|
}
|
|
|
|
froze.Actions = actions.IlmActionsCaster()
|
|
phases.PhasesCaster().Frozen = &froze
|
|
} else if policy != nil && policy.Phases.Frozen != nil {
|
|
// update
|
|
phases.PhasesCaster().Frozen = policy.Phases.Frozen
|
|
}
|
|
|
|
if cfg.HaveDelete() {
|
|
del := types.Phase{}
|
|
delete := types.DeleteAction{}
|
|
|
|
var actions types.IlmActionsVariant = esdsl.NewIlmActions()
|
|
|
|
if policy != nil {
|
|
// update
|
|
actions = policy.Phases.Delete.Actions
|
|
}
|
|
|
|
if cfg.DeleteMinAge != "" {
|
|
del.MinAge = cfg.DeleteMinAge
|
|
}
|
|
|
|
if cfg.DeleteSearchableSnapshots {
|
|
delete.DeleteSearchableSnapshot = &cfg.DeleteSearchableSnapshots
|
|
}
|
|
|
|
actions.IlmActionsCaster().Delete = &delete
|
|
del.Actions = actions.IlmActionsCaster()
|
|
phases.PhasesCaster().Delete = &del
|
|
} else if policy != nil && policy.Phases.Delete != nil {
|
|
// update
|
|
phases.PhasesCaster().Delete = policy.Phases.Delete
|
|
}
|
|
|
|
put := new(putlifecycle.Request{})
|
|
newpolicy := new(types.IlmPolicy{})
|
|
newpolicy.IlmPolicyCaster().Phases = *phases.PhasesCaster()
|
|
put.Policy = newpolicy
|
|
|
|
ilm.Request(put)
|
|
|
|
_, err = ilm.Do(context.Background())
|
|
if err != nil {
|
|
return fmt.Errorf("failed create ilm policy: %w", esErrorString(err))
|
|
}
|
|
|
|
return nil
|
|
}
|