Merge pull request 'Add logviewer: web UI for the ClickHouse log store' (#1) from benvin/initial into main
ci/woodpecker/tag/release Pipeline was successful
ci/woodpecker/tag/release Pipeline was successful
Reviewed-on: #1
This commit was merged in pull request #1.
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
dist/
|
||||
logviewer
|
||||
@@ -0,0 +1,19 @@
|
||||
exclude: ^web/static/(jquery|bootstrap)[^/]*\.min\.(js|css)$
|
||||
|
||||
repos:
|
||||
- repo: https://github.com/pre-commit/pre-commit-hooks
|
||||
rev: v4.6.0
|
||||
hooks:
|
||||
- id: trailing-whitespace
|
||||
- id: end-of-file-fixer
|
||||
- id: check-yaml
|
||||
- id: check-merge-conflict
|
||||
- id: mixed-line-ending
|
||||
args: [--fix=lf]
|
||||
|
||||
- repo: https://github.com/dnephin/pre-commit-golang
|
||||
rev: v0.5.1
|
||||
hooks:
|
||||
- id: go-fmt
|
||||
- id: go-vet
|
||||
- id: go-unit-tests
|
||||
@@ -0,0 +1,43 @@
|
||||
when:
|
||||
- event: pull_request
|
||||
|
||||
steps:
|
||||
- name: build
|
||||
image: golang:1.25
|
||||
commands:
|
||||
- make build
|
||||
backend_options:
|
||||
kubernetes:
|
||||
serviceAccountName: default
|
||||
resources:
|
||||
requests:
|
||||
memory: 512Mi
|
||||
cpu: 1
|
||||
limits:
|
||||
memory: 2Gi
|
||||
cpu: 2
|
||||
|
||||
# Validate the image builds without pushing. The CA-baked buildx plugin trusts
|
||||
# the internal registry's CA; the plain woodpeckerci plugin fails x509 here.
|
||||
- name: build-check
|
||||
image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
|
||||
settings:
|
||||
repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/logviewer
|
||||
dockerfile: Dockerfile
|
||||
dry_run: true
|
||||
platforms: linux/amd64
|
||||
build_args:
|
||||
VERSION: ${CI_COMMIT_SHA}
|
||||
buildkit_config: |
|
||||
[registry."artifactapi.k8s.syd1.au.unkin.net"]
|
||||
ca = ["/etc/docker/certs.d/artifactapi.k8s.syd1.au.unkin.net/ca.crt"]
|
||||
backend_options:
|
||||
kubernetes:
|
||||
serviceAccountName: default
|
||||
resources:
|
||||
requests:
|
||||
memory: 1Gi
|
||||
cpu: 1
|
||||
limits:
|
||||
memory: 4Gi
|
||||
cpu: 2
|
||||
@@ -0,0 +1,18 @@
|
||||
when:
|
||||
- event: pull_request
|
||||
|
||||
steps:
|
||||
- name: pre-commit
|
||||
image: git.unkin.net/unkin/almalinux9-gobuilder:20260606
|
||||
commands:
|
||||
- uvx pre-commit run --all-files
|
||||
backend_options:
|
||||
kubernetes:
|
||||
serviceAccountName: default
|
||||
resources:
|
||||
requests:
|
||||
memory: 512Mi
|
||||
cpu: 1
|
||||
limits:
|
||||
memory: 2Gi
|
||||
cpu: 2
|
||||
@@ -0,0 +1,30 @@
|
||||
when:
|
||||
- event: tag
|
||||
ref: refs/tags/v*
|
||||
|
||||
steps:
|
||||
- name: docker
|
||||
image: artifactapi.k8s.syd1.au.unkin.net/docker-internal/plugin-docker-buildx:latest
|
||||
settings:
|
||||
registry: artifactapi.k8s.syd1.au.unkin.net
|
||||
repo: artifactapi.k8s.syd1.au.unkin.net/docker-internal/logviewer
|
||||
dockerfile: Dockerfile
|
||||
platforms: linux/amd64
|
||||
build_args:
|
||||
VERSION: "${CI_COMMIT_TAG}"
|
||||
buildkit_config: |
|
||||
[registry."artifactapi.k8s.syd1.au.unkin.net"]
|
||||
ca = ["/etc/docker/certs.d/artifactapi.k8s.syd1.au.unkin.net/ca.crt"]
|
||||
tags:
|
||||
- "${CI_COMMIT_TAG}"
|
||||
- latest
|
||||
backend_options:
|
||||
kubernetes:
|
||||
serviceAccountName: default
|
||||
resources:
|
||||
requests:
|
||||
memory: 1Gi
|
||||
cpu: 1
|
||||
limits:
|
||||
memory: 4Gi
|
||||
cpu: 2
|
||||
@@ -0,0 +1,20 @@
|
||||
when:
|
||||
- event: pull_request
|
||||
|
||||
steps:
|
||||
- name: test
|
||||
image: golang:1.25
|
||||
commands:
|
||||
- go vet ./...
|
||||
- test -z "$(gofmt -l .)"
|
||||
- go test -v -race ./...
|
||||
backend_options:
|
||||
kubernetes:
|
||||
serviceAccountName: default
|
||||
resources:
|
||||
requests:
|
||||
memory: 512Mi
|
||||
cpu: 1
|
||||
limits:
|
||||
memory: 2Gi
|
||||
cpu: 2
|
||||
+18
@@ -0,0 +1,18 @@
|
||||
FROM golang:1.25-alpine AS builder
|
||||
|
||||
WORKDIR /build
|
||||
|
||||
COPY go.mod ./
|
||||
RUN go mod download
|
||||
|
||||
COPY . .
|
||||
|
||||
ARG VERSION=dev
|
||||
RUN CGO_ENABLED=0 go build -ldflags="-s -w -X main.version=${VERSION}" -o logviewer .
|
||||
|
||||
# distroless static ships ca-certificates and runs as an unprivileged user.
|
||||
FROM gcr.io/distroless/static-debian12:nonroot
|
||||
|
||||
COPY --from=builder /build/logviewer /usr/local/bin/logviewer
|
||||
|
||||
ENTRYPOINT ["logviewer"]
|
||||
@@ -0,0 +1,54 @@
|
||||
BINARY := logviewer
|
||||
DIST := dist
|
||||
VERSION := $(shell git describe --tags --always --dirty 2>/dev/null || echo dev)
|
||||
GOFLAGS := -ldflags="-s -w -X main.version=$(VERSION)"
|
||||
OS ?= $(shell go env GOOS)
|
||||
ARCH ?= $(shell go env GOARCH)
|
||||
|
||||
.PHONY: all build test lint fmt vet clean docker patch minor major _tag
|
||||
|
||||
all: build
|
||||
|
||||
build:
|
||||
CGO_ENABLED=0 GOOS=$(OS) GOARCH=$(ARCH) go build $(GOFLAGS) -o $(DIST)/$(BINARY) .
|
||||
|
||||
test:
|
||||
go test -v -race ./...
|
||||
|
||||
vet:
|
||||
go vet ./...
|
||||
|
||||
lint:
|
||||
golangci-lint run ./...
|
||||
|
||||
fmt:
|
||||
gofmt -w .
|
||||
|
||||
clean:
|
||||
rm -rf $(DIST)
|
||||
|
||||
docker:
|
||||
docker build --build-arg VERSION=$(VERSION) -t $(BINARY):$(VERSION) .
|
||||
|
||||
# Bump helpers — reads the latest semver tag and creates the next one.
|
||||
# If no tag exists yet, starts from v0.0.0.
|
||||
_LATEST := $(shell git tag --sort=-v:refname | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$$' | head -1)
|
||||
_BASE := $(if $(_LATEST),$(_LATEST),v0.0.0)
|
||||
_MAJ := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f1)
|
||||
_MIN := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f2)
|
||||
_PAT := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f3)
|
||||
|
||||
patch:
|
||||
@NEW=v$(_MAJ).$(_MIN).$(shell expr $(_PAT) + 1); \
|
||||
git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
|
||||
|
||||
minor:
|
||||
@NEW=v$(_MAJ).$(shell expr $(_MIN) + 1).0; \
|
||||
git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
|
||||
|
||||
major:
|
||||
@NEW=v$(shell expr $(_MAJ) + 1).0.0; \
|
||||
git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW
|
||||
|
||||
_tag:
|
||||
git push origin $(TAG)
|
||||
@@ -1,3 +1,57 @@
|
||||
# logviewer
|
||||
|
||||
Web UI for the ClickHouse log store: fuzzy find, tail and SQL-filter logs (logviewer.unkin.net)
|
||||
Web UI for the ClickHouse log store (`logs.raw`). Single Go binary with an
|
||||
embedded Bootstrap 3 + jQuery UI — no CDN assets, no build step.
|
||||
|
||||
logviewer does **no authentication itself**. It runs behind oauth2-proxy at
|
||||
`logviewer.unkin.net`; the proxy is the auth boundary. Query safety relies on
|
||||
the readonly ClickHouse user (`logreader`) plus server-side guardrails:
|
||||
|
||||
- **Time bounds are always enforced.** Missing `since`/`until` default to a
|
||||
15m window ending now; windows wider than 72h (the table TTL) are rejected.
|
||||
No request can scan the store unbounded.
|
||||
- **Everything is parameterized** (ClickHouse `param_*` bound parameters)
|
||||
except the explicit raw `sql` fragment, which is parenthesised and ANDed
|
||||
into the outer query — the enforced time bounds and `LIMIT` still apply,
|
||||
and the readonly user can't write.
|
||||
- `max_execution_time=30` is set on every query.
|
||||
|
||||
## Configuration
|
||||
|
||||
| Env | Default | |
|
||||
|---|---|---|
|
||||
| `CH_URL` | `http://clickhouse-logs.logging.svc.cluster.local:8123` | ClickHouse HTTP endpoint |
|
||||
| `CH_USER` | `logreader` | readonly ClickHouse user |
|
||||
| `CH_PASSWORD` | (empty) | password for `CH_USER` |
|
||||
| `LISTEN_ADDR` | `:8080` | HTTP listen address |
|
||||
|
||||
## API
|
||||
|
||||
- `GET /api/query` — `since`/`until` (RFC3339, unix, or relative `15m`/`6h`/`1d`),
|
||||
filters `namespace`/`host`/`pod`/`container`/`app`/`severity`/`stream`/`source`,
|
||||
`q` (space-separated case-insensitive substring terms, all must match),
|
||||
`sql` (raw WHERE fragment), `limit` (max 1000) / `offset`. Newest first.
|
||||
- `GET /api/tail` — same filters plus `cursor` (exclusive lower bound, unix ms).
|
||||
Returns ascending rows and the next `cursor`. Stale cursors are clamped to
|
||||
the last 15m.
|
||||
- `GET /api/facets` — top 20 namespaces / apps (`labels['app']`) / hosts by
|
||||
row count in the window.
|
||||
- `GET /healthz` — ClickHouse ping.
|
||||
|
||||
## UI
|
||||
|
||||
Time-range picker (15m/1h/6h/1d/custom), facet dropdowns, debounced fuzzy
|
||||
search, toggleable raw-SQL box, paged log table with expandable rows
|
||||
(labels/fields), and a Tail mode (1s polling, autoscroll, pauses when you
|
||||
scroll up).
|
||||
|
||||
## Development
|
||||
|
||||
```sh
|
||||
make build # dist/logviewer
|
||||
make test # go test -race ./...
|
||||
make patch # tag + push next vX.Y.Z → CI builds+pushes the image
|
||||
```
|
||||
|
||||
Releases: pushing a `v*` tag builds the container and pushes it to
|
||||
`artifactapi.k8s.syd1.au.unkin.net/docker-internal/logviewer`.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
@@ -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})
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"git.unkin.net/unkin/logviewer/internal/server"
|
||||
)
|
||||
|
||||
var version = "dev"
|
||||
|
||||
func envOr(key, def string) string {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func main() {
|
||||
chURL := envOr("CH_URL", "http://clickhouse-logs.logging.svc.cluster.local:8123")
|
||||
chUser := envOr("CH_USER", "logreader")
|
||||
chPassword := os.Getenv("CH_PASSWORD")
|
||||
addr := envOr("LISTEN_ADDR", ":8080")
|
||||
|
||||
srv := server.New(server.Config{
|
||||
CHURL: chURL,
|
||||
CHUser: chUser,
|
||||
CHPassword: chPassword,
|
||||
Version: version,
|
||||
})
|
||||
|
||||
hs := &http.Server{
|
||||
Addr: addr,
|
||||
Handler: srv,
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
ReadTimeout: 60 * time.Second,
|
||||
WriteTimeout: 60 * time.Second,
|
||||
IdleTimeout: 120 * time.Second,
|
||||
}
|
||||
|
||||
log.Printf("logviewer %s listening on %s (clickhouse %s as %s)", version, addr, chURL, chUser)
|
||||
if err := hs.ListenAndServe(); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
// Package web embeds the UI so the server binary is self-contained.
|
||||
package web
|
||||
|
||||
import "embed"
|
||||
|
||||
//go:embed index.html static
|
||||
var FS embed.FS
|
||||
+122
@@ -0,0 +1,122 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>logviewer</title>
|
||||
<link rel="stylesheet" href="/static/bootstrap.min.css">
|
||||
<link rel="stylesheet" href="/static/logviewer.css">
|
||||
</head>
|
||||
<body>
|
||||
<nav class="navbar navbar-inverse navbar-static-top">
|
||||
<div class="container-fluid">
|
||||
<div class="navbar-header">
|
||||
<a class="navbar-brand" href="/">logviewer</a>
|
||||
</div>
|
||||
<p class="navbar-text navbar-right" id="status-text"></p>
|
||||
</div>
|
||||
</nav>
|
||||
|
||||
<div class="container-fluid">
|
||||
<form id="search-form" onsubmit="return false;">
|
||||
<div class="row">
|
||||
<div class="col-md-4">
|
||||
<div class="btn-group" role="group" id="range-buttons">
|
||||
<button type="button" class="btn btn-default range-btn active" data-range="15m">15m</button>
|
||||
<button type="button" class="btn btn-default range-btn" data-range="1h">1h</button>
|
||||
<button type="button" class="btn btn-default range-btn" data-range="6h">6h</button>
|
||||
<button type="button" class="btn btn-default range-btn" data-range="1d">1d</button>
|
||||
<button type="button" class="btn btn-default range-btn" data-range="custom">custom</button>
|
||||
</div>
|
||||
</div>
|
||||
<div class="col-md-5">
|
||||
<div class="input-group">
|
||||
<input type="text" class="form-control" id="q" placeholder="fuzzy search (space-separated terms, all must match)">
|
||||
<span class="input-group-btn">
|
||||
<button class="btn btn-primary" type="button" id="search-btn">Search</button>
|
||||
</span>
|
||||
</div>
|
||||
</div>
|
||||
<div class="col-md-3 text-right">
|
||||
<button type="button" class="btn btn-default" id="sql-toggle">SQL</button>
|
||||
<button type="button" class="btn btn-default" id="tail-toggle">Tail</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="row" id="custom-range" style="display:none; margin-top:8px;">
|
||||
<div class="col-md-3">
|
||||
<div class="input-group">
|
||||
<span class="input-group-addon">since</span>
|
||||
<input type="datetime-local" class="form-control" id="since-input" step="1">
|
||||
</div>
|
||||
</div>
|
||||
<div class="col-md-3">
|
||||
<div class="input-group">
|
||||
<span class="input-group-addon">until</span>
|
||||
<input type="datetime-local" class="form-control" id="until-input" step="1">
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="row" id="sql-row" style="display:none; margin-top:8px;">
|
||||
<div class="col-md-12">
|
||||
<textarea class="form-control mono" id="sql" rows="2"
|
||||
placeholder="raw WHERE fragment, e.g. severity = 'error' AND message LIKE '%timeout%' (runs as readonly user, time bounds and LIMIT always enforced)"></textarea>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="row" style="margin-top:8px;">
|
||||
<div class="col-md-2"><select class="form-control" id="f-namespace"><option value="">namespace</option></select></div>
|
||||
<div class="col-md-2"><select class="form-control" id="f-app"><option value="">app</option></select></div>
|
||||
<div class="col-md-2"><select class="form-control" id="f-host"><option value="">host</option></select></div>
|
||||
<div class="col-md-2"><input type="text" class="form-control" id="f-pod" placeholder="pod"></div>
|
||||
<div class="col-md-1"><input type="text" class="form-control" id="f-container" placeholder="container"></div>
|
||||
<div class="col-md-2">
|
||||
<select class="form-control" id="f-severity">
|
||||
<option value="">severity</option>
|
||||
<option>fatal</option><option>error</option><option>warn</option>
|
||||
<option>info</option><option>debug</option><option>trace</option>
|
||||
</select>
|
||||
</div>
|
||||
<div class="col-md-1">
|
||||
<select class="form-control" id="f-stream">
|
||||
<option value="">stream</option>
|
||||
<option>stdout</option><option>stderr</option>
|
||||
</select>
|
||||
</div>
|
||||
</div>
|
||||
</form>
|
||||
|
||||
<div class="row" style="margin-top:10px;">
|
||||
<div class="col-md-6">
|
||||
<p id="result-info" class="text-muted"></p>
|
||||
</div>
|
||||
<div class="col-md-6 text-right" id="pager">
|
||||
<div class="btn-group">
|
||||
<button type="button" class="btn btn-default btn-sm" id="prev-btn" disabled>« newer</button>
|
||||
<button type="button" class="btn btn-default btn-sm" id="next-btn" disabled>older »</button>
|
||||
</div>
|
||||
<select id="page-size" class="btn btn-default btn-sm">
|
||||
<option>100</option><option>200</option><option>500</option><option>1000</option>
|
||||
</select>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<table class="table table-condensed table-hover" id="log-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th class="col-ts">timestamp (UTC)</th>
|
||||
<th class="col-src">namespace / pod</th>
|
||||
<th class="col-sev">severity</th>
|
||||
<th>message</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody id="log-body"></tbody>
|
||||
</table>
|
||||
</div>
|
||||
|
||||
<script src="/static/jquery.min.js"></script>
|
||||
<script src="/static/bootstrap.min.js"></script>
|
||||
<script src="/static/app.js"></script>
|
||||
</body>
|
||||
</html>
|
||||
@@ -0,0 +1,253 @@
|
||||
(function () {
|
||||
'use strict';
|
||||
|
||||
var state = {
|
||||
range: '15m',
|
||||
offset: 0,
|
||||
limit: 100,
|
||||
tail: false,
|
||||
tailTimer: null,
|
||||
tailCursor: null,
|
||||
tailPaused: false,
|
||||
lastCount: 0
|
||||
};
|
||||
|
||||
var SEV_CLASS = {
|
||||
fatal: 'label-danger', crit: 'label-danger', critical: 'label-danger',
|
||||
error: 'label-danger', err: 'label-danger',
|
||||
warn: 'label-warning', warning: 'label-warning',
|
||||
info: 'label-info',
|
||||
debug: 'label-default', trace: 'label-default'
|
||||
};
|
||||
|
||||
function sevLabel(sev) {
|
||||
var cls = SEV_CLASS[(sev || '').toLowerCase()] || 'label-default';
|
||||
return $('<span>').addClass('label ' + cls).text(sev || '-');
|
||||
}
|
||||
|
||||
function timeParams() {
|
||||
var p = {};
|
||||
if (state.range === 'custom') {
|
||||
var s = $('#since-input').val(), u = $('#until-input').val();
|
||||
if (s) p.since = new Date(s).toISOString();
|
||||
if (u) p.until = new Date(u).toISOString();
|
||||
} else {
|
||||
p.since = state.range;
|
||||
}
|
||||
return p;
|
||||
}
|
||||
|
||||
function filterParams() {
|
||||
var p = {};
|
||||
var map = {
|
||||
namespace: '#f-namespace', app: '#f-app', host: '#f-host',
|
||||
pod: '#f-pod', container: '#f-container',
|
||||
severity: '#f-severity', stream: '#f-stream'
|
||||
};
|
||||
Object.keys(map).forEach(function (k) {
|
||||
var v = $(map[k]).val();
|
||||
if (v) p[k] = v;
|
||||
});
|
||||
var q = $('#q').val().trim();
|
||||
if (q) p.q = q;
|
||||
if ($('#sql-row').is(':visible')) {
|
||||
var sql = $('#sql').val().trim();
|
||||
if (sql) p.sql = sql;
|
||||
}
|
||||
return p;
|
||||
}
|
||||
|
||||
function setStatus(msg, isError) {
|
||||
$('#status-text').text(msg || '').toggleClass('text-danger', !!isError);
|
||||
}
|
||||
|
||||
function renderRow(row, tbody) {
|
||||
var src = (row.namespace || '-') + ' / ' + (row.pod || '-');
|
||||
var tr = $('<tr>').addClass('log-row')
|
||||
.append($('<td>').addClass('col-ts').text(row.timestamp))
|
||||
.append($('<td>').addClass('col-src').text(src))
|
||||
.append($('<td>').addClass('col-sev').append(sevLabel(row.severity)))
|
||||
.append($('<td>').addClass('msg').text(row.message));
|
||||
tr.data('row', row);
|
||||
tbody.append(tr);
|
||||
}
|
||||
|
||||
function kvTable(title, obj) {
|
||||
var keys = Object.keys(obj || {});
|
||||
if (!keys.length) return null;
|
||||
var t = $('<table>').addClass('table table-condensed detail-table');
|
||||
t.append($('<caption>').text(title));
|
||||
keys.sort().forEach(function (k) {
|
||||
t.append($('<tr>').append($('<th>').text(k)).append($('<td>').text(obj[k])));
|
||||
});
|
||||
return t;
|
||||
}
|
||||
|
||||
function toggleDetail(tr) {
|
||||
var next = tr.next();
|
||||
if (next.hasClass('detail-row')) { next.remove(); return; }
|
||||
var row = tr.data('row');
|
||||
var cell = $('<td>').attr('colspan', 4);
|
||||
var meta = {
|
||||
host: row.host, source: row.source, namespace: row.namespace,
|
||||
pod: row.pod, container: row.container, stream: row.stream
|
||||
};
|
||||
[kvTable('meta', meta), kvTable('labels', row.labels), kvTable('fields', row.fields)]
|
||||
.forEach(function (t) { if (t) cell.append(t); });
|
||||
tr.after($('<tr>').addClass('detail-row').append(cell));
|
||||
}
|
||||
|
||||
function renderRows(rows, append) {
|
||||
var tbody = $('#log-body');
|
||||
if (!append) tbody.empty();
|
||||
rows.forEach(function (r) { renderRow(r, tbody); });
|
||||
}
|
||||
|
||||
function runQuery() {
|
||||
var params = $.extend({ limit: state.limit, offset: state.offset }, timeParams(), filterParams());
|
||||
setStatus('loading...');
|
||||
$.getJSON('/api/query', params)
|
||||
.done(function (res) {
|
||||
renderRows(res.rows, false);
|
||||
state.lastCount = res.count;
|
||||
setStatus('');
|
||||
$('#result-info').text(
|
||||
res.count + ' rows (' + res.since + ' → ' + res.until + '), offset ' + res.offset);
|
||||
$('#prev-btn').prop('disabled', state.offset === 0);
|
||||
$('#next-btn').prop('disabled', res.count < state.limit);
|
||||
})
|
||||
.fail(function (xhr) {
|
||||
setStatus((xhr.responseJSON && xhr.responseJSON.error) || 'query failed', true);
|
||||
});
|
||||
}
|
||||
|
||||
function tailTick() {
|
||||
if (state.tailPaused) return;
|
||||
var params = $.extend({}, filterParams());
|
||||
if (state.tailCursor !== null) params.cursor = state.tailCursor;
|
||||
$.getJSON('/api/tail', params)
|
||||
.done(function (res) {
|
||||
state.tailCursor = res.cursor;
|
||||
if (res.rows.length) {
|
||||
renderRows(res.rows, true);
|
||||
// keep the DOM bounded during long tails
|
||||
var body = $('#log-body');
|
||||
var extra = body.children('.log-row').length - 2000;
|
||||
if (extra > 0) body.children().slice(0, extra * 2).remove();
|
||||
if (!state.tailPaused) {
|
||||
window.scrollTo(0, document.body.scrollHeight);
|
||||
}
|
||||
}
|
||||
setStatus('tailing (cursor ' + res.cursor + ')');
|
||||
})
|
||||
.fail(function (xhr) {
|
||||
setStatus((xhr.responseJSON && xhr.responseJSON.error) || 'tail failed', true);
|
||||
});
|
||||
}
|
||||
|
||||
function startTail() {
|
||||
state.tail = true;
|
||||
state.tailCursor = null;
|
||||
state.tailPaused = false;
|
||||
$('#log-body').empty();
|
||||
$('#tail-toggle').addClass('active');
|
||||
$('#pager button, #page-size').prop('disabled', true);
|
||||
state.tailTimer = setInterval(tailTick, 1000);
|
||||
tailTick();
|
||||
}
|
||||
|
||||
function stopTail() {
|
||||
state.tail = false;
|
||||
clearInterval(state.tailTimer);
|
||||
state.tailTimer = null;
|
||||
$('#tail-toggle').removeClass('active');
|
||||
$('#pager button, #page-size').prop('disabled', false);
|
||||
setStatus('');
|
||||
}
|
||||
|
||||
function refresh() {
|
||||
if (state.tail) { stopTail(); startTail(); return; }
|
||||
state.offset = 0;
|
||||
runQuery();
|
||||
}
|
||||
|
||||
function loadFacets() {
|
||||
$.getJSON('/api/facets', timeParams()).done(function (res) {
|
||||
var fill = function (sel, items, placeholder) {
|
||||
var cur = $(sel).val();
|
||||
var el = $(sel).empty().append($('<option>').val('').text(placeholder));
|
||||
(items || []).forEach(function (it) {
|
||||
el.append($('<option>').val(it.value).text(it.value + ' (' + it.count + ')'));
|
||||
});
|
||||
el.val(cur);
|
||||
if (el.val() === null) el.val('');
|
||||
};
|
||||
fill('#f-namespace', res.namespaces, 'namespace');
|
||||
fill('#f-app', res.apps, 'app');
|
||||
fill('#f-host', res.hosts, 'host');
|
||||
});
|
||||
}
|
||||
|
||||
// --- events ---
|
||||
|
||||
$('#range-buttons').on('click', '.range-btn', function () {
|
||||
$('.range-btn').removeClass('active');
|
||||
$(this).addClass('active');
|
||||
state.range = $(this).data('range');
|
||||
$('#custom-range').toggle(state.range === 'custom');
|
||||
if (state.range !== 'custom') { loadFacets(); refresh(); }
|
||||
});
|
||||
|
||||
$('#custom-range input').on('change', function () {
|
||||
loadFacets();
|
||||
refresh();
|
||||
});
|
||||
|
||||
$('#search-btn').on('click', refresh);
|
||||
$('#search-form').on('change', 'select', refresh);
|
||||
$('#search-form').on('keypress', 'input', function (e) {
|
||||
if (e.which === 13) { e.preventDefault(); refresh(); }
|
||||
});
|
||||
|
||||
var debounce = null;
|
||||
$('#q').on('input', function () {
|
||||
clearTimeout(debounce);
|
||||
debounce = setTimeout(refresh, 300);
|
||||
});
|
||||
|
||||
$('#sql-toggle').on('click', function () {
|
||||
$(this).toggleClass('active');
|
||||
$('#sql-row').toggle($(this).hasClass('active'));
|
||||
});
|
||||
|
||||
$('#tail-toggle').on('click', function () {
|
||||
if (state.tail) stopTail(); else startTail();
|
||||
});
|
||||
|
||||
$('#log-body').on('click', 'tr.log-row', function () { toggleDetail($(this)); });
|
||||
|
||||
$('#prev-btn').on('click', function () {
|
||||
state.offset = Math.max(0, state.offset - state.limit);
|
||||
runQuery();
|
||||
});
|
||||
$('#next-btn').on('click', function () {
|
||||
state.offset += state.limit;
|
||||
runQuery();
|
||||
});
|
||||
$('#page-size').on('change', function () {
|
||||
state.limit = parseInt($(this).val(), 10);
|
||||
state.offset = 0;
|
||||
runQuery();
|
||||
});
|
||||
|
||||
// Pause tail autoscroll when the user scrolls up; resume at the bottom.
|
||||
$(window).on('scroll', function () {
|
||||
if (!state.tail) return;
|
||||
var atBottom = window.innerHeight + window.scrollY >= document.body.scrollHeight - 40;
|
||||
state.tailPaused = !atBottom;
|
||||
if (state.tailPaused) setStatus('tail paused (scroll to bottom to resume)');
|
||||
});
|
||||
|
||||
loadFacets();
|
||||
runQuery();
|
||||
})();
|
||||
Vendored
+6
File diff suppressed because one or more lines are too long
Vendored
+6
File diff suppressed because one or more lines are too long
Vendored
+2
File diff suppressed because one or more lines are too long
@@ -0,0 +1,14 @@
|
||||
.mono, #log-table td.msg, #log-table td.col-ts, .detail-table td {
|
||||
font-family: Menlo, Monaco, Consolas, "Liberation Mono", monospace;
|
||||
font-size: 12px;
|
||||
}
|
||||
#log-table td { vertical-align: top; word-break: break-all; }
|
||||
#log-table .col-ts { white-space: nowrap; width: 190px; }
|
||||
#log-table .col-src { width: 260px; }
|
||||
#log-table .col-sev { width: 80px; }
|
||||
#log-table tbody tr.log-row { cursor: pointer; }
|
||||
#log-table tr.detail-row td { background: #f7f7f9; }
|
||||
.detail-table { margin-bottom: 4px; }
|
||||
.detail-table th { width: 160px; }
|
||||
.sev-error, .sev-fatal, .sev-crit { }
|
||||
#tail-toggle.active, #sql-toggle.active { background: #337ab7; color: #fff; }
|
||||
Reference in New Issue
Block a user