diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3aebeb6e222..33cd0f54285 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,7 @@
 * [ENHANCEMENT] Add docker-compose example for GCS along with new backend options [#397](https://github.com/grafana/tempo/pull/397)
 * [ENHANCEMENT] tempo-cli list blocks usability improvements [#403](https://github.com/grafana/tempo/pull/403)
 * [ENHANCEMENT] Add Query Frontend module to allow scaling the query path [#400](https://github.com/grafana/tempo/pull/400)
+* [ENHANCEMENT] Added `tempo_distributor_bytes_received_total` as a per tenant counter of uncompressed bytes received. [#453](https://github.com/grafana/tempo/pull/453)
 * [BUGFIX] Compactor without GCS permissions fail silently [#379](https://github.com/grafana/tempo/issues/379)
 * [BUGFIX] Prevent race conditions between querier polling and ingesters clearing complete blocks [#421](https://github.com/grafana/tempo/issues/421)
 * [BUGFIX] Exclude blocks in last active window from compaction [#411](https://github.com/grafana/tempo/pull/411)
diff --git a/modules/distributor/distributor.go b/modules/distributor/distributor.go
index 9f049fd9e2f..4d996983b32 100644
--- a/modules/distributor/distributor.go
+++ b/modules/distributor/distributor.go
@@ -53,6 +53,11 @@ var (
 		Name:      "distributor_spans_received_total",
 		Help:      "The total number of spans received per tenant",
 	}, []string{"tenant"})
+	metricBytesIngested = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "tempo",
+		Name:      "distributor_bytes_received_total",
+		Help:      "The total number of proto bytes received per tenant",
+	}, []string{"tenant"})
 	metricTracesPerBatch = promauto.NewHistogram(prometheus.HistogramOpts{
 		Namespace: "tempo",
 		Name:      "distributor_traces_per_batch",
@@ -194,7 +199,11 @@ func (d *Distributor) Push(ctx context.Context, req *tempopb.PushRequest) (*temp
 		return nil, err
 	}
 
-	// Track metrics.
+	// calculate and metric size...
+	size := req.Size()
+	metricBytesIngested.WithLabelValues(userID).Add(float64(size))
+
+	// ... and spans
 	if req.Batch == nil {
 		return &tempopb.PushResponse{}, nil
 	}
@@ -207,6 +216,7 @@ func (d *Distributor) Push(ctx context.Context, req *tempopb.PushRequest) (*temp
 	}
 	metricSpansIngested.WithLabelValues(userID).Add(float64(spanCount))
 
+	// check limits
 	now := time.Now()
 	if !d.ingestionRateLimiter.AllowN(now, userID, spanCount) {
 		// Return a 4xx here to have the client discard the data and not retry. If a client