From b958fe0f5615370b8e84320e5df1d4bb59181a98 Mon Sep 17 00:00:00 2001
From: Injun Song <ijsong@gmail.com>
Date: Mon, 31 Jul 2023 18:21:18 +0900
Subject: [PATCH] feat(admin): check cluster id in mrmanager

This PR adds codes to check cluster id in mrmanager. Because the admin server doesn't have explicit
health checking, comparing cluster id is performed in ClusterMetadataViewer.
In the next PR, checking the replication factor will be added.
---
 internal/admin/mrmanager/manager.go | 25 ++++++++++++++++++++++++-
 tests/it/testenv.go                 |  1 +
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/internal/admin/mrmanager/manager.go b/internal/admin/mrmanager/manager.go
index c124327b2..fb8e48613 100644
--- a/internal/admin/mrmanager/manager.go
+++ b/internal/admin/mrmanager/manager.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 	"golang.org/x/sync/singleflight"
 
 	"github.com/kakao/varlog/pkg/mrc"
@@ -126,6 +127,7 @@ func New(ctx context.Context, opts ...Option) (MetadataRepositoryManager, error)
 			continue
 		}
 		return &mrManager{
+			config:    cfg,
 			dirty:     true,
 			connector: connector,
 		}, nil
@@ -375,11 +377,32 @@ func (mrm *mrManager) ClusterMetadata(ctx context.Context) (*varlogpb.MetadataDe
 	mrm.mu.RUnlock()
 
 	// slow path
+	return mrm.getClusterMetadataSlowly(ctx)
+}
+
+func (mrm *mrManager) getClusterMetadataSlowly(ctx context.Context) (*varlogpb.MetadataDescriptor, error) {
 	md, err, _ := mrm.sfg.Do("cluster_metadata", func() (interface{}, error) {
-		meta, err := mrm.clusterMetadata(ctx)
+		var (
+			meta *varlogpb.MetadataDescriptor
+			info *mrpb.ClusterInfo
+		)
+		eg, ctx := errgroup.WithContext(ctx)
+		eg.Go(func() (err error) {
+			meta, err = mrm.clusterMetadata(ctx)
+			return err
+		})
+		eg.Go(func() (err error) {
+			info, err = mrm.GetClusterInfo(ctx)
+			return err
+		})
+		err := eg.Wait()
 		if err != nil {
 			return nil, fmt.Errorf("cluster metadata: %w", err)
 		}
+		if info.ClusterID != mrm.cid {
+			return nil, fmt.Errorf("unexpected cluster id: expected %v, got %v", mrm.cid, info.ClusterID)
+		}
+
 		mrm.mu.Lock()
 		mrm.meta = meta
 		mrm.dirty = false
diff --git a/tests/it/testenv.go b/tests/it/testenv.go
index 5c83005ed..393bea169 100644
--- a/tests/it/testenv.go
+++ b/tests/it/testenv.go
@@ -1185,6 +1185,7 @@ func (clus *VarlogCluster) initVMS(t *testing.T) {
 	mrMgrOpts := append(clus.mrMgrOpts,
 		mrmanager.WithAddresses(clus.mrRPCEndpoints...),
 		mrmanager.WithLogger(clus.logger.Logger),
+		mrmanager.WithClusterID(clus.clusterID),
 	)
 	mrMgr, err := mrmanager.New(context.TODO(), mrMgrOpts...)
 	require.NoError(t, err)