logstream/logsapi.go

package main

import (
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"time"
)

// GET /api/v1/logs: incremental reading of the stored logs for automation
// tools (n8n, scripts). Each answer gives the rows that follow a cursor,
// oldest first, and the cursor to send next time; following the cursors
// returns every row once, without gaps or duplicates.
//
// The cursor is a position on _time, which is the reception time given by
// LogStream (see Entry.Record), so new rows always come after it. Rows only
// count once they are older than the lag (30 s by default): a message can
// wait a little in the queue or for its DNS name before VictoriaLogs makes it
// visible. Several rows can share the same _time to the nanosecond; inside
// such a group the rows are ordered by a hash of their content, and the
// cursor remembers how many of them were already returned.

const (
	logsAPIDefaultLimit = 1000
	logsAPIMaxLimit     = 10000
	logsAPIDefaultLag   = 30 * time.Second
	logsAPIMaxLag       = time.Hour
	logsAPIDefaultSince = time.Hour
	logsAPIMaxGroup     = 100000 // rows with the very same _time
)

// logsCursor: every row with _time < T has been returned, plus the first K
// rows (in hash order) of those with _time == T. Q is a hash of the query:
// K only means something for the same query.
type logsCursor struct {
	T time.Time `json:"t"`
	K int       `json:"k,omitempty"`
	Q string    `json:"q,omitempty"`
}

func (c logsCursor) encode() string {
	b, _ := json.Marshal(c)
	return base64.RawURLEncoding.EncodeToString(b)
}

func decodeCursor(s string) (logsCursor, bool) {
	var c logsCursor
	b, err := base64.RawURLEncoding.DecodeString(s)
	if err != nil || json.Unmarshal(b, &c) != nil || c.T.IsZero() || c.K < 0 {
		return logsCursor{}, false
	}
	return c, true
}

// parseSince reads the since parameter: a cursor from a previous answer, an
// RFC 3339 time, or a duration back from now (90s, 15m, 6h, 7d). Empty: the
// last hour.
func parseSince(s string, now time.Time) (logsCursor, error) {
	s = strings.TrimSpace(s)
	if s == "" {
		return logsCursor{T: now.Add(-logsAPIDefaultSince)}, nil
	}
	if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
		return logsCursor{T: t}, nil
	}
	if d, ok := parseAge(s); ok {
		return logsCursor{T: now.Add(-d)}, nil
	}
	if c, ok := decodeCursor(s); ok {
		return c, nil
	}
	return logsCursor{}, &codedError{code: "bad_since", msg: "since: expected a next_cursor, an RFC 3339 time or a duration such as 15m, 6h or 7d"}
}

// parseAge reads a Go duration, plus "Nd" for days.
func parseAge(s string) (time.Duration, bool) {
	if n, ok := strings.CutSuffix(s, "d"); ok {
		days, err := strconv.Atoi(n)
		if err != nil || days < 0 {
			return 0, false
		}
		return time.Duration(days) * 24 * time.Hour, true
	}
	d, err := time.ParseDuration(s)
	return d, err == nil && d >= 0
}

type logsPage struct {
	Rows       []map[string]any `json:"rows"`
	NextCursor string           `json:"next_cursor"`
	HasMore    bool             `json:"has_more"`          // call again right away: more rows are ready
	Waiting    bool             `json:"waiting,omitempty"` // the disk buffer is still being sent to VictoriaLogs
	Until      string           `json:"until"`             // rows are returned up to this time (now - lag)
}

// GET /api/v1/logs?query=<LogsQL filter>&since=<cursor|time|duration>&limit=<n>&lag=<duration>
func (a *API) logsV1(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	filter := strings.TrimSpace(q.Get("query"))
	if f, pipes := splitPipes(filter); strings.TrimSpace(pipes) != "" {
		writeErr(w, http.StatusBadRequest, &codedError{code: "api_pipes", msg: "query: LogsQL pipes (| ...) are not supported, keep only the filter part"})
		return
	} else {
		filter = strings.TrimSpace(f)
	}
	if filter == "" {
		filter = "*"
	}
	limit := logsAPIDefaultLimit
	if v := q.Get("limit"); v != "" {
		n, err := strconv.Atoi(v)
		if err != nil || n <= 0 || n > logsAPIMaxLimit {
			writeErr(w, http.StatusBadRequest, &codedError{code: "bad_limit", msg: fmt.Sprintf("limit: expected 1 to %d", logsAPIMaxLimit)})
			return
		}
		limit = n
	}
	lag := logsAPIDefaultLag
	if v := q.Get("lag"); v != "" {
		d, ok := parseAge(v)
		if !ok || d > logsAPIMaxLag {
			writeErr(w, http.StatusBadRequest, &codedError{code: "bad_lag", msg: "lag: expected a duration from 0s to 1h"})
			return
		}
		lag = d
	}
	now := time.Now()
	cur, err := parseSince(q.Get("since"), now)
	if err != nil {
		writeErr(w, http.StatusBadRequest, err)
		return
	}
	qh := queryHash(filter)
	if cur.Q != qh {
		cur.K = 0 // the query changed: the rank inside the group no longer applies
	}
	cur.Q = qh

	// Logs waiting in the disk buffer were received earlier: returning newer
	// rows now would move the cursor past them.
	if a.store.PendingSpool() > 0 {
		writeJSON(w, http.StatusOK, logsPage{Rows: []map[string]any{}, NextCursor: cur.encode(), Waiting: true, Until: logsqlTime(cur.T)})
		return
	}
	until := now.Add(-lag)
	page, err := readLogsPage(r.Context(), a.store.Query, filter, cur, until, limit)
	if err != nil {
		writeErr(w, queryStatus(err), err)
		return
	}
	a.annotateHosts(page.Rows)
	writeJSON(w, http.StatusOK, page)
}

func queryHash(filter string) string {
	sum := sha256.Sum256([]byte(filter))
	return hex.EncodeToString(sum[:4])
}

type queryFunc func(ctx context.Context, query string) ([]map[string]any, error)

// readLogsPage returns at most limit rows after cur, up to until included.
func readLogsPage(ctx context.Context, run queryFunc, filter string, cur logsCursor, until time.Time, limit int) (logsPage, error) {
	page := logsPage{Rows: []map[string]any{}, Until: logsqlTime(until)}
	next := cur
	start := cur.T
	if cur.K > 0 {
		// Rest of the group the previous page stopped in.
		group, err := readGroup(ctx, run, filter, cur.T)
		if err != nil {
			return page, err
		}
		rest := group[min(cur.K, len(group)):]
		if len(rest) >= limit {
			page.Rows = rest[:limit]
			next.K = cur.K + limit
			page.HasMore = true
			page.NextCursor = next.encode()
			return page, nil
		}
		page.Rows = append(page.Rows, rest...)
		start = cur.T.Add(time.Nanosecond)
	}

	left := limit - len(page.Rows)
	var rows []map[string]any
	if !start.After(until) {
		q := fmt.Sprintf("_time:[%s, %s] (%s) | sort by (_time) | limit %d", logsqlTime(start), logsqlTime(until), filter, left+1)
		var err error
		if rows, err = run(ctx, q); err != nil {
			return page, err
		}
	}
	if len(rows) <= left {
		// Everything up to until has been read.
		page.Rows = append(page.Rows, rows...)
		next = logsCursor{T: until.Add(time.Nanosecond), Q: cur.Q}
		if next.T.Before(start) {
			next.T = start
		}
		page.NextCursor = next.encode()
		return page, nil
	}

	// One row too many: the group of the last time may be cut. Whole groups
	// before it are returned; that last group starts the next page.
	page.HasMore = true
	last, err := rowTime(rows[left])
	if err != nil {
		return page, err
	}
	n := 0
	for n < left {
		t, err := rowTime(rows[n])
		if err != nil {
			return page, err
		}
		if !t.Before(last) {
			break
		}
		n++
	}
	if n > 0 || len(page.Rows) > 0 {
		page.Rows = append(page.Rows, rows[:n]...)
		page.NextCursor = logsCursor{T: last, Q: cur.Q}.encode()
		return page, nil
	}
	// The whole page is a single group: read it all to cut it in a stable order.
	group, err := readGroup(ctx, run, filter, last)
	if err != nil {
		return page, err
	}
	page.Rows = group[:min(left, len(group))]
	page.NextCursor = logsCursor{T: last, K: len(page.Rows), Q: cur.Q}.encode()
	return page, nil
}

// readGroup returns the rows whose _time is exactly t, in a stable order.
func readGroup(ctx context.Context, run queryFunc, filter string, t time.Time) ([]map[string]any, error) {
	ts := logsqlTime(t)
	rows, err := run(ctx, fmt.Sprintf("_time:[%s, %s] (%s) | limit %d", ts, ts, filter, logsAPIMaxGroup+1))
	if err != nil {
		return nil, err
	}
	if len(rows) > logsAPIMaxGroup {
		return nil, fmt.Errorf("more than %d logs share the time %s", logsAPIMaxGroup, ts)
	}
	keys := make([]string, len(rows))
	for i, row := range rows {
		b, _ := json.Marshal(row) // map keys are sorted: same row, same bytes
		sum := sha256.Sum256(b)
		keys[i] = string(sum[:])
	}
	idx := make([]int, len(rows))
	for i := range idx {
		idx[i] = i
	}
	sort.SliceStable(idx, func(i, j int) bool { return keys[idx[i]] < keys[idx[j]] })
	out := make([]map[string]any, len(rows))
	for i, k := range idx {
		out[i] = rows[k]
	}
	return out, nil
}

func rowTime(row map[string]any) (time.Time, error) {
	s, _ := row["_time"].(string)
	t, err := time.Parse(time.RFC3339Nano, s)
	if err != nil {
		return time.Time{}, errors.New("VictoriaLogs returned a row without a valid _time")
	}
	return t, nil
}