Repository: argoproj-labs/old-argo-dataflow (language: python)

Issue:
Total number of bytes processed metric
This would be a useful metric for:

* How well is an org adopting Argo Dataflow?
* What is the volume of data it is processing?

See #201


Discussion / hints:
@sarabala1979 this is a useful message in the Kafka metrics we have from out event bus. Might be a good one to add.
A bit more ideas on this:

* Maybe call this "message quantity"?
* This is a attribute needs to be counted for each replica and source, just list "total" and "rate"
* Message quantity is likely to end up large value in bytes, so we should use `resource.Quantity` not int.
* The lead replica should be responsible for emitting the Prometheus metric, same as other metrics.
* We should probably update the UI to show this value (low priority).
* We'll need to update our Wavefront dashboard.

Fix patch:
diff --git a/api/v1alpha1/metrics.go b/api/v1alpha1/metrics.go
index 509588fa..dfa0ece1 100644
--- a/api/v1alpha1/metrics.go
+++ b/api/v1alpha1/metrics.go
@@ -5,8 +5,9 @@ import (
 )
 
 type Metrics struct {
-	Total   uint64            `json:"total,omitempty" protobuf:"varint,1,opt,name=total"`
-	Errors  uint64            `json:"errors,omitempty" protobuf:"varint,2,opt,name=errors"`
-	Rate    resource.Quantity `json:"rate,omitempty" protobuf:"bytes,3,opt,name=rate"` // current rate of messages per second
-	Retries uint64            `json:"retries,omitempty" protobuf:"bytes,4,opt,name=retries"`
+	Total      uint64            `json:"total,omitempty" protobuf:"varint,1,opt,name=total"`
+	Errors     uint64            `json:"errors,omitempty" protobuf:"varint,2,opt,name=errors"`
+	Rate       resource.Quantity `json:"rate,omitempty" protobuf:"bytes,3,opt,name=rate"` // current rate of messages per second
+	Retries    uint64            `json:"retries,omitempty" protobuf:"bytes,4,opt,name=retries"`
+	TotalBytes uint64            `json:"totalBytes,omitempty" protobuf:"varint,5,opt,name=totalBytes"`
 }
diff --git a/api/v1alpha1/source_status.go b/api/v1alpha1/source_status.go
index 8c65c5a1..863894bb 100644
--- a/api/v1alpha1/source_status.go
+++ b/api/v1alpha1/source_status.go
@@ -46,3 +46,11 @@ func (in SourceStatus) GetRetries() uint64 {
 	}
 	return x
 }
+
+func (in SourceStatus) GetTotalBytes() uint64 {
+	var x uint64
+	for _, m := range in.Metrics {
+		x += m.TotalBytes
+	}
+	return x
+}
diff --git a/api/v1alpha1/source_statuses.go b/api/v1alpha1/source_statuses.go
index 5776b8d4..d49940ff 100644
--- a/api/v1alpha1/source_statuses.go
+++ b/api/v1alpha1/source_statuses.go
@@ -8,7 +8,7 @@ import (
 
 type SourceStatuses map[string]SourceStatus // key is source name
 
-func (in SourceStatuses) IncrTotal(name string, replica int, rate resource.Quantity) {
+func (in SourceStatuses) IncrTotal(name string, replica int, rate resource.Quantity, msgSize uint64) {
 	x := in[name]
 	if x.Metrics == nil {
 		x.Metrics = map[string]Metrics{}
@@ -16,6 +16,7 @@ func (in SourceStatuses) IncrTotal(name string, replica int, rate resource.Quant
 	m := x.Metrics[strconv.Itoa(replica)]
 	m.Total++
 	m.Rate = rate
+	m.TotalBytes += msgSize
 	x.Metrics[strconv.Itoa(replica)] = m
 	in[name] = x
 }
@@ -103,3 +104,11 @@ func (in SourceStatuses) IncrRetries(name string, replica int) {
 	x.Metrics[strconv.Itoa(replica)] = m
 	in[name] = x
 }
+
+func (in SourceStatuses) GetTotalBytes() uint64 {
+	var v uint64
+	for _, s := range in {
+		v += s.GetTotalBytes()
+	}
+	return v
+}
diff --git a/config/ci.yaml b/config/ci.yaml
index 1b9f823b..3b608133 100644
--- a/config/ci.yaml
+++ b/config/ci.yaml
@@ -13275,6 +13275,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
@@ -13306,6 +13309,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
diff --git a/config/crd/bases/dataflow.argoproj.io_steps.yaml b/config/crd/bases/dataflow.argoproj.io_steps.yaml
index 46e70e34..36f66b02 100644
--- a/config/crd/bases/dataflow.argoproj.io_steps.yaml
+++ b/config/crd/bases/dataflow.argoproj.io_steps.yaml
@@ -6428,6 +6428,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
@@ -6459,6 +6462,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
diff --git a/config/default.yaml b/config/default.yaml
index 1af4f617..8a7cb29b 100644
--- a/config/default.yaml
+++ b/config/default.yaml
@@ -13275,6 +13275,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
@@ -13306,6 +13309,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
diff --git a/config/dev.yaml b/config/dev.yaml
index 97970664..0cbfc1a1 100644
--- a/config/dev.yaml
+++ b/config/dev.yaml
@@ -13275,6 +13275,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
@@ -13306,6 +13309,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
diff --git a/config/quick-start.yaml b/config/quick-start.yaml
index 1af4f617..8a7cb29b 100644
--- a/config/quick-start.yaml
+++ b/config/quick-start.yaml
@@ -13275,6 +13275,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
@@ -13306,6 +13309,9 @@ spec:
                           total:
                             format: int64
                             type: integer
+                          totalBytes:
+                            format: int64
+                            type: integer
                         type: object
                       type: object
                     pending:
diff --git a/examples/git/go.mod b/examples/git/go.mod
index dd729144..ddaf9702 100644
--- a/examples/git/go.mod
+++ b/examples/git/go.mod
@@ -2,4 +2,4 @@ module github.com/argoproj-labs/argo-dataflow/examples/git
 
 go 1.16
 
-require github.com/argoproj-labs/argo-dataflow v0.0.90
+require github.com/argoproj-labs/argo-dataflow v0.0.93
diff --git a/examples/git/go.sum b/examples/git/go.sum
index ad1069c5..11e29277 100644
--- a/examples/git/go.sum
+++ b/examples/git/go.sum
@@ -70,8 +70,8 @@ github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRF
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
 github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
 github.com/antonmedv/expr v1.8.9/go.mod h1:5qsM3oLGDND7sDmQGDXHkYfkjYMUX14qsgqmHhwGEk8=
-github.com/argoproj-labs/argo-dataflow v0.0.90 h1:wbCzUSi3bZxeQudGCQi6Bxjv54bkxIXvPsLUoae3DZI=
-github.com/argoproj-labs/argo-dataflow v0.0.90/go.mod h1:HOdmu21+Xx59R28S8+e+wMx+/ApHWNtglPqLvgLRI2E=
+github.com/argoproj-labs/argo-dataflow v0.0.93 h1:yx6rscLQua7/QfVa80TkfQ6utkSdxOvqqWvFMdeOp74=
+github.com/argoproj-labs/argo-dataflow v0.0.93/go.mod h1:HOdmu21+Xx59R28S8+e+wMx+/ApHWNtglPqLvgLRI2E=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878/go.mod h1:3AMJUQhVx52RsWOnlkpikZr01T/yAVN2gn0861vByNg=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
diff --git a/runner/sidecar/sinks.go b/runner/sidecar/sinks.go
index c4907abd..f8ea1f84 100644
--- a/runner/sidecar/sinks.go
+++ b/runner/sidecar/sinks.go
@@ -84,7 +84,7 @@ func connectSinks(ctx context.Context) (func([]byte) error, error) {
 			counter := rateCounters[sinkName]
 			counter.Incr(1)
 			withLock(func() {
-				step.Status.SinkStatues.IncrTotal(sinkName, replica, rateToResourceQuantity(counter))
+				step.Status.SinkStatues.IncrTotal(sinkName, replica, rateToResourceQuantity(counter), uint64(len(msg)))
 			})
 			if err := f.Sink(msg); err != nil {
 				withLock(func() { step.Status.SinkStatues.IncrErrors(sinkName, replica) })
diff --git a/runner/sidecar/sources.go b/runner/sidecar/sources.go
index aef6e17e..4cb9039b 100644
--- a/runner/sidecar/sources.go
+++ b/runner/sidecar/sources.go
@@ -40,7 +40,7 @@ func connectSources(ctx context.Context, toMain func(context.Context, []byte) er
 		f := func(ctx context.Context, msg []byte) error {
 			rateCounter.Incr(1)
 			withLock(func() {
-				step.Status.SourceStatuses.IncrTotal(sourceName, replica, rateToResourceQuantity(rateCounter))
+				step.Status.SourceStatuses.IncrTotal(sourceName, replica, rateToResourceQuantity(rateCounter), uint64(len(msg)))
 			})
 			backoff := newBackoff(s.Retry)
 			for {
@@ -176,4 +176,15 @@ func newSourceMetrics(source dfv1.Source, sourceName string) {
 		defer mu.RUnlock()
 		return float64(step.Status.SourceStatuses.Get(sourceName).GetRetries())
 	})
+
+	promauto.NewCounterFunc(prometheus.CounterOpts{
+		Subsystem:   "sources",
+		Name:        "totalBytes",
+		Help:        "Total number of bytes processed, see https://github.com/argoproj-labs/argo-dataflow/blob/main/docs/METRICS.md#sources_retries",
+		ConstLabels: map[string]string{"sourceName": source.Name},
+	}, func() float64 {
+		mu.RLock()
+		defer mu.RUnlock()
+		return float64(step.Status.SourceStatuses.Get(sourceName).GetTotalBytes())
+	})
 }
diff --git a/runtimes/golang1-16/go.mod b/runtimes/golang1-16/go.mod
index 176405a6..fa4a138a 100644
--- a/runtimes/golang1-16/go.mod
+++ b/runtimes/golang1-16/go.mod
@@ -2,4 +2,4 @@ module github.com/argoproj-labs/argo-dataflow/runtimes/golang1-16
 
 go 1.16
 
-require github.com/argoproj-labs/argo-dataflow v0.0.90
+require github.com/argoproj-labs/argo-dataflow v0.0.93
diff --git a/runtimes/golang1-16/go.sum b/runtimes/golang1-16/go.sum
index ad1069c5..11e29277 100644
--- a/runtimes/golang1-16/go.sum
+++ b/runtimes/golang1-16/go.sum
@@ -70,8 +70,8 @@ github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRF
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
 github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
 github.com/antonmedv/expr v1.8.9/go.mod h1:5qsM3oLGDND7sDmQGDXHkYfkjYMUX14qsgqmHhwGEk8=
-github.com/argoproj-labs/argo-dataflow v0.0.90 h1:wbCzUSi3bZxeQudGCQi6Bxjv54bkxIXvPsLUoae3DZI=
-github.com/argoproj-labs/argo-dataflow v0.0.90/go.mod h1:HOdmu21+Xx59R28S8+e+wMx+/ApHWNtglPqLvgLRI2E=
+github.com/argoproj-labs/argo-dataflow v0.0.93 h1:yx6rscLQua7/QfVa80TkfQ6utkSdxOvqqWvFMdeOp74=
+github.com/argoproj-labs/argo-dataflow v0.0.93/go.mod h1:HOdmu21+Xx59R28S8+e+wMx+/ApHWNtglPqLvgLRI2E=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/armon/go-metrics v0.0.0-20190430140413-ec5e00d3c878/go.mod h1:3AMJUQhVx52RsWOnlkpikZr01T/yAVN2gn0861vByNg=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=


Test patch:
diff --git a/api/v1alpha1/source_status_test.go b/api/v1alpha1/source_status_test.go
index d2248b13..1bf7e2fc 100644
--- a/api/v1alpha1/source_status_test.go
+++ b/api/v1alpha1/source_status_test.go
@@ -44,3 +44,22 @@ func TestSourceStatus_GetRetries(t *testing.T) {
 		assert.Equal(t, uint64(2), x.GetRetries())
 	})
 }
+
+func TestSourceStatus_GetTotalBytes(t *testing.T) {
+	t.Run("None", func(t *testing.T) {
+		x := SourceStatus{}
+		assert.Equal(t, uint64(0), x.GetTotalBytes())
+	})
+	t.Run("One", func(t *testing.T) {
+		x := SourceStatus{
+			Metrics: map[string]Metrics{"one": {TotalBytes: 100}},
+		}
+		assert.Equal(t, uint64(100), x.GetTotalBytes())
+	})
+	t.Run("Two", func(t *testing.T) {
+		x := SourceStatus{
+			Metrics: map[string]Metrics{"one": {TotalBytes: 100}, "two": {TotalBytes: 100}},
+		}
+		assert.Equal(t, uint64(200), x.GetTotalBytes())
+	})
+}
diff --git a/api/v1alpha1/source_statuses_test.go b/api/v1alpha1/source_statuses_test.go
index 8931d939..b6081fa4 100644
--- a/api/v1alpha1/source_statuses_test.go
+++ b/api/v1alpha1/source_statuses_test.go
@@ -10,37 +10,40 @@ import (
 func TestSourceStatuses_Set(t *testing.T) {
 	ss := SourceStatuses{}
 
-	ss.IncrTotal("bar", 1, resource.MustParse("1"))
+	ss.IncrTotal("bar", 1, resource.MustParse("1"), 1)
 
 	if assert.Len(t, ss, 1) {
 		s := ss["bar"]
 		if assert.Len(t, s.Metrics, 1) {
 			assert.Equal(t, uint64(1), s.Metrics["1"].Total)
 			assert.Equal(t, resource.MustParse("1"), s.Metrics["1"].Rate)
+			assert.Equal(t, uint64(1), s.Metrics["1"].TotalBytes)
 		}
 	}
 
-	ss.IncrTotal("bar", 1, resource.MustParse("1"))
+	ss.IncrTotal("bar", 1, resource.MustParse("1"), 1)
 
 	if assert.Len(t, ss, 1) {
 		s := ss["bar"]
 		if assert.Len(t, s.Metrics, 1) {
 			assert.Equal(t, uint64(2), s.Metrics["1"].Total)
 			assert.Equal(t, resource.MustParse("1"), s.Metrics["1"].Rate)
+			assert.Equal(t, uint64(2), s.Metrics["1"].TotalBytes)
 		}
 	}
 
-	ss.IncrTotal("bar", 0, resource.MustParse("1"))
+	ss.IncrTotal("bar", 0, resource.MustParse("1"), 1)
 
 	if assert.Len(t, ss, 1) {
 		s := ss["bar"]
 		if assert.Len(t, s.Metrics, 2) {
 			assert.Equal(t, uint64(1), s.Metrics["0"].Total)
 			assert.Equal(t, uint64(2), s.Metrics["1"].Total)
+			assert.Equal(t, uint64(2), s.Metrics["1"].TotalBytes)
 		}
 	}
 
-	ss.IncrTotal("baz", 0, resource.MustParse("1"))
+	ss.IncrTotal("baz", 0, resource.MustParse("1"), 1)
 
 	if assert.Len(t, ss, 2) {
 		s := ss["baz"]
diff --git a/test/e2e/metrics_test.go b/test/e2e/metrics_test.go
index 3bb6b5c4..565e7f01 100644
--- a/test/e2e/metrics_test.go
+++ b/test/e2e/metrics_test.go
@@ -49,4 +49,7 @@ func TestMetrics(t *testing.T) {
 	ExpectMetric("sources_pending", 0)
 	ExpectMetric("sources_total", 1)
 	ExpectMetric("sources_retries", 0)
+	ExpectMetric("sources_totalBytes", 6)
+	SendMessageViaHTTP("my-msg")
+	ExpectMetric("sources_totalBytes", 12)
 }