Add logviewer: web UI for the ClickHouse log store
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/push/build Pipeline was successful

Single Go binary with embedded Bootstrap 3 + jQuery UI, querying logs.raw
over the ClickHouse HTTP interface as the readonly logreader user. Runs
behind oauth2-proxy; the app does no auth itself. Server-side enforced
time bounds (15m default, 72h max), parameterized queries, raw-SQL WHERE
fragment wrapped with enforced bounds and LIMIT, tail polling with a
clamped cursor, facets, healthz. Woodpecker build/test plus tag-driven
image push to artifactapi docker-internal.
This commit is contained in:
2026-08-23 16:39:22 +10:00
parent a81d3bb22a
commit abc81e60c8
19 changed files with 1587 additions and 1 deletions
+91
View File
@@ -0,0 +1,91 @@
package server
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"strconv"
"strings"
"time"
)
// maxExecutionSeconds bounds every ClickHouse query server-side.
const maxExecutionSeconds = 30
type chClient struct {
url string
user string
password string
http *http.Client
}
func newCHClient(chURL, user, password string) *chClient {
return &chClient{
url: strings.TrimRight(chURL, "/"),
user: user,
password: password,
http: &http.Client{Timeout: (maxExecutionSeconds + 5) * time.Second},
}
}
type chColumn struct {
Name string `json:"name"`
Type string `json:"type"`
}
type chResult struct {
Meta []chColumn `json:"meta"`
Data []map[string]any `json:"data"`
Rows int `json:"rows"`
}
// query POSTs sql to the ClickHouse HTTP interface. Every value in params is
// sent as a bound query parameter (param_<name>), never interpolated.
func (c *chClient) query(ctx context.Context, sql string, params map[string]string) (*chResult, error) {
q := url.Values{}
q.Set("max_execution_time", strconv.Itoa(maxExecutionSeconds))
q.Set("output_format_json_quote_64bit_integers", "0")
for k, v := range params {
q.Set("param_"+k, v)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url+"/?"+q.Encode(), strings.NewReader(sql))
if err != nil {
return nil, err
}
req.Header.Set("X-ClickHouse-User", c.user)
req.Header.Set("X-ClickHouse-Key", c.password)
req.Header.Set("Content-Type", "text/plain")
resp, err := c.http.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, 32<<20))
if err != nil {
return nil, err
}
if resp.StatusCode != http.StatusOK {
msg := strings.TrimSpace(string(body))
if len(msg) > 500 {
msg = msg[:500]
}
return nil, fmt.Errorf("clickhouse: %s: %s", resp.Status, msg)
}
var res chResult
if err := json.Unmarshal(body, &res); err != nil {
return nil, fmt.Errorf("clickhouse: decode response: %w", err)
}
return &res, nil
}
func (c *chClient) ping(ctx context.Context) error {
_, err := c.query(ctx, "SELECT 1 FORMAT JSON", nil)
return err
}
+218
View File
@@ -0,0 +1,218 @@
package server
import (
"fmt"
"net/url"
"strconv"
"strings"
"time"
)
const (
defaultWindow = 15 * time.Minute
// maxWindow matches the table TTL; anything wider scans nothing extra
// but signals a client bug, so reject it.
maxWindow = 72 * time.Hour
defaultLimit = 100
maxLimit = 1000
tailLimit = 500
)
// filterColumns maps API query parameters to the column expression they filter
// on. Only these names are ever accepted; values are always bound parameters.
var filterColumns = []struct {
param string
column string
}{
{"namespace", "namespace"},
{"host", "host"},
{"pod", "pod"},
{"container", "container"},
{"app", "labels['app']"},
{"severity", "severity"},
{"stream", "stream"},
{"source", "source"},
}
const selectColumns = "timestamp, toUnixTimestamp64Milli(timestamp) AS ts_ms, host, source, namespace, pod, container, stream, severity, message, labels, fields"
type timeRange struct {
since time.Time
until time.Time
}
// parseTimeRange enforces server-side time bounds: missing bounds default to a
// 15m window ending now, and windows wider than maxWindow are rejected so a
// query can never run unbounded over the store.
func parseTimeRange(sinceStr, untilStr string, now time.Time) (timeRange, error) {
until := now
if untilStr != "" {
t, err := parseTime(untilStr, now)
if err != nil {
return timeRange{}, fmt.Errorf("invalid until: %w", err)
}
until = t
}
since := until.Add(-defaultWindow)
if sinceStr != "" {
t, err := parseTime(sinceStr, now)
if err != nil {
return timeRange{}, fmt.Errorf("invalid since: %w", err)
}
since = t
}
if !since.Before(until) {
return timeRange{}, fmt.Errorf("since (%s) must be before until (%s)", since.UTC().Format(time.RFC3339), until.UTC().Format(time.RFC3339))
}
if until.Sub(since) > maxWindow {
return timeRange{}, fmt.Errorf("time window %s exceeds maximum %s", until.Sub(since), maxWindow)
}
return timeRange{since: since, until: until}, nil
}
// parseTime accepts RFC3339, "2006-01-02 15:04:05", unix seconds/millis, and
// relative durations like "15m" / "6h" / "1d" (meaning that long ago).
func parseTime(s string, now time.Time) (time.Time, error) {
if t, err := time.Parse(time.RFC3339, s); err == nil {
return t, nil
}
if t, err := time.Parse("2006-01-02 15:04:05", s); err == nil {
return t.UTC(), nil
}
if n, err := strconv.ParseInt(s, 10, 64); err == nil {
if n > 1e12 {
return time.UnixMilli(n).UTC(), nil
}
return time.Unix(n, 0).UTC(), nil
}
if d, err := parseDuration(strings.TrimPrefix(s, "-")); err == nil {
return now.Add(-d), nil
}
return time.Time{}, fmt.Errorf("unrecognised time %q", s)
}
func parseDuration(s string) (time.Duration, error) {
if strings.HasSuffix(s, "d") {
n, err := strconv.Atoi(strings.TrimSuffix(s, "d"))
if err != nil {
return 0, err
}
return time.Duration(n) * 24 * time.Hour, nil
}
return time.ParseDuration(s)
}
func parseLimit(s string, def int) (int, error) {
if s == "" {
return def, nil
}
n, err := strconv.Atoi(s)
if err != nil || n < 1 {
return 0, fmt.Errorf("invalid limit %q", s)
}
if n > maxLimit {
n = maxLimit
}
return n, nil
}
func parseOffset(s string) (int, error) {
if s == "" {
return 0, nil
}
n, err := strconv.Atoi(s)
if err != nil || n < 0 {
return 0, fmt.Errorf("invalid offset %q", s)
}
return n, nil
}
// validateSQLFragment gates the raw WHERE fragment. Real safety comes from the
// readonly ClickHouse user and the parameterized outer query; this only blocks
// statement separators.
func validateSQLFragment(frag string) error {
if strings.Contains(frag, ";") {
return fmt.Errorf("sql fragment must not contain ';'")
}
return nil
}
// encodeStringArray renders a ClickHouse Array(String) parameter value.
func encodeStringArray(items []string) string {
var b strings.Builder
b.WriteByte('[')
for i, it := range items {
if i > 0 {
b.WriteByte(',')
}
b.WriteByte('\'')
b.WriteString(strings.ReplaceAll(strings.ReplaceAll(it, `\`, `\\`), `'`, `\'`))
b.WriteByte('\'')
}
b.WriteByte(']')
return b.String()
}
// whereClause builds the parameterized WHERE clause shared by the endpoints.
// The time bounds are always present and always bound parameters, so no
// filter or raw sql fragment can widen the scanned range.
func whereClause(tr timeRange, form url.Values, params map[string]string, exclusiveSince bool) (string, error) {
var conds []string
params["since_ms"] = strconv.FormatInt(tr.since.UnixMilli(), 10)
params["until_ms"] = strconv.FormatInt(tr.until.UnixMilli(), 10)
sinceOp := ">="
if exclusiveSince {
sinceOp = ">"
}
conds = append(conds,
"timestamp "+sinceOp+" fromUnixTimestamp64Milli({since_ms:Int64})",
"timestamp < fromUnixTimestamp64Milli({until_ms:Int64})",
)
for _, f := range filterColumns {
if v := form.Get(f.param); v != "" {
p := "f_" + f.param
params[p] = v
conds = append(conds, fmt.Sprintf("%s = {%s:String}", f.column, p))
}
}
if q := strings.TrimSpace(form.Get("q")); q != "" {
tokens := strings.Fields(q)
params["q_tokens"] = encodeStringArray(tokens)
conds = append(conds, "arrayAll(t -> positionCaseInsensitive(message, t) > 0, {q_tokens:Array(String)})")
}
if frag := strings.TrimSpace(form.Get("sql")); frag != "" {
if err := validateSQLFragment(frag); err != nil {
return "", err
}
conds = append(conds, "( "+frag+" )")
}
return strings.Join(conds, "\n AND "), nil
}
func buildQuerySQL(where string, limit, offset int) string {
return fmt.Sprintf(
"SELECT %s\nFROM logs.raw\nWHERE %s\nORDER BY timestamp DESC\nLIMIT %d OFFSET %d\nFORMAT JSON",
selectColumns, where, limit, offset)
}
func buildTailSQL(where string) string {
return fmt.Sprintf(
"SELECT %s\nFROM logs.raw\nWHERE %s\nORDER BY timestamp ASC\nLIMIT %d\nFORMAT JSON",
selectColumns, where, tailLimit)
}
func buildFacetsSQL(where string) string {
facet := func(name, expr string) string {
return fmt.Sprintf(
"SELECT * FROM (SELECT '%s' AS facet, %s AS value, count() AS n FROM logs.raw WHERE %s GROUP BY value ORDER BY n DESC LIMIT 20)",
name, expr, where)
}
return facet("namespace", "namespace") +
"\nUNION ALL\n" + facet("app", "labels['app']") +
"\nUNION ALL\n" + facet("host", "host") +
"\nFORMAT JSON"
}
+218
View File
@@ -0,0 +1,218 @@
package server
import (
"encoding/json"
"log"
"net/http"
"strconv"
"time"
"git.unkin.net/unkin/logviewer/web"
)
type Config struct {
CHURL string
CHUser string
CHPassword string
Version string
}
type Server struct {
ch *chClient
version string
now func() time.Time
mux *http.ServeMux
}
func New(cfg Config) *Server {
s := &Server{
ch: newCHClient(cfg.CHURL, cfg.CHUser, cfg.CHPassword),
version: cfg.Version,
now: time.Now,
}
s.routes()
return s
}
func (s *Server) routes() {
s.mux = http.NewServeMux()
s.mux.HandleFunc("GET /api/query", s.handleQuery)
s.mux.HandleFunc("GET /api/tail", s.handleTail)
s.mux.HandleFunc("GET /api/facets", s.handleFacets)
s.mux.HandleFunc("GET /healthz", s.handleHealthz)
s.mux.Handle("GET /", http.FileServerFS(web.FS))
}
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
s.mux.ServeHTTP(w, r)
}
func writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
if err := json.NewEncoder(w).Encode(v); err != nil {
log.Printf("write response: %v", err)
}
}
func writeError(w http.ResponseWriter, status int, err error) {
writeJSON(w, status, map[string]string{"error": err.Error()})
}
func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
form := r.URL.Query()
tr, err := parseTimeRange(form.Get("since"), form.Get("until"), s.now())
if err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
limit, err := parseLimit(form.Get("limit"), defaultLimit)
if err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
offset, err := parseOffset(form.Get("offset"))
if err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
params := map[string]string{}
where, err := whereClause(tr, form, params, false)
if err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
res, err := s.ch.query(r.Context(), buildQuerySQL(where, limit, offset), params)
if err != nil {
writeError(w, http.StatusBadGateway, err)
return
}
writeJSON(w, http.StatusOK, map[string]any{
"rows": res.Data,
"count": len(res.Data),
"since": tr.since.UTC().Format(time.RFC3339Nano),
"until": tr.until.UTC().Format(time.RFC3339Nano),
"limit": limit,
"offset": offset,
})
}
func (s *Server) handleTail(w http.ResponseWriter, r *http.Request) {
form := r.URL.Query()
now := s.now()
// The cursor is the exclusive lower bound in ms. It is clamped to the
// default window so a stale client can never trigger a wide scan.
cursor := now.Add(-10 * time.Second).UnixMilli()
if c := form.Get("cursor"); c != "" {
n, err := strconv.ParseInt(c, 10, 64)
if err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
cursor = n
}
if min := now.Add(-defaultWindow).UnixMilli(); cursor < min {
cursor = min
}
// The tail bound is exclusive of the cursor so rows never repeat.
tr := timeRange{since: time.UnixMilli(cursor), until: now.Add(time.Second)}
params := map[string]string{}
where, err := whereClause(tr, form, params, true)
if err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
res, err := s.ch.query(r.Context(), buildTailSQL(where), params)
if err != nil {
writeError(w, http.StatusBadGateway, err)
return
}
next := cursor
for _, row := range res.Data {
if ts, ok := numField(row, "ts_ms"); ok && ts > next {
next = ts
}
}
writeJSON(w, http.StatusOK, map[string]any{
"rows": res.Data,
"count": len(res.Data),
"cursor": next,
})
}
// numField reads an int64-ish field that ClickHouse may return as a JSON
// number or (with 64-bit quoting) a string.
func numField(row map[string]any, key string) (int64, bool) {
switch v := row[key].(type) {
case float64:
return int64(v), true
case string:
n, err := strconv.ParseInt(v, 10, 64)
return n, err == nil
case json.Number:
n, err := v.Int64()
return n, err == nil
}
return 0, false
}
func (s *Server) handleFacets(w http.ResponseWriter, r *http.Request) {
form := r.URL.Query()
tr, err := parseTimeRange(form.Get("since"), form.Get("until"), s.now())
if err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
params := map[string]string{}
// Facets only take the time window; drop filters so dropdowns always
// show the full set for the window.
where, err := whereClause(tr, nil, params, false)
if err != nil {
writeError(w, http.StatusBadRequest, err)
return
}
res, err := s.ch.query(r.Context(), buildFacetsSQL(where), params)
if err != nil {
writeError(w, http.StatusBadGateway, err)
return
}
out := map[string][]map[string]any{
"namespaces": {},
"apps": {},
"hosts": {},
}
keys := map[string]string{"namespace": "namespaces", "app": "apps", "host": "hosts"}
for _, row := range res.Data {
facet, _ := row["facet"].(string)
key, ok := keys[facet]
if !ok {
continue
}
n, _ := numField(row, "n")
out[key] = append(out[key], map[string]any{"value": row["value"], "count": n})
}
writeJSON(w, http.StatusOK, out)
}
func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) {
if err := s.ch.ping(r.Context()); err != nil {
writeError(w, http.StatusServiceUnavailable, err)
return
}
writeJSON(w, http.StatusOK, map[string]string{"status": "ok", "version": s.version})
}
+396
View File
@@ -0,0 +1,396 @@
package server
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"net/url"
"strconv"
"strings"
"testing"
"time"
)
var testNow = time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC)
// mockCH captures the SQL body and bound parameters of each ClickHouse HTTP
// request and returns a canned FORMAT JSON response.
type mockCH struct {
*httptest.Server
lastSQL string
lastParams url.Values
respond func() (int, string)
}
func newMockCH(t *testing.T) *mockCH {
t.Helper()
m := &mockCH{respond: func() (int, string) { return 200, `{"meta":[],"data":[],"rows":0}` }}
m.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
m.lastSQL = string(body)
m.lastParams = r.URL.Query()
code, resp := m.respond()
w.WriteHeader(code)
w.Write([]byte(resp))
}))
t.Cleanup(m.Close)
return m
}
func newTestServer(t *testing.T) (*Server, *mockCH) {
t.Helper()
ch := newMockCH(t)
s := New(Config{CHURL: ch.URL, CHUser: "logreader", CHPassword: "x", Version: "test"})
s.now = func() time.Time { return testNow }
return s, ch
}
func get(t *testing.T, s *Server, path string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodGet, path, nil)
rec := httptest.NewRecorder()
s.ServeHTTP(rec, req)
return rec
}
func boundMS(t *testing.T, params url.Values, name string) int64 {
t.Helper()
n, err := strconv.ParseInt(params.Get("param_"+name), 10, 64)
if err != nil {
t.Fatalf("param %s not an int64: %v", name, err)
}
return n
}
// --- time bound enforcement ---
func TestQueryDefaultWindow(t *testing.T) {
s, ch := newTestServer(t)
rec := get(t, s, "/api/query")
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
since := boundMS(t, ch.lastParams, "since_ms")
until := boundMS(t, ch.lastParams, "until_ms")
if until != testNow.UnixMilli() {
t.Errorf("until = %d, want now (%d)", until, testNow.UnixMilli())
}
if until-since != defaultWindow.Milliseconds() {
t.Errorf("window = %dms, want default %dms", until-since, defaultWindow.Milliseconds())
}
for _, pred := range []string{
"timestamp >= fromUnixTimestamp64Milli({since_ms:Int64})",
"timestamp < fromUnixTimestamp64Milli({until_ms:Int64})",
} {
if !strings.Contains(ch.lastSQL, pred) {
t.Errorf("SQL missing enforced bound %q:\n%s", pred, ch.lastSQL)
}
}
}
func TestQueryExplicitWindow(t *testing.T) {
s, ch := newTestServer(t)
rec := get(t, s, "/api/query?since=2026-08-23T10:00:00Z&until=2026-08-23T11:00:00Z")
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
since := boundMS(t, ch.lastParams, "since_ms")
until := boundMS(t, ch.lastParams, "until_ms")
if until-since != time.Hour.Milliseconds() {
t.Errorf("window = %dms, want 1h", until-since)
}
}
func TestQueryRelativeSince(t *testing.T) {
s, ch := newTestServer(t)
if rec := get(t, s, "/api/query?since=6h"); rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
since := boundMS(t, ch.lastParams, "since_ms")
if want := testNow.Add(-6 * time.Hour).UnixMilli(); since != want {
t.Errorf("since = %d, want %d", since, want)
}
}
func TestQueryRejectsUnboundedWindow(t *testing.T) {
s, _ := newTestServer(t)
cases := []string{
"/api/query?since=1970-01-01T00:00:00Z", // wider than maxWindow
"/api/query?since=2026-08-01T00:00:00Z&until=30d", // nonsense + wide
"/api/query?since=0", // epoch
}
for _, c := range cases {
if rec := get(t, s, c); rec.Code != http.StatusBadRequest {
t.Errorf("%s: status %d, want 400 (body %s)", c, rec.Code, rec.Body)
}
}
}
func TestQueryRejectsInvertedAndBadTimes(t *testing.T) {
s, _ := newTestServer(t)
cases := []string{
"/api/query?since=2026-08-23T11:00:00Z&until=2026-08-23T10:00:00Z",
"/api/query?since=yesterdayish",
"/api/query?until=not-a-time",
}
for _, c := range cases {
if rec := get(t, s, c); rec.Code != http.StatusBadRequest {
t.Errorf("%s: status %d, want 400", c, rec.Code)
}
}
}
// --- filters and fuzzy search ---
func TestQueryFilterParamsAreBound(t *testing.T) {
s, ch := newTestServer(t)
rec := get(t, s, "/api/query?namespace=prod&app=encapi&severity=error")
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
if got := ch.lastParams.Get("param_f_namespace"); got != "prod" {
t.Errorf("param_f_namespace = %q", got)
}
if got := ch.lastParams.Get("param_f_app"); got != "encapi" {
t.Errorf("param_f_app = %q", got)
}
for _, pred := range []string{
"namespace = {f_namespace:String}",
"labels['app'] = {f_app:String}",
"severity = {f_severity:String}",
} {
if !strings.Contains(ch.lastSQL, pred) {
t.Errorf("SQL missing %q:\n%s", pred, ch.lastSQL)
}
}
// values must never be interpolated into the SQL text
if strings.Contains(ch.lastSQL, "prod") || strings.Contains(ch.lastSQL, "encapi") {
t.Errorf("filter value interpolated into SQL:\n%s", ch.lastSQL)
}
}
func TestQueryFuzzyTokens(t *testing.T) {
s, ch := newTestServer(t)
rec := get(t, s, "/api/query?q="+url.QueryEscape("timeout o'brien"))
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
if got, want := ch.lastParams.Get("param_q_tokens"), `['timeout','o\'brien']`; got != want {
t.Errorf("q_tokens = %q, want %q", got, want)
}
if !strings.Contains(ch.lastSQL, "arrayAll(t -> positionCaseInsensitive(message, t) > 0, {q_tokens:Array(String)})") {
t.Errorf("SQL missing fuzzy predicate:\n%s", ch.lastSQL)
}
}
// --- raw sql fragment ---
func TestQuerySQLFragmentWrapped(t *testing.T) {
s, ch := newTestServer(t)
frag := "severity = 'error' AND message LIKE '%oom%'"
rec := get(t, s, "/api/query?sql="+url.QueryEscape(frag))
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
if !strings.Contains(ch.lastSQL, "AND ( "+frag+" )") {
t.Errorf("fragment not parenthesised inside WHERE:\n%s", ch.lastSQL)
}
// bounds and LIMIT still enforced around the fragment
if !strings.Contains(ch.lastSQL, "timestamp >= fromUnixTimestamp64Milli({since_ms:Int64})") {
t.Errorf("time bounds missing with sql fragment:\n%s", ch.lastSQL)
}
if !strings.Contains(ch.lastSQL, "LIMIT 100 OFFSET 0") {
t.Errorf("LIMIT missing with sql fragment:\n%s", ch.lastSQL)
}
}
func TestQuerySQLFragmentRejectsSemicolon(t *testing.T) {
s, _ := newTestServer(t)
rec := get(t, s, "/api/query?sql="+url.QueryEscape("1=1; DROP TABLE logs.raw"))
if rec.Code != http.StatusBadRequest {
t.Fatalf("status %d, want 400", rec.Code)
}
}
// --- limit / offset ---
func TestQueryLimitCappedAndOffset(t *testing.T) {
s, ch := newTestServer(t)
rec := get(t, s, "/api/query?limit=999999&offset=200")
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
if !strings.Contains(ch.lastSQL, "LIMIT 1000 OFFSET 200") {
t.Errorf("limit not capped at %d:\n%s", maxLimit, ch.lastSQL)
}
for _, c := range []string{"limit=0", "limit=-5", "limit=abc", "offset=-1"} {
if rec := get(t, s, "/api/query?"+c); rec.Code != http.StatusBadRequest {
t.Errorf("%s: status %d, want 400", c, rec.Code)
}
}
}
// --- tail ---
func tailRows(ts ...int64) string {
rows := make([]map[string]any, len(ts))
for i, v := range ts {
rows[i] = map[string]any{"ts_ms": v, "message": "m"}
}
b, _ := json.Marshal(map[string]any{"meta": []any{}, "data": rows, "rows": len(rows)})
return string(b)
}
func TestTailCursorAdvances(t *testing.T) {
s, ch := newTestServer(t)
cursor := testNow.Add(-5 * time.Second).UnixMilli()
ch.respond = func() (int, string) { return 200, tailRows(cursor+100, cursor+250) }
rec := get(t, s, "/api/tail?cursor="+strconv.FormatInt(cursor, 10))
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
if got := boundMS(t, ch.lastParams, "since_ms"); got != cursor {
t.Errorf("since_ms = %d, want cursor %d", got, cursor)
}
if !strings.Contains(ch.lastSQL, "timestamp > fromUnixTimestamp64Milli({since_ms:Int64})") {
t.Errorf("tail lower bound must be exclusive:\n%s", ch.lastSQL)
}
if !strings.Contains(ch.lastSQL, "ORDER BY timestamp ASC") {
t.Errorf("tail must be ascending:\n%s", ch.lastSQL)
}
var resp struct {
Cursor int64 `json:"cursor"`
Count int `json:"count"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if resp.Cursor != cursor+250 {
t.Errorf("cursor = %d, want %d", resp.Cursor, cursor+250)
}
if resp.Count != 2 {
t.Errorf("count = %d, want 2", resp.Count)
}
}
func TestTailCursorUnchangedWhenNoRows(t *testing.T) {
s, _ := newTestServer(t)
cursor := testNow.Add(-3 * time.Second).UnixMilli()
rec := get(t, s, "/api/tail?cursor="+strconv.FormatInt(cursor, 10))
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
var resp struct {
Cursor int64 `json:"cursor"`
}
json.Unmarshal(rec.Body.Bytes(), &resp)
if resp.Cursor != cursor {
t.Errorf("cursor = %d, want unchanged %d", resp.Cursor, cursor)
}
}
func TestTailStaleCursorClamped(t *testing.T) {
s, ch := newTestServer(t)
stale := testNow.Add(-48 * time.Hour).UnixMilli()
rec := get(t, s, "/api/tail?cursor="+strconv.FormatInt(stale, 10))
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
if got, min := boundMS(t, ch.lastParams, "since_ms"), testNow.Add(-defaultWindow).UnixMilli(); got != min {
t.Errorf("stale cursor not clamped: since_ms = %d, want %d", got, min)
}
}
func TestTailBadCursor(t *testing.T) {
s, _ := newTestServer(t)
if rec := get(t, s, "/api/tail?cursor=nope"); rec.Code != http.StatusBadRequest {
t.Fatalf("status %d, want 400", rec.Code)
}
}
// --- facets ---
func TestFacets(t *testing.T) {
s, ch := newTestServer(t)
ch.respond = func() (int, string) {
rows := []map[string]any{
{"facet": "namespace", "value": "prod", "n": 100},
{"facet": "namespace", "value": "logging", "n": 50},
{"facet": "app", "value": "encapi", "n": 70},
{"facet": "host", "value": "node1", "n": 30},
}
b, _ := json.Marshal(map[string]any{"meta": []any{}, "data": rows, "rows": len(rows)})
return 200, string(b)
}
rec := get(t, s, "/api/facets?since=1h")
if rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
for _, expr := range []string{"GROUP BY value", "labels['app']", "UNION ALL", "LIMIT 20"} {
if !strings.Contains(ch.lastSQL, expr) {
t.Errorf("facets SQL missing %q:\n%s", expr, ch.lastSQL)
}
}
var resp map[string][]map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if len(resp["namespaces"]) != 2 || len(resp["apps"]) != 1 || len(resp["hosts"]) != 1 {
t.Errorf("facet grouping wrong: %v", resp)
}
if resp["namespaces"][0]["value"] != "prod" {
t.Errorf("namespaces[0] = %v", resp["namespaces"][0])
}
}
func TestFacetsRejectsWideWindow(t *testing.T) {
s, _ := newTestServer(t)
if rec := get(t, s, "/api/facets?since=2020-01-01T00:00:00Z"); rec.Code != http.StatusBadRequest {
t.Fatalf("status %d, want 400", rec.Code)
}
}
// --- healthz + plumbing ---
func TestHealthz(t *testing.T) {
s, ch := newTestServer(t)
if rec := get(t, s, "/healthz"); rec.Code != 200 {
t.Fatalf("status %d: %s", rec.Code, rec.Body)
}
ch.respond = func() (int, string) { return 500, "boom" }
if rec := get(t, s, "/healthz"); rec.Code != http.StatusServiceUnavailable {
t.Fatalf("status %d, want 503", rec.Code)
}
}
func TestQueryPropagatesCHError(t *testing.T) {
s, ch := newTestServer(t)
ch.respond = func() (int, string) { return 400, "Code: 62. DB::Exception: Syntax error" }
rec := get(t, s, "/api/query?sql="+url.QueryEscape("not valid sql"))
if rec.Code != http.StatusBadGateway {
t.Fatalf("status %d, want 502", rec.Code)
}
if !strings.Contains(rec.Body.String(), "Syntax error") {
t.Errorf("CH error not surfaced: %s", rec.Body)
}
}
func TestMaxExecutionTimeSet(t *testing.T) {
s, ch := newTestServer(t)
if rec := get(t, s, "/api/query"); rec.Code != 200 {
t.Fatalf("status %d", rec.Code)
}
if got := ch.lastParams.Get("max_execution_time"); got != "30" {
t.Errorf("max_execution_time = %q, want 30", got)
}
}
func TestIndexServed(t *testing.T) {
s, _ := newTestServer(t)
rec := get(t, s, "/")
if rec.Code != 200 || !strings.Contains(rec.Body.String(), "logviewer") {
t.Fatalf("index not served: %d", rec.Code)
}
}