Introduce payload size metrics (#6745)

* Introduce payload size metrics

* code cleanup

* formatting

Author: tubignat

common/metrics/defs.go

@@ -2106,6 +2106,7 @@ const (
 	PersistenceSampledCounter
 	PersistenceEmptyResponseCounter
 	PersistenceResponseRowSize
+	PersistenceResponsePayloadSize
 
 	PersistenceRequestsPerDomain
 	PersistenceRequestsPerShard
@@ -2812,6 +2813,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		PersistenceSampledCounter:                                    {metricName: "persistence_sampled", metricType: Counter},
 		PersistenceEmptyResponseCounter:                              {metricName: "persistence_empty_response", metricType: Counter},
 		PersistenceResponseRowSize:                                   {metricName: "persistence_response_row_size", metricType: Histogram, buckets: ResponseRowSizeBuckets},
+		PersistenceResponsePayloadSize:                               {metricName: "persistence_response_payload_size", metricType: Histogram, buckets: ResponsePayloadSizeBuckets},
 		PersistenceRequestsPerDomain:                                 {metricName: "persistence_requests_per_domain", metricRollupName: "persistence_requests", metricType: Counter},
 		PersistenceRequestsPerShard:                                  {metricName: "persistence_requests_per_shard", metricType: Counter},
 		PersistenceFailuresPerDomain:                                 {metricName: "persistence_errors_per_domain", metricRollupName: "persistence_errors", metricType: Counter},
@@ -3535,6 +3537,12 @@ var ResponseRowSizeBuckets = append(
 	tally.MustMakeExponentialValueBuckets(1, 2, 17)..., // 1..65536
 )
 
+// ResponsePayloadSizeBuckets contains buckets for tracking the size of the payload returned per persistence operation
+var ResponsePayloadSizeBuckets = append(
+	tally.ValueBuckets{0},                                 // need an explicit 0 or zero is reported as 1
+	tally.MustMakeExponentialValueBuckets(1024, 2, 20)..., // 1kB..1GB
+)
+
 // ErrorClass is an enum to help with classifying SLA vs. non-SLA errors (SLA = "service level agreement")
 type ErrorClass uint8
 

common/persistence/metered.go

@@ -23,6 +23,8 @@
 package persistence
 
 import (
+	"unsafe"
+
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/metrics"
 )
@@ -71,6 +73,248 @@ func (r GetAllHistoryTreeBranchesResponse) Len() int {
 	return len(r.Branches)
 }
 
+// For responses that require metrics for payload size ByteSize() uint64 should be defined.
+
+func (r *GetReplicationTasksResponse) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	size := uint64(int(unsafe.Sizeof(*r)) + len(r.NextPageToken))
+	for _, v := range r.Tasks {
+		size += v.ByteSize()
+	}
+
+	return size
+}
+
+func (r *ReplicationTaskInfo) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	return uint64(int(unsafe.Sizeof(*r)) + len(r.DomainID) + len(r.WorkflowID) + len(r.RunID) + len(r.BranchToken) + len(r.NewRunBranchToken))
+}
+
+func (r *GetTimerIndexTasksResponse) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	size := uint64(int(unsafe.Sizeof(*r)) + len(r.NextPageToken))
+	for _, v := range r.Timers {
+		size += v.ByteSize()
+	}
+
+	return size
+}
+
+func (r *TimerTaskInfo) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	return uint64(int(unsafe.Sizeof(*r)) + len(r.DomainID) + len(r.WorkflowID) + len(r.RunID))
+}
+
+func (r *GetTasksResponse) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	size := uint64(unsafe.Sizeof(*r))
+	for _, v := range r.Tasks {
+		size += v.ByteSize()
+	}
+
+	return size
+}
+
+func (r *TaskInfo) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	return uint64(int(unsafe.Sizeof(*r)) + len(r.DomainID) + len(r.WorkflowID) + len(r.RunID) + estimateStringMapSize(r.PartitionConfig))
+}
+
+func (r *ListDomainsResponse) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	size := uint64(int(unsafe.Sizeof(*r)) + len(r.NextPageToken))
+	for _, v := range r.Domains {
+		size += v.ByteSize()
+	}
+
+	return size
+}
+
+func (r *GetDomainResponse) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	return uint64(unsafe.Sizeof(*r)) + r.Info.ByteSize() + r.Config.ByteSize() + r.ReplicationConfig.ByteSize()
+}
+
+func (i *DomainInfo) ByteSize() uint64 {
+	if i == nil {
+		return 0
+	}
+
+	return uint64(int(unsafe.Sizeof(*i)) + len(i.ID) + len(i.Name) + len(i.Description) + len(i.OwnerEmail) + estimateStringMapSize(i.Data))
+}
+
+func (c *DomainConfig) ByteSize() uint64 {
+	if c == nil {
+		return 0
+	}
+
+	size := int(unsafe.Sizeof(*c)) + len(c.HistoryArchivalURI) + len(c.VisibilityArchivalURI)
+
+	asyncWorkflowConfigSize := int(unsafe.Sizeof(c.AsyncWorkflowConfig)) + len(c.AsyncWorkflowConfig.PredefinedQueueName) + len(c.AsyncWorkflowConfig.QueueType)
+	if c.AsyncWorkflowConfig.QueueConfig != nil {
+		size += len(c.AsyncWorkflowConfig.QueueConfig.Data)
+	}
+
+	binariesSize := 0
+	for key, value := range c.BadBinaries.Binaries {
+		binariesSize += len(key)
+		if value != nil {
+			binariesSize += len(value.Reason) + len(value.Operator)
+		}
+	}
+
+	isolationGroupsSize := 0
+	for key, value := range c.IsolationGroups {
+		binariesSize += len(key) + len(value.Name)
+	}
+
+	return uint64(size + asyncWorkflowConfigSize + binariesSize + isolationGroupsSize)
+}
+
+func (c *DomainReplicationConfig) ByteSize() uint64 {
+	if c == nil {
+		return 0
+	}
+
+	total := len(c.ActiveClusterName)
+	for _, v := range c.Clusters {
+		if v == nil {
+			continue
+		}
+		total += len(v.ClusterName)
+	}
+	return uint64(total)
+}
+
+func estimateStringMapSize(m map[string]string) int {
+	size := 0
+	for key, value := range m {
+		size += len(key) + len(value)
+	}
+	return size
+}
+
+func (r *ReadRawHistoryBranchResponse) Size2() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	total := uint64(int(unsafe.Sizeof(*r)) + len(r.NextPageToken))
+	for _, v := range r.HistoryEventBlobs {
+		total += v.ByteSize()
+	}
+	return total
+}
+
+func (d *DataBlob) ByteSize() uint64 {
+	if d == nil {
+		return 0
+	}
+
+	return uint64(int(unsafe.Sizeof(*d)) + len(d.Data))
+}
+
+func (r *ListCurrentExecutionsResponse) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	total := uint64(int(unsafe.Sizeof(*r)) + len(r.PageToken))
+	for _, v := range r.Executions {
+		total += v.ByteSize()
+	}
+
+	return total
+}
+
+func (r *CurrentWorkflowExecution) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	return uint64(int(unsafe.Sizeof(*r)) + len(r.DomainID) + len(r.WorkflowID) + len(r.RunID) + len(r.CurrentRunID))
+}
+
+func (r *GetTransferTasksResponse) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	total := uint64(int(unsafe.Sizeof(*r)) + len(r.NextPageToken))
+	for _, v := range r.Tasks {
+		total += v.ByteSize()
+	}
+
+	return total
+}
+
+func (r *TransferTaskInfo) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	return uint64(int(unsafe.Sizeof(*r)) + len(r.DomainID) + len(r.WorkflowID) + len(r.RunID) +
+		len(r.TargetDomainID) + len(r.TargetWorkflowID) + len(r.TargetRunID) + len(r.TaskList))
+}
+
+func (r QueueMessageList) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	total := uint64(0)
+	for _, v := range r {
+		total += v.ByteSize()
+	}
+
+	return total
+}
+
+func (r *QueueMessage) ByteSize() uint64 {
+	if r == nil {
+		return 0
+	}
+
+	return uint64(int(unsafe.Sizeof(*r)) + len(r.Payload))
+}
+
+func (r GetAllHistoryTreeBranchesResponse) ByteSize() uint64 {
+	total := uint64(int(unsafe.Sizeof(r)) + len(r.NextPageToken))
+	for _, v := range r.Branches {
+		total += v.ByteSize()
+	}
+
+	return total
+}
+
+func (r HistoryBranchDetail) ByteSize() uint64 {
+	return uint64(int(unsafe.Sizeof(r)) + len(r.TreeID) + len(r.BranchID) + len(r.Info))
+}
+
 // If MetricTags() []metrics.Tag is defined, then metrics will be emitted for the request.
 
 func (r ReadHistoryBranchRequest) MetricTags() []metrics.Tag {