feat(shard-distributor): add streaming WatchNamespaceState endpoint (#7426)

**What changed?**
Added a new streaming gRPC endpoint `WatchNamespaceState` to the shard
distributor service that allows spectators to observe namespace state
changes in real-time.

- Added `WatchNamespaceState` RPC to shard distributor proto with
streaming response
- Implemented types and mappers for `WatchNamespaceStateRequest`,
`WatchNamespaceStateResponse`, and `ExecutorShardAssignment`
- Updated grpc.tmpl template to support server-side streaming endpoints
- Updated metered.tmpl to dynamically detect request parameter for
namespace tagging
- Added metrics scope for `WatchNamespaceState` operation
- Handler implementation returns "not implemented" (stub for future
work)

**Why?**
To enable real-time monitoring of shard distribution state across
executors without polling, providing better observability for debugging
and operational awareness.

**How did you test it?**
Unit tests

**Potential risks**
Low risk - the endpoint is not yet implemented (returns error), only the
API surface and supporting infrastructure are added.

**Release notes**
- Added experimental streaming endpoint `WatchNamespaceState` to shard
distributor (not yet implemented)

**Documentation Changes**
None required at this stage.

---------

Signed-off-by: Jakob Haahr Taankvist <[email protected]>

Author: jakobht

.gen/proto/sharddistributor/v1/service.pb.go

@@ -1468,6 +1468,7 @@ const (
 const (
 	// ShardDistributorGetShardOwnerScope tracks GetShardOwner API calls received by service
 	ShardDistributorGetShardOwnerScope = iota + NumWorkerScopes
+	ShardDistributorWatchNamespaceStateScope
 	ShardDistributorHeartbeatScope
 	ShardDistributorAssignLoopScope
 
@@ -2153,6 +2154,7 @@ var ScopeDefs = map[ServiceIdx]map[ScopeIdx]scopeDefinition{
 	},
 	ShardDistributor: {
 		ShardDistributorGetShardOwnerScope:         {operation: "GetShardOwner"},
+		ShardDistributorWatchNamespaceStateScope:   {operation: "WatchNamespaceState"},
 		ShardDistributorHeartbeatScope:             {operation: "ExecutorHeartbeat"},
 		ShardDistributorAssignLoopScope:            {operation: "ShardAssignLoop"},
 		ShardDistributorExecutorScope:              {operation: "Executor"},

.gen/proto/sharddistributor/v1/service.pb.yarpc.go

@@ -279,3 +279,82 @@ func toMigrationMode(modeSD types.MigrationMode) sharddistributorv1.MigrationMod
 	}
 	return mode
 }
+
+// FromShardDistributorWatchNamespaceStateRequest converts a types.WatchNamespaceStateRequest to a sharddistributor.WatchNamespaceStateRequest
+func FromShardDistributorWatchNamespaceStateRequest(t *types.WatchNamespaceStateRequest) *sharddistributorv1.WatchNamespaceStateRequest {
+	if t == nil {
+		return nil
+	}
+	return &sharddistributorv1.WatchNamespaceStateRequest{
+		Namespace: t.GetNamespace(),
+	}
+}
+
+// ToShardDistributorWatchNamespaceStateRequest converts a sharddistributor.WatchNamespaceStateRequest to a types.WatchNamespaceStateRequest
+func ToShardDistributorWatchNamespaceStateRequest(t *sharddistributorv1.WatchNamespaceStateRequest) *types.WatchNamespaceStateRequest {
+	if t == nil {
+		return nil
+	}
+	return &types.WatchNamespaceStateRequest{
+		Namespace: t.GetNamespace(),
+	}
+}
+
+// FromShardDistributorWatchNamespaceStateResponse converts a types.WatchNamespaceStateResponse to a sharddistributor.WatchNamespaceStateResponse
+func FromShardDistributorWatchNamespaceStateResponse(t *types.WatchNamespaceStateResponse) *sharddistributorv1.WatchNamespaceStateResponse {
+	if t == nil {
+		return nil
+	}
+
+	var executors []*sharddistributorv1.ExecutorInfo
+
+	for _, executor := range t.GetExecutors() {
+		// Convert the Shards
+		shards := make([]*sharddistributorv1.Shard, 0, len(executor.GetAssignedShards()))
+		for _, shard := range executor.GetAssignedShards() {
+			shards = append(shards, &sharddistributorv1.Shard{
+				ShardKey: shard.GetShardKey(),
+			})
+		}
+		executors = append(executors, &sharddistributorv1.ExecutorInfo{
+			ExecutorId: executor.GetExecutorID(),
+			Metadata:   executor.GetMetadata(),
+			Shards:     shards,
+		})
+	}
+
+	return &sharddistributorv1.WatchNamespaceStateResponse{
+		Executors: executors,
+	}
+}
+
+// ToShardDistributorWatchNamespaceStateResponse converts a sharddistributor.WatchNamespaceStateResponse to a types.WatchNamespaceStateResponse
+func ToShardDistributorWatchNamespaceStateResponse(t *sharddistributorv1.WatchNamespaceStateResponse) *types.WatchNamespaceStateResponse {
+	if t == nil {
+		return nil
+	}
+
+	var executors []*types.ExecutorShardAssignment
+	if t.GetExecutors() != nil {
+		executors = make([]*types.ExecutorShardAssignment, 0, len(t.GetExecutors()))
+		for _, executor := range t.GetExecutors() {
+			// Convert the Shards
+			shards := make([]*types.Shard, 0, len(executor.GetShards()))
+			for _, shard := range executor.GetShards() {
+				shards = append(shards, &types.Shard{
+					ShardKey: shard.GetShardKey(),
+				})
+			}
+
+			executors = append(executors, &types.ExecutorShardAssignment{
+				ExecutorID:     executor.GetExecutorId(),
+				Metadata:       executor.GetMetadata(),
+				AssignedShards: shards,
+			})
+		}
+	}
+
+	return &types.WatchNamespaceStateResponse{
+		Executors: executors,
+	}
+}

common/metrics/defs.go

@@ -54,3 +54,15 @@ func TestToShardDistributorExecutorHeartbeatResponse(t *testing.T) {
 		assert.Equal(t, item, ToShardDistributorExecutorHeartbeatResponse(FromShardDistributorExecutorHeartbeatResponse(item)))
 	}
 }
+
+func TestFromShardDistributorWatchNamespaceStateRequest(t *testing.T) {
+	for _, item := range []*types.WatchNamespaceStateRequest{nil, {}, &testdata.ShardDistributorWatchNamespaceStateRequest} {
+		assert.Equal(t, item, ToShardDistributorWatchNamespaceStateRequest(FromShardDistributorWatchNamespaceStateRequest(item)))
+	}
+}
+
+func TestFromShardDistributorWatchNamespaceStateResponse(t *testing.T) {
+	for _, item := range []*types.WatchNamespaceStateResponse{nil, {}, &testdata.ShardDistributorWatchNamespaceStateResponse} {
+		assert.Equal(t, item, ToShardDistributorWatchNamespaceStateResponse(FromShardDistributorWatchNamespaceStateResponse(item)))
+	}
+}

common/types/mapper/proto/sharddistributor.go

@@ -226,3 +226,63 @@ const (
 	MigrationModeDISTRIBUTEDPASSTHROUGH MigrationMode = 3
 	MigrationModeONBOARDED              MigrationMode = 4
 )
+
+type WatchNamespaceStateRequest struct {
+	Namespace string
+}
+
+func (v *WatchNamespaceStateRequest) GetNamespace() (o string) {
+	if v != nil {
+		return v.Namespace
+	}
+	return
+}
+
+type WatchNamespaceStateResponse struct {
+	Executors []*ExecutorShardAssignment
+}
+
+func (v *WatchNamespaceStateResponse) GetExecutors() (o []*ExecutorShardAssignment) {
+	if v != nil {
+		return v.Executors
+	}
+	return
+}
+
+type ExecutorShardAssignment struct {
+	ExecutorID     string
+	AssignedShards []*Shard
+	Metadata       map[string]string
+}
+
+func (v *ExecutorShardAssignment) GetExecutorID() (o string) {
+	if v != nil {
+		return v.ExecutorID
+	}
+	return
+}
+
+func (v *ExecutorShardAssignment) GetAssignedShards() (o []*Shard) {
+	if v != nil {
+		return v.AssignedShards
+	}
+	return
+}
+
+func (v *ExecutorShardAssignment) GetMetadata() (o map[string]string) {
+	if v != nil {
+		return v.Metadata
+	}
+	return
+}
+
+type Shard struct {
+	ShardKey string
+}
+
+func (v *Shard) GetShardKey() (o string) {
+	if v != nil {
+		return v.ShardKey
+	}
+	return
+}