Reading API /api/v1/logs with a stable cursor, and API tokens
GET /api/v1/logs?query=<LogsQL>&since=<cursor|time|duration>&limit=<n> returns the logs after a cursor, oldest first, with next_cursor and has_more. The cursor is a position on _time (the reception time) plus the number of rows already returned among those sharing that exact time, ordered by a hash of their content, so following the cursors gives every row once even within groups cut by a page. Rows count once older than the lag (30 s), and nothing is returned while the disk spool still holds logs. API tokens (Settings > API) give read-only access to /api/v1/ as "Authorization: Bearer lst_...", in every auth mode; only their SHA-256 is kept in /data/api_tokens.json. README (EN/FR) documents the route and an n8n workflow keeping the cursor in $getWorkflowStaticData. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
cedricandClaude Opus 5.5 committed
1 parent
d61384ed2c
commit
c8f45044a3
12 files changed
+1327
-7
No files matched your search
+276
@@ -0,0 +1,276 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"math/rand"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// cursorVL plays VictoriaLogs for the two queries of /api/v1/logs: a time
|
||||
// range sorted by _time, and an exact time. Like VictoriaLogs, it gives no
|
||||
// order among rows of the same _time: they are shuffled on every query.
|
||||
type cursorVL struct {
|
||||
mu sync.Mutex
|
||||
rows []map[string]any
|
||||
rnd *rand.Rand
|
||||
queries []string
|
||||
}
|
||||
|
||||
var cursorQuery = regexp.MustCompile(`^_time:\[([^,]+), ([^\]]+)] \((.*?)\)( \| sort by \(_time\))? \| limit (\d+)$`)
|
||||
|
||||
func (v *cursorVL) add(t time.Time, msg string) {
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
v.rows = append(v.rows, map[string]any{"_time": t.UTC().Format(time.RFC3339Nano), "_msg": msg, "host": "h1"})
|
||||
}
|
||||
|
||||
func (v *cursorVL) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
_ = r.ParseForm()
|
||||
q := r.PostForm.Get("query")
|
||||
m := cursorQuery.FindStringSubmatch(q)
|
||||
if m == nil {
|
||||
return &http.Response{StatusCode: http.StatusBadRequest, Body: io.NopCloser(strings.NewReader("unexpected query " + q)), Header: http.Header{}}, nil
|
||||
}
|
||||
from, _ := time.Parse(time.RFC3339Nano, m[1])
|
||||
to, _ := time.Parse(time.RFC3339Nano, m[2])
|
||||
limit, _ := strconv.Atoi(m[5])
|
||||
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
v.queries = append(v.queries, q)
|
||||
var out []map[string]any
|
||||
for _, row := range v.rows {
|
||||
t, _ := time.Parse(time.RFC3339Nano, row["_time"].(string))
|
||||
if !t.Before(from) && !t.After(to) {
|
||||
out = append(out, row)
|
||||
}
|
||||
}
|
||||
v.rnd.Shuffle(len(out), func(i, j int) { out[i], out[j] = out[j], out[i] })
|
||||
if m[4] != "" {
|
||||
sort.SliceStable(out, func(i, j int) bool {
|
||||
a, _ := time.Parse(time.RFC3339Nano, out[i]["_time"].(string))
|
||||
b, _ := time.Parse(time.RFC3339Nano, out[j]["_time"].(string))
|
||||
return a.Before(b)
|
||||
})
|
||||
}
|
||||
if len(out) > limit {
|
||||
out = out[:limit]
|
||||
}
|
||||
var b strings.Builder
|
||||
for _, row := range out {
|
||||
line, _ := json.Marshal(row)
|
||||
b.Write(line)
|
||||
b.WriteByte('\n')
|
||||
}
|
||||
return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(b.String())), Header: http.Header{}}, nil
|
||||
}
|
||||
|
||||
func newCursorVL(seed int64) (*cursorVL, *Store) {
|
||||
vl := &cursorVL{rnd: rand.New(rand.NewSource(seed))}
|
||||
store := NewStore("http://vl", 10, 10, time.Second, nil)
|
||||
store.client.Transport = vl
|
||||
store.streamClient.Transport = vl
|
||||
return vl, store
|
||||
}
|
||||
|
||||
// fill adds rows by groups of the same _time, some groups larger than a
|
||||
// page, some with identical rows, and returns the multiset of the rows.
|
||||
func fill(vl *cursorVL, start time.Time, groups []int, tag string) map[string]int {
|
||||
want := map[string]int{}
|
||||
t := start
|
||||
for g, size := range groups {
|
||||
for i := 0; i < size; i++ {
|
||||
msg := fmt.Sprintf("%s g%d r%d", tag, g, i%4) // r0..r3 repeat: identical rows
|
||||
vl.add(t, msg)
|
||||
want[rowKey(t, msg)]++
|
||||
}
|
||||
t = t.Add(time.Duration(1+g%3) * time.Microsecond)
|
||||
}
|
||||
return want
|
||||
}
|
||||
|
||||
func rowKey(t time.Time, msg string) string { return t.UTC().Format(time.RFC3339Nano) + " " + msg }
|
||||
|
||||
func TestLogsCursorNoLossNoDuplicate(t *testing.T) {
|
||||
base := time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC)
|
||||
groups := []int{1, 3, 1, 1, 7, 2, 1, 12, 1, 4, 4, 1, 25, 1, 2}
|
||||
for _, limit := range []int{1, 2, 3, 5, 8, 1000} {
|
||||
for seed := int64(1); seed <= 5; seed++ {
|
||||
vl, store := newCursorVL(seed)
|
||||
want := fill(vl, base, groups, "a")
|
||||
got := map[string]int{}
|
||||
cur := logsCursor{T: base.Add(-time.Minute)}
|
||||
until := base.Add(time.Second)
|
||||
var last time.Time
|
||||
for calls := 0; ; calls++ {
|
||||
if calls > 500 {
|
||||
t.Fatalf("limit %d: no end", limit)
|
||||
}
|
||||
page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, limit)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(page.Rows) > limit {
|
||||
t.Fatalf("limit %d: page of %d rows", limit, len(page.Rows))
|
||||
}
|
||||
for _, row := range page.Rows {
|
||||
rt, _ := rowTime(row)
|
||||
if rt.Before(last) {
|
||||
t.Fatalf("limit %d: rows out of order (%s after %s)", limit, rt, last)
|
||||
}
|
||||
last = rt
|
||||
got[rowKey(rt, row["_msg"].(string))]++
|
||||
}
|
||||
var ok bool
|
||||
if cur, ok = decodeCursor(page.NextCursor); !ok {
|
||||
t.Fatalf("bad cursor %q", page.NextCursor)
|
||||
}
|
||||
if !page.HasMore {
|
||||
break
|
||||
}
|
||||
if len(page.Rows) == 0 {
|
||||
t.Fatalf("limit %d: has_more with an empty page", limit)
|
||||
}
|
||||
}
|
||||
if fmt.Sprint(got) != fmt.Sprint(want) {
|
||||
t.Fatalf("limit %d seed %d:\n got %v\nwant %v", limit, seed, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// New logs arrive between the calls, as with a scheduled n8n workflow.
|
||||
func TestLogsCursorFollowsNewLogs(t *testing.T) {
|
||||
vl, store := newCursorVL(7)
|
||||
base := time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC)
|
||||
want := map[string]int{}
|
||||
got := map[string]int{}
|
||||
cur := logsCursor{T: base}
|
||||
for round := 0; round < 6; round++ {
|
||||
start := base.Add(time.Duration(round) * time.Second)
|
||||
for k, n := range fill(vl, start, []int{2, 5, 1, 3}, fmt.Sprintf("round%d", round)) {
|
||||
want[k] += n
|
||||
}
|
||||
until := start.Add(500 * time.Millisecond) // after this round's rows
|
||||
for {
|
||||
page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, 4)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, row := range page.Rows {
|
||||
rt, _ := rowTime(row)
|
||||
got[rowKey(rt, row["_msg"].(string))]++
|
||||
}
|
||||
cur, _ = decodeCursor(page.NextCursor)
|
||||
if !page.HasMore {
|
||||
break
|
||||
}
|
||||
}
|
||||
// Polling again with nothing new returns nothing.
|
||||
page, err := readLogsPage(context.Background(), store.Query, "*", cur, until, 4)
|
||||
if err != nil || len(page.Rows) != 0 || page.HasMore {
|
||||
t.Fatalf("round %d: second poll returned %d rows (has_more %v, err %v)", round, len(page.Rows), page.HasMore, err)
|
||||
}
|
||||
}
|
||||
if fmt.Sprint(got) != fmt.Sprint(want) {
|
||||
t.Fatalf("got %v\nwant %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSince(t *testing.T) {
|
||||
now := time.Date(2026, 11, 1, 12, 0, 0, 0, time.UTC)
|
||||
c := logsCursor{T: now.Add(-time.Minute), K: 3, Q: "abcd"}
|
||||
for _, tc := range []struct {
|
||||
in string
|
||||
want logsCursor
|
||||
}{
|
||||
{"", logsCursor{T: now.Add(-time.Hour)}},
|
||||
{"15m", logsCursor{T: now.Add(-15 * time.Minute)}},
|
||||
{"7d", logsCursor{T: now.Add(-7 * 24 * time.Hour)}},
|
||||
{"2026-11-01T08:30:00.5Z", logsCursor{T: time.Date(2026, 11, 1, 8, 30, 0, 5e8, time.UTC)}},
|
||||
{c.encode(), c},
|
||||
} {
|
||||
got, err := parseSince(tc.in, now)
|
||||
if err != nil || !got.T.Equal(tc.want.T) || got.K != tc.want.K || got.Q != tc.want.Q {
|
||||
t.Errorf("parseSince(%q) = %+v, %v; want %+v", tc.in, got, err, tc.want)
|
||||
}
|
||||
}
|
||||
for _, bad := range []string{"yesterday", "-5m", "e30"} {
|
||||
if _, err := parseSince(bad, now); err == nil {
|
||||
t.Errorf("parseSince(%q): expected an error", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLogsV1Handler(t *testing.T) {
|
||||
vl, store := newCursorVL(3)
|
||||
a := &API{store: store, rdns: NewReverseDNS(false, "")}
|
||||
old := time.Now().Add(-10 * time.Minute)
|
||||
fill(vl, old, []int{2, 1}, "x")
|
||||
vl.add(time.Now().Add(time.Hour), "in the future: after until")
|
||||
|
||||
get := func(url string) (int, logsPage, map[string]string) {
|
||||
rec := httptest.NewRecorder()
|
||||
a.logsV1(rec, httptest.NewRequest("GET", url, nil))
|
||||
var page logsPage
|
||||
var e map[string]string
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &page)
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &e)
|
||||
return rec.Code, page, e
|
||||
}
|
||||
code, page, _ := get("/api/v1/logs?query=host:h1&since=1h")
|
||||
if code != http.StatusOK || len(page.Rows) != 3 || page.HasMore || page.NextCursor == "" {
|
||||
t.Fatalf("status %d, %d rows, has_more %v", code, len(page.Rows), page.HasMore)
|
||||
}
|
||||
if q := vl.queries[len(vl.queries)-1]; !strings.Contains(q, "(host:h1)") {
|
||||
t.Fatalf("query filter not applied: %s", q)
|
||||
}
|
||||
code, page, _ = get("/api/v1/logs?query=host:h1&since=" + page.NextCursor)
|
||||
if code != http.StatusOK || len(page.Rows) != 0 {
|
||||
t.Fatalf("second call: status %d, %d rows", code, len(page.Rows))
|
||||
}
|
||||
for url, wantCode := range map[string]string{
|
||||
"/api/v1/logs?query=" + "*%20|%20stats%20count()": "api_pipes",
|
||||
"/api/v1/logs?limit=0": "bad_limit",
|
||||
"/api/v1/logs?limit=20000": "bad_limit",
|
||||
"/api/v1/logs?lag=2h": "bad_lag",
|
||||
"/api/v1/logs?since=nope": "bad_since",
|
||||
} {
|
||||
if code, _, e := get(url); code != http.StatusBadRequest || e["code"] != wantCode {
|
||||
t.Errorf("%s: status %d, %v; want 400 %s", url, code, e, wantCode)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLogsV1WaitsForSpool(t *testing.T) {
|
||||
spool, err := OpenSpool(t.TempDir(), 1<<20)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := spool.Write([]byte(`{"_msg":"x"}`+"\n"), 1); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
vl, _ := newCursorVL(1)
|
||||
store := NewStore("http://vl", 10, 10, time.Second, spool)
|
||||
store.streamClient.Transport = vl
|
||||
a := &API{store: store}
|
||||
rec := httptest.NewRecorder()
|
||||
a.logsV1(rec, httptest.NewRequest("GET", "/api/v1/logs?since=2026-11-01T10:00:00Z", nil))
|
||||
var page logsPage
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &page)
|
||||
c, _ := decodeCursor(page.NextCursor)
|
||||
if rec.Code != http.StatusOK || !page.Waiting || len(page.Rows) != 0 || !c.T.Equal(time.Date(2026, 11, 1, 10, 0, 0, 0, time.UTC)) || len(vl.queries) != 0 {
|
||||
t.Fatalf("status %d, waiting %v, %d rows, cursor %v, %d queries", rec.Code, page.Waiting, len(page.Rows), c.T, len(vl.queries))
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user