Address comments, parse Archetype from TemporalNamespaceDivision

Author: awln-temporal

api/adminservice/v1/request_response.go-helpers.pb.go

@@ -991,6 +991,11 @@ so forwarding by endpoint ID will not work out of the box.`,
 		true,
 		`FrontendEnableBatcher enables batcher-related RPCs in the frontend`,
 	)
+	FrontendMaxConcurrentAdminBatchOperationPerNamespace = NewNamespaceIntSetting(
+		"frontend.MaxConcurrentAdminBatchOperationPerNamespace",
+		1,
+		`FrontendMaxConcurrentAdminBatchOperationPerNamespace is the max concurrent admin batch operation job count per namespace`,
+	)
 
 	FrontendEnableUpdateWorkflowExecution = NewNamespaceBoolSetting(
 		"frontend.enableUpdateWorkflowExecution",

api/adminservice/v1/request_response.pb.go

@@ -61,9 +61,6 @@ type (
 	MatchingRawClient matchingservice.MatchingServiceClient
 	MatchingClient    matchingservice.MatchingServiceClient
 
-	AdminRawClient adminservice.AdminServiceClient
-	AdminClient    adminservice.AdminServiceClient
-
 	RuntimeMetricsReporterParams struct {
 		fx.In
 
@@ -95,7 +92,6 @@ var Module = fx.Options(
 	fx.Provide(ClientFactoryProvider),
 	fx.Provide(ClientBeanProvider),
 	fx.Provide(FrontendClientProvider),
-	fx.Provide(AdminRawClientProvider),
 	fx.Provide(AdminClientProvider),
 	fx.Provide(GrpcListenerProvider),
 	fx.Provide(RuntimeMetricsReporterProvider),
@@ -253,16 +249,16 @@ func FrontendClientProvider(clientBean client.Bean) workflowservice.WorkflowServ
 	)
 }
 
-func AdminRawClientProvider(clientBean client.Bean, clusterMetadata cluster.Metadata) (AdminRawClient, error) {
-	return clientBean.GetRemoteAdminClient(clusterMetadata.GetCurrentClusterName())
-}
-
-func AdminClientProvider(adminRawClient AdminRawClient) AdminClient {
+func AdminClientProvider(clientBean client.Bean, clusterMetadata cluster.Metadata) (adminservice.AdminServiceClient, error) {
+	adminRawClient, err := clientBean.GetRemoteAdminClient(clusterMetadata.GetCurrentClusterName())
+	if err != nil {
+		return nil, err
+	}
 	return admin.NewRetryableClient(
 		adminRawClient,
 		common.CreateFrontendClientRetryPolicy(),
 		common.IsServiceClientTransientError,
-	)
+	), nil
 }
 
 func RuntimeMetricsReporterProvider(

api/batch/v1/request_response.pb.go

@@ -606,6 +606,7 @@ message ForceUnloadTaskQueuePartitionResponse {
 }
 
 // StartAdminBatchOperationRequest starts an admin batch operation.
+// WARNING: Batch Operations are exposed to all users of the namespace. Admin Batch Operations should be exercised with caution.
 message StartAdminBatchOperationRequest {
   // Namespace that contains the batch operation.
   string namespace = 1;
@@ -626,19 +627,16 @@ message StartAdminBatchOperationRequest {
 
   // The admin batch operation to perform.
   oneof operation {
-    BatchOperationRefreshWorkflowTasks refresh_workflow_tasks_operation = 10;
+    BatchOperationRefreshTasks refresh_tasks_operation = 10;
   }
 }
 
 message StartAdminBatchOperationResponse {
 }
 
-// BatchOperationRefreshWorkflowTasks refreshes workflow tasks for batch workflows.
-// This regenerates all pending tasks for each workflow.
-message BatchOperationRefreshWorkflowTasks {
+// BatchOperationRefreshTasks refreshes tasks for batch executions.
+// This regenerates all pending tasks for each execution.
+message BatchOperationRefreshTasks {
   // The identity of the worker/client.
   string identity = 1;
-
-  // Archetype to use for the refresh operation. If not specified, defaults to workflow archetype.
-  string archetype = 2;
 }

common/dynamicconfig/constants.go

@@ -19,11 +19,12 @@ message BatchOperationInput {
 
   repeated string non_retryable_errors = 5;
 
+  // Only needed if StartBatchOperationRequest request is set.
   temporal.api.enums.v1.BatchOperationType batch_type = 6;
 
-  // The request to start the batch operation.
+  // The request to start the batch operation. Mutually exclusive with StartAdminBatchOperationRequest admin_request.
   temporal.api.workflowservice.v1.StartBatchOperationRequest request = 7;
 
-  // The request to start an admin batch operation.
+  // The request to start an admin batch operation. Mutually exclusive with StartBatchOperationRequest request.
   temporal.server.api.adminservice.v1.StartAdminBatchOperationRequest admin_request = 8;
 }

common/resource/fx.go

@@ -1598,18 +1598,18 @@ func (adh *AdminHandler) StartAdminBatchOperation(
 	countResp, err := adh.visibilityMgr.CountWorkflowExecutions(ctx, &manager.CountWorkflowExecutionsRequest{
 		NamespaceID: namespaceID,
 		Namespace:   namespace.Name(request.GetNamespace()),
-		Query:       batcher.OpenBatchOperationQuery,
+		Query:       batcher.OpenAdminBatchOperationQuery,
 	})
 	if err != nil {
 		return nil, err
 	}
 
-	openBatchOperationCount := int(countResp.Count)
-	if openBatchOperationCount >= maxConcurrentBatchOperation {
+	openAdminBatchOperationCount := int(countResp.Count)
+	if openAdminBatchOperationCount >= maxConcurrentBatchOperation {
 		return nil, &serviceerror.ResourceExhausted{
 			Cause:   enumspb.RESOURCE_EXHAUSTED_CAUSE_CONCURRENT_LIMIT,
 			Scope:   enumspb.RESOURCE_EXHAUSTED_SCOPE_NAMESPACE,
-			Message: "Max concurrent batch operations is reached",
+			Message: "Max concurrent admin batch operations is reached",
 		}
 	}
 
@@ -1621,9 +1621,9 @@ func (adh *AdminHandler) StartAdminBatchOperation(
 	var identity string
 	var batchTypeMemo string
 	switch op := request.Operation.(type) {
-	case *adminservice.StartAdminBatchOperationRequest_RefreshWorkflowTasksOperation:
+	case *adminservice.StartAdminBatchOperationRequest_RefreshTasksOperation:
 		batchTypeMemo = "refresh_workflow_tasks"
-		identity = op.RefreshWorkflowTasksOperation.GetIdentity()
+		identity = op.RefreshTasksOperation.GetIdentity()
 	default:
 		return nil, serviceerror.NewInvalidArgumentf("The operation type %T is not supported", op)
 	}
@@ -1642,7 +1642,7 @@ func (adh *AdminHandler) StartAdminBatchOperation(
 
 	var searchAttributes *commonpb.SearchAttributes
 	searchattribute.AddSearchAttribute(&searchAttributes, sadefs.BatcherUser, payload.EncodeString(identity))
-	searchattribute.AddSearchAttribute(&searchAttributes, sadefs.TemporalNamespaceDivision, payload.EncodeString(batcher.NamespaceDivision))
+	searchattribute.AddSearchAttribute(&searchAttributes, sadefs.TemporalNamespaceDivision, payload.EncodeString(batcher.AdminNamespaceDivision))
 
 	startReq := &workflowservice.StartWorkflowExecutionRequest{
 		Namespace:                request.Namespace,
@@ -1656,7 +1656,6 @@ func (adh *AdminHandler) StartAdminBatchOperation(
 		WorkflowIdReusePolicy:    enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
 		Memo:                     memo,
 		SearchAttributes:         searchAttributes,
-		Priority:                 &commonpb.Priority{}, // ie default priority
 	}
 
 	_, err = adh.historyClient.StartWorkflowExecution(
@@ -1691,7 +1690,7 @@ func validateAdminBatchOperation(params *adminservice.StartAdminBatchOperationRe
 	}
 
 	switch op := params.GetOperation().(type) {
-	case *adminservice.StartAdminBatchOperationRequest_RefreshWorkflowTasksOperation:
+	case *adminservice.StartAdminBatchOperationRequest_RefreshTasksOperation:
 		// No additional validation needed
 		return nil
 	default:

proto/internal/temporal/server/api/adminservice/v1/request_response.proto

@@ -173,6 +173,8 @@ type Config struct {
 	// Batch operation dynamic configs
 	MaxConcurrentBatchOperation     dynamicconfig.IntPropertyFnWithNamespaceFilter
 	MaxExecutionCountBatchOperation dynamicconfig.IntPropertyFnWithNamespaceFilter
+	// Admin Batch operation dynamic config
+	MaxConcurrentAdminBatchOperation dynamicconfig.IntPropertyFnWithNamespaceFilter
 
 	EnableUpdateWorkflowExecution                              dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	EnableUpdateWorkflowExecutionAsyncAccepted                 dynamicconfig.BoolPropertyFnWithNamespaceFilter