/* 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 . */ package es import ( "context" "errors" "fmt" "log/slog" "codeberg.org/scip/esctl/pkg/cfg" "codeberg.org/scip/esctl/pkg/printer" ) func getRemoteName(conf *cfg.Config) (string, error) { res, err := conf.DefaultCluster.ES().Cluster.RemoteInfo(). Do(context.Background()) if err != nil { return "", fmt.Errorf("failed to retrieve follower info: %w", esErrorString(err)) } remote := "" for name := range res { remote = name break } if remote == "" { return "", errors.New("cluster doesn't have a follower") } return remote, nil } func wrapError(call func(*cfg.Config, string) error, conf *cfg.Config, index string) error { err := call(conf, index) if conf.Force { fmt.Printf("caught error: %s, continuing anyway", err) } else { return err } return nil } func CcrFollowerRenew(conf *cfg.Config, index string) error { if err := wrapError(IndexClose, conf, index); err != nil { return err } fmt.Printf("closed %s", index) if err := wrapError(CcrFollowerPause, conf, index); err != nil { return err } fmt.Printf("paused %s", index) if err := wrapError(CcrFollowerUnfollow, conf, index); err != nil { return err } fmt.Printf("unfollowed %s", index) if err := wrapError(IndexDelete, conf, index); err != nil { return err } fmt.Printf("deleted %s", index) if err := CcrFollowerAdd(conf, index); err != nil { return err } fmt.Printf("added follower %s", index) return nil } func CcrFollowerResume(conf *cfg.Config, index string) error { _, err := conf.DefaultCluster.ES().Ccr.ResumeFollow(index). Do(context.Background()) if err != nil { return fmt.Errorf("failed to resume ccr following: %w", esErrorString(err)) } return nil } func CcrFollowerPause(conf *cfg.Config, index string) error { _, err := conf.DefaultCluster.ES().Ccr.PauseFollow(index). Do(context.Background()) if err != nil { return fmt.Errorf("failed to pause ccr following: %w", esErrorString(err)) } return nil } func CcrFollowerUnfollow(conf *cfg.Config, index string) error { _, err := conf.DefaultCluster.ES().Ccr.ForgetFollower(index). Do(context.Background()) if err != nil { return fmt.Errorf("failed to unfollow index: %w", esErrorString(err)) } return nil } func CcrFollowerAdd(conf *cfg.Config, index string) error { remote, err := getRemoteName(conf) if err != nil { return err } create := conf.DefaultCluster.ES().Ccr.Follow(index). LeaderIndex(index). RemoteCluster(remote) if conf.Wait { create.WaitForActiveShards("all") } _, err = create.Do(context.Background()) if err != nil { return fmt.Errorf("failed to create follower index: %w", esErrorString(err)) } return nil } func CcrFollowerShow(conf *cfg.Config, index string) error { res, err := conf.DefaultCluster.ES().Ccr.FollowStats(index). Do(context.Background()) if err != nil { return fmt.Errorf("failed to retrieve follower index info: %w", esErrorString(err)) } slog.Debug("ES result", "follower stats", res.Indices) if len(res.Indices) == 0 { return fmt.Errorf("cluster did not return any follower stats for index %s", index) } if len(res.Indices[0].Shards) == 0 { return fmt.Errorf("cluster did not return any shard stats on follower index %s", index) } follower := res.Indices[0].Shards[0] table := printer.NewTable(conf, 2, 9) table.Addheaders("ccr follower property", "value") table.Entries = [][]any{ {"name", index}, {"remote_cluster", follower.RemoteCluster}, {"leader_checkpoint", follower.LeaderGlobalCheckpoint}, {"follower_checkpoint", follower.FollowerGlobalCheckpoint}, {"bytes_read", follower.BytesRead}, {"failed_read_requests", follower.FailedReadRequests}, {"failed_write_requests", follower.FailedWriteRequests}, {"successful_read_requests", follower.SuccessfulReadRequests}, {"successful_write_requests", follower.SuccessfulWriteRequests}, } if err := table.Print(); err != nil { return err } return nil }