diff --git a/app/vlstorage/main.go b/app/vlstorage/main.go index 826366f4ff..b6b5927112 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 9bd2235c0b..261517d9ae 100644 --- a/app/vlstorage/netinsert/netinsert.go +++ b/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,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{} @@ -332,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 { @@ -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 } diff --git a/app/vlstorage/netinsert/netinsert_test.go b/app/vlstorage/netinsert/netinsert_test.go index 45d83484f2..8ef3581eb5 100644 --- a/app/vlstorage/netinsert/netinsert_test.go +++ b/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() diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index 7eed66005c..7651b5ae79 100644 --- a/docs/victorialogs/CHANGELOG.md +++ b/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). ## [v1.52.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.52.0)