lib/fs: disable OS readahead for data files opened for random access

Signed-off-by: Zasda Mikail <zmikail@victoriametrics.com>

docs/victoriametrics/changelog/CHANGELOG.md

@@ -25,6 +25,9 @@ The sandbox cluster installation runs under the constant load generated by
 See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-releases/).
 
 ## tip
+* FEATURE: all VictoriaMetrics components: expose the `process_cpu_cgroup_limited` metric, which is set to `1` when the cgroup CPU quota is lower than the number of logical CPU cores available to the process and `0` otherwise.
+* FEATURE: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): hint the OS with `fadvise(FADV_RANDOM)` and `madvise(MADV_RANDOM)` that data and index files are read at random offsets during queries. This disables OS readahead for these files and reduces disk read amplification on hosts with high `read_ahead_kb` settings. The hints can be disabled with the `-fs.disableFadviseRandomRead` command-line flag.
+* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): include `vm_account_id` and `vm_project_id` labels in the instance identity used by [monitoring data exchange](https://docs.victoriametrics.com/victoriametrics/vmagent/#monitoring-data-exchange) filtering. Previously, when [multitenant handlers](https://docs.victoriametrics.com/victoriametrics/vmagent/#multitenancy) were enabled, series from different tenants sharing the same `job` and `instance` labels could be misidentified as coming from a discovered VictoriaMetrics instance and leak into the `-remoteWrite.mdx.enable` destination. See [#11381](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11381).
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/), `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) and [vmctl](https://docs.victoriametrics.com/victoriametrics/vmctl/): properly parse small fractional Unix timestamps in timestamp args such as `start` and `end` in `/api/v1/query_range` and `--vm-native-filter-time-start` and `--vm-native-filter-time-end` in `vmctl`. Previously, fractional Unix timestamps with the integer part below `9223372` were interpreted with the wrong unit, for example `12.0` was parsed as `12000` seconds instead of `12` seconds. See [#11324](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11324).
 
 ## [v1.149.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.149.0)

lib/cgroup/cpu.go

@@ -21,7 +21,7 @@ func AvailableCPUs() int {
 }
 
 func init() {
-	cpuQuota := getCPUQuota()
+	cpuQuota, cpuCgroupLimited := getCPUQuota()
 	if cpuQuota > 0 {
 		updateGOMAXPROCSToCPUQuota(cpuQuota)
 	}
@@ -32,6 +32,12 @@ func init() {
 	metrics.NewGauge(`process_cpu_cores_available`, func() float64 {
 		return cpuCoresAvailable
 	})
+	metrics.NewGauge(`process_cpu_cgroup_limited`, func() float64 {
+		if cpuCgroupLimited {
+			return 1
+		}
+		return 0
+	})
 }
 
 // updateGOMAXPROCSToCPUQuota updates GOMAXPROCS to cpuQuota if GOMAXPROCS isn't set in environment var.
@@ -60,17 +66,21 @@ func updateGOMAXPROCSToCPUQuota(cpuQuota float64) {
 	runtime.GOMAXPROCS(gomaxprocs)
 }
 
-func getCPUQuota() float64 {
+func getCPUQuota() (float64, bool) {
 	cpuQuota, err := getCPUQuotaGeneric()
 	if err != nil {
-		return 0
+		return 0, false
 	}
 	if cpuQuota <= 0 {
 		// The quota isn't set. This may be the case in multilevel containers.
 		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/685#issuecomment-674423728
-		return getOnlineCPUCount()
+		return getOnlineCPUCount(), false
 	}
-	return cpuQuota
+	return cpuQuota, isCPUCgroupLimited(cpuQuota, runtime.NumCPU())
+}
+
+func isCPUCgroupLimited(cpuQuota float64, cpuCount int) bool {
+	return cpuQuota > 0 && cpuQuota < float64(cpuCount)
 }
 
 func getCPUQuotaGeneric() (float64, error) {

lib/cgroup/cpu_test.go

@@ -4,6 +4,23 @@ import (
 	"testing"
 )
 
+func TestIsCPUCgroupLimited(t *testing.T) {
+	f := func(cpuQuota float64, cpuCount int, expected bool) {
+		t.Helper()
+		got := isCPUCgroupLimited(cpuQuota, cpuCount)
+		if got != expected {
+			t.Fatalf("unexpected result from isCPUCgroupLimited(%f, %d); got %v; want %v", cpuQuota, cpuCount, got, expected)
+		}
+	}
+
+	f(-1, 8, false)
+	f(0, 8, false)
+	f(8, 8, false)
+	f(16, 8, false)
+	f(7.5, 8, true)
+	f(0.5, 8, true)
+}
+
 func TestCountCPUs(t *testing.T) {
 	f := func(s string, nExpected int) {
 		t.Helper()

lib/fs/fadvise_darwin.go

@@ -8,3 +8,13 @@ func fadviseSequentialRead(_ *os.File, _ bool) error {
 	// TODO: implement this properly
 	return nil
 }
+
+func fadviseRandomRead(_ *os.File) error {
+	// TODO: implement this properly
+	return nil
+}
+
+func madviseRandomRead(_ []byte) error {
+	// TODO: implement this properly
+	return nil
+}

lib/fs/fadvise_openbsd.go

@@ -8,3 +8,13 @@ func fadviseSequentialRead(f *os.File, prefetch bool) error {
 	// TODO: implement this properly
 	return nil
 }
+
+func fadviseRandomRead(_ *os.File) error {
+	// TODO: implement this properly
+	return nil
+}
+
+func madviseRandomRead(_ []byte) error {
+	// TODO: implement this properly
+	return nil
+}

lib/fs/fadvise_solaris.go

@@ -6,3 +6,13 @@ func fadviseSequentialRead(f *os.File, prefetch bool) error {
 	// TODO: implement this properly
 	return nil
 }
+
+func fadviseRandomRead(_ *os.File) error {
+	// TODO: implement this properly
+	return nil
+}
+
+func madviseRandomRead(_ []byte) error {
+	// TODO: implement this properly
+	return nil
+}

lib/fs/fadvise_unix.go

@@ -20,3 +20,18 @@ func fadviseSequentialRead(f *os.File, prefetch bool) error {
 	}
 	return nil
 }
+
+func fadviseRandomRead(f *os.File) error {
+	fd := int(f.Fd())
+	if err := unix.Fadvise(fd, 0, 0, unix.FADV_RANDOM); err != nil {
+		return fmt.Errorf("error returned from unix.Fadvise(FADV_RANDOM): %w", err)
+	}
+	return nil
+}
+
+func madviseRandomRead(data []byte) error {
+	if err := unix.Madvise(data, unix.MADV_RANDOM); err != nil {
+		return fmt.Errorf("error returned from unix.Madvise(MADV_RANDOM): %w", err)
+	}
+	return nil
+}

lib/fs/fs_windows.go

@@ -116,6 +116,16 @@ func fadviseSequentialRead(_ *os.File, _ bool) error {
 	return nil
 }
 
+// stub
+func fadviseRandomRead(_ *os.File) error {
+	return nil
+}
+
+// stub
+func madviseRandomRead(_ []byte) error {
+	return nil
+}
+
 // https://docs.microsoft.com/en-us/windows/win32/api/minwinbase/ns-minwinbase-overlapped
 func newOverlapped() (*windows.Overlapped, error) {
 	event, err := windows.CreateEvent(nil, 1, 1, nil)

lib/fs/reader_at.go

@@ -17,6 +17,11 @@ var disableMmap = flag.Bool("fs.disableMmap", is32BitPtr, "Whether to use pread(
 	"By default, mmap() is used for 64-bit arches and pread() is used for 32-bit arches, since they cannot read data files bigger than 2^32 bytes in memory. "+
 	"mmap() is usually faster for reading small data chunks than pread()")
 
+var disableFadviseRandomRead = flag.Bool("fs.disableFadviseRandomRead", false, "Whether to disable fadvise(FADV_RANDOM) and madvise(MADV_RANDOM) hints "+
+	"for data files opened for random access. These hints disable OS readahead for such files. This reduces the amount of unneeded data read from disk "+
+	"during queries, which select small number of blocks scattered across big data files. "+
+	"Disabling the hints may improve performance for queries, which read the most of the data in big data files")
+
 var disableMincore = flag.Bool("fs.disableMincore", false, "Whether to disable the mincore() syscall for checking mmap()ed files. "+
 	"By default, mincore() is used to detect whether mmap()ed file pages are resident in memory. "+
 	"Disabling mincore() may be needed on older ZFS filesystems (below 2.1.5), since it may trigger ZFS bug. "+
@@ -50,6 +55,9 @@ type ReaderAt struct {
 	mrLock sync.Mutex
 
 	useLocalStats bool
+
+	// useRandomReadHint instructs hinting the OS that the file is read at random offsets.
+	useRandomReadHint bool
 }
 
 // Path returns path to r.
@@ -103,6 +111,9 @@ func (r *ReaderAt) getMmapReader() *mmapReader {
 	mr = r.mr.Load()
 	if mr == nil {
 		mr = newMmapReaderFromPath(r.path)
+		if r.useRandomReadHint && !*disableFadviseRandomRead {
+			mr.mustHintRandomRead(r.path)
+		}
 		r.mr.Store(mr)
 	}
 	r.mrLock.Unlock()
@@ -155,10 +166,14 @@ func (r *ReaderAt) MustFadviseSequentialRead(prefetch bool) {
 
 // MustOpenReaderAt opens ReaderAt for reading from the file located at path.
 //
+// The OS is hinted that the file is read at random offsets, so it must skip readahead
+// on page cache misses. Use NewReaderAt for files read mostly sequentially.
+//
 // MustClose must be called on the returned ReaderAt when it is no longer needed.
 func MustOpenReaderAt(path string) *ReaderAt {
 	var r ReaderAt
 	r.path = path
+	r.useRandomReadHint = true
 	return &r
 }
 
@@ -229,6 +244,22 @@ func newMmapReaderFromFile(f *os.File) *mmapReader {
 	return mr
 }
 
+// mustHintRandomRead hints the OS that mr is read at random offsets,
+// so the OS must skip readahead on page cache misses.
+//
+// fadvise(FADV_RANDOM) covers reads via pread() syscall at mustReadAtViaSyscall,
+// while madvise(MADV_RANDOM) covers page faults on the mmap()ed region.
+func (mr *mmapReader) mustHintRandomRead(path string) {
+	if err := fadviseRandomRead(mr.f); err != nil {
+		logger.Panicf("FATAL: error in fadviseRandomRead(%q): %s", path, err)
+	}
+	if len(mr.mmapData) > 0 {
+		if err := madviseRandomRead(mr.mmapData[:cap(mr.mmapData)]); err != nil {
+			logger.Panicf("FATAL: error in madviseRandomRead(%q): %s", path, err)
+		}
+	}
+}
+
 func (mr *mmapReader) mustClose() {
 	fname := mr.f.Name()
 	if len(mr.mmapData) > 0 {

lib/mdx/filter.go

@@ -23,6 +23,8 @@ const (
 	vmAppLabelName         = "victoriametrics_app"
 	vmAppLabelValue        = "true"
 	vmAppVersionMetricName = "vm_app_version"
+	accountIDLabelName     = "vm_account_id"
+	projectIDLabelName     = "vm_project_id"
 )
 
 // Ctx defines filtering context
@@ -37,6 +39,8 @@ type Ctx struct {
 	hasFilterLabelValue  bool
 	jobLabelValue        string
 	instanceLabelValue   string
+	accountIDLabelValue  string
+	projectIDLabelValue  string
 }
 
 func (ctx *Ctx) reset() {
@@ -49,6 +53,8 @@ func (ctx *Ctx) reset() {
 	ctx.hasFilterLabelValue = false
 	ctx.jobLabelValue = ""
 	ctx.instanceLabelValue = ""
+	ctx.accountIDLabelValue = ""
+	ctx.projectIDLabelValue = ""
 }
 
 var ctxPool = &sync.Pool{
@@ -70,7 +76,8 @@ func PutContext(ctx *Ctx) {
 	ctxPool.Put(ctx)
 }
 
-// Filter manages the list of VictoriaMetrics instances grouped by job:instance labels.
+// Filter manages the list of VictoriaMetrics instances grouped by job:instance labels
+// and vm_account_id:vm_project_id labels when they are present.
 // job and instance must present at timeseries.
 //
 // Filter keeps timeseries with any of the following conditions:
@@ -168,6 +175,10 @@ func (ctx *Ctx) prepare(labels []prompb.Label, filterByLabelName, label string)
 			ctx.jobLabelValue = l.Value
 		case "instance":
 			ctx.instanceLabelValue = l.Value
+		case accountIDLabelName:
+			ctx.accountIDLabelValue = l.Value
+		case projectIDLabelName:
+			ctx.projectIDLabelValue = l.Value
 		case vmAppLabelName:
 			if l.Value == vmAppLabelValue {
 				ctx.hasVMAppLabel = true
@@ -197,6 +208,10 @@ func (ctx *Ctx) formatTimeSeriesKey() string {
 	buf = strconv.AppendQuote(buf, ctx.jobLabelValue)
 	buf = append(buf, ':')
 	buf = strconv.AppendQuote(buf, ctx.instanceLabelValue)
+	buf = append(buf, ':')
+	buf = strconv.AppendQuote(buf, ctx.accountIDLabelValue)
+	buf = append(buf, ':')
+	buf = strconv.AppendQuote(buf, ctx.projectIDLabelValue)
 	ctx.buf = buf
 	return bytesutil.ToUnsafeString(buf)
 }

lib/mdx/filter_test.go

@@ -375,6 +375,132 @@ func TestMdxInstanceFilter(t *testing.T) {
 				},
 			}})
 
+	// metrics from another tenant with the same job and instance must be dropped.
+	f([]prompb.TimeSeries{
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "vm_app_version"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "shared:8428"},
+			{Name: "vm_account_id", Value: "1042"},
+			{Name: "vm_project_id", Value: "0"},
+		}},
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "http_requests_total"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "shared:8428"},
+			{Name: "vm_account_id", Value: "2077"},
+			{Name: "vm_project_id", Value: "0"},
+		}},
+	}, []prompb.TimeSeries{
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "vm_app_version"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "shared:8428"},
+			{Name: "vm_account_id", Value: "1042"},
+			{Name: "vm_project_id", Value: "0"},
+			{Name: "victoriametrics_app", Value: "true"},
+		}},
+	})
+
+	// metrics from the same tenant and instance must be preserved.
+	f([]prompb.TimeSeries{
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "vm_app_version"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "shared:8428"},
+			{Name: "vm_account_id", Value: "1042"},
+			{Name: "vm_project_id", Value: "0"},
+		}},
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "http_requests_total"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "shared:8428"},
+			{Name: "vm_account_id", Value: "1042"},
+			{Name: "vm_project_id", Value: "0"},
+		}},
+	}, []prompb.TimeSeries{
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "vm_app_version"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "shared:8428"},
+			{Name: "vm_account_id", Value: "1042"},
+			{Name: "vm_project_id", Value: "0"},
+			{Name: "victoriametrics_app", Value: "true"},
+		}},
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "http_requests_total"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "shared:8428"},
+			{Name: "vm_account_id", Value: "1042"},
+			{Name: "vm_project_id", Value: "0"},
+			{Name: "victoriametrics_app", Value: "true"},
+		}},
+	})
+
+	// instances with and without tenant labels must remain distinct.
+	f([]prompb.TimeSeries{
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "vm_app_version"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "x"},
+		}},
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "http_requests_total"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "x"},
+			{Name: "vm_account_id", Value: "1"},
+		}},
+	}, []prompb.TimeSeries{
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "vm_app_version"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "x"},
+			{Name: "victoriametrics_app", Value: "true"},
+		}},
+	})
+
+	// partial tenant labels must remain distinct from absent and full tenant labels.
+	f([]prompb.TimeSeries{
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "vm_app_version"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "partial:8428"},
+			{Name: "vm_account_id", Value: "1"},
+		}},
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "http_requests_total"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "partial:8428"},
+			{Name: "vm_account_id", Value: "1"},
+		}},
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "http_requests_total"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "partial:8428"},
+		}},
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "http_requests_total"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "partial:8428"},
+			{Name: "vm_account_id", Value: "1"},
+			{Name: "vm_project_id", Value: "0"},
+		}},
+	}, []prompb.TimeSeries{
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "vm_app_version"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "partial:8428"},
+			{Name: "vm_account_id", Value: "1"},
+			{Name: "victoriametrics_app", Value: "true"},
+		}},
+		{Labels: []prompb.Label{
+			{Name: "__name__", Value: "http_requests_total"},
+			{Name: "job", Value: "test"},
+			{Name: "instance", Value: "partial:8428"},
+			{Name: "vm_account_id", Value: "1"},
+			{Name: "victoriametrics_app", Value: "true"},
+		}},
+	})
 }
 
 func TestMdxInstanceFilterConcurrent(t *testing.T) {