app/vlstorage/netinsert: drain buffered data on graceful shutdown ins…

app/vlstorage/netinsert/netinsert.go

@@ -1,6 +1,7 @@
 package netinsert
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -11,7 +12,6 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/contextutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
@@ -24,6 +24,11 @@ import (
 	"github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage"
 )
 
+// the maximum duration for sending a single data block to a storage node.
+//
+// This is consistent with -remoteWrite.sendTimeout in vlagent.
+const sendTimeout = time.Minute
+
 // the maximum size of a single data block sent to storage node.
 const maxInsertBlockSize = 2 * 1024 * 1024
 
@@ -257,7 +262,7 @@ func (sn *storageNode) sendInsertRequest(pendingData *bytesutil.ByteBuffer) erro
 }
 
 func (sn *storageNode) doRequest(path string, body io.Reader) error {
-	ctx, cancel := contextutil.NewStopChanContext(sn.s.stopCh)
+	ctx, cancel := context.WithTimeout(context.Background(), sendTimeout)
 	defer cancel()
 
 	method := "GET"

app/vlstorage/netinsert/netinsert_test.go

@@ -2,13 +2,62 @@ package netinsert
 
 import (
 	"fmt"
+	"io"
 	"math"
 	"math/rand"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/cespare/xxhash/v2"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+
+	"github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage"
 )
 
+func TestStorageDrainsPendingDataOnShutdown(t *testing.T) {
+	var insertRequests atomic.Int64
+	var receivedBytes atomic.Int64
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/internal/insert" {
+			body, _ := io.ReadAll(r.Body)
+			receivedBytes.Add(int64(len(body)))
+			insertRequests.Add(1)
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer ts.Close()
+
+	ac, err := (&promauth.Options{}).NewConfig()
+	if err != nil {
+		t.Fatalf("cannot create auth config: %s", err)
+	}
+
+	addr := strings.TrimPrefix(ts.URL, "http://")
+	s := NewStorage([]string{addr}, []*promauth.Config{ac}, []bool{false}, 1, true)
+
+	// Buffer a single small row. It stays pending (it doesn't reach maxInsertBlockSize),
+	// so it is only sent by the final flush performed during shutdown.
+	r := &logstorage.InsertRow{
+		Timestamp: 1,
+		Fields:    []logstorage.Field{{Name: "foo", Value: "bar"}},
+	}
+	s.AddRow(0, r)
+
+	// MustStop must drain the buffered row to the storage node instead of dropping it.
+	s.MustStop()
+
+	if n := insertRequests.Load(); n == 0 {
+		t.Fatalf("expected the buffered data to be drained to the storage node on shutdown; got no insert requests")
+	}
+	if n := receivedBytes.Load(); n == 0 {
+		t.Fatalf("expected the storage node to receive non-empty buffered data on shutdown")
+	}
+}
+
 func TestStreamRowsTracker(t *testing.T) {
 	f := func(rowsCount, streamsCount, nodesCount int) {
 		t.Helper()

docs/victorialogs/CHANGELOG.md

@@ -38,6 +38,7 @@ according to the following docs:
 * BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): avoid `cannot connect to storage node at ...: EOF` errors after `vlselect` or `vlinsert` was idle for more than 60 seconds. See [#1440](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1440).
 * BUGFIX: [vlselect](https://docs.victoriametrics.com/victorialogs/cluster/): return `502 Bad Gateway` HTTP response code for incoming queries when one of the `vlstorage` nodes runs a VictoriaLogs version with an incompatible internal API instead of `400 Bad Request`. This is consistent with the `502 Bad Gateway` response returned when a `vlstorage` node is unavailable, and it allows building a proper failover scheme in high-availability setups. See [these docs](https://docs.victoriametrics.com/victorialogs/cluster/#high-availability).
 * BUGFIX: [multi-level cluster setup](https://docs.victoriametrics.com/victorialogs/cluster/#multi-level-cluster-setup): properly return `502 Bad Gateway` HTTP response code when a `vlselect` node queries other `vlselect` nodes and the underlying `vlstorage` is unavailable, as described at [high availability](https://docs.victoriametrics.com/victorialogs/cluster/#high-availability) docs. This allows configuring proper failover schemes to a healthy cluster.
+* BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): prevent `vlinsert` from dropping buffered log data on graceful shutdown, e.g. during autoscaling scale-in or rolling restarts. Previously the final flush reused the already-canceled shutdown context, so in-flight data blocks failed immediately with `context canceled` and were dropped. `vlinsert` now sends every data block to `vlstorage` nodes with an independent `1m` timeout instead of the shutdown context, consistent with `-remoteWrite.sendTimeout` in `vlagent`, so the final flush during shutdown can actually complete. See [#1572](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1572).
 
 ## [v1.51.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.51.0)