Files
esctl/pkg/es/cluster_reroute.go
2026-06-29 13:08:05 +02:00

126 lines
3.0 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"
"fmt"
"codeberg.org/scip/esctl/pkg/cfg"
"github.com/elastic/go-elasticsearch/v9/typedapi/esdsl"
"github.com/elastic/go-elasticsearch/v9/typedapi/types"
)
func ClusterRerouteMove(conf *cfg.Config, index string) error {
move := conf.DefaultCluster.ES().Cluster.Reroute()
commands := esdsl.NewCommand()
moveCommand := &types.CommandMoveAction{
Shard: conf.Shards,
FromNode: conf.FromNode,
ToNode: conf.ToNode,
Index: index,
}
commands.CommandCaster().Move = moveCommand
move.Commands(commands)
_, err := move.Do(context.Background())
if err != nil {
return fmt.Errorf("failed to reroute move: %w", err)
}
return nil
}
func ClusterRerouteAllocateReplica(conf *cfg.Config, index string) error {
move := conf.DefaultCluster.ES().Cluster.Reroute()
commands := esdsl.NewCommand()
allocCommand := &types.CommandAllocateReplicaAction{
Shard: conf.Shards,
Node: conf.ToNode,
Index: index,
}
commands.CommandCaster().AllocateReplica = allocCommand
move.Commands(commands)
_, err := move.Do(context.Background())
if err != nil {
return fmt.Errorf("failed to allocate a replica shard: %w", err)
}
return nil
}
func ClusterRerouteCancel(conf *cfg.Config, index string) error {
move := conf.DefaultCluster.ES().Cluster.Reroute()
commands := esdsl.NewCommand()
cancelCommand := &types.CommandCancelAction{
Shard: conf.Shards,
Node: conf.ToNode,
Index: index,
AllowPrimary: &conf.AllowPrimary,
}
commands.CommandCaster().Cancel = cancelCommand
move.Commands(commands)
_, err := move.Do(context.Background())
if err != nil {
return fmt.Errorf("failed to cancel a reroute process: %w", err)
}
return nil
}
func ClusterRerouteAllocatePrimary(conf *cfg.Config, index string, stale bool) error {
move := conf.DefaultCluster.ES().Cluster.Reroute()
commands := esdsl.NewCommand()
allocCommand := &types.CommandAllocatePrimaryAction{
Shard: conf.Shards,
Node: conf.ToNode,
Index: index,
AcceptDataLoss: conf.AcceptDataLoss,
}
if stale {
commands.CommandCaster().AllocateStalePrimary = allocCommand
} else {
commands.CommandCaster().AllocateEmptyPrimary = allocCommand
}
move.Commands(commands)
_, err := move.Do(context.Background())
if err != nil {
which := "empty"
if stale {
which = "stale"
}
return fmt.Errorf("failed to allocate %s primary shard: %w", which, err)
}
return nil
}