mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2026-08-23 11:49:19 +03:00
Compare commits
11 Commits
export-tra
...
issue-1132
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b2d459e854 | ||
|
|
420f18b219 | ||
|
|
824ce428c9 | ||
|
|
fadd701496 | ||
|
|
5ee7223f56 | ||
|
|
364b3b6823 | ||
|
|
400f01a114 | ||
|
|
00abfbf043 | ||
|
|
2f2c6bcdec | ||
|
|
636a50bb9c | ||
|
|
166cfe4597 |
@@ -1,6 +1,6 @@
|
||||
# VictoriaMetrics
|
||||
|
||||
[](https://github.com/VictoriaMetrics/VictoriaMetrics/releases)
|
||||
[](https://github.com/VictoriaMetrics/VictoriaMetrics/releases)
|
||||
[](https://hub.docker.com/u/victoriametrics)
|
||||
[](https://github.com/VictoriaMetrics/VictoriaMetrics/actions/workflows/build.yml)
|
||||
[](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/LICENSE)
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/appmetrics"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
|
||||
@@ -34,9 +35,10 @@ var (
|
||||
"This can be changed with -promscrape.config.strictParse=false command-line flag")
|
||||
maxIngestionRate = flag.Int("maxIngestionRate", 0, "The maximum number of samples vmsingle can receive per second. Data ingestion is paused when the limit is exceeded. "+
|
||||
"By default there are no limits on samples ingestion rate.")
|
||||
vmselectMaxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
|
||||
"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
|
||||
"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
|
||||
vmselectMaxConcurrentRequests = flagutil.NewIntWithDynamicDefault("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "vmselect.getDefaultMaxConcurrentRequests()",
|
||||
"The maximum number of concurrent search requests. "+
|
||||
"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
|
||||
"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
|
||||
vmselectMaxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests "+
|
||||
"limit is reached; see also -search.maxQueryDuration")
|
||||
)
|
||||
@@ -90,7 +92,9 @@ func main() {
|
||||
}
|
||||
logger.Infof("starting VictoriaMetrics at %q...", listenAddrs)
|
||||
startTime := time.Now()
|
||||
|
||||
vmstorage.Init(*vmselectMaxConcurrentRequests, *vmselectMaxQueueDuration, promql.ResetRollupResultCacheIfNeeded)
|
||||
appmetrics.MustCreateUncleanShutdownMarker(vmstorage.DataPath())
|
||||
vmselect.Init(*vmselectMaxConcurrentRequests, *vmselectMaxQueueDuration)
|
||||
vminsertcommon.StartIngestionRateLimiter(*maxIngestionRate)
|
||||
vminsert.Init()
|
||||
@@ -120,6 +124,7 @@ func main() {
|
||||
|
||||
vmstorage.Stop()
|
||||
vmselect.Stop()
|
||||
appmetrics.MustRemoveUncleanShutdownMarker(vmstorage.DataPath())
|
||||
|
||||
logger.Infof("the VictoriaMetrics has been stopped in %.3f seconds", time.Since(startTime).Seconds())
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
"github.com/VictoriaMetrics/metrics"
|
||||
"github.com/cespare/xxhash/v2"
|
||||
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/appmetrics"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bloomfilter"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
|
||||
@@ -62,9 +63,10 @@ var (
|
||||
"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. "+
|
||||
"Useful when -remoteWrite.url is changed temporarily and persistent queue files will be needed later on.")
|
||||
queues = flagutil.NewArrayInt("remoteWrite.queues", cgroup.AvailableCPUs()*2, "The number of concurrent queues to each -remoteWrite.url. Set more queues if default number of queues "+
|
||||
"isn't enough for sending high volume of collected data to remote storage. "+
|
||||
"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage")
|
||||
queues = flagutil.NewArrayIntWithDynamicDefault("remoteWrite.queues", cgroup.AvailableCPUs()*2, "2*cgroup.AvailableCPUs()",
|
||||
"The number of concurrent queues to each -remoteWrite.url. Set more queues if default number of queues "+
|
||||
"isn't enough for sending high volume of collected data to remote storage. "+
|
||||
"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage")
|
||||
inmemoryQueues = flagutil.NewArrayInt("remoteWrite.inmemoryQueues", 0, "The number of additional workers per each -remoteWrite.url, which send only recently ingested data from the in-memory queue, "+
|
||||
"while the file-based queue at -remoteWrite.tmpDataPath is drained by workers configured via -remoteWrite.queues. "+
|
||||
"This reduces delivery lag for fresh samples when the file-based queue contains a backlog accumulated during remote storage outages.")
|
||||
@@ -233,6 +235,7 @@ func Init() {
|
||||
initStreamAggrConfigGlobal()
|
||||
|
||||
initRemoteWriteCtxs(*remoteWriteURLs)
|
||||
appmetrics.MustCreateUncleanShutdownMarker(*tmpDataPath)
|
||||
|
||||
disableOnDiskQueues := []bool(*disableOnDiskQueue)
|
||||
disableOnDiskQueueAny = slices.Contains(disableOnDiskQueues, true)
|
||||
@@ -391,6 +394,8 @@ func Stop() {
|
||||
if sl := dailySeriesLimiter; sl != nil {
|
||||
sl.MustStop()
|
||||
}
|
||||
|
||||
appmetrics.MustRemoveUncleanShutdownMarker(*tmpDataPath)
|
||||
}
|
||||
|
||||
// PushDropSamplesOnFailure pushes wr to the configured remote storage systems set via -remoteWrite.url
|
||||
|
||||
@@ -36,9 +36,10 @@ var (
|
||||
idleConnectionTimeout = flag.Duration("remoteWrite.idleConnTimeout", 50*time.Second, `Defines a duration for idle (keep-alive connections) to exist. Consider settings this value less to the value of "-http.idleConnTimeout". It must prevent possible "write: broken pipe" and "read: connection reset by peer" errors.`)
|
||||
maxIdleConnections = flag.Int("remoteWrite.maxIdleConnections", 100, `Defines the number of idle (keep-alive connections) to -remoteWrite.url for the vmalert-tool debug writer, which sends every series in a separate request. Too low a value may result in a high number of sockets in TIME_WAIT state.`)
|
||||
|
||||
maxQueueSize = flag.Int("remoteWrite.maxQueueSize", defaultMaxQueueSize, "Defines the max number of pending datapoints to remote write endpoint")
|
||||
maxBatchSize = flag.Int("remoteWrite.maxBatchSize", defaultMaxBatchSize, "Defines max number of timeseries to be flushed at once")
|
||||
concurrency = flag.Int("remoteWrite.concurrency", defaultConcurrency, "Defines number of writers for concurrent writing into remote write endpoint. Default value depends on the number of available CPU cores.")
|
||||
maxQueueSize = flag.Int("remoteWrite.maxQueueSize", defaultMaxQueueSize, "Defines the max number of pending datapoints to remote write endpoint")
|
||||
maxBatchSize = flag.Int("remoteWrite.maxBatchSize", defaultMaxBatchSize, "Defines max number of timeseries to be flushed at once")
|
||||
concurrency = flagutil.NewIntWithDynamicDefault("remoteWrite.concurrency", defaultConcurrency, "2*cgroup.AvailableCPUs()",
|
||||
"Defines number of writers for concurrent writing into remote write endpoint. Default value depends on the number of available CPU cores.")
|
||||
flushInterval = flag.Duration("remoteWrite.flushInterval", defaultFlushInterval, "Defines interval of flushes to remote write endpoint")
|
||||
|
||||
tlsInsecureSkipVerify = flag.Bool("remoteWrite.tlsInsecureSkipVerify", false, "Whether to skip tls verification when connecting to -remoteWrite.url")
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage/metricnamestats"
|
||||
@@ -30,11 +31,12 @@ var (
|
||||
maxSamplesPerSeries = flag.Int("search.maxSamplesPerSeries", 30e6, "The maximum number of raw samples a single query can scan per each time series. This option allows limiting memory usage")
|
||||
maxSamplesPerQuery = flag.Int("search.maxSamplesPerQuery", 1e9, "The maximum number of raw samples a single query can process across all time series. "+
|
||||
"This protects from heavy queries, which select unexpectedly high number of raw samples. See also -search.maxSamplesPerSeries")
|
||||
maxWorkersPerQuery = flag.Int("search.maxWorkersPerQuery", defaultMaxWorkersPerQuery, "The maximum number of CPU cores a single query can use. "+
|
||||
"The default value should work good for most cases. "+
|
||||
"The flag can be set to lower values for improving performance of big number of concurrently executed queries. "+
|
||||
"The flag can be set to bigger values for improving performance of heavy queries, which scan big number of time series (>10K) and/or big number of samples (>100M). "+
|
||||
"There is no sense in setting this flag to values bigger than the number of CPU cores available on the system")
|
||||
maxWorkersPerQuery = flagutil.NewIntWithDynamicDefault("search.maxWorkersPerQuery", defaultMaxWorkersPerQuery, "netstorage.defaultMaxWorkersPerQuery()",
|
||||
"The maximum number of CPU cores a single query can use. "+
|
||||
"The default value should work good for most cases. "+
|
||||
"The flag can be set to lower values for improving performance of big number of concurrently executed queries. "+
|
||||
"The flag can be set to bigger values for improving performance of heavy queries, which scan big number of time series (>10K) and/or big number of samples (>100M). "+
|
||||
"There is no sense in setting this flag to values bigger than the number of CPU cores available on the system")
|
||||
)
|
||||
|
||||
// Result is a single timeseries result.
|
||||
|
||||
@@ -16,6 +16,19 @@ groups:
|
||||
Job {{ $labels.job }} (instance {{ $labels.instance }}) has restarted more than twice in the last 15 minutes.
|
||||
It might be crashlooping.
|
||||
|
||||
- alert: UncleanShutdown
|
||||
expr: vm_app_prev_shutdown_unclean == 1 and time() - vm_app_start_timestamp < 600
|
||||
labels:
|
||||
severity: warning
|
||||
annotations:
|
||||
summary: "{{ $labels.job }} on instance {{ $labels.instance }} started after an unclean shutdown"
|
||||
description: |
|
||||
The previous process run didn't shut down cleanly. Check the logs for OOM, SIGKILL,
|
||||
a host failure, or another unexpected termination. In Kubernetes, a pod may be forcefully
|
||||
killed with SIGKILL if the shutdown takes longer than terminationGracePeriodSeconds.
|
||||
This alert stops firing 10 minutes after startup.
|
||||
See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8443 for more details.
|
||||
|
||||
- alert: ServiceDown
|
||||
expr: up{job=~".*(victoriametrics|vmselect|vminsert|vmstorage|vmagent|vmalert|vmsingle|vmalertmanager|vmauth).*"} == 0
|
||||
for: 2m
|
||||
|
||||
@@ -90,12 +90,9 @@ endif
|
||||
sed -i 's/\t/ /g' docs/victoriametrics/victoria_metrics_common_flags.md
|
||||
sed -i 's/\t/ /g' docs/victoriametrics/victoria_metrics_enterprise_flags.md
|
||||
|
||||
# adjust flags with dynamic default values
|
||||
# remove after https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680 implemented
|
||||
sed -i '/The maximum number of concurrent insert requests/ s/(default [0-9]\+)/(default 2*cgroup.AvailableCPUs())/' docs/victoriametrics/victoria_metrics_common_flags.md
|
||||
sed -i '/The maximum number of concurrent search requests\./ s/(default [0-9]\+)/(default vmselect.getDefaultMaxConcurrentRequests())/' docs/victoriametrics/victoria_metrics_common_flags.md
|
||||
sed -i '/The maximum number of CPU cores a single query can use\./ s/(default [0-9]\+)/(default netstorage.defaultMaxWorkersPerQuery())/' docs/victoriametrics/victoria_metrics_common_flags.md
|
||||
sed -i '/The maximum number of concurrent goroutines to work with files;/ s/(default [0-9]\+)/(default fsutil.getDefaultConcurrency())/' docs/victoriametrics/victoria_metrics_common_flags.md
|
||||
# hide the machine-specific value of dynamic defaults, keeping the formula.
|
||||
# the flagutil.New*WithDynamicDefault constructors print them as "(default <value> = <formula>)".
|
||||
sed -i 's/(default [0-9]\+ = \(.*\))$$/(default \1)/' docs/victoriametrics/victoria_metrics_common_flags.md
|
||||
|
||||
docs-update-vmauth-flags:
|
||||
ifndef TAG
|
||||
@@ -119,9 +116,9 @@ endif
|
||||
sed -i 's/\t/ /g' docs/victoriametrics/vmauth_common_flags.md
|
||||
sed -i 's/\t/ /g' docs/victoriametrics/vmauth_enterprise_flags.md
|
||||
|
||||
# adjust flags with dynamic default values
|
||||
# remove after https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680 implemented
|
||||
sed -i '/The maximum number of concurrent goroutines to work with files;/ s/(default [0-9]\+)/(default fsutil.getDefaultConcurrency())/' docs/victoriametrics/vmauth_common_flags.md
|
||||
# hide the machine-specific value of dynamic defaults, keeping the formula.
|
||||
# the flagutil.New*WithDynamicDefault constructors print them as "(default <value> = <formula>)".
|
||||
sed -i 's/(default [0-9]\+ = \(.*\))$$/(default \1)/' docs/victoriametrics/vmauth_common_flags.md
|
||||
|
||||
docs-update-vmagent-flags:
|
||||
ifndef TAG
|
||||
@@ -145,11 +142,9 @@ endif
|
||||
sed -i 's/\t/ /g' docs/victoriametrics/vmagent_common_flags.md
|
||||
sed -i 's/\t/ /g' docs/victoriametrics/vmagent_enterprise_flags.md
|
||||
|
||||
# adjust flags with dynamic default values
|
||||
# remove after https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680 implemented
|
||||
sed -i '/The maximum number of concurrent insert requests/ s/(default [0-9]\+)/(default 2*cgroup.AvailableCPUs())/' docs/victoriametrics/vmagent_common_flags.md
|
||||
sed -i '/The number of concurrent queues to each -remoteWrite.url./ s/(default [0-9]\+)/(default 2*cgroup.AvailableCPUs())/' docs/victoriametrics/vmagent_common_flags.md
|
||||
sed -i '/The maximum number of concurrent goroutines to work with files;/ s/(default [0-9]\+)/(default fsutil.getDefaultConcurrency())/' docs/victoriametrics/vmagent_common_flags.md
|
||||
# hide the machine-specific value of dynamic defaults, keeping the formula.
|
||||
# the flagutil.New*WithDynamicDefault constructors print them as "(default <value> = <formula>)".
|
||||
sed -i 's/(default [0-9]\+ = \(.*\))$$/(default \1)/' docs/victoriametrics/vmagent_common_flags.md
|
||||
|
||||
docs-update-vmalert-flags:
|
||||
ifndef TAG
|
||||
@@ -173,10 +168,9 @@ endif
|
||||
sed -i 's/\t/ /g' docs/victoriametrics/vmalert_common_flags.md
|
||||
sed -i 's/\t/ /g' docs/victoriametrics/vmalert_enterprise_flags.md
|
||||
|
||||
# adjust flags with dynamic default values
|
||||
# remove after https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680 implemented
|
||||
sed -i '/Defines number of writers for concurrent writing into remote write endpoint./ s/(default [0-9]\+)/(default 2*cgroup.AvailableCPUs())/' docs/victoriametrics/vmalert_common_flags.md
|
||||
sed -i '/The maximum number of concurrent goroutines to work with files;/ s/(default [0-9]\+)/(default fsutil.getDefaultConcurrency())/' docs/victoriametrics/vmalert_common_flags.md
|
||||
# hide the machine-specific value of dynamic defaults, keeping the formula.
|
||||
# the flagutil.New*WithDynamicDefault constructors print them as "(default <value> = <formula>)".
|
||||
sed -i 's/(default [0-9]\+ = \(.*\))$$/(default \1)/' docs/victoriametrics/vmalert_common_flags.md
|
||||
|
||||
docs-update-vmselect-flags:
|
||||
ifndef TAG
|
||||
|
||||
@@ -26,7 +26,10 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
|
||||
|
||||
## tip
|
||||
|
||||
* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/), [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/), `vmstorage` and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): expose the `vm_app_prev_shutdown_unclean` gauge. It is set to `1` when the previous process run didn't shut down cleanly. Added the `UncleanShutdown` [alerting rule](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules/alerts-health.yml), which fires for 10 minutes after an unclean shutdown is detected. See [#8443](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8443).
|
||||
* FEATURE: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): show the selected time zone UTC offset next to the date/time controls and allow opening time zone settings from it. See [#11332](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11332).
|
||||
* FEATURE: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/), [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/), [vmalert](https://docs.victoriametrics.com/victoriametrics/vmalert/), and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): show how the default value is calculated for command-line flags which derive it from the number of available CPU cores. For example, `-maxConcurrentInserts` now prints `(default 16 = 2*cgroup.AvailableCPUs())` in `-help` output instead of `(default 16)`. Updated flags: `-search.maxConcurrentRequests`, `-search.maxWorkersPerQuery`, `-fs.maxConcurrency`, `-remoteWrite.concurrency`, `-remoteWrite.queues`. See [#9680](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9680). Thanks to @Vandit1604 for contribution.
|
||||
* FEATURE: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): accept Unix timestamp values with negative scientific notation exponents in [Prometheus querying API](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#prometheus-querying-api-usage) time parameters such as `start` and `end` in `/api/v1/query_range`. Previously, values such as `1000e-1` were rejected. See [#11427](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11427).
|
||||
|
||||
* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fix infinite loop in the OpenTelemetry Firehose ingestion endpoint (`/opentelemetry/api/v1/push`) when receiving a malformed record with an incomplete varint in the `data` field. Previously this caused the goroutine to spin forever, permanently consuming CPU until the process was restarted.
|
||||
* BUGFIX: [vmalert-tool](https://docs.victoriametrics.com/victoriametrics/vmalert-tool/): reuse connections to `-remoteWrite.url` when writing the results of recording rules and alerts. Previously every series was sent over a new connection, which left a lot of sockets in `TIME_WAIT` state and could exhaust the ephemeral port range. The number of idle connections can be tuned via the new `-remoteWrite.maxIdleConnections` command-line flag. Thanks @evkuzin for contribution.
|
||||
|
||||
@@ -50,7 +50,7 @@ If you don't see an option to create a data source - try contacting system admin
|
||||
Create [Prometheus datasource](https://grafana.com/docs/grafana/latest/datasources/prometheus/configure/)
|
||||
in Grafana. Follow the same connection instructions as for [VictoriaMetrics datasource](#VictoriaMetrics-datasource).
|
||||
|
||||
In the "Type and version" section set the type to "Prometheus" and the version to at least "2.24.x".
|
||||
In the "Performance" section set the Prometheus type to "Prometheus" and the Prometheus version to at least "2.24.x".
|
||||
This allows Grafana to use a more efficient API to get label values:
|
||||
|
||||

|
||||
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 7.1 KiB After Width: | Height: | Size: 10 KiB |
@@ -1127,13 +1127,18 @@ Or for all rules within the [group](#groups) {{% available_from "v1.117.0" %}}.
|
||||
Just set `debug: true` in configuration and vmalert will start printing additional log messages:
|
||||
|
||||
```sh
|
||||
2022-09-15T13:35:41.155Z DEBUG alerting rule "TestGroup":"Conns" (2601299393013563564) at 2022-09-15T15:35:41+02:00: query returned 0 series (elapsed: 5.896041ms, isPartial: false)
|
||||
2022-09-15T13:35:56.149Z DEBUG datasource request: executing POST request with params "denyPartialResponse=true&query=sum%28vm_tcplistener_conns%7Binstance%3D%22localhost%3A8429%22%7D%29+by%28instance%29+%3E+0&step=15s&time=1663248945"
|
||||
2022-09-15T13:35:56.178Z DEBUG alerting rule "TestGroup":"Conns" (2601299393013563564) at 2022-09-15T15:35:56+02:00: query returned 1 series (elapsed: 28.368208ms, isPartial: false)
|
||||
2022-09-15T13:35:56.178Z DEBUG datasource request: executing POST request with params "denyPartialResponse=true&query=sum%28vm_tcplistener_conns%7Binstance%3D%22localhost%3A8429%22%7D%29&step=15s&time=1663248945"
|
||||
2022-09-15T13:35:56.179Z DEBUG alerting rule "TestGroup":"Conns" (2601299393013563564) at 2022-09-15T15:35:56+02:00: alert 10705778000901301787 {alertgroup="TestGroup",alertname="Conns",cluster="east-1",instance="localhost:8429",replica="a"} created in state PENDING
|
||||
2026-08-20T08:21:29.464Z info VictoriaMetrics/app/vmalert/datasource/client.go:262 DEBUG datasource request: executing POST request with params "http://victoriametrics:8428/api/v1/query?query=up%7Bjob%3D~%22.%2A%28victoriametrics%7Cvmselect%7Cvminsert%7Cvmstorage%7Cvmagent%7Cvmalert%7Cvmsingle%7Cvmalertmanager%7Cvmauth%29.%2A%22%7D&step=300s&time=2026-08-20T08%3A20%3A00Z"
|
||||
2026-08-20T08:21:29.465Z info VictoriaMetrics/app/vmalert/rule/alerting.go:273 DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:20:00Z: query returned 0 series (series_fetched: 0, elapsed: 1.075166ms, isPartial: false)
|
||||
...
|
||||
2022-09-15T13:36:56.153Z DEBUG alerting rule "TestGroup":"Conns" (2601299393013563564) at 2022-09-15T15:36:56+02:00: alert 10705778000901301787 {alertgroup="TestGroup",alertname="Conns",cluster="east-1",instance="localhost:8429",replica="a"} PENDING => FIRING: 1m0s since becoming active at 2022-09-15 15:35:56.126006 +0200 CEST m=+39.384575417
|
||||
2026-08-20T08:22:29.466Z info VictoriaMetrics/app/vmalert/datasource/client.go:262 DEBUG datasource request: executing POST request with params "http://victoriametrics:8428/api/v1/query?query=up%7Bjob%3D~%22.%2A%28victoriametrics%7Cvmselect%7Cvminsert%7Cvmstorage%7Cvmagent%7Cvmalert%7Cvmsingle%7Cvmalertmanager%7Cvmauth%29.%2A%22%7D&step=300s&time=2026-08-20T08%3A21%3A00Z"
|
||||
2026-08-20T08:22:29.468Z info VictoriaMetrics/app/vmalert/rule/alerting.go:273 DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:21:00Z: query returned 2 series (series_fetched: 2, elapsed: 2.055916ms, isPartial: false)
|
||||
2026-08-20T08:22:29.469Z info VictoriaMetrics/app/vmalert/rule/alerting.go:273 DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:21:00Z: alert 4671711516378822929 {alertgroup="vm-health",alertname="ServiceDown",instance="victoriametrics:8428",job="victoriametrics",severity="critical"} created in state PENDING
|
||||
2026-08-20T08:22:29.469Z info VictoriaMetrics/app/vmalert/rule/alerting.go:273 DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:21:00Z: alert 6230585559362831632 {alertgroup="vm-health",alertname="ServiceDown",instance="vmagent:8429",job="vmagent",severity="critical"} created in state PENDING
|
||||
...
|
||||
2026-08-20T08:23:29.463Z info VictoriaMetrics/app/vmalert/datasource/client.go:262 DEBUG datasource request: executing POST request with params "http://victoriametrics:8428/api/v1/query?query=up%7Bjob%3D~%22.%2A%28victoriametrics%7Cvmselect%7Cvminsert%7Cvmstorage%7Cvmagent%7Cvmalert%7Cvmsingle%7Cvmalertmanager%7Cvmauth%29.%2A%22%7D&step=300s&time=2026-08-20T08%3A22%3A00Z"
|
||||
2026-08-20T08:23:29.465Z info VictoriaMetrics/app/vmalert/rule/alerting.go:273 DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:22:00Z: query returned 2 series (series_fetched: 2, elapsed: 1.391416ms, isPartial: false)
|
||||
2026-08-20T08:23:29.466Z info VictoriaMetrics/app/vmalert/rule/alerting.go:273 DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:22:00Z: alert 4671711516378822929 {alertgroup="vm-health",alertname="ServiceDown",instance="victoriametrics:8428",job="victoriametrics",severity="critical"} PENDING => FIRING: 1m0s since becoming active at 2026-08-20 08:21:00 +0000 UTC
|
||||
2026-08-20T08:23:29.466Z info VictoriaMetrics/app/vmalert/rule/alerting.go:273 DEBUG alerting rule "/etc/alerts/alerts-health.yml", "vm-health":"ServiceDown" (1340947595484135783) at 2026-08-20T08:22:00Z: alert 6230585559362831632 {alertgroup="vm-health",alertname="ServiceDown",instance="vmagent:8429",job="vmagent",severity="critical"} PENDING => FIRING: 1m0s since becoming active at 2026-08-20 08:21:00 +0000 UTC
|
||||
```
|
||||
|
||||
Sensitive info is stripped from the `curl` examples - see [security](#security) section for more details.
|
||||
|
||||
@@ -65,6 +65,9 @@ func writePrometheusMetrics(w io.Writer) {
|
||||
// Export start time and uptime in seconds
|
||||
metrics.WriteGaugeUint64(w, "vm_app_start_timestamp", uint64(startTime.Unix()))
|
||||
metrics.WriteGaugeUint64(w, "vm_app_uptime_seconds", uint64(time.Since(startTime).Seconds()))
|
||||
if uncleanShutdownEnabled.Load() {
|
||||
metrics.WriteGaugeUint64(w, "vm_app_prev_shutdown_unclean", uncleanShutdown)
|
||||
}
|
||||
|
||||
// Export flags as metrics.
|
||||
isSetMap := make(map[string]bool)
|
||||
|
||||
@@ -13,13 +13,13 @@ type osInfo struct {
|
||||
release string
|
||||
}
|
||||
|
||||
var os osInfo
|
||||
var hostOS osInfo
|
||||
var initOSOnce sync.Once
|
||||
|
||||
func writeOSMetrics(w io.Writer) {
|
||||
initOSOnce.Do(initOS)
|
||||
|
||||
if os.name != "" {
|
||||
metrics.WriteGaugeUint64(w, fmt.Sprintf(`vm_os_info{os=%q, release=%q}`, os.name, os.release), 1)
|
||||
if hostOS.name != "" {
|
||||
metrics.WriteGaugeUint64(w, fmt.Sprintf(`vm_os_info{os=%q, release=%q}`, hostOS.name, hostOS.release), 1)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,7 +8,7 @@ import (
|
||||
)
|
||||
|
||||
func initOS() {
|
||||
os = osInfo{name: "darwin"}
|
||||
hostOS = osInfo{name: "darwin"}
|
||||
|
||||
out, err := exec.Command("sysctl", "-n", "kern.osrelease").Output()
|
||||
if err != nil {
|
||||
@@ -16,5 +16,5 @@ func initOS() {
|
||||
return
|
||||
}
|
||||
|
||||
os.release = strings.TrimSpace(string(out))
|
||||
hostOS.release = strings.TrimSpace(string(out))
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ import (
|
||||
)
|
||||
|
||||
func initOS() {
|
||||
os = osInfo{name: "linux"}
|
||||
hostOS = osInfo{name: "linux"}
|
||||
|
||||
var uname syscall.Utsname
|
||||
if err := syscall.Uname(&uname); err != nil {
|
||||
@@ -22,5 +22,5 @@ func initOS() {
|
||||
}
|
||||
ur = append(ur, byte(v))
|
||||
}
|
||||
os.release = string(ur)
|
||||
hostOS.release = string(ur)
|
||||
}
|
||||
|
||||
@@ -8,12 +8,12 @@ import (
|
||||
)
|
||||
|
||||
func initOS() {
|
||||
os = osInfo{name: "windows"}
|
||||
hostOS = osInfo{name: "windows"}
|
||||
|
||||
ver := windows.RtlGetVersion()
|
||||
if ver == nil {
|
||||
logger.Warnf("vm_os_info metric will miss release info since windows.RtlGetVersion returned nil version")
|
||||
return
|
||||
}
|
||||
os.release = fmt.Sprintf("%d.%d.%d", ver.MajorVersion, ver.MinorVersion, ver.BuildNumber)
|
||||
hostOS.release = fmt.Sprintf("%d.%d.%d", ver.MajorVersion, ver.MinorVersion, ver.BuildNumber)
|
||||
}
|
||||
|
||||
53
lib/appmetrics/unclean_shutdown.go
Normal file
53
lib/appmetrics/unclean_shutdown.go
Normal file
@@ -0,0 +1,53 @@
|
||||
package appmetrics
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
|
||||
)
|
||||
|
||||
// UncleanShutdownMarkerFilename is the marker file used to detect a previous unclean shutdown.
|
||||
const UncleanShutdownMarkerFilename = ".vm_app_running"
|
||||
|
||||
var (
|
||||
uncleanShutdownEnabled atomic.Bool
|
||||
uncleanShutdown uint64
|
||||
)
|
||||
|
||||
// MustCreateUncleanShutdownMarker creates an UncleanShutdownMarkerFilename marker file in the given dirPath.
|
||||
// Must be called once on program startup and paired with a single MustRemoveUncleanShutdownMarker call on exit.
|
||||
//
|
||||
// If the marker file already exists on startup, it indicates a previous unclean shutdown and uncleanShutdown is set to 1.
|
||||
func MustCreateUncleanShutdownMarker(dirPath string) {
|
||||
if !uncleanShutdownEnabled.CompareAndSwap(false, true) {
|
||||
logger.Fatalf("BUG: unclean shutdown marker was already initialized. It could only be called once")
|
||||
}
|
||||
marker := filepath.Join(dirPath, UncleanShutdownMarkerFilename)
|
||||
f, err := os.OpenFile(marker, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
|
||||
if err == nil {
|
||||
fs.MustClose(f)
|
||||
return
|
||||
}
|
||||
if os.IsExist(err) {
|
||||
uncleanShutdown = 1
|
||||
logger.Warnf("Previous shutdown was unclean since file %q exists. Please check logs and investigate the reason of unclean shutdown", marker)
|
||||
return
|
||||
}
|
||||
logger.Panicf("FATAL: cannot create unclean shutdown marker %q: %s", marker, err)
|
||||
}
|
||||
|
||||
// MustRemoveUncleanShutdownMarker removes the UncleanShutdownMarkerFilename marker file created by MustCreateUncleanShutdownMarker.
|
||||
// Must be called once, as late as possible before program exit.
|
||||
func MustRemoveUncleanShutdownMarker(dirPath string) {
|
||||
if !uncleanShutdownEnabled.Load() {
|
||||
logger.Fatalf("BUG: unclean shutdown marker was not initialized with MustCreateUncleanShutdownMarker call")
|
||||
}
|
||||
marker := filepath.Join(dirPath, UncleanShutdownMarkerFilename)
|
||||
if err := os.Remove(marker); err != nil {
|
||||
logger.Fatalf("FATAL: cannot remove unclean shutdown marker %q: %s", marker, err)
|
||||
}
|
||||
fs.MustSyncPath(dirPath)
|
||||
}
|
||||
60
lib/appmetrics/unclean_shutdown_test.go
Normal file
60
lib/appmetrics/unclean_shutdown_test.go
Normal file
@@ -0,0 +1,60 @@
|
||||
package appmetrics
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestUncleanShutdownLifecycle(t *testing.T) {
|
||||
|
||||
t.Cleanup(func() {
|
||||
uncleanShutdownEnabled.Store(false)
|
||||
})
|
||||
dirPath := t.TempDir()
|
||||
markerPath := filepath.Join(dirPath, UncleanShutdownMarkerFilename)
|
||||
|
||||
// unclean logic is disabled. the unclean shutdown metric should not be exposed
|
||||
var bb bytes.Buffer
|
||||
writePrometheusMetrics(&bb)
|
||||
if strings.Contains(bb.String(), "vm_app_prev_shutdown_unclean") {
|
||||
t.Fatalf("unexpected unclean shutdown metric before starting the marker")
|
||||
}
|
||||
|
||||
// clean start, the metric must report 0
|
||||
MustCreateUncleanShutdownMarker(dirPath)
|
||||
mustContainUncleanShutdownMetric(t, 0)
|
||||
if _, err := os.Stat(markerPath); err != nil {
|
||||
t.Fatalf("cannot stat the running marker after the first start: %s", err)
|
||||
}
|
||||
MustRemoveUncleanShutdownMarker(dirPath)
|
||||
if _, err := os.Stat(markerPath); !os.IsNotExist(err) {
|
||||
t.Fatalf("unexpected running marker after a clean shutdown; got error %v; want os.ErrNotExist", err)
|
||||
}
|
||||
uncleanShutdownEnabled.Store(false)
|
||||
|
||||
// simulate prev unclean shutdown, the metric must report 1
|
||||
if err := os.WriteFile(markerPath, nil, 0600); err != nil {
|
||||
t.Fatalf("cannot create test marker: %s", err)
|
||||
}
|
||||
MustCreateUncleanShutdownMarker(dirPath)
|
||||
mustContainUncleanShutdownMetric(t, 1)
|
||||
MustRemoveUncleanShutdownMarker(dirPath)
|
||||
if _, err := os.Stat(markerPath); !os.IsNotExist(err) {
|
||||
t.Fatalf("unexpected running marker after a clean shutdown; got error %v; want os.ErrNotExist", err)
|
||||
}
|
||||
}
|
||||
|
||||
func mustContainUncleanShutdownMetric(t *testing.T, value uint64) {
|
||||
t.Helper()
|
||||
|
||||
var bb bytes.Buffer
|
||||
writePrometheusMetrics(&bb)
|
||||
want := "vm_app_prev_shutdown_unclean " + strconv.FormatUint(value, 10) + "\n"
|
||||
if !strings.Contains(bb.String(), want) {
|
||||
t.Fatalf("missing %q in the exported app metrics", want)
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/appmetrics"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/backupnames"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
|
||||
)
|
||||
@@ -107,7 +108,7 @@ func appendFilesInternal(dst []string, d *os.File) ([]string, error) {
|
||||
}
|
||||
|
||||
func isSpecialFile(name string) bool {
|
||||
return name == "flock.lock" || name == backupnames.RestoreInProgressFilename || name == backupnames.RestoreMarkFileName || strings.HasSuffix(name, ".tmp")
|
||||
return name == "flock.lock" || name == appmetrics.UncleanShutdownMarkerFilename || name == backupnames.RestoreInProgressFilename || name == backupnames.RestoreMarkFileName || strings.HasSuffix(name, ".tmp")
|
||||
}
|
||||
|
||||
// RemoveEmptyDirs recursively removes empty directories under the given dir.
|
||||
|
||||
@@ -39,8 +39,29 @@ func NewArrayBool(name, description string) *ArrayBool {
|
||||
}
|
||||
|
||||
// NewArrayInt returns new ArrayInt with the given name, defaultValue and description.
|
||||
//
|
||||
// -help shows defaultValue as a plain number. Use NewArrayIntWithDynamicDefault when
|
||||
// defaultValue is calculated at runtime.
|
||||
func NewArrayInt(name string, defaultValue int, description string) *ArrayInt {
|
||||
description += fmt.Sprintf(" (default %d)", defaultValue)
|
||||
return newArrayInt(name, defaultValue, strconv.Itoa(defaultValue), description)
|
||||
}
|
||||
|
||||
// NewArrayIntWithDynamicDefault returns new ArrayInt with the given name, defaultValue and description.
|
||||
//
|
||||
// Use it instead of NewArrayInt when defaultValue is calculated at runtime.
|
||||
// See NewIntWithDynamicDefault for why such a value needs a hint.
|
||||
func NewArrayIntWithDynamicDefault(name string, defaultValue int, defaultValueHint, description string) *ArrayInt {
|
||||
if defaultValueHint == "" {
|
||||
panic(fmt.Sprintf("BUG: missing defaultValueHint for -%s", name))
|
||||
}
|
||||
return newArrayInt(name, defaultValue, fmt.Sprintf("%d = %s", defaultValue, defaultValueHint), description)
|
||||
}
|
||||
|
||||
// newArrayInt registers an int array flag, which shows defaultValueText as its default in -help.
|
||||
//
|
||||
// Array flags keep the default in the description, since flag.Var hides an empty DefValue.
|
||||
func newArrayInt(name string, defaultValue int, defaultValueText, description string) *ArrayInt {
|
||||
description += fmt.Sprintf(" (default %s)", defaultValueText)
|
||||
description += "\nSupports `array` of values separated by comma or specified via multiple flags."
|
||||
description += "\nEmpty values are set to default value."
|
||||
a := &ArrayInt{
|
||||
|
||||
@@ -7,6 +7,25 @@ import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
// NewIntWithDynamicDefault returns a new int flag with the given name, defaultValue and description.
|
||||
//
|
||||
// Use it instead of flag.Int when defaultValue is calculated at runtime, for example
|
||||
// from the number of CPU cores. Such a value differs per machine, so -help shows both
|
||||
// the value and defaultValueHint, for example "16 = 2 * availableCPUs".
|
||||
//
|
||||
// Only -help output changes. The flag value stays defaultValue.
|
||||
func NewIntWithDynamicDefault(name string, defaultValue int, defaultValueHint, description string) *int {
|
||||
if defaultValueHint == "" {
|
||||
panic(fmt.Sprintf("BUG: missing defaultValueHint for -%s", name))
|
||||
}
|
||||
p := flag.Int(name, defaultValue, description)
|
||||
|
||||
// DefValue is only the text shown by -help: "default value (as text); for usage message".
|
||||
flag.Lookup(name).DefValue = fmt.Sprintf("%d = %s", defaultValue, defaultValueHint)
|
||||
|
||||
return p
|
||||
}
|
||||
|
||||
// WriteFlags writes all the explicitly set flags to w.
|
||||
func WriteFlags(w io.Writer) {
|
||||
flag.Visit(func(f *flag.Flag) {
|
||||
|
||||
51
lib/flagutil/flag_test.go
Normal file
51
lib/flagutil/flag_test.go
Normal file
@@ -0,0 +1,51 @@
|
||||
package flagutil
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The flags are registered at package level, since flag registration panics when it repeats.
|
||||
var (
|
||||
fooFlagIntDynamicDefault = NewIntWithDynamicDefault("fooFlagIntDynamicDefault", 42, "2 * availableCPUs", "test")
|
||||
fooFlagArrayIntDynamicDefault = NewArrayIntWithDynamicDefault("fooFlagArrayIntDynamicDefault", 42, "2 * availableCPUs", "test")
|
||||
fooFlagArrayIntPlainDefault = NewArrayInt("fooFlagArrayIntPlainDefault", 42, "test")
|
||||
)
|
||||
|
||||
func TestNewIntWithDynamicDefaultSuccess(t *testing.T) {
|
||||
// -help must show the value together with the hint.
|
||||
f := flag.Lookup("fooFlagIntDynamicDefault")
|
||||
if f.DefValue != "42 = 2 * availableCPUs" {
|
||||
t.Fatalf("unexpected DefValue; got %q; want %q", f.DefValue, "42 = 2 * availableCPUs")
|
||||
}
|
||||
|
||||
// the flag value must stay the calculated one.
|
||||
if *fooFlagIntDynamicDefault != 42 {
|
||||
t.Fatalf("unexpected flag value; got %d; want %d", *fooFlagIntDynamicDefault, 42)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewArrayIntWithDynamicDefaultSuccess(t *testing.T) {
|
||||
// array flags keep the default in the description, so the hint must go there.
|
||||
f := flag.Lookup("fooFlagArrayIntDynamicDefault")
|
||||
if !strings.Contains(f.Usage, "(default 42 = 2 * availableCPUs)") {
|
||||
t.Fatalf("missing the hint in the flag description; got %q", f.Usage)
|
||||
}
|
||||
|
||||
// the default value must stay the calculated one.
|
||||
if n := fooFlagArrayIntDynamicDefault.GetOptionalArg(0); n != 42 {
|
||||
t.Fatalf("unexpected default value; got %d; want %d", n, 42)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewArrayIntKeepsPlainDefault(t *testing.T) {
|
||||
// NewArrayInt must keep showing a plain number, since it shares the body with the dynamic one.
|
||||
f := flag.Lookup("fooFlagArrayIntPlainDefault")
|
||||
if !strings.Contains(f.Usage, "(default 42)") {
|
||||
t.Fatalf("unexpected flag description; got %q", f.Usage)
|
||||
}
|
||||
if n := fooFlagArrayIntPlainDefault.GetOptionalArg(0); n != 42 {
|
||||
t.Fatalf("unexpected default value; got %d; want %d", n, 42)
|
||||
}
|
||||
}
|
||||
@@ -1,14 +1,15 @@
|
||||
package fsutil
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"sync"
|
||||
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
|
||||
)
|
||||
|
||||
var maxConcurrency = flag.Int("fs.maxConcurrency", getDefaultConcurrency(), "The maximum number of concurrent goroutines to work with files; smaller values may help reducing Go scheduling latency "+
|
||||
"on systems with small number of CPU cores; higher values may help reducing data ingestion latency on systems with high-latency storage such as NFS or Ceph")
|
||||
var maxConcurrency = flagutil.NewIntWithDynamicDefault("fs.maxConcurrency", getDefaultConcurrency(), "fsutil.getDefaultConcurrency()",
|
||||
"The maximum number of concurrent goroutines to work with files; smaller values may help reducing Go scheduling latency "+
|
||||
"on systems with small number of CPU cores; higher values may help reducing data ingestion latency on systems with high-latency storage such as NFS or Ceph")
|
||||
|
||||
func getDefaultConcurrency() int {
|
||||
n := min(16*cgroup.AvailableCPUs(), 256)
|
||||
|
||||
@@ -79,7 +79,7 @@ func ParseTimeAt(s string, currentTimestamp int64) (int64, error) {
|
||||
// Parse YYYY
|
||||
return parseTimeAt("2006", s, tzOffset, sOrig)
|
||||
}
|
||||
if !strings.Contains(sOrig, "-") {
|
||||
if !strings.Contains(sOrig, "-") || getExpIndex(sOrig) >= 0 {
|
||||
nsec, ok := TryParseUnixTimestamp(sOrig)
|
||||
if !ok {
|
||||
return 0, fmt.Errorf("cannot parse numeric timestamp %q", sOrig)
|
||||
@@ -186,80 +186,84 @@ func getExpIndex(s string) int {
|
||||
}
|
||||
|
||||
func tryParseScientificUnixTimestamp(s string, decimalExp int64) (int64, bool) {
|
||||
if decimalExp < 0 {
|
||||
// Negative exponents on a fractional mantissa are intentionally not
|
||||
// supported. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11268
|
||||
intStr, fracStr, ok := expandScientificUnixTimestamp(s, decimalExp)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
dotIdx := strings.IndexByte(s, '.')
|
||||
if dotIdx < 0 {
|
||||
n, ok := tryParseInt64(s)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
n, ok = multiplyByDecimalExp(n, decimalExp)
|
||||
if fracStr == "" {
|
||||
n, ok := tryParseInt64(intStr)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
return getUnixTimestampNanoseconds(n), true
|
||||
}
|
||||
|
||||
intStr := s[:dotIdx]
|
||||
fracStr := s[dotIdx+1:]
|
||||
if decimalExp >= int64(len(fracStr)) {
|
||||
// The exponent shifts the decimal point past every fractional digit.
|
||||
n, ok := tryParseDecimalMantissaAsInt(intStr, fracStr)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
decimalExp -= int64(len(fracStr))
|
||||
n, ok = multiplyByDecimalExp(n, decimalExp)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
return getUnixTimestampNanoseconds(n), true
|
||||
}
|
||||
|
||||
// The exponent leaves fractional digits, e.g. 1.784144612388E9 == 1784144612.388
|
||||
if decimalExp >= int64(len(decimalMultipliers)) {
|
||||
return 0, false
|
||||
}
|
||||
decimalExpInt := int(decimalExp)
|
||||
intStr = s[:dotIdx] + fracStr[:decimalExpInt]
|
||||
fracStr = fracStr[decimalExpInt:]
|
||||
return tryParseFractionalUnixTimestamp(intStr, fracStr)
|
||||
}
|
||||
|
||||
func tryParseDecimalMantissaAsInt(intStr, fracStr string) (int64, bool) {
|
||||
n, ok := tryParseInt64(intStr)
|
||||
if !ok {
|
||||
return 0, false
|
||||
func expandScientificUnixTimestamp(s string, decimalExp int64) (string, string, bool) {
|
||||
dotIdx := strings.IndexByte(s, '.')
|
||||
intStr := s
|
||||
fracStr := ""
|
||||
if dotIdx >= 0 {
|
||||
intStr = s[:dotIdx]
|
||||
fracStr = s[dotIdx+1:]
|
||||
}
|
||||
if _, ok := tryParseInt64(intStr); !ok {
|
||||
return "", "", false
|
||||
}
|
||||
if !isDecimalString(fracStr) {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
decimalExp := int64(len(fracStr))
|
||||
num, ok := multiplyByDecimalExp(n, decimalExp)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
|
||||
frac, ok := tryParseInt64(fracStr)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
|
||||
if num >= 0 {
|
||||
if num > math.MaxInt64-frac {
|
||||
return 0, false
|
||||
isNegativeExp := decimalExp < 0
|
||||
if isNegativeExp {
|
||||
if decimalExp <= -int64(len(decimalMultipliers)) {
|
||||
return "", "", false
|
||||
}
|
||||
num += frac
|
||||
decimalExp = -decimalExp
|
||||
}
|
||||
if decimalExp > int64(math.MaxInt) || decimalExp < int64(math.MinInt) {
|
||||
return "", "", false
|
||||
}
|
||||
|
||||
intStr = strings.TrimPrefix(intStr, "+")
|
||||
isNegative := strings.HasPrefix(intStr, "-")
|
||||
if isNegative {
|
||||
intStr = intStr[1:]
|
||||
}
|
||||
|
||||
var shiftedIntStr, shiftedFracStr string
|
||||
if isNegativeExp {
|
||||
// e.g.
|
||||
// 1. the integer and fractional part of 1.23e-5 should be 0 and 0000123 respectively.
|
||||
// 2. the integer and fractional part of 123.4e-1 should be 12 and 34 respectively.
|
||||
if decimalExp >= int64(len(intStr)) {
|
||||
zerosToAdd := decimalExp - int64(len(intStr))
|
||||
shiftedIntStr = "0"
|
||||
shiftedFracStr = strings.Repeat("0", int(zerosToAdd)) + intStr + fracStr
|
||||
} else {
|
||||
decimalExpInt := int(decimalExp)
|
||||
shiftedIntStr = intStr[:len(intStr)-decimalExpInt]
|
||||
shiftedFracStr = intStr[len(intStr)-decimalExpInt:] + fracStr
|
||||
}
|
||||
} else if decimalExp >= int64(len(fracStr)) {
|
||||
zerosToAdd := decimalExp - int64(len(fracStr))
|
||||
if zerosToAdd >= int64(len(decimalMultipliers)) {
|
||||
return "", "", false
|
||||
}
|
||||
// e.g. the integer part and fractional part of 1.23e5 should be 123000 and 0 respectively.
|
||||
shiftedIntStr = intStr + fracStr + strings.Repeat("0", int(zerosToAdd))
|
||||
shiftedFracStr = ""
|
||||
} else {
|
||||
if num < math.MinInt64+frac {
|
||||
return 0, false
|
||||
}
|
||||
num -= frac
|
||||
decimalExpInt := int(decimalExp)
|
||||
shiftedIntStr = intStr + fracStr[:decimalExpInt]
|
||||
shiftedFracStr = fracStr[decimalExpInt:]
|
||||
}
|
||||
|
||||
return num, true
|
||||
if isNegative {
|
||||
shiftedIntStr = "-" + shiftedIntStr
|
||||
}
|
||||
return shiftedIntStr, shiftedFracStr, true
|
||||
}
|
||||
|
||||
func tryParseFractionalUnixTimestamp(intStr, fracStr string) (int64, bool) {
|
||||
@@ -270,15 +274,11 @@ func tryParseFractionalUnixTimestamp(intStr, fracStr string) (int64, bool) {
|
||||
isNegative := n < 0 || n == 0 && strings.HasPrefix(intStr, "-")
|
||||
|
||||
multiplier, maxFracDigits := getUnixTimestampMultiplier(n)
|
||||
if !isDecimalString(fracStr) {
|
||||
return 0, false
|
||||
}
|
||||
// Truncate the fractional digits to valid length according to the unit precision.
|
||||
if len(fracStr) > maxFracDigits {
|
||||
// 1.123456789XXX is invalid.
|
||||
tail := fracStr[maxFracDigits:]
|
||||
for i := 0; i < len(tail); i++ {
|
||||
if tail[i] < '0' || tail[i] > '9' {
|
||||
return 0, false
|
||||
}
|
||||
}
|
||||
fracStr = fracStr[:maxFracDigits]
|
||||
}
|
||||
if len(fracStr) == 0 {
|
||||
@@ -309,26 +309,6 @@ func tryParseFractionalUnixTimestamp(intStr, fracStr string) (int64, bool) {
|
||||
return n + frac, true
|
||||
}
|
||||
|
||||
func multiplyByDecimalExp(n int64, decimalExp int64) (int64, bool) {
|
||||
if decimalExp < 0 {
|
||||
return 0, false
|
||||
}
|
||||
if decimalExp >= int64(len(decimalMultipliers)) {
|
||||
return 0, false
|
||||
}
|
||||
if decimalExp == 0 {
|
||||
return n, true
|
||||
}
|
||||
|
||||
m := decimalMultipliers[decimalExp]
|
||||
|
||||
if n >= 0 && n > math.MaxInt64/m || n < 0 && n < math.MinInt64/m {
|
||||
return 0, false
|
||||
}
|
||||
|
||||
return n * m, true
|
||||
}
|
||||
|
||||
var decimalMultipliers = [...]int64{0, 1e1, 1e2, 1e3, 1e4, 1e5, 1e6, 1e7, 1e8, 1e9, 1e10, 1e11, 1e12, 1e13, 1e14, 1e15, 1e16, 1e17, 1e18}
|
||||
|
||||
const (
|
||||
@@ -369,3 +349,12 @@ func tryParseInt64(s string) (int64, bool) {
|
||||
}
|
||||
return n, true
|
||||
}
|
||||
|
||||
func isDecimalString(s string) bool {
|
||||
for i := 0; i < len(s); i++ {
|
||||
if s[i] < '0' || s[i] > '9' {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -76,6 +76,7 @@ func TestTryParseUnixTimestamp_Success(t *testing.T) {
|
||||
f("-1.23456789e9", -1234567890_000_000_000)
|
||||
f("1.234567890123456789e18", 1234567890_123_456_789)
|
||||
f("-1.234567890123456789e18", -1234567890_123_456_789)
|
||||
f("0.0000000001e19", 1000000000_000_000_000)
|
||||
f("0.23456789e9", 234567890_000_000_000)
|
||||
f("123.456789123e9", 123456789_123_000_000)
|
||||
f("-1234.5678912e9", -1234567891_200_000_000)
|
||||
@@ -99,7 +100,19 @@ func TestTryParseUnixTimestamp_Success(t *testing.T) {
|
||||
f("1.23456789e9", 1234567890_000_000_000) // exponent consumes all frac digits (integer result)
|
||||
f("1.23e1", 12_300_000_000) // == 12.3
|
||||
f("1.234e0", 1_234_000_000) // == 1.234
|
||||
f("-0.1e1", -1_000_000_000) // == -1
|
||||
f("1234567890123456789.0e0", 1234567890_123_456_789)
|
||||
f("1000e-1", 100_000_000_000)
|
||||
f("1000E-1", 100_000_000_000)
|
||||
f("-1000e-1", -100_000_000_000)
|
||||
f("1e-1", 100_000_000)
|
||||
f("-1e-1", -100_000_000)
|
||||
f("1000.01e-1", 100_001_000_000)
|
||||
f("-1000.01e-1", -100_001_000_000)
|
||||
f("1.2E-1", 120_000_000)
|
||||
f("1.2345678901e-1", 123_456_789)
|
||||
f("0.12345678901", 123_456_789)
|
||||
f("1.2345678901234567890123e-1", 123_456_789)
|
||||
|
||||
}
|
||||
|
||||
@@ -138,7 +151,8 @@ func TestTryParseUnixTimestamp_Failure(t *testing.T) {
|
||||
f("1.3e123456789090123")
|
||||
|
||||
// negative decimal exponent
|
||||
f("1E-1")
|
||||
f("1e-9223372036854775808")
|
||||
f("1e-123456789090123")
|
||||
f("1.3e-123456789090123")
|
||||
|
||||
}
|
||||
@@ -164,6 +178,8 @@ func TestParseTimeAtSuccess(t *testing.T) {
|
||||
f("1562529662.678", now, 1562529662_678_000_000)
|
||||
f("1562529662.678123", now, 1562529662_678_123_000)
|
||||
f("1562529662.678123456", now, 1562529662_678_123_456)
|
||||
f("1000e-1", now, 100_000_000_000)
|
||||
f("1000.01e-1", now, 100_001_000_000)
|
||||
|
||||
// unix timestamp in milliseconds
|
||||
f("1562529662678", now, 1562529662_678_000_000)
|
||||
|
||||
@@ -10,16 +10,18 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
|
||||
"github.com/VictoriaMetrics/metrics"
|
||||
)
|
||||
|
||||
var (
|
||||
maxConcurrentInserts = flag.Int("maxConcurrentInserts", 2*cgroup.AvailableCPUs(), "The maximum number of concurrent insert requests. "+
|
||||
"Set higher value when clients send data over slow networks. "+
|
||||
"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage. "+
|
||||
"See also -insert.maxQueueDuration")
|
||||
maxConcurrentInserts = flagutil.NewIntWithDynamicDefault("maxConcurrentInserts", 2*cgroup.AvailableCPUs(), "2*cgroup.AvailableCPUs()",
|
||||
"The maximum number of concurrent insert requests. "+
|
||||
"Set higher value when clients send data over slow networks. "+
|
||||
"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage. "+
|
||||
"See also -insert.maxQueueDuration")
|
||||
maxQueueDuration = flag.Duration("insert.maxQueueDuration", time.Minute, "The maximum duration to wait in the queue when -maxConcurrentInserts "+
|
||||
"concurrent insert requests are executed")
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user