app/vlinsert/insertutil: optimize LogMessageProcessor #1021

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

valyala merged 85 commits into master from lmp-atomics

Feb 5, 2026

app/vlinsert/insertutil/common_params.go

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -219,6 +219,9 @@ type logMessageProcessor struct {
  
    	rowsIngestedTotal  *metrics.Counter

    	bytesIngestedTotal *metrics.Counter

    	flushDuration      *metrics.Summary

    	unflushedRows  int

    	unflushedBytes int

    }

    func (lmp *logMessageProcessor) initPeriodicFlush() {

    @@ -249,9 +252,12 @@ func (lmp *logMessageProcessor) initPeriodicFlush() {
  
    // If streamFieldsLen >= 0, then the given number of the initial fields is used as log stream fields

    // instead of the pre-configured stream fields.

    func (lmp *logMessageProcessor) AddRow(timestamp int64, fields []logstorage.Field, streamFieldsLen int) {

    	lmp.rowsIngestedTotal.Inc()

    	lmp.mu.Lock()

    	defer lmp.mu.Unlock()

    	lmp.unflushedRows++

    	n := logstorage.EstimatedJSONRowLen(fields)

    	lmp.bytesIngestedTotal.Add(n)

    	lmp.unflushedBytes += n

    	if len(fields) > *MaxFieldsPerLine {

    		line := logstorage.MarshalFieldsToJSON(nil, fields)

    @@ -260,9 +266,6 @@ func (lmp *logMessageProcessor) AddRow(timestamp int64, fields []logstorage.Fiel
  
    		return

    	}

    	lmp.mu.Lock()

    	defer lmp.mu.Unlock()

    	lmp.lr.MustAdd(lmp.cp.TenantID, timestamp, fields, streamFieldsLen)

    	if lmp.cp.Debug {

    @@ -285,9 +288,12 @@ type InsertRowProcessor interface {
  
    // AddInsertRow adds r to lmp.

    func (lmp *logMessageProcessor) AddInsertRow(r *logstorage.InsertRow) {

    	lmp.rowsIngestedTotal.Inc()

    	lmp.mu.Lock()

    	defer lmp.mu.Unlock()

    	lmp.unflushedRows++

    	n := logstorage.EstimatedJSONRowLen(r.Fields)

    	lmp.bytesIngestedTotal.Add(n)

    	lmp.unflushedBytes += n

    	if len(r.Fields) > *MaxFieldsPerLine {

    		line := logstorage.MarshalFieldsToJSON(nil, r.Fields)

    @@ -296,9 +302,6 @@ func (lmp *logMessageProcessor) AddInsertRow(r *logstorage.InsertRow) {
  
    		return

    	}

    	lmp.mu.Lock()

    	defer lmp.mu.Unlock()

    	lmp.lr.MustAddInsertRow(r)

    	if lmp.cp.Debug {

    @@ -319,7 +322,13 @@ func (lmp *logMessageProcessor) flushLocked() {
  
    	lmp.lastFlushTime = start

    	logRowsStorage.MustAddRows(lmp.lr)

    	lmp.lr.ResetKeepSettings()

    	lmp.flushDuration.UpdateDuration(start)

    	lmp.rowsIngestedTotal.Add(lmp.unflushedRows)

    	lmp.bytesIngestedTotal.Add(lmp.unflushedBytes)

    	lmp.unflushedRows = 0

    	lmp.unflushedBytes = 0

    }

    // MustClose flushes the remaining data to the underlying storage and closes lmp.

    @@ -338,9 +347,11 @@ func (lmp *logMessageProcessor) MustClose() {
  
    // MustClose() must be called on the returned LogMessageProcessor when it is no longer needed.

    func (cp *CommonParams) NewLogMessageProcessor(protocolName string, isStreamMode bool) LogMessageProcessor {

    	lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields, cp.DecolorizeFields, cp.ExtraFields, *defaultMsgValue)

    	rowsIngestedTotal := metrics.GetOrCreateCounter(fmt.Sprintf("vl_rows_ingested_total{type=%q}", protocolName))

    	bytesIngestedTotal := metrics.GetOrCreateCounter(fmt.Sprintf("vl_bytes_ingested_total{type=%q}", protocolName))

    	flushDuration := metrics.GetOrCreateSummary(fmt.Sprintf("vl_insert_flush_duration_seconds{type=%q}", protocolName))

    	lmp := &logMessageProcessor{

    		cp: cp,

    		lr: lr,

app/vlinsert/insertutil/common_params_timing_test.go

-Original file line number
+Diff line change
@@ -0,0 +1,66 @@
+    package insertutil
+    import (
+    	"net/http"
+    	"testing"
+    	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+    	"github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage"
+    )
+    func BenchmarkAddRow(b *testing.B) {
+    	r, err := http.NewRequest(http.MethodPost, "https://foo.bar/baz?_stream_fields=kubernetes.pod_name,kubernetes.pod_labels.app", nil)
+    	if err != nil {
+    		b.Fatal(err)
+    	}
+    	cp, err := GetCommonParams(r)
+    	if err != nil {
+    		b.Fatal(err)
+    	}
+    	SetLogRowsStorage(benchmarkStorage{})
+    	ts := int64(fasttime.UnixTimestamp()) * 1e9
+    	rows := [][]logstorage.Field{
+    		{
+    			{Name: "kubernetes.pod_name", Value: "victoria-logs-single-0"},
+    			{Name: "kubernetes.pod_labels.app", Value: "VictoriaLogs"},
+    			{Name: "kubernetes.container_name", Value: "oauth2-proxy"},
+    			{Name: "_msg", Value: `GET - "/ready" HTTP/1.1 "kube-probe/1.34" 200 2 0.000`},
+    		},
+    		{
+    			{Name: "kubernetes.pod_name", Value: "victoria-logs-single-0"},
+    			{Name: "kubernetes.pod_labels.app", Value: "VictoriaLogs"},
+    			{Name: "kubernetes.container_name", Value: "oauth2-proxy"},
+    			{Name: "_msg", Value: `GET - "/ready" HTTP/1.1 "GoogleHC/1.0" 200 2 0.000`},
+    		},
+    	}
+    	var n int
+    	for _, row := range rows {
+    		n += logstorage.EstimatedJSONRowLen(row)
+    	}
+    	b.SetBytes(int64(n))
+    	b.ReportAllocs()
+    	b.RunParallel(func(pb *testing.PB) {
+    		lmp := cp.NewLogMessageProcessor("test", false)
+    		defer lmp.MustClose()
+    		for pb.Next() {
+    			for _, row := range rows {
+    				lmp.AddRow(ts, row, -1)
+    			}
+    		}
+    	})
+    }
+    type benchmarkStorage struct{}
+    func (benchmarkStorage) MustAddRows(*logstorage.LogRows) {
+    }
+    func (benchmarkStorage) CanWriteData() error {
+    	return nil
+    }

docs/victorialogs/CHANGELOG.md

-Original file line number
+Diff line change
@@ Expand Up / @@ -57,6 +57,7 @@ Released at 2026-01-27 @@
     * FEATURE: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/): add an ability to avoid [flattening](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) of JSON values for the given keys enumerated via `preserve_json_keys` query arg or via `VL-Preserve-JSON-Keys` http request header. This may be useful if the input JSON-encoded logs contain high-cardinality key names such as `sku-*` in the `{"items":{"sku-1":{...},...,"sku-N":{...}}}`. Then `preserve_json_keys=items` would preserve the `items` field value as is, e.g. it will be equal to the string `{"sku-1":{...},...,"sku-N":{...}}`. See more details about these options at [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/#http-parameters) and at [#1002](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1002).
     * FEATURE: [dashboards/internal](https://grafana.com/grafana/dashboards/24585): add Grafana dashboard for monitoring VictoriaLogs internal state. The source of the dashboard is available [here](https://github.com/VictoriaMetrics/VictoriaLogs/blob/master/dashboards/victorialogs-internal.json).
     * FEATURE: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): add `pattern_match_prefix()` and `pattern_match_suffx()` filters for matching the given pattern at the beginning or at the end of the log field value. See [these docs](https://docs.victoriametrics.com/victorialogs/logsql/#pattern-match-filter) and [#762](https://github.com/VictoriaMetrics/VictoriaLogs/issues/762).
+    * FEATURE: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/): improve scalability for the data ingestion performance on systems with many CPU cores. See [#1021](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1021).
     * FEATURE: [Kubernetes Collector](https://docs.victoriametrics.com/victorialogs/vlagent/#collect-kubernetes-pod-logs): add an ability to change default [`_stream`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) fields via `-kubernetesCollector.streamFields` command-line flag. See [#998](https://github.com/VictoriaMetrics/VictoriaLogs/issues/998).
     * FEATURE: [querying HTTP API](https://docs.victoriametrics.com/victorialogs/querying/#http-api): accept `step` arg with nanosecond precision at [`/select/logsql/hits`](https://docs.victoriametrics.com/victorialogs/querying/#querying-hits-stats) and [`/select/logsql/stats_query_range`](https://docs.victoriametrics.com/victorialogs/querying/#querying-log-range-stats) endpoints. Previously these endpoints were accepting the `step` query arg only with millisecond precision.
     * FEATURE: [web UI](https://docs.victoriametrics.com/victorialogs/querying/#web-ui): add pipe titles with links to [docs.victoriametrics.com](https://docs.victoriametrics.com/victorialogs/logsql/) in autocomplete popup.
@@ Expand Down @@

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

app/vlinsert/insertutil: optimize LogMessageProcessor #1021

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!