replication: add histograms for task-ack timers

Signed-off-by: Diana Zawadzki <dzawa@live.de>

Author: zawadzkidiana

common/metrics/defs.go

@@ -2726,11 +2726,16 @@ const (
 	ReplicationTasksApplied
 	ReplicationTasksFailed
 	ReplicationTasksLag
+	ExponentialReplicationTasksLag
 	ReplicationTasksLagRaw
+	ExponentialReplicationTasksLagRaw
 	ReplicationTasksDelay
 	ReplicationTasksFetched
+	ExponentialReplicationTasksFetched
 	ReplicationTasksReturned
+	ExponentialReplicationTasksReturned
 	ReplicationTasksReturnedDiff
+	ExponentialReplicationTasksReturnedDiff
 	ReplicationTasksAppliedLatency
 	ReplicationTasksBatchSize
 	ReplicationDynamicTaskBatchSizerDecision
@@ -3541,11 +3546,16 @@ var MetricDefs = map[ServiceIdx]map[MetricIdx]metricDefinition{
 		ReplicationTasksApplied:                                      {metricName: "replication_tasks_applied", metricType: Counter},
 		ReplicationTasksFailed:                                       {metricName: "replication_tasks_failed", metricType: Counter},
 		ReplicationTasksLag:                                          {metricName: "replication_tasks_lag", metricType: Timer},
+		ExponentialReplicationTasksLag:                               {metricName: "replication_tasks_lag_counts", metricType: Histogram, intExponentialBuckets: Mid1To16k},
 		ReplicationTasksLagRaw:                                       {metricName: "replication_tasks_lag_raw", metricType: Timer},
+		ExponentialReplicationTasksLagRaw:                            {metricName: "replication_tasks_lag_raw_counts", metricType: Histogram, intExponentialBuckets: Mid1To16k},
 		ReplicationTasksDelay:                                        {metricName: "replication_tasks_delay", metricType: Histogram, buckets: ReplicationTaskDelayBucket},
 		ReplicationTasksFetched:                                      {metricName: "replication_tasks_fetched", metricType: Timer},
+		ExponentialReplicationTasksFetched:                           {metricName: "replication_tasks_fetched_counts", metricType: Histogram, intExponentialBuckets: Mid1To16k},
 		ReplicationTasksReturned:                                     {metricName: "replication_tasks_returned", metricType: Timer},
+		ExponentialReplicationTasksReturned:                          {metricName: "replication_tasks_returned_counts", metricType: Histogram, intExponentialBuckets: Mid1To16k},
 		ReplicationTasksReturnedDiff:                                 {metricName: "replication_tasks_returned_diff", metricType: Timer},
+		ExponentialReplicationTasksReturnedDiff:                      {metricName: "replication_tasks_returned_diff_counts", metricType: Histogram, intExponentialBuckets: Mid1To16k},
 		ReplicationTasksAppliedLatency:                               {metricName: "replication_tasks_applied_latency", metricType: Timer},
 		ReplicationTasksBatchSize:                                    {metricName: "replication_tasks_batch_size", metricType: Gauge},
 		ReplicationDynamicTaskBatchSizerDecision:                     {metricName: "replication_dynamic_task_batch_sizer_decision", metricType: Counter},

service/history/replication/task_ack_manager.go

@@ -147,6 +147,7 @@ func (t *TaskAckManager) getTasks(ctx context.Context, pollingCluster string, la
 		return nil, err
 	}
 	t.scope.RecordTimer(metrics.ReplicationTasksFetched, time.Duration(len(taskInfos)))
+	t.scope.IntExponentialHistogram(metrics.ExponentialReplicationTasksFetched, len(taskInfos))
 
 	// Happy path assumption - we will push all tasks to replication tasks.
 	msgs := &types.ReplicationMessages{
@@ -167,7 +168,10 @@ func (t *TaskAckManager) getTasks(ctx context.Context, pollingCluster string, la
 		oldestUnprocessedTaskTimestamp = t.timeSource.Now().UnixNano()
 	}
 
-	t.scope.RecordTimer(metrics.ReplicationTasksLagRaw, time.Duration(t.ackLevels.UpdateIfNeededAndGetQueueMaxReadLevel(persistence.HistoryTaskCategoryReplication, pollingCluster).GetTaskID()-oldestUnprocessedTaskID))
+	maxReadLevel := t.ackLevels.UpdateIfNeededAndGetQueueMaxReadLevel(persistence.HistoryTaskCategoryReplication, pollingCluster).GetTaskID()
+	lagRaw := maxReadLevel - oldestUnprocessedTaskID
+	t.scope.RecordTimer(metrics.ReplicationTasksLagRaw, time.Duration(lagRaw))
+	t.scope.IntExponentialHistogram(metrics.ExponentialReplicationTasksLagRaw, clampInt64ToInt(lagRaw))
 	t.scope.RecordHistogramDuration(metrics.ReplicationTasksDelay, time.Duration(oldestUnprocessedTaskTimestamp-t.timeSource.Now().UnixNano()))
 
 	// hydrate the tasks
@@ -199,9 +203,16 @@ func (t *TaskAckManager) getTasks(ctx context.Context, pollingCluster string, la
 		return nil, err
 	}
 
-	t.scope.RecordTimer(metrics.ReplicationTasksLag, time.Duration(t.ackLevels.UpdateIfNeededAndGetQueueMaxReadLevel(persistence.HistoryTaskCategoryReplication, pollingCluster).GetTaskID()-msgs.LastRetrievedMessageID))
+	lag := maxReadLevel - msgs.LastRetrievedMessageID
+	t.scope.RecordTimer(metrics.ReplicationTasksLag, time.Duration(lag))
+	t.scope.IntExponentialHistogram(metrics.ExponentialReplicationTasksLag, clampInt64ToInt(lag))
+
 	t.scope.RecordTimer(metrics.ReplicationTasksReturned, time.Duration(len(msgs.ReplicationTasks)))
-	t.scope.RecordTimer(metrics.ReplicationTasksReturnedDiff, time.Duration(len(taskInfos)-len(msgs.ReplicationTasks)))
+	t.scope.IntExponentialHistogram(metrics.ExponentialReplicationTasksReturned, len(msgs.ReplicationTasks))
+
+	returnedDiff := len(taskInfos) - len(msgs.ReplicationTasks)
+	t.scope.RecordTimer(metrics.ReplicationTasksReturnedDiff, time.Duration(returnedDiff))
+	t.scope.IntExponentialHistogram(metrics.ExponentialReplicationTasksReturnedDiff, returnedDiff)
 
 	t.ackLevel(pollingCluster, lastReadTaskID)
 
@@ -221,6 +232,17 @@ func (t *TaskAckManager) getTasks(ctx context.Context, pollingCluster string, la
 	}, nil
 }
 
+func clampInt64ToInt(v int64) int {
+	if v <= 0 {
+		return 0
+	}
+	maxInt := int64(int(^uint(0) >> 1))
+	if v > maxInt {
+		return int(maxInt)
+	}
+	return int(v)
+}
+
 // ackLevel updates the ack level for the given cluster
 func (t *TaskAckManager) ackLevel(pollingCluster string, lastReadTaskID int64) {
 	if err := t.ackLevels.UpdateQueueClusterAckLevel(persistence.HistoryTaskCategoryReplication, pollingCluster, persistence.NewImmediateTaskKey(lastReadTaskID)); err != nil {