vlinsert: prevent a slow vlstorage node from stalling ingestion to the whole cluster

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. "+

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,33 +206,36 @@ 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)
 	if err == nil {
 		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
 

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")
+	}
+}

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).
 

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