Create admin API to get task queue tasks

temporalio · Nov 23, 2021 · 243fd24 · 243fd24
1 parent 17b9d4c
commit 243fd24
Show file tree

Hide file tree

Showing 10 changed files with 1,109 additions and 251 deletions.
diff --git a/api/adminservice/v1/request_response.pb.go b/api/adminservice/v1/request_response.pb.go
diff --git a/api/adminservice/v1/service.pb.go b/api/adminservice/v1/service.pb.go
diff --git a/api/adminservicemock/v1/service.pb.mock.go b/api/adminservicemock/v1/service.pb.mock.go
diff --git a/client/admin/client.go b/client/admin/client.go
@@ -414,6 +414,20 @@ func (c *clientImpl) ResendReplicationTasks(
 	return client.ResendReplicationTasks(ctx, request, opts...)
 }
 
+func (c *clientImpl) GetTaskQueueTasks(
+	ctx context.Context,
+	request *adminservice.GetTaskQueueTasksRequest,
+	opts ...grpc.CallOption,
+) (*adminservice.GetTaskQueueTasksResponse, error) {
+	client, err := c.getRandomClient()
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := c.createContext(ctx)
+	defer cancel()
+	return client.GetTaskQueueTasks(ctx, request, opts...)
+}
+
 func (c *clientImpl) createContext(parent context.Context) (context.Context, context.CancelFunc) {
 	return context.WithTimeout(parent, c.timeout)
 }

diff --git a/client/admin/metricClient.go b/client/admin/metricClient.go
@@ -488,3 +488,20 @@ func (c *metricClient) ResendReplicationTasks(
 	}
 	return resp, err
 }
+
+func (c *metricClient) GetTaskQueueTasks(
+	ctx context.Context,
+	request *adminservice.GetTaskQueueTasksRequest,
+	opts ...grpc.CallOption,
+) (*adminservice.GetTaskQueueTasksResponse, error) {
+
+	c.metricsClient.IncCounter(metrics.AdminClientResendReplicationTasksScope, metrics.ClientRequests)
+	sw := c.metricsClient.StartTimer(metrics.AdminClientResendReplicationTasksScope, metrics.ClientLatency)
+	resp, err := c.client.GetTaskQueueTasks(ctx, request, opts...)
+	sw.Stop()
+
+	if err != nil {
+		c.metricsClient.IncCounter(metrics.AdminClientResendReplicationTasksScope, metrics.ClientFailures)
+	}
+	return resp, err
+}
diff --git a/client/admin/retryableClient.go b/client/admin/retryableClient.go
@@ -443,3 +443,19 @@ func (c *retryableClient) ResendReplicationTasks(
 	err := backoff.Retry(op, c.policy, c.isRetryable)
 	return resp, err
 }
+
+func (c *retryableClient) GetTaskQueueTasks(
+	ctx context.Context,
+	request *adminservice.GetTaskQueueTasksRequest,
+	opts ...grpc.CallOption,
+) (*adminservice.GetTaskQueueTasksResponse, error) {
+
+	var resp *adminservice.GetTaskQueueTasksResponse
+	op := func() error {
+		var err error
+		resp, err = c.client.GetTaskQueueTasks(ctx, request, opts...)
+		return err
+	}
+	err := backoff.Retry(op, c.policy, c.isRetryable)
+	return resp, err
+}