app/vmselect: fail the query request directly when there is not enough disk space

Previously, vmselect may crash if there is not enough free disk space to store tmp files from vmstorage. vmselect stores data blocks received from vmstorage if response size exceeds in-memory limit.
 It reduced vmselect availability, because it makes impossible to serve small queries. And huge query may crash vmselect for all users.

 So this commit adds a error check instead. If there is not enough disk space, vmselect will stop query execution and return error back to user.

fixes https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4688

app/vmselect/netstorage/netstorage.go

@@ -1807,6 +1807,21 @@ type limitExceededErr struct {
 // Error satisfies error interface
 func (e limitExceededErr) Error() string { return e.err.Error() }
 
+// tmpBlocksFileErr generated by vmselect when it cannot store
+// the data received from vmstorage nodes at the temporary blocks file -
+// for example, on the disk space shortage at -cacheDataPath.
+type tmpBlocksFileErr struct {
+	err error
+}
+
+func (e *tmpBlocksFileErr) Error() string {
+	return e.err.Error()
+}
+
+func (e *tmpBlocksFileErr) Unwrap() error {
+	return e.err
+}
+
 // ProcessSearchQuery performs sq until the given deadline.
 //
 // Results.RunParallel or Results.Cancel must be called on the returned Results.
@@ -1844,7 +1859,9 @@ func ProcessSearchQuery(qt *querytracer.Tracer, denyPartialResponse bool, sq *st
 		}
 
 		if err := tbfw.RegisterAndWriteBlock(mb, workerID); err != nil {
-			return fmt.Errorf("cannot write MetricBlock to temporary blocks file: %w", err)
+			return &tmpBlocksFileErr{
+				err: fmt.Errorf("cannot write MetricBlock to temporary blocks file: %w", err),
+			}
 		}
 		return nil
 	}
@@ -2109,6 +2126,17 @@ func (snr *storageNodesRequest) collectResults(partialResultsCounter *metrics.Co
 				snr.finishQueryTracers("cancel request because query complexity limit was exceeded")
 				return false, err
 			}
+			var tbfErr *tmpBlocksFileErr
+			if errors.As(err, &tbfErr) {
+				// Immediately return the error, since vmselect cannot store the data received
+				// from vmstorage nodes due to file system issues like disk space shortage.
+				snr.finishQueryTracers("cancel request because vmselect cannot store the received data")
+				err = &httpserver.ErrorWithStatusCode{
+					Err:        err,
+					StatusCode: http.StatusServiceUnavailable,
+				}
+				return false, err
+			}
 
 			errsPartialPerGroup[group] = append(errsPartialPerGroup[group], err)
 			if snr.denyPartialResponse && len(errsPartialPerGroup[group]) >= group.replicationFactor {
@@ -2468,10 +2496,12 @@ func (sn *storageNode) execOnConnWithPossibleRetry(qt *querytracer.Tracer, funcN
 	var er *errRemote
 	var ne net.Error
 	var le *limitExceededErr
-	if errors.As(err, &le) || errors.As(err, &er) || errors.As(err, &ne) && ne.Timeout() || deadline.Exceeded() || errors.Is(err, errCannotObtainConn) {
+	var tbfErr *tmpBlocksFileErr
+	if errors.As(err, &le) || errors.As(err, &tbfErr) || errors.As(err, &er) || errors.As(err, &ne) && ne.Timeout() || deadline.Exceeded() || errors.Is(err, errCannotObtainConn) {
 		// There is no sense in repeating the query on the following errors:
 		//
 		//   - exceeded complexity limits (limitExceededErr)
+		//   - vmselect cannot store the received data (tmpBlocksFileErr)
 		//   - induced by vmstorage (errRemote)
 		//   - network timeout errors
 		//   - request deadline exceeded errors

app/vmselect/netstorage/tmp_blocks_file.go

@@ -63,6 +63,9 @@ type tmpBlocksFile struct {
 	r *fs.ReaderAt
 
 	offset uint64
+
+	// err stores the first error occurred while writing the temporary blocks file.
+	err error
 }
 
 func getTmpBlocksFile() *tmpBlocksFile {
@@ -83,6 +86,7 @@ func putTmpBlocksFile(tbf *tmpBlocksFile) {
 	tbf.f = nil
 	tbf.r = nil
 	tbf.offset = 0
+	tbf.err = nil
 	tmpBlocksFilePool.Put(tbf)
 }
 
@@ -109,8 +113,16 @@ var (
 //
 // It returns errors since the operation may fail on space shortage
 // and this must be handled.
+//
+// The tbf is left unusable after the first error, since a failed write cannot be undone:
+// the returned addresses are derived from tbf.offset before the data is flushed from tbf.buf
+// to the file, the failed flush may be partial, and the remaining buffer is dropped.
 func (tbf *tmpBlocksFile) WriteBlockData(b []byte, tbfIdx uint) (tmpBlockAddr, error) {
 	var addr tmpBlockAddr
+	if tbf.err != nil {
+		// Do not write anything to the tbf after the first failed write
+		return addr, tbf.err
+	}
 	addr.tbfIdx = tbfIdx
 	addr.offset = tbf.offset
 	addr.size = len(b)
@@ -125,7 +137,8 @@ func (tbf *tmpBlocksFile) WriteBlockData(b []byte, tbfIdx uint) (tmpBlockAddr, e
 	if tbf.f == nil {
 		f, err := os.CreateTemp(tmpBlocksDir, "")
 		if err != nil {
-			return addr, err
+			tbf.err = fmt.Errorf("cannot create temporary blocks file at %q: %w", tmpBlocksDir, err)
+			return addr, tbf.err
 		}
 		tbf.f = f
 		tmpBlocksFilesCreated.Inc()
@@ -133,7 +146,9 @@ func (tbf *tmpBlocksFile) WriteBlockData(b []byte, tbfIdx uint) (tmpBlockAddr, e
 	_, err := tbf.f.Write(tbf.buf)
 	tbf.buf = append(tbf.buf[:0], b...)
 	if err != nil {
-		return addr, fmt.Errorf("cannot write block to %q: %w", tbf.f.Name(), err)
+		// The blocks buffered at tbf.buf could be partially lost, mark the tbf as unusable.
+		tbf.err = fmt.Errorf("cannot write block to %q: %w", tbf.f.Name(), err)
+		return addr, tbf.err
 	}
 	return addr, nil
 }
@@ -144,12 +159,16 @@ func (tbf *tmpBlocksFile) Len() uint64 {
 }
 
 func (tbf *tmpBlocksFile) Finalize() error {
+	if tbf.err != nil {
+		return tbf.err
+	}
 	if tbf.f == nil {
 		return nil
 	}
 	fname := tbf.f.Name()
 	if _, err := tbf.f.Write(tbf.buf); err != nil {
-		return fmt.Errorf("cannot write the remaining %d bytes to %q: %w", len(tbf.buf), fname, err)
+		tbf.err = fmt.Errorf("cannot write the remaining %d bytes to %q: %w", len(tbf.buf), fname, err)
+		return tbf.err
 	}
 	tbf.buf = tbf.buf[:0]
 	r := fs.NewReaderAt(tbf.f)
@@ -169,6 +188,10 @@ func (tbf *tmpBlocksFile) Finalize() error {
 }
 
 func (tbf *tmpBlocksFile) MustReadBlockAt(dst *storage.Block, addr tmpBlockAddr) {
+	if tbf.err != nil {
+		// This should never happen, since Finalize() already returns the error for such a tbf.
+		logger.Panicf("BUG: cannot read block at %s from the temporary blocks file with the failed write: %s", addr, tbf.err)
+	}
 	var buf []byte
 	if tbf.r == nil {
 		buf = tbf.buf[addr.offset : addr.offset+uint64(addr.size)]

app/vmselect/netstorage/tmp_blocks_file_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math/rand"
 	"os"
+	"path/filepath"
 	"reflect"
 	"testing"
 	"time"
@@ -29,6 +30,39 @@ func TestTmpBlocksFileSerial(t *testing.T) {
 	}
 }
 
+func TestTmpBlocksFileWriteFailure(t *testing.T) {
+	// Emulate the disk write failure by pointing tmpBlocksDir to a non-existing directory.
+	tmpBlocksDirOrig := tmpBlocksDir
+	tmpBlocksDir = filepath.Join(tmpBlocksDirOrig, "non-existing-dir")
+	defer func() {
+		tmpBlocksDir = tmpBlocksDirOrig
+	}()
+
+	tbf := getTmpBlocksFile()
+	defer putTmpBlocksFile(tbf)
+
+	// Write blocks until tbf.buf is flushed to the file. The flush must fail.
+	b := make([]byte, 64*1024)
+	var writeErr error
+	for range maxInmemoryTmpBlocksFile()/len(b) + 2 {
+		if _, err := tbf.WriteBlockData(b, 0); err != nil {
+			writeErr = err
+			break
+		}
+	}
+	if writeErr == nil {
+		t.Fatalf("expecting non-nil error from WriteBlockData")
+	}
+
+	if _, err := tbf.WriteBlockData(b, 0); err != writeErr {
+		t.Fatalf("expecting non-nil error from WriteBlockData after the failed write")
+	}
+
+	if err := tbf.Finalize(); err != writeErr {
+		t.Fatalf("expecting non-nil error from Finalize after the failed write")
+	}
+}
+
 func TestTmpBlocksFileConcurrent(t *testing.T) {
 	concurrency := 3
 	ch := make(chan error, concurrency)

docs/victoriametrics/changelog/CHANGELOG.md

@@ -33,6 +33,7 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fix infinite loop in the OpenTelemetry Firehose ingestion endpoint (`/opentelemetry/api/v1/push`) when receiving a malformed record with an incomplete varint in the `data` field. Previously this caused the goroutine to spin forever, permanently consuming CPU until the process was restarted.
 * BUGFIX: [vmalert-tool](https://docs.victoriametrics.com/victoriametrics/vmalert-tool/): reuse connections to `-remoteWrite.url` when writing the results of recording rules and alerts. Previously every series was sent over a new connection, which left a lot of sockets in `TIME_WAIT` state and could exhaust the ephemeral port range. The number of idle connections can be tuned via the new `-remoteWrite.maxIdleConnections` command-line flag. Thanks @evkuzin for contribution.
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): prevent process crash in `sort_by_label_numeric()` and `sort_by_label_numeric_desc()` when a label value contains a number with 309 or more digits. See [#11423](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11423).
+* BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fail the query request directly when there is not enough disk space to store temporary search results. Previously, such queries could lead to vmselect crash. See [#4688](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4688).
 
 ## [v1.150.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.150.0)