Merge pull request #144 from flashcatcloud/fix/victorialogs-raw-time-parity

fix(monit-query): normalize raw-mode time args for loki/victorialogs

Author: ysyneu

internal/cli/monit_query.go

@@ -3,6 +3,7 @@ package cli
 import (
 	"encoding/json"
 	"fmt"
+	"strconv"
 
 	"github.com/flashcatcloud/go-flashduty"
 	"github.com/spf13/cobra"
@@ -70,8 +71,8 @@ func newMonitQueryDiagnoseCmd() *cobra.Command {
 	cmd.Flags().StringVar(&dsType, "ds-type", "", "Datasource type: prometheus|victorialogs|loki|mysql (required)")
 	cmd.Flags().StringVar(&dsName, "ds-name", "", "Datasource name as configured (required)")
 	registerEnumFlag(cmd, "ds-type", "prometheus", "victorialogs", "loki", "mysql")
-	cmd.Flags().StringVar(&timeStart, "time-start", "15m", "Window start (relative '15m'/'1h', unix seconds, or 'now')")
-	cmd.Flags().StringVar(&timeEnd, "time-end", "now", "Window end (relative, unix seconds, or 'now'; span capped at 6h)")
+	cmd.Flags().StringVar(&timeStart, "time-start", "15m", "Window start: relative duration ('15m'/'1h'), 'now', a date/RFC3339 timestamp, or a unix epoch in seconds or milliseconds")
+	cmd.Flags().StringVar(&timeEnd, "time-end", "now", "Window end: same formats as --time-start; span capped at 6h")
 	cmd.Flags().StringVar(&inputQuery, "input-query", "", "Filter-only log query OR matrix PromQL (required)")
 	cmd.Flags().StringVar(&operation, "operation", "", "log_patterns or metric_trends (default inferred from ds-type)")
 	cmd.Flags().IntVar(&maxLogs, "max-logs", 0, "Max log lines scanned (default 10000, cap 50000)")
@@ -99,6 +100,9 @@ func newMonitQueryRowsCmd() *cobra.Command {
 			if err != nil {
 				return fmt.Errorf("invalid --args: %w", err)
 			}
+			if err := normalizeRawTimeArgs(dsType, argsMap); err != nil {
+				return err
+			}
 
 			return runCommand(cmd, args, func(ctx *RunContext) error {
 				input := &flashduty.QueryRowsRequest{
@@ -136,7 +140,35 @@ func newMonitQueryRowsCmd() *cobra.Command {
 	cmd.Flags().StringVar(&dsName, "ds-name", "", "Datasource name (required)")
 	registerEnumFlag(cmd, "ds-type", "prometheus", "victorialogs", "loki", "mysql")
 	cmd.Flags().StringVar(&expr, "expr", "", "Query expression (required)")
-	cmd.Flags().StringSliceVar(&argsKV, "args", nil, "Arg entries KEY=VALUE (repeatable; values must be strings per monit-query contract)")
+	cmd.Flags().StringSliceVar(&argsKV, "args", nil, "Arg entries KEY=VALUE (repeatable; values must be strings per monit-query contract). "+
+		"For loki/victorialogs raw mode, <ds-type>.start/<ds-type>.end accept a relative duration ('15m'), 'now', a date/RFC3339 timestamp, "+
+		"or a unix epoch in seconds or milliseconds — normalized to the form the datasource requires before sending")
 
 	return cmd
 }
+
+// normalizeRawTimeArgs rewrites the raw-mode time-window args of a
+// monit-query rows call (<ds-type>.start / <ds-type>.end) into the unix-
+// seconds form the server requires, accepting any format timeutil.Parse
+// understands (RFC3339, date/datetime, relative duration, unix seconds or
+// milliseconds). Loki and VictoriaLogs are the only ds-types whose raw mode
+// consumes these keys; other ds-types ignore args entirely, so nothing is
+// touched for them.
+func normalizeRawTimeArgs(dsType string, args map[string]string) error {
+	if dsType != "loki" && dsType != "victorialogs" {
+		return nil
+	}
+	for _, suffix := range []string{"start", "end"} {
+		key := dsType + "." + suffix
+		v, ok := args[key]
+		if !ok || v == "" {
+			continue
+		}
+		ts, err := timeutil.Parse(v)
+		if err != nil {
+			return fmt.Errorf("invalid --args %s=%s: %w", key, v, err)
+		}
+		args[key] = strconv.FormatInt(ts, 10)
+	}
+	return nil
+}

internal/cli/monit_query_test.go

@@ -3,8 +3,10 @@ package cli
 import (
 	"encoding/json"
 	"fmt"
+	"strconv"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestMonitQueryDiagnoseFlags(t *testing.T) {
@@ -290,6 +292,120 @@ func TestMonitQueryRowsRequiredFlags(t *testing.T) {
 	}
 }
 
+// --- normalizeRawTimeArgs --------------------------------------------------
+
+func TestNormalizeRawTimeArgsAcceptedFormats(t *testing.T) {
+	cases := []struct {
+		name  string
+		input string
+	}{
+		{"rfc3339 utc", "2026-08-11T09:40:00Z"},
+		{"rfc3339 offset", "2026-08-11T09:40:00+08:00"},
+		{"unix seconds", "1786497600"},
+		{"unix milliseconds", "1786497600000"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			args := map[string]string{"victorialogs.start": tc.input, "victorialogs.end": tc.input}
+			if err := normalizeRawTimeArgs("victorialogs", args); err != nil {
+				t.Fatalf("normalizeRawTimeArgs(%q): unexpected error: %v", tc.input, err)
+			}
+			for _, key := range []string{"victorialogs.start", "victorialogs.end"} {
+				if _, err := strconv.ParseInt(args[key], 10, 64); err != nil {
+					t.Errorf("%s: expected normalized unix-seconds string, got %q", key, args[key])
+				}
+			}
+		})
+	}
+}
+
+func TestNormalizeRawTimeArgsLokiPrefix(t *testing.T) {
+	args := map[string]string{"loki.start": "2026-08-11T09:40:00Z", "loki.end": "2026-08-11T10:05:00Z"}
+	if err := normalizeRawTimeArgs("loki", args); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	wantStart := strconv.FormatInt(time.Date(2026, 8, 11, 9, 40, 0, 0, time.UTC).Unix(), 10)
+	wantEnd := strconv.FormatInt(time.Date(2026, 8, 11, 10, 5, 0, 0, time.UTC).Unix(), 10)
+	if args["loki.start"] != wantStart || args["loki.end"] != wantEnd {
+		t.Errorf("unexpected normalized loki args: %#v, want start=%s end=%s", args, wantStart, wantEnd)
+	}
+}
+
+func TestNormalizeRawTimeArgsIgnoresOtherDsTypes(t *testing.T) {
+	args := map[string]string{"prometheus.start": "not-a-time"}
+	if err := normalizeRawTimeArgs("prometheus", args); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if args["prometheus.start"] != "not-a-time" {
+		t.Errorf("expected prometheus args untouched, got %#v", args)
+	}
+}
+
+func TestNormalizeRawTimeArgsIgnoresUnrelatedKeys(t *testing.T) {
+	args := map[string]string{"victorialogs.type": "raw", "victorialogs.timespan.value": "15"}
+	if err := normalizeRawTimeArgs("victorialogs", args); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if args["victorialogs.type"] != "raw" || args["victorialogs.timespan.value"] != "15" {
+		t.Errorf("expected unrelated args untouched, got %#v", args)
+	}
+}
+
+func TestNormalizeRawTimeArgsInvalidValue(t *testing.T) {
+	args := map[string]string{"victorialogs.start": "not-a-time"}
+	err := normalizeRawTimeArgs("victorialogs", args)
+	if err == nil {
+		t.Fatal("expected error for invalid victorialogs.start, got nil")
+	}
+	if !strings.Contains(err.Error(), "victorialogs.start") {
+		t.Errorf("expected error to mention victorialogs.start, got %q", err.Error())
+	}
+}
+
+// TestMonitQueryRowsRawModeNormalizesRFC3339 is the regression test for the
+// raw-vs-stats time format inconsistency: a raw-mode VictoriaLogs query given
+// RFC3339 --args timestamps must reach the server as the unix-seconds form
+// the raw query path requires.
+func TestMonitQueryRowsRawModeNormalizesRFC3339(t *testing.T) {
+	saveAndResetGlobals(t)
+	stub := newGFStub(t)
+	stub.data = []any{}
+
+	_, err := execCommand(
+		"monit-query", "rows",
+		"--ds-type", "victorialogs",
+		"--ds-name", "vl-prod",
+		"--expr", `{app="api"} |= "error"`,
+		"--args", "victorialogs.type=raw",
+		"--args", "victorialogs.start=2026-08-11T09:40:00Z",
+		"--args", "victorialogs.end=2026-08-11T10:05:00Z",
+	)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	body := stub.lastBody
+	argsSent, _ := body["args"].(map[string]any)
+	start, ok := argsSent["victorialogs.start"].(string)
+	if !ok {
+		t.Fatalf("expected victorialogs.start in request args, got %#v", argsSent)
+	}
+	if _, err := strconv.ParseInt(start, 10, 64); err != nil {
+		t.Errorf("expected victorialogs.start to be unix-seconds, got %q", start)
+	}
+	end, ok := argsSent["victorialogs.end"].(string)
+	if !ok {
+		t.Fatalf("expected victorialogs.end in request args, got %#v", argsSent)
+	}
+	if _, err := strconv.ParseInt(end, 10, 64); err != nil {
+		t.Errorf("expected victorialogs.end to be unix-seconds, got %q", end)
+	}
+	wantStart := time.Date(2026, 8, 11, 9, 40, 0, 0, time.UTC).Unix()
+	wantEnd := time.Date(2026, 8, 11, 10, 5, 0, 0, time.UTC).Unix()
+	if start != strconv.FormatInt(wantStart, 10) || end != strconv.FormatInt(wantEnd, 10) {
+		t.Errorf("expected start=%d end=%d, got start=%s end=%s", wantStart, wantEnd, start, end)
+	}
+}
+
 func TestMonitQueryRowsInvalidArgs(t *testing.T) {
 	saveAndResetGlobals(t)
 	stub := newGFStub(t)

internal/timeutil/parse.go

@@ -17,7 +17,9 @@ import (
 //   - Date: "2026-04-01" (parsed as local midnight)
 //   - Datetime: "2026-04-01 10:00:00" or "2026-04-01T10:00:00" (parsed as local time)
 //   - RFC3339 with timezone: "2026-04-01T10:00:00+08:00" / "...Z" (the format the SDK emits)
-//   - Unix timestamp: "1712000000" (passed through)
+//   - Unix timestamp in seconds: "1712000000" (passed through)
+//   - Unix timestamp in milliseconds: "1712000000000" (divided down to seconds;
+//     distinguished from seconds by magnitude — see msThreshold below)
 func Parse(s string) (int64, error) {
 	s = strings.TrimSpace(s)
 	if s == "" || s == "now" {
@@ -67,14 +69,26 @@ func Parse(s string) (int64, error) {
 		}
 	}
 
-	// Try unix timestamp
-	if ts, err := strconv.ParseInt(s, 10, 64); err == nil && ts > 1000000000 {
-		return ts, nil
+	// Try unix timestamp, seconds or milliseconds. A value at or above
+	// msThreshold (100 billion) would be seconds only for a date past the
+	// year 5138, so any realistic timestamp that large must be milliseconds.
+	if ts, err := strconv.ParseInt(s, 10, 64); err == nil {
+		switch {
+		case ts >= msThreshold:
+			return ts / 1000, nil
+		case ts > 1000000000:
+			return ts, nil
+		}
 	}
 
-	return 0, fmt.Errorf("unable to parse time %q: expected duration (24h), RFC3339 (2006-01-02T15:04:05Z07:00), date (2006-01-02), datetime (2006-01-02 15:04:05), or unix timestamp", s)
+	return 0, fmt.Errorf("unable to parse time %q: expected duration (24h), RFC3339 (2006-01-02T15:04:05Z07:00), date (2006-01-02), datetime (2006-01-02 15:04:05), or unix timestamp in seconds or milliseconds", s)
 }
 
+// msThreshold is the magnitude cutoff separating a unix timestamp in seconds
+// from one in milliseconds: 100,000,000,000 as seconds is the year 5138, so
+// any input at or above it is treated as milliseconds instead.
+const msThreshold = 100_000_000_000
+
 // expandDays converts day shorthand (e.g. "7d", "30d") to hours for time.ParseDuration.
 // If the string does not end with "d" or is not purely numeric before it, returns as-is.
 func expandDays(s string) string {

internal/timeutil/parse_test.go

@@ -194,6 +194,29 @@ func TestParse(t *testing.T) {
 			wantExact:  time.Date(2026, 5, 29, 14, 0, 0, 0, time.Local).Unix(),
 			exactMatch: true,
 		},
+		// 25. Unix timestamp in milliseconds (13 digits) → divided down to seconds
+		{
+			name:       "unix milliseconds 1712000000000",
+			input:      "1712000000000",
+			wantExact:  1712000000,
+			exactMatch: true,
+		},
+		// 26. Boundary at the seconds/milliseconds threshold: just below is
+		// still treated as seconds, however implausible a date that is.
+		{
+			name:       "boundary below ms threshold treated as seconds",
+			input:      "99999999999",
+			wantExact:  99999999999,
+			exactMatch: true,
+		},
+		// 27. Boundary at the seconds/milliseconds threshold: at or above is
+		// treated as milliseconds.
+		{
+			name:       "boundary at ms threshold treated as milliseconds",
+			input:      "100000000000",
+			wantExact:  100000000000 / 1000,
+			exactMatch: true,
+		},
 	}
 
 	for _, tc := range tests {