diff --git a/app/vlstorage/main.go b/app/vlstorage/main.go index 826366f4ff..1dd8ebdcb3 100644 --- a/app/vlstorage/main.go +++ b/app/vlstorage/main.go @@ -68,7 +68,7 @@ var ( storageNodeAddrs = flagutil.NewArrayString("storageNode", "Comma-separated list of TCP addresses for storage nodes to route the ingested logs to and to send select queries to. "+ "If the list is empty, then the ingested logs are stored and queried locally from -storageDataPath") - insertConcurrency = flag.Int("insert.concurrency", 2, "The average number of concurrent data ingestion requests, which can be sent to every -storageNode") + insertConcurrency = flag.Int("insert.concurrency", 2, "The maximum number of concurrent data ingestion requests, which can be sent to every -storageNode; requests above this limit are re-routed to other -storageNode nodes") 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") selectDisableCompression = flag.Bool("select.disableCompression", false, "Whether to disable compression for select query responses received from -storageNode nodes. "+ diff --git a/app/vlstorage/netinsert/netinsert.go b/app/vlstorage/netinsert/netinsert.go index 9bd2235c0b..cbd7efd571 100644 --- a/app/vlstorage/netinsert/netinsert.go +++ b/app/vlstorage/netinsert/netinsert.go @@ -40,8 +40,6 @@ type Storage struct { srt *streamRowsTracker - pendingDataBuffers chan *bytesutil.ByteBuffer - stopCh chan struct{} wg sync.WaitGroup } @@ -75,9 +73,20 @@ type storageNode struct { // isReachable is set to true if the given storageNode is available for data writing. isReachable atomic.Bool + + // concurrencyCh limits the number of concurrent in-flight insert requests to addr. + // When it is full, the data is re-routed to less busy storage nodes. + concurrencyCh chan struct{} + + // concurrencyLimitReached counts insert requests rejected because concurrencyCh is full. + concurrencyLimitReached *metrics.Counter + + // concurrencyLimitLogger throttles warnings emitted when the data is re-routed because concurrencyCh is full, + // since such warnings can be emitted every second while the storage node at addr remains slow. + concurrencyLimitLogger *logger.LogThrottler } -func newStorageNode(s *Storage, addr string, ac *promauth.Config, isTLS bool) *storageNode { +func newStorageNode(s *Storage, addr string, ac *promauth.Config, isTLS bool, concurrency int) *storageNode { tr := httputil.NewTransport(false, "vlinsert_backend") tr.TLSHandshakeTimeout = 20 * time.Second tr.DisableCompression = true @@ -104,6 +113,11 @@ func newStorageNode(s *Storage, addr string, ac *promauth.Config, isTLS bool) *s sendErrors: metrics.GetOrCreateCounter(fmt.Sprintf(`vl_insert_remote_send_errors_total{addr=%q}`, addr)), pendingData: &bytesutil.ByteBuffer{}, + + concurrencyCh: make(chan struct{}, concurrency), + + concurrencyLimitReached: metrics.GetOrCreateCounter(fmt.Sprintf(`vl_insert_remote_concurrency_limit_reached_total{addr=%q}`, addr)), + concurrencyLimitLogger: logger.WithThrottler("concurrencyLimitReached-"+addr, 5*time.Second), } sn.isReachable.Store(true) @@ -192,15 +206,16 @@ var bbPool bytesutil.ByteBufferPool func (sn *storageNode) grabPendingDataForFlushLocked() *bytesutil.ByteBuffer { sn.pendingDataLastFlush = time.Now() pendingData := sn.pendingData - sn.pendingData = <-sn.s.pendingDataBuffers + sn.pendingData = pendingDataPool.Get() return pendingData } +var pendingDataPool bytesutil.ByteBufferPool + func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer) { defer func() { - pendingData.Reset() - sn.s.pendingDataBuffers <- pendingData + pendingDataPool.Put(pendingData) }() err := sn.sendInsertRequest(pendingData) @@ -208,17 +223,19 @@ func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer) return } - if !errors.Is(err, errTemporarilyDisabled) { + if errors.Is(err, errConcurrencyLimitReached) { + sn.concurrencyLimitLogger.Warnf("cannot send data block to the storage node %q, since it has too many in-flight insert requests; re-routing the data block to the remaining nodes", sn.addr) + } else if !errors.Is(err, errTemporarilyDisabled) { logger.Warnf("%s; re-routing the data block to the remaining nodes", err) } 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") + logger.Errorf("cannot send pending data to storage nodes, since all of them are unavailable or overloaded; re-trying to send the data in a second") t := timerpool.Get(time.Second) select { case <-sn.s.stopCh: timerpool.Put(t) - logger.Errorf("dropping %d bytes of data, since there are no available storage nodes", pendingData.Len()) + logger.Errorf("dropping %d bytes of data, since all storage nodes are unavailable or overloaded", pendingData.Len()) return case <-t.C: timerpool.Put(t) @@ -238,6 +255,16 @@ func (sn *storageNode) sendInsertRequest(pendingData *bytesutil.ByteBuffer) erro return errTemporarilyDisabled } + select { + case sn.concurrencyCh <- struct{}{}: + default: + sn.concurrencyLimitReached.Inc() + return errConcurrencyLimitReached + } + defer func() { + <-sn.concurrencyCh + }() + var body io.Reader if !sn.s.disableCompression { bb := zstdBufPool.Get() @@ -319,26 +346,20 @@ var zstdBufPool bytesutil.ByteBufferPool // NewStorage returns new Storage for the given addrs with the given authCfgs. // -// The concurrency is the average number of concurrent connections per every addr. +// The concurrency limits the number of concurrent insert requests sent to every addr. // // 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 { - pendingDataBuffers := make(chan *bytesutil.ByteBuffer, concurrency*len(addrs)) - for range cap(pendingDataBuffers) { - pendingDataBuffers <- &bytesutil.ByteBuffer{} - } - s := &Storage{ disableCompression: disableCompression, - pendingDataBuffers: pendingDataBuffers, stopCh: make(chan struct{}), } sns := make([]*storageNode, len(addrs)) for i, addr := range addrs { - sns[i] = newStorageNode(s, addr, authCfgs[i], isTLSs[i]) + sns[i] = newStorageNode(s, addr, authCfgs[i], isTLSs[i], concurrency) } s.sns = sns @@ -383,13 +404,13 @@ func (s *Storage) AddRow(streamHash uint64, r *logstorage.InsertRow) { sn.addRow(r) } -// sendInsertRequestToAnyNode controls the rerouting logic when storage node is unavailable. +// sendInsertRequestToAnyNode controls the rerouting logic when storage node is unavailable or overloaded. func (s *Storage) sendInsertRequestToAnyNode(pendingData *bytesutil.ByteBuffer) bool { // collect available storage node indexes availableIdx := make([]int, 0, len(s.sns)) currentTime := fasttime.UnixTimestamp() for idx, sn := range s.sns { - if sn.disabledUntil.Load() <= currentTime { + if sn.disabledUntil.Load() <= currentTime && len(sn.concurrencyCh) < cap(sn.concurrencyCh) { availableIdx = append(availableIdx, idx) } } @@ -403,7 +424,7 @@ func (s *Storage) sendInsertRequestToAnyNode(pendingData *bytesutil.ByteBuffer) if err == nil { return true } - if !errors.Is(err, errTemporarilyDisabled) { + if !errors.Is(err, errTemporarilyDisabled) && !errors.Is(err, errConcurrencyLimitReached) { logger.Warnf("cannot send pending data to the storage node %q: %s; trying to send it to another storage node", sn.addr, err) } } @@ -412,6 +433,8 @@ func (s *Storage) sendInsertRequestToAnyNode(pendingData *bytesutil.ByteBuffer) var errTemporarilyDisabled = fmt.Errorf("writing to the node is temporarily disabled") +var errConcurrencyLimitReached = fmt.Errorf("the node has too many in-flight insert requests") + type streamRowsTracker struct { mu sync.Mutex diff --git a/app/vlstorage/netinsert/netinsert_test.go b/app/vlstorage/netinsert/netinsert_test.go index 45d83484f2..bc2ae23372 100644 --- a/app/vlstorage/netinsert/netinsert_test.go +++ b/app/vlstorage/netinsert/netinsert_test.go @@ -1,11 +1,18 @@ package netinsert import ( + "errors" "fmt" "math" "math/rand" + "net/http" + "net/http/httptest" + "sync" "testing" + "time" + "github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil" + "github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth" "github.com/cespare/xxhash/v2" ) @@ -55,3 +62,64 @@ func TestStreamRowsTracker(t *testing.T) { nodesCount = 9 f(rowsCount, streamsCount, nodesCount) } + +func TestStorageReroutesDataFromUnresponsiveNode(t *testing.T) { + // The unresponsive node accepts the request, then blocks until the test releases it. + releaseCh := make(chan struct{}) + reachedCh := make(chan struct{}) + unresponsive := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { + reachedCh <- struct{}{} + <-releaseCh + })) + defer unresponsive.Close() + + // The healthy node responds immediately. + healthy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer healthy.Close() + + addrs := []string{unresponsive.Listener.Addr().String(), healthy.Listener.Addr().String()} + authCfgs := make([]*promauth.Config, len(addrs)) + for i := range authCfgs { + ac, err := (&promauth.Options{}).NewConfig() + if err != nil { + t.Fatalf("cannot create auth config: %s", err) + } + authCfgs[i] = ac + } + + s := NewStorage(addrs, authCfgs, []bool{false, false}, 2, true) + defer s.MustStop() + + data := &bytesutil.ByteBuffer{} + data.MustWrite([]byte("log data block")) + + // Saturate the unresponsive node by filling all its in-flight slots with blocked sends. + snUnresponsive := s.sns[0] + var wg sync.WaitGroup + defer func() { + close(releaseCh) + wg.Wait() + }() + for range cap(snUnresponsive.concurrencyCh) { + wg.Go(func() { + _ = snUnresponsive.sendInsertRequest(data) + }) + select { + case <-reachedCh: + case <-time.After(2 * time.Second): + t.Fatalf("timed out waiting for the unresponsive node to become saturated") + } + } + + // A further request to the saturated node must be rejected, not block. + if err := snUnresponsive.sendInsertRequest(data); !errors.Is(err, errConcurrencyLimitReached) { + t.Fatalf("unexpected error when sending to the saturated node; got %v; want %v", err, errConcurrencyLimitReached) + } + + // The data must still reach the healthy node while the other node is unresponsive. + if !s.sendInsertRequestToAnyNode(data) { + t.Fatalf("cannot send data to any storage node while one node is unresponsive") + } +} diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index 41e90b6874..a43c9e4788 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/): prevent a single slow or unresponsive `vlstorage` node from stalling data ingestion to the remaining healthy `vlstorage` nodes. The `-insert.concurrency` command-line flag now caps the number of concurrent insert requests per `vlstorage` node and re-routes requests above the cap to other nodes. The re-routing is reported with the new `vl_insert_remote_concurrency_limit_reached_total` metric and a throttled warning in `vlinsert` logs. See [#1512](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1512). * 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). diff --git a/docs/victorialogs/metrics.md b/docs/victorialogs/metrics.md index c7f0102548..7d4f3c23a1 100644 --- a/docs/victorialogs/metrics.md +++ b/docs/victorialogs/metrics.md @@ -397,6 +397,14 @@ These metrics follow the Prometheus exposition format and can be used for monito **Description:** Remote storage node availability status where 1 means reachable and 0 means unreachable. Becomes 0 when send errors occur and temporarily disabled for 10 seconds, returns to 1 when successful requests resume. Cluster health monitoring. +### vl_insert_remote_concurrency_limit_reached_total +**Type:** Counter + +**Labels:** +- `addr`: storage node address + +**Description:** Data ingestion requests rejected because the storage node reached the concurrency limit set by `-insert.concurrency`. Rejected requests are re-routed to other storage nodes. Growing values on a single node indicate the node is slower than the rest of the cluster. + ### vl_select_remote_send_errors_total **Type:** Counter