Introduce a metric to track the number of bytes we're sending to Kusto

ingestor/adx/uploader.go

@@ -284,7 +284,7 @@ func (n *uploader) upload(ctx context.Context) error {
 					}
 				}
 
-				mr := io.MultiReader(readers...)
+				mr := newCountingReader(io.MultiReader(readers...))
 
 				now := time.Now()
 				if err := n.uploadReader(mr, database, table, mapping); err != nil {
@@ -310,6 +310,8 @@ func (n *uploader) upload(ctx context.Context) error {
 						metrics.LogsUploaded.WithLabelValues(database, table).Add(float64(sampleCount))
 					}
 				}
+
+				metrics.IngestorUploadedBytes.WithLabelValues(database, table).Set(float64(mr.Count()))
 			}()
 
 		}
@@ -388,3 +390,32 @@ type clusterDetails struct {
 	ClockDescription       string    `kusto:"ClockDescription"`
 	RuntimeDescription     string    `kusto:"RuntimeDescription"`
 }
+
+// countingReader wraps an io.Reader and counts bytes read without allocations.
+type countingReader struct {
+	reader io.Reader
+	count  int64
+}
+
+// Read implements io.Reader without allocations.
+func (c *countingReader) Read(p []byte) (int, error) {
+	n, err := c.reader.Read(p)
+	c.count += int64(n)
+	return n, err
+}
+
+// Count returns the total bytes read.
+func (c *countingReader) Count() int64 {
+	return c.count
+}
+
+// newCountingReader returns a countingReader wrapping the provided reader.
+func newCountingReader(r io.Reader) *countingReader {
+	return &countingReader{reader: r}
+}
+
+// Reset allows reusing countingReader by resetting its internal state.
+func (c *countingReader) Reset(r io.Reader) {
+	c.reader = r
+	c.count = 0
+}

ingestor/adx/uploader_test.go

@@ -1,7 +1,9 @@
 package adx
 
 import (
+	"bytes"
 	"context"
+	"io"
 	"testing"
 
 	"github.com/Azure/adx-mon/pkg/testutils"
@@ -32,3 +34,20 @@ func TestClusterRequiresDirectIngest(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, requiresDirectIngest)
 }
+
+func BenchmarkCountingReader(b *testing.B) {
+	data := bytes.Repeat([]byte("testdata"), 1024)
+	reader := bytes.NewReader(data)
+	countReader := newCountingReader(reader)
+
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		reader.Reset(data)
+		countReader.Reset(reader)
+
+		_, _ = io.Copy(io.Discard, countReader)
+		if countReader.Count() != int64(len(data)) {
+			b.Fatalf("expected %d bytes read, got %d", len(data), countReader.Count())
+		}
+	}
+}

metrics/metrics.go

@@ -93,6 +93,13 @@ var (
 		Help:      "Counter of the number of logs uploaded to Kusto",
 	}, []string{"database", "table"})
 
+	IngestorUploadedBytes = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: Namespace,
+		Subsystem: "ingestor",
+		Name:      "uploaded_size_bytes",
+		Help:      "Gauge for the size of uploaded data to Kusto in bytes",
+	}, []string{"database", "table"})
+
 	InvalidLogsDropped = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: Namespace,
 		Subsystem: "ingestor",