frostfs-node/cmd/frostfs-adm/internal/modules/metabase/upgrade.go

package metabase

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-adm/internal/commonflags"
	"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config"
	engineconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/engine"
	shardconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/engine/shard"
	morphconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/morph"
	nodeconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/node"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client"
	morphcontainer "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/container"
	"github.com/spf13/cobra"
	"golang.org/x/sync/errgroup"
)

const (
	noCompactFlag = "no-compact"
)

var (
	errNoPathsFound          = errors.New("no metabase paths found")
	errNoMorphEndpointsFound = errors.New("no morph endpoints found")
)

var UpgradeCmd = &cobra.Command{
	Use:   "upgrade",
	Short: "Upgrade metabase to latest version",
	RunE:  upgrade,
}

func upgrade(cmd *cobra.Command, _ []string) error {
	configFile, err := cmd.Flags().GetString(commonflags.ConfigFlag)
	if err != nil {
		return err
	}
	configDir, err := cmd.Flags().GetString(commonflags.ConfigDirFlag)
	if err != nil {
		return err
	}
	appCfg := config.New(configFile, configDir, config.EnvPrefix)
	paths, err := getMetabasePaths(appCfg)
	if err != nil {
		return err
	}
	if len(paths) == 0 {
		return errNoPathsFound
	}
	cmd.Println("found", len(paths), "metabases:")
	for i, path := range paths {
		cmd.Println(i+1, ":", path)
	}
	mc, err := createMorphClient(cmd.Context(), appCfg)
	if err != nil {
		return err
	}
	defer mc.Close()
	civ, err := createContainerInfoProvider(mc)
	if err != nil {
		return err
	}
	noCompact, _ := cmd.Flags().GetBool(noCompactFlag)
	result := make(map[string]bool)
	var resultGuard sync.Mutex
	eg, ctx := errgroup.WithContext(cmd.Context())
	for _, path := range paths {
		eg.Go(func() error {
			var success bool
			cmd.Println("upgrading metabase", path, "...")
			if err := meta.Upgrade(ctx, path, !noCompact, civ, func(a ...any) {
				cmd.Println(append([]any{time.Now().Format(time.RFC3339), ":", path, ":"}, a...)...)
			}); err != nil {
				cmd.Println("error: failed to upgrade metabase", path, ":", err)
			} else {
				success = true
				cmd.Println("metabase", path, "upgraded successfully")
			}
			resultGuard.Lock()
			result[path] = success
			resultGuard.Unlock()
			return nil
		})
	}
	if err := eg.Wait(); err != nil {
		return err
	}
	for mb, ok := range result {
		if ok {
			cmd.Println(mb, ": success")
		} else {
			cmd.Println(mb, ": failed")
		}
	}
	return nil
}

func getMetabasePaths(appCfg *config.Config) ([]string, error) {
	var paths []string
	if err := engineconfig.IterateShards(appCfg, false, func(sc *shardconfig.Config) error {
		paths = append(paths, sc.Metabase().Path())
		return nil
	}); err != nil {
		return nil, fmt.Errorf("failed to get metabase paths: %w", err)
	}
	return paths, nil
}

func createMorphClient(ctx context.Context, appCfg *config.Config) (*client.Client, error) {
	addresses := morphconfig.RPCEndpoint(appCfg)
	if len(addresses) == 0 {
		return nil, errNoMorphEndpointsFound
	}
	key := nodeconfig.Key(appCfg)
	cli, err := client.New(ctx,
		key,
		client.WithDialTimeout(morphconfig.DialTimeout(appCfg)),
		client.WithEndpoints(addresses...),
		client.WithSwitchInterval(morphconfig.SwitchInterval(appCfg)),
	)
	if err != nil {
		return nil, fmt.Errorf("create morph client:%w", err)
	}
	return cli, nil
}

func createContainerInfoProvider(cli *client.Client) (container.InfoProvider, error) {
	sh, err := cli.NNSContractAddress(client.NNSContainerContractName)
	if err != nil {
		return nil, fmt.Errorf("resolve container contract hash: %w", err)
	}
	cc, err := morphcontainer.NewFromMorph(cli, sh, 0, morphcontainer.TryNotary())
	if err != nil {
		return nil, fmt.Errorf("create morph container client: %w", err)
	}
	return container.NewInfoProvider(func() (container.Source, error) {
		return morphcontainer.AsContainerSource(cc), nil
	}), nil
}

func initUpgradeCommand() {
	flags := UpgradeCmd.Flags()
	flags.Bool(noCompactFlag, false, "Do not compact upgraded metabase file")
}