Fixed nil pointer error from PushStream in ingester when Push return nil response with an error (#6795)

Signed-off-by: Alex Le <[email protected]>

Author: alexqyle

integration/ingester_stream_push_test.go

@@ -116,3 +116,69 @@ func TestIngesterStreamPushConnection(t *testing.T) {
 	assertServiceMetricsPrefixes(t, Ingester, ingester2)
 	assertServiceMetricsPrefixes(t, Ingester, ingester3)
 }
+
+func TestIngesterStreamPushConnectionWithError(t *testing.T) {
+
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	maxGlobalSeriesPerMetric := 300
+	maxGlobalSeriesPerTenant := 1000
+
+	flags := BlocksStorageFlags()
+	flags["-distributor.use-stream-push"] = "true"
+	flags["-distributor.replication-factor"] = "1"
+	flags["-distributor.shard-by-all-labels"] = "true"
+	flags["-distributor.sharding-strategy"] = "shuffle-sharding"
+	flags["-distributor.ingestion-tenant-shard-size"] = "1"
+	flags["-ingester.max-series-per-user"] = "0"
+	flags["-ingester.max-series-per-metric"] = "0"
+	flags["-ingester.max-global-series-per-user"] = strconv.Itoa(maxGlobalSeriesPerTenant)
+	flags["-ingester.max-global-series-per-metric"] = strconv.Itoa(maxGlobalSeriesPerMetric)
+	flags["-ingester.heartbeat-period"] = "1s"
+	flags["-ingester.instance-limits.max-series"] = "1" // trigger "max series limit reached" error which returns nil response
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+	// Start Cortex components.
+	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ingester1 := e2ecortex.NewIngester("ingester-1", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ingester2 := e2ecortex.NewIngester("ingester-2", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ingester3 := e2ecortex.NewIngester("ingester-3", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester1, ingester2, ingester3))
+
+	// Wait until distributor has updated the ring.
+	require.NoError(t, distributor.WaitSumMetricsWithOptions(e2e.Equals(3), []string{"cortex_ring_members"}, e2e.WithLabelMatchers(
+		labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"),
+		labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE"))))
+
+	// Wait until ingesters have heartbeated the ring after all ingesters were active,
+	// in order to update the number of instances. Since we have no metric, we have to
+	// rely on a ugly sleep.
+	time.Sleep(2 * time.Second)
+
+	now := time.Now()
+	client, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", userID)
+	require.NoError(t, err)
+
+	for i := 0; i < 5; i++ {
+		series, _ := generateSeries("test_limit_per_metric", now,
+			prompb.Label{
+				Name:  "cardinality",
+				Value: strconv.Itoa(rand.Int()),
+			},
+		)
+		_, err = client.Push(series)
+		require.NoError(t, err)
+		err = ingester1.WaitForRunning()
+		require.NoError(t, err)
+		err = ingester2.WaitForRunning()
+		require.NoError(t, err)
+		err = ingester3.WaitForRunning()
+		require.NoError(t, err)
+	}
+}

pkg/ingester/ingester.go

@@ -1577,6 +1577,9 @@ func (i *Ingester) PushStream(srv client.Ingester_PushStreamServer) error {
 		}
 		ctx = user.InjectOrgID(ctx, req.TenantID)
 		resp, err := i.Push(ctx, req.Request)
+		if resp == nil {
+			resp = &cortexpb.WriteResponse{}
+		}
 		resp.Code = http.StatusOK
 		if err != nil {
 			httpResponse, isGRPCError := httpgrpc.HTTPResponseFromError(err)