docs/victorialogs/CHANGELOG.md: cut v1.23.0-victorialogs release

This is a follow-up for 5bb012b67b

Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9013

docs/victorialogs/CHANGELOG.md

@@ -18,13 +18,19 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
 
 ## tip
 
+## [v1.23.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.23.0-victorialogs)
+
+Released at 2025-05-28
+
 * FEATURE: [web UI](https://docs.victoriametrics.com/victorialogs/querying/#web-ui): add "Live" tab that allows monitoring logs in real-time as they arrive. This feature helps users to observe the most recent log entries without manual refreshing, making troubleshooting and monitoring more efficient. See [#7046](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7046).
 * FEATURE: [dashboards/cluster](https://grafana.com/grafana/dashboards/23274) and [dashboards/single](https://grafana.com/grafana/dashboards/22084): add panels for [Pressure Stall Information (PSI)](https://docs.kernel.org/accounting/psi.html) metrics to dashboards. They could help to identify shortage of resources for VictoriaLogs components.
+* FEATURE: [`math` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#math-pipe): add `now()` function, which returns the current [Unix timestamp](https://en.wikipedia.org/wiki/Unix_time) in nanoseconds.
 
 * BUGFIX: [OpenTelemetry](https://docs.victoriametrics.com/victorialogs/data-ingestion/opentelemetry/): properly handle nested attributes by expanding them into separate top-level fields. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8862).
 * BUGFIX: [Datadog](https://docs.victoriametrics.com/victorialogs/data-ingestion/datadog/): respond HTTP 202 instead of HTTP 200 on successful Datadog endpoint ingestion as it's strictly required by Datadog agent. See [#8956](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8956).
 * BUGFIX: [web UI](https://docs.victoriametrics.com/victorialogs/querying/#web-ui): properly escape special characters in field values shown in autocomplete suggestions. See [#8925](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8925).
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): Properly handle time filters when querying vlstorage directly or through vlselect. See [#8985](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8985).
+* BUGFIX: Self-healing from OOM interruption during the creation of a daily partition. See [#8873](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8873).
 
 ## [v1.22.2](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.22.2-victorialogs)
 

docs/victorialogs/LogsQL.md

@@ -2403,6 +2403,7 @@ The following mathematical operations are supported by `math` pipe:
 - `ln(arg)` - returns [natural logarithm](https://en.wikipedia.org/wiki/Natural_logarithm) for the given `arg`
 - `max(arg1, ..., argN)` - returns the maximum value among the given `arg1`, ..., `argN`
 - `min(arg1, ..., argN)` - returns the minimum value among the given `arg1`, ..., `argN`
+- `now()` - returns the current [Unix timestamp](https://en.wikipedia.org/wiki/Unix_time) in nanoseconds.
 - `rand()` - returns pseudo-random number in the range `[0...1)`.
 - `round(arg)` - returns rounded to integer value for the given `arg`. The `round()` accepts optional `nearest` arg, which allows rounding the number to the given `nearest` multiple.
   For example, `round(temperature, 0.1)` rounds `temperature` field to one decimal digit after the point.
@@ -3668,7 +3669,7 @@ See also:
 
 ### json_values stats
 
-`json_values(field1, ..., fieldN)` [stats pipe function](#stats-pipe-functions) returns packs the given fields into JSON per every log entry and returns JSON array,
+`json_values(field1, ..., fieldN)` [stats pipe function](#stats-pipe-functions) packs the given fields into JSON per every log entry and returns JSON array,
 which can be unrolled with [`unroll` pipe](#unroll-pipe).
 
 For example, the following query returns per-`app` JSON arrays containing [`_time`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#time-field)

docs/victorialogs/logsql-examples.md

@@ -482,3 +482,20 @@ plus up to 100 logs after the given log message:
 ```logsql
 _time:5m stacktrace | stream_context before 10 after 100
 ```
+
+
+## How to get the duration since the last seen log entry matching the given filter?
+
+Use the following query:
+
+```logsql
+_time:1d ERROR
+  | stats max(_time) as max_time
+  | math round((now() - max_time) / 1s) as duration_seconds
+```
+
+It uses [`max()` stats function](https://docs.victoriametrics.com/victorialogs/logsql/#max-stats) for obtaining the maximum value
+for the [`_time` field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#time-field) across all the logs for the last day,
+which contain the `ERROR` word in the [`_msg` field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#message-field).
+Then it uses `now()` function at [`math` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#math-pipe) for calculating
+the duration since the last seen log entry with the `ERROR` word.

lib/logstorage/datadb.go

@@ -1167,6 +1167,31 @@ func mustReadPartNames(path string) []string {
 	partNamesPath := filepath.Join(path, partsFilename)
 	data, err := os.ReadFile(partNamesPath)
 	if err != nil {
+		if os.IsNotExist(err) {
+			// The parts.json file is missing. This can happen if VictoriaLogs shuts down uncleanly
+			// (via OOM crash, a panic, SIGKILL or hardware shutdown) in the middle of creating
+			// new per-day partition inside the mustCreatePartition() function.
+			// Check if there are any part directories in the datadb directory.
+			des := fs.MustReadDir(path)
+			var partDirs []string
+			for _, de := range des {
+				if !fs.IsDirOrSymlink(de) {
+					continue
+				}
+				partDirs = append(partDirs, de.Name())
+			}
+
+			if len(partDirs) == 0 {
+				logger.Warnf("creating missing %s with empty parts list, since no part directories found in %s", partNamesPath, path)
+				mustWritePartNames(path, nil, nil)
+				return []string{}
+			}
+
+			// Parts exist but parts.json is missing - this is an unexpected state that requires manual intervention
+			logger.Panicf("FATAL: cannot read %s: %s; found part directories %v in %s. "+
+				"This indicates corruption. Manually remove the %s partition directory to resolve the corruption (the partition data will be lost)",
+				partNamesPath, err, partDirs, path, path)
+		}
 		logger.Panicf("FATAL: cannot read %s: %s", partNamesPath, err)
 	}
 	var partNames []string

lib/logstorage/partition.go

@@ -62,8 +62,22 @@ func mustDeletePartition(path string) {
 func mustOpenPartition(s *Storage, path string) *partition {
 	name := filepath.Base(path)
 
-	// Open indexdb
 	indexdbPath := filepath.Join(path, indexdbDirname)
+	isIndexDBExist := fs.IsPathExist(indexdbPath)
+
+	datadbPath := filepath.Join(path, datadbDirname)
+	isDatadbExist := fs.IsPathExist(datadbPath)
+
+	if !isIndexDBExist {
+		if isDatadbExist {
+			logger.Panicf("FATAL: indexdb directory %s is missing, but datadb directory %s exists. "+
+				"This indicates corruption. Manually remove the %s partition to resolve it (partition data will be lost)",
+				indexdbPath, datadbPath, path)
+		}
+
+		logger.Warnf("creating missing indexdb directory %s, this could happen if VictoriaLogs shuts down uncleanly (via OOM crash, a panic, SIGKILL or hardware shutdown) while creating new per-day partition", indexdbPath)
+		mustCreateIndexdb(indexdbPath)
+	}
 	idb := mustOpenIndexdb(indexdbPath, name, s)
 
 	// Start initializing the partition
@@ -74,8 +88,11 @@ func mustOpenPartition(s *Storage, path string) *partition {
 		idb:  idb,
 	}
 
-	// Open datadb
-	datadbPath := filepath.Join(path, datadbDirname)
+	if !isDatadbExist {
+		logger.Warnf("creating missing datadb directory %s, this could happen if VictoriaLogs shuts down uncleanly (via OOM crash, a panic, SIGKILL or hardware shutdown) while creating new per-day partition", datadbPath)
+		mustCreateDatadb(datadbPath)
+	}
+
 	pt.ddb = mustOpenDatadb(pt, datadbPath, s.flushInterval)
 
 	return pt

lib/logstorage/pipe_math.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math"
 	"strings"
+	"time"
 
 	"github.com/valyala/fastrand"
 
@@ -580,6 +581,8 @@ func parseMathExprOperand(lex *lexer) (*mathExpr, error) {
 		return parseMathExprMax(lex)
 	case lex.isKeyword("min"):
 		return parseMathExprMin(lex)
+	case lex.isKeyword("now"):
+		return parseMathExprNow(lex)
 	case lex.isKeyword("rand"):
 		return parseMathExprRand(lex)
 	case lex.isKeyword("round"):
@@ -656,6 +659,26 @@ func parseMathExprMin(lex *lexer) (*mathExpr, error) {
 	return me, nil
 }
 
+func parseMathExprNow(lex *lexer) (*mathExpr, error) {
+	if !lex.isKeyword("now") {
+		return nil, fmt.Errorf("missing 'now' keyword")
+	}
+	lex.nextToken()
+
+	args, err := parseMathFuncArgs(lex)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse args for 'now' function: %w", err)
+	}
+	if len(args) != 0 {
+		return nil, fmt.Errorf("'now' function must have no args; got %d args", len(args))
+	}
+	me := &mathExpr{
+		op: "now",
+		f:  mathFuncNow,
+	}
+	return me, nil
+}
+
 func parseMathExprRand(lex *lexer) (*mathExpr, error) {
 	if !lex.isKeyword("rand") {
 		return nil, fmt.Errorf("missing 'rand' keyword")
@@ -1012,6 +1035,13 @@ func mathFuncRand(result []float64, _ [][]float64) {
 	}
 }
 
+func mathFuncNow(result []float64, _ [][]float64) {
+	nowNanos := float64(time.Now().UnixNano())
+	for i := range result {
+		result[i] = nowNanos
+	}
+}
+
 func mathFuncRound(result []float64, args [][]float64) {
 	arg := args[0]
 	if len(args) == 1 {

lib/logstorage/pipe_math_test.go

@@ -30,6 +30,8 @@ func TestParsePipeMathSuccess(t *testing.T) {
 	f(`math (x / 1d * 1s) as x`)
 	f(`math (x - y + z) as x`)
 	f(`math (x - (y + z)) as x`)
+	f(`math now() as current_time`)
+	f(`math round((now() - max_time) / 1s) as duration_seconds`)
 }
 
 func TestParsePipeMathFailure(t *testing.T) {
@@ -49,6 +51,7 @@ func TestParsePipeMathFailure(t *testing.T) {
 	f(`math round() as x`)
 	f(`math round(a, b, c) as x`)
 	f(`math rand(123) as x`)
+	f(`math now(123) as x`)
 }
 
 func TestPipeMath(t *testing.T) {