resource_group: implement async loading for resource groups

errors.toml

@@ -921,6 +921,11 @@ error = '''
 keyspace not found with name: %s
 '''
 
+["PD:resourcemanager:ErrResourceGroupsLoading"]
+error = '''
+resource groups are still being loaded, please try again later
+'''
+
 ["PD:scatter:ErrEmptyRegion"]
 error = '''
 empty region

pkg/errs/errno.go

@@ -67,6 +67,16 @@ var (
 	ErrGRPCRateLimitExceeded            = func(err error) error {
 		return status.Error(codes.ResourceExhausted, err.Error())
 	}
+	// ErrResourceGroupsLoadingGRPC converts the retryable
+	// ErrResourceGroupsLoading into codes.Unavailable, so generic
+	// client-side retry logic can act on it instead of seeing an opaque
+	// codes.Unknown. Other errors pass through unchanged.
+	ErrResourceGroupsLoadingGRPC = func(err error) error {
+		if ErrResourceGroupsLoading.Equal(err) {
+			return status.Error(codes.Unavailable, err.Error())
+		}
+		return err
+	}
 
 	// FailedPrecondition indicates operation was rejected because the
 	// system is not in a state required for the operation's execution.
@@ -538,6 +548,7 @@ var (
 	ErrResourceGroupNotExists  = errors.Normalize("the %s resource group does not exist", errors.RFCCodeText("PD:resourcemanager:ErrGroupNotExists"))
 	ErrDeleteReservedGroup     = errors.Normalize("cannot delete reserved group", errors.RFCCodeText("PD:resourcemanager:ErrDeleteReservedGroup"))
 	ErrInvalidGroup            = errors.Normalize("invalid group settings, please check %s", errors.RFCCodeText("PD:resourcemanager:ErrInvalidGroup"))
+	ErrResourceGroupsLoading   = errors.Normalize("resource groups are still being loaded, please try again later", errors.RFCCodeText("PD:resourcemanager:ErrResourceGroupsLoading"))
 )
 
 // Microservice errors

pkg/mcs/resourcemanager/metadataapi/config_service.go

@@ -513,6 +513,13 @@ func (*ConfigService) respondStoreReadError(c *gin.Context, err error) {
 		c.String(http.StatusNotFound, err.Error())
 		return
 	}
+	// Resource groups are still being loaded asynchronously: the request can
+	// succeed once loading completes, so report it as retryable rather than as
+	// an internal error.
+	if errs.ErrResourceGroupsLoading.Equal(err) {
+		c.String(http.StatusServiceUnavailable, err.Error())
+		return
+	}
 	c.String(http.StatusInternalServerError, err.Error())
 }
 
@@ -525,5 +532,9 @@ func (*ConfigService) respondStoreWriteError(c *gin.Context, err error) {
 		c.String(http.StatusNotFound, err.Error())
 		return
 	}
+	if errs.ErrResourceGroupsLoading.Equal(err) {
+		c.String(http.StatusServiceUnavailable, err.Error())
+		return
+	}
 	c.String(http.StatusInternalServerError, err.Error())
 }

pkg/mcs/resourcemanager/metadataapi/config_service_test.go

@@ -226,6 +226,26 @@ func TestConfigServiceGroupCRUDAndErrorCodes(t *testing.T) {
 	}
 }
 
+// TestConfigServiceLoadingReturns503 asserts the "resource groups are still
+// being loaded" error is reported as retryable. It is a transient startup state,
+// not an internal error, so callers and load balancers must be able to tell the
+// difference and retry.
+func TestConfigServiceLoadingReturns503(t *testing.T) {
+	t.Parallel()
+
+	re := require.New(t)
+	store := newTestStore()
+	handler := newTestHTTPHandler(store)
+
+	store.listErr = pderrors.ErrResourceGroupsLoading
+	resp := doJSONRequest(re, handler, http.MethodGet, "/resource-manager/api/v1/config/groups", nil)
+	re.Equal(http.StatusServiceUnavailable, resp.Code)
+
+	store.listErr = errors.New("boom")
+	resp = doJSONRequest(re, handler, http.MethodGet, "/resource-manager/api/v1/config/groups", nil)
+	re.Equal(http.StatusInternalServerError, resp.Code)
+}
+
 func TestConfigServiceControllerAllOrNothing(t *testing.T) {
 	t.Parallel()
 
@@ -351,6 +371,7 @@ type testStore struct {
 	serviceLimits                map[uint32]float64
 	addErr                       error
 	setServiceLimitErr           error
+	listErr                      error
 	updatedControllerConfigItems []string
 }
 
@@ -406,7 +427,10 @@ func (s *testStore) GetResourceGroup(keyspaceID uint32, name string, withStats b
 	return group.Clone(withStats), nil
 }
 
-func (*testStore) GetResourceGroupList(_ uint32, _ bool) ([]*rmserver.ResourceGroup, error) {
+func (s *testStore) GetResourceGroupList(_ uint32, _ bool) ([]*rmserver.ResourceGroup, error) {
+	if s.listErr != nil {
+		return nil, s.listErr
+	}
 	return []*rmserver.ResourceGroup{}, nil
 }
 

pkg/mcs/resourcemanager/server/grpc_service.go

@@ -110,7 +110,7 @@ func (s *Service) GetResourceGroup(_ context.Context, req *rmpb.GetResourceGroup
 	keyspaceID := ExtractKeyspaceID(req.GetKeyspaceId())
 	rg, err := s.manager.GetResourceGroup(keyspaceID, req.ResourceGroupName, req.WithRuStats)
 	if err != nil {
-		return nil, err
+		return nil, errs.ErrResourceGroupsLoadingGRPC(err)
 	}
 	if rg == nil {
 		return nil, errs.ErrResourceGroupNotExists.FastGenByArgs(req.ResourceGroupName)
@@ -129,7 +129,7 @@ func (s *Service) ListResourceGroups(_ context.Context, req *rmpb.ListResourceGr
 	keyspaceID := ExtractKeyspaceID(req.GetKeyspaceId())
 	groups, err := s.manager.GetResourceGroupList(keyspaceID, req.WithRuStats)
 	if err != nil {
-		return nil, err
+		return nil, errs.ErrResourceGroupsLoadingGRPC(err)
 	}
 	resps := &rmpb.ListResourceGroupsResponse{
 		Groups: make([]*rmpb.ResourceGroup, 0, len(groups)),
@@ -151,7 +151,7 @@ func (s *Service) AddResourceGroup(_ context.Context, req *rmpb.PutResourceGroup
 	}
 	err := s.manager.AddResourceGroup(req.GetGroup())
 	if err != nil {
-		return nil, err
+		return nil, errs.ErrResourceGroupsLoadingGRPC(err)
 	}
 	return &rmpb.PutResourceGroupResponse{Body: "Success!"}, nil
 }
@@ -166,7 +166,7 @@ func (s *Service) DeleteResourceGroup(_ context.Context, req *rmpb.DeleteResourc
 	}
 	err := s.manager.DeleteResourceGroup(ExtractKeyspaceID(req.GetKeyspaceId()), req.ResourceGroupName)
 	if err != nil {
-		return nil, err
+		return nil, errs.ErrResourceGroupsLoadingGRPC(err)
 	}
 	return &rmpb.DeleteResourceGroupResponse{Body: "Success!"}, nil
 }
@@ -181,7 +181,7 @@ func (s *Service) ModifyResourceGroup(_ context.Context, req *rmpb.PutResourceGr
 	}
 	err := s.manager.ModifyResourceGroup(req.GetGroup())
 	if err != nil {
-		return nil, err
+		return nil, errs.ErrResourceGroupsLoadingGRPC(err)
 	}
 	return &rmpb.PutResourceGroupResponse{Body: "Success!"}, nil
 }
@@ -230,18 +230,25 @@ func (s *Service) AcquireTokenBuckets(stream rmpb.ResourceManager_AcquireTokenBu
 				zap.Uint32("keyspace-id", keyspaceID),
 				zap.String("resource-group", resourceGroupName),
 			)
+			// Get the resource group from manager to acquire token buckets. This also
+			// triggers lazy loading of the group if async loading hasn't completed yet,
+			// so it must happen before accessKeyspaceResourceGroupManager below.
+			// Lazy loading can fail transiently (a storage read error, or retries
+			// exhausted while async loading is still in progress). Skip just this
+			// request instead of returning: the error belongs to one resource group,
+			// while returning would tear down the whole stream and force every
+			// client multiplexed on it to reconnect.
+			rg, err := s.manager.GetMutableResourceGroup(keyspaceID, resourceGroupName)
+			if rg == nil {
+				log.Warn("resource group is unavailable", append(requestFields, zap.Error(err))...)
+				continue
+			}
 			// Get keyspace resource group manager to apply service limit later.
 			krgm, err := s.manager.accessKeyspaceResourceGroupManager(keyspaceID, resourceGroupName)
 			if krgm == nil {
 				log.Warn("keyspace resource group manager not found", append(requestFields, zap.Error(err))...)
 				continue
 			}
-			// Get the resource group from manager to acquire token buckets.
-			rg, err := s.manager.GetMutableResourceGroup(keyspaceID, resourceGroupName)
-			if rg == nil {
-				log.Warn("resource group not found", append(requestFields, zap.Error(err))...)
-				continue
-			}
 			// Send the consumption to update the metrics.
 			err = s.manager.dispatchConsumption(req)
 			if err != nil {