rebase from stream connection

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

docs/configuration/config-file-reference.md

@@ -2768,16 +2768,16 @@ ha_tracker:
 # CLI flag: -distributor.sign-write-requests
 [sign_write_requests: <boolean> | default = false]
 
-# EXPERIMENTAL: If true, accept prometheus remote write v2 protocol push
-# request.
-# CLI flag: -distributor.remote-write2-enabled
-[remote_write2_enabled: <boolean> | default = false]
-
 # EXPERIMENTAL: If enabled, distributor would use stream connection to send
 # requests to ingesters.
 # CLI flag: -distributor.use-stream-push
 [use_stream_push: <boolean> | default = false]
 
+# EXPERIMENTAL: If true, accept prometheus remote write v2 protocol push
+# request.
+# CLI flag: -distributor.remote-write2-enabled
+[remote_write2_enabled: <boolean> | default = false]
+
 ring:
   kvstore:
     # Backend storage to use for the ring. Supported values are: consul, etcd,

pkg/cortexpb/cortex.proto

@@ -36,14 +36,12 @@ message StreamWriteRequest {
 message WriteResponse {
   int32 code = 1;
   string message = 2;
-}
-message WriteResponse {
   // Samples represents X-Prometheus-Remote-Write-Written-Samples
-  int64 Samples = 1;
+  int64 Samples = 3;
   // Histograms represents X-Prometheus-Remote-Write-Written-Histograms
-  int64 Histograms = 2;
+  int64 Histograms = 4;
   // Exemplars represents X-Prometheus-Remote-Write-Written-Exemplars
-  int64 Exemplars = 3;
+  int64 Exemplars = 5;
 }
 
 message TimeSeries {

pkg/distributor/distributor.go

@@ -1178,13 +1178,14 @@ func (d *Distributor) send(ctx context.Context, ingester ring.InstanceDesc, time
 	d.inflightClientRequests.Inc()
 	defer d.inflightClientRequests.Dec()
 
+	var resp *cortexpb.WriteResponse
 	if d.cfg.UseStreamPush {
 		req := &cortexpb.WriteRequest{
 			Timeseries: timeseries,
 			Metadata:   metadata,
 			Source:     source,
 		}
-		_, err = c.PushStreamConnection(ctx, req)
+		resp, err = c.PushStreamConnection(ctx, req)
 	} else {
 		req := cortexpb.PreallocWriteRequestFromPool()
 		req.Timeseries = timeseries