add ilm management support (#43)

This commit is contained in:
T. von Dein
2026-06-22 13:52:02 +02:00
parent 802217158c
commit 9b5ba0775b
9 changed files with 700 additions and 1 deletions

View File

@@ -2,7 +2,7 @@ matrix:
platform: platform:
- linux/amd64 - linux/amd64
goversion: goversion:
- 1.25 - 1.25.8
labels: labels:
platform: ${platform} platform: ${platform}

View File

@@ -31,6 +31,7 @@ const (
Crole Crole
Ccluster Ccluster
Capi Capi
Cilm
) )
func complete(cmd *cli.Command, what int) { func complete(cmd *cli.Command, what int) {
@@ -61,6 +62,8 @@ func complete(cmd *cli.Command, what int) {
list, err = es.DatastreamNames(conf) list, err = es.DatastreamNames(conf)
case Capi: case Capi:
list = es.ApiPathNames() list = es.ApiPathNames()
case Cilm:
list, err = es.IlmNames(conf)
} }
if err != nil { if err != nil {

View File

@@ -197,3 +197,24 @@ func DatastreamRollover(conf *cfg.Config) *cli.Command {
}, },
} }
} }
func DatastreamIlm(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "ilm",
Usage: "show ilm status",
UsageText: "ds ilm <name>",
ShellComplete: func(ctx context.Context, cmd *cli.Command) {
complete(cmd, Cdatastream)
},
Action: func(ctx context.Context, cmd *cli.Command) error {
ds := cmd.Args().Get(0)
if ds == "" {
return errors.New("no ds name specified")
}
return es.IlmExplain(conf, ds)
},
}
}

277
cmd/ilm.go Normal file
View File

@@ -0,0 +1,277 @@
/*
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"
"codeberg.org/scip/esctl/pkg/cfg"
"codeberg.org/scip/esctl/pkg/es"
"github.com/urfave/cli/v3"
)
func Ilm(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "ilm",
Usage: "manage index lifecycle",
Commands: []*cli.Command{
IlmRetry(conf),
IlmStatus(conf),
IlmList(conf),
IlmShow(conf),
IlmCreate(conf),
},
}
}
func IlmRetry(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "retry",
Usage: "retry applying an ILM profile to an index",
UsageText: "retry <index>",
Action: func(ctx context.Context, cmd *cli.Command) error {
return es.IlmRetry(conf, cmd.Args().Get(0))
},
}
}
func IlmStatus(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "status",
Usage: "get the current index lifecycle management status",
Action: func(ctx context.Context, cmd *cli.Command) error {
return es.IlmStatus(conf)
},
}
}
func IlmList(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "list",
Aliases: []string{"ls"},
Usage: "list index lifecycle policies",
UsageText: "list [<policy-pattern>]",
Action: func(ctx context.Context, cmd *cli.Command) error {
return es.IlmList(conf, cmd.Args().Get(0))
},
}
}
func IlmShow(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "show",
Aliases: []string{"sh"},
Usage: "show details about an index lifecycle policy",
UsageText: "show <policy>",
Action: func(ctx context.Context, cmd *cli.Command) error {
policy := cmd.Args().Get(0)
if policy == "" {
return errors.New("no policy specified")
}
return es.IlmShow(conf, cmd.Args().Get(0))
},
ShellComplete: func(ctx context.Context, cmd *cli.Command) {
complete(cmd, Cilm)
},
}
}
/*
*
{
"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",
Aliases: []string{"+"},
Usage: "create a index lifecycle policy",
UsageText: "create [options] <policy>",
Flags: []cli.Flag{
&cli.StringFlag{
Name: "description",
Usage: "policy description",
Destination: &conf.Ilm.Description,
Aliases: []string{"D"},
},
// hot
&cli.StringFlag{
Name: "hot-min-age",
Usage: "hot phase: min age",
Destination: &conf.Ilm.HotMinAge,
},
&cli.StringFlag{
Name: "hot-rollover-max-age",
Usage: "hot phase: rollover after",
Destination: &conf.Ilm.HotRolloverMaxAge,
},
&cli.Int64Flag{
Name: "hot-rollover-max-primary-shard-size",
Usage: "hot phase: max primary shard size",
Destination: &conf.Ilm.HotRolloverMaxPrimaryShardSize,
},
&cli.Int64Flag{
Name: "hot-rollover-max-docs",
Usage: "hot phase: max docs",
Destination: &conf.Ilm.HotRolloverMaxDocs,
},
&cli.BoolFlag{
Name: "hot-readonly",
Usage: "hot phase: make index ro before rollover",
Destination: &conf.Ilm.HotReadonly,
},
// warm
&cli.StringFlag{
Name: "warm-min-age",
Usage: "warm phase: min age",
Destination: &conf.Ilm.WarmMinAge,
},
&cli.IntFlag{
Name: "warm-force-merge-segments",
Usage: "warm phase: force merge segments per shard",
Destination: &conf.Ilm.WarmForceMerge,
},
&cli.IntFlag{
Name: "warm-shrink-shards",
Usage: "warm phase: shrink on number of shards",
Destination: &conf.Ilm.WarmShrinkShards,
},
&cli.IntFlag{
Name: "warm-priority",
Usage: "warm phase: priority",
Destination: &conf.Ilm.WarmPriority,
},
// cold
&cli.StringFlag{
Name: "cold-min-age",
Usage: "cold phase: min age",
Destination: &conf.Ilm.ColdMinAge,
},
&cli.StringFlag{
Name: "cold-searchable-snapshot-repo",
Usage: "cold phase: searchable snapshot repo",
Destination: &conf.Ilm.ColdSearchableSnapshotRepo,
},
&cli.IntFlag{
Name: "cold-force-merge-segments",
Usage: "cold phase: force merge segments per shard",
Destination: &conf.Ilm.ColdForceMerge,
},
&cli.IntFlag{
Name: "cold-priority",
Usage: "cold phase: priority",
Destination: &conf.Ilm.ColdPriority,
},
// frozen
&cli.StringFlag{
Name: "frozen-min-age",
Usage: "frozen phase: min age",
Destination: &conf.Ilm.FrozenMinAge,
},
// delete
&cli.StringFlag{
Name: "delete-min-age",
Usage: "delete phase: min age",
Destination: &conf.Ilm.DeleteMinAge,
},
&cli.BoolFlag{
Name: "delete-searchable-snapshots",
Usage: "delete phase: delete searchable snapshots",
Destination: &conf.Ilm.DeleteSearchableSnapshots,
},
},
Action: func(ctx context.Context, cmd *cli.Command) error {
policy := cmd.Args().Get(0)
if policy == "" {
return errors.New("no policy name specified")
}
return es.IlmCreate(conf, cmd.Args().Get(0))
},
}
}

View File

@@ -42,6 +42,7 @@ func Index(conf *cfg.Config) *cli.Command {
IndexAllocation(conf), IndexAllocation(conf),
IndexModify(conf), IndexModify(conf),
IndexFields(conf), IndexFields(conf),
IndexIlm(conf),
// sub commands // sub commands
IndexAlias(conf), IndexAlias(conf),
@@ -300,3 +301,24 @@ func IndexFields(conf *cfg.Config) *cli.Command {
}, },
} }
} }
func IndexIlm(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "ilm",
Usage: "show ilm status",
UsageText: "index ilm <index>",
ShellComplete: func(ctx context.Context, cmd *cli.Command) {
complete(cmd, Cindex)
},
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
if index == "" {
return errors.New("no index specified")
}
return es.IlmExplain(conf, index)
},
}
}

View File

@@ -98,6 +98,7 @@ func Main() int {
Cluster(conf), Cluster(conf),
Ccr(conf), Ccr(conf),
Index(conf), Index(conf),
Ilm(conf),
Datastream(conf), Datastream(conf),
Shard(conf), Shard(conf),
Snapshot(conf), Snapshot(conf),

View File

@@ -110,6 +110,8 @@ type Config struct {
DryRun bool // rollover: -n DryRun bool // rollover: -n
Tag string // api ls: -t Tag string // api ls: -t
Ilm Ilm // ilm create
} }
func NewConfig() *Config { func NewConfig() *Config {

72
pkg/cfg/ilm.go Normal file
View File

@@ -0,0 +1,72 @@
/*
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 cfg
type Ilm struct {
Description string
HotMinAge string
HotRolloverMaxAge string
HotRolloverMaxPrimaryShardSize int64
HotRolloverMaxDocs int64
HotReadonly bool
WarmMinAge string
WarmForceMerge int
WarmShrinkShards int
WarmPriority int
ColdMinAge string
ColdSearchableSnapshotRepo string
ColdForceMerge int
ColdPriority int
FrozenMinAge string
DeleteMinAge string
DeleteSearchableSnapshots bool
}
func (cfg *Ilm) HaveHot() bool {
return cfg.HotMinAge != "" ||
cfg.HotRolloverMaxAge != "" ||
cfg.HotRolloverMaxDocs > 0 ||
cfg.HotRolloverMaxPrimaryShardSize > 0 ||
cfg.HotReadonly
}
func (cfg *Ilm) HaveWarm() bool {
return cfg.WarmMinAge != "" ||
cfg.WarmForceMerge > 0 ||
cfg.WarmShrinkShards > 0 ||
cfg.WarmPriority > 0
}
func (cfg *Ilm) HaveCold() bool {
return cfg.ColdMinAge != "" ||
cfg.ColdSearchableSnapshotRepo != "" ||
cfg.ColdForceMerge > 0 ||
cfg.ColdPriority > 0
}
func (cfg *Ilm) HaveFrozen() bool {
return cfg.FrozenMinAge != ""
}
func (cfg *Ilm) HaveDelete() bool {
return cfg.DeleteSearchableSnapshots || cfg.DeleteMinAge != ""
}

301
pkg/es/ilm.go Normal file
View File

@@ -0,0 +1,301 @@
/*
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"
"log/slog"
"strings"
"codeberg.org/scip/esctl/pkg/cfg"
"codeberg.org/scip/esctl/pkg/printer"
"github.com/alecthomas/repr"
"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: %s", 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: %s", 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: %s", esErrorString(err))
}
names := make([]string, len(res))
idx := 0
for name := range res {
names[idx] = name
}
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: %s", esErrorString(err))
}
if conf.Debug {
repr.Println(res)
}
table := printer.NewTable(conf, 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: %s", esErrorString(err))
}
if conf.Debug {
repr.Println(res)
}
ilm, exists := res[policy]
if !exists {
return errors.New("no ilm policy retrieved")
}
table := printer.NewTable(conf, 2, 5)
table.Addheaders("ilm policy setting", "value")
table.Entries = [][]string{
{"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()
}
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, fmt.Sprintf("searchable-snapshot:%s", 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: %s", 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, 2, 0)
table.Addheaders("ilm status field", "value")
table.Entries = [][]string{
{"index", ilm.Index},
{"ilm policy", *ilm.Policy},
{"action", *ilm.Action},
{"step", *ilm.Step},
{"age", fmt.Sprintf("%s", ilm.Age)},
{"managed", fmt.Sprintf("%t", ilm.Managed)},
{"phase", *ilm.Phase},
{"phase execution", ilmPhaseString(ilm.PhaseExecution.PhaseDefinition, false)},
}
return table.Print()
}
func IlmCreate(conf *cfg.Config, policyname string) error {
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 := &types.RolloverAction{}
haveroll := false
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 != 0 {
rollover.MaxPrimaryShardSize = &cfg.HotRolloverMaxPrimaryShardSize
}
if haveroll {
actions.IlmActionsCaster().Rollover = rollover
}
hot.Actions = actions.IlmActionsCaster()
phases.PhasesCaster().Hot = &hot
}
if cfg.HaveWarm() {
warm := types.Phase{}
var actions types.IlmActionsVariant = esdsl.NewIlmActions()
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
}
if cfg.HaveFrozen() {
froze := phases.PhasesCaster().Frozen
if cfg.FrozenMinAge != "" {
froze.MinAge = cfg.FrozenMinAge
}
}
if cfg.HaveDelete() {
del := phases.PhasesCaster().Delete
delete := types.DeleteAction{}
var actions types.IlmActionsVariant = esdsl.NewIlmActions()
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
}
put := &putlifecycle.Request{}
policy := &types.IlmPolicy{}
policy.IlmPolicyCaster().Phases = *phases.PhasesCaster()
put.Policy = policy
ilm.Request(put)
_, err := ilm.Do(context.Background())
if err != nil {
return fmt.Errorf("failed create ilm policy: %s", esErrorString(err))
}
return nil
}