CBG-4864 get maxcas for each vbucket

Author: torcolvin

base/bucket.go

@@ -60,6 +60,7 @@ type CouchbaseBucketStore interface {
 	HttpClient(context.Context) *http.Client
 	GetSpec() BucketSpec
 	GetMaxVbno() (uint16, error)
+	GetCCVStartingCas(context.Context) (map[uint16]uint64, error)
 
 	// GetStatsVbSeqno retrieves the high sequence number for all vbuckets and returns
 	// a map of UUIDS and a map of high sequence numbers (map from vbno -> seq)
@@ -575,3 +576,5 @@ func GetNewDatabaseSleeperFunc() RetrySleeper {
 		5,  // InitialRetrySleepTimeMS
 	)
 }
+
+var _ CouchbaseBucketStore = &GocbV2Bucket{}

base/bucket_gocb_test.go

@@ -2632,3 +2632,15 @@ func TestWriteUpdateWithXattrsDocumentTombstone(t *testing.T) {
 	require.JSONEq(t, string(xattrModifiedBody), string(xattrs[xattr1Key]))
 	require.NotContains(t, xattrs, xattr2Key)
 }
+
+func TestGetCCVStartingCAS(t *testing.T) {
+	ctx := TestCtx(t)
+	bucket := GetTestBucket(t)
+	defer bucket.Close(ctx)
+
+	cbStore, ok := AsCouchbaseBucketStore(bucket)
+	require.True(t, ok)
+	cas, err := cbStore.GetCCVStartingCas(ctx)
+	require.NoError(t, err)
+	require.NotZero(t, cas)
+}

base/collection.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -352,6 +353,42 @@ func (b *GocbV2Bucket) GetMaxVbno() (uint16, error) {
 	return uint16(vbNo), nil
 }
 
+func (b *GocbV2Bucket) GetCCVStartingCas(ctx context.Context) (map[uint16]uint64, error) {
+	uri := "/pools/default/buckets/" + b.GetName()
+	output, status, err := b.MgmtRequest(ctx, http.MethodGet, uri, "application/json", nil)
+	if err != nil {
+		return nil, fmt.Errorf("error executing query for vbucket CAS, status code: %d error: %v output: %s", status, err, string(output))
+	}
+	if status != http.StatusOK {
+		return nil, fmt.Errorf("error executing query for vbucket CAS, status code: %d output: %s", status, string(output))
+	}
+	var bucketInfo struct {
+		EnableCrossClusterVersioning bool     `json:"enableCrossClusterVersioning"`
+		MaxCas                       []string `json:"vBucketsMaxCas"`
+	}
+	err = JSONUnmarshal(output, &bucketInfo)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing output from bucket %q error:%v", string(output), err)
+	}
+
+	numVBuckets, err := b.GetMaxVbno()
+	if err != nil {
+		return nil, fmt.Errorf("error getting vbucket count: %v", err)
+	}
+	if len(bucketInfo.MaxCas) != int(numVBuckets) {
+		return nil, fmt.Errorf("error getting vbucket CAS, expected %d vbucket CAS values, got %q", numVBuckets, bucketInfo.MaxCas)
+	}
+	highCas := make(map[uint16]uint64, len(bucketInfo.MaxCas))
+	for i, casStr := range bucketInfo.MaxCas {
+		cas, err := strconv.ParseUint(casStr, 10, 64)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing vbucket CAS value %q for vbucket %d: %v", casStr, i, err)
+		}
+		highCas[uint16(i)] = cas
+	}
+	return highCas, nil
+}
+
 func (b *GocbV2Bucket) getConfigSnapshot() (*gocbcore.ConfigSnapshot, error) {
 	agent, err := b.GetGoCBAgent()
 	if err != nil {