cp fix global service safe point (tikv#312) (tikv#370)

Signed-off-by: ystaticy <y_static_y@sina.com>

Author: ystaticy

pkg/gc/safepoint_v2.go

@@ -212,3 +212,13 @@ func (manager *SafePointV2Manager) RemoveServiceSafePoint(keyspaceID uint32, ser
 	}
 	return minServiceSafePoint, nil
 }
+
+// LoadMinServiceSafePointV2 update keyspace service safe point with the given serviceSafePoint.
+func (manager *SafePointV2Manager) LoadMinServiceSafePointV2(serviceSafePoint *endpoint.ServiceSafePointV2, now time.Time) (*endpoint.ServiceSafePointV2, error) {
+	return manager.v2Storage.LoadMinServiceSafePointV2(serviceSafePoint.KeyspaceID, now)
+}
+
+// LoadServiceSafePointV2 update keyspace service safe point with the given serviceSafePoint.
+func (manager *SafePointV2Manager) LoadServiceSafePointV2(keyspaceID uint32, serviceID string) (*endpoint.ServiceSafePointV2, error) {
+	return manager.v2Storage.LoadServiceSafePointV2(keyspaceID, serviceID)
+}

pkg/storage/endpoint/safepoint_v2.go

@@ -27,6 +27,11 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	// NullKeyspaceID is used for api v1 or legacy path where is keyspace agnostic.
+	nullKeyspaceID = uint32(0xFFFFFFFF)
+)
+
 // GCSafePointV2 represents the overall safe point for a specific keyspace.
 type GCSafePointV2 struct {
 	KeyspaceID uint32 `json:"keyspace_id"`
@@ -118,6 +123,17 @@ func (se *StorageEndpoint) LoadMinServiceSafePointV2(keyspaceID uint32, now time
 
 	hasGCWorker := false
 	min := &ServiceSafePointV2{KeyspaceID: keyspaceID, SafePoint: math.MaxUint64}
+
+	// Load global service safe point
+	serviceSafePointV1, err := se.loadServiceGCSafePointV1(keypath.NativeBRServiceID)
+	if err != nil {
+		return nil, err
+	}
+	if serviceSafePointV1 != nil {
+		min.KeyspaceID = nullKeyspaceID
+		min.SafePoint = serviceSafePointV1.SafePoint
+	}
+
 	for i, key := range keys {
 		serviceSafePoint := &ServiceSafePointV2{}
 		if err = json.Unmarshal([]byte(values[i]), serviceSafePoint); err != nil {
@@ -157,6 +173,22 @@ func (se *StorageEndpoint) LoadMinServiceSafePointV2(keyspaceID uint32, now time
 	return min, nil
 }
 
+// loadServiceGCSafePointV1 loads current GC safe point from storage.
+func (se *StorageEndpoint) loadServiceGCSafePointV1(serviceID string) (*ServiceSafePoint, error) {
+	serviceIDPath := keypath.GCSafePointServicePrefixPath() + serviceID
+	value, err := se.Load(serviceIDPath)
+	if err != nil || value == "" {
+		return nil, err
+	}
+
+	ssp := &ServiceSafePoint{}
+	if err := json.Unmarshal([]byte(value), ssp); err != nil {
+		return nil, err
+	}
+
+	return ssp, nil
+}
+
 // LoadServiceSafePointV2 returns ServiceSafePointV2 for given keyspaceID and serviceID.
 func (se *StorageEndpoint) LoadServiceSafePointV2(keyspaceID uint32, serviceID string) (*ServiceSafePointV2, error) {
 	key := keypath.ServiceSafePointV2Path(keyspaceID, serviceID)

pkg/storage/storage_gc_test.go

@@ -150,3 +150,45 @@ func TestLoadEmpty(t *testing.T) {
 	re.NoError(err)
 	re.Nil(serviceSafePoint)
 }
+
+func TestGlobalServiceSafePoint(t *testing.T) {
+	// expectMinServiceSafePoint is a very small value,
+	// we set NativeBRServiceID's service safe point to expectMinServiceSafePoint,
+	// and check min service safe point V2 is expectMinServiceSafePoint.
+	expectMinServiceSafePoint := uint64(50)
+
+	re := require.New(t)
+	storage := NewStorageWithMemoryBackend()
+	currentTime := time.Now()
+	expireAt1 := currentTime.Add(1000 * time.Second).Unix()
+	expireAt2 := currentTime.Add(2000 * time.Second).Unix()
+	expireAt3 := currentTime.Add(3000 * time.Second).Unix()
+
+	testKeyspaceID := uint32(1)
+	serviceSafePoints := []*endpoint.ServiceSafePointV2{
+		{KeyspaceID: testKeyspaceID, ServiceID: "0", ExpiredAt: expireAt1, SafePoint: 300},
+		{KeyspaceID: testKeyspaceID, ServiceID: "1", ExpiredAt: expireAt2, SafePoint: 400},
+		{KeyspaceID: testKeyspaceID, ServiceID: "2", ExpiredAt: expireAt3, SafePoint: 500},
+	}
+
+	globalServiceIDs := []string{keypath.NativeBRServiceID, "test01"}
+	minGlobalSafePoints := []uint64{expectMinServiceSafePoint, expectMinServiceSafePoint + 1}
+
+	for i, globalServiceID := range globalServiceIDs {
+		ssp := &endpoint.ServiceSafePoint{
+			ServiceID: globalServiceID,
+			ExpiredAt: expireAt3,
+			SafePoint: minGlobalSafePoints[i],
+		}
+		storage.SaveServiceGCSafePoint(ssp)
+	}
+
+	for _, serviceSafePoint := range serviceSafePoints {
+		re.NoError(storage.SaveServiceSafePointV2(serviceSafePoint))
+	}
+	// enabling failpoint to make expired key removal immediately observable
+	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/storage/endpoint/removeExpiredKeys", "return(true)"))
+	minSafePoint, err := storage.LoadMinServiceSafePointV2(testKeyspaceID, currentTime)
+	re.NoError(err)
+	re.Equal(expectMinServiceSafePoint, minSafePoint.SafePoint)
+}

pkg/utils/keypath/key_path.go

@@ -46,19 +46,17 @@ const (
 	CustomSchedulerConfigPath = "scheduler_config"
 	// GCWorkerServiceSafePointID is the service id of GC worker.
 	GCWorkerServiceSafePointID = "gc_worker"
-	// NativeBRServiceID is CSE native br service id.
-	NativeBRServiceID         = "native_br"
-	minResolvedTS             = "min_resolved_ts"
-	externalTimeStamp         = "external_timestamp"
-	keyspaceSafePointPrefix   = "keyspaces/gc_safepoint"
-	keyspaceGCSafePointSuffix = "gc"
-	keyspacePrefix            = "keyspaces"
-	keyspaceMetaInfix         = "meta"
-	keyspaceIDInfix           = "id"
-	keyspaceAllocID           = "alloc_id"
-	gcSafePointInfix          = "gc_safe_point"
-	serviceSafePointInfix     = "service_safe_point"
-	regionPathPrefix          = "raft/r"
+	minResolvedTS              = "min_resolved_ts"
+	externalTimeStamp          = "external_timestamp"
+	keyspaceSafePointPrefix    = "keyspaces/gc_safepoint"
+	keyspaceGCSafePointSuffix  = "gc"
+	keyspacePrefix             = "keyspaces"
+	keyspaceMetaInfix          = "meta"
+	keyspaceIDInfix            = "id"
+	keyspaceAllocID            = "alloc_id"
+	gcSafePointInfix           = "gc_safe_point"
+	serviceSafePointInfix      = "service_safe_point"
+	regionPathPrefix           = "raft/r"
 	// resource group storage endpoint has prefix `resource_group`
 	// ResourceGroupSettingsPath is the path to save the resource group settings.
 	ResourceGroupSettingsPath = "settings"
@@ -85,6 +83,9 @@ const (
 	// Keyspace safe point version.
 	keyspaceGlobalInfix                 = "global"
 	keyspaceGlobalSafePointVersionInfix = "safe_point_version"
+
+	// NativeBRServiceID is CSE native br service id.
+	NativeBRServiceID = "native_br"
 )
 
 // PDRootPath returns the PD root path.

tests/integrations/client/gc_client_test.go

@@ -218,3 +218,34 @@ func (suite *gcClientTestSuite) mustGetRevision(re *require.Assertions, keyspace
 	re.NoError(err)
 	return res.Header.GetRevision()
 }
+
+// mustUpdateServiceSafePoint updates the service safe point of the given keyspace id.
+func (suite *gcClientTestSuite) mustUpdateServiceSafePoint(keyspaceID uint32, safePoint uint64, ttl int64, serviceID string) uint64 {
+	res, err := suite.client.UpdateServiceSafePointV2(suite.server.Context(), keyspaceID, serviceID, ttl, safePoint)
+	suite.NoError(err)
+	return res
+}
+
+// mustUpdateServiceSafePoint updates the service safe point of the given keyspace id.
+func (suite *gcClientTestSuite) mustLoadServiceSafePoint(keyspaceID uint32, serviceID string) uint64 {
+	res, err := suite.server.GetSafePointV2Manager().LoadServiceSafePointV2(keyspaceID, serviceID)
+	suite.NoError(err)
+	return res.SafePoint
+}
+
+func (suite *gcClientTestSuite) TestUpdateServiceSafePointV2() {
+	serviceID := "test_service_id"
+	ttl := int64(10000)
+
+	keyspaceID01 := uint32(1)
+	serviceSafePoint01 := uint64(1)
+	_ = suite.mustUpdateServiceSafePoint(keyspaceID01, serviceSafePoint01, ttl, serviceID)
+	minSafePoint01 := suite.mustLoadServiceSafePoint(keyspaceID01, serviceID)
+	suite.Equal(serviceSafePoint01, minSafePoint01)
+
+	keyspaceID02 := uint32(2)
+	serviceSafePoint02 := uint64(2)
+	_ = suite.mustUpdateServiceSafePoint(keyspaceID02, serviceSafePoint02, ttl, serviceID)
+	minSafePoint02 := suite.mustLoadServiceSafePoint(keyspaceID02, serviceID)
+	suite.Equal(serviceSafePoint02, minSafePoint02)
+}