add tests, unit and func

Author: ychebotarev

api/matchingservice/v1/request_response.pb.go

@@ -60,7 +60,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.34.0
 	go.opentelemetry.io/otel/sdk/metric v1.34.0
 	go.opentelemetry.io/otel/trace v1.34.0
-	go.temporal.io/api v1.51.0
+	go.temporal.io/api v1.51.1-0.20250725001547-f146f460d5c9
 	go.temporal.io/sdk v1.34.0
 	go.uber.org/automaxprocs v1.6.0
 	go.uber.org/fx v1.23.0

go.mod

@@ -399,8 +399,8 @@ go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC
 go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE=
 go.opentelemetry.io/proto/otlp v1.5.0 h1:xJvq7gMzB31/d406fB8U5CBdyQGw4P399D1aQWU/3i4=
 go.opentelemetry.io/proto/otlp v1.5.0/go.mod h1:keN8WnHxOy8PG0rQZjJJ5A2ebUoafqWp0eVQ4yIXvJ4=
-go.temporal.io/api v1.51.0 h1:9+e14GrIa7nWoWoudqj/PSwm33yYjV+u8TAR9If7s/g=
-go.temporal.io/api v1.51.0/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM=
+go.temporal.io/api v1.51.1-0.20250725001547-f146f460d5c9 h1:TC5floyNPVF18aZHx10KnjDE1YRAq3sssDMAY7DEq9s=
+go.temporal.io/api v1.51.1-0.20250725001547-f146f460d5c9/go.mod h1:iaxoP/9OXMJcQkETTECfwYq4cw/bj4nwov8b3ZLVnXM=
 go.temporal.io/sdk v1.34.0 h1:VLg/h6ny7GvLFVoQPqz2NcC93V9yXboQwblkRvZ1cZE=
 go.temporal.io/sdk v1.34.0/go.mod h1:iE4U5vFrH3asOhqpBBphpj9zNtw8btp8+MSaf5A0D3w=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=

go.sum

@@ -616,6 +616,5 @@ message DescribeWorkerRequest {
 }
 
 message DescribeWorkerResponse {
-    temporal.api.worker.v1.WorkerInfo workers_info = 1;
+    temporal.api.worker.v1.WorkerInfo worker_info = 1;
 }
-

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

@@ -6218,10 +6218,27 @@ func (wh *WorkflowHandler) UpdateWorkerConfig(_ context.Context, request *workfl
 	return nil, serviceerror.NewUnimplemented("UpdateWorkerConfig command is not enabled.")
 }
 
-func (wh *WorkflowHandler) DescribeWorker(_ context.Context, request *workflowservice.DescribeWorkerRequest,
+func (wh *WorkflowHandler) DescribeWorker(ctx context.Context, request *workflowservice.DescribeWorkerRequest,
 ) (*workflowservice.DescribeWorkerResponse, error) {
 	if !wh.config.ListWorkersEnabled(request.GetNamespace()) {
 		return nil, serviceerror.NewUnimplemented("DescribeWorker command is not enabled.")
 	}
-	return nil, serviceerror.NewUnimplemented("DescribeWorker command is not enabled.")
+	namespaceName := namespace.Name(request.GetNamespace())
+	namespaceID, err := wh.namespaceRegistry.GetNamespaceID(namespaceName)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := wh.matchingClient.DescribeWorker(ctx, &matchingservice.DescribeWorkerRequest{
+		NamespaceId: namespaceID.String(),
+		Request:     request,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return &workflowservice.DescribeWorkerResponse{
+		WorkerInfo: resp.GetWorkerInfo(),
+	}, nil
 }

service/frontend/workflow_handler.go

@@ -601,3 +601,19 @@ func (h *Handler) UpdateTaskQueueConfig(
 ) (*matchingservice.UpdateTaskQueueConfigResponse, error) {
 	return h.engine.UpdateTaskQueueConfig(ctx, request)
 }
+
+func (h *Handler) DescribeWorker(
+	_ context.Context, request *matchingservice.DescribeWorkerRequest,
+) (*matchingservice.DescribeWorkerResponse, error) {
+	nsID := namespace.ID(request.GetNamespaceId())
+	hb, err := h.workersRegistry.DescribeWorker(
+		nsID, request.Request.GetWorkerInstanceKey())
+	if err != nil {
+		return nil, err
+	}
+	return &matchingservice.DescribeWorkerResponse{
+		WorkerInfo: &workerpb.WorkerInfo{
+			WorkerHeartbeat: hb,
+		},
+	}, nil
+}

service/matching/handler.go

@@ -9,5 +9,6 @@ type (
 	Registry interface {
 		RecordWorkerHeartbeats(nsID namespace.ID, workerHeartbeat []*workerpb.WorkerHeartbeat)
 		ListWorkers(nsID namespace.ID, queue string, nextPageToken []byte) ([]*workerpb.WorkerHeartbeat, error)
+		DescribeWorker(nsID namespace.ID, workerInstanceKey string) (*workerpb.WorkerHeartbeat, error)
 	}
 )

service/matching/workers/registry.go

@@ -7,6 +7,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"go.temporal.io/api/serviceerror"
 	workerpb "go.temporal.io/api/worker/v1"
 	"go.temporal.io/server/common/namespace"
 	"go.uber.org/fx"
@@ -116,6 +117,23 @@ func (b *bucket) filterWorkers(
 	return out
 }
 
+func (b *bucket) getWorkerHeartbeat(nsID namespace.ID, workerInstanceKey string) (*workerpb.WorkerHeartbeat, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	mp, ok := b.namespaces[nsID]
+	if !ok {
+		return nil, serviceerror.NewNotFoundf("namespace not found: %s", nsID.String())
+	}
+
+	e, exists := mp[workerInstanceKey]
+	if !exists {
+		return nil, serviceerror.NewNotFoundf("worker not found: %s", workerInstanceKey)
+	}
+
+	return e.hb, nil
+}
+
 // evictByTTL removes entries older than expireBefore from this bucket.
 // Returns the number of entries removed.
 func (b *bucket) evictByTTL(expireBefore time.Time) int {
@@ -224,7 +242,6 @@ func (m *registryImpl) filterWorkers(
 		return nil
 	}
 	return b.filterWorkers(nsID, predicate)
-
 }
 
 // evictLoop periodically triggers TTL and capacity-based eviction.
@@ -289,17 +306,26 @@ func (m *registryImpl) RecordWorkerHeartbeats(nsID namespace.ID, workerHeartbeat
 }
 
 func (m *registryImpl) ListWorkers(nsID namespace.ID, query string, _ []byte) ([]*workerpb.WorkerHeartbeat, error) {
-	predicate := func(_ *workerpb.WorkerHeartbeat) bool { return true }
-	if query != "" {
-		queryEngine, err := newWorkerQueryEngine(nsID.String(), query)
-		if err != nil {
-			return nil, err
-		}
+	if query == "" {
+		return m.filterWorkers(nsID, func(_ *workerpb.WorkerHeartbeat) bool { return true }), nil
+	}
 
-		predicate = func(heartbeat *workerpb.WorkerHeartbeat) bool {
-			result, err := queryEngine.EvaluateWorker(heartbeat)
-			return err == nil && result
-		}
+	queryEngine, err := newWorkerQueryEngine(nsID.String(), query)
+	if err != nil {
+		return nil, err
+	}
+
+	predicate := func(heartbeat *workerpb.WorkerHeartbeat) bool {
+		result, err := queryEngine.EvaluateWorker(heartbeat)
+		return err == nil && result
 	}
 	return m.filterWorkers(nsID, predicate), nil
 }
+
+func (m *registryImpl) DescribeWorker(nsID namespace.ID, workerInstanceKey string) (*workerpb.WorkerHeartbeat, error) {
+	b := m.getBucket(nsID)
+	if b == nil {
+		return nil, serviceerror.NewNotFoundf("namespace not found: %s", nsID.String())
+	}
+	return b.getWorkerHeartbeat(nsID, workerInstanceKey)
+}

service/matching/workers/registry_impl.go

@@ -194,3 +194,95 @@ func TestRegistryImpl_ListWorkers(t *testing.T) {
 		})
 	}
 }
+
+func TestRegistryImpl_DescribeWorker(t *testing.T) {
+	tests := []struct {
+		name              string
+		setup             func(*registryImpl)
+		nsID              namespace.ID
+		workerInstanceKey string
+		expectError       bool
+	}{
+		{
+			name:              "list workers from non-existent namespace",
+			setup:             func(r *registryImpl) {},
+			nsID:              "non-existent",
+			workerInstanceKey: "worker",
+			expectError:       true,
+		},
+		{
+			name: "list workers from empty namespace",
+			setup: func(r *registryImpl) {
+			},
+			nsID:              "empty-ns",
+			workerInstanceKey: "worker",
+			expectError:       true,
+		},
+		{
+			name: "list empty worker",
+			setup: func(r *registryImpl) {
+				r.upsertHeartbeats("namespace1", []*workerpb.WorkerHeartbeat{{
+					WorkerInstanceKey: "worker1",
+				}})
+			},
+			nsID:              "namespace1",
+			workerInstanceKey: "",
+			expectError:       true,
+		},
+		{
+			name: "list single worker, doesn't exist",
+			setup: func(r *registryImpl) {
+				r.upsertHeartbeats("namespace1", []*workerpb.WorkerHeartbeat{{
+					WorkerInstanceKey: "worker1",
+				}})
+			},
+			nsID:              "namespace1",
+			workerInstanceKey: "worker2",
+			expectError:       true,
+		},
+		{
+			name: "list single worker",
+			setup: func(r *registryImpl) {
+				r.upsertHeartbeats("namespace1", []*workerpb.WorkerHeartbeat{{
+					WorkerInstanceKey: "worker1",
+				}})
+			},
+			nsID:              "namespace1",
+			workerInstanceKey: "worker1",
+		},
+		{
+			name: "list workers from specific namespace only",
+			setup: func(r *registryImpl) {
+				// Setup namespace1
+				r.upsertHeartbeats("namespace1", []*workerpb.WorkerHeartbeat{{
+					WorkerInstanceKey: "worker1",
+				}})
+				// Setup namespace2
+				r.upsertHeartbeats("namespace2", []*workerpb.WorkerHeartbeat{{
+					WorkerInstanceKey: "worker2",
+				}})
+			},
+			nsID:              "namespace2",
+			workerInstanceKey: "worker2",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := newRegistryImpl(
+				defaultBuckets, defaultEntryTTL, defaultMinEvictAge, defaultMaxEntries, defaultEvictionInterval,
+			)
+			tt.setup(r)
+
+			result, err := r.DescribeWorker(tt.nsID, tt.workerInstanceKey)
+			if tt.expectError {
+				assert.Error(t, err, "expected an error for non-existent namespace")
+				assert.Nil(t, result, "result should be nil when an error occurs")
+				return
+			}
+			assert.NoError(t, err, "unexpected error when listing workers")
+			assert.NotNil(t, result, "result should not be nil when worker exists")
+			assert.Equal(t, tt.workerInstanceKey, result.WorkerInstanceKey)
+		})
+	}
+}