From 6eec825533abb6a330caa68115a7f9898bfb21d8 Mon Sep 17 00:00:00 2001 From: Vincent Miszczak Date: Fri, 3 Jul 2026 14:34:14 +0200 Subject: [PATCH 1/4] app/vlstorage/netinsert: drain buffered data on graceful shutdown instead of dropping it vlinsert dropped in-flight buffered log blocks whenever a pod terminated (autoscaling scale-in or rolling restarts): the shutdown flush reused the already-canceled stopCh context, so requests failed immediately with "context canceled" and the data was dropped. Thread an explicit context through the send path so the shutdown flush uses a fresh, bounded context. Buffered data is now drained to storage nodes on shutdown, bounded by the new -insert.drainTimeout flag (default 10s). --- app/vlstorage/netinsert/netinsert.go | 55 ++++++++++++------- app/vlstorage/netinsert/netinsert_test.go | 54 ++++++++++++++++++ docs/victorialogs/CHANGELOG.md | 1 + .../victoria_logs_common_flags.md | 2 + docs/victorialogs/vlagent_common_flags.md | 2 + 5 files changed, 94 insertions(+), 20 deletions(-) diff --git a/app/vlstorage/netinsert/netinsert.go b/app/vlstorage/netinsert/netinsert.go index c79f5c02f5..b327186428 100644 --- a/app/vlstorage/netinsert/netinsert.go +++ b/app/vlstorage/netinsert/netinsert.go @@ -1,7 +1,9 @@ package netinsert import ( + "context" "errors" + "flag" "fmt" "io" "net/http" @@ -24,6 +26,8 @@ import ( "github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage" ) +var drainTimeout = flag.Duration("insert.drainTimeout", 10*time.Second, "The maximum duration for draining in-memory buffered data to storage nodes during graceful shutdown before it is dropped") + // the maximum size of a single data block sent to storage node. const maxInsertBlockSize = 2 * 1024 * 1024 @@ -43,7 +47,14 @@ type Storage struct { pendingDataBuffers chan *bytesutil.ByteBuffer stopCh chan struct{} - wg sync.WaitGroup + + // reqCtx is the context used for regular data ingestion requests. + // It is canceled when stopCh is closed. The final shutdown flush uses a + // separate bounded context instead, so buffered data can still be drained. + reqCtx context.Context + reqCancel context.CancelFunc + + wg sync.WaitGroup } type storageNode struct { @@ -127,15 +138,20 @@ func (sn *storageNode) backgroundFlusher() { for { select { case <-sn.s.stopCh: - sn.flushPendingData(true) + // sn.s.reqCtx is already canceled by the closed stopCh, so use a fresh + // bounded context to drain buffered data to storage nodes during graceful + // shutdown instead of dropping it immediately. + ctx, cancel := context.WithTimeout(context.Background(), *drainTimeout) + sn.flushPendingData(ctx, true) + cancel() return case <-t.C: - sn.flushPendingData(false) + sn.flushPendingData(sn.s.reqCtx, false) } } } -func (sn *storageNode) flushPendingData(force bool) { +func (sn *storageNode) flushPendingData(ctx context.Context, force bool) { sn.pendingDataMu.Lock() if !force && time.Since(sn.pendingDataLastFlush) < time.Second { // nothing to flush @@ -146,15 +162,15 @@ func (sn *storageNode) flushPendingData(force bool) { pendingData := sn.grabPendingDataForFlushLocked() sn.pendingDataMu.Unlock() - sn.mustSendInsertRequest(pendingData) + sn.mustSendInsertRequest(ctx, pendingData) } func (sn *storageNode) debugFlush() { // Send pending samples to sn. - sn.flushPendingData(true) + sn.flushPendingData(sn.s.reqCtx, true) // Instruct sn to convert the received samples into searchable parts. - if err := sn.doRequest("/internal/force_flush", nil); err != nil { + if err := sn.doRequest(sn.s.reqCtx, "/internal/force_flush", nil); err != nil { logger.Errorf("cannot convert pending samples into searchable parts: %s", err) } } @@ -183,7 +199,7 @@ func (sn *storageNode) addRow(r *logstorage.InsertRow) { bbPool.Put(bb) if pendingData != nil { - sn.mustSendInsertRequest(pendingData) + sn.mustSendInsertRequest(sn.s.reqCtx, pendingData) } } @@ -197,13 +213,13 @@ func (sn *storageNode) grabPendingDataForFlushLocked() *bytesutil.ByteBuffer { return pendingData } -func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer) { +func (sn *storageNode) mustSendInsertRequest(ctx context.Context, pendingData *bytesutil.ByteBuffer) { defer func() { pendingData.Reset() sn.s.pendingDataBuffers <- pendingData }() - err := sn.sendInsertRequest(pendingData) + err := sn.sendInsertRequest(ctx, pendingData) if err == nil { return } @@ -211,12 +227,12 @@ func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer) if !errors.Is(err, errTemporarilyDisabled) { logger.Warnf("%s; re-routing the data block to the remaining nodes", err) } - for !sn.s.sendInsertRequestToAnyNode(pendingData) { + for !sn.s.sendInsertRequestToAnyNode(ctx, pendingData) { logger.Errorf("cannot send pending data to storage nodes, since all of them are unavailable; re-trying to send the data in a second") t := timerpool.Get(time.Second) select { - case <-sn.s.stopCh: + case <-ctx.Done(): timerpool.Put(t) logger.Errorf("dropping %d bytes of data, since there are no available storage nodes", pendingData.Len()) return @@ -226,7 +242,7 @@ func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer) } } -func (sn *storageNode) sendInsertRequest(pendingData *bytesutil.ByteBuffer) error { +func (sn *storageNode) sendInsertRequest(ctx context.Context, pendingData *bytesutil.ByteBuffer) error { dataLen := pendingData.Len() if dataLen == 0 { // Nothing to send. @@ -249,17 +265,14 @@ func (sn *storageNode) sendInsertRequest(pendingData *bytesutil.ByteBuffer) erro body = pendingData.NewReader() } - if err := sn.doRequest("/internal/insert", body); err != nil { + if err := sn.doRequest(ctx, "/internal/insert", body); err != nil { return fmt.Errorf("cannot send data block with the length %d: %w", pendingData.Len(), err) } return nil } -func (sn *storageNode) doRequest(path string, body io.Reader) error { - ctx, cancel := contextutil.NewStopChanContext(sn.s.stopCh) - defer cancel() - +func (sn *storageNode) doRequest(ctx context.Context, path string, body io.Reader) error { method := "GET" if body != nil { method = "POST" @@ -335,6 +348,7 @@ func NewStorage(addrs []string, authCfgs []*promauth.Config, isTLSs []bool, conc pendingDataBuffers: pendingDataBuffers, stopCh: make(chan struct{}), } + s.reqCtx, s.reqCancel = contextutil.NewStopChanContext(s.stopCh) sns := make([]*storageNode, len(addrs)) for i, addr := range addrs { @@ -364,6 +378,7 @@ func (s *Storage) getActiveStreams() int { func (s *Storage) MustStop() { close(s.stopCh) s.wg.Wait() + s.reqCancel() s.sns = nil } @@ -383,12 +398,12 @@ func (s *Storage) AddRow(streamHash uint64, r *logstorage.InsertRow) { sn.addRow(r) } -func (s *Storage) sendInsertRequestToAnyNode(pendingData *bytesutil.ByteBuffer) bool { +func (s *Storage) sendInsertRequestToAnyNode(ctx context.Context, pendingData *bytesutil.ByteBuffer) bool { startIdx := int(fastrand.Uint32n(uint32(len(s.sns)))) for i := range s.sns { idx := (startIdx + i) % len(s.sns) sn := s.sns[idx] - err := sn.sendInsertRequest(pendingData) + err := sn.sendInsertRequest(ctx, pendingData) if err == nil { return true } diff --git a/app/vlstorage/netinsert/netinsert_test.go b/app/vlstorage/netinsert/netinsert_test.go index 45d83484f2..fdbf71f7b1 100644 --- a/app/vlstorage/netinsert/netinsert_test.go +++ b/app/vlstorage/netinsert/netinsert_test.go @@ -2,13 +2,67 @@ package netinsert import ( "fmt" + "io" "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) { + oldTimeout := *drainTimeout + *drainTimeout = 2 * time.Second + defer func() { *drainTimeout = oldTimeout }() + + 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() diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index 5cfffe1d15..6f0e9a64aa 100644 --- a/docs/victorialogs/CHANGELOG.md +++ b/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 drains buffered data to `vlstorage` nodes during shutdown, bounded by the new `-insert.drainTimeout` command-line flag (default `10s`). See [#1572](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1572). ## [v1.51.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.51.0) diff --git a/docs/victorialogs/victoria_logs_common_flags.md b/docs/victorialogs/victoria_logs_common_flags.md index ce3b1b3f5c..2de97c2043 100644 --- a/docs/victorialogs/victoria_logs_common_flags.md +++ b/docs/victorialogs/victoria_logs_common_flags.md @@ -109,6 +109,8 @@ See the docs at https://docs.victoriametrics.com/victorialogs/ Whether to disable both /insert/* and /internal/insert HTTP endpoints. Useful for dedicated vlselect nodes. See also -internalinsert.disable. See https://docs.victoriametrics.com/victorialogs/cluster/#security -insert.disableCompression Whether to disable compression when sending the ingested data to -storageNode nodes. Disabled compression reduces CPU usage at the cost of higher network usage + -insert.drainTimeout duration + The maximum duration for draining in-memory buffered data to storage nodes during graceful shutdown before it is dropped (default 10s) -insert.maxFieldsPerLine int The maximum number of log fields per line, which can be read by /insert/* handlers; see https://docs.victoriametrics.com/victorialogs/faq/#how-many-fields-a-single-log-entry-may-contain (default 1000) -insert.maxLineSizeBytes size diff --git a/docs/victorialogs/vlagent_common_flags.md b/docs/victorialogs/vlagent_common_flags.md index 09ded3751b..ed0b27c8d1 100644 --- a/docs/victorialogs/vlagent_common_flags.md +++ b/docs/victorialogs/vlagent_common_flags.md @@ -132,6 +132,8 @@ See the docs at https://docs.victoriametrics.com/victorialogs/vlagent/ . Empty values are set to false. -insert.disable Whether to disable both /insert/* and /internal/insert HTTP endpoints. Useful for dedicated vlselect nodes. See also -internalinsert.disable. See https://docs.victoriametrics.com/victorialogs/cluster/#security + -insert.drainTimeout duration + The maximum duration for draining in-memory buffered data to storage nodes during graceful shutdown before it is dropped (default 10s) -insert.maxFieldsPerLine int The maximum number of log fields per line, which can be read by /insert/* handlers; see https://docs.victoriametrics.com/victorialogs/faq/#how-many-fields-a-single-log-entry-may-contain (default 1000) -insert.maxLineSizeBytes size From 8287d84c9ceeda43ae4faf597b6ddd4e9639ef6d Mon Sep 17 00:00:00 2001 From: Vincent Miszczak Date: Fri, 3 Jul 2026 17:47:57 +0200 Subject: [PATCH 2/4] app/vlstorage/netinsert: bound send requests by a fixed timeout instead of the shutdown context The final flush during graceful shutdown reused the already-canceled shutdown context, so the last buffered data block failed instantly with "context canceled" and was dropped. Send every data block with its own timeout instead, consistent with -remoteWrite.sendTimeout in vlagent. --- app/vlstorage/netinsert/netinsert.go | 58 ++++++++----------- app/vlstorage/netinsert/netinsert_test.go | 5 -- docs/victorialogs/CHANGELOG.md | 2 +- .../victoria_logs_common_flags.md | 2 - docs/victorialogs/vlagent_common_flags.md | 2 - 5 files changed, 25 insertions(+), 44 deletions(-) diff --git a/app/vlstorage/netinsert/netinsert.go b/app/vlstorage/netinsert/netinsert.go index b327186428..7fb2623aba 100644 --- a/app/vlstorage/netinsert/netinsert.go +++ b/app/vlstorage/netinsert/netinsert.go @@ -3,7 +3,6 @@ package netinsert import ( "context" "errors" - "flag" "fmt" "io" "net/http" @@ -13,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" @@ -26,7 +24,10 @@ import ( "github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage" ) -var drainTimeout = flag.Duration("insert.drainTimeout", 10*time.Second, "The maximum duration for draining in-memory buffered data to storage nodes during graceful shutdown before it is dropped") +// 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 @@ -47,14 +48,7 @@ type Storage struct { pendingDataBuffers chan *bytesutil.ByteBuffer stopCh chan struct{} - - // reqCtx is the context used for regular data ingestion requests. - // It is canceled when stopCh is closed. The final shutdown flush uses a - // separate bounded context instead, so buffered data can still be drained. - reqCtx context.Context - reqCancel context.CancelFunc - - wg sync.WaitGroup + wg sync.WaitGroup } type storageNode struct { @@ -138,20 +132,15 @@ func (sn *storageNode) backgroundFlusher() { for { select { case <-sn.s.stopCh: - // sn.s.reqCtx is already canceled by the closed stopCh, so use a fresh - // bounded context to drain buffered data to storage nodes during graceful - // shutdown instead of dropping it immediately. - ctx, cancel := context.WithTimeout(context.Background(), *drainTimeout) - sn.flushPendingData(ctx, true) - cancel() + sn.flushPendingData(true) return case <-t.C: - sn.flushPendingData(sn.s.reqCtx, false) + sn.flushPendingData(false) } } } -func (sn *storageNode) flushPendingData(ctx context.Context, force bool) { +func (sn *storageNode) flushPendingData(force bool) { sn.pendingDataMu.Lock() if !force && time.Since(sn.pendingDataLastFlush) < time.Second { // nothing to flush @@ -162,15 +151,15 @@ func (sn *storageNode) flushPendingData(ctx context.Context, force bool) { pendingData := sn.grabPendingDataForFlushLocked() sn.pendingDataMu.Unlock() - sn.mustSendInsertRequest(ctx, pendingData) + sn.mustSendInsertRequest(pendingData) } func (sn *storageNode) debugFlush() { // Send pending samples to sn. - sn.flushPendingData(sn.s.reqCtx, true) + sn.flushPendingData(true) // Instruct sn to convert the received samples into searchable parts. - if err := sn.doRequest(sn.s.reqCtx, "/internal/force_flush", nil); err != nil { + if err := sn.doRequest("/internal/force_flush", nil); err != nil { logger.Errorf("cannot convert pending samples into searchable parts: %s", err) } } @@ -199,7 +188,7 @@ func (sn *storageNode) addRow(r *logstorage.InsertRow) { bbPool.Put(bb) if pendingData != nil { - sn.mustSendInsertRequest(sn.s.reqCtx, pendingData) + sn.mustSendInsertRequest(pendingData) } } @@ -213,13 +202,13 @@ func (sn *storageNode) grabPendingDataForFlushLocked() *bytesutil.ByteBuffer { return pendingData } -func (sn *storageNode) mustSendInsertRequest(ctx context.Context, pendingData *bytesutil.ByteBuffer) { +func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer) { defer func() { pendingData.Reset() sn.s.pendingDataBuffers <- pendingData }() - err := sn.sendInsertRequest(ctx, pendingData) + err := sn.sendInsertRequest(pendingData) if err == nil { return } @@ -227,12 +216,12 @@ func (sn *storageNode) mustSendInsertRequest(ctx context.Context, pendingData *b if !errors.Is(err, errTemporarilyDisabled) { logger.Warnf("%s; re-routing the data block to the remaining nodes", err) } - for !sn.s.sendInsertRequestToAnyNode(ctx, pendingData) { + for !sn.s.sendInsertRequestToAnyNode(pendingData) { logger.Errorf("cannot send pending data to storage nodes, since all of them are unavailable; re-trying to send the data in a second") t := timerpool.Get(time.Second) select { - case <-ctx.Done(): + case <-sn.s.stopCh: timerpool.Put(t) logger.Errorf("dropping %d bytes of data, since there are no available storage nodes", pendingData.Len()) return @@ -242,7 +231,7 @@ func (sn *storageNode) mustSendInsertRequest(ctx context.Context, pendingData *b } } -func (sn *storageNode) sendInsertRequest(ctx context.Context, pendingData *bytesutil.ByteBuffer) error { +func (sn *storageNode) sendInsertRequest(pendingData *bytesutil.ByteBuffer) error { dataLen := pendingData.Len() if dataLen == 0 { // Nothing to send. @@ -265,14 +254,17 @@ func (sn *storageNode) sendInsertRequest(ctx context.Context, pendingData *bytes body = pendingData.NewReader() } - if err := sn.doRequest(ctx, "/internal/insert", body); err != nil { + if err := sn.doRequest("/internal/insert", body); err != nil { return fmt.Errorf("cannot send data block with the length %d: %w", pendingData.Len(), err) } return nil } -func (sn *storageNode) doRequest(ctx context.Context, path string, body io.Reader) error { +func (sn *storageNode) doRequest(path string, body io.Reader) error { + ctx, cancel := context.WithTimeout(context.Background(), sendTimeout) + defer cancel() + method := "GET" if body != nil { method = "POST" @@ -348,7 +340,6 @@ func NewStorage(addrs []string, authCfgs []*promauth.Config, isTLSs []bool, conc pendingDataBuffers: pendingDataBuffers, stopCh: make(chan struct{}), } - s.reqCtx, s.reqCancel = contextutil.NewStopChanContext(s.stopCh) sns := make([]*storageNode, len(addrs)) for i, addr := range addrs { @@ -378,7 +369,6 @@ func (s *Storage) getActiveStreams() int { func (s *Storage) MustStop() { close(s.stopCh) s.wg.Wait() - s.reqCancel() s.sns = nil } @@ -398,12 +388,12 @@ func (s *Storage) AddRow(streamHash uint64, r *logstorage.InsertRow) { sn.addRow(r) } -func (s *Storage) sendInsertRequestToAnyNode(ctx context.Context, pendingData *bytesutil.ByteBuffer) bool { +func (s *Storage) sendInsertRequestToAnyNode(pendingData *bytesutil.ByteBuffer) bool { startIdx := int(fastrand.Uint32n(uint32(len(s.sns)))) for i := range s.sns { idx := (startIdx + i) % len(s.sns) sn := s.sns[idx] - err := sn.sendInsertRequest(ctx, pendingData) + err := sn.sendInsertRequest(pendingData) if err == nil { return true } diff --git a/app/vlstorage/netinsert/netinsert_test.go b/app/vlstorage/netinsert/netinsert_test.go index fdbf71f7b1..05ca1d3f6d 100644 --- a/app/vlstorage/netinsert/netinsert_test.go +++ b/app/vlstorage/netinsert/netinsert_test.go @@ -10,7 +10,6 @@ import ( "strings" "sync/atomic" "testing" - "time" "github.com/cespare/xxhash/v2" @@ -20,10 +19,6 @@ import ( ) func TestStorageDrainsPendingDataOnShutdown(t *testing.T) { - oldTimeout := *drainTimeout - *drainTimeout = 2 * time.Second - defer func() { *drainTimeout = oldTimeout }() - var insertRequests atomic.Int64 var receivedBytes atomic.Int64 ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index 6f0e9a64aa..658c90ef37 100644 --- a/docs/victorialogs/CHANGELOG.md +++ b/docs/victorialogs/CHANGELOG.md @@ -38,7 +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 drains buffered data to `vlstorage` nodes during shutdown, bounded by the new `-insert.drainTimeout` command-line flag (default `10s`). See [#1572](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1572). +* 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) diff --git a/docs/victorialogs/victoria_logs_common_flags.md b/docs/victorialogs/victoria_logs_common_flags.md index 2de97c2043..ce3b1b3f5c 100644 --- a/docs/victorialogs/victoria_logs_common_flags.md +++ b/docs/victorialogs/victoria_logs_common_flags.md @@ -109,8 +109,6 @@ See the docs at https://docs.victoriametrics.com/victorialogs/ Whether to disable both /insert/* and /internal/insert HTTP endpoints. Useful for dedicated vlselect nodes. See also -internalinsert.disable. See https://docs.victoriametrics.com/victorialogs/cluster/#security -insert.disableCompression Whether to disable compression when sending the ingested data to -storageNode nodes. Disabled compression reduces CPU usage at the cost of higher network usage - -insert.drainTimeout duration - The maximum duration for draining in-memory buffered data to storage nodes during graceful shutdown before it is dropped (default 10s) -insert.maxFieldsPerLine int The maximum number of log fields per line, which can be read by /insert/* handlers; see https://docs.victoriametrics.com/victorialogs/faq/#how-many-fields-a-single-log-entry-may-contain (default 1000) -insert.maxLineSizeBytes size diff --git a/docs/victorialogs/vlagent_common_flags.md b/docs/victorialogs/vlagent_common_flags.md index ed0b27c8d1..09ded3751b 100644 --- a/docs/victorialogs/vlagent_common_flags.md +++ b/docs/victorialogs/vlagent_common_flags.md @@ -132,8 +132,6 @@ See the docs at https://docs.victoriametrics.com/victorialogs/vlagent/ . Empty values are set to false. -insert.disable Whether to disable both /insert/* and /internal/insert HTTP endpoints. Useful for dedicated vlselect nodes. See also -internalinsert.disable. See https://docs.victoriametrics.com/victorialogs/cluster/#security - -insert.drainTimeout duration - The maximum duration for draining in-memory buffered data to storage nodes during graceful shutdown before it is dropped (default 10s) -insert.maxFieldsPerLine int The maximum number of log fields per line, which can be read by /insert/* handlers; see https://docs.victoriametrics.com/victorialogs/faq/#how-many-fields-a-single-log-entry-may-contain (default 1000) -insert.maxLineSizeBytes size From 34cbcba64c4263c8a79e887ff90f48442c79a691 Mon Sep 17 00:00:00 2001 From: Cuong Le Date: Sat, 18 Jul 2026 20:18:26 +0700 Subject: [PATCH 3/4] app/vlstorage/netinsert: bound graceful shutdown drain with -insert.drainTimeout --- app/vlstorage/main.go | 4 ++- app/vlstorage/netinsert/netinsert.go | 25 +++++++++++--- app/vlstorage/netinsert/netinsert_test.go | 42 ++++++++++++++++++----- docs/victorialogs/CHANGELOG.md | 2 +- 4 files changed, 58 insertions(+), 15 deletions(-) diff --git a/app/vlstorage/main.go b/app/vlstorage/main.go index bce1b14577..459632b4d7 100644 --- a/app/vlstorage/main.go +++ b/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) diff --git a/app/vlstorage/netinsert/netinsert.go b/app/vlstorage/netinsert/netinsert.go index 7fb2623aba..5520b58066 100644 --- a/app/vlstorage/netinsert/netinsert.go +++ b/app/vlstorage/netinsert/netinsert.go @@ -25,8 +25,6 @@ import ( ) // 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. @@ -43,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 { @@ -262,7 +266,7 @@ func (sn *storageNode) sendInsertRequest(pendingData *bytesutil.ByteBuffer) erro } func (sn *storageNode) doRequest(path string, body io.Reader) error { - ctx, cancel := context.WithTimeout(context.Background(), sendTimeout) + ctx, cancel := context.WithTimeout(sn.s.sendCtx, sendTimeout) defer cancel() method := "GET" @@ -329,7 +333,7 @@ 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{} @@ -337,9 +341,11 @@ func NewStorage(addrs []string, authCfgs []*promauth.Config, isTLSs []bool, conc 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 { @@ -367,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 } diff --git a/app/vlstorage/netinsert/netinsert_test.go b/app/vlstorage/netinsert/netinsert_test.go index 05ca1d3f6d..8ef3581eb5 100644 --- a/app/vlstorage/netinsert/netinsert_test.go +++ b/app/vlstorage/netinsert/netinsert_test.go @@ -2,7 +2,6 @@ package netinsert import ( "fmt" - "io" "math" "math/rand" "net/http" @@ -10,6 +9,7 @@ import ( "strings" "sync/atomic" "testing" + "time" "github.com/cespare/xxhash/v2" @@ -20,11 +20,8 @@ import ( 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) @@ -37,7 +34,7 @@ func TestStorageDrainsPendingDataOnShutdown(t *testing.T) { } addr := strings.TrimPrefix(ts.URL, "http://") - s := NewStorage([]string{addr}, []*promauth.Config{ac}, []bool{false}, 1, true) + 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. @@ -47,14 +44,43 @@ func TestStorageDrainsPendingDataOnShutdown(t *testing.T) { } 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 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) } } diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index 658c90ef37..9a647d2f2e 100644 --- a/docs/victorialogs/CHANGELOG.md +++ b/docs/victorialogs/CHANGELOG.md @@ -38,7 +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). +* BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): `vlinsert` no longer loses buffered logs on graceful shutdown, e.g. during autoscaling scale-in or rolling restarts. It now sends the buffered logs to `vlstorage` nodes before exiting, within the time limit set by the new `-insert.drainTimeout` command-line flag (default `5s`), so a slow or unresponsive `vlstorage` node cannot delay the shutdown beyond the container termination grace period. See [#1572](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1572). ## [v1.51.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.51.0) From 35fccc3696affeb2e8498983a96270788771a0ff Mon Sep 17 00:00:00 2001 From: Cuong Le Date: Thu, 23 Jul 2026 10:55:43 +0700 Subject: [PATCH 4/4] app/vlstorage/netinsert: keep re-trying to send pending data within -insert.drainTimeout on shutdown --- app/vlstorage/netinsert/netinsert.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/vlstorage/netinsert/netinsert.go b/app/vlstorage/netinsert/netinsert.go index 2f168f35b0..261517d9ae 100644 --- a/app/vlstorage/netinsert/netinsert.go +++ b/app/vlstorage/netinsert/netinsert.go @@ -225,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