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

app/vlstorage/main.go

@@ -71,6 +71,8 @@ var (
 	insertConcurrency        = flag.Int("insert.concurrency", 2, "The average number of concurrent data ingestion requests, which can be sent to every -storageNode")
 	insertDisableCompression = flag.Bool("insert.disableCompression", false, "Whether to disable compression when sending the ingested data to -storageNode nodes. "+
 		"Disabled compression reduces CPU usage at the cost of higher network usage")
+	insertDrainTimeout = flag.Duration("insert.drainTimeout", 5*time.Second, "The maximum duration for draining the in-memory buffered logs to -storageNode nodes on graceful shutdown. "+
+		"It must be smaller than the container termination grace period minus -http.shutdownDelay, otherwise the process may be killed before the drain completes and the buffered logs are lost")
 	selectDisableCompression = flag.Bool("select.disableCompression", false, "Whether to disable compression for select query responses received from -storageNode nodes. "+
 		"Disabled compression reduces CPU usage at the cost of higher network usage")
 
@@ -175,7 +177,7 @@ func initNetworkStorage() {
 	}
 
 	logger.Infof("starting insert service for nodes %s", *storageNodeAddrs)
-	netstorageInsert = netinsert.NewStorage(*storageNodeAddrs, authCfgs, isTLSs, *insertConcurrency, *insertDisableCompression)
+	netstorageInsert = netinsert.NewStorage(*storageNodeAddrs, authCfgs, isTLSs, *insertConcurrency, *insertDisableCompression, *insertDrainTimeout)
 
 	logger.Infof("initializing select service for nodes %s", *storageNodeAddrs)
 	netstorageSelect = netselect.NewStorage(*storageNodeAddrs, authCfgs, isTLSs, *selectDisableCompression)

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,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage"
 )
 
+// the maximum duration for sending a single data block to a storage node.
+const sendTimeout = time.Minute
+
 // the maximum size of a single data block sent to storage node.
 const maxInsertBlockSize = 2 * 1024 * 1024
 
@@ -38,12 +41,18 @@ type Storage struct {
 
 	disableCompression bool
 
+	drainTimeout time.Duration
+
 	srt *streamRowsTracker
 
 	pendingDataBuffers chan *bytesutil.ByteBuffer
 
 	stopCh chan struct{}
-	wg     sync.WaitGroup
+
+	sendCtx    context.Context
+	sendCancel context.CancelFunc
+
+	wg sync.WaitGroup
 }
 
 type storageNode struct {
@@ -216,7 +225,7 @@ func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer)
 
 		t := timerpool.Get(time.Second)
 		select {
-		case <-sn.s.stopCh:
+		case <-sn.s.sendCtx.Done():
 			timerpool.Put(t)
 			logger.Errorf("dropping %d bytes of data, since there are no available storage nodes", pendingData.Len())
 			return
@@ -257,7 +266,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(sn.s.sendCtx, sendTimeout)
 	defer cancel()
 
 	method := "GET"
@@ -324,17 +333,19 @@ var zstdBufPool bytesutil.ByteBufferPool
 // If disableCompression is set, then the data is sent uncompressed to the remote storage.
 //
 // Call MustStop on the returned storage when it is no longer needed.
-func NewStorage(addrs []string, authCfgs []*promauth.Config, isTLSs []bool, concurrency int, disableCompression bool) *Storage {
+func NewStorage(addrs []string, authCfgs []*promauth.Config, isTLSs []bool, concurrency int, disableCompression bool, drainTimeout time.Duration) *Storage {
 	pendingDataBuffers := make(chan *bytesutil.ByteBuffer, concurrency*len(addrs))
 	for range cap(pendingDataBuffers) {
 		pendingDataBuffers <- &bytesutil.ByteBuffer{}
 	}
 
 	s := &Storage{
 		disableCompression: disableCompression,
+		drainTimeout:       drainTimeout,
 		pendingDataBuffers: pendingDataBuffers,
 		stopCh:             make(chan struct{}),
 	}
+	s.sendCtx, s.sendCancel = context.WithCancel(context.Background())
 
 	sns := make([]*storageNode, len(addrs))
 	for i, addr := range addrs {
@@ -362,8 +373,17 @@ func (s *Storage) getActiveStreams() int {
 
 // MustStop stops the s.
 func (s *Storage) MustStop() {
+	// Drain the buffered data to storage nodes on shutdown, bounded by drainTimeout so an
+	// unresponsive storage node cannot block MustStop.
+	t := time.AfterFunc(s.drainTimeout, func() {
+		logger.Warnf("cannot drain the buffered data to -storageNode nodes within -insert.drainTimeout=%s on graceful shutdown; "+
+			"the remaining buffered data is dropped; consider increasing -insert.drainTimeout", s.drainTimeout)
+		s.sendCancel()
+	})
 	close(s.stopCh)
 	s.wg.Wait()
+	t.Stop()
+	s.sendCancel()
 	s.sns = nil
 }
 

app/vlstorage/netinsert/netinsert_test.go

@@ -4,11 +4,86 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"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
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/internal/insert" {
+			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, 5*time.Second)
+
+	// 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)
+
+	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")
+	}
+}
+
+func TestStorageDrainTimeoutOnUnresponsiveNode(t *testing.T) {
+	block := make(chan struct{})
+	ts := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
+		<-block
+	}))
+	defer ts.Close()
+	defer close(block)
+
+	ac, err := (&promauth.Options{}).NewConfig()
+	if err != nil {
+		t.Fatalf("cannot create auth config: %s", err)
+	}
+
+	addr := strings.TrimPrefix(ts.URL, "http://")
+	drainTimeout := 100 * time.Millisecond
+	s := NewStorage([]string{addr}, []*promauth.Config{ac}, []bool{false}, 1, true, drainTimeout)
+	s.AddRow(0, &logstorage.InsertRow{
+		Timestamp: 1,
+		Fields:    []logstorage.Field{{Name: "foo", Value: "bar"}},
+	})
+
+	done := make(chan struct{})
+	go func() {
+		s.MustStop()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatalf("MustStop is blocked on the unresponsive storage node; it must return within drainTimeout=%s", drainTimeout)
+	}
+}
+
 func TestStreamRowsTracker(t *testing.T) {
 	f := func(rowsCount, streamsCount, nodesCount int) {
 		t.Helper()

docs/victorialogs/CHANGELOG.md

@@ -22,6 +22,7 @@ according to the following docs:
 
 ## tip
 
+* BUGFIX: [vlinsert](https://docs.victoriametrics.com/victorialogs/cluster/): now drains buffered logs to `vlstorage` nodes on graceful shutdown instead of dropping them, bounded by the new `-insert.drainTimeout` command-line flag (default `5s`). See [#1572](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1572).
 * BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): evenly spread rerouted data across available `vlstorage` nodes. Previously, healthy nodes adjacent to unavailable nodes in the `-storageNode` list could receive much more data, resulting in uneven resource usage. See [#1548](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1548).
 * BUGFIX: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/) and [querying](https://docs.victoriametrics.com/victorialogs/querying/): properly handle logs containing duplicate [stream field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) names. Previously, [v1.52.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.52.0) could panic when ingesting such logs in single-node VictoriaLogs, drop them during ingestion in VictoriaLogs cluster, or panic when querying such data written by earlier releases. See [#1603](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1603) and [#1604](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1604).