esctl/pkg/es/ccr_follower.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"
	"log/slog"

	"codeberg.org/scip/esctl/pkg/cfg"
)

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", err)
	}

	remote := ""
	for name := range res {
		remote = name
		break
	}

	if remote == "" {
		return "", fmt.Errorf("cluster doesn't have a follower: %s", err)
	}

	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", 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", 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", 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", 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", 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 := NewTable(2, 9)
	table.Addheaders("field", "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.PrintMarkdown(); err != nil {
		return err
	}

	return nil
}