Compare commits

..

1 Commits

Author SHA1 Message Date
Zakhar Bessarab
c188d5cf08 lib/promrelabel/debug: make metric name easier to distinguish from labels
Previously, users could get confused in cases when metric name which is valid for MetricsQL but not valid for Prom was used in relabeling debug. See #8584 for example.

This adds highlight to a metric name and prints empty labels set in order to explicitly show metric name and set of labels.

Signed-off-by: Zakhar Bessarab <z.bessarab@victoriametrics.com>
2025-03-27 15:41:55 +04:00
1968 changed files with 122889 additions and 115843 deletions

View File

@@ -5,10 +5,10 @@ body:
- type: markdown
attributes:
value: |
Before filling a bug report it would be great to [upgrade](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-upgrade)
Before filling a bug report it would be great to [upgrade](https://docs.victoriametrics.com/#how-to-upgrade)
to [the latest available release](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest)
and verify whether the bug is reproducible there.
It's also recommended to read the [troubleshooting docs](https://docs.victoriametrics.com/victoriametrics/troubleshooting/) first.
It's also recommended to read the [troubleshooting docs](https://docs.victoriametrics.com/troubleshooting/) first.
- type: textarea
id: describe-the-bug
attributes:
@@ -64,8 +64,8 @@ body:
* [Grafana dashboard for VictoriaMetrics cluster](https://grafana.com/grafana/dashboards/11176)
See how to setup monitoring here:
* [monitoring for single-node VictoriaMetrics](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#monitoring)
* [monitoring for VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#monitoring)
* [monitoring for single-node VictoriaMetrics](https://docs.victoriametrics.com/#monitoring)
* [monitoring for VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/#monitoring)
validations:
required: false
- type: textarea

View File

@@ -24,9 +24,9 @@ body:
label: Troubleshooting docs
description: I am familiar with the following troubleshooting docs
options:
- label: General - https://docs.victoriametrics.com/victoriametrics/troubleshooting/
- label: General - https://docs.victoriametrics.com/troubleshooting/
required: false
- label: vmagent - https://docs.victoriametrics.com/victoriametrics/vmagent/#troubleshooting
- label: vmagent - https://docs.victoriametrics.com/vmagent/#troubleshooting
required: false
- label: vmalert - https://docs.victoriametrics.com/victoriametrics/vmalert/#troubleshooting
- label: vmalert - https://docs.victoriametrics.com/vmalert/#troubleshooting
required: false

View File

@@ -6,4 +6,4 @@ Please provide a brief description of the changes you made. Be as specific as po
The following checks are **mandatory**:
- [ ] My change adheres to [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/victoriametrics/contributing/#pull-request-checklist).
- [ ] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).

1
.gitignore vendored
View File

@@ -27,4 +27,3 @@ _site
coverage.txt
cspell.json
*~
deployment/docker/provisioning/plugins/

View File

@@ -1 +1 @@
The document has been moved [here](https://docs.victoriametrics.com/victoriametrics/contributing/).
The document has been moved [here](https://docs.victoriametrics.com/contributing/).

View File

@@ -5,6 +5,7 @@ MAKE_PARALLEL := $(MAKE) -j $(MAKE_CONCURRENCY)
DATEINFO_TAG ?= $(shell date -u +'%Y%m%d-%H%M%S')
BUILDINFO_TAG ?= $(shell echo $$(git describe --long --all | tr '/' '-')$$( \
git diff-index --quiet HEAD -- || echo '-dirty-'$$(git diff-index -u HEAD | openssl sha1 | cut -d' ' -f2 | cut -c 1-8)))
LATEST_TAG ?= latest
PKG_TAG ?= $(shell git tag -l --points-at HEAD)
ifeq ($(PKG_TAG),)
@@ -195,31 +196,12 @@ vmutils-crossbuild: \
vmutils-openbsd-amd64 \
vmutils-windows-amd64
publish-latest:
PKG_TAG=$(TAG) APP_NAME=victoria-metrics $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG) APP_NAME=vmagent $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG) APP_NAME=vmalert $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG) APP_NAME=vmalert-tool $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG) APP_NAME=vmauth $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG) APP_NAME=vmbackup $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG) APP_NAME=vmrestore $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG) APP_NAME=vmctl $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG)-cluster APP_NAME=vminsert $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG)-cluster APP_NAME=vmselect $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG)-cluster APP_NAME=vmstorage $(MAKE) publish-via-docker-latest && \
PKG_TAG=$(TAG)-enterprise APP_NAME=vmgateway $(MAKE) publish-via-docker-latest
PKG_TAG=$(TAG)-enterprise APP_NAME=vmbackupmanager $(MAKE) publish-via-docker-latest
publish-victoria-logs-latest:
PKG_TAG=$(TAG) APP_NAME=victoria-logs $(MAKE) publish-via-docker-latest
PKG_TAG=$(TAG) APP_NAME=vlogscli $(MAKE) publish-via-docker-latest
publish-release:
rm -rf bin/*
git checkout $(TAG) && $(MAKE) release && $(MAKE) publish && \
git checkout $(TAG)-cluster && $(MAKE) release && $(MAKE) publish && \
git checkout $(TAG)-enterprise && $(MAKE) release && $(MAKE) publish && \
git checkout $(TAG)-enterprise-cluster && $(MAKE) release && $(MAKE) publish
git checkout $(TAG) && $(MAKE) release && LATEST_TAG=stable $(MAKE) publish && \
git checkout $(TAG)-cluster && $(MAKE) release && LATEST_TAG=cluster-stable $(MAKE) publish && \
git checkout $(TAG)-enterprise && $(MAKE) release && LATEST_TAG=enterprise-stable $(MAKE) publish && \
git checkout $(TAG)-enterprise-cluster && $(MAKE) release && LATEST_TAG=enterprise-cluster-stable $(MAKE) publish
release:
$(MAKE_PARALLEL) \
@@ -522,7 +504,7 @@ fmt:
gofmt -l -w -s ./apptest
vet:
GOEXPERIMENT=synctest go vet ./lib/...
go vet ./lib/...
go vet ./app/...
go vet ./apptest/...
@@ -531,35 +513,35 @@ check-all: fmt vet golangci-lint govulncheck
clean-checkers: remove-golangci-lint remove-govulncheck
test:
GOEXPERIMENT=synctest go test ./lib/... ./app/...
go test ./lib/... ./app/...
test-race:
GOEXPERIMENT=synctest go test -race ./lib/... ./app/...
go test -race ./lib/... ./app/...
test-pure:
GOEXPERIMENT=synctest CGO_ENABLED=0 go test ./lib/... ./app/...
CGO_ENABLED=0 go test ./lib/... ./app/...
test-full:
GOEXPERIMENT=synctest go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
test-full-386:
GOEXPERIMENT=synctest GOARCH=386 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
GOARCH=386 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
integration-test: victoria-metrics vmagent vmalert vmauth vmctl vmbackup vmrestore victoria-logs
integration-test: victoria-metrics vmagent vmalert vmauth
go test ./apptest/... -skip="^TestCluster.*"
benchmark:
GOEXPERIMENT=synctest go test -bench=. ./lib/...
go test -bench=. ./lib/...
go test -bench=. ./app/...
benchmark-pure:
GOEXPERIMENT=synctest CGO_ENABLED=0 go test -bench=. ./lib/...
CGO_ENABLED=0 go test -bench=. ./lib/...
CGO_ENABLED=0 go test -bench=. ./app/...
vendor-update:
go get -u ./lib/...
go get -u ./app/...
go mod tidy -compat=1.24
go mod tidy -compat=1.23
go mod vendor
app-local:
@@ -582,7 +564,7 @@ install-qtc:
golangci-lint: install-golangci-lint
GOEXPERIMENT=synctest golangci-lint run
golangci-lint run
install-golangci-lint:
which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.64.7

View File

@@ -11,9 +11,9 @@
![Reddit](https://img.shields.io/reddit/subreddit-subscribers/VictoriaMetrics?style=flat&label=Join&labelColor=red&logoColor=white&logo=reddit&link=https%3A%2F%2Fwww.reddit.com%2Fr%2FVictoriaMetrics)
<picture>
<source srcset="docs/victoriametrics/logo_white.webp" media="(prefers-color-scheme: dark)">
<source srcset="docs/victoriametrics/logo.webp" media="(prefers-color-scheme: light)">
<img src="docs/victoriametrics/logo.webp" width="300" alt="VictoriaMetrics logo">
<source srcset="docs/logo_white.webp" media="(prefers-color-scheme: dark)">
<source srcset="docs/logo.webp" media="(prefers-color-scheme: light)">
<img src="docs/logo.webp" width="300" alt="VictoriaMetrics logo">
</picture>
VictoriaMetrics is a fast, cost-saving, and scalable solution for monitoring and managing time series data. It delivers high performance and reliability, making it an ideal choice for businesses of all sizes.
@@ -21,10 +21,10 @@ VictoriaMetrics is a fast, cost-saving, and scalable solution for monitoring and
Here are some resources and information about VictoriaMetrics:
- Documentation: [docs.victoriametrics.com](https://docs.victoriametrics.com)
- Case studies: [Grammarly, Roblox, Wix,...](https://docs.victoriametrics.com/victoriametrics/casestudies/).
- Case studies: [Grammarly, Roblox, Wix,...](https://docs.victoriametrics.com/casestudies/).
- Available: [Binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest), docker images [Docker Hub](https://hub.docker.com/r/victoriametrics/victoria-metrics/) and [Quay](https://quay.io/repository/victoriametrics/victoria-metrics), [Source code](https://github.com/VictoriaMetrics/VictoriaMetrics)
- Deployment types: [Single-node version](https://docs.victoriametrics.com/), [Cluster version](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/), and [Enterprise version](https://docs.victoriametrics.com/victoriametrics/enterprise/)
- Changelog: [CHANGELOG](https://docs.victoriametrics.com/victoriametrics/changelog/), and [How to upgrade](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-upgrade-victoriametrics)
- Deployment types: [Single-node version](https://docs.victoriametrics.com/), [Cluster version](https://docs.victoriametrics.com/cluster-victoriametrics/), and [Enterprise version](https://docs.victoriametrics.com/enterprise/)
- Changelog: [CHANGELOG](https://docs.victoriametrics.com/changelog/), and [How to upgrade](https://docs.victoriametrics.com/#how-to-upgrade-victoriametrics)
- Community: [Slack](https://slack.victoriametrics.com/), [X (Twitter)](https://x.com/VictoriaMetrics), [LinkedIn](https://www.linkedin.com/company/victoriametrics/), [YouTube](https://www.youtube.com/@VictoriaMetrics)
Yes, we open-source both the single-node VictoriaMetrics and the cluster version.
@@ -35,22 +35,22 @@ VictoriaMetrics is optimized for timeseries data, even when old time series are
* **Long-term storage for Prometheus** or as a drop-in replacement for Prometheus and Graphite in Grafana.
* **Powerful stream aggregation**: Can be used as a StatsD alternative.
* **Ideal for big data**: Works well with large amounts of time series data from APM, Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various [Enterprise workloads](https://docs.victoriametrics.com/victoriametrics/enterprise/).
* **Ideal for big data**: Works well with large amounts of time series data from APM, Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various [Enterprise workloads](https://docs.victoriametrics.com/enterprise/).
* **Query language**: Supports both PromQL and the more performant MetricsQL.
* **Easy to setup**: No dependencies, single [small binary](https://medium.com/@valyala/stripping-dependency-bloat-in-victoriametrics-docker-image-983fb5912b0d), configuration through command-line flags, but the default is also fine-tuned; backup and restore with [instant snapshots](https://medium.com/@valyala/how-victoriametrics-makes-instant-snapshots-for-multi-terabyte-time-series-data-e1f3fb0e0282).
* **Global query view**: Multiple Prometheus instances or any other data sources may ingest data into VictoriaMetrics and queried via a single query.
* **Various Protocols**: Support metric scraping, ingestion and backfilling in various protocol.
* [Prometheus exporters](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-scrape-prometheus-exporters-such-as-node-exporter), [Prometheus remote write API](https://docs.victoriametrics.com/victoriametrics/integrations/prometheus/), [Prometheus exposition format](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-import-data-in-prometheus-exposition-format).
* [InfluxDB line protocol](https://docs.victoriametrics.com/victoriametrics/integrations/influxdb/) over HTTP, TCP and UDP.
* [Graphite plaintext protocol](https://docs.victoriametrics.com/victoriametrics/integrations/graphite/#ingesting) with [tags](https://graphite.readthedocs.io/en/latest/tags.html#carbon).
* [OpenTSDB put message](https://docs.victoriametrics.com/victoriametrics/integrations/opentsdb/#sending-data-via-telnet).
* [HTTP OpenTSDB /api/put requests](https://docs.victoriametrics.com/victoriametrics/integrations/opentsdb/#sending-data-via-http).
* [JSON line format](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-import-data-in-json-line-format).
* [Arbitrary CSV data](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-import-csv-data).
* [Native binary format](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-import-data-in-native-format).
* [DataDog agent or DogStatsD](https://docs.victoriametrics.com/victoriametrics/integrations/datadog/).
* [NewRelic infrastructure agent](https://docs.victoriametrics.com/victoriametrics/integrations/newrelic/#sending-data-from-agent).
* [OpenTelemetry metrics format](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#sending-data-via-opentelemetry).
* [Prometheus exporters](https://docs.victoriametrics.com/#how-to-scrape-prometheus-exporters-such-as-node-exporter), [Prometheus remote write API](https://docs.victoriametrics.com/#prometheus-setup), [Prometheus exposition format](https://docs.victoriametrics.com/#how-to-import-data-in-prometheus-exposition-format).
* [InfluxDB line protocol](https://docs.victoriametrics.com/#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf) over HTTP, TCP and UDP.
* [Graphite plaintext protocol](https://docs.victoriametrics.com/#how-to-send-data-from-graphite-compatible-agents-such-as-statsd) with [tags](https://graphite.readthedocs.io/en/latest/tags.html#carbon).
* [OpenTSDB put message](https://docs.victoriametrics.com/#sending-data-via-telnet-put-protocol).
* [HTTP OpenTSDB /api/put requests](https://docs.victoriametrics.com/#sending-opentsdb-data-via-http-apiput-requests).
* [JSON line format](https://docs.victoriametrics.com/#how-to-import-data-in-json-line-format).
* [Arbitrary CSV data](https://docs.victoriametrics.com/#how-to-import-csv-data).
* [Native binary format](https://docs.victoriametrics.com/#how-to-import-data-in-native-format).
* [DataDog agent or DogStatsD](https://docs.victoriametrics.com/#how-to-send-data-from-datadog-agent).
* [NewRelic infrastructure agent](https://docs.victoriametrics.com/#how-to-send-data-from-newrelic-agent).
* [OpenTelemetry metrics format](https://docs.victoriametrics.com/#sending-data-via-opentelemetry).
* **NFS-based storages**: Supports storing data on NFS-based storages such as Amazon EFS, Google Filestore.
* And many other features such as metrics relabeling, cardinality limiter, etc.
@@ -62,9 +62,9 @@ In addition, the Enterprise version includes extra features:
- **Backup automation**: Automates regular backup procedures.
- **Multiple retentions**: Reducing storage costs by specifying different retentions for different datasets.
- **Downsampling**: Reducing storage costs and increasing performance for queries over historical data.
- **Stable releases** with long-term support lines ([LTS](https://docs.victoriametrics.com/victoriametrics/lts-releases/)).
- **Stable releases** with long-term support lines ([LTS](https://docs.victoriametrics.com/lts-releases/)).
- **Comprehensive support**: First-class consulting, feature requests and technical support provided by the core VictoriaMetrics dev team.
- Many other features, which you can read about on [the Enterprise page](https://docs.victoriametrics.com/victoriametrics/enterprise/).
- Many other features, which you can read about on [the Enterprise page](https://docs.victoriametrics.com/enterprise/).
[Contact us](mailto:info@victoriametrics.com) if you need enterprise support for VictoriaMetrics. Or you can request a free trial license [here](https://victoriametrics.com/products/enterprise/trial/), downloaded Enterprise binaries are available at [Github Releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest).
@@ -77,7 +77,7 @@ Some good benchmarks VictoriaMetrics achieved:
* **Minimal memory footprint**: handling millions of unique timeseries with [10x less RAM](https://medium.com/@valyala/insert-benchmarks-with-inch-influxdb-vs-victoriametrics-e31a41ae2893) than InfluxDB, up to [7x less RAM](https://valyala.medium.com/prometheus-vs-victoriametrics-benchmark-on-node-exporter-metrics-4ca29c75590f) than Prometheus, Thanos or Cortex.
* **Highly scalable and performance** for [data ingestion](https://medium.com/@valyala/high-cardinality-tsdb-benchmarks-victoriametrics-vs-timescaledb-vs-influxdb-13e6ee64dd6b) and [querying](https://medium.com/@valyala/when-size-matters-benchmarking-victoriametrics-vs-timescale-and-influxdb-6035811952d4), [20x outperforms](https://medium.com/@valyala/insert-benchmarks-with-inch-influxdb-vs-victoriametrics-e31a41ae2893) InfluxDB and TimescaleDB.
* **High data compression**: [70x more data points](https://medium.com/@valyala/when-size-matters-benchmarking-victoriametrics-vs-timescale-and-influxdb-6035811952d4) may be stored into limited storage than TimescaleDB, [7x less storage](https://valyala.medium.com/prometheus-vs-victoriametrics-benchmark-on-node-exporter-metrics-4ca29c75590f) space is required than Prometheus, Thanos or Cortex.
* **Reducing storage costs**: [10x more effective](https://docs.victoriametrics.com/victoriametrics/casestudies/#grammarly) than Graphite according to the Grammarly case study.
* **Reducing storage costs**: [10x more effective](https://docs.victoriametrics.com/casestudies/#grammarly) than Graphite according to the Grammarly case study.
* **A single-node VictoriaMetrics** can replace medium-sized clusters built with competing solutions such as Thanos, M3DB, Cortex, InfluxDB or TimescaleDB. See [VictoriaMetrics vs Thanos](https://medium.com/@valyala/comparing-thanos-to-victoriametrics-cluster-b193bea1683), [Measuring vertical scalability](https://medium.com/@valyala/measuring-vertical-scalability-for-time-series-databases-in-google-cloud-92550d78d8ae), [Remote write storage wars - PromCon 2019](https://promcon.io/2019-munich/talks/remote-write-storage-wars/).
* **Optimized for storage**: [Works well with high-latency IO](https://medium.com/@valyala/high-cardinality-tsdb-benchmarks-victoriametrics-vs-timescaledb-vs-influxdb-13e6ee64dd6b) and low IOPS (HDD and network storage in AWS, Google Cloud, Microsoft Azure, etc.).
@@ -93,7 +93,7 @@ Feel free asking any questions regarding VictoriaMetrics:
* [Telegram-ru](https://t.me/VictoriaMetrics_ru1)
* [Mastodon](https://mastodon.social/@victoriametrics/)
If you like VictoriaMetrics and want to contribute, then please [read these docs](https://docs.victoriametrics.com/victoriametrics/contributing/).
If you like VictoriaMetrics and want to contribute, then please [read these docs](https://docs.victoriametrics.com/contributing/).
## VictoriaMetrics Logo

View File

@@ -6,9 +6,9 @@ The following versions of VictoriaMetrics receive regular security fixes:
| Version | Supported |
|---------|--------------------|
| [latest release](https://docs.victoriametrics.com/victoriametrics/changelog/) | :white_check_mark: |
| v1.102.x [LTS line](https://docs.victoriametrics.com/victoriametrics/lts-releases/) | :white_check_mark: |
| v1.110.x [LTS line](https://docs.victoriametrics.com/victoriametrics/lts-releases/) | :white_check_mark: |
| [latest release](https://docs.victoriametrics.com/changelog/) | :white_check_mark: |
| v1.102.x [LTS line](https://docs.victoriametrics.com/lts-releases/) | :white_check_mark: |
| v1.110.x [LTS line](https://docs.victoriametrics.com/lts-releases/) | :white_check_mark: |
| other releases | :x: |
See [this page](https://victoriametrics.com/security/) for more details.

View File

@@ -8,7 +8,6 @@ import (
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlselect"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
@@ -45,13 +44,9 @@ func main() {
vlstorage.Init()
vlselect.Init()
insertutil.SetLogRowsStorage(&vlstorage.Storage{})
vlinsert.Init()
go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{
UseProxyProtocol: useProxyProtocol,
})
go httpserver.Serve(listenAddrs, useProxyProtocol, requestHandler)
logger.Infof("started VictoriaLogs in %.3f seconds; see https://docs.victoriametrics.com/victorialogs/", time.Since(startTime).Seconds())
pushmetrics.Init()

View File

@@ -32,7 +32,7 @@ var (
"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt . "+
"With enabled proxy protocol http server cannot serve regular /metrics endpoint. Use -pushmetrics.url for metrics pushing")
minScrapeInterval = flag.Duration("dedup.minScrapeInterval", 0, "Leave only the last sample in every time series per each discrete interval "+
"equal to -dedup.minScrapeInterval > 0. See also -streamAggr.dedupInterval and https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
"equal to -dedup.minScrapeInterval > 0. See also -streamAggr.dedupInterval and https://docs.victoriametrics.com/#deduplication")
dryRun = flag.Bool("dryRun", false, "Whether to check config files without running VictoriaMetrics. The following config files are checked: "+
"-promscrape.config, -relabelConfig and -streamAggr.config. Unknown config entries aren't allowed in -promscrape.config by default. "+
"This can be changed with -promscrape.config.strictParse=false command-line flag")
@@ -45,7 +45,7 @@ var (
finalDedupScheduleInterval = flag.Duration("storage.finalDedupScheduleCheckInterval", time.Hour, "The interval for checking when final deduplication process should be started."+
"Storage unconditionally adds 25% jitter to the interval value on each check evaluation."+
" Changing the interval to the bigger values may delay downsampling, deduplication for historical data."+
" See also https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
" See also https://docs.victoriametrics.com/#deduplication")
)
func main() {
@@ -101,9 +101,7 @@ func main() {
startSelfScraper()
go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{
UseProxyProtocol: useProxyProtocol,
})
go httpserver.Serve(listenAddrs, useProxyProtocol, requestHandler)
logger.Infof("started VictoriaMetrics in %.3f seconds", time.Since(startTime).Seconds())
pushmetrics.Init()

View File

@@ -0,0 +1,619 @@
package main
import (
"bytes"
"encoding/json"
"errors"
"flag"
"fmt"
"io"
"log"
"math/rand"
"net"
"net/http"
"os"
"path/filepath"
"reflect"
"strconv"
"strings"
"testing"
"time"
testutil "github.com/VictoriaMetrics/VictoriaMetrics/app/victoria-metrics/test"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
)
const (
testFixturesDir = "testdata"
testStorageSuffix = "vm-test-storage"
testHTTPListenAddr = ":7654"
testStatsDListenAddr = ":2003"
testOpenTSDBListenAddr = ":4242"
testOpenTSDBHTTPListenAddr = ":4243"
testLogLevel = "INFO"
)
const (
testReadHTTPPath = "http://127.0.0.1" + testHTTPListenAddr
testWriteHTTPPath = "http://127.0.0.1" + testHTTPListenAddr + "/write"
testOpenTSDBWriteHTTPPath = "http://127.0.0.1" + testOpenTSDBHTTPListenAddr + "/api/put"
testPromWriteHTTPPath = "http://127.0.0.1" + testHTTPListenAddr + "/api/v1/write"
testImportCSVWriteHTTPPath = "http://127.0.0.1" + testHTTPListenAddr + "/api/v1/import/csv"
testHealthHTTPPath = "http://127.0.0.1" + testHTTPListenAddr + "/health"
)
const (
testStorageInitTimeout = 10 * time.Second
)
var (
storagePath string
insertionTime = time.Now().UTC()
)
type test struct {
Name string `json:"name"`
Data []string `json:"data"`
InsertQuery string `json:"insert_query"`
Query []string `json:"query"`
ResultMetrics []Metric `json:"result_metrics"`
ResultSeries Series `json:"result_series"`
ResultQuery Query `json:"result_query"`
Issue string `json:"issue"`
ExpectedResultLinesCount int `json:"expected_result_lines_count"`
}
type Metric struct {
Metric map[string]string `json:"metric"`
Values []float64 `json:"values"`
Timestamps []int64 `json:"timestamps"`
}
func (r *Metric) UnmarshalJSON(b []byte) error {
type plain Metric
return json.Unmarshal(testutil.PopulateTimeTpl(b, insertionTime), (*plain)(r))
}
type Series struct {
Status string `json:"status"`
Data []map[string]string `json:"data"`
}
type Query struct {
Status string `json:"status"`
Data struct {
ResultType string `json:"resultType"`
Result json.RawMessage `json:"result"`
} `json:"data"`
}
const rtVector, rtMatrix = "vector", "matrix"
func (q *Query) metrics() ([]Metric, error) {
switch q.Data.ResultType {
case rtVector:
var r QueryInstant
if err := json.Unmarshal(q.Data.Result, &r.Result); err != nil {
return nil, err
}
return r.metrics()
case rtMatrix:
var r QueryRange
if err := json.Unmarshal(q.Data.Result, &r.Result); err != nil {
return nil, err
}
return r.metrics()
default:
return nil, fmt.Errorf("unknown result type %q", q.Data.ResultType)
}
}
type QueryInstant struct {
Result []struct {
Labels map[string]string `json:"metric"`
TV [2]any `json:"value"`
} `json:"result"`
}
func (q QueryInstant) metrics() ([]Metric, error) {
result := make([]Metric, len(q.Result))
for i, res := range q.Result {
f, err := strconv.ParseFloat(res.TV[1].(string), 64)
if err != nil {
return nil, fmt.Errorf("metric %v, unable to parse float64 from %s: %w", res, res.TV[1], err)
}
var m Metric
m.Metric = res.Labels
m.Timestamps = append(m.Timestamps, int64(res.TV[0].(float64)))
m.Values = append(m.Values, f)
result[i] = m
}
return result, nil
}
type QueryRange struct {
Result []struct {
Metric map[string]string `json:"metric"`
Values [][]any `json:"values"`
} `json:"result"`
}
func (q QueryRange) metrics() ([]Metric, error) {
var result []Metric
for i, res := range q.Result {
var m Metric
for _, tv := range res.Values {
f, err := strconv.ParseFloat(tv[1].(string), 64)
if err != nil {
return nil, fmt.Errorf("metric %v, unable to parse float64 from %s: %w", res, tv[1], err)
}
m.Values = append(m.Values, f)
m.Timestamps = append(m.Timestamps, int64(tv[0].(float64)))
}
if len(m.Values) < 1 || len(m.Timestamps) < 1 {
return nil, fmt.Errorf("metric %v contains no values", res)
}
m.Metric = q.Result[i].Metric
result = append(result, m)
}
return result, nil
}
func (q *Query) UnmarshalJSON(b []byte) error {
type plain Query
return json.Unmarshal(testutil.PopulateTimeTpl(b, insertionTime), (*plain)(q))
}
func TestMain(m *testing.M) {
setUp()
code := m.Run()
tearDown()
os.Exit(code)
}
func setUp() {
storagePath = filepath.Join(os.TempDir(), testStorageSuffix)
processFlags()
logger.Init()
vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
vmselect.Init()
vminsert.Init()
go httpserver.Serve(*httpListenAddrs, useProxyProtocol, requestHandler)
readyStorageCheckFunc := func() bool {
resp, err := http.Get(testHealthHTTPPath)
if err != nil {
return false
}
_ = resp.Body.Close()
return resp.StatusCode == 200
}
if err := waitFor(testStorageInitTimeout, readyStorageCheckFunc); err != nil {
log.Fatalf("http server can't start for %s seconds, err %s", testStorageInitTimeout, err)
}
}
func processFlags() {
flag.Parse()
for _, fv := range []struct {
flag string
value string
}{
{flag: "storageDataPath", value: storagePath},
{flag: "httpListenAddr", value: testHTTPListenAddr},
{flag: "graphiteListenAddr", value: testStatsDListenAddr},
{flag: "opentsdbListenAddr", value: testOpenTSDBListenAddr},
{flag: "loggerLevel", value: testLogLevel},
{flag: "opentsdbHTTPListenAddr", value: testOpenTSDBHTTPListenAddr},
} {
// panics if flag doesn't exist
if err := flag.Lookup(fv.flag).Value.Set(fv.value); err != nil {
log.Fatalf("unable to set %q with value %q, err: %v", fv.flag, fv.value, err)
}
}
}
func waitFor(timeout time.Duration, f func() bool) error {
fraction := timeout / 10
for i := fraction; i < timeout; i += fraction {
if f() {
return nil
}
time.Sleep(fraction)
}
return fmt.Errorf("timeout")
}
func tearDown() {
if err := httpserver.Stop(*httpListenAddrs); err != nil {
log.Printf("cannot stop the webservice: %s", err)
}
vminsert.Stop()
vmstorage.Stop()
vmselect.Stop()
fs.MustRemoveAll(storagePath)
}
func TestWriteRead(t *testing.T) {
t.Run("write", testWrite)
time.Sleep(500 * time.Millisecond)
vmstorage.Storage.DebugFlush()
time.Sleep(1500 * time.Millisecond)
t.Run("read", testRead)
}
func testWrite(t *testing.T) {
t.Run("prometheus", func(t *testing.T) {
for _, test := range readIn("prometheus", insertionTime) {
if test.Data == nil {
continue
}
r := testutil.WriteRequest{}
testData := strings.Join(test.Data, "\n")
if err := json.Unmarshal([]byte(testData), &r.Timeseries); err != nil {
panic(fmt.Errorf("BUG: cannot unmarshal TimeSeries: %s\ntest data\n%s", err, testData))
}
if n := len(r.Timeseries); n <= 0 {
panic(fmt.Errorf("BUG: expecting non-empty Timeseries in test data:\n%s", testData))
}
data := testutil.Compress(r)
httpWrite(t, testPromWriteHTTPPath, test.InsertQuery, bytes.NewBuffer(data))
}
})
t.Run("csv", func(t *testing.T) {
for _, test := range readIn("csv", insertionTime) {
if test.Data == nil {
continue
}
httpWrite(t, testImportCSVWriteHTTPPath, test.InsertQuery, bytes.NewBuffer([]byte(strings.Join(test.Data, "\n"))))
}
})
t.Run("influxdb", func(t *testing.T) {
for _, x := range readIn("influxdb", insertionTime) {
test := x
t.Run(test.Name, func(t *testing.T) {
t.Parallel()
httpWrite(t, testWriteHTTPPath, test.InsertQuery, bytes.NewBufferString(strings.Join(test.Data, "\n")))
})
}
})
t.Run("graphite", func(t *testing.T) {
for _, x := range readIn("graphite", insertionTime) {
test := x
t.Run(test.Name, func(t *testing.T) {
t.Parallel()
tcpWrite(t, "127.0.0.1"+testStatsDListenAddr, strings.Join(test.Data, "\n"))
})
}
})
t.Run("opentsdb", func(t *testing.T) {
for _, x := range readIn("opentsdb", insertionTime) {
test := x
t.Run(test.Name, func(t *testing.T) {
t.Parallel()
tcpWrite(t, "127.0.0.1"+testOpenTSDBListenAddr, strings.Join(test.Data, "\n"))
})
}
})
t.Run("opentsdbhttp", func(t *testing.T) {
for _, x := range readIn("opentsdbhttp", insertionTime) {
test := x
t.Run(test.Name, func(t *testing.T) {
t.Parallel()
logger.Infof("writing %s", test.Data)
httpWrite(t, testOpenTSDBWriteHTTPPath, test.InsertQuery, bytes.NewBufferString(strings.Join(test.Data, "\n")))
})
}
})
}
func testRead(t *testing.T) {
for _, engine := range []string{"csv", "prometheus", "graphite", "opentsdb", "influxdb", "opentsdbhttp"} {
t.Run(engine, func(t *testing.T) {
for _, x := range readIn(engine, insertionTime) {
test := x
t.Run(test.Name, func(t *testing.T) {
t.Parallel()
for _, q := range test.Query {
q = testutil.PopulateTimeTplString(q, insertionTime)
if test.Issue != "" {
test.Issue = "\nRegression in " + test.Issue
}
switch {
case strings.HasPrefix(q, "/api/v1/export/csv"):
data := strings.Split(string(httpReadData(t, testReadHTTPPath, q)), "\n")
if len(data) == test.ExpectedResultLinesCount {
t.Fatalf("not expected number of csv lines want=%d\ngot=%d test=%s.%s\n\response=%q", len(data), test.ExpectedResultLinesCount, q, test.Issue, strings.Join(data, "\n"))
}
case strings.HasPrefix(q, "/api/v1/export"):
if err := checkMetricsResult(httpReadMetrics(t, testReadHTTPPath, q), test.ResultMetrics); err != nil {
t.Fatalf("Export. %s fails with error %s.%s", q, err, test.Issue)
}
case strings.HasPrefix(q, "/api/v1/series"):
s := Series{}
httpReadStruct(t, testReadHTTPPath, q, &s)
if err := checkSeriesResult(s, test.ResultSeries); err != nil {
t.Fatalf("Series. %s fails with error %s.%s", q, err, test.Issue)
}
case strings.HasPrefix(q, "/api/v1/query"):
queryResult := Query{}
httpReadStruct(t, testReadHTTPPath, q, &queryResult)
gotMetrics, err := queryResult.metrics()
if err != nil {
t.Fatalf("failed to parse query response: %s", err)
}
expMetrics, err := test.ResultQuery.metrics()
if err != nil {
t.Fatalf("failed to parse expected response: %s", err)
}
if err := checkMetricsResult(gotMetrics, expMetrics); err != nil {
t.Fatalf("%q fails with error %s.%s", q, err, test.Issue)
}
default:
t.Fatalf("unsupported read query %s", q)
}
}
})
}
})
}
}
func readIn(readFor string, insertTime time.Time) []test {
testDir := filepath.Join(testFixturesDir, readFor)
var tt []test
err := filepath.Walk(testDir, func(path string, _ os.FileInfo, err error) error {
if err != nil {
return err
}
if filepath.Ext(path) != ".json" {
return nil
}
b, err := os.ReadFile(path)
if err != nil {
panic(fmt.Errorf("BUG: cannot read %s: %s", path, err))
}
item := test{}
if err := json.Unmarshal(b, &item); err != nil {
panic(fmt.Errorf("cannot parse %T from %s: %s; data:\n%s", &item, path, err, b))
}
for i := range item.Data {
item.Data[i] = testutil.PopulateTimeTplString(item.Data[i], insertTime)
}
tt = append(tt, item)
return nil
})
if err != nil {
panic(fmt.Errorf("BUG: cannot read test data at %s: %w", testDir, err))
}
if len(tt) == 0 {
panic(fmt.Errorf("BUG: no tests found in %s", testDir))
}
return tt
}
func httpWrite(t *testing.T, address, query string, r io.Reader) {
t.Helper()
requestURL := address + query
resp, err := http.Post(requestURL, "", r)
if err != nil {
t.Fatalf("cannot send request to %s: %s", requestURL, err)
}
_ = resp.Body.Close()
if resp.StatusCode != 204 {
t.Fatalf("unexpected status code received from %s; got %d; want 204", requestURL, resp.StatusCode)
}
}
func tcpWrite(t *testing.T, address, data string) {
t.Helper()
conn, err := net.Dial("tcp", address)
if err != nil {
t.Fatalf("cannot dial %s: %s", address, err)
}
defer func() {
_ = conn.Close()
}()
n, err := conn.Write([]byte(data))
if err != nil {
t.Fatalf("cannot write %d bytes to %s: %s", len(data), address, err)
}
if n != len(data) {
panic(fmt.Errorf("BUG: conn.Write() returned unexpected number of written bytes to %s; got %d; want %d", address, n, len(data)))
}
}
func httpReadMetrics(t *testing.T, address, query string) []Metric {
t.Helper()
requestURL := address + query
resp, err := http.Get(requestURL)
if err != nil {
t.Fatalf("cannot send request to %s: %s", requestURL, err)
}
defer func() {
_ = resp.Body.Close()
}()
if resp.StatusCode != 200 {
t.Fatalf("unexpected status code received from %s; got %d; want 200", requestURL, resp.StatusCode)
}
var rows []Metric
dec := json.NewDecoder(resp.Body)
for {
var row Metric
err := dec.Decode(&row)
if err != nil {
if errors.Is(err, io.EOF) {
return rows
}
t.Fatalf("cannot decode %T from response received from %s: %s", &row, requestURL, err)
}
rows = append(rows, row)
}
}
func httpReadStruct(t *testing.T, address, query string, dst any) {
t.Helper()
requestURL := address + query
resp, err := http.Get(requestURL)
if err != nil {
t.Fatalf("cannot send request to %s: %s", requestURL, err)
}
defer func() {
_ = resp.Body.Close()
}()
if resp.StatusCode != 200 {
t.Fatalf("unexpected status code received from %s; got %d; want 200", requestURL, resp.StatusCode)
}
err = json.NewDecoder(resp.Body).Decode(dst)
if err != nil {
t.Fatalf("cannot decode %T from response received from %s: %s", dst, requestURL, err)
}
}
func httpReadData(t *testing.T, address, query string) []byte {
t.Helper()
requestURL := address + query
resp, err := http.Get(requestURL)
if err != nil {
t.Fatalf("cannot send request to %s: %s", requestURL, err)
}
defer func() {
_ = resp.Body.Close()
}()
if resp.StatusCode != 200 {
t.Fatalf("unexpected status code received from %s; got %d; want 200", requestURL, resp.StatusCode)
}
data, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("cannot read response from %s: %s", requestURL, err)
}
return data
}
func checkMetricsResult(got, want []Metric) error {
for _, r := range append([]Metric(nil), got...) {
want = removeIfFoundMetrics(r, want)
}
if len(want) > 0 {
return fmt.Errorf("expected metrics %+v not found in %+v", want, got)
}
return nil
}
func removeIfFoundMetrics(r Metric, contains []Metric) []Metric {
for i, item := range contains {
if reflect.DeepEqual(r.Metric, item.Metric) && reflect.DeepEqual(r.Values, item.Values) &&
reflect.DeepEqual(r.Timestamps, item.Timestamps) {
contains[i] = contains[len(contains)-1]
return contains[:len(contains)-1]
}
}
return contains
}
func checkSeriesResult(got, want Series) error {
if got.Status != want.Status {
return fmt.Errorf("status mismatch %q - %q", want.Status, got.Status)
}
wantData := append([]map[string]string(nil), want.Data...)
for _, r := range got.Data {
wantData = removeIfFoundSeries(r, wantData)
}
if len(wantData) > 0 {
return fmt.Errorf("expected seria(s) %+v not found in %+v", wantData, got.Data)
}
return nil
}
func removeIfFoundSeries(r map[string]string, contains []map[string]string) []map[string]string {
for i, item := range contains {
if reflect.DeepEqual(r, item) {
contains[i] = contains[len(contains)-1]
return contains[:len(contains)-1]
}
}
return contains
}
func TestImportJSONLines(t *testing.T) {
f := func(labelsCount, labelLen int) {
t.Helper()
reqURL := fmt.Sprintf("http://localhost%s/api/v1/import", testHTTPListenAddr)
line := generateJSONLine(labelsCount, labelLen)
req, err := http.NewRequest("POST", reqURL, bytes.NewBufferString(line))
if err != nil {
t.Fatalf("cannot create request: %s", err)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("cannot perform request for labelsCount=%d, labelLen=%d: %s", labelsCount, labelLen, err)
}
if resp.StatusCode != 204 {
t.Fatalf("unexpected statusCode for labelsCount=%d, labelLen=%d; got %d; want 204", labelsCount, labelLen, resp.StatusCode)
}
}
// labels with various lengths
for i := 0; i < 500; i++ {
f(10, i*5)
}
// Too many labels
f(1000, 100)
// Too long labels
f(1, 100_000)
f(10, 100_000)
f(10, 10_000)
}
func generateJSONLine(labelsCount, labelLen int) string {
m := make(map[string]string, labelsCount)
m["__name__"] = generateSizedRandomString(labelLen)
for j := 1; j < labelsCount; j++ {
labelName := generateSizedRandomString(labelLen)
labelValue := generateSizedRandomString(labelLen)
m[labelName] = labelValue
}
type jsonLine struct {
Metric map[string]string `json:"metric"`
Values []float64 `json:"values"`
Timestamps []int64 `json:"timestamps"`
}
line := &jsonLine{
Metric: m,
Values: []float64{1.34},
Timestamps: []int64{time.Now().UnixNano() / 1e6},
}
data, err := json.Marshal(&line)
if err != nil {
panic(fmt.Errorf("cannot marshal JSON: %w", err))
}
data = append(data, '\n')
return string(data)
}
const alphabetSample = `qwertyuiopasdfghjklzxcvbnm`
func generateSizedRandomString(size int) string {
dst := make([]byte, size)
for i := range dst {
dst[i] = alphabetSample[rand.Intn(len(alphabetSample))]
}
return string(dst)
}

View File

@@ -9,5 +9,4 @@ COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certifica
EXPOSE 8428
ENTRYPOINT ["/victoria-metrics-prod"]
ARG TARGETARCH
ARG BINARY_SUFFIX=non-existing
COPY victoria-metrics-linux-${TARGETARCH}-prod${BINARY_SUFFIX} ./victoria-metrics-prod
COPY victoria-metrics-linux-${TARGETARCH}-prod ./victoria-metrics-prod

View File

@@ -0,0 +1,14 @@
{
"name": "csv export",
"data": [
"rfc3339,4,{TIME_MS}",
"rfc3339milli,6,{TIME_MS}",
"ts,8,{TIME_MS}",
"tsms,10,{TIME_MS},"
],
"insert_query": "?format=1:label:tfmt,2:metric:test_csv,3:time:unix_ms",
"query": [
"/api/v1/export/csv?format=__name__,tfmt,__value__,__timestamp__:rfc3339&match[]={__name__=\"test_csv\"}&step=30s&start={TIME_MS-180s}"
],
"expected_result_lines_count": 4
}

View File

@@ -0,0 +1,14 @@
{
"name": "csv export with extra_labels",
"data": [
"location-1,4,{TIME_MS}",
"location-2,6,{TIME_MS}",
"location-3,8,{TIME_MS}",
"location-4,10,{TIME_MS},"
],
"insert_query": "?format=1:label:location,2:metric:test_csv_labels,3:time:unix_ms&extra_label=location=location-1",
"query": [
"/api/v1/export/csv?format=__name__,location,__value__,__timestamp__:unix_ms&match[]={__name__=\"test_csv\"}&step=30s&start={TIME_MS-180s}"
],
"expected_result_lines_count": 4
}

View File

@@ -0,0 +1,8 @@
{
"name": "basic_insertion",
"data": ["graphite.foo.bar.baz;tag1=value1;tag2=value2 123 {TIME_S}"],
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"graphite.foo.bar.baz","tag1":"value1","tag2":"value2"},"values":[123], "timestamps": ["{TIME_MSZ}"]}
]
}

View File

@@ -0,0 +1,16 @@
{
"name": "comparison-not-inf-not-nan",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/150",
"data": [
"not_nan_not_inf;item=x 1 {TIME_S-1m}",
"not_nan_not_inf;item=x 1 {TIME_S-2m}",
"not_nan_not_inf;item=y 3 {TIME_S-1m}",
"not_nan_not_inf;item=y 1 {TIME_S-2m}"],
"query": ["/api/v1/query_range?query=1/(not_nan_not_inf-1)!=inf!=nan&start={TIME_S-3m}&end={TIME_S}&step=60"],
"result_query": {
"status":"success",
"data":{"resultType":"matrix",
"result":[
{"metric":{"item":"y"},"values":[["{TIME_S-1m}","0.5"], ["{TIME_S}","0.5"]]}
]}}
}

View File

@@ -0,0 +1,16 @@
{
"name": "empty-label-match",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/395",
"data": [
"empty_label_match 1 {TIME_S-1m}",
"empty_label_match;foo=bar 2 {TIME_S-1m}",
"empty_label_match;foo=baz 3 {TIME_S-1m}"],
"query": ["/api/v1/query_range?query=empty_label_match{foo=~'bar|'}&start={TIME_S-1m}&end={TIME_S}&step=60"],
"result_query": {
"status":"success",
"data":{"resultType":"matrix",
"result":[
{"metric":{"__name__":"empty_label_match"},"values":[["{TIME_S-1m}","1"],["{TIME_S}","1"]]},
{"metric":{"__name__":"empty_label_match","foo":"bar"},"values":[["{TIME_S-1m}","2"],["{TIME_S}","2"]]}
]}}
}

View File

@@ -0,0 +1,17 @@
{
"name": "graphite-selector",
"issue": "",
"data": [
"graphite-selector.bar.baz 1 {TIME_S-1m}",
"graphite-selector.xxx.yy 2 {TIME_S-1m}",
"graphite-selector.bb.cc 3 {TIME_S-1m}",
"graphite-selector.a.baz 4 {TIME_S-1m}"],
"query": ["/api/v1/query?query=sort({__graphite__='graphite-selector.*.baz'})&time={TIME_S-1m}"],
"result_query": {
"status":"success",
"data":{"resultType":"vector","result":[
{"metric":{"__name__":"graphite-selector.bar.baz"},"value":["{TIME_S-1m}","1"]},
{"metric":{"__name__":"graphite-selector.a.baz"},"value":["{TIME_S-1m}","4"]}
]}
}
}

View File

@@ -0,0 +1,23 @@
{
"name": "max_lookback_set",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/209",
"data": [
"max_lookback_set 1 {TIME_S-30s}",
"max_lookback_set 2 {TIME_S-60s}",
"max_lookback_set 3 {TIME_S-120s}",
"max_lookback_set 4 {TIME_S-150s}"
],
"query": ["/api/v1/query_range?query=max_lookback_set&start={TIME_S-150s}&end={TIME_S}&step=10s&max_lookback=1s"],
"result_query": {
"status":"success",
"data":{"resultType":"matrix",
"result":[{"metric":{"__name__":"max_lookback_set"},"values":[
["{TIME_S-150s}","4"],
["{TIME_S-120s}","3"],
["{TIME_S-60s}","2"],
["{TIME_S-30s}","1"],
["{TIME_S-20s}","1"],
["{TIME_S-10s}","1"],
["{TIME_S-0s}","1"]
]}]}}
}

View File

@@ -0,0 +1,31 @@
{
"name": "max_lookback_unset",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/209",
"data": [
"max_lookback_unset 1 {TIME_S-30s}",
"max_lookback_unset 2 {TIME_S-60s}",
"max_lookback_unset 3 {TIME_S-120s}",
"max_lookback_unset 4 {TIME_S-150s}"
],
"query": ["/api/v1/query_range?query=max_lookback_unset&start={TIME_S-150s}&end={TIME_S}&step=10s"],
"result_query": {
"status":"success",
"data":{"resultType":"matrix",
"result":[{"metric":{"__name__":"max_lookback_unset"},"values":[
["{TIME_S-150s}","4"],
["{TIME_S-140s}","4"],
["{TIME_S-130s}","4"],
["{TIME_S-120s}","3"],
["{TIME_S-110s}","3"],
["{TIME_S-100s}","3"],
["{TIME_S-90s}","3"],
["{TIME_S-80s}","3"],
["{TIME_S-60s}","2"],
["{TIME_S-50s}","2"],
["{TIME_S-40s}","2"],
["{TIME_S-30s}","1"],
["{TIME_S-20s}","1"],
["{TIME_S-10s}","1"],
["{TIME_S-0s}","1"]
]}]}}
}

View File

@@ -0,0 +1,16 @@
{
"name": "name-plus-negative-filter",
"issue": "",
"data": [
"name-plus-negative-filter;foo=123 1 {TIME_S-1m}",
"name-plus-negative-filter;bar=123 2 {TIME_S-1m}",
"name-plus-negative-filter;foo=qwe 3 {TIME_S-1m}"
],
"query": ["/api/v1/query?query={__name__='name-plus-negative-filter',foo!='123'}&time={TIME_S-1m}"],
"result_query": {
"status":"success",
"data":{"resultType":"vector","result":[
{"metric":{"__name__":"name-plus-negative-filter","foo":"qwe"},"value":["{TIME_S-1m}","3"]}
]}
}
}

View File

@@ -0,0 +1,18 @@
{
"name": "not-nan-as-missing-data",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/153",
"data": [
"not_nan_as_missing_data;item=x 2 {TIME_S-2m}",
"not_nan_as_missing_data;item=x 1 {TIME_S-1m}",
"not_nan_as_missing_data;item=y 4 {TIME_S-2m}",
"not_nan_as_missing_data;item=y 3 {TIME_S-1m}"
],
"query": ["/api/v1/query_range?query=not_nan_as_missing_data>1&start={TIME_S-2m}&end={TIME_S}&step=60"],
"result_query": {
"status":"success",
"data":{"resultType":"matrix",
"result":[
{"metric":{"__name__":"not_nan_as_missing_data","item":"x"},"values":[["{TIME_S-2m}","2"]]},
{"metric":{"__name__":"not_nan_as_missing_data","item":"y"},"values":[["{TIME_S-2m}","4"],["{TIME_S-1m}","3"],["{TIME_S}", "3"]]}
]}}
}

View File

@@ -0,0 +1,14 @@
{
"name": "subquery-aggregation",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/184",
"data": [
"forms_daily_count;item=x 1 {TIME_S-59s}",
"forms_daily_count;item=x 2 {TIME_S-1m59s}",
"forms_daily_count;item=y 3 {TIME_S-59s}",
"forms_daily_count;item=y 4 {TIME_S-1m59s}"],
"query": ["/api/v1/query?query=min%20by%20(item)%20(min_over_time(forms_daily_count[10m:1m]))&time={TIME_S-1m}&latency_offset=1ms"],
"result_query": {
"status":"success",
"data":{"resultType":"vector","result":[{"metric":{"item":"x"},"value":["{TIME_S-1m}","2"]},{"metric":{"item":"y"},"value":["{TIME_S-1m}","4"]}]}
}
}

View File

@@ -0,0 +1,9 @@
{
"name": "basic_insertion",
"data": ["measurement,tag1=value1,tag2=value2 field1=1.23,field2=123 {TIME_NS}"],
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"measurement_field2","tag1":"value1","tag2":"value2"},"values":[123], "timestamps": ["{TIME_MS}"]},
{"metric":{"__name__":"measurement_field1","tag1":"value1","tag2":"value2"},"values":[1.23], "timestamps": ["{TIME_MS}"]}
]
}

View File

@@ -0,0 +1,10 @@
{
"name": "insert_with_extra_labels",
"data": ["measurement,tag1=value1,tag2=value2 field6=1.23,field5=123 {TIME_NS}"],
"insert_query": "?extra_label=job=test&extra_label=tag2=value10",
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"measurement_field5","tag1":"value1","job": "test","tag2":"value10"},"values":[123], "timestamps": ["{TIME_MS}"]},
{"metric":{"__name__":"measurement_field6","tag1":"value1","job": "test","tag2":"value10"},"values":[1.23], "timestamps": ["{TIME_MS}"]}
]
}

View File

@@ -0,0 +1,8 @@
{
"name": "basic_insertion",
"data": ["put openstdb.foo.bar.baz {TIME_S} 123 tag1=value1 tag2=value2"],
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"openstdb.foo.bar.baz","tag1":"value1","tag2":"value2"},"values":[123], "timestamps": ["{TIME_MSZ}"]}
]
}

View File

@@ -0,0 +1,8 @@
{
"name": "basic_insertion",
"data": ["{\"metric\": \"opentsdbhttp.foo\", \"value\": 1001, \"timestamp\": {TIME_S}, \"tags\": {\"bar\":\"baz\", \"x\": \"y\"}}"],
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"opentsdbhttp.foo","bar":"baz","x":"y"},"values":[1001], "timestamps": ["{TIME_MSZ}"]}
]
}

View File

@@ -0,0 +1,9 @@
{
"name": "multiline",
"data": ["[{\"metric\": \"opentsdbhttp.multiline1\", \"value\": 1001, \"timestamp\": \"{TIME_S}\", \"tags\": {\"bar\":\"baz\", \"x\": \"y\"}}, {\"metric\": \"opentsdbhttp.multiline2\", \"value\": 1002, \"timestamp\": {TIME_S}}]"],
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"opentsdbhttp.multiline1","bar":"baz","x":"y"},"values":[1001], "timestamps": ["{TIME_MSZ}"]},
{"metric":{"__name__":"opentsdbhttp.multiline2"},"values":[1002], "timestamps": ["{TIME_MSZ}"]}
]
}

View File

@@ -0,0 +1,9 @@
{
"name": "insert_with_extra_labels",
"data": ["{\"metric\": \"opentsdbhttp.foobar\", \"value\": 1001, \"timestamp\": {TIME_S}, \"tags\": {\"bar\":\"baz\", \"x\": \"y\"}}"],
"insert_query": "?extra_label=job=open-test&extra_label=x=z",
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"opentsdbhttp.foobar","bar":"baz","x":"z","job": "open-test"},"values":[1001], "timestamps": ["{TIME_MSZ}"]}
]
}

View File

@@ -0,0 +1,8 @@
{
"name": "basic_insertion",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.bar\"},{\"name\":\"baz\",\"value\":\"qux\"}],\"samples\":[{\"value\":100000,\"timestamp\":\"{TIME_MS}\"}]}]"],
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"prometheus.bar","baz":"qux"},"values":[100000], "timestamps": ["{TIME_MS}"]}
]
}

View File

@@ -0,0 +1,10 @@
{
"name": "case-sensitive-regex",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/161",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.sensitiveRegex\"},{\"name\":\"label\",\"value\":\"sensitiveRegex\"}],\"samples\":[{\"value\":2,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.sensitiveRegex\"},{\"name\":\"label\",\"value\":\"SensitiveRegex\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]}]"],
"query": ["/api/v1/export?match={label=~'(?i)sensitiveregex'}"],
"result_metrics": [
{"metric":{"__name__":"prometheus.sensitiveRegex","label":"sensitiveRegex"},"values":[2], "timestamps": ["{TIME_MS}"]},
{"metric":{"__name__":"prometheus.sensitiveRegex","label":"SensitiveRegex"},"values":[1], "timestamps": ["{TIME_MS}"]}
]
}

View File

@@ -0,0 +1,9 @@
{
"name": "duplicate_label",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/172",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.duplicate_label\"},{\"name\":\"duplicate\",\"value\":\"label\"},{\"name\":\"duplicate\",\"value\":\"label\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]}]"],
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"prometheus.duplicate_label","duplicate":"label"},"values":[1], "timestamps": ["{TIME_MS}"]}
]
}

View File

@@ -0,0 +1,12 @@
{
"name": "instant query with look-behind window",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"foo\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS-60s}\"}]}]"],
"query": ["/api/v1/query?query=foo[5m]"],
"result_query": {
"status": "success",
"data":{
"resultType":"matrix",
"result":[{"metric":{"__name__":"foo"},"values":[["{TIME_S-60s}", "1"]]}]
}
}
}

View File

@@ -0,0 +1,11 @@
{
"name": "instant scalar query",
"query": ["/api/v1/query?query=42&time={TIME_S}"],
"result_query": {
"status": "success",
"data":{
"resultType":"vector",
"result":[{"metric":{},"value":["{TIME_S}", "42"]}]
}
}
}

View File

@@ -0,0 +1,13 @@
{
"name": "too big look-behind window",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5553",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"foo\"},{\"name\":\"issue\",\"value\":\"5553\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS-60s}\"}]}]"],
"query": ["/api/v1/query?query=foo{issue=\"5553\"}[100y]"],
"result_query": {
"status": "success",
"data":{
"resultType":"matrix",
"result":[{"metric":{"__name__":"foo", "issue": "5553"},"values":[["{TIME_S-60s}", "1"]]}]
}
}
}

View File

@@ -0,0 +1,15 @@
{
"name": "match_series",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/155",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"MatchSeries\"},{\"name\":\"db\",\"value\":\"TenMinute\"},{\"name\":\"TurbineType\",\"value\":\"V112\"},{\"name\":\"Park\",\"value\":\"1\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"MatchSeries\"},{\"name\":\"db\",\"value\":\"TenMinute\"},{\"name\":\"TurbineType\",\"value\":\"V112\"},{\"name\":\"Park\",\"value\":\"2\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"MatchSeries\"},{\"name\":\"db\",\"value\":\"TenMinute\"},{\"name\":\"TurbineType\",\"value\":\"V112\"},{\"name\":\"Park\",\"value\":\"3\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"MatchSeries\"},{\"name\":\"db\",\"value\":\"TenMinute\"},{\"name\":\"TurbineType\",\"value\":\"V112\"},{\"name\":\"Park\",\"value\":\"4\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS}\"}]}]"],
"query": ["/api/v1/series?match[]={__name__='MatchSeries'}", "/api/v1/series?match[]={__name__=~'MatchSeries.*'}"],
"result_series": {
"status": "success",
"data": [
{"__name__":"MatchSeries","db":"TenMinute","Park":"1","TurbineType":"V112"},
{"__name__":"MatchSeries","db":"TenMinute","Park":"2","TurbineType":"V112"},
{"__name__":"MatchSeries","db":"TenMinute","Park":"3","TurbineType":"V112"},
{"__name__":"MatchSeries","db":"TenMinute","Park":"4","TurbineType":"V112"}
]
}
}

View File

@@ -0,0 +1,18 @@
{
"name": "query range",
"issue": "https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5553",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"bar\"}],\"samples\":[{\"value\":1,\"timestamp\":\"{TIME_MS-60s}\"}, {\"value\":2,\"timestamp\":\"{TIME_MS-120s}\"}, {\"value\":1,\"timestamp\":\"{TIME_MS-180s}\"}]}]"],
"query": ["/api/v1/query_range?query=bar&step=30s&start={TIME_MS-180s}"],
"result_query": {
"status": "success",
"data":{
"resultType":"matrix",
"result":[
{
"metric":{"__name__":"bar"},
"values":[["{TIME_S-180s}", "1"],["{TIME_S-150s}", "1"],["{TIME_S-120s}", "2"],["{TIME_S-90s}", "2"], ["{TIME_S-60s}", "1"], ["{TIME_S-30s}", "1"], ["{TIME_S}", "1"]]
}
]
}
}
}

View File

@@ -0,0 +1,9 @@
{
"name": "basic_insertion_with_extra_labels",
"insert_query": "?extra_label=job=prom-test&extra_label=baz=bar",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.foobar\"},{\"name\":\"baz\",\"value\":\"qux\"}],\"samples\":[{\"value\":100000,\"timestamp\":\"{TIME_MS}\"}]}]"],
"query": ["/api/v1/export?match={__name__!=''}"],
"result_metrics": [
{"metric":{"__name__":"prometheus.foobar","baz":"bar","job": "prom-test"},"values":[100000], "timestamps": ["{TIME_MS}"]}
]
}

View File

@@ -0,0 +1,8 @@
{
"name": "basic_select_with_extra_labels",
"data": ["[{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.tenant.limits\"},{\"name\":\"baz\",\"value\":\"qux\"},{\"name\":\"tenant\",\"value\":\"dev\"}],\"samples\":[{\"value\":100000,\"timestamp\":\"{TIME_MS}\"}]},{\"labels\":[{\"name\":\"__name__\",\"value\":\"prometheus.up\"},{\"name\":\"baz\",\"value\":\"qux\"}],\"samples\":[{\"value\":100000,\"timestamp\":\"{TIME_MS}\"}]}]"],
"query": ["/api/v1/export?match={__name__!=''}&extra_label=tenant=dev"],
"result_metrics": [
{"metric":{"__name__":"prometheus.tenant.limits","baz":"qux","tenant": "dev"},"values":[100000], "timestamps": ["{TIME_MS}"]}
]
}

View File

@@ -11,6 +11,7 @@ import (
"github.com/valyala/fastjson"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -32,10 +33,10 @@ var parserPool fastjson.ParserPool
// RequestHandler processes Datadog insert requests
func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
switch path {
case "/insert/datadog/api/v1/validate":
case "/api/v1/validate":
fmt.Fprintf(w, `{}`)
return true
case "/insert/datadog/api/v2/logs":
case "/api/v2/logs":
return datadogLogsIngestion(w, r)
default:
return false
@@ -73,7 +74,7 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
cp.IgnoreFields = *datadogIgnoreFields
}
if err := insertutil.CanWriteData(); err != nil {
if err := vlstorage.CanWriteData(); err != nil {
httpserver.Errorf(w, r, "%s", err)
return true
}
@@ -94,14 +95,13 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
// There is no need in updating v2LogsRequestDuration for request errors,
// since their timings are usually much smaller than the timing for successful request parsing.
v2LogsRequestDuration.UpdateDuration(startTime)
w.WriteHeader(http.StatusAccepted)
fmt.Fprintf(w, `{}`)
return true
}
var (
v2LogsRequestsTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/datadog/api/v2/logs"}`)
v2LogsRequestDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/datadog/api/v2/logs"}`)
v2LogsRequestDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/datadog/api/v2/logs"}`)
)
// datadog message field has two formats:

View File

@@ -11,6 +11,7 @@ import (
"github.com/VictoriaMetrics/metrics"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bufferedwriter"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -30,38 +31,36 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
// This header is needed for Logstash
w.Header().Set("X-Elastic-Product", "Elasticsearch")
if strings.HasPrefix(path, "/insert/elasticsearch/_ilm/policy") {
if strings.HasPrefix(path, "/_ilm/policy") {
// Return fake response for Elasticsearch ilm request.
fmt.Fprintf(w, `{}`)
return true
}
if strings.HasPrefix(path, "/insert/elasticsearch/_index_template") {
if strings.HasPrefix(path, "/_index_template") {
// Return fake response for Elasticsearch index template request.
fmt.Fprintf(w, `{}`)
return true
}
if strings.HasPrefix(path, "/insert/elasticsearch/_ingest") {
if strings.HasPrefix(path, "/_ingest") {
// Return fake response for Elasticsearch ingest pipeline request.
// See: https://www.elastic.co/guide/en/elasticsearch/reference/8.8/put-pipeline-api.html
fmt.Fprintf(w, `{}`)
return true
}
if strings.HasPrefix(path, "/insert/elasticsearch/_nodes") {
if strings.HasPrefix(path, "/_nodes") {
// Return fake response for Elasticsearch nodes discovery request.
// See: https://www.elastic.co/guide/en/elasticsearch/reference/8.8/cluster.html
fmt.Fprintf(w, `{}`)
return true
}
if strings.HasPrefix(path, "/insert/elasticsearch/logstash") || strings.HasPrefix(path, "/insert/elasticsearch/_logstash") {
if strings.HasPrefix(path, "/logstash") || strings.HasPrefix(path, "/_logstash") {
// Return fake response for Logstash APIs requests.
// See: https://www.elastic.co/guide/en/elasticsearch/reference/8.8/logstash-apis.html
fmt.Fprintf(w, `{}`)
return true
}
switch path {
// some clients may omit trailing slash
// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8353
case "/insert/elasticsearch/", "/insert/elasticsearch":
case "/", "":
switch r.Method {
case http.MethodGet:
// Return fake response for Elasticsearch ping request.
@@ -76,7 +75,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
}
return true
case "/insert/elasticsearch/_license":
case "/_license":
// Return fake response for Elasticsearch license request.
fmt.Fprintf(w, `{
"license": {
@@ -87,7 +86,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
}
}`)
return true
case "/insert/elasticsearch/_bulk":
case "/_bulk":
startTime := time.Now()
bulkRequestsTotal.Inc()
@@ -96,14 +95,14 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
httpserver.Errorf(w, r, "%s", err)
return true
}
if err := insertutil.CanWriteData(); err != nil {
if err := vlstorage.CanWriteData(); err != nil {
httpserver.Errorf(w, r, "%s", err)
return true
}
lmp := cp.NewLogMessageProcessor("elasticsearch_bulk", true)
encoding := r.Header.Get("Content-Encoding")
streamName := fmt.Sprintf("remoteAddr=%s, requestURI=%q", httpserver.GetQuotedRemoteAddr(r), r.RequestURI)
n, err := readBulkRequest(streamName, r.Body, encoding, cp.TimeFields, cp.MsgFields, lmp)
n, err := readBulkRequest(streamName, r.Body, encoding, cp.TimeField, cp.MsgFields, lmp)
lmp.MustClose()
if err != nil {
logger.Warnf("cannot decode log message #%d in /_bulk request: %s, stream fields: %s", n, err, cp.StreamFields)
@@ -129,10 +128,10 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
var (
bulkRequestsTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/elasticsearch/_bulk"}`)
bulkRequestDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/elasticsearch/_bulk"}`)
bulkRequestDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/elasticsearch/_bulk"}`)
)
func readBulkRequest(streamName string, r io.Reader, encoding string, timeFields, msgFields []string, lmp insertutil.LogMessageProcessor) (int, error) {
func readBulkRequest(streamName string, r io.Reader, encoding string, timeField string, msgFields []string, lmp insertutil.LogMessageProcessor) (int, error) {
// See https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
reader, err := protoparserutil.GetUncompressedReader(r, encoding)
@@ -148,7 +147,7 @@ func readBulkRequest(streamName string, r io.Reader, encoding string, timeFields
n := 0
for {
ok, err := readBulkLine(lr, timeFields, msgFields, lmp)
ok, err := readBulkLine(lr, timeField, msgFields, lmp)
wcr.DecConcurrency()
if err != nil || !ok {
return n, err
@@ -157,7 +156,7 @@ func readBulkRequest(streamName string, r io.Reader, encoding string, timeFields
}
}
func readBulkLine(lr *insertutil.LineReader, timeFields, msgFields []string, lmp insertutil.LogMessageProcessor) (bool, error) {
func readBulkLine(lr *insertutil.LineReader, timeField string, msgFields []string, lmp insertutil.LogMessageProcessor) (bool, error) {
var line []byte
// Read the command, must be "create" or "index"
@@ -191,7 +190,7 @@ func readBulkLine(lr *insertutil.LineReader, timeFields, msgFields []string, lmp
return false, fmt.Errorf("cannot parse json-encoded log entry: %w", err)
}
ts, err := extractTimestampFromFields(timeFields, p.Fields)
ts, err := extractTimestampFromFields(timeField, p.Fields)
if err != nil {
return false, fmt.Errorf("cannot parse timestamp: %w", err)
}
@@ -205,20 +204,18 @@ func readBulkLine(lr *insertutil.LineReader, timeFields, msgFields []string, lmp
return true, nil
}
func extractTimestampFromFields(timeFields []string, fields []logstorage.Field) (int64, error) {
for _, timeField := range timeFields {
for i := range fields {
f := &fields[i]
if f.Name != timeField {
continue
}
timestamp, err := parseElasticsearchTimestamp(f.Value)
if err != nil {
return 0, err
}
f.Value = ""
return timestamp, nil
func extractTimestampFromFields(timeField string, fields []logstorage.Field) (int64, error) {
for i := range fields {
f := &fields[i]
if f.Name != timeField {
continue
}
timestamp, err := parseElasticsearchTimestamp(f.Value)
if err != nil {
return 0, err
}
f.Value = ""
return timestamp, nil
}
return 0, nil
}

View File

@@ -19,7 +19,7 @@ func TestReadBulkRequest_Failure(t *testing.T) {
tlp := &insertutil.TestLogMessageProcessor{}
r := bytes.NewBufferString(data)
rows, err := readBulkRequest("test", r, "", []string{"_time"}, []string{"_msg"}, tlp)
rows, err := readBulkRequest("test", r, "", "_time", []string{"_msg"}, tlp)
if err == nil {
t.Fatalf("expecting non-empty error")
}
@@ -40,13 +40,12 @@ func TestReadBulkRequest_Success(t *testing.T) {
f := func(data, encoding, timeField, msgField string, timestampsExpected []int64, resultExpected string) {
t.Helper()
timeFields := []string{"non_existing_foo", timeField, "non_existing_bar"}
msgFields := []string{"non_existing_foo", msgField, "non_exiting_bar"}
tlp := &insertutil.TestLogMessageProcessor{}
// Read the request without compression
r := bytes.NewBufferString(data)
rows, err := readBulkRequest("test", r, "", timeFields, msgFields, tlp)
rows, err := readBulkRequest("test", r, "", timeField, msgFields, tlp)
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
@@ -63,7 +62,7 @@ func TestReadBulkRequest_Success(t *testing.T) {
data = compressData(data, encoding)
}
r = bytes.NewBufferString(data)
rows, err = readBulkRequest("test", r, encoding, timeFields, msgFields, tlp)
rows, err = readBulkRequest("test", r, encoding, timeField, msgFields, tlp)
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
@@ -112,7 +111,7 @@ func compressData(s string, encoding string) string {
case "zstd":
zw, _ = zstd.NewWriter(&bb)
case "snappy":
return string(snappy.Encode(nil, []byte(s)))
zw = snappy.NewBufferedWriter(&bb)
case "deflate":
zw = zlib.NewWriter(&bb)
default:

View File

@@ -40,7 +40,7 @@ func benchmarkReadBulkRequest(b *testing.B, encoding string) {
}
dataBytes := bytesutil.ToUnsafeBytes(data)
timeFields := []string{"@timestamp"}
timeField := "@timestamp"
msgFields := []string{"message"}
blp := &insertutil.BenchmarkLogMessageProcessor{}
@@ -50,7 +50,7 @@ func benchmarkReadBulkRequest(b *testing.B, encoding string) {
r := &bytes.Reader{}
for pb.Next() {
r.Reset(dataBytes)
_, err := readBulkRequest("test", r, encoding, timeFields, msgFields, blp)
_, err := readBulkRequest("test", r, encoding, timeField, msgFields, blp)
if err != nil {
panic(fmt.Errorf("unexpected error: %w", err))
}

View File

@@ -11,6 +11,7 @@ import (
"github.com/VictoriaMetrics/metrics"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -27,15 +28,13 @@ var (
//
// See https://docs.victoriametrics.com/victorialogs/data-ingestion/#http-parameters
type CommonParams struct {
TenantID logstorage.TenantID
TimeFields []string
MsgFields []string
StreamFields []string
IgnoreFields []string
DecolorizeFields []string
ExtraFields []logstorage.Field
TenantID logstorage.TenantID
TimeField string
MsgFields []string
StreamFields []string
IgnoreFields []string
ExtraFields []logstorage.Field
IsTimeFieldSet bool
Debug bool
DebugRequestURI string
DebugRemoteAddr string
@@ -49,17 +48,14 @@ func GetCommonParams(r *http.Request) (*CommonParams, error) {
return nil, err
}
var isTimeFieldSet bool
timeFields := []string{"_time"}
if tfs := httputil.GetArray(r, "_time_field", "VL-Time-Field"); len(tfs) > 0 {
isTimeFieldSet = true
timeFields = tfs
timeField := "_time"
if tf := httputil.GetRequestValue(r, "_time_field", "VL-Time-Field"); tf != "" {
timeField = tf
}
msgFields := httputil.GetArray(r, "_msg_field", "VL-Msg-Field")
streamFields := httputil.GetArray(r, "_stream_fields", "VL-Stream-Fields")
ignoreFields := httputil.GetArray(r, "ignore_fields", "VL-Ignore-Fields")
decolorizeFields := httputil.GetArray(r, "decolorize_fields", "VL-Decolorize-Fields")
extraFields, err := getExtraFields(r)
if err != nil {
@@ -81,15 +77,12 @@ func GetCommonParams(r *http.Request) (*CommonParams, error) {
}
cp := &CommonParams{
TenantID: tenantID,
TimeFields: timeFields,
MsgFields: msgFields,
StreamFields: streamFields,
IgnoreFields: ignoreFields,
DecolorizeFields: decolorizeFields,
ExtraFields: extraFields,
IsTimeFieldSet: isTimeFieldSet,
TenantID: tenantID,
TimeField: timeField,
MsgFields: msgFields,
StreamFields: streamFields,
IgnoreFields: ignoreFields,
ExtraFields: extraFields,
Debug: debug,
DebugRequestURI: debugRequestURI,
DebugRemoteAddr: debugRemoteAddr,
@@ -119,7 +112,7 @@ func getExtraFields(r *http.Request) ([]logstorage.Field, error) {
}
// GetCommonParamsForSyslog returns common params needed for parsing syslog messages and storing them to the given tenantID.
func GetCommonParamsForSyslog(tenantID logstorage.TenantID, streamFields, ignoreFields, decolorizeFields []string, extraFields []logstorage.Field) *CommonParams {
func GetCommonParamsForSyslog(tenantID logstorage.TenantID, streamFields, ignoreFields []string, extraFields []logstorage.Field) *CommonParams {
// See https://docs.victoriametrics.com/victorialogs/logsql/#unpack_syslog-pipe
if streamFields == nil {
streamFields = []string{
@@ -129,45 +122,19 @@ func GetCommonParamsForSyslog(tenantID logstorage.TenantID, streamFields, ignore
}
}
cp := &CommonParams{
TenantID: tenantID,
TimeFields: []string{
"timestamp",
},
TenantID: tenantID,
TimeField: "timestamp",
MsgFields: []string{
"message",
},
StreamFields: streamFields,
IgnoreFields: ignoreFields,
DecolorizeFields: decolorizeFields,
ExtraFields: extraFields,
StreamFields: streamFields,
IgnoreFields: ignoreFields,
ExtraFields: extraFields,
}
return cp
}
// LogRowsStorage is an interface for ingesting logs into the storage.
type LogRowsStorage interface {
// MustAddRows must add lr to the underlying storage.
MustAddRows(lr *logstorage.LogRows)
// CanWriteData must returns non-nil error if logs cannot be added to the underlying storage.
CanWriteData() error
}
var logRowsStorage LogRowsStorage
// SetLogRowsStorage sets the storage for writing data to via LogMessageProcessor.
//
// This function must be called before using LogMessageProcessor and CanWriteData from this package.
func SetLogRowsStorage(storage LogRowsStorage) {
logRowsStorage = storage
}
// CanWriteData returns non-nil error if data cannot be written to the underlying storage.
func CanWriteData() error {
return logRowsStorage.CanWriteData()
}
// LogMessageProcessor is an interface for log message processors.
type LogMessageProcessor interface {
// AddRow must add row to the LogMessageProcessor with the given timestamp and fields.
@@ -252,46 +219,10 @@ func (lmp *logMessageProcessor) AddRow(timestamp int64, fields, streamFields []l
}
}
// InsertRowProcessor is used by native data ingestion protocol parser.
type InsertRowProcessor interface {
// AddInsertRow must add r to the underlying storage.
AddInsertRow(r *logstorage.InsertRow)
}
// AddInsertRow adds r to lmp.
func (lmp *logMessageProcessor) AddInsertRow(r *logstorage.InsertRow) {
lmp.rowsIngestedTotal.Inc()
n := logstorage.EstimatedJSONRowLen(r.Fields)
lmp.bytesIngestedTotal.Add(n)
if len(r.Fields) > *MaxFieldsPerLine {
line := logstorage.MarshalFieldsToJSON(nil, r.Fields)
logger.Warnf("dropping log line with %d fields; it exceeds -insert.maxFieldsPerLine=%d; %s", len(r.Fields), *MaxFieldsPerLine, line)
rowsDroppedTotalTooManyFields.Inc()
return
}
lmp.mu.Lock()
defer lmp.mu.Unlock()
lmp.lr.MustAddInsertRow(r)
if lmp.cp.Debug {
s := lmp.lr.GetRowString(0)
lmp.lr.ResetKeepSettings()
logger.Infof("remoteAddr=%s; requestURI=%s; ignoring log entry because of `debug` arg: %s", lmp.cp.DebugRemoteAddr, lmp.cp.DebugRequestURI, s)
rowsDroppedTotalDebug.Inc()
return
}
if lmp.lr.NeedFlush() {
lmp.flushLocked()
}
}
// flushLocked must be called under locked lmp.mu.
func (lmp *logMessageProcessor) flushLocked() {
lmp.lastFlushTime = time.Now()
logRowsStorage.MustAddRows(lmp.lr)
vlstorage.MustAddRows(lmp.lr)
lmp.lr.ResetKeepSettings()
}
@@ -309,7 +240,7 @@ 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)
lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields, 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))
lmp := &logMessageProcessor{

View File

@@ -56,7 +56,6 @@ func NewLineReader(name string, r io.Reader) *LineReader {
// Check for Err in this case.
func (lr *LineReader) NextLine() bool {
for {
lr.Line = nil
if lr.bufOffset >= len(lr.buf) {
if lr.err != nil || lr.eofReached {
return false
@@ -102,11 +101,9 @@ func (lr *LineReader) readMoreData() bool {
bufLen := len(lr.buf)
if bufLen >= MaxLineSizeBytes.IntN() {
ok, skippedBytes := lr.skipUntilNextLine()
logger.Warnf("%s: the line length exceeds -insert.maxLineSizeBytes=%d; skipping it; total skipped bytes=%d",
lr.name, MaxLineSizeBytes.IntN(), skippedBytes)
logger.Warnf("%s: the line length exceeds -insert.maxLineSizeBytes=%d; skipping it; line contents=%q", lr.name, MaxLineSizeBytes.IntN(), lr.buf)
tooLongLinesSkipped.Inc()
return ok
return lr.skipUntilNextLine()
}
lr.buf = slicesutil.SetLength(lr.buf, MaxLineSizeBytes.IntN())
@@ -124,35 +121,26 @@ func (lr *LineReader) readMoreData() bool {
var tooLongLinesSkipped = metrics.NewCounter("vl_too_long_lines_skipped_total")
func (lr *LineReader) skipUntilNextLine() (bool, int) {
// Initialize skipped bytes count with MaxLineSizeBytes because
// we've already read that many bytes without encountering a newline,
// indicating the line size exceeds the maximum allowed limit.
skipSizeBytes := MaxLineSizeBytes.IntN()
func (lr *LineReader) skipUntilNextLine() bool {
for {
lr.buf = slicesutil.SetLength(lr.buf, MaxLineSizeBytes.IntN())
n, err := lr.r.Read(lr.buf)
skipSizeBytes += n
lr.buf = lr.buf[:n]
if err != nil {
if errors.Is(err, io.EOF) {
lr.eofReached = true
lr.buf = lr.buf[:0]
return true, skipSizeBytes
return true
}
lr.err = fmt.Errorf("cannot skip the current line: %s", err)
return false, skipSizeBytes
return false
}
if n := bytes.IndexByte(lr.buf, '\n'); n >= 0 {
// Include skipped bytes before \n, including the newline itself.
skipSizeBytes += n + 1 - len(lr.buf)
// Include \n in the buf, so too long line is replaced with an empty line.
// This is needed for maintaining synchorinzation consistency between lines
// in protocols such as Elasticsearch bulk import.
lr.buf = append(lr.buf[:0], lr.buf[n:]...)
return true, skipSizeBytes
return true
}
}
}

View File

@@ -24,9 +24,6 @@ func TestLineReader_Success(t *testing.T) {
if lr.NextLine() {
t.Fatalf("expecting error on the second call to NextLine()")
}
if len(lr.Line) > 0 {
t.Fatalf("unexpected non-empty line after failed NextLine(): %q", lr.Line)
}
if !reflect.DeepEqual(lines, linesExpected) {
t.Fatalf("unexpected lines\ngot\n%q\nwant\n%q", lines, linesExpected)
}

View File

@@ -10,38 +10,33 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
)
// ExtractTimestampFromFields extracts timestamp in nanoseconds from the first field the name from timeFields at fields.
// ExtractTimestampFromFields extracts timestamp in nanoseconds from the field with the name timeField at fields.
//
// The value for the corresponding timeFields is set to empty string after returning from the function,
// The value for the timeField is set to empty string after returning from the function,
// so it could be ignored during data ingestion.
//
// The current timestamp is returned if fields do not contain a field with timeField name or if the timeField value is empty.
func ExtractTimestampFromFields(timeFields []string, fields []logstorage.Field) (int64, error) {
for _, timeField := range timeFields {
for i := range fields {
f := &fields[i]
if f.Name != timeField {
continue
}
nsecs, err := parseTimestamp(f.Value)
if err != nil {
return 0, fmt.Errorf("cannot parse timestamp from field %q: %s", f.Name, err)
}
f.Value = ""
if nsecs == 0 {
nsecs = time.Now().UnixNano()
}
return nsecs, nil
func ExtractTimestampFromFields(timeField string, fields []logstorage.Field) (int64, error) {
for i := range fields {
f := &fields[i]
if f.Name != timeField {
continue
}
nsecs, err := parseTimestamp(f.Value)
if err != nil {
return 0, fmt.Errorf("cannot parse timestamp from field %q: %s", timeField, err)
}
f.Value = ""
if nsecs == 0 {
nsecs = time.Now().UnixNano()
}
return nsecs, nil
}
return time.Now().UnixNano(), nil
}
func parseTimestamp(s string) (int64, error) {
// "-" is a nil timestamp value, if the syslog
// application is incapable of obtaining system time
// https://datatracker.ietf.org/doc/html/rfc5424#section-6.2.3
if s == "" || s == "0" || s == "-" {
if s == "" || s == "0" {
return time.Now().UnixNano(), nil
}
if len(s) <= len("YYYY") || s[len("YYYY")] != '-' {

View File

@@ -67,7 +67,7 @@ func TestExtractTimestampFromFields_Success(t *testing.T) {
f := func(timeField string, fields []logstorage.Field, nsecsExpected int64) {
t.Helper()
nsecs, err := ExtractTimestampFromFields([]string{timeField}, fields)
nsecs, err := ExtractTimestampFromFields(timeField, fields)
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
@@ -133,33 +133,6 @@ func TestExtractTimestampFromFields_Success(t *testing.T) {
}, 1718773640000000000)
}
func TestExtractTimestampFromFields_Now(t *testing.T) {
f := func(timeField string, fields []logstorage.Field) {
t.Helper()
nsecs, err := ExtractTimestampFromFields([]string{timeField}, fields)
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
if nsecs < 1 {
t.Fatalf("expected generated timestamp, got error: %s", err)
}
}
// RFC5424 allows `-` for nil timestamp (log ingestion time)
f("time", []logstorage.Field{
{Name: "time", Value: "-"},
})
f("time", []logstorage.Field{
{Name: "time", Value: ""},
})
f("time", []logstorage.Field{
{Name: "time", Value: "0"},
})
}
func TestExtractTimestampFromFields_Error(t *testing.T) {
f := func(s string) {
t.Helper()
@@ -167,7 +140,7 @@ func TestExtractTimestampFromFields_Error(t *testing.T) {
fields := []logstorage.Field{
{Name: "time", Value: s},
}
nsecs, err := ExtractTimestampFromFields([]string{"time"}, fields)
nsecs, err := ExtractTimestampFromFields("time", fields)
if err == nil {
t.Fatalf("expecting non-nil error")
}

View File

@@ -1,88 +0,0 @@
package internalinsert
import (
"fmt"
"net/http"
"time"
"github.com/VictoriaMetrics/metrics"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage/netinsert"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
)
var (
maxRequestSize = flagutil.NewBytes("internalinsert.maxRequestSize", 64*1024*1024, "The maximum size in bytes of a single request, which can be accepted at /internal/insert HTTP endpoint")
)
// RequestHandler processes /internal/insert requests.
func RequestHandler(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
if r.Method != "POST" {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
version := r.FormValue("version")
if version != netinsert.ProtocolVersion {
httpserver.Errorf(w, r, "unsupported protocol version=%q; want %q", version, netinsert.ProtocolVersion)
return
}
requestsTotal.Inc()
cp, err := insertutil.GetCommonParams(r)
if err != nil {
httpserver.Errorf(w, r, "%s", err)
return
}
if err := insertutil.CanWriteData(); err != nil {
httpserver.Errorf(w, r, "%s", err)
return
}
encoding := r.Header.Get("Content-Encoding")
err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
lmp := cp.NewLogMessageProcessor("internalinsert", false)
irp := lmp.(insertutil.InsertRowProcessor)
err := parseData(irp, data)
lmp.MustClose()
return err
})
if err != nil {
errorsTotal.Inc()
httpserver.Errorf(w, r, "cannot parse internal insert request: %s", err)
return
}
requestDuration.UpdateDuration(startTime)
}
func parseData(irp insertutil.InsertRowProcessor, data []byte) error {
r := logstorage.GetInsertRow()
src := data
i := 0
for len(src) > 0 {
tail, err := r.UnmarshalInplace(src)
if err != nil {
return fmt.Errorf("cannot parse row #%d: %s", i, err)
}
src = tail
i++
irp.AddInsertRow(r)
}
logstorage.PutInsertRow(r)
return nil
}
var (
requestsTotal = metrics.NewCounter(`vl_http_requests_total{path="/internal/insert"}`)
errorsTotal = metrics.NewCounter(`vl_http_errors_total{path="/internal/insert"}`)
requestDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/internal/insert"}`)
)

View File

@@ -3,30 +3,29 @@ package journald
import (
"bytes"
"encoding/binary"
"errors"
"flag"
"fmt"
"io"
"net/http"
"regexp"
"slices"
"strconv"
"strings"
"sync"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
"github.com/VictoriaMetrics/metrics"
)
// See https://github.com/systemd/systemd/blob/main/src/libsystemd/sd-journal/journal-file.c#L1703
const maxFieldNameLen = 64
const journaldEntryMaxNameLen = 64
var allowedJournaldEntryNameChars = regexp.MustCompile(`^[A-Z_][A-Z0-9_]*`)
var (
journaldStreamFields = flagutil.NewArrayString("journald.streamFields", "Comma-separated list of fields to use as log stream fields for logs ingested over journald protocol. "+
@@ -37,7 +36,9 @@ var (
"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#time-field")
journaldTenantID = flag.String("journald.tenantID", "0:0", "TenantID for logs ingested via the Journald endpoint. "+
"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#multitenancy")
journaldIncludeEntryMetadata = flag.Bool("journald.includeEntryMetadata", false, "Include Journald fields with double underscore prefixes")
journaldIncludeEntryMetadata = flag.Bool("journald.includeEntryMetadata", false, "Include journal entry fields, which with double underscores.")
maxRequestSize = flagutil.NewBytes("journald.maxRequestSize", 64*1024*1024, "The maximum size in bytes of a single journald request")
)
func getCommonParams(r *http.Request) (*insertutil.CommonParams, error) {
@@ -52,12 +53,11 @@ func getCommonParams(r *http.Request) (*insertutil.CommonParams, error) {
}
cp.TenantID = tenantID
}
if !cp.IsTimeFieldSet {
cp.TimeFields = []string{*journaldTimeField}
if cp.TimeField != "" {
cp.TimeField = *journaldTimeField
}
if len(cp.StreamFields) == 0 {
cp.StreamFields = getStreamFields()
cp.StreamFields = *journaldStreamFields
}
if len(cp.IgnoreFields) == 0 {
cp.IgnoreFields = *journaldIgnoreFields
@@ -66,23 +66,10 @@ func getCommonParams(r *http.Request) (*insertutil.CommonParams, error) {
return cp, nil
}
func getStreamFields() []string {
if len(*journaldStreamFields) > 0 {
return *journaldStreamFields
}
return defaultStreamFields
}
var defaultStreamFields = []string{
"_MACHINE_ID",
"_HOSTNAME",
"_SYSTEMD_UNIT",
}
// RequestHandler processes Journald Export insert requests
func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
switch path {
case "/insert/journald/upload":
case "/upload":
if r.Header.Get("Content-Type") != "application/vnd.fdo.journal" {
httpserver.Errorf(w, r, "only application/vnd.fdo.journal encoding is supported for Journald")
return true
@@ -97,7 +84,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
// handleJournald parses Journal binary entries
func handleJournald(r *http.Request, w http.ResponseWriter) {
startTime := time.Now()
requestsTotal.Inc()
requestsJournaldTotal.Inc()
cp, err := getCommonParams(r)
if err != nil {
@@ -106,25 +93,19 @@ func handleJournald(r *http.Request, w http.ResponseWriter) {
return
}
if err := insertutil.CanWriteData(); err != nil {
if err := vlstorage.CanWriteData(); err != nil {
errorsTotal.Inc()
httpserver.Errorf(w, r, "%s", err)
return
}
encoding := r.Header.Get("Content-Encoding")
reader, err := protoparserutil.GetUncompressedReader(r.Body, encoding)
if err != nil {
errorsTotal.Inc()
logger.Errorf("cannot decode journald request: %s", err)
return
}
lmp := cp.NewLogMessageProcessor("journald", true)
streamName := fmt.Sprintf("remoteAddr=%s, requestURI=%q", httpserver.GetQuotedRemoteAddr(r), r.RequestURI)
err = processStreamInternal(streamName, reader, lmp, cp)
protoparserutil.PutUncompressedReader(reader)
lmp.MustClose()
err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
lmp := cp.NewLogMessageProcessor("journald", false)
err := parseJournaldRequest(data, lmp, cp)
lmp.MustClose()
return err
})
if err != nil {
errorsTotal.Inc()
httpserver.Errorf(w, r, "cannot read journald protocol data: %s", err)
@@ -136,185 +117,102 @@ func handleJournald(r *http.Request, w http.ResponseWriter) {
// See https://github.com/systemd/systemd/pull/34822
w.Header().Set("Accept-Encoding", "zstd")
// update requestDuration only for successfully parsed requests
// There is no need in updating requestDuration for request errors,
// update requestJournaldDuration only for successfully parsed requests
// There is no need in updating requestJournaldDuration for request errors,
// since their timings are usually much smaller than the timing for successful request parsing.
requestDuration.UpdateDuration(startTime)
requestJournaldDuration.UpdateDuration(startTime)
}
var (
requestsTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/journald/upload"}`)
errorsTotal = metrics.NewCounter(`vl_http_errors_total{path="/insert/journald/upload"}`)
requestDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/journald/upload"}`)
requestsJournaldTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/journald/upload"}`)
errorsTotal = metrics.NewCounter(`vl_http_errors_total{path="/insert/journald/upload"}`)
requestJournaldDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/journald/upload"}`)
)
func processStreamInternal(streamName string, r io.Reader, lmp insertutil.LogMessageProcessor, cp *insertutil.CommonParams) error {
wcr := writeconcurrencylimiter.GetReader(r)
defer writeconcurrencylimiter.PutReader(wcr)
lr := insertutil.NewLineReader("journald", wcr)
for {
err := readJournaldLogEntry(streamName, lr, lmp, cp)
wcr.DecConcurrency()
if err != nil {
if errors.Is(err, io.EOF) {
return nil
}
return fmt.Errorf("%s: %w", streamName, err)
}
}
}
type fieldsBuf struct {
fields []logstorage.Field
buf []byte
name []byte
value []byte
}
func (fb *fieldsBuf) reset() {
fb.fields = fb.fields[:0]
fb.buf = fb.buf[:0]
fb.name = fb.name[:0]
fb.value = fb.value[:0]
}
func (fb *fieldsBuf) addField(name, value string) {
bufLen := len(fb.buf)
fb.buf = append(fb.buf, name...)
nameCopy := bytesutil.ToUnsafeString(fb.buf[bufLen:])
bufLen = len(fb.buf)
fb.buf = append(fb.buf, value...)
valueCopy := bytesutil.ToUnsafeString(fb.buf[bufLen:])
fb.fields = append(fb.fields, logstorage.Field{
Name: nameCopy,
Value: valueCopy,
})
}
func (fb *fieldsBuf) appendNextLineToValue(lr *insertutil.LineReader) error {
if !lr.NextLine() {
if err := lr.Err(); err != nil {
return err
}
return fmt.Errorf("unexpected end of stream")
}
fb.value = append(fb.value, lr.Line...)
fb.value = append(fb.value, '\n')
return nil
}
func getFieldsBuf() *fieldsBuf {
fb := fieldsBufPool.Get()
if fb == nil {
return &fieldsBuf{}
}
return fb.(*fieldsBuf)
}
func putFieldsBuf(fb *fieldsBuf) {
fb.reset()
fieldsBufPool.Put(fb)
}
var fieldsBufPool sync.Pool
// readJournaldLogEntry reads a single log entry in Journald format.
//
// See https://systemd.io/JOURNAL_EXPORT_FORMATS/#journal-export-format
func readJournaldLogEntry(streamName string, lr *insertutil.LineReader, lmp insertutil.LogMessageProcessor, cp *insertutil.CommonParams) error {
func parseJournaldRequest(data []byte, lmp insertutil.LogMessageProcessor, cp *insertutil.CommonParams) error {
var fields []logstorage.Field
var ts int64
var size uint64
var name, value string
var line []byte
fb := getFieldsBuf()
defer putFieldsBuf(fb)
currentTimestamp := time.Now().UnixNano()
if !lr.NextLine() {
if err := lr.Err(); err != nil {
return fmt.Errorf("cannot read the first field: %w", err)
}
return io.EOF
}
for {
line := lr.Line
if len(line) == 0 {
// The end of a single log entry. Write it to the storage
if len(fb.fields) > 0 {
for len(data) > 0 {
idx := bytes.IndexByte(data, '\n')
switch {
case idx > 0:
// process fields
line = data[:idx]
data = data[idx+1:]
case idx == 0:
// next message or end of file
// double new line is a separator for the next message
if len(fields) > 0 {
if ts == 0 {
ts = time.Now().UnixNano()
ts = currentTimestamp
}
lmp.AddRow(ts, fb.fields, nil)
lmp.AddRow(ts, fields, nil)
fields = fields[:0]
}
return nil
// skip newline separator
data = data[1:]
continue
case idx < 0:
return fmt.Errorf("missing new line separator, unread data left=%d", len(data))
}
// line could be either "key=value" or "key"
// according to https://systemd.io/JOURNAL_EXPORT_FORMATS/#journal-export-format
if n := bytes.IndexByte(line, '='); n >= 0 {
// line = "key=value"
fb.name = append(fb.name[:0], line[:n]...)
name = bytesutil.ToUnsafeString(fb.name)
fb.value = append(fb.value[:0], line[n+1:]...)
value = bytesutil.ToUnsafeString(fb.value)
idx = bytes.IndexByte(line, '=')
// could b either e key=value\n pair
// or just key\n
// with binary data at the buffer
if idx > 0 {
name = bytesutil.ToUnsafeString(line[:idx])
value = bytesutil.ToUnsafeString(line[idx+1:])
} else {
// line = "key"
// Parse the binary-encoded value from the next line according to "key\n<little_endian_size_64>value\n" format
fb.name = append(fb.name[:0], line...)
name = bytesutil.ToUnsafeString(fb.name)
fb.value = fb.value[:0]
for len(fb.value) < 8 {
if err := fb.appendNextLineToValue(lr); err != nil {
return fmt.Errorf("cannot read value size: %w", err)
}
name = bytesutil.ToUnsafeString(line)
if len(data) == 0 {
return fmt.Errorf("unexpected zero data for binary field value of key=%s", name)
}
size := binary.LittleEndian.Uint64(fb.value[:8])
// Read the value until its lenth exceeds the given size - the last char in the read value will always be '\n'
// because it is appended by appendNextLineToValue().
for uint64(len(fb.value[8:])) <= size {
if err := fb.appendNextLineToValue(lr); err != nil {
return fmt.Errorf("cannot read %q value with size %d bytes; read only %d bytes: %w", fb.name, size, len(fb.value[8:]), err)
}
}
value = bytesutil.ToUnsafeString(fb.value[8 : len(fb.value)-1])
if uint64(len(value)) != size {
return fmt.Errorf("unexpected %q value size; got %d bytes; want %d bytes; value: %q", fb.name, len(value), size, value)
}
}
if !lr.NextLine() {
if err := lr.Err(); err != nil {
return fmt.Errorf("cannot read the next log field: %w", err)
}
// add the last log field below before the return
}
if len(name) > maxFieldNameLen {
logger.Errorf("%s: field name size should not exceed %d bytes; got %d bytes: %q; skipping this field", streamName, maxFieldNameLen, len(name), name)
continue
}
if !isValidFieldName(name) {
logger.Errorf("%s: invalid field name %q; it must consist of `A-Z0-9_` chars and must start from non-digit char; skipping this field", streamName, name)
continue
}
if slices.Contains(cp.TimeFields, name) {
t, err := strconv.ParseInt(value, 10, 64)
// size of binary data encoded as le i64 at the begging
idx, err := binary.Decode(data, binary.LittleEndian, &size)
if err != nil {
logger.Errorf("%s: cannot parse timestamp from the field %q: %w; using the current timestamp", streamName, name, err)
ts = 0
} else {
// Convert journald microsecond timestamp to nanoseconds
ts = t * 1e3
return fmt.Errorf("failed to extract binary field %q value size: %w", name, err)
}
// skip binary data size
data = data[idx:]
if size == 0 {
return fmt.Errorf("unexpected zero binary data size decoded %d", size)
}
if int(size) > len(data) {
return fmt.Errorf("binary data size=%d cannot exceed size of the data at buffer=%d", size, len(data))
}
value = bytesutil.ToUnsafeString(data[:size])
data = data[int(size):]
// binary data must has new line separator for the new line or next field
if len(data) == 0 {
return fmt.Errorf("unexpected empty buffer after binary field=%s read", name)
}
lastB := data[0]
if lastB != '\n' {
return fmt.Errorf("expected new line separator after binary field=%s, got=%s", name, string(lastB))
}
data = data[1:]
}
if len(name) > journaldEntryMaxNameLen {
return fmt.Errorf("journald entry name should not exceed %d symbols, got: %q", journaldEntryMaxNameLen, name)
}
if !allowedJournaldEntryNameChars.MatchString(name) {
return fmt.Errorf("journald entry name should consist of `A-Z0-9_` characters and must start from non-digit symbol")
}
if name == cp.TimeField {
n, err := strconv.ParseInt(value, 10, 64)
if err != nil {
return fmt.Errorf("failed to parse Journald timestamp, %w", err)
}
ts = n * 1e3
continue
}
@@ -322,56 +220,18 @@ func readJournaldLogEntry(streamName string, lr *insertutil.LineReader, lmp inse
name = "_msg"
}
if name == "PRIORITY" {
priority := journaldPriorityToLevel(value)
fb.addField("level", priority)
}
if !strings.HasPrefix(name, "__") || *journaldIncludeEntryMetadata {
fb.addField(name, value)
if *journaldIncludeEntryMetadata || !strings.HasPrefix(name, "__") {
fields = append(fields, logstorage.Field{
Name: name,
Value: value,
})
}
}
}
func journaldPriorityToLevel(priority string) string {
// See https://wiki.archlinux.org/title/Systemd/Journal#Priority_level
// and https://grafana.com/docs/grafana/latest/explore/logs-integration/#log-level
switch priority {
case "0":
return "emerg"
case "1":
return "alert"
case "2":
return "critical"
case "3":
return "error"
case "4":
return "warning"
case "5":
return "notice"
case "6":
return "info"
case "7":
return "debug"
default:
return priority
}
}
func isValidFieldName(s string) bool {
if len(s) == 0 {
return false
}
c := s[0]
if !(c >= 'A' && c <= 'Z' || c == '_') {
return false
}
for i := 1; i < len(s); i++ {
c := s[i]
if !(c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_') {
return false
if len(fields) > 0 {
if ts == 0 {
ts = currentTimestamp
}
lmp.AddRow(ts, fields, nil)
}
return true
return nil
}

View File

@@ -1,81 +1,20 @@
package journald
import (
"bytes"
"net/http"
"testing"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
)
func TestIsValidFieldName(t *testing.T) {
f := func(name string, resultExpected bool) {
t.Helper()
result := isValidFieldName(name)
if result != resultExpected {
t.Fatalf("unexpected result for isValidJournaldFieldName(%q); got %v; want %v", name, result, resultExpected)
}
}
f("", false)
f("a", false)
f("1", false)
f("_", true)
f("X", true)
f("Xa", false)
f("X_343", true)
f("X_0123456789_AZ", true)
f("SDDFD sdf", false)
}
func TestGetCommonParams_TimeField(t *testing.T) {
f := func(timeFieldHeader, expectedTimeField string) {
t.Helper()
req, err := http.NewRequest("POST", "/insert/journald/upload", nil)
if err != nil {
t.Fatalf("unexpected error creating request: %s", err)
}
if timeFieldHeader != "" {
req.Header.Set("VL-Time-Field", timeFieldHeader)
}
cp, err := getCommonParams(req)
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
if len(cp.TimeFields) != 1 || cp.TimeFields[0] != expectedTimeField {
t.Fatalf("unexpected TimeFields; got %v; want [%s]", cp.TimeFields, expectedTimeField)
}
}
// Test default behavior - when no custom time field is specified, journald uses __REALTIME_TIMESTAMP
f("", "__REALTIME_TIMESTAMP")
// Test custom time field - when a custom time field is specified via HTTP header, it's respected
f("custom_time", "custom_time")
}
func TestPushJournald_Success(t *testing.T) {
func TestPushJournaldOk(t *testing.T) {
f := func(src string, timestampsExpected []int64, resultExpected string) {
t.Helper()
tlp := &insertutil.TestLogMessageProcessor{}
r, err := http.NewRequest("GET", "https://foo.bar/baz", nil)
if err != nil {
t.Fatalf("cannot create request: %s", err)
cp := &insertutil.CommonParams{
TimeField: "__REALTIME_TIMESTAMP",
MsgFields: []string{"MESSAGE"},
}
cp, err := getCommonParams(r)
if err != nil {
t.Fatalf("cannot create commonParams: %s", err)
}
buf := bytes.NewBufferString(src)
if err := processStreamInternal("test", buf, tlp, cp); err != nil {
if err := parseJournaldRequest([]byte(src), tlp, cp); err != nil {
t.Fatalf("unexpected error: %s", err)
}
@@ -83,17 +22,16 @@ func TestPushJournald_Success(t *testing.T) {
t.Fatal(err)
}
}
// Single event
f("__REALTIME_TIMESTAMP=91723819283\nMESSAGE=Test message\n\n",
f("__REALTIME_TIMESTAMP=91723819283\nMESSAGE=Test message\n",
[]int64{91723819283000},
"{\"_msg\":\"Test message\"}",
)
// Multiple events
f("__REALTIME_TIMESTAMP=91723819283\nPRIORITY=3\nMESSAGE=Test message\n\n__REALTIME_TIMESTAMP=91723819284\nMESSAGE=Test message2\n",
f("__REALTIME_TIMESTAMP=91723819283\nMESSAGE=Test message\n\n__REALTIME_TIMESTAMP=91723819284\nMESSAGE=Test message2\n",
[]int64{91723819283000, 91723819284000},
"{\"level\":\"error\",\"PRIORITY\":\"3\",\"_msg\":\"Test message\"}\n{\"_msg\":\"Test message2\"}",
"{\"_msg\":\"Test message\"}\n{\"_msg\":\"Test message2\"}",
)
// Parse binary data
@@ -101,66 +39,30 @@ func TestPushJournald_Success(t *testing.T) {
[]int64{1729698775704404000},
"{\"E\":\"JobStateChanged\",\"_BOOT_ID\":\"f778b6e2f7584a77b991a2366612a7b5\",\"_UID\":\"0\",\"_GID\":\"0\",\"_MACHINE_ID\":\"a4a970370c30a925df02a13c67167847\",\"_HOSTNAME\":\"ecd5e4555787\",\"_RUNTIME_SCOPE\":\"system\",\"_TRANSPORT\":\"journal\",\"_CAP_EFFECTIVE\":\"1ffffffffff\",\"_SYSTEMD_CGROUP\":\"/init.scope\",\"_SYSTEMD_UNIT\":\"init.scope\",\"_SYSTEMD_SLICE\":\"-.slice\",\"CODE_FILE\":\"\\u003cstdin>\",\"CODE_LINE\":\"1\",\"CODE_FUNC\":\"\\u003cmodule>\",\"SYSLOG_IDENTIFIER\":\"python3\",\"_COMM\":\"python3\",\"_EXE\":\"/usr/bin/python3.12\",\"_CMDLINE\":\"python3\",\"_msg\":\"foo\\nbar\\n\\n\\nasda\\nasda\",\"_PID\":\"2763\",\"_SOURCE_REALTIME_TIMESTAMP\":\"1729698775704375\"}",
)
// Parse binary data with trailing newline
f("__REALTIME_TIMESTAMP=1729698775704404\n_CMDLINE=python3\nMESSAGE\n\x14\x00\x00\x00\x00\x00\x00\x00foo\nbar\n\n\nasda\nasda\n\n_PID=2763\n\n",
[]int64{1729698775704404000},
`{"_CMDLINE":"python3","_msg":"foo\nbar\n\n\nasda\nasda\n","_PID":"2763"}`,
)
f("__REALTIME_TIMESTAMP=1729698775704404\n_CMDLINE=python3\nMESSAGE\n\x00\x00\x00\x00\x00\x00\x00\x00\n_PID=2763\n\n",
[]int64{1729698775704404000},
`{"_CMDLINE":"python3","_PID":"2763"}`,
)
f("__REALTIME_TIMESTAMP=1729698775704404\n_CMDLINE=python3\nMESSAGE\n\x0A\x00\x00\x00\x00\x00\x00\x00123456789\n\n_PID=2763\n\n",
[]int64{1729698775704404000},
`{"_CMDLINE":"python3","_msg":"123456789\n","_PID":"2763"}`,
)
f("__REALTIME_TIMESTAMP=1729698775704404\n_CMDLINE=python3\nMESSAGE\n\x0A\x00\x00\x00\x00\x00\x00\x001234567890\n_PID=2763\n\n",
[]int64{1729698775704404000},
`{"_CMDLINE":"python3","_msg":"1234567890","_PID":"2763"}`,
)
// Empty field name must be ignored
f("__REALTIME_TIMESTAMP=91723819283\na=b\n=Test message", nil, "")
f("__REALTIME_TIMESTAMP=91723819284\nMESSAGE=Test message2\n\n__REALTIME_TIMESTAMP=91723819283\n=Test message\n", []int64{91723819284000}, `{"_msg":"Test message2"}`)
// field name starting with number must be ignored
f("__REALTIME_TIMESTAMP=91723819283\n1incorrect=Test message\n\n__REALTIME_TIMESTAMP=91723819284\nMESSAGE=Test message2\n\n", []int64{91723819284000}, `{"_msg":"Test message2"}`)
// field name exceeding 64 bytes limit must be ignored
f("__REALTIME_TIMESTAMP=91723819283\ntoolooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongcorrecooooooooooooong=Test message\n", nil, "")
// field name with invalid chars must be ignored
f("__REALTIME_TIMESTAMP=91723819283\nbadC!@$!@$as=Test message\n", nil, "")
}
func TestPushJournald_Failure(t *testing.T) {
f := func(data string) {
t.Helper()
tlp := &insertutil.TestLogMessageProcessor{}
r, err := http.NewRequest("GET", "https://foo.bar/baz", nil)
if err != nil {
t.Fatalf("cannot create request: %s", err)
cp := &insertutil.CommonParams{
TimeField: "__REALTIME_TIMESTAMP",
MsgFields: []string{"MESSAGE"},
}
cp, err := getCommonParams(r)
if err != nil {
t.Fatalf("cannot create commonParams: %s", err)
}
buf := bytes.NewBufferString(data)
if err := processStreamInternal("test", buf, tlp, cp); err == nil {
t.Fatalf("expecting non-nil error")
if err := parseJournaldRequest([]byte(data), tlp, cp); err == nil {
t.Fatalf("expected non nil error")
}
}
// too short binary encoded message
f("__CURSOR=s=e0afe8412a6a49d2bfcf66aa7927b588;i=1f06;b=f778b6e2f7584a77b991a2366612a7b5;m=300bdfd420;t=62526e1182354;x=930dc44b370963b7\n__REALTIME_TIMESTAMP=1729698775704404\nMESSAGE\n\x13\x00\x00\x00\x00\x00\x00\x00foo\nbar\n\n\nasdaasd")
f("__REALTIME_TIMESTAMP=1729698775704404\n_CMDLINE=python3\nMESSAGE\n\x00\x00\x00\x00\x00\x00\x00\x00_PID=2763\n\n")
f("__REALTIME_TIMESTAMP=1729698775704404\n_CMDLINE=python3\nMESSAGE\n\x0A\x00\x00\x00\x00\x00\x00\x001234567890_PID=2763\n\n")
f("__REALTIME_TIMESTAMP=1729698775704404\n_CMDLINE=python3\nMESSAGE\n\x0A\x00\x00\x00\x00\x00\x00\x00123456789\n_PID=2763\n\n")
// too long binary encoded message
f("__CURSOR=s=e0afe8412a6a49d2bfcf66aa7927b588;i=1f06;b=f778b6e2f7584a77b991a2366612a7b5;m=300bdfd420;t=62526e1182354;x=930dc44b370963b7\n__REALTIME_TIMESTAMP=1729698775704404\nMESSAGE\n\x13\x00\x00\x00\x00\x00\x00\x00foo\nbar\n\n\nasdaasdakljlsfd")
// missing new line terminator for binary encoded message
f("__CURSOR=s=e0afe8412a6a49d2bfcf66aa7927b588;i=1f06;b=f778b6e2f7584a77b991a2366612a7b5;m=300bdfd420;t=62526e1182354;x=930dc44b370963b7\n__REALTIME_TIMESTAMP=1729698775704404\nMESSAGE\n\x13\x00\x00\x00\x00\x00\x00\x00foo\nbar\n\n\nasdaasda2")
// missing new line terminator
f("__REALTIME_TIMESTAMP=91723819283\n=Test message")
// empty field name
f("__REALTIME_TIMESTAMP=91723819283\n=Test message\n")
// field name starting with number
f("__REALTIME_TIMESTAMP=91723819283\n1incorrect=Test message\n")
// field name exceeds 64 limit
f("__REALTIME_TIMESTAMP=91723819283\ntoolooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooongcorrecooooooooooooong=Test message\n")
// Only allow A-Z0-9 and '_'
f("__REALTIME_TIMESTAMP=91723819283\nbadC!@$!@$as=Test message\n")
}

View File

@@ -1,82 +0,0 @@
package journald
import (
"bytes"
"encoding/binary"
"fmt"
"strings"
"testing"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
)
func BenchmarkIsValidFieldName(b *testing.B) {
b.ReportAllocs()
b.SetBytes(int64(len(benchmarkFields)))
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
for _, field := range benchmarkFields {
if !isValidFieldName(field) {
panic(fmt.Errorf("cannot validate field %q", field))
}
}
}
})
}
var benchmarkFields = strings.Split(
"E,_BOOT_ID,_UID,_GID,_MACHINE_ID,_HOSTNAME,_RUNTIME_SCOPE,_TRANSPORT,_CAP_EFFECTIVE,_SYSTEMD_CGROUP,_SYSTEMD_UNIT,"+
"_SYSTEMD_SLICE,CODE_FILE,CODE_LINE,CODE_FUNC,SYSLOG_IDENTIFIER,_COMM,_EXE,_CMDLINE,MESSAGE,_PID,_SOURCE_REALTIME_TIMESTAMP,_REALTIME_TIMESTAMP",
",")
func BenchmarkPushJournaldPerformance(b *testing.B) {
cp := &insertutil.CommonParams{
TimeFields: []string{"__REALTIME_TIMESTAMP"},
MsgFields: []string{"MESSAGE"},
}
const dataChunkSize = 1024 * 1024
data := generateJournaldData(dataChunkSize)
b.ReportAllocs()
b.SetBytes(int64(len(data)))
b.RunParallel(func(pb *testing.PB) {
r := &bytes.Reader{}
blp := &insertutil.BenchmarkLogMessageProcessor{}
for pb.Next() {
r.Reset(data)
if err := processStreamInternal("performance_test", r, blp, cp); err != nil {
panic(fmt.Errorf("unexpected error: %w", err))
}
}
})
}
func generateJournaldData(size int) []byte {
var buf []byte
timestamp := time.Now().UnixMicro()
binaryMsg := []byte("binary message data for performance test")
var sizeBuf [8]byte
for len(buf) < size {
timestamp++
var entry string
// Generate a mix of simple and binary messages
if timestamp%10 == 0 {
// Generate binary message
binary.LittleEndian.PutUint64(sizeBuf[:], uint64(len(binaryMsg)))
entry = fmt.Sprintf("__REALTIME_TIMESTAMP=%d\nMESSAGE\n%s%s\n\n",
timestamp,
sizeBuf[:],
binaryMsg,
)
} else {
// Generate simple message
entry = fmt.Sprintf("__REALTIME_TIMESTAMP=%d\nMESSAGE=Performance test message %d\n\n", timestamp, timestamp)
}
buf = append(buf, entry...)
}
return buf
}

View File

@@ -7,6 +7,7 @@ import (
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
@@ -32,7 +33,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
httpserver.Errorf(w, r, "%s", err)
return
}
if err := insertutil.CanWriteData(); err != nil {
if err := vlstorage.CanWriteData(); err != nil {
httpserver.Errorf(w, r, "%s", err)
return
}
@@ -47,49 +48,34 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
lmp := cp.NewLogMessageProcessor("jsonline", true)
streamName := fmt.Sprintf("remoteAddr=%s, requestURI=%q", httpserver.GetQuotedRemoteAddr(r), r.RequestURI)
err = processStreamInternal(streamName, reader, cp.TimeFields, cp.MsgFields, lmp)
processStreamInternal(streamName, reader, cp.TimeField, cp.MsgFields, lmp)
lmp.MustClose()
if err != nil {
httpserver.Errorf(w, r, "cannot process jsonline request; error: %s", err)
return
}
requestDuration.UpdateDuration(startTime)
}
func processStreamInternal(streamName string, r io.Reader, timeFields, msgFields []string, lmp insertutil.LogMessageProcessor) error {
func processStreamInternal(streamName string, r io.Reader, timeField string, msgFields []string, lmp insertutil.LogMessageProcessor) {
wcr := writeconcurrencylimiter.GetReader(r)
defer writeconcurrencylimiter.PutReader(wcr)
lr := insertutil.NewLineReader(streamName, wcr)
n := 0
errors := 0
var lastError error
for {
ok, err := readLine(lr, timeFields, msgFields, lmp)
ok, err := readLine(lr, timeField, msgFields, lmp)
wcr.DecConcurrency()
if err != nil {
lastError = err
errors++
errorsTotal.Inc()
logger.Warnf("jsonline: cannot read line #%d in /jsonline request: %s", n, err)
}
if !ok {
break
return
}
n++
}
errorsTotal.Add(errors)
if errors > 0 && n == errors {
// Return an error if no logs were processed and there were errors
return lastError
}
return nil
}
func readLine(lr *insertutil.LineReader, timeFields, msgFields []string, lmp insertutil.LogMessageProcessor) (bool, error) {
func readLine(lr *insertutil.LineReader, timeField string, msgFields []string, lmp insertutil.LogMessageProcessor) (bool, error) {
var line []byte
for len(line) == 0 {
if !lr.NextLine() {
@@ -103,11 +89,11 @@ func readLine(lr *insertutil.LineReader, timeFields, msgFields []string, lmp ins
defer logstorage.PutJSONParser(p)
if err := p.ParseLogMessage(line); err != nil {
return true, fmt.Errorf("%s; line contents: %q", err, line)
return true, fmt.Errorf("cannot parse json-encoded line: %w; line contents: %q", err, line)
}
ts, err := insertutil.ExtractTimestampFromFields(timeFields, p.Fields)
ts, err := insertutil.ExtractTimestampFromFields(timeField, p.Fields)
if err != nil {
return true, fmt.Errorf("%s; line contents: %q", err, line)
return true, fmt.Errorf("cannot get timestamp from json-encoded line: %w; line contents: %q", err, line)
}
logstorage.RenameField(p.Fields, msgFields, "_msg")
lmp.AddRow(ts, p.Fields, nil)
@@ -119,5 +105,5 @@ var (
requestsTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/jsonline"}`)
errorsTotal = metrics.NewCounter(`vl_http_errors_total{path="/insert/jsonline"}`)
requestDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/jsonline"}`)
requestDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/jsonline"}`)
)

View File

@@ -2,23 +2,19 @@ package jsonline
import (
"bytes"
"strings"
"testing"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
)
func TestProcessStreamInternalSuccess(t *testing.T) {
func TestProcessStreamInternal(t *testing.T) {
f := func(data, timeField, msgField string, timestampsExpected []int64, resultExpected string) {
t.Helper()
timeFields := []string{timeField}
msgFields := []string{msgField}
tlp := &insertutil.TestLogMessageProcessor{}
r := bytes.NewBufferString(data)
if err := processStreamInternal("test", r, timeFields, msgFields, tlp); err != nil {
t.Fatalf("unexpected error: %s", err)
}
processStreamInternal("test", r, timeField, msgFields, tlp)
if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
t.Fatal(err)
@@ -48,6 +44,22 @@ func TestProcessStreamInternalSuccess(t *testing.T) {
{"message":"baz"}`
f(data, timeField, msgField, timestampsExpected, resultExpected)
// invalid json
data = "foobar"
timeField = "@timestamp"
msgField = "aaa"
timestampsExpected = nil
resultExpected = ``
f(data, timeField, msgField, timestampsExpected, resultExpected)
// invalid timestamp field
data = `{"time":"foobar"}`
timeField = "time"
msgField = "abc"
timestampsExpected = nil
resultExpected = ``
f(data, timeField, msgField, timestampsExpected, resultExpected)
// invalid lines among valid lines
data = `
dsfodmasd
@@ -65,33 +77,3 @@ asbsdf
{"_msg":"baz"}`
f(data, timeField, msgField, timestampsExpected, resultExpected)
}
func TestProcessStreamInternalFailure(t *testing.T) {
f := func(data string) {
t.Helper()
tlp := &insertutil.TestLogMessageProcessor{}
r := strings.NewReader(data)
if err := processStreamInternal("test", r, []string{"time"}, nil, tlp); err == nil {
t.Fatalf("expected error, got nil")
}
if err := tlp.Verify(nil, ""); err != nil {
t.Fatalf("unexpected error: %s", err)
}
}
// invalid json
f("foobar")
f(`foo
bar`)
f(`
foo
`)
// invalid timestamp field
f(`{"time":"foobar"}`)
}

View File

@@ -16,10 +16,10 @@ var disableMessageParsing = flag.Bool("loki.disableMessageParsing", false, "Whet
// RequestHandler processes Loki insert requests
func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
switch path {
case "/insert/loki/api/v1/push":
case "/api/v1/push":
handleInsert(r, w)
return true
case "/insert/loki/ready":
case "/ready":
// See https://grafana.com/docs/loki/latest/api/#identify-ready-loki-instance
w.WriteHeader(http.StatusOK)
w.Write([]byte("ready"))

View File

@@ -9,6 +9,7 @@ import (
"github.com/valyala/fastjson"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -29,7 +30,7 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
return
}
if err := insertutil.CanWriteData(); err != nil {
if err := vlstorage.CanWriteData(); err != nil {
httpserver.Errorf(w, r, "%s", err)
return
}
@@ -58,7 +59,7 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
var (
requestsJSONTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="json"}`)
requestJSONDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="json"}`)
requestJSONDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="json"}`)
)
func parseJSONRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields, parseMessage bool) error {

View File

@@ -9,6 +9,7 @@ import (
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
@@ -28,7 +29,7 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
return
}
if err := insertutil.CanWriteData(); err != nil {
if err := vlstorage.CanWriteData(); err != nil {
httpserver.Errorf(w, r, "%s", err)
return
}
@@ -62,7 +63,7 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
var (
requestsProtobufTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="protobuf"}`)
requestProtobufDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="protobuf"}`)
requestProtobufDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="protobuf"}`)
)
func parseProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields, parseMessage bool) error {

View File

@@ -1,25 +1,17 @@
package vlinsert
import (
"flag"
"fmt"
"net/http"
"strings"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/datadog"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/elasticsearch"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/internalinsert"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/journald"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/jsonline"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/loki"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/opentelemetry"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/syslog"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
)
var (
disableInsert = flag.Bool("insert.disable", false, "Whether to disable /insert/* HTTP endpoints")
disableInternal = flag.Bool("internalinsert.disable", false, "Whether to disable /internal/insert HTTP endpoint")
)
// Init initializes vlinsert
@@ -34,55 +26,44 @@ func Stop() {
// RequestHandler handles insert requests for VictoriaLogs
func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
path := strings.ReplaceAll(r.URL.Path, "//", "/")
path := r.URL.Path
if strings.HasPrefix(path, "/insert/") {
if *disableInsert {
httpserver.Errorf(w, r, "requests to /insert/* are disabled with -insert.disable command-line flag")
return true
}
return insertHandler(w, r, path)
if !strings.HasPrefix(path, "/insert/") {
// Skip requests, which do not start with /insert/, since these aren't our requests.
return false
}
path = strings.TrimPrefix(path, "/insert")
path = strings.ReplaceAll(path, "//", "/")
if path == "/internal/insert" {
if *disableInternal || *disableInsert {
httpserver.Errorf(w, r, "requests to /internal/insert are disabled with -internalinsert.disable or -insert.disable command-line flag")
return true
}
internalinsert.RequestHandler(w, r)
return true
}
return false
}
func insertHandler(w http.ResponseWriter, r *http.Request, path string) bool {
switch path {
case "/insert/jsonline":
case "/jsonline":
jsonline.RequestHandler(w, r)
return true
case "/insert/ready":
case "/ready":
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(200)
fmt.Fprintf(w, `{"status":"ok"}`)
return true
}
switch {
// some clients may omit trailing slash at elasticsearch protocol.
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8353
case strings.HasPrefix(path, "/insert/elasticsearch"):
case strings.HasPrefix(path, "/elasticsearch"):
// some clients may omit trailing slash
// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8353
path = strings.TrimPrefix(path, "/elasticsearch")
return elasticsearch.RequestHandler(path, w, r)
case strings.HasPrefix(path, "/insert/loki/"):
case strings.HasPrefix(path, "/loki/"):
path = strings.TrimPrefix(path, "/loki")
return loki.RequestHandler(path, w, r)
case strings.HasPrefix(path, "/insert/opentelemetry/"):
case strings.HasPrefix(path, "/opentelemetry/"):
path = strings.TrimPrefix(path, "/opentelemetry")
return opentelemetry.RequestHandler(path, w, r)
case strings.HasPrefix(path, "/insert/journald/"):
case strings.HasPrefix(path, "/journald/"):
path = strings.TrimPrefix(path, "/journald")
return journald.RequestHandler(path, w, r)
case strings.HasPrefix(path, "/insert/datadog/"):
case strings.HasPrefix(path, "/datadog/"):
path = strings.TrimPrefix(path, "/datadog")
return datadog.RequestHandler(path, w, r)
default:
return false
}
return false
}

View File

@@ -6,11 +6,13 @@ import (
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/pb"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
"github.com/VictoriaMetrics/metrics"
)
@@ -21,7 +23,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
switch path {
// use the same path as opentelemetry collector
// https://opentelemetry.io/docs/specs/otlp/#otlphttp-request
case "/insert/opentelemetry/v1/logs":
case "/v1/logs":
if r.Header.Get("Content-Type") == "application/json" {
httpserver.Errorf(w, r, "json encoding isn't supported for opentelemetry format. Use protobuf encoding")
return true
@@ -42,7 +44,7 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
return
}
if err := insertutil.CanWriteData(); err != nil {
if err := vlstorage.CanWriteData(); err != nil {
httpserver.Errorf(w, r, "%s", err)
return
}
@@ -51,7 +53,7 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
lmp := cp.NewLogMessageProcessor("opentelelemtry_protobuf", false)
useDefaultStreamFields := len(cp.StreamFields) == 0
err := pushProtobufRequest(data, lmp, cp.MsgFields, useDefaultStreamFields)
err := pushProtobufRequest(data, lmp, useDefaultStreamFields)
lmp.MustClose()
return err
})
@@ -70,10 +72,10 @@ var (
requestsProtobufTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/opentelemetry/v1/logs",format="protobuf"}`)
errorsTotal = metrics.NewCounter(`vl_http_errors_total{path="/insert/opentelemetry/v1/logs",format="protobuf"}`)
requestProtobufDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/opentelemetry/v1/logs",format="protobuf"}`)
requestProtobufDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/opentelemetry/v1/logs",format="protobuf"}`)
)
func pushProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields bool) error {
func pushProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, useDefaultStreamFields bool) error {
var req pb.ExportLogsServiceRequest
if err := req.UnmarshalProtobuf(data); err != nil {
errorsTotal.Inc()
@@ -82,31 +84,35 @@ func pushProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFie
var commonFields []logstorage.Field
for _, rl := range req.ResourceLogs {
commonFields = commonFields[:0]
commonFields = appendKeyValues(commonFields, rl.Resource.Attributes, "")
attributes := rl.Resource.Attributes
commonFields = slicesutil.SetLength(commonFields, len(attributes))
for i, attr := range attributes {
commonFields[i].Name = attr.Key
commonFields[i].Value = attr.Value.FormatString(true)
}
commonFieldsLen := len(commonFields)
for _, sc := range rl.ScopeLogs {
commonFields = pushFieldsFromScopeLogs(&sc, commonFields[:commonFieldsLen], lmp, msgFields, useDefaultStreamFields)
commonFields = pushFieldsFromScopeLogs(&sc, commonFields[:commonFieldsLen], lmp, useDefaultStreamFields)
}
}
return nil
}
func pushFieldsFromScopeLogs(sc *pb.ScopeLogs, commonFields []logstorage.Field, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields bool) []logstorage.Field {
func pushFieldsFromScopeLogs(sc *pb.ScopeLogs, commonFields []logstorage.Field, lmp insertutil.LogMessageProcessor, useDefaultStreamFields bool) []logstorage.Field {
fields := commonFields
for _, lr := range sc.LogRecords {
fields = fields[:len(commonFields)]
if lr.Body.KeyValueList != nil {
fields = appendKeyValues(fields, lr.Body.KeyValueList.Values, "")
logstorage.RenameField(fields[len(commonFields):], msgFields, "_msg")
} else {
fields = append(fields, logstorage.Field{
Name: "_msg",
Value: lr.Body.FormatString(true),
})
for _, attr := range lr.Attributes {
fields = append(fields, logstorage.Field{
Name: "_msg",
Value: lr.Body.FormatString(true),
Name: attr.Key,
Value: attr.Value.FormatString(true),
})
}
fields = appendKeyValues(fields, lr.Attributes, "")
if len(lr.TraceID) > 0 {
fields = append(fields, logstorage.Field{
Name: "trace_id",
@@ -132,22 +138,3 @@ func pushFieldsFromScopeLogs(sc *pb.ScopeLogs, commonFields []logstorage.Field,
}
return fields
}
func appendKeyValues(fields []logstorage.Field, kvs []*pb.KeyValue, parentField string) []logstorage.Field {
for _, attr := range kvs {
fieldName := attr.Key
if parentField != "" {
fieldName = parentField + "." + fieldName
}
if attr.Value.KeyValueList != nil {
fields = appendKeyValues(fields, attr.Value.KeyValueList.Values, fieldName)
} else {
fields = append(fields, logstorage.Field{
Name: fieldName,
Value: attr.Value.FormatString(true),
})
}
}
return fields
}

View File

@@ -16,7 +16,7 @@ func TestPushProtoOk(t *testing.T) {
pData := lr.MarshalProtobuf(nil)
tlp := &insertutil.TestLogMessageProcessor{}
if err := pushProtobufRequest(pData, tlp, nil, false); err != nil {
if err := pushProtobufRequest(pData, tlp, false); err != nil {
t.Fatalf("unexpected error: %s", err)
}
@@ -41,26 +41,6 @@ func TestPushProtoOk(t *testing.T) {
`{"_msg":"log-line-message","severity":"Trace"}`,
)
// severities mapping
f([]pb.ResourceLogs{
{
ScopeLogs: []pb.ScopeLogs{
{
LogRecords: []pb.LogRecord{
{Attributes: []*pb.KeyValue{}, TimeUnixNano: 1234, SeverityNumber: 1, Body: pb.AnyValue{StringValue: ptrTo("log-line-message")}},
{Attributes: []*pb.KeyValue{}, TimeUnixNano: 1234, SeverityNumber: 13, Body: pb.AnyValue{StringValue: ptrTo("log-line-message")}},
{Attributes: []*pb.KeyValue{}, TimeUnixNano: 1234, SeverityNumber: 24, Body: pb.AnyValue{StringValue: ptrTo("log-line-message")}},
},
},
},
},
},
[]int64{1234, 1234, 1234},
`{"_msg":"log-line-message","severity":"Trace"}
{"_msg":"log-line-message","severity":"Warn"}
{"_msg":"log-line-message","severity":"Fatal4"}`,
)
// multi-line with resource attributes
f([]pb.ResourceLogs{
{
@@ -80,7 +60,7 @@ func TestPushProtoOk(t *testing.T) {
{
LogRecords: []pb.LogRecord{
{Attributes: []*pb.KeyValue{}, TimeUnixNano: 1234, SeverityNumber: 1, Body: pb.AnyValue{StringValue: ptrTo("log-line-message")}},
{Attributes: []*pb.KeyValue{}, TimeUnixNano: 1235, SeverityNumber: 25, Body: pb.AnyValue{StringValue: ptrTo("log-line-message-msg-2")}},
{Attributes: []*pb.KeyValue{}, TimeUnixNano: 1235, SeverityNumber: 21, Body: pb.AnyValue{StringValue: ptrTo("log-line-message-msg-2")}},
{Attributes: []*pb.KeyValue{}, TimeUnixNano: 1236, SeverityNumber: -1, Body: pb.AnyValue{StringValue: ptrTo("log-line-message-msg-2")}},
},
},
@@ -88,9 +68,9 @@ func TestPushProtoOk(t *testing.T) {
},
},
[]int64{1234, 1235, 1236},
`{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","_msg":"log-line-message","severity":"Trace"}
{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","_msg":"log-line-message-msg-2","severity":"Unspecified"}
{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","_msg":"log-line-message-msg-2","severity":"Unspecified"}`,
`{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message","severity":"Trace"}
{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Unspecified"}
{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Unspecified"}`,
)
// multi-scope with resource attributes and multi-line
@@ -136,71 +116,14 @@ func TestPushProtoOk(t *testing.T) {
},
},
[]int64{1234, 1235, 2345, 2346, 2347, 2348, 3333},
`{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","_msg":"log-line-message","severity":"Trace"}
{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","_msg":"log-line-message-msg-2","severity":"Debug"}
`{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message","severity":"Trace"}
{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Debug"}
{"_msg":"log-line-resource-scope-1-0-0","severity":"Info2"}
{"_msg":"log-line-resource-scope-1-0-1","severity":"Info2"}
{"_msg":"log-line-resource-scope-1-1-0","severity":"Info4"}
{"_msg":"log-line-resource-scope-1-1-1","trace_id":"1234","span_id":"45","severity":"Info4"}
{"_msg":"log-line-resource-scope-1-1-2","trace_id":"4bf92f3577b34da6a3ce929d0e0e4736","span_id":"00f067aa0ba902b7","severity":"Unspecified"}`,
)
// nested fields
f([]pb.ResourceLogs{
{
ScopeLogs: []pb.ScopeLogs{
{
LogRecords: []pb.LogRecord{
{
TimeUnixNano: 1234,
Body: pb.AnyValue{StringValue: ptrTo("nested fields")},
Attributes: []*pb.KeyValue{
{Key: "error", Value: &pb.AnyValue{KeyValueList: &pb.KeyValueList{Values: []*pb.KeyValue{
{
Key: "type",
Value: &pb.AnyValue{StringValue: ptrTo("document_parsing_exception")},
},
{
Key: "reason",
Value: &pb.AnyValue{StringValue: ptrTo("failed to parse field [_msg] of type [text]")},
},
{
Key: "caused_by",
Value: &pb.AnyValue{KeyValueList: &pb.KeyValueList{Values: []*pb.KeyValue{
{
Key: "type",
Value: &pb.AnyValue{StringValue: ptrTo("x_content_parse_exception")},
},
{
Key: "reason",
Value: &pb.AnyValue{StringValue: ptrTo("unexpected end-of-input in VALUE_STRING")},
},
{
Key: "caused_by",
Value: &pb.AnyValue{KeyValueList: &pb.KeyValueList{Values: []*pb.KeyValue{
{
Key: "type",
Value: &pb.AnyValue{StringValue: ptrTo("json_e_o_f_exception")},
},
{
Key: "reason",
Value: &pb.AnyValue{StringValue: ptrTo("eof")},
},
}}},
},
}}},
},
}}}},
},
},
},
},
},
},
}, []int64{1234},
`{"_msg":"nested fields","error.type":"document_parsing_exception","error.reason":"failed to parse field [_msg] of type [text]",`+
`"error.caused_by.type":"x_content_parse_exception","error.caused_by.reason":"unexpected end-of-input in VALUE_STRING",`+
`"error.caused_by.caused_by.type":"json_e_o_f_exception","error.caused_by.caused_by.reason":"eof","severity":"Unspecified"}`)
}
func ptrTo[T any](s T) *T {

View File

@@ -27,7 +27,7 @@ func benchmarkParseProtobufRequest(b *testing.B, streams, rows, labels int) {
b.RunParallel(func(pb *testing.PB) {
body := getProtobufBody(streams, rows, labels)
for pb.Next() {
if err := pushProtobufRequest(body, blp, nil, false); err != nil {
if err := pushProtobufRequest(body, blp, false); err != nil {
panic(fmt.Errorf("unexpected error: %w", err))
}
}

View File

@@ -17,6 +17,7 @@ import (
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
@@ -44,11 +45,6 @@ var (
ignoreFieldsUDP = flagutil.NewArrayString("syslog.ignoreFields.udp", "Fields to ignore at logs ingested via the corresponding -syslog.listenAddr.udp. "+
`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#dropping-fields`)
decolorizeFieldsTCP = flagutil.NewArrayString("syslog.decolorizeFields.tcp", "Fields to remove ANSI color codes across logs ingested via the corresponding -syslog.listenAddr.tcp. "+
`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#decolorizing-fields`)
decolorizeFieldsUDP = flagutil.NewArrayString("syslog.decolorizeFields.udp", "Fields to remove ANSI color codes across logs ingested via the corresponding -syslog.listenAddr.udp. "+
`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#decolorizing-fields`)
extraFieldsTCP = flagutil.NewArrayString("syslog.extraFields.tcp", "Fields to add to logs ingested via the corresponding -syslog.listenAddr.tcp. "+
`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#adding-extra-fields`)
extraFieldsUDP = flagutil.NewArrayString("syslog.extraFields.udp", "Fields to add to logs ingested via the corresponding -syslog.listenAddr.udp. "+
@@ -192,12 +188,6 @@ func runUDPListener(addr string, argIdx int) {
logger.Fatalf("cannot parse -syslog.ignoreFields.udp=%q for -syslog.listenAddr.udp=%q: %s", ignoreFieldsStr, addr, err)
}
decolorizeFieldsStr := decolorizeFieldsUDP.GetOptionalArg(argIdx)
decolorizeFields, err := parseFieldsList(decolorizeFieldsStr)
if err != nil {
logger.Fatalf("cannot parse -syslog.decolorizeFields.udp=%q for -syslog.listenAddr.udp=%q: %s", decolorizeFieldsStr, addr, err)
}
extraFieldsStr := extraFieldsUDP.GetOptionalArg(argIdx)
extraFields, err := parseExtraFields(extraFieldsStr)
if err != nil {
@@ -206,7 +196,7 @@ func runUDPListener(addr string, argIdx int) {
doneCh := make(chan struct{})
go func() {
serveUDP(ln, tenantID, compressMethod, useLocalTimestamp, streamFields, ignoreFields, decolorizeFields, extraFields)
serveUDP(ln, tenantID, compressMethod, useLocalTimestamp, streamFields, ignoreFields, extraFields)
close(doneCh)
}()
@@ -259,12 +249,6 @@ func runTCPListener(addr string, argIdx int) {
logger.Fatalf("cannot parse -syslog.ignoreFields.tcp=%q for -syslog.listenAddr.tcp=%q: %s", ignoreFieldsStr, addr, err)
}
decolorizeFieldsStr := decolorizeFieldsTCP.GetOptionalArg(argIdx)
decolorizeFields, err := parseFieldsList(decolorizeFieldsStr)
if err != nil {
logger.Fatalf("cannot parse -syslog.decolorizeFields.tcp=%q for -syslog.listenAddr.tcp=%q: %s", decolorizeFieldsStr, addr, err)
}
extraFieldsStr := extraFieldsTCP.GetOptionalArg(argIdx)
extraFields, err := parseExtraFields(extraFieldsStr)
if err != nil {
@@ -273,7 +257,7 @@ func runTCPListener(addr string, argIdx int) {
doneCh := make(chan struct{})
go func() {
serveTCP(ln, tenantID, compressMethod, useLocalTimestamp, streamFields, ignoreFields, decolorizeFields, extraFields)
serveTCP(ln, tenantID, compressMethod, useLocalTimestamp, streamFields, ignoreFields, extraFields)
close(doneCh)
}()
@@ -295,7 +279,7 @@ func checkCompressMethod(compressMethod, addr, protocol string) {
}
}
func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string, useLocalTimestamp bool, streamFields, ignoreFields, decolorizeFields []string, extraFields []logstorage.Field) {
func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
gomaxprocs := cgroup.AvailableCPUs()
var wg sync.WaitGroup
localAddr := ln.LocalAddr()
@@ -303,7 +287,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string,
wg.Add(1)
go func() {
defer wg.Done()
cp := insertutil.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, decolorizeFields, extraFields)
cp := insertutil.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
var bb bytesutil.ByteBuffer
bb.B = bytesutil.ResizeNoCopyNoOverallocate(bb.B, 64*1024)
for {
@@ -337,7 +321,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string,
wg.Wait()
}
func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, useLocalTimestamp bool, streamFields, ignoreFields, decolorizeFields []string, extraFields []logstorage.Field) {
func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
var cm ingestserver.ConnsMap
cm.Init("syslog")
@@ -367,7 +351,7 @@ func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, us
wg.Add(1)
go func() {
cp := insertutil.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, decolorizeFields, extraFields)
cp := insertutil.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
if err := processStream("tcp", c, encoding, useLocalTimestamp, cp); err != nil {
logger.Errorf("syslog: cannot process TCP data at %q: %s", addr, err)
}
@@ -384,7 +368,7 @@ func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, us
// processStream parses a stream of syslog messages from r and ingests them into vlstorage.
func processStream(protocol string, r io.Reader, encoding string, useLocalTimestamp bool, cp *insertutil.CommonParams) error {
if err := insertutil.CanWriteData(); err != nil {
if err := vlstorage.CanWriteData(); err != nil {
return err
}
@@ -551,7 +535,7 @@ func processLine(line []byte, currentYear int, timezone *time.Location, useLocal
if useLocalTimestamp {
ts = time.Now().UnixNano()
} else {
nsecs, err := insertutil.ExtractTimestampFromFields(timeFields, p.Fields)
nsecs, err := insertutil.ExtractTimestampFromFields("timestamp", p.Fields)
if err != nil {
return fmt.Errorf("cannot get timestamp from syslog line %q: %w", line, err)
}
@@ -564,7 +548,6 @@ func processLine(line []byte, currentYear int, timezone *time.Location, useLocal
return nil
}
var timeFields = []string{"timestamp"}
var msgFields = []string{"message"}
var (

View File

@@ -101,8 +101,8 @@ func TestProcessStreamInternal_Success(t *testing.T) {
currentYear := 2023
timestampsExpected := []int64{1685794113000000000, 1685880513000000000, 1685814132345000000}
resultExpected := `{"format":"rfc3164","hostname":"abcd","app_name":"systemd","_msg":"Starting Update the local ESM caches..."}
{"priority":"165","facility_keyword":"local4","level":"notice","facility":"20","severity":"5","format":"rfc3164","hostname":"abcd","app_name":"systemd","proc_id":"345","_msg":"abc defg"}
{"priority":"123","facility_keyword":"solaris-cron","level":"error","facility":"15","severity":"3","format":"rfc5424","hostname":"mymachine.example.com","app_name":"appname","proc_id":"12345","msg_id":"ID47","exampleSDID@32473.iut":"3","exampleSDID@32473.eventSource":"Application 123 = ] 56","exampleSDID@32473.eventID":"11211","_msg":"This is a test message with structured data."}`
{"priority":"165","facility":"20","severity":"5","format":"rfc3164","hostname":"abcd","app_name":"systemd","proc_id":"345","_msg":"abc defg"}
{"priority":"123","facility":"15","severity":"3","format":"rfc5424","hostname":"mymachine.example.com","app_name":"appname","proc_id":"12345","msg_id":"ID47","exampleSDID@32473.iut":"3","exampleSDID@32473.eventSource":"Application 123 = ] 56","exampleSDID@32473.eventID":"11211","_msg":"This is a test message with structured data."}`
f(data, currentYear, timestampsExpected, resultExpected)
}

View File

@@ -17,7 +17,7 @@ func isTerminal() bool {
return isatty.IsTerminal(os.Stdout.Fd()) && isatty.IsTerminal(os.Stderr.Fd())
}
func readWithLess(r io.Reader, disableColors, wrapLongLines bool) error {
func readWithLess(r io.Reader, wrapLongLines bool) error {
if !isTerminal() {
// Just write everything to stdout if no terminal is available.
_, err := io.Copy(os.Stdout, r)
@@ -49,9 +49,6 @@ func readWithLess(r io.Reader, disableColors, wrapLongLines bool) error {
return fmt.Errorf("cannot find 'less' command: %w", err)
}
opts := []string{"less", "-F", "-X"}
if !disableColors {
opts = append(opts, "-R")
}
if !wrapLongLines {
opts = append(opts, "-S")
}

View File

@@ -91,7 +91,6 @@ func runReadlineLoop(rl *readline.Instance, incompleteLine *string) {
}
outputMode := outputModeJSONMultiline
disableColors := true
wrapLongLines := false
s := ""
for {
@@ -101,7 +100,7 @@ func runReadlineLoop(rl *readline.Instance, incompleteLine *string) {
case io.EOF:
if s != "" {
// This is non-interactive query execution.
executeQuery(context.Background(), rl, s, outputMode, disableColors, wrapLongLines)
executeQuery(context.Background(), rl, s, outputMode, wrapLongLines)
}
return
case readline.ErrInterrupt:
@@ -177,24 +176,6 @@ func runReadlineLoop(rl *readline.Instance, incompleteLine *string) {
s = ""
continue
}
if s == `\disable_colors` {
if !disableColors {
disableColors = true
fmt.Fprintf(rl, `disabled colors in compact output mode; enter \enable_colors for enabling it`+"\n")
}
historyLines = pushToHistory(rl, historyLines, s)
s = ""
continue
}
if s == `\enable_colors` {
if disableColors {
disableColors = false
fmt.Fprintf(rl, `enabled colors in compact output mode; type \disable_colors for disabling it`+"\n")
}
historyLines = pushToHistory(rl, historyLines, s)
s = ""
continue
}
if line != "" && !strings.HasSuffix(line, ";") {
// Assume the query is incomplete and allow the user finishing the query on the next line
s += "\n"
@@ -204,7 +185,7 @@ func runReadlineLoop(rl *readline.Instance, incompleteLine *string) {
// Execute the query
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
executeQuery(ctx, rl, s, outputMode, disableColors, wrapLongLines)
executeQuery(ctx, rl, s, outputMode, wrapLongLines)
cancel()
historyLines = pushToHistory(rl, historyLines, s)
@@ -292,15 +273,13 @@ func printCommandsHelp(w io.Writer) {
\c - compact output mode
\logfmt - logfmt output mode
\wrap_long_lines - toggles wrapping long lines
\enable_colors - enable ANSI colors in compact output mode
\disable_colors - disable ANSI colors in compact output mode
\tail <query> - live tail <query> results
See https://docs.victoriametrics.com/victorialogs/querying/vlogscli/ for more details
`)
}
func executeQuery(ctx context.Context, output io.Writer, qStr string, outputMode outputMode, disableColors, wrapLongLines bool) {
func executeQuery(ctx context.Context, output io.Writer, qStr string, outputMode outputMode, wrapLongLines bool) {
if strings.HasPrefix(qStr, `\tail `) {
tailQuery(ctx, output, qStr, outputMode)
return
@@ -314,7 +293,7 @@ func executeQuery(ctx context.Context, output io.Writer, qStr string, outputMode
_ = respBody.Close()
}()
if err := readWithLess(respBody, disableColors, wrapLongLines); err != nil {
if err := readWithLess(respBody, wrapLongLines); err != nil {
fmt.Fprintf(output, "error when reading query response: %s\n", err)
return
}

View File

@@ -25,8 +25,8 @@ var (
addr = flag.String("addr", "stdout", "HTTP address to push the generated logs to; if it is set to stdout, then logs are generated to stdout")
workers = flag.Int("workers", 1, "The number of workers to use to push logs to -addr")
start = newTimeFlag("start", "-1d", "Generated logs start from this time; see https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#timestamp-formats")
end = newTimeFlag("end", "0s", "Generated logs end at this time; see https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#timestamp-formats")
start = newTimeFlag("start", "-1d", "Generated logs start from this time; see https://docs.victoriametrics.com/#timestamp-formats")
end = newTimeFlag("end", "0s", "Generated logs end at this time; see https://docs.victoriametrics.com/#timestamp-formats")
activeStreams = flag.Int("activeStreams", 100, "The number of active log streams to generate; see https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields")
totalStreams = flag.Int("totalStreams", 0, "The number of total log streams; if -totalStreams > -activeStreams, then some active streams are substituted with new streams "+
"during data generation")
@@ -206,8 +206,6 @@ func generateAndPushLogs(cfg *workerConfig, workerID int) {
if err != nil {
logger.Fatalf("cannot perform request to %q: %s", cfg.url, err)
}
defer resp.Body.Close()
if resp.StatusCode/100 != 2 {
logger.Fatalf("unexpected status code got from %q: %d; want 2xx", cfg.url, err)
}

View File

@@ -1,316 +0,0 @@
package internalselect
import (
"context"
"fmt"
"net/http"
"strconv"
"sync"
"time"
"github.com/VictoriaMetrics/metrics"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage/netselect"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/atomicutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
)
// RequestHandler processes requests to /internal/select/*
func RequestHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
path := r.URL.Path
rh := requestHandlers[path]
if rh == nil {
httpserver.Errorf(w, r, "unsupported endpoint requested: %s", path)
return
}
metrics.GetOrCreateCounter(fmt.Sprintf(`vl_http_requests_total{path=%q}`, path)).Inc()
if err := rh(ctx, w, r); err != nil && !netutil.IsTrivialNetworkError(err) {
metrics.GetOrCreateCounter(fmt.Sprintf(`vl_http_request_errors_total{path=%q}`, path)).Inc()
httpserver.Errorf(w, r, "%s", err)
// The return is skipped intentionally in order to track the duration of failed queries.
}
metrics.GetOrCreateSummary(fmt.Sprintf(`vl_http_request_duration_seconds{path=%q}`, path)).UpdateDuration(startTime)
}
var requestHandlers = map[string]func(ctx context.Context, w http.ResponseWriter, r *http.Request) error{
"/internal/select/query": processQueryRequest,
"/internal/select/field_names": processFieldNamesRequest,
"/internal/select/field_values": processFieldValuesRequest,
"/internal/select/stream_field_names": processStreamFieldNamesRequest,
"/internal/select/stream_field_values": processStreamFieldValuesRequest,
"/internal/select/streams": processStreamsRequest,
"/internal/select/stream_ids": processStreamIDsRequest,
}
func processQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
cp, err := getCommonParams(r, netselect.QueryProtocolVersion)
if err != nil {
return err
}
w.Header().Set("Content-Type", "application/octet-stream")
var wLock sync.Mutex
var dataLenBuf []byte
sendBuf := func(bb *bytesutil.ByteBuffer) error {
if len(bb.B) == 0 {
return nil
}
data := bb.B
if !cp.DisableCompression {
bufLen := len(bb.B)
bb.B = zstd.CompressLevel(bb.B, bb.B, 1)
data = bb.B[bufLen:]
}
wLock.Lock()
dataLenBuf = encoding.MarshalUint64(dataLenBuf[:0], uint64(len(data)))
_, err := w.Write(dataLenBuf)
if err == nil {
_, err = w.Write(data)
}
wLock.Unlock()
// Reset the sent buf
bb.Reset()
return err
}
var bufs atomicutil.Slice[bytesutil.ByteBuffer]
var errGlobalLock sync.Mutex
var errGlobal error
writeBlock := func(workerID uint, db *logstorage.DataBlock) {
if errGlobal != nil {
return
}
bb := bufs.Get(workerID)
bb.B = db.Marshal(bb.B)
if len(bb.B) < 1024*1024 {
// Fast path - the bb is too small to be sent to the client yet.
return
}
// Slow path - the bb must be sent to the client.
if err := sendBuf(bb); err != nil {
errGlobalLock.Lock()
if errGlobal != nil {
errGlobal = err
}
errGlobalLock.Unlock()
}
}
if err := vlstorage.RunQuery(ctx, cp.TenantIDs, cp.Query, writeBlock); err != nil {
return err
}
if errGlobal != nil {
return errGlobal
}
// Send the remaining data
for _, bb := range bufs.All() {
if err := sendBuf(bb); err != nil {
return err
}
}
return nil
}
func processFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
cp, err := getCommonParams(r, netselect.FieldNamesProtocolVersion)
if err != nil {
return err
}
fieldNames, err := vlstorage.GetFieldNames(ctx, cp.TenantIDs, cp.Query)
if err != nil {
return fmt.Errorf("cannot obtain field names: %w", err)
}
return writeValuesWithHits(w, fieldNames, cp.DisableCompression)
}
func processFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
cp, err := getCommonParams(r, netselect.FieldValuesProtocolVersion)
if err != nil {
return err
}
fieldName := r.FormValue("field")
limit, err := getInt64FromRequest(r, "limit")
if err != nil {
return err
}
fieldValues, err := vlstorage.GetFieldValues(ctx, cp.TenantIDs, cp.Query, fieldName, uint64(limit))
if err != nil {
return fmt.Errorf("cannot obtain field values: %w", err)
}
return writeValuesWithHits(w, fieldValues, cp.DisableCompression)
}
func processStreamFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
cp, err := getCommonParams(r, netselect.StreamFieldNamesProtocolVersion)
if err != nil {
return err
}
fieldNames, err := vlstorage.GetStreamFieldNames(ctx, cp.TenantIDs, cp.Query)
if err != nil {
return fmt.Errorf("cannot obtain stream field names: %w", err)
}
return writeValuesWithHits(w, fieldNames, cp.DisableCompression)
}
func processStreamFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
cp, err := getCommonParams(r, netselect.StreamFieldValuesProtocolVersion)
if err != nil {
return err
}
fieldName := r.FormValue("field")
limit, err := getInt64FromRequest(r, "limit")
if err != nil {
return err
}
fieldValues, err := vlstorage.GetStreamFieldValues(ctx, cp.TenantIDs, cp.Query, fieldName, uint64(limit))
if err != nil {
return fmt.Errorf("cannot obtain stream field values: %w", err)
}
return writeValuesWithHits(w, fieldValues, cp.DisableCompression)
}
func processStreamsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
cp, err := getCommonParams(r, netselect.StreamsProtocolVersion)
if err != nil {
return err
}
limit, err := getInt64FromRequest(r, "limit")
if err != nil {
return err
}
streams, err := vlstorage.GetStreams(ctx, cp.TenantIDs, cp.Query, uint64(limit))
if err != nil {
return fmt.Errorf("cannot obtain streams: %w", err)
}
return writeValuesWithHits(w, streams, cp.DisableCompression)
}
func processStreamIDsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) error {
cp, err := getCommonParams(r, netselect.StreamIDsProtocolVersion)
if err != nil {
return err
}
limit, err := getInt64FromRequest(r, "limit")
if err != nil {
return err
}
streamIDs, err := vlstorage.GetStreamIDs(ctx, cp.TenantIDs, cp.Query, uint64(limit))
if err != nil {
return fmt.Errorf("cannot obtain streams: %w", err)
}
return writeValuesWithHits(w, streamIDs, cp.DisableCompression)
}
type commonParams struct {
TenantIDs []logstorage.TenantID
Query *logstorage.Query
DisableCompression bool
}
func getCommonParams(r *http.Request, expectedProtocolVersion string) (*commonParams, error) {
version := r.FormValue("version")
if version != expectedProtocolVersion {
return nil, fmt.Errorf("unexpected version=%q; want %q", version, expectedProtocolVersion)
}
tenantIDsStr := r.FormValue("tenant_ids")
tenantIDs, err := logstorage.UnmarshalTenantIDs([]byte(tenantIDsStr))
if err != nil {
return nil, fmt.Errorf("cannot unmarshal tenant_ids=%q: %w", tenantIDsStr, err)
}
timestamp, err := getInt64FromRequest(r, "timestamp")
if err != nil {
return nil, err
}
qStr := r.FormValue("query")
q, err := logstorage.ParseQueryAtTimestamp(qStr, timestamp)
if err != nil {
return nil, fmt.Errorf("cannot unmarshal query=%q: %w", qStr, err)
}
s := r.FormValue("disable_compression")
disableCompression, err := strconv.ParseBool(s)
if err != nil {
return nil, fmt.Errorf("cannot parse disable_compression=%q: %w", s, err)
}
cp := &commonParams{
TenantIDs: tenantIDs,
Query: q,
DisableCompression: disableCompression,
}
return cp, nil
}
func writeValuesWithHits(w http.ResponseWriter, vhs []logstorage.ValueWithHits, disableCompression bool) error {
var b []byte
for i := range vhs {
b = vhs[i].Marshal(b)
}
if !disableCompression {
b = zstd.CompressLevel(nil, b, 1)
}
w.Header().Set("Content-Type", "application/octet-stream")
if _, err := w.Write(b); err != nil {
return fmt.Errorf("cannot send response to the client: %w", err)
}
return nil
}
func getInt64FromRequest(r *http.Request, argName string) (int64, error) {
s := r.FormValue(argName)
n, err := strconv.ParseInt(s, 10, 64)
if err != nil {
return 0, fmt.Errorf("cannot parse %s=%q: %w", argName, s, err)
}
return n, nil
}

View File

@@ -0,0 +1,47 @@
package logsql
import (
"bufio"
"io"
"sync"
)
func getBufferedWriter(w io.Writer) *bufferedWriter {
v := bufferedWriterPool.Get()
if v == nil {
return &bufferedWriter{
bw: bufio.NewWriter(w),
}
}
bw := v.(*bufferedWriter)
bw.bw.Reset(w)
return bw
}
func putBufferedWriter(bw *bufferedWriter) {
bw.reset()
bufferedWriterPool.Put(bw)
}
var bufferedWriterPool sync.Pool
type bufferedWriter struct {
mu sync.Mutex
bw *bufio.Writer
}
func (bw *bufferedWriter) reset() {
// nothing to do
}
func (bw *bufferedWriter) WriteIgnoreErrors(p []byte) {
bw.mu.Lock()
_, _ = bw.bw.Write(p)
bw.mu.Unlock()
}
func (bw *bufferedWriter) FlushIgnoreErrors() {
bw.mu.Lock()
_ = bw.bw.Flush()
bw.mu.Unlock()
}

View File

@@ -3,7 +3,6 @@ package logsql
import (
"context"
"fmt"
"io"
"math"
"net/http"
"regexp"
@@ -12,14 +11,12 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/VictoriaMetrics/metrics"
"github.com/valyala/fastjson"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/atomicutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
@@ -55,21 +52,15 @@ func ProcessFacetsRequest(ctx context.Context, w http.ResponseWriter, r *http.Re
}
keepConstFields := httputil.GetBool(r, "keep_const_fields")
// Pipes must be dropped, since it is expected facets are obtained
// from the real logs stored in the database.
q.DropAllPipes()
q.AddFacetsPipe(limit, maxValuesPerField, maxValueLen, keepConstFields)
var mLock sync.Mutex
m := make(map[string][]facetEntry)
writeBlock := func(_ uint, db *logstorage.DataBlock) {
rowsCount := db.RowsCount()
if rowsCount == 0 {
writeBlock := func(_ uint, _ []int64, columns []logstorage.BlockColumn) {
if len(columns) == 0 || len(columns[0].Values) == 0 {
return
}
columns := db.Columns
if len(columns) != 3 {
logger.Panicf("BUG: expecting 3 columns; got %d columns", len(columns))
}
@@ -159,27 +150,23 @@ func ProcessHitsRequest(ctx context.Context, w http.ResponseWriter, r *http.Requ
fieldsLimit = 0
}
// Pipes must be dropped, since it is expected hits are obtained
// from the real logs stored in the database.
// Prepare the query for hits count.
q.DropAllPipes()
q.AddCountByTimePipe(int64(step), int64(offset), fields)
var mLock sync.Mutex
m := make(map[string]*hitsSeries)
writeBlock := func(_ uint, db *logstorage.DataBlock) {
rowsCount := db.RowsCount()
if rowsCount == 0 {
writeBlock := func(_ uint, timestamps []int64, columns []logstorage.BlockColumn) {
if len(columns) == 0 || len(columns[0].Values) == 0 {
return
}
columns := db.Columns
timestampValues := columns[0].Values
hitsValues := columns[len(columns)-1].Values
columns = columns[1 : len(columns)-1]
bb := blockResultPool.Get()
for i := 0; i < rowsCount; i++ {
for i := range timestamps {
timestampStr := strings.Clone(timestampValues[i])
hitsStr := strings.Clone(hitsValues[i])
hits, err := strconv.ParseUint(hitsStr, 10, 64)
@@ -218,8 +205,6 @@ func ProcessHitsRequest(ctx context.Context, w http.ResponseWriter, r *http.Requ
WriteHitsSeries(w, m)
}
var blockResultPool bytesutil.ByteBufferPool
func getTopHitsSeries(m map[string]*hitsSeries, fieldsLimit int) map[string]*hitsSeries {
if fieldsLimit <= 0 || fieldsLimit >= len(m) {
return m
@@ -295,10 +280,6 @@ func ProcessFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *htt
return
}
// Pipes must be dropped, since it is expected field names are obtained
// from the real logs stored in the database.
q.DropAllPipes()
// Obtain field names for the given query
fieldNames, err := vlstorage.GetFieldNames(ctx, tenantIDs, q)
if err != nil {
@@ -338,10 +319,6 @@ func ProcessFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *ht
limit = 0
}
// Pipes must be dropped, since it is expected field values are obtained
// from the real logs stored in the database.
q.DropAllPipes()
// Obtain unique values for the given field
values, err := vlstorage.GetFieldValues(ctx, tenantIDs, q, fieldName, uint64(limit))
if err != nil {
@@ -364,10 +341,6 @@ func ProcessStreamFieldNamesRequest(ctx context.Context, w http.ResponseWriter,
return
}
// Pipes must be dropped, since it is expected stream field names are obtained
// from the real logs stored in the database.
q.DropAllPipes()
// Obtain stream field names for the given query
names, err := vlstorage.GetStreamFieldNames(ctx, tenantIDs, q)
if err != nil {
@@ -406,10 +379,6 @@ func ProcessStreamFieldValuesRequest(ctx context.Context, w http.ResponseWriter,
limit = 0
}
// Pipes must be dropped, since it is expected stream field values are obtained
// from the real logs stored in the database.
q.DropAllPipes()
// Obtain stream field values for the given query and the given fieldName
values, err := vlstorage.GetStreamFieldValues(ctx, tenantIDs, q, fieldName, uint64(limit))
if err != nil {
@@ -441,10 +410,6 @@ func ProcessStreamIDsRequest(ctx context.Context, w http.ResponseWriter, r *http
limit = 0
}
// Pipes must be dropped, since it is expected stream ids are obtained
// from the real logs stored in the database.
q.DropAllPipes()
// Obtain streamIDs for the given query
streamIDs, err := vlstorage.GetStreamIDs(ctx, tenantIDs, q, uint64(limit))
if err != nil {
@@ -476,10 +441,6 @@ func ProcessStreamsRequest(ctx context.Context, w http.ResponseWriter, r *http.R
limit = 0
}
// Pipes must be dropped, since it is expected stream are obtained
// from the real logs stored in the database.
q.DropAllPipes()
// Obtain streams for the given query
streams, err := vlstorage.GetStreams(ctx, tenantIDs, q, uint64(limit))
if err != nil {
@@ -543,11 +504,6 @@ func ProcessLiveTailRequest(ctx context.Context, w http.ResponseWriter, r *http.
if !ok {
logger.Panicf("BUG: it is expected that http.ResponseWriter (%T) supports http.Flusher interface", w)
}
w.Header().Set("Content-Type", "application/x-ndjson")
w.Header().Set("Access-Control-Allow-Origin", "*")
flusher.Flush()
qOrig := q
for {
q = qOrig.CloneWithTimeFilter(end, start, end)
@@ -610,8 +566,8 @@ func newTailProcessor(cancel func()) *tailProcessor {
}
}
func (tp *tailProcessor) writeBlock(_ uint, db *logstorage.DataBlock) {
if db.RowsCount() == 0 {
func (tp *tailProcessor) writeBlock(_ uint, timestamps []int64, columns []logstorage.BlockColumn) {
if len(timestamps) == 0 {
return
}
@@ -623,8 +579,14 @@ func (tp *tailProcessor) writeBlock(_ uint, db *logstorage.DataBlock) {
}
// Make sure columns contain _time field, since it is needed for proper tail work.
timestamps, ok := db.GetTimestamps(nil)
if !ok {
hasTime := false
for _, c := range columns {
if c.Name == "_time" {
hasTime = true
break
}
}
if !hasTime {
tp.err = fmt.Errorf("missing _time field")
tp.cancel()
return
@@ -633,8 +595,8 @@ func (tp *tailProcessor) writeBlock(_ uint, db *logstorage.DataBlock) {
// Copy block rows to tp.perStreamRows
for i, timestamp := range timestamps {
streamID := ""
fields := make([]logstorage.Field, len(db.Columns))
for j, c := range db.Columns {
fields := make([]logstorage.Field, len(columns))
for j, c := range columns {
name := strings.Clone(c.Name)
value := strings.Clone(c.Values[i])
@@ -726,15 +688,12 @@ func ProcessStatsQueryRangeRequest(ctx context.Context, w http.ResponseWriter, r
m := make(map[string]*statsSeries)
var mLock sync.Mutex
writeBlock := func(_ uint, db *logstorage.DataBlock) {
rowsCount := db.RowsCount()
columns := db.Columns
writeBlock := func(_ uint, timestamps []int64, columns []logstorage.BlockColumn) {
clonedColumnNames := make([]string, len(columns))
for i, c := range columns {
clonedColumnNames[i] = strings.Clone(c.Name)
}
for i := 0; i < rowsCount; i++ {
for i := range timestamps {
// Do not move q.GetTimestamp() outside writeBlock, since ts
// must be initialized to query timestamp for every processed log row.
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8312
@@ -843,14 +802,12 @@ func ProcessStatsQueryRequest(ctx context.Context, w http.ResponseWriter, r *htt
var rowsLock sync.Mutex
timestamp := q.GetTimestamp()
writeBlock := func(_ uint, db *logstorage.DataBlock) {
rowsCount := db.RowsCount()
columns := db.Columns
writeBlock := func(_ uint, timestamps []int64, columns []logstorage.BlockColumn) {
clonedColumnNames := make([]string, len(columns))
for i, c := range columns {
clonedColumnNames[i] = strings.Clone(c.Name)
}
for i := 0; i < rowsCount; i++ {
for i := range timestamps {
labels := make([]logstorage.Field, 0, len(byFields))
for j, c := range columns {
if slices.Contains(byFields, c.Name) {
@@ -912,21 +869,11 @@ func ProcessQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Req
return
}
sw := &syncWriter{
w: w,
}
var bwShards atomicutil.Slice[bufferedWriter]
bwShards.Init = func(shard *bufferedWriter) {
shard.sw = sw
}
bw := getBufferedWriter(w)
defer func() {
shards := bwShards.All()
for _, shard := range shards {
shard.FlushIgnoreErrors()
}
bw.FlushIgnoreErrors()
putBufferedWriter(bw)
}()
w.Header().Set("Content-Type", "application/stream+json")
if limit > 0 {
@@ -936,34 +883,32 @@ func ProcessQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Req
httpserver.Errorf(w, r, "%s", err)
return
}
bw := bwShards.Get(0)
bb := blockResultPool.Get()
b := bb.B
for i := range rows {
bw.buf = logstorage.MarshalFieldsToJSON(bw.buf, rows[i].fields)
bw.buf = append(bw.buf, '\n')
if len(bw.buf) > 16*1024 {
bw.FlushIgnoreErrors()
}
b = logstorage.MarshalFieldsToJSON(b[:0], rows[i].fields)
b = append(b, '\n')
bw.WriteIgnoreErrors(b)
}
bb.B = b
blockResultPool.Put(bb)
return
}
q.AddPipeLimit(uint64(limit))
}
writeBlock := func(workerID uint, db *logstorage.DataBlock) {
rowsCount := db.RowsCount()
if rowsCount == 0 {
writeBlock := func(_ uint, timestamps []int64, columns []logstorage.BlockColumn) {
if len(columns) == 0 || len(columns[0].Values) == 0 {
return
}
columns := db.Columns
bw := bwShards.Get(workerID)
for i := 0; i < rowsCount; i++ {
WriteJSONRow(bw, columns, i)
if len(bw.buf) > 16*1024 {
bw.FlushIgnoreErrors()
}
bb := blockResultPool.Get()
for i := range timestamps {
WriteJSONRow(bb, columns, i)
}
bw.WriteIgnoreErrors(bb.B)
blockResultPool.Put(bb)
}
if err := vlstorage.RunQuery(ctx, tenantIDs, q, writeBlock); err != nil {
@@ -972,37 +917,14 @@ func ProcessQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Req
}
}
type syncWriter struct {
mu sync.Mutex
w io.Writer
var blockResultPool bytesutil.ByteBufferPool
type row struct {
timestamp int64
fields []logstorage.Field
}
func (sw *syncWriter) Write(p []byte) (int, error) {
sw.mu.Lock()
n, err := sw.w.Write(p)
sw.mu.Unlock()
return n, err
}
type bufferedWriter struct {
buf []byte
sw *syncWriter
}
func (bw *bufferedWriter) Write(p []byte) (int, error) {
bw.buf = append(bw.buf, p...)
// Do not send bw.buf to bw.sw here, since the data at bw.buf may be incomplete (it must end with '\n')
return len(p), nil
}
func (bw *bufferedWriter) FlushIgnoreErrors() {
_, _ = bw.sw.Write(bw.buf)
bw.buf = bw.buf[:0]
}
func getLastNQueryResults(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit int) ([]logRow, error) {
func getLastNQueryResults(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit int) ([]row, error) {
limitUpper := 2 * limit
q.AddPipeLimit(uint64(limitUpper))
@@ -1071,39 +993,28 @@ func getLastNQueryResults(ctx context.Context, tenantIDs []logstorage.TenantID,
}
}
func getLastNRows(rows []logRow, limit int) []logRow {
sortLogRows(rows)
func getLastNRows(rows []row, limit int) []row {
sort.Slice(rows, func(i, j int) bool {
return rows[i].timestamp < rows[j].timestamp
})
if len(rows) > limit {
rows = rows[len(rows)-limit:]
}
return rows
}
func getQueryResultsWithLimit(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit int) ([]logRow, error) {
func getQueryResultsWithLimit(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit int) ([]row, error) {
ctxWithCancel, cancel := context.WithCancel(ctx)
defer cancel()
var missingTimeColumn atomic.Bool
var rows []logRow
var rows []row
var rowsLock sync.Mutex
writeBlock := func(_ uint, db *logstorage.DataBlock) {
if missingTimeColumn.Load() {
return
}
columns := db.Columns
writeBlock := func(_ uint, timestamps []int64, columns []logstorage.BlockColumn) {
clonedColumnNames := make([]string, len(columns))
for i, c := range columns {
clonedColumnNames[i] = strings.Clone(c.Name)
}
timestamps, ok := db.GetTimestamps(nil)
if !ok {
missingTimeColumn.Store(true)
cancel()
return
}
for i, timestamp := range timestamps {
fields := make([]logstorage.Field, len(columns))
for j := range columns {
@@ -1113,7 +1024,7 @@ func getQueryResultsWithLimit(ctx context.Context, tenantIDs []logstorage.Tenant
}
rowsLock.Lock()
rows = append(rows, logRow{
rows = append(rows, row{
timestamp: timestamp,
fields: fields,
})
@@ -1124,13 +1035,11 @@ func getQueryResultsWithLimit(ctx context.Context, tenantIDs []logstorage.Tenant
cancel()
}
}
err := vlstorage.RunQuery(ctxWithCancel, tenantIDs, q, writeBlock)
if missingTimeColumn.Load() {
return nil, fmt.Errorf("missing _time column in the result for the query [%s]", q)
if err := vlstorage.RunQuery(ctxWithCancel, tenantIDs, q, writeBlock); err != nil {
return nil, err
}
return rows, err
return rows, nil
}
func parseCommonArgs(r *http.Request) (*logstorage.Query, []logstorage.TenantID, error) {
@@ -1189,22 +1098,20 @@ func parseCommonArgs(r *http.Request) (*logstorage.Query, []logstorage.TenantID,
}
// Parse optional extra_filters
for _, extraFiltersStr := range r.Form["extra_filters"] {
extraFilters, err := parseExtraFilters(extraFiltersStr)
if err != nil {
return nil, nil, err
}
q.AddExtraFilters(extraFilters)
extraFiltersStr := r.FormValue("extra_filters")
extraFilters, err := parseExtraFilters(extraFiltersStr)
if err != nil {
return nil, nil, err
}
q.AddExtraFilters(extraFilters)
// Parse optional extra_stream_filters
for _, extraStreamFiltersStr := range r.Form["extra_stream_filters"] {
extraStreamFilters, err := parseExtraStreamFilters(extraStreamFiltersStr)
if err != nil {
return nil, nil, err
}
q.AddExtraFilters(extraStreamFilters)
extraStreamFiltersStr := r.FormValue("extra_stream_filters")
extraStreamFilters, err := parseExtraStreamFilters(extraStreamFiltersStr)
if err != nil {
return nil, nil, err
}
q.AddExtraFilters(extraStreamFilters)
return q, tenantIDs, nil
}

View File

@@ -9,7 +9,6 @@ import (
"strings"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlselect/internalselect"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlselect/logsql"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -25,9 +24,6 @@ var (
maxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the search request waits for execution when -search.maxConcurrentRequests "+
"limit is reached; see also -search.maxQueryDuration")
maxQueryDuration = flag.Duration("search.maxQueryDuration", time.Second*30, "The maximum duration for query execution. It can be overridden to a smaller value on a per-query basis via 'timeout' query arg")
disableSelect = flag.Bool("select.disable", false, "Whether to disable /select/* HTTP endpoints")
disableInternal = flag.Bool("internalselect.disable", false, "Whether to disable /internal/select/* HTTP endpoints")
)
func getDefaultMaxConcurrentRequests() int {
@@ -74,31 +70,13 @@ var vmuiFileServer = http.FileServer(http.FS(vmuiFiles))
// RequestHandler handles select requests for VictoriaLogs
func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
path := strings.ReplaceAll(r.URL.Path, "//", "/")
path := r.URL.Path
if strings.HasPrefix(path, "/select/") {
if *disableSelect {
httpserver.Errorf(w, r, "requests to /select/* are disabled with -select.disable command-line flag")
return true
}
return selectHandler(w, r, path)
if !strings.HasPrefix(path, "/select/") {
// Skip requests, which do not start with /select/, since these aren't our requests.
return false
}
if strings.HasPrefix(path, "/internal/select/") {
if *disableInternal || *disableSelect {
httpserver.Errorf(w, r, "requests to /internal/select/* are disabled with -internalselect.disable or -select.disable command-line flag")
return true
}
internalselect.RequestHandler(r.Context(), w, r)
return true
}
return false
}
func selectHandler(w http.ResponseWriter, r *http.Request, path string) bool {
ctx := r.Context()
path = strings.ReplaceAll(path, "//", "/")
if path == "/select/vmui" {
// VMUI access via incomplete url without `/` in the end. Redirect to complete url.
@@ -121,16 +99,9 @@ func selectHandler(w http.ResponseWriter, r *http.Request, path string) bool {
return true
}
if path == "/select/logsql/tail" {
logsqlTailRequests.Inc()
// Process live tailing request without timeout, since it is OK to run live tailing requests for very long time.
// Also do not apply concurrency limit to tail requests, since these limits are intended for non-tail requests.
logsql.ProcessLiveTailRequest(ctx, w, r)
return true
}
// Limit the number of concurrent queries, which can consume big amounts of CPU time.
startTime := time.Now()
ctx := r.Context()
d := getMaxQueryDuration(r)
ctxWithTimeout, cancel := context.WithTimeout(ctx, d)
defer cancel()
@@ -140,22 +111,28 @@ func selectHandler(w http.ResponseWriter, r *http.Request, path string) bool {
}
defer decRequestConcurrency()
if path == "/select/logsql/tail" {
logsqlTailRequests.Inc()
// Process live tailing request without timeout (e.g. use ctx instead of ctxWithTimeout),
// since it is OK to run live tailing requests for very long time.
logsql.ProcessLiveTailRequest(ctx, w, r)
return true
}
ok := processSelectRequest(ctxWithTimeout, w, r, path)
if !ok {
return false
}
logRequestErrorIfNeeded(ctxWithTimeout, w, r, startTime)
return true
}
func logRequestErrorIfNeeded(ctx context.Context, w http.ResponseWriter, r *http.Request, startTime time.Time) {
err := ctx.Err()
err := ctxWithTimeout.Err()
switch err {
case nil:
// nothing to do
case context.Canceled:
// do not log canceled requests, since they are expected and legal.
remoteAddr := httpserver.GetQuotedRemoteAddr(r)
requestURI := httpserver.GetRequestURI(r)
logger.Infof("client has canceled the request after %.3f seconds: remoteAddr=%s, requestURI: %q",
time.Since(startTime).Seconds(), remoteAddr, requestURI)
case context.DeadlineExceeded:
err = &httpserver.ErrorWithStatusCode{
Err: fmt.Errorf("the request couldn't be executed in %.3f seconds; possible solutions: "+
@@ -166,6 +143,8 @@ func logRequestErrorIfNeeded(ctx context.Context, w http.ResponseWriter, r *http
default:
httpserver.Errorf(w, r, "unexpected error: %s", err)
}
return true
}
func incRequestConcurrency(ctx context.Context, w http.ResponseWriter, r *http.Request) bool {

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@@ -35,10 +35,10 @@
<meta property="og:title" content="UI for VictoriaLogs">
<meta property="og:url" content="https://victoriametrics.com/products/victorialogs/">
<meta property="og:description" content="Explore your log data with VictoriaLogs UI">
<script type="module" crossorigin src="./assets/index-721xTF8u.js"></script>
<link rel="modulepreload" crossorigin href="./assets/vendor-V4vnRsM-.js">
<script type="module" crossorigin src="./assets/index-BgdvCSTM.js"></script>
<link rel="modulepreload" crossorigin href="./assets/vendor-DojlIpLz.js">
<link rel="stylesheet" crossorigin href="./assets/vendor-D1GxaB_c.css">
<link rel="stylesheet" crossorigin href="./assets/index-C36SC0pJ.css">
<link rel="stylesheet" crossorigin href="./assets/index-u4IOGr0E.css">
</head>
<body>
<noscript>You need to enable JavaScript to run this app.</noscript>

View File

@@ -10,14 +10,11 @@ import (
"github.com/VictoriaMetrics/metrics"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage/netinsert"
"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage/netselect"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
)
var (
@@ -41,59 +38,15 @@ var (
minFreeDiskSpaceBytes = flagutil.NewBytes("storage.minFreeDiskSpaceBytes", 10e6, "The minimum free disk space at -storageDataPath after which "+
"the storage stops accepting new data")
forceMergeAuthKey = flagutil.NewPassword("forceMergeAuthKey", "authKey, which must be passed in query string to /internal/force_merge . It overrides -httpAuth.* . "+
"See https://docs.victoriametrics.com/victorialogs/#forced-merge")
forceFlushAuthKey = flagutil.NewPassword("forceFlushAuthKey", "authKey, which must be passed in query string to /internal/force_flush . It overrides -httpAuth.* . "+
"See https://docs.victoriametrics.com/victorialogs/#forced-flush")
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")
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. "+
"Disabled compression reduces CPU usage at the cost of higher network usage")
storageNodeUsername = flagutil.NewArrayString("storageNode.username", "Optional basic auth username to use for the corresponding -storageNode")
storageNodePassword = flagutil.NewArrayString("storageNode.password", "Optional basic auth password to use for the corresponding -storageNode")
storageNodePasswordFile = flagutil.NewArrayString("storageNode.passwordFile", "Optional path to basic auth password to use for the corresponding -storageNode. "+
"The file is re-read every second")
storageNodeBearerToken = flagutil.NewArrayString("storageNode.bearerToken", "Optional bearer auth token to use for the corresponding -storageNode")
storageNodeBearerTokenFile = flagutil.NewArrayString("storageNode.bearerTokenFile", "Optional path to bearer token file to use for the corresponding -storageNode. "+
"The token is re-read from the file every second")
storageNodeTLS = flagutil.NewArrayBool("storageNode.tls", "Whether to use TLS (HTTPS) protocol for communicating with the corresponding -storageNode. "+
"By default communication is performed via HTTP")
storageNodeTLSCAFile = flagutil.NewArrayString("storageNode.tlsCAFile", "Optional path to TLS CA file to use for verifying connections to the corresponding -storageNode. "+
"By default, system CA is used")
storageNodeTLSCertFile = flagutil.NewArrayString("storageNode.tlsCertFile", "Optional path to client-side TLS certificate file to use when connecting "+
"to the corresponding -storageNode")
storageNodeTLSKeyFile = flagutil.NewArrayString("storageNode.tlsKeyFile", "Optional path to client-side TLS certificate key to use when connecting to the corresponding -storageNode")
storageNodeTLSServerName = flagutil.NewArrayString("storageNode.tlsServerName", "Optional TLS server name to use for connections to the corresponding -storageNode. "+
"By default, the server name from -storageNode is used")
storageNodeTLSInsecureSkipVerify = flagutil.NewArrayBool("storageNode.tlsInsecureSkipVerify", "Whether to skip tls verification when connecting to the corresponding -storageNode")
forceMergeAuthKey = flagutil.NewPassword("forceMergeAuthKey", "authKey, which must be passed in query string to /internal/force_merge pages. It overrides -httpAuth.*")
)
var localStorage *logstorage.Storage
var localStorageMetrics *metrics.Set
var netstorageInsert *netinsert.Storage
var netstorageSelect *netselect.Storage
// Init initializes vlstorage.
//
// Stop must be called when vlstorage is no longer needed
func Init() {
if len(*storageNodeAddrs) == 0 {
initLocalStorage()
} else {
initNetworkStorage()
}
}
func initLocalStorage() {
if localStorage != nil {
logger.Panicf("BUG: initLocalStorage() has been already called")
if strg != nil {
logger.Panicf("BUG: Init() has been already called")
}
if retentionPeriod.Duration() < 24*time.Hour {
@@ -110,160 +63,60 @@ func initLocalStorage() {
}
logger.Infof("opening storage at -storageDataPath=%s", *storageDataPath)
startTime := time.Now()
localStorage = logstorage.MustOpenStorage(*storageDataPath, cfg)
strg = logstorage.MustOpenStorage(*storageDataPath, cfg)
var ss logstorage.StorageStats
localStorage.UpdateStats(&ss)
strg.UpdateStats(&ss)
logger.Infof("successfully opened storage in %.3f seconds; smallParts: %d; bigParts: %d; smallPartBlocks: %d; bigPartBlocks: %d; smallPartRows: %d; bigPartRows: %d; "+
"smallPartSize: %d bytes; bigPartSize: %d bytes",
time.Since(startTime).Seconds(), ss.SmallParts, ss.BigParts, ss.SmallPartBlocks, ss.BigPartBlocks, ss.SmallPartRowsCount, ss.BigPartRowsCount,
ss.CompressedSmallPartSize, ss.CompressedBigPartSize)
// register local storage metrics
localStorageMetrics = metrics.NewSet()
localStorageMetrics.RegisterMetricsWriter(func(w io.Writer) {
writeStorageMetrics(w, localStorage)
// register storage metrics
storageMetrics = metrics.NewSet()
storageMetrics.RegisterMetricsWriter(func(w io.Writer) {
writeStorageMetrics(w, strg)
})
metrics.RegisterSet(localStorageMetrics)
}
func initNetworkStorage() {
if netstorageInsert != nil || netstorageSelect != nil {
logger.Panicf("BUG: initNetworkStorage() has been already called")
}
authCfgs := make([]*promauth.Config, len(*storageNodeAddrs))
isTLSs := make([]bool, len(*storageNodeAddrs))
for i := range authCfgs {
authCfgs[i] = newAuthConfigForStorageNode(i)
isTLSs[i] = storageNodeTLS.GetOptionalArg(i)
}
logger.Infof("starting insert service for nodes %s", *storageNodeAddrs)
netstorageInsert = netinsert.NewStorage(*storageNodeAddrs, authCfgs, isTLSs, *insertConcurrency, *insertDisableCompression)
logger.Infof("initializing select service for nodes %s", *storageNodeAddrs)
netstorageSelect = netselect.NewStorage(*storageNodeAddrs, authCfgs, isTLSs, *selectDisableCompression)
logger.Infof("initialized all the network services")
}
func newAuthConfigForStorageNode(argIdx int) *promauth.Config {
username := storageNodeUsername.GetOptionalArg(argIdx)
password := storageNodePassword.GetOptionalArg(argIdx)
passwordFile := storageNodePasswordFile.GetOptionalArg(argIdx)
var basicAuthCfg *promauth.BasicAuthConfig
if username != "" || password != "" || passwordFile != "" {
basicAuthCfg = &promauth.BasicAuthConfig{
Username: username,
Password: promauth.NewSecret(password),
PasswordFile: passwordFile,
}
}
token := storageNodeBearerToken.GetOptionalArg(argIdx)
tokenFile := storageNodeBearerTokenFile.GetOptionalArg(argIdx)
tlsCfg := &promauth.TLSConfig{
CAFile: storageNodeTLSCAFile.GetOptionalArg(argIdx),
CertFile: storageNodeTLSCertFile.GetOptionalArg(argIdx),
KeyFile: storageNodeTLSKeyFile.GetOptionalArg(argIdx),
ServerName: storageNodeTLSServerName.GetOptionalArg(argIdx),
InsecureSkipVerify: storageNodeTLSInsecureSkipVerify.GetOptionalArg(argIdx),
}
opts := &promauth.Options{
BasicAuth: basicAuthCfg,
BearerToken: token,
BearerTokenFile: tokenFile,
TLSConfig: tlsCfg,
}
ac, err := opts.NewConfig()
if err != nil {
logger.Panicf("FATAL: cannot populate auth config for storage node #%d: %s", argIdx, err)
}
return ac
metrics.RegisterSet(storageMetrics)
}
// Stop stops vlstorage.
func Stop() {
if localStorage != nil {
metrics.UnregisterSet(localStorageMetrics, true)
localStorageMetrics = nil
metrics.UnregisterSet(storageMetrics, true)
storageMetrics = nil
localStorage.MustClose()
localStorage = nil
} else {
netstorageInsert.MustStop()
netstorageInsert = nil
netstorageSelect.MustStop()
netstorageSelect = nil
}
strg.MustClose()
strg = nil
}
// RequestHandler is a storage request handler.
func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
path := r.URL.Path
switch path {
case "/internal/force_merge":
return processForceMerge(w, r)
case "/internal/force_flush":
return processForceFlush(w, r)
if path == "/internal/force_merge" {
if !httpserver.CheckAuthFlag(w, r, forceMergeAuthKey) {
return true
}
// Run force merge in background
partitionNamePrefix := r.FormValue("partition_prefix")
go func() {
activeForceMerges.Inc()
defer activeForceMerges.Dec()
logger.Infof("forced merge for partition_prefix=%q has been started", partitionNamePrefix)
startTime := time.Now()
strg.MustForceMerge(partitionNamePrefix)
logger.Infof("forced merge for partition_prefix=%q has been successfully finished in %.3f seconds", partitionNamePrefix, time.Since(startTime).Seconds())
}()
return true
}
return false
}
func processForceMerge(w http.ResponseWriter, r *http.Request) bool {
if localStorage == nil {
// Force merge isn't supported by non-local storage
return false
}
var strg *logstorage.Storage
var storageMetrics *metrics.Set
if !httpserver.CheckAuthFlag(w, r, forceMergeAuthKey) {
return true
}
// Run force merge in background
partitionNamePrefix := r.FormValue("partition_prefix")
go func() {
activeForceMerges.Inc()
defer activeForceMerges.Dec()
logger.Infof("forced merge for partition_prefix=%q has been started", partitionNamePrefix)
startTime := time.Now()
localStorage.MustForceMerge(partitionNamePrefix)
logger.Infof("forced merge for partition_prefix=%q has been successfully finished in %.3f seconds", partitionNamePrefix, time.Since(startTime).Seconds())
}()
return true
}
func processForceFlush(w http.ResponseWriter, r *http.Request) bool {
if localStorage == nil {
// Force merge isn't supported by non-local storage
return false
}
if !httpserver.CheckAuthFlag(w, r, forceFlushAuthKey) {
return true
}
logger.Infof("flushing storage to make pending data available for reading")
localStorage.DebugFlush()
return true
}
// Storage implements insertutil.LogRowsStorage interface
type Storage struct{}
// CanWriteData returns non-nil error if it cannot write data to vlstorage
func (*Storage) CanWriteData() error {
if localStorage == nil {
// The data can be always written in non-local mode.
return nil
}
if localStorage.IsReadOnly() {
// CanWriteData returns non-nil error if it cannot write data to vlstorage.
func CanWriteData() error {
if strg.IsReadOnly() {
return &httpserver.ErrorWithStatusCode{
Err: fmt.Errorf("cannot add rows into storage in read-only mode; the storage can be in read-only mode "+
"because of lack of free disk space at -storageDataPath=%s", *storageDataPath),
@@ -276,78 +129,51 @@ func (*Storage) CanWriteData() error {
// MustAddRows adds lr to vlstorage
//
// It is advised to call CanWriteData() before calling MustAddRows()
func (*Storage) MustAddRows(lr *logstorage.LogRows) {
if localStorage != nil {
// Store lr in the local storage.
localStorage.MustAddRows(lr)
} else {
// Store lr across the remote storage nodes.
lr.ForEachRow(netstorageInsert.AddRow)
}
func MustAddRows(lr *logstorage.LogRows) {
strg.MustAddRows(lr)
}
// RunQuery runs the given q and calls writeBlock for the returned data blocks
func RunQuery(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, writeBlock logstorage.WriteDataBlockFunc) error {
if localStorage != nil {
return localStorage.RunQuery(ctx, tenantIDs, q, writeBlock)
}
return netstorageSelect.RunQuery(ctx, tenantIDs, q, writeBlock)
func RunQuery(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, writeBlock logstorage.WriteBlockFunc) error {
return strg.RunQuery(ctx, tenantIDs, q, writeBlock)
}
// GetFieldNames executes q and returns field names seen in results.
func GetFieldNames(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query) ([]logstorage.ValueWithHits, error) {
if localStorage != nil {
return localStorage.GetFieldNames(ctx, tenantIDs, q)
}
return netstorageSelect.GetFieldNames(ctx, tenantIDs, q)
return strg.GetFieldNames(ctx, tenantIDs, q)
}
// GetFieldValues executes q and returns unique values for the fieldName seen in results.
//
// If limit > 0, then up to limit unique values are returned.
func GetFieldValues(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, fieldName string, limit uint64) ([]logstorage.ValueWithHits, error) {
if localStorage != nil {
return localStorage.GetFieldValues(ctx, tenantIDs, q, fieldName, limit)
}
return netstorageSelect.GetFieldValues(ctx, tenantIDs, q, fieldName, limit)
return strg.GetFieldValues(ctx, tenantIDs, q, fieldName, limit)
}
// GetStreamFieldNames executes q and returns stream field names seen in results.
func GetStreamFieldNames(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query) ([]logstorage.ValueWithHits, error) {
if localStorage != nil {
return localStorage.GetStreamFieldNames(ctx, tenantIDs, q)
}
return netstorageSelect.GetStreamFieldNames(ctx, tenantIDs, q)
return strg.GetStreamFieldNames(ctx, tenantIDs, q)
}
// GetStreamFieldValues executes q and returns stream field values for the given fieldName seen in results.
//
// If limit > 0, then up to limit unique stream field values are returned.
func GetStreamFieldValues(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, fieldName string, limit uint64) ([]logstorage.ValueWithHits, error) {
if localStorage != nil {
return localStorage.GetStreamFieldValues(ctx, tenantIDs, q, fieldName, limit)
}
return netstorageSelect.GetStreamFieldValues(ctx, tenantIDs, q, fieldName, limit)
return strg.GetStreamFieldValues(ctx, tenantIDs, q, fieldName, limit)
}
// GetStreams executes q and returns streams seen in query results.
//
// If limit > 0, then up to limit unique streams are returned.
func GetStreams(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit uint64) ([]logstorage.ValueWithHits, error) {
if localStorage != nil {
return localStorage.GetStreams(ctx, tenantIDs, q, limit)
}
return netstorageSelect.GetStreams(ctx, tenantIDs, q, limit)
return strg.GetStreams(ctx, tenantIDs, q, limit)
}
// GetStreamIDs executes q and returns streamIDs seen in query results.
//
// If limit > 0, then up to limit unique streamIDs are returned.
func GetStreamIDs(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit uint64) ([]logstorage.ValueWithHits, error) {
if localStorage != nil {
return localStorage.GetStreamIDs(ctx, tenantIDs, q, limit)
}
return netstorageSelect.GetStreamIDs(ctx, tenantIDs, q, limit)
return strg.GetStreamIDs(ctx, tenantIDs, q, limit)
}
func writeStorageMetrics(w io.Writer, strg *logstorage.Storage) {

View File

@@ -1,369 +0,0 @@
package netinsert
import (
"errors"
"fmt"
"io"
"net/http"
"net/url"
"sync"
"sync/atomic"
"time"
"github.com/valyala/fastrand"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/contextutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
)
// the maximum size of a single data block sent to storage node.
const maxInsertBlockSize = 2 * 1024 * 1024
// ProtocolVersion is the version of the data ingestion protocol.
//
// It must be changed every time the data encoding at /internal/insert HTTP endpoint is changed.
const ProtocolVersion = "v1"
// Storage is a network storage for sending data to remote storage nodes in the cluster.
type Storage struct {
sns []*storageNode
disableCompression bool
srt *streamRowsTracker
pendingDataBuffers chan *bytesutil.ByteBuffer
stopCh chan struct{}
wg sync.WaitGroup
}
type storageNode struct {
// scheme is http or https scheme to communicate with addr
scheme string
// addr is TCP address of storage node to send the ingested data to
addr string
// s is a storage, which holds the given storageNode
s *Storage
// c is an http client used for sending data blocks to addr.
c *http.Client
// ac is auth config used for setting request headers such as Authorization and Host.
ac *promauth.Config
// pendingData contains pending data, which must be sent to the storage node at the addr.
pendingDataMu sync.Mutex
pendingData *bytesutil.ByteBuffer
pendingDataLastFlush time.Time
// the unix timestamp until the storageNode is disabled for data writing.
disabledUntil atomic.Uint64
}
func newStorageNode(s *Storage, addr string, ac *promauth.Config, isTLS bool) *storageNode {
tr := httputil.NewTransport(false, "vlinsert_backend")
tr.TLSHandshakeTimeout = 20 * time.Second
tr.DisableCompression = true
scheme := "http"
if isTLS {
scheme = "https"
}
sn := &storageNode{
scheme: scheme,
addr: addr,
s: s,
c: &http.Client{
Transport: ac.NewRoundTripper(tr),
},
ac: ac,
pendingData: &bytesutil.ByteBuffer{},
}
s.wg.Add(1)
go func() {
defer s.wg.Done()
sn.backgroundFlusher()
}()
return sn
}
func (sn *storageNode) backgroundFlusher() {
t := time.NewTicker(time.Second)
defer t.Stop()
for {
select {
case <-sn.s.stopCh:
return
case <-t.C:
sn.flushPendingData()
}
}
}
func (sn *storageNode) flushPendingData() {
sn.pendingDataMu.Lock()
if time.Since(sn.pendingDataLastFlush) < time.Second {
// nothing to flush
sn.pendingDataMu.Unlock()
return
}
pendingData := sn.grabPendingDataForFlushLocked()
sn.pendingDataMu.Unlock()
sn.mustSendInsertRequest(pendingData)
}
func (sn *storageNode) addRow(r *logstorage.InsertRow) {
bb := bbPool.Get()
b := bb.B
b = r.Marshal(b)
if len(b) > maxInsertBlockSize {
logger.Warnf("skipping too long log entry, since its length exceeds %d bytes; the actual log entry length is %d bytes; log entry contents: %s", maxInsertBlockSize, len(b), b)
return
}
var pendingData *bytesutil.ByteBuffer
sn.pendingDataMu.Lock()
if sn.pendingData.Len()+len(b) > maxInsertBlockSize {
pendingData = sn.grabPendingDataForFlushLocked()
}
sn.pendingData.MustWrite(b)
sn.pendingDataMu.Unlock()
bb.B = b
bbPool.Put(bb)
if pendingData != nil {
sn.mustSendInsertRequest(pendingData)
}
}
var bbPool bytesutil.ByteBufferPool
func (sn *storageNode) grabPendingDataForFlushLocked() *bytesutil.ByteBuffer {
sn.pendingDataLastFlush = time.Now()
pendingData := sn.pendingData
sn.pendingData = <-sn.s.pendingDataBuffers
return pendingData
}
func (sn *storageNode) mustSendInsertRequest(pendingData *bytesutil.ByteBuffer) {
defer func() {
pendingData.Reset()
sn.s.pendingDataBuffers <- pendingData
}()
err := sn.sendInsertRequest(pendingData)
if err == nil {
return
}
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 all storage nodes, since all of them are unavailable; 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())
return
case <-t.C:
timerpool.Put(t)
}
}
}
func (sn *storageNode) sendInsertRequest(pendingData *bytesutil.ByteBuffer) error {
dataLen := pendingData.Len()
if dataLen == 0 {
// Nothing to send.
return nil
}
if sn.disabledUntil.Load() > fasttime.UnixTimestamp() {
return errTemporarilyDisabled
}
ctx, cancel := contextutil.NewStopChanContext(sn.s.stopCh)
defer cancel()
var body io.Reader
if !sn.s.disableCompression {
bb := zstdBufPool.Get()
defer zstdBufPool.Put(bb)
bb.B = zstd.CompressLevel(bb.B[:0], pendingData.B, 1)
body = bb.NewReader()
} else {
body = pendingData.NewReader()
}
reqURL := sn.getRequestURL("/internal/insert")
req, err := http.NewRequestWithContext(ctx, "POST", reqURL, body)
if err != nil {
logger.Panicf("BUG: unexpected error when creating an http request: %s", err)
}
req.Header.Set("Content-Type", "application/octet-stream")
if !sn.s.disableCompression {
req.Header.Set("Content-Encoding", "zstd")
}
if err := sn.ac.SetHeaders(req, true); err != nil {
return fmt.Errorf("cannot set auth headers for %q: %w", reqURL, err)
}
resp, err := sn.c.Do(req)
if err != nil {
// Disable sn for data writing for 10 seconds.
sn.disabledUntil.Store(fasttime.UnixTimestamp() + 10)
return fmt.Errorf("cannot send data block with the length %d to %q: %s", pendingData.Len(), reqURL, err)
}
defer resp.Body.Close()
if resp.StatusCode/100 == 2 {
return nil
}
respBody, err := io.ReadAll(resp.Body)
if err != nil {
respBody = []byte(fmt.Sprintf("%s", err))
}
// Disable sn for data writing for 10 seconds.
sn.disabledUntil.Store(fasttime.UnixTimestamp() + 10)
return fmt.Errorf("unexpected status code returned when sending data block to %q: %d; want 2xx; response body: %q", reqURL, resp.StatusCode, respBody)
}
func (sn *storageNode) getRequestURL(path string) string {
return fmt.Sprintf("%s://%s%s?version=%s", sn.scheme, sn.addr, path, url.QueryEscape(ProtocolVersion))
}
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.
//
// 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 i := 0; i < cap(pendingDataBuffers); i++ {
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])
}
s.sns = sns
s.srt = newStreamRowsTracker(len(sns))
return s
}
// MustStop stops the s.
func (s *Storage) MustStop() {
close(s.stopCh)
s.wg.Wait()
s.sns = nil
}
// AddRow adds the given log row into s.
func (s *Storage) AddRow(streamHash uint64, r *logstorage.InsertRow) {
idx := s.srt.getNodeIdx(streamHash)
sn := s.sns[idx]
sn.addRow(r)
}
func (s *Storage) sendInsertRequestToAnyNode(pendingData *bytesutil.ByteBuffer) bool {
startIdx := int(fastrand.Uint32n(uint32(len(s.sns))))
for i := range s.sns {
idx := (startIdx + i) % len(s.sns)
sn := s.sns[idx]
err := sn.sendInsertRequest(pendingData)
if err == nil {
return true
}
if !errors.Is(err, errTemporarilyDisabled) {
logger.Warnf("cannot send pending data to the storage node %q: %s; trying to send it to another storage node", sn.addr, err)
}
}
return false
}
var errTemporarilyDisabled = fmt.Errorf("writing to the node is temporarily disabled")
type streamRowsTracker struct {
mu sync.Mutex
nodesCount int64
rowsPerStream map[uint64]uint64
}
func newStreamRowsTracker(nodesCount int) *streamRowsTracker {
return &streamRowsTracker{
nodesCount: int64(nodesCount),
rowsPerStream: make(map[uint64]uint64),
}
}
func (srt *streamRowsTracker) getNodeIdx(streamHash uint64) uint64 {
if srt.nodesCount == 1 {
// Fast path for a single node.
return 0
}
srt.mu.Lock()
defer srt.mu.Unlock()
streamRows := srt.rowsPerStream[streamHash] + 1
srt.rowsPerStream[streamHash] = streamRows
if streamRows <= 1000 {
// Write the initial rows for the stream to a single storage node for better locality.
// This should work great for log streams containing small number of logs, since will be distributed
// evenly among available storage nodes because they have different streamHash.
return streamHash % uint64(srt.nodesCount)
}
// The log stream contains more than 1000 rows. Distribute them among storage nodes at random
// in order to improve query performance over this stream (the data for the log stream
// can be processed in parallel on all the storage nodes).
//
// The random distribution is preferred over round-robin distribution in order to avoid possible
// dependency between the order of the ingested logs and the number of storage nodes,
// which may lead to non-uniform distribution of logs among storage nodes.
return uint64(fastrand.Uint32n(uint32(srt.nodesCount)))
}

View File

@@ -1,57 +0,0 @@
package netinsert
import (
"fmt"
"math"
"math/rand"
"testing"
"github.com/cespare/xxhash/v2"
)
func TestStreamRowsTracker(t *testing.T) {
f := func(rowsCount, streamsCount, nodesCount int) {
t.Helper()
// generate stream hashes
streamHashes := make([]uint64, streamsCount)
for i := range streamHashes {
streamHashes[i] = xxhash.Sum64([]byte(fmt.Sprintf("stream %d.", i)))
}
srt := newStreamRowsTracker(nodesCount)
rng := rand.New(rand.NewSource(0))
rowsPerNode := make([]uint64, nodesCount)
for i := 0; i < rowsCount; i++ {
streamIdx := rng.Intn(streamsCount)
h := streamHashes[streamIdx]
nodeIdx := srt.getNodeIdx(h)
rowsPerNode[nodeIdx]++
}
// Verify that rows are uniformly distributed among nodes.
expectedRowsPerNode := float64(rowsCount) / float64(nodesCount)
for nodeIdx, nodeRows := range rowsPerNode {
if math.Abs(float64(nodeRows)-expectedRowsPerNode)/expectedRowsPerNode > 0.15 {
t.Fatalf("non-uniform distribution of rows among nodes; node %d has %d rows, while it must have %v rows; rowsPerNode=%d",
nodeIdx, nodeRows, expectedRowsPerNode, rowsPerNode)
}
}
}
rowsCount := 10000
streamsCount := 9
nodesCount := 2
f(rowsCount, streamsCount, nodesCount)
rowsCount = 10000
streamsCount = 100
nodesCount = 2
f(rowsCount, streamsCount, nodesCount)
rowsCount = 100000
streamsCount = 1000
nodesCount = 9
f(rowsCount, streamsCount, nodesCount)
}

View File

@@ -1,472 +0,0 @@
package netselect
import (
"context"
"errors"
"fmt"
"io"
"math"
"net/http"
"net/url"
"strings"
"sync"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/contextutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
)
const (
// FieldNamesProtocolVersion is the version of the protocol used for /internal/select/field_names HTTP endpoint.
//
// It must be updated every time the protocol changes.
FieldNamesProtocolVersion = "v1"
// FieldValuesProtocolVersion is the version of the protocol used for /internal/select/field_values HTTP endpoint.
//
// It must be updated every time the protocol changes.
FieldValuesProtocolVersion = "v1"
// StreamFieldNamesProtocolVersion is the version of the protocol used for /internal/select/stream_field_names HTTP endpoint.
//
// It must be updated every time the protocol changes.
StreamFieldNamesProtocolVersion = "v1"
// StreamFieldValuesProtocolVersion is the version of the protocol used for /internal/select/stream_field_values HTTP endpoint.
//
// It must be updated every time the protocol changes.
StreamFieldValuesProtocolVersion = "v1"
// StreamsProtocolVersion is the version of the protocol used for /internal/select/streams HTTP endpoint.
//
// It must be updated every time the protocol changes.
StreamsProtocolVersion = "v1"
// StreamIDsProtocolVersion is the version of the protocol used for /internal/select/stream_ids HTTP endpoint.
//
// It must be updated every time the protocol changes.
StreamIDsProtocolVersion = "v1"
// QueryProtocolVersion is the version of the protocol used for /internal/select/query HTTP endpoint.
//
// It must be updated every time the protocol changes.
QueryProtocolVersion = "v1"
)
// Storage is a network storage for querying remote storage nodes in the cluster.
type Storage struct {
sns []*storageNode
disableCompression bool
}
type storageNode struct {
// scheme is http or https scheme to communicate with addr
scheme string
// addr is TCP address of the storage node to query
addr string
// s is a storage, which holds the given storageNode
s *Storage
// c is an http client used for querying storage node at addr.
c *http.Client
// ac is auth config used for setting request headers such as Authorization and Host.
ac *promauth.Config
}
func newStorageNode(s *Storage, addr string, ac *promauth.Config, isTLS bool) *storageNode {
tr := httputil.NewTransport(false, "vlselect_backend")
tr.TLSHandshakeTimeout = 20 * time.Second
tr.DisableCompression = true
scheme := "http"
if isTLS {
scheme = "https"
}
sn := &storageNode{
scheme: scheme,
addr: addr,
s: s,
c: &http.Client{
Transport: ac.NewRoundTripper(tr),
},
ac: ac,
}
return sn
}
func (sn *storageNode) runQuery(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, processBlock func(db *logstorage.DataBlock)) error {
args := sn.getCommonArgs(QueryProtocolVersion, tenantIDs, q)
reqURL := sn.getRequestURL("/internal/select/query", args)
req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil)
if err != nil {
logger.Panicf("BUG: unexpected error when creating a request: %s", err)
}
if err := sn.ac.SetHeaders(req, true); err != nil {
return fmt.Errorf("cannot set auth headers for %q: %w", reqURL, err)
}
// send the request to the storage node
resp, err := sn.c.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
responseBody, err := io.ReadAll(resp.Body)
if err != nil {
responseBody = []byte(err.Error())
}
return fmt.Errorf("unexpected status code for the request to %q: %d; want %d; response: %q", reqURL, resp.StatusCode, http.StatusOK, responseBody)
}
// read the response
var dataLenBuf [8]byte
var buf []byte
var db logstorage.DataBlock
var valuesBuf []string
for {
if _, err := io.ReadFull(resp.Body, dataLenBuf[:]); err != nil {
if errors.Is(err, io.EOF) {
// The end of response stream
return nil
}
return fmt.Errorf("cannot read block size from %q: %w", reqURL, err)
}
blockLen := encoding.UnmarshalUint64(dataLenBuf[:])
if blockLen > math.MaxInt {
return fmt.Errorf("too big data block: %d bytes; mustn't exceed %v bytes", blockLen, math.MaxInt)
}
buf = slicesutil.SetLength(buf, int(blockLen))
if _, err := io.ReadFull(resp.Body, buf); err != nil {
return fmt.Errorf("cannot read block with size of %d bytes from %q: %w", blockLen, reqURL, err)
}
src := buf
if !sn.s.disableCompression {
bufLen := len(buf)
var err error
buf, err = zstd.Decompress(buf, buf)
if err != nil {
return fmt.Errorf("cannot decompress data block: %w", err)
}
src = buf[bufLen:]
}
for len(src) > 0 {
tail, vb, err := db.UnmarshalInplace(src, valuesBuf[:0])
if err != nil {
return fmt.Errorf("cannot unmarshal data block received from %q: %w", reqURL, err)
}
valuesBuf = vb
src = tail
processBlock(&db)
clear(valuesBuf)
}
}
}
func (sn *storageNode) getFieldNames(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query) ([]logstorage.ValueWithHits, error) {
args := sn.getCommonArgs(FieldNamesProtocolVersion, tenantIDs, q)
return sn.getValuesWithHits(ctx, "/internal/select/field_names", args)
}
func (sn *storageNode) getFieldValues(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, fieldName string, limit uint64) ([]logstorage.ValueWithHits, error) {
args := sn.getCommonArgs(FieldValuesProtocolVersion, tenantIDs, q)
args.Set("field", fieldName)
args.Set("limit", fmt.Sprintf("%d", limit))
return sn.getValuesWithHits(ctx, "/internal/select/field_values", args)
}
func (sn *storageNode) getStreamFieldNames(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query) ([]logstorage.ValueWithHits, error) {
args := sn.getCommonArgs(StreamFieldNamesProtocolVersion, tenantIDs, q)
return sn.getValuesWithHits(ctx, "/internal/select/stream_field_names", args)
}
func (sn *storageNode) getStreamFieldValues(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, fieldName string, limit uint64) ([]logstorage.ValueWithHits, error) {
args := sn.getCommonArgs(StreamFieldValuesProtocolVersion, tenantIDs, q)
args.Set("field", fieldName)
args.Set("limit", fmt.Sprintf("%d", limit))
return sn.getValuesWithHits(ctx, "/internal/select/stream_field_values", args)
}
func (sn *storageNode) getStreams(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit uint64) ([]logstorage.ValueWithHits, error) {
args := sn.getCommonArgs(StreamsProtocolVersion, tenantIDs, q)
args.Set("limit", fmt.Sprintf("%d", limit))
return sn.getValuesWithHits(ctx, "/internal/select/streams", args)
}
func (sn *storageNode) getStreamIDs(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit uint64) ([]logstorage.ValueWithHits, error) {
args := sn.getCommonArgs(StreamIDsProtocolVersion, tenantIDs, q)
args.Set("limit", fmt.Sprintf("%d", limit))
return sn.getValuesWithHits(ctx, "/internal/select/stream_ids", args)
}
func (sn *storageNode) getCommonArgs(version string, tenantIDs []logstorage.TenantID, q *logstorage.Query) url.Values {
args := url.Values{}
args.Set("version", version)
args.Set("tenant_ids", string(logstorage.MarshalTenantIDs(nil, tenantIDs)))
args.Set("query", q.String())
args.Set("timestamp", fmt.Sprintf("%d", q.GetTimestamp()))
args.Set("disable_compression", fmt.Sprintf("%v", sn.s.disableCompression))
return args
}
func (sn *storageNode) getValuesWithHits(ctx context.Context, path string, args url.Values) ([]logstorage.ValueWithHits, error) {
data, err := sn.executeRequestAt(ctx, path, args)
if err != nil {
return nil, err
}
return unmarshalValuesWithHits(data)
}
func (sn *storageNode) executeRequestAt(ctx context.Context, path string, args url.Values) ([]byte, error) {
reqURL := sn.getRequestURL(path, args)
req, err := http.NewRequestWithContext(ctx, "GET", reqURL, nil)
if err != nil {
logger.Panicf("BUG: unexpected error when creating a request: %s", err)
}
if err := sn.ac.SetHeaders(req, true); err != nil {
return nil, fmt.Errorf("cannot set auth headers for %q: %w", reqURL, err)
}
// send the request to the storage node
resp, err := sn.c.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
responseBody, err := io.ReadAll(resp.Body)
if err != nil {
responseBody = []byte(err.Error())
}
return nil, fmt.Errorf("unexpected status code for the request to %q: %d; want %d; response: %q", reqURL, resp.StatusCode, http.StatusOK, responseBody)
}
// read the response
var bb bytesutil.ByteBuffer
if _, err := bb.ReadFrom(resp.Body); err != nil {
return nil, fmt.Errorf("cannot read response from %q: %w", reqURL, err)
}
if sn.s.disableCompression {
return bb.B, nil
}
bbLen := len(bb.B)
bb.B, err = zstd.Decompress(bb.B, bb.B)
if err != nil {
return nil, err
}
return bb.B[bbLen:], nil
}
func (sn *storageNode) getRequestURL(path string, args url.Values) string {
return fmt.Sprintf("%s://%s%s?%s", sn.scheme, sn.addr, path, args.Encode())
}
// NewStorage returns new Storage for the given addrs and the given authCfgs.
//
// If disableCompression is set, then uncompressed responses are received from storage nodes.
//
// Call MustStop on the returned storage when it is no longer needed.
func NewStorage(addrs []string, authCfgs []*promauth.Config, isTLSs []bool, disableCompression bool) *Storage {
s := &Storage{
disableCompression: disableCompression,
}
sns := make([]*storageNode, len(addrs))
for i, addr := range addrs {
sns[i] = newStorageNode(s, addr, authCfgs[i], isTLSs[i])
}
s.sns = sns
return s
}
// MustStop stops the s.
func (s *Storage) MustStop() {
s.sns = nil
}
// RunQuery runs the given q and calls writeBlock for the returned data blocks
func (s *Storage) RunQuery(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, writeBlock logstorage.WriteDataBlockFunc) error {
nqr, err := logstorage.NewNetQueryRunner(ctx, tenantIDs, q, s.RunQuery, writeBlock)
if err != nil {
return err
}
search := func(stopCh <-chan struct{}, q *logstorage.Query, writeBlock logstorage.WriteDataBlockFunc) error {
return s.runQuery(stopCh, tenantIDs, q, writeBlock)
}
concurrency := q.GetConcurrency()
return nqr.Run(ctx, concurrency, search)
}
func (s *Storage) runQuery(stopCh <-chan struct{}, tenantIDs []logstorage.TenantID, q *logstorage.Query, writeBlock logstorage.WriteDataBlockFunc) error {
ctxWithCancel, cancel := contextutil.NewStopChanContext(stopCh)
defer cancel()
errs := make([]error, len(s.sns))
var wg sync.WaitGroup
for i := range s.sns {
wg.Add(1)
go func(nodeIdx int) {
defer wg.Done()
sn := s.sns[nodeIdx]
err := sn.runQuery(ctxWithCancel, tenantIDs, q, func(db *logstorage.DataBlock) {
writeBlock(uint(nodeIdx), db)
})
if err != nil {
// Cancel the remaining parallel queries
cancel()
}
errs[nodeIdx] = err
}(i)
}
wg.Wait()
return getFirstNonCancelError(errs)
}
// GetFieldNames executes q and returns field names seen in results.
func (s *Storage) GetFieldNames(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query) ([]logstorage.ValueWithHits, error) {
return s.getValuesWithHits(ctx, 0, false, func(ctx context.Context, sn *storageNode) ([]logstorage.ValueWithHits, error) {
return sn.getFieldNames(ctx, tenantIDs, q)
})
}
// GetFieldValues executes q and returns unique values for the fieldName seen in results.
//
// If limit > 0, then up to limit unique values are returned.
func (s *Storage) GetFieldValues(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, fieldName string, limit uint64) ([]logstorage.ValueWithHits, error) {
return s.getValuesWithHits(ctx, limit, true, func(ctx context.Context, sn *storageNode) ([]logstorage.ValueWithHits, error) {
return sn.getFieldValues(ctx, tenantIDs, q, fieldName, limit)
})
}
// GetStreamFieldNames executes q and returns stream field names seen in results.
func (s *Storage) GetStreamFieldNames(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query) ([]logstorage.ValueWithHits, error) {
return s.getValuesWithHits(ctx, 0, false, func(ctx context.Context, sn *storageNode) ([]logstorage.ValueWithHits, error) {
return sn.getStreamFieldNames(ctx, tenantIDs, q)
})
}
// GetStreamFieldValues executes q and returns stream field values for the given fieldName seen in results.
//
// If limit > 0, then up to limit unique stream field values are returned.
func (s *Storage) GetStreamFieldValues(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, fieldName string, limit uint64) ([]logstorage.ValueWithHits, error) {
return s.getValuesWithHits(ctx, limit, true, func(ctx context.Context, sn *storageNode) ([]logstorage.ValueWithHits, error) {
return sn.getStreamFieldValues(ctx, tenantIDs, q, fieldName, limit)
})
}
// GetStreams executes q and returns streams seen in query results.
//
// If limit > 0, then up to limit unique streams are returned.
func (s *Storage) GetStreams(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit uint64) ([]logstorage.ValueWithHits, error) {
return s.getValuesWithHits(ctx, limit, true, func(ctx context.Context, sn *storageNode) ([]logstorage.ValueWithHits, error) {
return sn.getStreams(ctx, tenantIDs, q, limit)
})
}
// GetStreamIDs executes q and returns streamIDs seen in query results.
//
// If limit > 0, then up to limit unique streamIDs are returned.
func (s *Storage) GetStreamIDs(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit uint64) ([]logstorage.ValueWithHits, error) {
return s.getValuesWithHits(ctx, limit, true, func(ctx context.Context, sn *storageNode) ([]logstorage.ValueWithHits, error) {
return sn.getStreamIDs(ctx, tenantIDs, q, limit)
})
}
func (s *Storage) getValuesWithHits(ctx context.Context, limit uint64, resetHitsOnLimitExceeded bool,
callback func(ctx context.Context, sn *storageNode) ([]logstorage.ValueWithHits, error)) ([]logstorage.ValueWithHits, error) {
ctxWithCancel, cancel := context.WithCancel(ctx)
defer cancel()
results := make([][]logstorage.ValueWithHits, len(s.sns))
errs := make([]error, len(s.sns))
var wg sync.WaitGroup
for i := range s.sns {
wg.Add(1)
go func(nodeIdx int) {
defer wg.Done()
sn := s.sns[nodeIdx]
vhs, err := callback(ctxWithCancel, sn)
results[nodeIdx] = vhs
errs[nodeIdx] = err
if err != nil {
// Cancel the remaining parallel requests
cancel()
}
}(i)
}
wg.Wait()
if err := getFirstNonCancelError(errs); err != nil {
return nil, err
}
vhs := logstorage.MergeValuesWithHits(results, limit, resetHitsOnLimitExceeded)
return vhs, nil
}
func getFirstNonCancelError(errs []error) error {
for _, err := range errs {
if err != nil && !errors.Is(err, context.Canceled) {
return err
}
}
return nil
}
func unmarshalValuesWithHits(src []byte) ([]logstorage.ValueWithHits, error) {
var vhs []logstorage.ValueWithHits
for len(src) > 0 {
var vh logstorage.ValueWithHits
tail, err := vh.UnmarshalInplace(src)
if err != nil {
return nil, fmt.Errorf("cannot unmarshal ValueWithHits #%d: %w", len(vhs), err)
}
src = tail
// Clone vh.Value, since it points to src.
vh.Value = strings.Clone(vh.Value)
vhs = append(vhs, vh)
}
return vhs, nil
}

View File

@@ -1,3 +1,3 @@
See vmagent docs [here](https://docs.victoriametrics.com/victoriametrics/vmagent/).
See vmagent docs [here](https://docs.victoriametrics.com/vmagent/).
vmagent docs can be edited at [docs/vmagent.md](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/victoriametrics/vmagent.md).
vmagent docs can be edited at [docs/vmagent.md](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/vmagent.md).

View File

@@ -105,7 +105,7 @@ func main() {
// Some workloads may need increased GOGC values. Then such values can be set via GOGC environment variable.
// It is recommended increasing GOGC if go_memstats_gc_cpu_fraction metric exposed at /metrics page
// exceeds 0.05 for extended periods of time.
cgroup.SetGOGC(50)
cgroup.SetGOGC(30)
// Write flags and help message to stdout, since it is easier to grep or pipe.
flag.CommandLine.SetOutput(os.Stdout)
@@ -165,9 +165,7 @@ func main() {
promscrape.Init(remotewrite.PushDropSamplesOnFailure)
go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{
UseProxyProtocol: useProxyProtocol,
})
go httpserver.Serve(listenAddrs, useProxyProtocol, requestHandler)
logger.Infof("started vmagent in %.3f seconds", time.Since(startTime).Seconds())
pushmetrics.Init()
@@ -244,7 +242,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
}
w.Header().Add("Content-Type", "text/html; charset=utf-8")
fmt.Fprintf(w, "<h2>vmagent</h2>")
fmt.Fprintf(w, "See docs at <a href='https://docs.victoriametrics.com/victoriametrics/vmagent/'>https://docs.victoriametrics.com/victoriametrics/vmagent/</a></br>")
fmt.Fprintf(w, "See docs at <a href='https://docs.victoriametrics.com/vmagent/'>https://docs.victoriametrics.com/vmagent/</a></br>")
fmt.Fprintf(w, "Useful endpoints:</br>")
httpserver.WriteAPIHelp(w, [][2]string{
{"targets", "status for discovered active targets"},
@@ -445,10 +443,8 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
case "/prometheus/api/v1/targets", "/api/v1/targets":
promscrapeAPIV1TargetsRequests.Inc()
w.Header().Set("Content-Type", "application/json")
// https://prometheus.io/docs/prometheus/latest/querying/api/#targets
state := r.FormValue("state")
scrapePool := r.FormValue("scrapePool")
promscrape.WriteAPIV1Targets(w, state, scrapePool)
promscrape.WriteAPIV1Targets(w, state)
return true
case "/prometheus/target_response", "/target_response":
promscrapeTargetResponseRequests.Inc()
@@ -754,7 +750,7 @@ func usage() {
const s = `
vmagent collects metrics data via popular data ingestion protocols and routes it to VictoriaMetrics.
See the docs at https://docs.victoriametrics.com/victoriametrics/vmagent/ .
See the docs at https://docs.victoriametrics.com/vmagent/ .
`
flagutil.Usage(s)
}

View File

@@ -9,5 +9,4 @@ COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certifica
EXPOSE 8429
ENTRYPOINT ["/vmagent-prod"]
ARG TARGETARCH
ARG BINARY_SUFFIX=non-existing
COPY vmagent-linux-${TARGETARCH}-prod${BINARY_SUFFIX} ./vmagent-prod
COPY vmagent-linux-${TARGETARCH}-prod ./vmagent-prod

View File

@@ -10,29 +10,27 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/awsapi"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/ratelimiter"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
"github.com/VictoriaMetrics/metrics"
"github.com/golang/snappy"
)
var (
forcePromProto = flagutil.NewArrayBool("remoteWrite.forcePromProto", "Whether to force Prometheus remote write protocol for sending data "+
"to the corresponding -remoteWrite.url . See https://docs.victoriametrics.com/victoriametrics/vmagent/#victoriametrics-remote-write-protocol")
"to the corresponding -remoteWrite.url . See https://docs.victoriametrics.com/vmagent/#victoriametrics-remote-write-protocol")
forceVMProto = flagutil.NewArrayBool("remoteWrite.forceVMProto", "Whether to force VictoriaMetrics remote write protocol for sending data "+
"to the corresponding -remoteWrite.url . See https://docs.victoriametrics.com/victoriametrics/vmagent/#victoriametrics-remote-write-protocol")
"to the corresponding -remoteWrite.url . See https://docs.victoriametrics.com/vmagent/#victoriametrics-remote-write-protocol")
rateLimit = flagutil.NewArrayInt("remoteWrite.rateLimit", 0, "Optional rate limit in bytes per second for data sent to the corresponding -remoteWrite.url. "+
"By default, the rate limit is disabled. It can be useful for limiting load on remote storage when big amounts of buffered data "+
@@ -90,8 +88,7 @@ type client struct {
remoteWriteURL string
// Whether to use VictoriaMetrics remote write protocol for sending the data to remoteWriteURL
useVMProto atomic.Bool
canDowngradeVMProto atomic.Bool
useVMProto bool
fq *persistentqueue.FastQueue
hc *http.Client
@@ -129,7 +126,8 @@ func newHTTPClient(argIdx int, remoteWriteURL, sanitizedURL string, fq *persiste
logger.Fatalf("cannot initialize AWS Config for -remoteWrite.url=%q: %s", remoteWriteURL, err)
}
tr := httputil.NewTransport(false, "vmagent_remotewrite")
tr := httputil.NewTransport(false)
tr.DialContext = netutil.NewStatDialFunc("vmagent_remotewrite")
tr.TLSHandshakeTimeout = tlsHandshakeTimeout.GetOptionalArg(argIdx)
tr.MaxConnsPerHost = 2 * concurrency
tr.MaxIdleConnsPerHost = 2 * concurrency
@@ -170,11 +168,17 @@ func newHTTPClient(argIdx int, remoteWriteURL, sanitizedURL string, fq *persiste
logger.Fatalf("-remoteWrite.useVMProto and -remoteWrite.usePromProto cannot be set simultaneously for -remoteWrite.url=%s", sanitizedURL)
}
if !useVMProto && !usePromProto {
// The VM protocol could be downgraded later at runtime if unsupported media type response status is received.
useVMProto = true
c.canDowngradeVMProto.Store(true)
// Auto-detect whether the remote storage supports VictoriaMetrics remote write protocol.
doRequest := func(url string) (*http.Response, error) {
return c.doRequest(url, nil)
}
useVMProto = protoparserutil.HandleVMProtoClientHandshake(c.remoteWriteURL, doRequest)
if !useVMProto {
logger.Infof("the remote storage at %q doesn't support VictoriaMetrics remote write protocol. Switching to Prometheus remote write protocol. "+
"See https://docs.victoriametrics.com/vmagent/#victoriametrics-remote-write-protocol", sanitizedURL)
}
}
c.useVMProto.Store(useVMProto)
c.useVMProto = useVMProto
return c
}
@@ -382,7 +386,7 @@ func (c *client) newRequest(url string, body []byte) (*http.Request, error) {
h := req.Header
h.Set("User-Agent", "vmagent")
h.Set("Content-Type", "application/x-protobuf")
if encoding.IsZstd(body) {
if c.useVMProto {
h.Set("Content-Encoding", "zstd")
h.Set("X-VictoriaMetrics-Remote-Write-Version", "1")
} else {
@@ -418,7 +422,7 @@ again:
if retryDuration > maxRetryDuration {
retryDuration = maxRetryDuration
}
remoteWriteRetryLogger.Warnf("couldn't send a block with size %d bytes to %q: %s; re-sending the block in %.3f seconds",
logger.Warnf("couldn't send a block with size %d bytes to %q: %s; re-sending the block in %.3f seconds",
len(block), c.sanitizedURL, err, retryDuration.Seconds())
t := timerpool.Get(retryDuration)
select {
@@ -431,7 +435,6 @@ again:
c.retriesCount.Inc()
goto again
}
statusCode := resp.StatusCode
if statusCode/100 == 2 {
_ = resp.Body.Close()
@@ -440,43 +443,21 @@ again:
c.blocksSent.Inc()
return true
}
metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_requests_total{url=%q, status_code="%d"}`, c.sanitizedURL, statusCode)).Inc()
if statusCode == 409 {
logBlockRejected(block, c.sanitizedURL, resp)
// Just drop block on 409 status code like Prometheus does.
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/873
// and https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1149
if statusCode == 409 || statusCode == 400 {
body, err := io.ReadAll(resp.Body)
_ = resp.Body.Close()
c.packetsDropped.Inc()
return true
// - Remote Write v1 specification implicitly expects a `400 Bad Request` when the encoding is not supported.
// - Remote Write v2 specification explicitly specifies a `415 Unsupported Media Type` for unsupported encodings.
// - Real-world implementations of v1 use both 400 and 415 status codes.
// See more in research: https://github.com/VictoriaMetrics/VictoriaMetrics/pull/8462#issuecomment-2786918054
} else if statusCode == 415 || statusCode == 400 {
if c.canDowngradeVMProto.Swap(false) {
logger.Infof("received unsupported media type or bad request from remote storage at %q. Downgrading protocol from VictoriaMetrics to Prometheus remote write for all future requests. "+
"See https://docs.victoriametrics.com/victoriametrics/vmagent/#victoriametrics-remote-write-protocol", c.sanitizedURL)
c.useVMProto.Store(false)
if err != nil {
remoteWriteRejectedLogger.Errorf("sending a block with size %d bytes to %q was rejected (skipping the block): status code %d; "+
"failed to read response body: %s",
len(block), c.sanitizedURL, statusCode, err)
} else {
remoteWriteRejectedLogger.Errorf("sending a block with size %d bytes to %q was rejected (skipping the block): status code %d; response body: %s",
len(block), c.sanitizedURL, statusCode, string(body))
}
if encoding.IsZstd(block) {
logger.Infof("received unsupported media type or bad request from remote storage at %q. Re-packing the block to Prometheus remote write and retrying."+
"See https://docs.victoriametrics.com/victoriametrics/vmagent/#victoriametrics-remote-write-protocol", c.sanitizedURL)
block = mustRepackBlockFromZstdToSnappy(block)
c.retriesCount.Inc()
_ = resp.Body.Close()
goto again
}
// Just drop snappy blocks on 400 or 415 status codes like Prometheus does.
// Just drop block on 409 and 400 status codes like Prometheus does.
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/873
// and https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1149
logBlockRejected(block, c.sanitizedURL, resp)
_ = resp.Body.Close()
c.packetsDropped.Inc()
return true
@@ -509,7 +490,6 @@ again:
}
var remoteWriteRejectedLogger = logger.WithThrottler("remoteWriteRejected", 5*time.Second)
var remoteWriteRetryLogger = logger.WithThrottler("remoteWriteRetry", 5*time.Second)
// getRetryDuration returns retry duration.
// retryAfterDuration has the highest priority.
@@ -532,28 +512,6 @@ func getRetryDuration(retryAfterDuration, retryDuration, maxRetryDuration time.D
return retryDuration
}
func mustRepackBlockFromZstdToSnappy(zstdBlock []byte) []byte {
plainBlock := make([]byte, 0, len(zstdBlock)*2)
plainBlock, err := zstd.Decompress(plainBlock, zstdBlock)
if err != nil {
logger.Panicf("FATAL: cannot re-pack block with size %d bytes from Zstd to Snappy: %s", len(zstdBlock), err)
}
return snappy.Encode(nil, plainBlock)
}
func logBlockRejected(block []byte, sanitizedURL string, resp *http.Response) {
body, err := io.ReadAll(resp.Body)
if err != nil {
remoteWriteRejectedLogger.Errorf("sending a block with size %d bytes to %q was rejected (skipping the block): status code %d; "+
"failed to read response body: %s",
len(block), sanitizedURL, resp.StatusCode, err)
} else {
remoteWriteRejectedLogger.Errorf("sending a block with size %d bytes to %q was rejected (skipping the block): status code %d; response body: %s",
len(block), sanitizedURL, resp.StatusCode, string(body))
}
}
// parseRetryAfterHeader parses `Retry-After` value retrieved from HTTP response header.
// retryAfterString should be in either HTTP-date or a number of seconds.
// It will return time.Duration(0) if `retryAfterString` does not follow RFC 7231.

View File

@@ -5,9 +5,6 @@ import (
"net/http"
"testing"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
"github.com/golang/snappy"
)
func TestCalculateRetryDuration(t *testing.T) {
@@ -100,19 +97,3 @@ func helper(d time.Duration) time.Duration {
return d + dv
}
func TestRepackBlockFromZstdToSnappy(t *testing.T) {
expectedPlainBlock := []byte(`foobar`)
zstdBlock := encoding.CompressZSTDLevel(nil, expectedPlainBlock, 1)
snappyBlock := mustRepackBlockFromZstdToSnappy(zstdBlock)
actualPlainBlock, err := snappy.Decode(nil, snappyBlock)
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
if string(actualPlainBlock) != string(expectedPlainBlock) {
t.Fatalf("unexpected plain block; got %q; want %q", actualPlainBlock, expectedPlainBlock)
}
}

View File

@@ -16,7 +16,6 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
"github.com/VictoriaMetrics/metrics"
"github.com/golang/snappy"
@@ -29,7 +28,7 @@ var (
maxRowsPerBlock = flag.Int("remoteWrite.maxRowsPerBlock", 10000, "The maximum number of samples to send in each block to remote storage. Higher number may improve performance at the cost of the increased memory usage. See also -remoteWrite.maxBlockSize")
vmProtoCompressLevel = flag.Int("remoteWrite.vmProtoCompressLevel", 0, "The compression level for VictoriaMetrics remote write protocol. "+
"Higher values reduce network traffic at the cost of higher CPU usage. Negative values reduce CPU usage at the cost of increased network traffic. "+
"See https://docs.victoriametrics.com/victoriametrics/vmagent/#victoriametrics-remote-write-protocol")
"See https://docs.victoriametrics.com/vmagent/#victoriametrics-remote-write-protocol")
)
type pendingSeries struct {
@@ -40,7 +39,7 @@ type pendingSeries struct {
periodicFlusherWG sync.WaitGroup
}
func newPendingSeries(fq *persistentqueue.FastQueue, isVMRemoteWrite *atomic.Bool, significantFigures, roundDigits int) *pendingSeries {
func newPendingSeries(fq *persistentqueue.FastQueue, isVMRemoteWrite bool, significantFigures, roundDigits int) *pendingSeries {
var ps pendingSeries
ps.wr.fq = fq
ps.wr.isVMRemoteWrite = isVMRemoteWrite
@@ -100,7 +99,7 @@ type writeRequest struct {
fq *persistentqueue.FastQueue
// Whether to encode the write request with VictoriaMetrics remote write protocol.
isVMRemoteWrite *atomic.Bool
isVMRemoteWrite bool
// How many significant figures must be left before sending the writeRequest to fq.
significantFigures int
@@ -138,7 +137,7 @@ func (wr *writeRequest) reset() {
// This is needed in order to properly save in-memory data to persistent queue on graceful shutdown.
func (wr *writeRequest) mustFlushOnStop() {
wr.wr.Timeseries = wr.tss
if !tryPushWriteRequest(&wr.wr, wr.mustWriteBlock, wr.isVMRemoteWrite.Load()) {
if !tryPushWriteRequest(&wr.wr, wr.mustWriteBlock, wr.isVMRemoteWrite) {
logger.Panicf("BUG: final flush must always return true")
}
wr.reset()
@@ -152,7 +151,7 @@ func (wr *writeRequest) mustWriteBlock(block []byte) bool {
func (wr *writeRequest) tryFlush() bool {
wr.wr.Timeseries = wr.tss
wr.lastFlushTime.Store(fasttime.UnixTimestamp())
if !tryPushWriteRequest(&wr.wr, wr.fq.TryWriteBlock, wr.isVMRemoteWrite.Load()) {
if !tryPushWriteRequest(&wr.wr, wr.fq.TryWriteBlock, wr.isVMRemoteWrite) {
return false
}
wr.reset()
@@ -198,43 +197,28 @@ func (wr *writeRequest) tryPush(src []prompbmarshal.TimeSeries) bool {
}
func (wr *writeRequest) copyTimeSeries(dst, src *prompbmarshal.TimeSeries) {
labelsSrc := src.Labels
// Pre-allocate memory for labels.
labelsDst := wr.labels
labelsLen := len(wr.labels)
wr.labels = slicesutil.SetLength(wr.labels, labelsLen+len(labelsSrc))
labelsDst := wr.labels[labelsLen:]
samplesDst := wr.samples
buf := wr.buf
for i := range src.Labels {
labelsDst = append(labelsDst, prompbmarshal.Label{})
dstLabel := &labelsDst[len(labelsDst)-1]
srcLabel := &src.Labels[i]
// Pre-allocate memory for byte slice needed for storing label names and values.
neededBufLen := 0
for i := range labelsSrc {
label := &labelsSrc[i]
neededBufLen += len(label.Name) + len(label.Value)
}
bufLen := len(wr.buf)
wr.buf = slicesutil.SetLength(wr.buf, bufLen+neededBufLen)
buf := wr.buf[:bufLen]
// Copy labels
for i := range labelsSrc {
dstLabel := &labelsDst[i]
srcLabel := &labelsSrc[i]
bufLen := len(buf)
buf = append(buf, srcLabel.Name...)
dstLabel.Name = bytesutil.ToUnsafeString(buf[bufLen:])
bufLen = len(buf)
dstLabel.Name = bytesutil.ToUnsafeString(buf[len(buf)-len(srcLabel.Name):])
buf = append(buf, srcLabel.Value...)
dstLabel.Value = bytesutil.ToUnsafeString(buf[bufLen:])
dstLabel.Value = bytesutil.ToUnsafeString(buf[len(buf)-len(srcLabel.Value):])
}
wr.buf = buf
dst.Labels = labelsDst
dst.Labels = labelsDst[labelsLen:]
// Copy samples
samplesLen := len(wr.samples)
wr.samples = append(wr.samples, src.Samples...)
dst.Samples = wr.samples[samplesLen:]
samplesDst = append(samplesDst, src.Samples...)
dst.Samples = samplesDst[len(samplesDst)-len(src.Samples):]
wr.samples = samplesDst
wr.labels = labelsDst
wr.buf = buf
}
// marshalConcurrency limits the maximum number of concurrent workers, which marshal and compress WriteRequest.

View File

@@ -7,13 +7,10 @@ import (
"strings"
"sync"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
"github.com/VictoriaMetrics/metrics"
)
var (
@@ -22,10 +19,10 @@ var (
relabelConfigPathGlobal = flag.String("remoteWrite.relabelConfig", "", "Optional path to file with relabeling configs, which are applied "+
"to all the metrics before sending them to -remoteWrite.url. See also -remoteWrite.urlRelabelConfig. "+
"The path can point either to local file or to http url. "+
"See https://docs.victoriametrics.com/victoriametrics/relabeling/")
"See https://docs.victoriametrics.com/vmagent/#relabeling")
relabelConfigPaths = flagutil.NewArrayString("remoteWrite.urlRelabelConfig", "Optional path to relabel configs for the corresponding -remoteWrite.url. "+
"See also -remoteWrite.relabelConfig. The path can point either to local file or to http url. "+
"See https://docs.victoriametrics.com/victoriametrics/relabeling/")
"See https://docs.victoriametrics.com/vmagent/#relabeling")
usePromCompatibleNaming = flag.Bool("usePromCompatibleNaming", false, "Whether to replace characters unsupported by Prometheus with underscores "+
"in the ingested metric names and label names. For example, foo.bar{a.b='c'} is transformed into foo_bar{a_b='c'} during data ingestion if this flag is set. "+
@@ -34,59 +31,12 @@ var (
var labelsGlobal []prompbmarshal.Label
var (
relabelConfigReloads *metrics.Counter
relabelConfigReloadErrors *metrics.Counter
relabelConfigSuccess *metrics.Gauge
relabelConfigTimestamp *metrics.Counter
)
func initRelabelMetrics() {
relabelConfigReloads = metrics.NewCounter(`vmagent_relabel_config_reloads_total`)
relabelConfigReloadErrors = metrics.NewCounter(`vmagent_relabel_config_reloads_errors_total`)
relabelConfigSuccess = metrics.NewGauge(`vmagent_relabel_config_last_reload_successful`, nil)
relabelConfigTimestamp = metrics.NewCounter(`vmagent_relabel_config_last_reload_success_timestamp_seconds`)
}
// CheckRelabelConfigs checks -remoteWrite.relabelConfig and -remoteWrite.urlRelabelConfig.
func CheckRelabelConfigs() error {
_, err := loadRelabelConfigs()
return err
}
func initRelabelConfigs() {
rcs, err := loadRelabelConfigs()
if err != nil {
logger.Fatalf("cannot initialize relabel configs: %s", err)
}
allRelabelConfigs.Store(rcs)
if rcs.isSet() {
initRelabelMetrics()
relabelConfigSuccess.Set(1)
relabelConfigTimestamp.Set(fasttime.UnixTimestamp())
}
}
func reloadRelabelConfigs() {
rcs := allRelabelConfigs.Load()
if !rcs.isSet() {
return
}
relabelConfigReloads.Inc()
logger.Infof("reloading relabel configs pointed by -remoteWrite.relabelConfig and -remoteWrite.urlRelabelConfig")
rcs, err := loadRelabelConfigs()
if err != nil {
relabelConfigReloadErrors.Inc()
relabelConfigSuccess.Set(0)
logger.Errorf("cannot reload relabel configs; preserving the previous configs; error: %s", err)
return
}
allRelabelConfigs.Store(rcs)
relabelConfigSuccess.Set(1)
relabelConfigTimestamp.Set(fasttime.UnixTimestamp())
logger.Infof("successfully reloaded relabel configs")
}
func loadRelabelConfigs() (*relabelConfigs, error) {
var rcs relabelConfigs
if *relabelConfigPathGlobal != "" {
@@ -120,21 +70,6 @@ type relabelConfigs struct {
perURL []*promrelabel.ParsedConfigs
}
func (rcs *relabelConfigs) isSet() bool {
if rcs == nil {
return false
}
if rcs.global.Len() > 0 {
return true
}
for _, pc := range rcs.perURL {
if pc.Len() > 0 {
return true
}
}
return false
}
// initLabelsGlobal must be called after parsing command-line flags.
func initLabelsGlobal() {
labelsGlobal = nil

View File

@@ -15,7 +15,7 @@ import (
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bloomfilter"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/consistenthash"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -39,21 +39,22 @@ var (
"Pass multiple -remoteWrite.url options in order to replicate the collected data to multiple remote storage systems. "+
"The data can be sharded among the configured remote storage systems if -remoteWrite.shardByURL flag is set")
enableMultitenantHandlers = flag.Bool("enableMultitenantHandlers", false, "Whether to process incoming data via multitenant insert handlers according to "+
"https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#url-format . By default incoming data is processed via single-node insert handlers "+
"according to https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-import-time-series-data ."+
"See https://docs.victoriametrics.com/victoriametrics/vmagent/#multitenancy for details")
"https://docs.victoriametrics.com/cluster-victoriametrics/#url-format . By default incoming data is processed via single-node insert handlers "+
"according to https://docs.victoriametrics.com/#how-to-import-time-series-data ."+
"See https://docs.victoriametrics.com/vmagent/#multitenancy for details")
shardByURL = flag.Bool("remoteWrite.shardByURL", false, "Whether to shard outgoing series across all the remote storage systems enumerated via -remoteWrite.url. "+
"By default the data is replicated across all the -remoteWrite.url . See https://docs.victoriametrics.com/victoriametrics/vmagent/#sharding-among-remote-storages . "+
shardByURL = flag.Bool("remoteWrite.shardByURL", false, "Whether to shard outgoing series across all the remote storage systems enumerated via -remoteWrite.url . "+
"By default the data is replicated across all the -remoteWrite.url . See https://docs.victoriametrics.com/vmagent/#sharding-among-remote-storages . "+
"See also -remoteWrite.shardByURLReplicas")
shardByURLReplicas = flag.Int("remoteWrite.shardByURLReplicas", 1, "How many copies of data to make among remote storage systems enumerated via -remoteWrite.url "+
"when -remoteWrite.shardByURL is set. See https://docs.victoriametrics.com/victoriametrics/vmagent/#sharding-among-remote-storages")
"when -remoteWrite.shardByURL is set. See https://docs.victoriametrics.com/vmagent/#sharding-among-remote-storages")
shardByURLLabels = flagutil.NewArrayString("remoteWrite.shardByURL.labels", "Optional list of labels, which must be used for sharding outgoing samples "+
"among remote storage systems if -remoteWrite.shardByURL command-line flag is set. By default all the labels are used for sharding in order to gain "+
"even distribution of series over the specified -remoteWrite.url systems. See also -remoteWrite.shardByURL.ignoreLabels")
shardByURLIgnoreLabels = flagutil.NewArrayString("remoteWrite.shardByURL.ignoreLabels", "Optional list of labels, which must be ignored when sharding outgoing samples "+
"among remote storage systems if -remoteWrite.shardByURL command-line flag is set. By default all the labels are used for sharding in order to gain "+
"even distribution of series over the specified -remoteWrite.url systems. See also -remoteWrite.shardByURL.labels")
tmpDataPath = flag.String("remoteWrite.tmpDataPath", "vmagent-remotewrite-data", "Path to directory for storing pending data, which isn't sent to the configured -remoteWrite.url . "+
"See also -remoteWrite.maxDiskUsagePerURL and -remoteWrite.disableOnDiskQueue")
keepDanglingQueues = flag.Bool("remoteWrite.keepDanglingQueues", false, "Keep persistent queues contents at -remoteWrite.tmpDataPath in case there are no matching -remoteWrite.url. "+
@@ -80,30 +81,28 @@ var (
`For example, if m{k1="v1",k2="v2"} may be sent as m{k2="v2",k1="v1"}`+
`Enabled sorting for labels can slow down ingestion performance a bit`)
maxHourlySeries = flag.Int("remoteWrite.maxHourlySeries", 0, "The maximum number of unique series vmagent can send to remote storage systems during the last hour. "+
"Excess series are logged and dropped. This can be useful for limiting series cardinality. See https://docs.victoriametrics.com/victoriametrics/vmagent/#cardinality-limiter")
"Excess series are logged and dropped. This can be useful for limiting series cardinality. See https://docs.victoriametrics.com/vmagent/#cardinality-limiter")
maxDailySeries = flag.Int("remoteWrite.maxDailySeries", 0, "The maximum number of unique series vmagent can send to remote storage systems during the last 24 hours. "+
"Excess series are logged and dropped. This can be useful for limiting series churn rate. See https://docs.victoriametrics.com/victoriametrics/vmagent/#cardinality-limiter")
"Excess series are logged and dropped. This can be useful for limiting series churn rate. See https://docs.victoriametrics.com/vmagent/#cardinality-limiter")
maxIngestionRate = flag.Int("maxIngestionRate", 0, "The maximum number of samples vmagent can receive per second. Data ingestion is paused when the limit is exceeded. "+
"By default there are no limits on samples ingestion rate. See also -remoteWrite.rateLimit")
disableOnDiskQueue = flagutil.NewArrayBool("remoteWrite.disableOnDiskQueue", "Whether to disable storing pending data to -remoteWrite.tmpDataPath "+
"when the remote storage system at the corresponding -remoteWrite.url cannot keep up with the data ingestion rate. "+
"See https://docs.victoriametrics.com/victoriametrics/vmagent/#disabling-on-disk-persistence . See also -remoteWrite.dropSamplesOnOverload")
"See https://docs.victoriametrics.com/vmagent#disabling-on-disk-persistence . See also -remoteWrite.dropSamplesOnOverload")
dropSamplesOnOverload = flag.Bool("remoteWrite.dropSamplesOnOverload", false, "Whether to drop samples when -remoteWrite.disableOnDiskQueue is set and if the samples "+
"cannot be pushed into the configured -remoteWrite.url systems in a timely manner. See https://docs.victoriametrics.com/victoriametrics/vmagent/#disabling-on-disk-persistence")
"cannot be pushed into the configured -remoteWrite.url systems in a timely manner. See https://docs.victoriametrics.com/vmagent#disabling-on-disk-persistence")
)
var (
// rwctxsGlobal contains statically populated entries when -remoteWrite.url is specified.
rwctxsGlobal []*remoteWriteCtx
rwctxsGlobalIdx []int
rwctxConsistentHashGlobal *consistenthash.ConsistentHash
rwctxsGlobal []*remoteWriteCtx
// ErrQueueFullHTTPRetry must be returned when TryPush() returns false.
ErrQueueFullHTTPRetry = &httpserver.ErrorWithStatusCode{
Err: fmt.Errorf("remote storage systems cannot keep up with the data ingestion rate; retry the request later " +
"or remove -remoteWrite.disableOnDiskQueue from vmagent command-line flags, so it could save pending data to -remoteWrite.tmpDataPath; " +
"see https://docs.victoriametrics.com/victoriametrics/vmagent/#disabling-on-disk-persistence"),
"see https://docs.victoriametrics.com/vmagent/#disabling-on-disk-persistence"),
StatusCode: http.StatusTooManyRequests,
}
@@ -184,7 +183,7 @@ func Init() {
if len(*shardByURLLabels) > 0 && len(*shardByURLIgnoreLabels) > 0 {
logger.Fatalf("-remoteWrite.shardByURL.labels and -remoteWrite.shardByURL.ignoreLabels cannot be set simultaneously; " +
"see https://docs.victoriametrics.com/victoriametrics/vmagent/#sharding-among-remote-storages")
"see https://docs.victoriametrics.com/vmagent/#sharding-among-remote-storages")
}
shardByURLLabelsMap = newMapFromStrings(*shardByURLLabels)
shardByURLIgnoreLabelsMap = newMapFromStrings(*shardByURLIgnoreLabels)
@@ -196,11 +195,17 @@ func Init() {
// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1240
sighupCh := procutil.NewSighupChan()
initRelabelConfigs()
rcs, err := loadRelabelConfigs()
if err != nil {
logger.Fatalf("cannot load relabel configs: %s", err)
}
allRelabelConfigs.Store(rcs)
relabelConfigSuccess.Set(1)
relabelConfigTimestamp.Set(fasttime.UnixTimestamp())
initStreamAggrConfigGlobal()
initRemoteWriteCtxs(*remoteWriteURLs)
rwctxsGlobal = newRemoteWriteCtxs(*remoteWriteURLs)
disableOnDiskQueues := []bool(*disableOnDiskQueue)
disableOnDiskQueueAny = slices.Contains(disableOnDiskQueues, true)
@@ -262,7 +267,30 @@ func dropDanglingQueues() {
}
}
func initRemoteWriteCtxs(urls []string) {
func reloadRelabelConfigs() {
relabelConfigReloads.Inc()
logger.Infof("reloading relabel configs pointed by -remoteWrite.relabelConfig and -remoteWrite.urlRelabelConfig")
rcs, err := loadRelabelConfigs()
if err != nil {
relabelConfigReloadErrors.Inc()
relabelConfigSuccess.Set(0)
logger.Errorf("cannot reload relabel configs; preserving the previous configs; error: %s", err)
return
}
allRelabelConfigs.Store(rcs)
relabelConfigSuccess.Set(1)
relabelConfigTimestamp.Set(fasttime.UnixTimestamp())
logger.Infof("successfully reloaded relabel configs")
}
var (
relabelConfigReloads = metrics.NewCounter(`vmagent_relabel_config_reloads_total`)
relabelConfigReloadErrors = metrics.NewCounter(`vmagent_relabel_config_reloads_errors_total`)
relabelConfigSuccess = metrics.NewGauge(`vmagent_relabel_config_last_reload_successful`, nil)
relabelConfigTimestamp = metrics.NewCounter(`vmagent_relabel_config_last_reload_success_timestamp_seconds`)
)
func newRemoteWriteCtxs(urls []string) []*remoteWriteCtx {
if len(urls) == 0 {
logger.Panicf("BUG: urls must be non-empty")
}
@@ -278,7 +306,6 @@ func initRemoteWriteCtxs(urls []string) {
maxInmemoryBlocks = 2
}
rwctxs := make([]*remoteWriteCtx, len(urls))
rwctxIdx := make([]int, len(urls))
for i, remoteWriteURLRaw := range urls {
remoteWriteURL, err := url.Parse(remoteWriteURLRaw)
if err != nil {
@@ -289,19 +316,8 @@ func initRemoteWriteCtxs(urls []string) {
sanitizedURL = fmt.Sprintf("%d:%s", i+1, remoteWriteURL)
}
rwctxs[i] = newRemoteWriteCtx(i, remoteWriteURL, maxInmemoryBlocks, sanitizedURL)
rwctxIdx[i] = i
}
if *shardByURL {
consistentHashNodes := make([]string, 0, len(urls))
for i, url := range urls {
consistentHashNodes = append(consistentHashNodes, fmt.Sprintf("%d:%s", i+1, url))
}
rwctxConsistentHashGlobal = consistenthash.NewConsistentHash(consistentHashNodes, 0)
}
rwctxsGlobal = rwctxs
rwctxsGlobalIdx = rwctxIdx
return rwctxs
}
var (
@@ -485,10 +501,6 @@ func tryPush(at *auth.Token, wr *prompbmarshal.WriteRequest, forceDropSamplesOnF
return true
}
// getEligibleRemoteWriteCtxs checks whether writes to configured remote storage systems are blocked and
// returns only the unblocked rwctx.
//
// calculateHealthyRwctxIdx will rely on the order of rwctx to be in ascending order.
func getEligibleRemoteWriteCtxs(tss []prompbmarshal.TimeSeries, forceDropSamplesOnFailure bool) ([]*remoteWriteCtx, bool) {
if !disableOnDiskQueueAny {
return rwctxsGlobal, true
@@ -505,12 +517,6 @@ func getEligibleRemoteWriteCtxs(tss []prompbmarshal.TimeSeries, forceDropSamples
return nil, false
}
rowsCount := getRowsCount(tss)
if *shardByURL {
// Todo: When shardByURL is enabled, the following metrics won't be 100% accurate. Because vmagent don't know
// which rwctx should data be pushed to yet. Let's consider the hashing algorithm fair and will distribute
// data to all rwctxs evenly.
rowsCount = rowsCount / len(rwctxsGlobal)
}
rwctx.rowsDroppedOnPushFailure.Add(rowsCount)
}
}
@@ -522,7 +528,6 @@ func pushToRemoteStoragesTrackDropped(tss []prompbmarshal.TimeSeries) {
if len(rwctxs) == 0 {
return
}
if !tryPushBlockToRemoteStorages(rwctxs, tss, true) {
logger.Panicf("BUG: tryPushBlockToRemoteStorages() must return true when forceDropSamplesOnFailure=true")
}
@@ -573,68 +578,7 @@ func tryShardingBlockAmongRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []pr
defer putTSSShards(x)
shards := x.shards
shardAmountRemoteWriteCtx(tssBlock, shards, rwctxs, replicas)
// Push sharded samples to remote storage systems in parallel in order to reduce
// the time needed for sending the data to multiple remote storage systems.
var wg sync.WaitGroup
var anyPushFailed atomic.Bool
for i, rwctx := range rwctxs {
shard := shards[i]
if len(shard) == 0 {
continue
}
wg.Add(1)
go func(rwctx *remoteWriteCtx, tss []prompbmarshal.TimeSeries) {
defer wg.Done()
if !rwctx.TryPush(tss, forceDropSamplesOnFailure) {
anyPushFailed.Store(true)
}
}(rwctx, shard)
}
wg.Wait()
return !anyPushFailed.Load()
}
// calculateHealthyRwctxIdx returns the index of healthyRwctxs in rwctxsGlobal.
// It relies on the order of rwctx in healthyRwctxs, which is appended by getEligibleRemoteWriteCtxs.
func calculateHealthyRwctxIdx(healthyRwctxs []*remoteWriteCtx) ([]int, []int) {
// fast path: all rwctxs are healthy.
if len(healthyRwctxs) == len(rwctxsGlobal) {
return rwctxsGlobalIdx, nil
}
unhealthyIdx := make([]int, 0, len(rwctxsGlobal))
healthyIdx := make([]int, 0, len(rwctxsGlobal))
var i int
for j := range rwctxsGlobal {
if i < len(healthyRwctxs) && rwctxsGlobal[j].idx == healthyRwctxs[i].idx {
healthyIdx = append(healthyIdx, j)
i++
} else {
unhealthyIdx = append(unhealthyIdx, j)
}
}
return healthyIdx, unhealthyIdx
}
// shardAmountRemoteWriteCtx distribute time series to shards by consistent hashing.
func shardAmountRemoteWriteCtx(tssBlock []prompbmarshal.TimeSeries, shards [][]prompbmarshal.TimeSeries, rwctxs []*remoteWriteCtx, replicas int) {
tmpLabels := promutil.GetLabels()
defer promutil.PutLabels(tmpLabels)
healthyIdx, unhealthyIdx := calculateHealthyRwctxIdx(rwctxs)
// shardsIdxMap is a map to find which the shard idx by rwctxs idx.
// rwctxConsistentHashGlobal will tell which the rwctxs idx a time series should be written to.
// And this time series should be appended to the shards by correct shard idx.
shardsIdxMap := make(map[int]int, len(healthyIdx))
for idx, rwctxsIdx := range healthyIdx {
shardsIdxMap[rwctxsIdx] = idx
}
for _, ts := range tssBlock {
hashLabels := ts.Labels
if len(shardByURLLabelsMap) > 0 {
@@ -655,25 +599,41 @@ func shardAmountRemoteWriteCtx(tssBlock []prompbmarshal.TimeSeries, shards [][]p
tmpLabels.Labels = hashLabels
}
h := getLabelsHash(hashLabels)
// Get the rwctxIdx through consistent hashing and then map it to the index in shards.
// The rwctxIdx is not always equal to the shardIdx, for example, when some rwctx are not available.
rwctxIdx := rwctxConsistentHashGlobal.GetNodeIdx(h, unhealthyIdx)
shardIdx := shardsIdxMap[rwctxIdx]
replicated := 0
idx := h % uint64(len(shards))
i := 0
for {
shards[shardIdx] = append(shards[shardIdx], ts)
replicated++
if replicated >= replicas {
shards[idx] = append(shards[idx], ts)
i++
if i >= replicas {
break
}
shardIdx++
if shardIdx >= len(shards) {
shardIdx = 0
idx++
if idx >= uint64(len(shards)) {
idx = 0
}
}
}
promutil.PutLabels(tmpLabels)
// Push sharded samples to remote storage systems in parallel in order to reduce
// the time needed for sending the data to multiple remote storage systems.
var wg sync.WaitGroup
var anyPushFailed atomic.Bool
for i, rwctx := range rwctxs {
shard := shards[i]
if len(shard) == 0 {
continue
}
wg.Add(1)
go func(rwctx *remoteWriteCtx, tss []prompbmarshal.TimeSeries) {
defer wg.Done()
if !rwctx.TryPush(tss, forceDropSamplesOnFailure) {
anyPushFailed.Store(true)
}
}(rwctx, shard)
}
wg.Wait()
return !anyPushFailed.Load()
}
type tssShards struct {
@@ -847,7 +807,7 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks in
}
pss := make([]*pendingSeries, pssLen)
for i := range pss {
pss[i] = newPendingSeries(fq, &c.useVMProto, sf, rd)
pss[i] = newPendingSeries(fq, c.useVMProto, sf, rd)
}
rwctx := &remoteWriteCtx{

View File

@@ -4,17 +4,13 @@ import (
"fmt"
"math"
"reflect"
"strconv"
"sync/atomic"
"testing"
"time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/consistenthash"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
"github.com/VictoriaMetrics/metrics"
)
@@ -72,9 +68,7 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
allRelabelConfigs.Store(rcs)
pss := make([]*pendingSeries, 1)
isVMProto := &atomic.Bool{}
isVMProto.Store(true)
pss[0] = newPendingSeries(nil, isVMProto, 0, 100)
pss[0] = newPendingSeries(nil, true, 0, 100)
rwctx := &remoteWriteCtx{
idx: 0,
streamAggrKeepInput: keepInput,
@@ -177,173 +171,3 @@ metric{env="dev"} 15
metric{env="bar"} 25
`)
}
func TestShardAmountRemoteWriteCtx(t *testing.T) {
// 1. distribute 100000 series to n nodes.
// 2. remove the last node from healthy list.
// 3. distribute the same 10000 series to (n-1) node again.
// 4. check active time series change rate:
// change rate must < (3/total nodes). e.g. +30% if 10 you have 10 nodes.
f := func(remoteWriteCount int, healthyIdx []int, replicas int) {
t.Helper()
defer func() {
rwctxsGlobal = nil
rwctxsGlobalIdx = nil
rwctxConsistentHashGlobal = nil
}()
rwctxsGlobal = make([]*remoteWriteCtx, remoteWriteCount)
rwctxsGlobalIdx = make([]int, remoteWriteCount)
rwctxs := make([]*remoteWriteCtx, 0, len(healthyIdx))
for i := range remoteWriteCount {
rwCtx := &remoteWriteCtx{
idx: i,
}
rwctxsGlobalIdx[i] = i
if i >= len(healthyIdx) {
rwctxsGlobal[i] = rwCtx
continue
}
hIdx := healthyIdx[i]
if hIdx != i {
rwctxs = append(rwctxs, &remoteWriteCtx{
idx: hIdx,
})
} else {
rwctxs = append(rwctxs, rwCtx)
}
rwctxsGlobal[i] = rwCtx
}
seriesCount := 100000
// build 1000000 series
tssBlock := make([]prompbmarshal.TimeSeries, 0, seriesCount)
for i := 0; i < seriesCount; i++ {
tssBlock = append(tssBlock, prompbmarshal.TimeSeries{
Labels: []prompbmarshal.Label{
{
Name: "label",
Value: strconv.Itoa(i),
},
},
Samples: []prompbmarshal.Sample{
{
Timestamp: 0,
Value: 0,
},
},
})
}
// build consistent hash for x remote write context
// build active time series set
nodes := make([]string, 0, remoteWriteCount)
activeTimeSeriesByNodes := make([]map[string]struct{}, remoteWriteCount)
for i := 0; i < remoteWriteCount; i++ {
nodes = append(nodes, fmt.Sprintf("node%d", i))
activeTimeSeriesByNodes[i] = make(map[string]struct{})
}
rwctxConsistentHashGlobal = consistenthash.NewConsistentHash(nodes, 0)
// create shards
x := getTSSShards(len(rwctxs))
shards := x.shards
// execute
shardAmountRemoteWriteCtx(tssBlock, shards, rwctxs, replicas)
for i, nodeIdx := range healthyIdx {
for _, ts := range shards[i] {
// add it to node[nodeIdx]'s active time series
activeTimeSeriesByNodes[nodeIdx][prompbmarshal.LabelsToString(ts.Labels)] = struct{}{}
}
}
totalActiveTimeSeries := 0
for _, activeTimeSeries := range activeTimeSeriesByNodes {
totalActiveTimeSeries += len(activeTimeSeries)
}
avgActiveTimeSeries1 := totalActiveTimeSeries / remoteWriteCount
putTSSShards(x)
// removed last node
rwctxs = rwctxs[:len(rwctxs)-1]
healthyIdx = healthyIdx[:len(healthyIdx)-1]
x = getTSSShards(len(rwctxs))
shards = x.shards
// execute
shardAmountRemoteWriteCtx(tssBlock, shards, rwctxs, replicas)
for i, nodeIdx := range healthyIdx {
for _, ts := range shards[i] {
// add it to node[nodeIdx]'s active time series
activeTimeSeriesByNodes[nodeIdx][prompbmarshal.LabelsToString(ts.Labels)] = struct{}{}
}
}
totalActiveTimeSeries = 0
for _, activeTimeSeries := range activeTimeSeriesByNodes {
totalActiveTimeSeries += len(activeTimeSeries)
}
avgActiveTimeSeries2 := totalActiveTimeSeries / remoteWriteCount
changed := math.Abs(float64(avgActiveTimeSeries2-avgActiveTimeSeries1) / float64(avgActiveTimeSeries1))
threshold := 3 / float64(remoteWriteCount)
if changed >= threshold {
t.Fatalf("average active time series before: %d, after: %d, changed: %.2f. threshold: %.2f", avgActiveTimeSeries1, avgActiveTimeSeries2, changed, threshold)
}
}
f(5, []int{0, 1, 2, 3, 4}, 1)
f(5, []int{0, 1, 2, 3, 4}, 2)
f(10, []int{0, 1, 2, 3, 4, 5, 6, 7, 9}, 1)
f(10, []int{0, 1, 2, 3, 4, 5, 6, 7, 9}, 3)
}
func TestCalculateHealthyRwctxIdx(t *testing.T) {
f := func(total int, healthyIdx []int, unhealthyIdx []int) {
t.Helper()
healthyMap := make(map[int]bool)
for _, idx := range healthyIdx {
healthyMap[idx] = true
}
rwctxsGlobal = make([]*remoteWriteCtx, total)
rwctxsGlobalIdx = make([]int, total)
rwctxs := make([]*remoteWriteCtx, 0, len(healthyIdx))
for i := range rwctxsGlobal {
rwctx := &remoteWriteCtx{idx: i}
rwctxsGlobal[i] = rwctx
if healthyMap[i] {
rwctxs = append(rwctxs, rwctx)
}
rwctxsGlobalIdx[i] = i
}
gotHealthyIdx, gotUnhealthyIdx := calculateHealthyRwctxIdx(rwctxs)
if !reflect.DeepEqual(healthyIdx, gotHealthyIdx) {
t.Errorf("calculateHealthyRwctxIdx want healthyIdx = %v, got %v", healthyIdx, gotHealthyIdx)
}
if !reflect.DeepEqual(unhealthyIdx, gotUnhealthyIdx) {
t.Errorf("calculateHealthyRwctxIdx want unhealthyIdx = %v, got %v", unhealthyIdx, gotUnhealthyIdx)
}
}
f(5, []int{0, 1, 2, 3, 4}, nil)
f(5, []int{0, 1, 2, 4}, []int{3})
f(5, []int{2, 4}, []int{0, 1, 3})
f(5, []int{0, 2, 4}, []int{1, 3})
f(5, []int{}, []int{0, 1, 2, 3, 4})
f(5, []int{4}, []int{0, 1, 2, 3})
f(1, []int{0}, nil)
f(1, []int{}, []int{0})
}

View File

@@ -16,55 +16,55 @@ import (
var (
// Global config
streamAggrGlobalConfig = flag.String("streamAggr.config", "", "Optional path to file with stream aggregation config. "+
"See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/ . "+
"See https://docs.victoriametrics.com/stream-aggregation/ . "+
"See also -streamAggr.keepInput, -streamAggr.dropInput and -streamAggr.dedupInterval")
streamAggrGlobalKeepInput = flag.Bool("streamAggr.keepInput", false, "Whether to keep all the input samples after the aggregation "+
"with -streamAggr.config. By default, only aggregates samples are dropped, while the remaining samples "+
"are written to remote storages write. See also -streamAggr.dropInput and https://docs.victoriametrics.com/victoriametrics/stream-aggregation/")
"are written to remote storages write. See also -streamAggr.dropInput and https://docs.victoriametrics.com/stream-aggregation/")
streamAggrGlobalDropInput = flag.Bool("streamAggr.dropInput", false, "Whether to drop all the input samples after the aggregation "+
"with -remoteWrite.streamAggr.config. By default, only aggregates samples are dropped, while the remaining samples "+
"are written to remote storages write. See also -streamAggr.keepInput and https://docs.victoriametrics.com/victoriametrics/stream-aggregation/")
"are written to remote storages write. See also -streamAggr.keepInput and https://docs.victoriametrics.com/stream-aggregation/")
streamAggrGlobalDedupInterval = flag.Duration("streamAggr.dedupInterval", 0, "Input samples are de-duplicated with this interval on "+
"aggregator before optional aggregation with -streamAggr.config . "+
"See also -dedup.minScrapeInterval and https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#deduplication")
"See also -dedup.minScrapeInterval and https://docs.victoriametrics.com/stream-aggregation/#deduplication")
streamAggrGlobalIgnoreOldSamples = flag.Bool("streamAggr.ignoreOldSamples", false, "Whether to ignore input samples with old timestamps outside the "+
"current aggregation interval for aggregator. "+
"See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#ignoring-old-samples")
"See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples")
streamAggrGlobalIgnoreFirstIntervals = flag.Int("streamAggr.ignoreFirstIntervals", 0, "Number of aggregation intervals to skip after the start for "+
"aggregator. Increase this value if you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving unordered delayed data from "+
"clients pushing data into the vmagent. See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#ignore-aggregation-intervals-on-start")
"clients pushing data into the vmagent. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
streamAggrGlobalDropInputLabels = flagutil.NewArrayString("streamAggr.dropInputLabels", "An optional list of labels to drop from samples for aggregator "+
"before stream de-duplication and aggregation . See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#dropping-unneeded-labels")
"before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels")
streamAggrGlobalEnableWindows = flag.Bool("streamAggr.enableWindows", false, "Enables aggregation within fixed windows for all global aggregators. "+
"This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. "+
"See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#aggregation-windows.")
"See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.")
// Per URL config
streamAggrConfig = flagutil.NewArrayString("remoteWrite.streamAggr.config", "Optional path to file with stream aggregation config for the corresponding -remoteWrite.url. "+
"See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/ . "+
"See https://docs.victoriametrics.com/stream-aggregation/ . "+
"See also -remoteWrite.streamAggr.keepInput, -remoteWrite.streamAggr.dropInput and -remoteWrite.streamAggr.dedupInterval")
streamAggrDropInput = flagutil.NewArrayBool("remoteWrite.streamAggr.dropInput", "Whether to drop all the input samples after the aggregation "+
"with -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. By default, only aggregates samples are dropped, while the remaining samples "+
"are written to the corresponding -remoteWrite.url . See also -remoteWrite.streamAggr.keepInput and https://docs.victoriametrics.com/victoriametrics/stream-aggregation/")
"are written to the corresponding -remoteWrite.url . See also -remoteWrite.streamAggr.keepInput and https://docs.victoriametrics.com/stream-aggregation/")
streamAggrKeepInput = flagutil.NewArrayBool("remoteWrite.streamAggr.keepInput", "Whether to keep all the input samples after the aggregation "+
"with -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. By default, only aggregates samples are dropped, while the remaining samples "+
"are written to the corresponding -remoteWrite.url . See also -remoteWrite.streamAggr.dropInput and https://docs.victoriametrics.com/victoriametrics/stream-aggregation/")
"are written to the corresponding -remoteWrite.url . See also -remoteWrite.streamAggr.dropInput and https://docs.victoriametrics.com/stream-aggregation/")
streamAggrDedupInterval = flagutil.NewArrayDuration("remoteWrite.streamAggr.dedupInterval", 0, "Input samples are de-duplicated with this interval before optional aggregation "+
"with -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. See also -dedup.minScrapeInterval and https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#deduplication")
"with -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. See also -dedup.minScrapeInterval and https://docs.victoriametrics.com/stream-aggregation/#deduplication")
streamAggrIgnoreOldSamples = flagutil.NewArrayBool("remoteWrite.streamAggr.ignoreOldSamples", "Whether to ignore input samples with old timestamps outside the current "+
"aggregation interval for the corresponding -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. "+
"See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#ignoring-old-samples")
"See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples")
streamAggrIgnoreFirstIntervals = flagutil.NewArrayInt("remoteWrite.streamAggr.ignoreFirstIntervals", 0, "Number of aggregation intervals to skip after the start "+
"for the corresponding -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. Increase this value if "+
"you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving buffered delayed data from clients pushing data into the vmagent. "+
"See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#ignore-aggregation-intervals-on-start")
"See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
streamAggrDropInputLabels = flagutil.NewArrayString("remoteWrite.streamAggr.dropInputLabels", "An optional list of labels to drop from samples "+
"before stream de-duplication and aggregation with -remoteWrite.streamAggr.config and -remoteWrite.streamAggr.dedupInterval at the corresponding -remoteWrite.url. "+
"Multiple labels per remoteWrite.url must be delimited by '^^': -remoteWrite.streamAggr.dropInputLabels='replica^^az,replica'. "+
"See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#dropping-unneeded-labels")
"See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels")
streamAggrEnableWindows = flagutil.NewArrayBool("remoteWrite.streamAggr.enableWindows", "Enables aggregation within fixed windows for all remote write's aggregators. "+
"This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. "+
"See https://docs.victoriametrics.com/victoriametrics/stream-aggregation/#aggregation-windows.")
"See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.")
)
// CheckStreamAggrConfigs checks -remoteWrite.streamAggr.config and -streamAggr.config.

View File

@@ -1,3 +1,3 @@
See vmalert-tool docs [here](https://docs.victoriametrics.com/victoriametrics/vmalert-tool/).
See vmalert-tool docs [here](https://docs.victoriametrics.com/vmalert-tool.html).
vmalert-tool docs can be edited at [docs/vmalert-tool.md](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/victoriametrics/vmalert-tool.md).
vmalert-tool docs can be edited at [docs/vmalert-tool.md](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/vmalert-tool.md).

View File

@@ -1,8 +0,0 @@
ARG base_image=non-existing
FROM $base_image
EXPOSE 8880
ENTRYPOINT ["/vmalert-tool-prod"]
ARG src_binary=non-existing
COPY $src_binary ./vmalert-tool-prod

View File

@@ -17,13 +17,13 @@ func main() {
app := &cli.App{
Name: "vmalert-tool",
Usage: "VMAlert command-line tool",
UsageText: "More info in https://docs.victoriametrics.com/victoriametrics/vmalert-tool/",
UsageText: "More info in https://docs.victoriametrics.com/vmalert-tool.html",
Version: buildinfo.Version,
Commands: []*cli.Command{
{
Name: "unittest",
Usage: "Run unittest for alerting and recording rules.",
UsageText: "More info in https://docs.victoriametrics.com/victoriametrics/vmalert-tool/#unit-testing-for-rules",
UsageText: "More info in https://docs.victoriametrics.com/vmalert-tool.html#Unit-testing-for-rules",
Flags: []cli.Flag{
&cli.StringSliceFlag{
Name: "files",

View File

@@ -9,5 +9,4 @@ COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certifica
EXPOSE 8429
ENTRYPOINT ["/vmalert-tool-prod"]
ARG TARGETARCH
ARG BINARY_SUFFIX=non-existing
COPY vmalert-tool-linux-${TARGETARCH}-prod${BINARY_SUFFIX} ./vmalert-tool-prod
COPY vmalert-tool-linux-${TARGETARCH}-prod ./vmalert-tool-prod

View File

@@ -18,7 +18,7 @@ import (
"github.com/VictoriaMetrics/metricsql"
)
var numReg = regexp.MustCompile(`(?i)[+x-]?(?:\d+(?:\.\d*)?|\.\d+|inf|nan|_)(?:e[+-]?\d+)?[+x-]?`)
var numReg = regexp.MustCompile(`\D?\d*\.?\d*\D?`)
// series holds input_series defined in the test file
type series struct {
@@ -94,18 +94,12 @@ func parseInputSeries(input []series, interval *promutil.Duration, startStamp ti
// parseInputValue support input like "1", "1+1x1 _ -4 3+20x1", see more examples in test.
func parseInputValue(input string, origin bool) ([]sequenceValue, error) {
var res []sequenceValue
items := strings.Fields(input)
if len(items) == 0 {
return nil, fmt.Errorf("values cannot be an empty string")
}
items := strings.Split(input, " ")
for _, item := range items {
if item == "stale" {
res = append(res, sequenceValue{Value: decimal.StaleNaN})
continue
}
if strings.Contains(item, "stale") {
return nil, fmt.Errorf("stale metric doesn't support operations")
}
vals := numReg.FindAllString(item, -1)
switch len(vals) {
case 1:

View File

@@ -1,7 +1,6 @@
package unittest
import (
"math"
"testing"
"time"
@@ -13,20 +12,17 @@ func TestParseInputValue_Failure(t *testing.T) {
f := func(input string) {
t.Helper()
val, err := parseInputValue(input, true)
_, err := parseInputValue(input, true)
if err == nil {
t.Fatalf("expecting non-nil error, got value %+v", val)
t.Fatalf("expecting non-nil error")
}
}
f("")
f("x4")
f("testfailed")
// stale doesn't support operations
f("stalex3")
f("1+stalex3")
}
func TestParseInputValue_Success(t *testing.T) {
@@ -46,8 +42,7 @@ func TestParseInputValue_Success(t *testing.T) {
t.Fatalf("unexpected Omitted field in the output\ngot\n%v\nwant\n%v", output, outputExpected)
}
if outputExpected[i].Value != output[i].Value {
if (math.IsNaN(outputExpected[i].Value) && math.IsNaN(output[i].Value)) ||
(decimal.IsStaleNaN(outputExpected[i].Value) && decimal.IsStaleNaN(output[i].Value)) {
if decimal.IsStaleNaN(outputExpected[i].Value) && decimal.IsStaleNaN(output[i].Value) {
continue
}
t.Fatalf("unexpected Value field in the output\ngot\n%v\nwant\n%v", output, outputExpected)
@@ -61,8 +56,6 @@ func TestParseInputValue_Success(t *testing.T) {
f("stale", []sequenceValue{{Value: decimal.StaleNaN}})
f(" stale 2", []sequenceValue{{Value: decimal.StaleNaN}, {Value: 2}})
f("-4x1", []sequenceValue{{Value: -4}, {Value: -4}})
f("_x1", []sequenceValue{{Omitted: true}})
@@ -72,14 +65,6 @@ func TestParseInputValue_Success(t *testing.T) {
f("2-1x4", []sequenceValue{{Value: 2}, {Value: 1}, {Value: 0}, {Value: -1}, {Value: -2}})
f("1+1x1 _ -4 stale 3+20x1", []sequenceValue{{Value: 1}, {Value: 2}, {Omitted: true}, {Value: -4}, {Value: decimal.StaleNaN}, {Value: 3}, {Value: 23}})
f("Inf +Inf -Inf", []sequenceValue{{Value: math.Inf(1)}, {Value: math.Inf(1)}, {Value: math.Inf(-1)}})
f("Nan Infx2", []sequenceValue{{Value: math.NaN()}, {Value: math.Inf(1)}, {Value: math.Inf(1)}, {Value: math.Inf(1)}})
f("NaNx2", []sequenceValue{{Value: math.NaN()}, {Value: math.NaN()}, {Value: math.NaN()}})
f("4e9+1x2 -3e-2x2 1e-2-1x1", []sequenceValue{{Value: 4000000000}, {Value: 4000000001}, {Value: 4000000002}, {Value: -0.03}, {Value: -0.03}, {Value: -0.03}, {Value: 0.01}, {Value: -0.99}})
}
func TestParseInputSeries_Success(t *testing.T) {

View File

@@ -13,6 +13,7 @@ import (
"path/filepath"
"reflect"
"sort"
"strconv"
"strings"
"syscall"
"time"
@@ -98,11 +99,8 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
}()
}
tmpFolder, err := os.MkdirTemp(os.TempDir(), testStoragePath)
if err != nil {
logger.Fatalf("failed to create tmp dir for tests: %v", err)
}
storagePath = tmpFolder
// adding time.Now().UnixNano() to avoid possible file conflict when multiple processes run on a single host
storagePath = filepath.Join(os.TempDir(), testStoragePath, strconv.FormatInt(time.Now().UnixNano(), 10))
processFlags()
vminsert.Init()
vmselect.Init()

View File

@@ -74,7 +74,7 @@ test-vmalert:
go test -v -race -cover ./app/vmalert/notifier
go test -v -race -cover ./app/vmalert/config
go test -v -race -cover ./app/vmalert/remotewrite
go test -v -race -cover ./app/vmalert/vmalertutil
go test -v -race -cover ./app/vmalert/utils
run-vmalert: vmalert
./bin/vmalert -rule=app/vmalert/config/testdata/rules/rules2-good.rules \

Some files were not shown because too many files have changed in this diff Show More