docs: add OpenTelemetry integration article

See https://itnext.io/opentelemetry-otel-collectors-in-kubernetes-and-victoriametrics-stack-integration-d907ed0a15a0

Signed-off-by: hagen1778 <roman@victoriametrics.com>

docs/victoriametrics/Articles.md

@@ -114,6 +114,7 @@ See also [case studies](https://docs.victoriametrics.com/victoriametrics/casestu
 * [QCon London 2026: Wrangling Telemetry at Scale, a Guide to Self-Hosted Observability](https://www.infoq.com/news/2026/03/self-hosted-observability/)
 * [How We Made Telemetry Queries 10x Faster: Chunk-Split Caching for Metrics, Logs, and Traces](https://mirastacklabs.ai/blog/chunk-split-caching/)
 * [Claude Code: creating Kubernetes debugging AI Agent for VictoriaMetrics](https://rtfm.co.ua/en/claude-code-creating-kubernetes-debugging-ai-agent-for-victoriametrics/)
+* [OpenTelemetry: OTel Collectors in Kubernetes and VictoriaMetrics Stack integration](https://itnext.io/opentelemetry-otel-collectors-in-kubernetes-and-victoriametrics-stack-integration-d907ed0a15a0)
 
 ## Third-party articles and slides about VictoriaLogs
 

docs/victoriametrics/changelog/CHANGELOG.md

@@ -25,6 +25,7 @@ The sandbox cluster installation runs under the constant load generated by
 See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-releases/).
 
 ## tip
+* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): flush pending persistent queue data to chunk file before updating the metadata. This prevents the metadata writer offset from getting ahead of the chunk file size and avoids losing the persistent queue after an unclean shutdown. See [#11192](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11192).
 
 * FEATURE: [MetricsQL](https://docs.victoriametrics.com/victoriametrics/metricsql/): support `fill` modifiers to allow missing series on either side of a binary operation to be filled with a provided default value. See [#10598](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10598).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): Improve background discovery performance for [http_sd](https://docs.victoriametrics.com/victoriametrics/sd_configs/#http_sd_configs) discovery. See [#8838](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8838).
@@ -36,7 +37,6 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 * BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): properly apply limit to metrics metadata response. See [#11139](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11139).
 * BUGFIX: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): keep only one header navigation dropdown (`Explore`, `Tools`) open at a time. Previously, hovering across two dropdowns could briefly leave both open due to the close delay. See [#11224](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11224). Thanks to @antedotee for contribution.
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): fix a possible data race when processing OpenTelemetry metadata. See [#11238](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11238). Thanks to @nevgeny for contribution.
-* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): flush pending persistent queue data to chunk file before updating the metadata. This prevents the metadata writer offset from getting ahead of the chunk file size and avoids losing the persistent queue after an unclean shutdown. See [#11192](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11192).
 
 ## [v1.147.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.147.0)
 

lib/storage/partition.go

@@ -11,7 +11,9 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"unsafe"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/atomicutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
@@ -38,6 +40,14 @@ const maxInmemoryParts = 60
 // See appendPartsToMerge tests for details.
 const defaultPartsToMerge = 15
 
+// The number of shards for rawRow entries per partition.
+//
+// Higher number of shards reduces CPU contention and increases the max bandwidth on multi-core systems.
+var rawRowsShardsPerPartition = cgroup.AvailableCPUs()
+
+// The interval for flushing buffered rows into parts, so they become visible to search.
+const pendingRowsFlushInterval = 2 * time.Second
+
 // The interval for guaranteed flush of recently ingested data from memory to on-disk parts, so they survive process crash.
 var dataFlushInterval = 5 * time.Second
 
@@ -56,6 +66,11 @@ func SetDataFlushInterval(d time.Duration) {
 	dataFlushInterval = d
 }
 
+// The maximum number of rawRow items in rawRowsShard.
+//
+// Limit the maximum shard size to 8Mb, since this gives the lowest CPU usage under high ingestion rate.
+const maxRawRowsPerShard = (8 << 20) / int(unsafe.Sizeof(rawRow{}))
+
 // partition represents a partition.
 type partition struct {
 	activeInmemoryMerges atomic.Int64
@@ -465,29 +480,150 @@ func (pt *partition) AddRows(rows []rawRow) {
 		}
 	}
 
-	pt.rawRows.addRows(pt.flushRowssToInmemoryParts, rows)
+	pt.rawRows.addRows(pt, rows)
 }
 
 var isDebug = false
 
-func (pt *partition) flushRowssToInmemoryParts(rowss [][]rawRow) {
-	var nonEmptyRowss [][]rawRow
-	for _, rows := range rowss {
-		if len(rows) > 0 {
-			nonEmptyRowss = append(nonEmptyRowss, rows)
-		}
+type rawRowsShards struct {
+	flushDeadlineMs atomic.Int64
+
+	shardIdx atomic.Uint32
+
+	// Shards reduce lock contention when adding rows on multi-CPU systems.
+	shards []rawRowsShard
+
+	rowssToFlushLock sync.Mutex
+	rowssToFlush     [][]rawRow
+}
+
+func (rrss *rawRowsShards) init() {
+	rrss.shards = make([]rawRowsShard, rawRowsShardsPerPartition)
+}
+
+func (rrss *rawRowsShards) addRows(pt *partition, rows []rawRow) {
+	shards := rrss.shards
+	shardsLen := uint32(len(shards))
+	for len(rows) > 0 {
+		n := rrss.shardIdx.Add(1)
+		idx := n % shardsLen
+		tailRows, rowsToFlush := shards[idx].addRows(rows)
+		rrss.addRowsToFlush(pt, rowsToFlush)
+		rows = tailRows
 	}
-	if len(nonEmptyRowss) == 0 {
+}
+
+func (rrss *rawRowsShards) addRowsToFlush(pt *partition, rowsToFlush []rawRow) {
+	if len(rowsToFlush) == 0 {
+		return
+	}
+
+	var rowssToMerge [][]rawRow
+
+	rrss.rowssToFlushLock.Lock()
+	if len(rrss.rowssToFlush) == 0 {
+		rrss.updateFlushDeadline()
+	}
+	rrss.rowssToFlush = append(rrss.rowssToFlush, rowsToFlush)
+	if len(rrss.rowssToFlush) >= defaultPartsToMerge {
+		rowssToMerge = rrss.rowssToFlush
+		rrss.rowssToFlush = nil
+	}
+	rrss.rowssToFlushLock.Unlock()
+
+	pt.flushRowssToInmemoryParts(rowssToMerge)
+}
+
+func (rrss *rawRowsShards) Len() int {
+	n := 0
+	for i := range rrss.shards[:] {
+		n += rrss.shards[i].Len()
+	}
+
+	rrss.rowssToFlushLock.Lock()
+	for _, rows := range rrss.rowssToFlush {
+		n += len(rows)
+	}
+	rrss.rowssToFlushLock.Unlock()
+
+	return n
+}
+
+func (rrss *rawRowsShards) updateFlushDeadline() {
+	rrss.flushDeadlineMs.Store(time.Now().Add(pendingRowsFlushInterval).UnixMilli())
+}
+
+type rawRowsShardNopad struct {
+	flushDeadlineMs atomic.Int64
+
+	mu   sync.Mutex
+	rows []rawRow
+}
+
+type rawRowsShard struct {
+	rawRowsShardNopad
+
+	// The padding prevents false sharing
+	_ [atomicutil.CacheLineSize - unsafe.Sizeof(rawRowsShardNopad{})%atomicutil.CacheLineSize]byte
+}
+
+func (rrs *rawRowsShard) Len() int {
+	rrs.mu.Lock()
+	n := len(rrs.rows)
+	rrs.mu.Unlock()
+	return n
+}
+
+func (rrs *rawRowsShard) addRows(rows []rawRow) ([]rawRow, []rawRow) {
+	var rowsToFlush []rawRow
+
+	rrs.mu.Lock()
+	if cap(rrs.rows) == 0 {
+		rrs.rows = newRawRows()
+	}
+	if len(rrs.rows) == 0 {
+		rrs.updateFlushDeadline()
+	}
+	n := copy(rrs.rows[len(rrs.rows):cap(rrs.rows)], rows)
+	rrs.rows = rrs.rows[:len(rrs.rows)+n]
+	rows = rows[n:]
+	if len(rows) > 0 {
+		rowsToFlush = rrs.rows
+		rrs.rows = newRawRows()
+		rrs.updateFlushDeadline()
+		n = copy(rrs.rows[:cap(rrs.rows)], rows)
+		rrs.rows = rrs.rows[:n]
+		rows = rows[n:]
+	}
+	rrs.mu.Unlock()
+
+	return rows, rowsToFlush
+}
+
+func newRawRows() []rawRow {
+	return make([]rawRow, 0, maxRawRowsPerShard)
+}
+
+func (pt *partition) flushRowssToInmemoryParts(rowss [][]rawRow) {
+	if len(rowss) == 0 {
 		return
 	}
 
 	// Convert rowss into in-memory parts.
-	pws := make([]*partWrapper, len(nonEmptyRowss))
+	var pwsLock sync.Mutex
+	pws := make([]*partWrapper, 0, len(rowss))
 	wg := getWaitGroup()
-	for i, rows := range nonEmptyRowss {
+	for _, rows := range rowss {
 		inmemoryPartsConcurrencyCh <- struct{}{}
+
 		wg.Go(func() {
-			pws[i] = pt.mustCreateInmemoryPart(rows)
+			pw := pt.createInmemoryPart(rows)
+			if pw != nil {
+				pwsLock.Lock()
+				pws = append(pws, pw)
+				pwsLock.Unlock()
+			}
+
 			<-inmemoryPartsConcurrencyCh
 		})
 	}
@@ -742,14 +878,9 @@ func (pt *partition) mustMergeInmemoryPartsFinal(pws []*partWrapper) *partWrappe
 	return newPartWrapperFromInmemoryPart(mpDst, flushToDiskDeadline)
 }
 
-// mustCreateInmemoryPart creates a new in-memory part from rawRows.
-//
-// The number of rawRows cannot be zero. Otherwise the method will panic.
-//
-// The returned value is always a non-nil partWrapper.
-func (pt *partition) mustCreateInmemoryPart(rows []rawRow) *partWrapper {
+func (pt *partition) createInmemoryPart(rows []rawRow) *partWrapper {
 	if len(rows) == 0 {
-		logger.Panicf("BUG: a part cannot be created from 0 rawRows")
+		return nil
 	}
 	mp := getInmemoryPart()
 	mp.InitFromRows(rows)
@@ -1011,7 +1142,7 @@ func (pt *partition) pendingRowsFlusher() {
 }
 
 func (pt *partition) flushPendingRows(isFinal bool) {
-	pt.rawRows.flush(pt.flushRowssToInmemoryParts, isFinal)
+	pt.rawRows.flush(pt, isFinal)
 }
 
 func (pt *partition) flushInmemoryRowsToFiles() {
@@ -1037,6 +1168,66 @@ func (pt *partition) flushInmemoryPartsToFiles(isFinal bool) {
 	}
 }
 
+func (rrss *rawRowsShards) flush(pt *partition, isFinal bool) {
+	var dst [][]rawRow
+
+	currentTimeMs := time.Now().UnixMilli()
+	flushDeadlineMs := rrss.flushDeadlineMs.Load()
+	if isFinal || currentTimeMs >= flushDeadlineMs {
+		rrss.rowssToFlushLock.Lock()
+		dst = rrss.rowssToFlush
+		rrss.rowssToFlush = nil
+		rrss.rowssToFlushLock.Unlock()
+	}
+
+	for i := range rrss.shards {
+		dst = rrss.shards[i].appendRawRowsToFlush(dst, currentTimeMs, isFinal)
+	}
+
+	pt.flushRowssToInmemoryParts(dst)
+}
+
+func (rrs *rawRowsShard) appendRawRowsToFlush(dst [][]rawRow, currentTimeMs int64, isFinal bool) [][]rawRow {
+	flushDeadlineMs := rrs.flushDeadlineMs.Load()
+	if !isFinal && currentTimeMs < flushDeadlineMs {
+		// Fast path - nothing to flush
+		return dst
+	}
+
+	// Slow path - move rrs.rows to dst.
+	rrs.mu.Lock()
+	dst = appendRawRowss(dst, rrs.rows)
+	rrs.rows = rrs.rows[:0]
+	rrs.mu.Unlock()
+
+	return dst
+}
+
+func (rrs *rawRowsShard) updateFlushDeadline() {
+	rrs.flushDeadlineMs.Store(time.Now().Add(pendingRowsFlushInterval).UnixMilli())
+}
+
+func appendRawRowss(dst [][]rawRow, src []rawRow) [][]rawRow {
+	if len(src) == 0 {
+		return dst
+	}
+	if len(dst) == 0 {
+		dst = append(dst, newRawRows())
+	}
+	prows := &dst[len(dst)-1]
+	n := copy((*prows)[len(*prows):cap(*prows)], src)
+	*prows = (*prows)[:len(*prows)+n]
+	src = src[n:]
+	for len(src) > 0 {
+		rows := newRawRows()
+		n := copy(rows[:cap(rows)], src)
+		rows = rows[:len(rows)+n]
+		src = src[n:]
+		dst = append(dst, rows)
+	}
+	return dst
+}
+
 func (pt *partition) mergePartsToFiles(pws []*partWrapper, stopCh <-chan struct{}, concurrencyCh chan struct{}, useSparseCache bool) error {
 	pwsLen := len(pws)
 

lib/storage/raw_row.go

@@ -3,29 +3,11 @@ package storage
 import (
 	"sort"
 	"sync"
-	"sync/atomic"
-	"time"
-	"unsafe"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/atomicutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 )
 
-// The number of shards for rawRow entries.
-//
-// Higher number of shards reduces CPU contention and increases the max bandwidth on multi-core systems.
-var numRawRowsShards = cgroup.AvailableCPUs()
-
-// The interval for flushing buffered rows into parts, so they become visible to search.
-const pendingRowsFlushInterval = 2 * time.Second
-
-// The maximum number of rawRow items in rawRowsShard.
-//
-// Limit the maximum shard size to 8Mb, since this gives the lowest CPU usage under high ingestion rate.
-const maxRawRowsPerShard = (8 << 20) / int(unsafe.Sizeof(rawRow{}))
-
 // rawRow represents raw timeseries row.
 type rawRow struct {
 	// TSID is time series id.
@@ -167,182 +149,3 @@ func putRawRowsMarshaler(rrm *rawRowsMarshaler) {
 }
 
 var rrmPool sync.Pool
-
-type rawRowsShards struct {
-	flushDeadlineMs atomic.Int64
-
-	shardIdx atomic.Uint32
-
-	// Shards reduce lock contention when adding rows on multi-CPU systems.
-	shards []rawRowsShard
-
-	rowssToFlushLock sync.Mutex
-	rowssToFlush     [][]rawRow
-}
-
-func (rrss *rawRowsShards) init() {
-	rrss.shards = make([]rawRowsShard, numRawRowsShards)
-}
-
-func (rrss *rawRowsShards) Len() int {
-	n := 0
-	for i := range rrss.shards[:] {
-		n += rrss.shards[i].Len()
-	}
-
-	rrss.rowssToFlushLock.Lock()
-	for _, rows := range rrss.rowssToFlush {
-		n += len(rows)
-	}
-	rrss.rowssToFlushLock.Unlock()
-
-	return n
-}
-
-func (rrss *rawRowsShards) addRows(flush func([][]rawRow), rows []rawRow) {
-	shards := rrss.shards
-	shardsLen := uint32(len(shards))
-	for len(rows) > 0 {
-		n := rrss.shardIdx.Add(1)
-		idx := n % shardsLen
-		tailRows, rowsToFlush := shards[idx].addRows(rows)
-		rrss.addRowsToFlush(flush, rowsToFlush)
-		rows = tailRows
-	}
-}
-
-func (rrss *rawRowsShards) addRowsToFlush(flush func([][]rawRow), rowsToFlush []rawRow) {
-	if len(rowsToFlush) == 0 {
-		return
-	}
-
-	var rowssToMerge [][]rawRow
-
-	rrss.rowssToFlushLock.Lock()
-	if len(rrss.rowssToFlush) == 0 {
-		rrss.updateFlushDeadline()
-	}
-	rrss.rowssToFlush = append(rrss.rowssToFlush, rowsToFlush)
-	if len(rrss.rowssToFlush) >= defaultPartsToMerge {
-		rowssToMerge = rrss.rowssToFlush
-		rrss.rowssToFlush = nil
-	}
-	rrss.rowssToFlushLock.Unlock()
-
-	flush(rowssToMerge)
-}
-
-func (rrss *rawRowsShards) updateFlushDeadline() {
-	rrss.flushDeadlineMs.Store(time.Now().Add(pendingRowsFlushInterval).UnixMilli())
-}
-
-func (rrss *rawRowsShards) flush(flush func(rrs [][]rawRow), isFinal bool) {
-	var dst [][]rawRow
-
-	currentTimeMs := time.Now().UnixMilli()
-	flushDeadlineMs := rrss.flushDeadlineMs.Load()
-	if isFinal || currentTimeMs >= flushDeadlineMs {
-		rrss.rowssToFlushLock.Lock()
-		dst = rrss.rowssToFlush
-		rrss.rowssToFlush = nil
-		rrss.rowssToFlushLock.Unlock()
-	}
-
-	for i := range rrss.shards {
-		dst = rrss.shards[i].appendRawRowsToFlush(dst, currentTimeMs, isFinal)
-	}
-
-	flush(dst)
-}
-
-type rawRowsShardNopad struct {
-	flushDeadlineMs atomic.Int64
-
-	mu   sync.Mutex
-	rows []rawRow
-}
-
-type rawRowsShard struct {
-	rawRowsShardNopad
-
-	// The padding prevents false sharing
-	_ [atomicutil.CacheLineSize - unsafe.Sizeof(rawRowsShardNopad{})%atomicutil.CacheLineSize]byte
-}
-
-func (rrs *rawRowsShard) Len() int {
-	rrs.mu.Lock()
-	n := len(rrs.rows)
-	rrs.mu.Unlock()
-	return n
-}
-
-func (rrs *rawRowsShard) addRows(rows []rawRow) ([]rawRow, []rawRow) {
-	var rowsToFlush []rawRow
-
-	rrs.mu.Lock()
-	if cap(rrs.rows) == 0 {
-		rrs.rows = newRawRows()
-	}
-	if len(rrs.rows) == 0 {
-		rrs.updateFlushDeadline()
-	}
-	n := copy(rrs.rows[len(rrs.rows):cap(rrs.rows)], rows)
-	rrs.rows = rrs.rows[:len(rrs.rows)+n]
-	rows = rows[n:]
-	if len(rows) > 0 {
-		rowsToFlush = rrs.rows
-		rrs.rows = newRawRows()
-		rrs.updateFlushDeadline()
-		n = copy(rrs.rows[:cap(rrs.rows)], rows)
-		rrs.rows = rrs.rows[:n]
-		rows = rows[n:]
-	}
-	rrs.mu.Unlock()
-
-	return rows, rowsToFlush
-}
-
-func newRawRows() []rawRow {
-	return make([]rawRow, 0, maxRawRowsPerShard)
-}
-
-func (rrs *rawRowsShard) updateFlushDeadline() {
-	rrs.flushDeadlineMs.Store(time.Now().Add(pendingRowsFlushInterval).UnixMilli())
-}
-
-func (rrs *rawRowsShard) appendRawRowsToFlush(dst [][]rawRow, currentTimeMs int64, isFinal bool) [][]rawRow {
-	flushDeadlineMs := rrs.flushDeadlineMs.Load()
-	if !isFinal && currentTimeMs < flushDeadlineMs {
-		// Fast path - nothing to flush
-		return dst
-	}
-
-	// Slow path - move rrs.rows to dst.
-	rrs.mu.Lock()
-	dst = appendRawRowss(dst, rrs.rows)
-	rrs.rows = rrs.rows[:0]
-	rrs.mu.Unlock()
-
-	return dst
-}
-
-func appendRawRowss(dst [][]rawRow, src []rawRow) [][]rawRow {
-	if len(src) == 0 {
-		return dst
-	}
-	if len(dst) == 0 {
-		dst = append(dst, newRawRows())
-	}
-	prows := &dst[len(dst)-1]
-	n := copy((*prows)[len(*prows):cap(*prows)], src)
-	*prows = (*prows)[:len(*prows)+n]
-	src = src[n:]
-	for len(src) > 0 {
-		rows := newRawRows()
-		n := copy(rows[:cap(rows)], src)
-		rows = rows[:len(rows)+n]
-		src = src[n:]
-		dst = append(dst, rows)
-	}
-	return dst
-}