/* 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" "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(). Header("content-type", "application/json"). Header("accept", "application/json"). Do(context.Background()) if err != nil { return "", fmt.Errorf("failed to retrieve follower info: %s", esErrorString(err)) } remote := "" for name := range res { remote = name break } if remote == "" { return "", fmt.Errorf("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 { create := conf.DefaultCluster.ES.Ccr.ResumeFollow(index). Header("content-type", "application/json"). Header("accept", "application/json") _, err := create.Do(context.Background()) if err != nil { return fmt.Errorf("failed to resume ccr following: %s", esErrorString(err)) } return nil } func CcrFollowerPause(conf *cfg.Config, index string) error { create := conf.DefaultCluster.ES.Ccr.PauseFollow(index). Header("content-type", "application/json"). Header("accept", "application/json") _, err := create.Do(context.Background()) if err != nil { return fmt.Errorf("failed to pause ccr following: %s", esErrorString(err)) } return nil } func CcrFollowerUnfollow(conf *cfg.Config, index string) error { create := conf.DefaultCluster.ES.Ccr.ForgetFollower(index). Header("content-type", "application/json"). Header("accept", "application/json") _, err := create.Do(context.Background()) if err != nil { return fmt.Errorf("failed to unfollow index: %s", 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). Header("content-type", "application/json"). Header("accept", "application/json") if conf.Wait { create.WaitForActiveShards("all") } _, err = create.Do(context.Background()) if err != nil { return fmt.Errorf("failed to create follower index: %s", esErrorString(err)) } return nil } func CcrFollowerShow(conf *cfg.Config, index string) error { res, err := conf.DefaultCluster.ES.Ccr.FollowStats(index). Header("content-type", "application/json"). Header("accept", "application/json"). Do(context.Background()) if err != nil { return fmt.Errorf("failed to retrieve follower index info: %s", 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 = [][]string{ {"name", index}, {"remote_cluster", follower.RemoteCluster}, {"leader_checkpoint", fmt.Sprintf("%d", follower.LeaderGlobalCheckpoint)}, {"follower_checkpoint", fmt.Sprintf("%d", follower.FollowerGlobalCheckpoint)}, {"bytes_read", fmt.Sprintf("%d", follower.BytesRead)}, {"failed_read_requests", fmt.Sprintf("%d", follower.FailedReadRequests)}, {"failed_write_requests", fmt.Sprintf("%d", follower.FailedWriteRequests)}, {"successful_read_requests", fmt.Sprintf("%d", follower.SuccessfulReadRequests)}, {"successful_write_requests", fmt.Sprintf("%d", follower.SuccessfulWriteRequests)}, } if err := table.Print(); err != nil { return err } return nil }