introduce latency metrics and refactored shard statistics

Author: arzonus

common/metrics/defs.go

@@ -1479,6 +1479,7 @@ const (
 	ShardDistributorStoreAssignShardScope
 	ShardDistributorStoreAssignShardsScope
 	ShardDistributorStoreDeleteExecutorsScope
+	ShardDistributorStoreGetShardStatsScope
 	ShardDistributorStoreDeleteShardStatsScope
 	ShardDistributorStoreGetHeartbeatScope
 	ShardDistributorStoreGetStateScope
@@ -2167,6 +2168,7 @@ var ScopeDefs = map[ServiceIdx]map[ScopeIdx]scopeDefinition{
 		ShardDistributorStoreAssignShardScope:                  {operation: "StoreAssignShard"},
 		ShardDistributorStoreAssignShardsScope:                 {operation: "StoreAssignShards"},
 		ShardDistributorStoreDeleteExecutorsScope:              {operation: "StoreDeleteExecutors"},
+		ShardDistributorStoreGetShardStatsScope:                {operation: "StoreGetShardStats"},
 		ShardDistributorStoreDeleteShardStatsScope:             {operation: "StoreDeleteShardStats"},
 		ShardDistributorStoreGetHeartbeatScope:                 {operation: "StoreGetHeartbeat"},
 		ShardDistributorStoreGetStateScope:                     {operation: "StoreGetState"},
@@ -2966,6 +2968,13 @@ const (
 	ShardDistributorStoreRequestsPerNamespace
 	ShardDistributorStoreLatencyHistogramPerNamespace
 
+	// ShardDistributorShardAssignmentDistributionLatency measures the time taken between assignment of a shard
+	// and the time it is fully distributed to executors
+	ShardDistributorShardAssignmentDistributionLatency
+
+	// ShardDistributorShardHandoverLatency measures the time taken to hand over a shard from one executor to another
+	ShardDistributorShardHandoverLatency
+
 	NumShardDistributorMetrics
 )
 
@@ -3749,6 +3758,9 @@ var MetricDefs = map[ServiceIdx]map[MetricIdx]metricDefinition{
 		ShardDistributorStoreFailuresPerNamespace:         {metricName: "shard_distributor_store_failures_per_namespace", metricType: Counter},
 		ShardDistributorStoreRequestsPerNamespace:         {metricName: "shard_distributor_store_requests_per_namespace", metricType: Counter},
 		ShardDistributorStoreLatencyHistogramPerNamespace: {metricName: "shard_distributor_store_latency_histogram_per_namespace", metricType: Histogram, buckets: ShardDistributorExecutorStoreLatencyBuckets},
+
+		ShardDistributorShardAssignmentDistributionLatency: {metricName: "shard_distributor_shard_assignment_distribution_latency", metricType: Histogram, buckets: Default1ms100s.buckets()},
+		ShardDistributorShardHandoverLatency:               {metricName: "shard_distributor_shard_handover_latency", metricType: Histogram, buckets: Default1ms100s.buckets()},
 	},
 }
 

common/metrics/tags.go

@@ -352,6 +352,10 @@ func NamespaceTypeTag(namespaceType string) Tag {
 	return metricWithUnknown("namespace_type", namespaceType)
 }
 
+func HandoverTypeTag(handoverType string) Tag {
+	return metricWithUnknown("handover_type", handoverType)
+}
+
 func TaskCategoryTag(category string) Tag {
 	return metricWithUnknown("task_category", category)
 }

common/ptr/ptr.go

@@ -0,0 +1,6 @@
+package ptr
+
+// ToPtr returns a pointer to the given value.
+func ToPtr[T any](v T) *T {
+	return &v
+}

common/types/sharddistributor.go

@@ -22,9 +22,11 @@
 
 package types
 
-import "fmt"
+import (
+	"fmt"
+)
 
-//go:generate enumer -type=ExecutorStatus,ShardStatus,AssignmentStatus,MigrationMode -trimprefix=ExecutorStatus,ShardStatus,AssignmentStatus,MigrationMode -json -output sharddistributor_statuses_enumer_generated.go
+//go:generate enumer -type=ExecutorStatus,ShardStatus,AssignmentStatus,MigrationMode,HandoverType -trimprefix=ExecutorStatus,ShardStatus,AssignmentStatus,MigrationMode,HandoverType -json -output sharddistributor_statuses_enumer_generated.go
 
 type GetShardOwnerRequest struct {
 	ShardKey  string
@@ -217,6 +219,30 @@ const (
 	AssignmentStatusREADY   AssignmentStatus = 1
 )
 
+// HandoverType is used to indicate the type of handover that occurred during shard reassignment.
+// Type is persisted to the DB with a string value mapping.
+// Beware - if we want to change the name - it should be backward compatible and should be done in two steps.
+type HandoverType int32
+
+const (
+	HandoverTypeINVALID HandoverType = 0
+
+	// HandoverTypeGRACEFUL
+	// Graceful handover indicates that the shard was transferred in a way that allowed
+	// the previous owner had a chance to finish processing before the shard was reassigned.
+	HandoverTypeGRACEFUL HandoverType = 1
+
+	// HandoverTypeEMERGENCY
+	// Emergency handover indicates that the shard was transferred abruptly without
+	// allowing the previous owner to finish processing.
+	HandoverTypeEMERGENCY HandoverType = 2
+)
+
+// Ptr returns a pointer to the HandoverType value.
+func (t HandoverType) Ptr() *HandoverType {
+	return &t
+}
+
 type MigrationMode int32
 
 const (