mirror of
https://github.com/gesellix/Bose-SoundTouch.git
synced 2026-08-18 08:36:13 +00:00
feat(service): add Logs tab streaming the live stderr trace
Cloud-deploy operators on Discussion #295 needed to leave the admin UI for docker logs / journalctl to see what the service was doing. Mirror log.Default() output into an in-memory ring buffer and expose it under /setup/logs so the admin UI can show a live trace alongside the existing tabs. The buffer is a second sink under log.SetOutput(io.MultiWriter( os.Stderr, buf)) — stderr keeps receiving every line verbatim, so docker logs / journalctl are unaffected. Default capacity 2000 lines (~400 KB), tunable via SOUNDTOUCH_LOG_BUFFER_LINES. - pkg/service/logbuf: io.Writer ring with \n splitting, partial-line buffering, monotonic Seq, Since(since, limit) reporting dropped count when the caller falls behind. - New /setup/logs (GET) returns {entries, nextSince, dropped, capacity}. Polls at 1.5s while the tab is active; paused on document.hidden. - "8. Logs" tab with substring filter, tail-follow toggle (auto-disables when the user scrolls up), monospace dark view. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Tobias Gesellchen
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
f791145976
commit
c3723dc0e6
@@ -0,0 +1,86 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/logbuf"
|
||||
)
|
||||
|
||||
// logsResponse is the wire shape for GET /setup/logs. Entries is
|
||||
// ordered by Seq ascending. NextSince is the highest Seq returned,
|
||||
// suitable for the client's next poll's `since` parameter. Dropped
|
||||
// is the count of entries the client missed because the ring
|
||||
// evicted them before the client polled — a non-zero value
|
||||
// signals the client to surface a gap to the operator.
|
||||
type logsResponse struct {
|
||||
Entries []logbuf.Entry `json:"entries"`
|
||||
NextSince uint64 `json:"nextSince"`
|
||||
Dropped uint64 `json:"dropped"`
|
||||
Capacity int `json:"capacity"`
|
||||
}
|
||||
|
||||
// HandleGetLogs returns recent log entries from the in-process
|
||||
// ring buffer. Query parameters:
|
||||
//
|
||||
// since — return entries with Seq strictly greater than this
|
||||
// value. Omitted or "0" → full snapshot.
|
||||
// limit — cap the number of entries returned. Omitted → no cap
|
||||
// beyond the buffer's capacity.
|
||||
//
|
||||
// When no log buffer is attached (e.g. tests, or the env opted
|
||||
// out via SOUNDTOUCH_LOG_BUFFER_LINES=0), the response is an
|
||||
// empty snapshot rather than an error so the UI degrades
|
||||
// gracefully.
|
||||
func (s *Server) HandleGetLogs(w http.ResponseWriter, r *http.Request) {
|
||||
q := r.URL.Query()
|
||||
|
||||
since, err := parseUint64Query(q.Get("since"), 0)
|
||||
if err != nil {
|
||||
writeJSONError(w, http.StatusBadRequest, "invalid since: "+err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
limit, err := parseIntQuery(q.Get("limit"), 0)
|
||||
if err != nil {
|
||||
writeJSONError(w, http.StatusBadRequest, "invalid limit: "+err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
resp := logsResponse{
|
||||
Entries: []logbuf.Entry{},
|
||||
NextSince: since,
|
||||
}
|
||||
|
||||
if buf := s.LogBuffer(); buf != nil {
|
||||
entries, nextSince, dropped := buf.Since(since, limit)
|
||||
resp.Entries = entries
|
||||
resp.NextSince = nextSince
|
||||
resp.Dropped = dropped
|
||||
resp.Capacity = buf.Capacity()
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
|
||||
if err := json.NewEncoder(w).Encode(resp); err != nil {
|
||||
http.Error(w, "Failed to encode response", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func parseUint64Query(raw string, defaultVal uint64) (uint64, error) {
|
||||
if raw == "" {
|
||||
return defaultVal, nil
|
||||
}
|
||||
|
||||
return strconv.ParseUint(raw, 10, 64)
|
||||
}
|
||||
|
||||
func parseIntQuery(raw string, defaultVal int) (int, error) {
|
||||
if raw == "" {
|
||||
return defaultVal, nil
|
||||
}
|
||||
|
||||
return strconv.Atoi(raw)
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/logbuf"
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
func newLogsTestServer(t *testing.T, buf *logbuf.Buffer) *httptest.Server {
|
||||
t.Helper()
|
||||
|
||||
_, server := setupRouter("http://localhost:8001", nil)
|
||||
server.SetLogBuffer(buf)
|
||||
|
||||
r := chi.NewRouter()
|
||||
r.Get("/setup/logs", server.HandleGetLogs)
|
||||
|
||||
ts := httptest.NewServer(r)
|
||||
t.Cleanup(ts.Close)
|
||||
|
||||
return ts
|
||||
}
|
||||
|
||||
func TestHandleGetLogs_FullSnapshot(t *testing.T) {
|
||||
buf := logbuf.New(16)
|
||||
for _, line := range []string{"first\n", "second\n", "third\n"} {
|
||||
_, _ = buf.Write([]byte(line))
|
||||
}
|
||||
|
||||
ts := newLogsTestServer(t, buf)
|
||||
|
||||
res, err := http.Get(ts.URL + "/setup/logs")
|
||||
if err != nil {
|
||||
t.Fatalf("GET: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
if res.StatusCode != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d", res.StatusCode)
|
||||
}
|
||||
|
||||
var resp logsResponse
|
||||
if err := json.NewDecoder(res.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
|
||||
if len(resp.Entries) != 3 {
|
||||
t.Fatalf("expected 3 entries, got %d", len(resp.Entries))
|
||||
}
|
||||
|
||||
if resp.Entries[0].Message != "first" || resp.Entries[2].Message != "third" {
|
||||
t.Errorf("unexpected ordering: %+v", resp.Entries)
|
||||
}
|
||||
|
||||
if resp.NextSince != 3 {
|
||||
t.Errorf("nextSince: want 3, got %d", resp.NextSince)
|
||||
}
|
||||
|
||||
if resp.Capacity != 16 {
|
||||
t.Errorf("capacity: want 16, got %d", resp.Capacity)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleGetLogs_SinceRoundTrip(t *testing.T) {
|
||||
buf := logbuf.New(16)
|
||||
for i := 0; i < 5; i++ {
|
||||
_, _ = buf.Write([]byte("line\n"))
|
||||
}
|
||||
|
||||
ts := newLogsTestServer(t, buf)
|
||||
|
||||
res, err := http.Get(ts.URL + "/setup/logs?since=2")
|
||||
if err != nil {
|
||||
t.Fatalf("GET: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
var resp logsResponse
|
||||
if err := json.NewDecoder(res.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
|
||||
if len(resp.Entries) != 3 {
|
||||
t.Errorf("expected 3 entries with since=2, got %d", len(resp.Entries))
|
||||
}
|
||||
|
||||
if resp.NextSince != 5 {
|
||||
t.Errorf("nextSince: want 5, got %d", resp.NextSince)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleGetLogs_Limit(t *testing.T) {
|
||||
buf := logbuf.New(16)
|
||||
for i := 0; i < 8; i++ {
|
||||
_, _ = buf.Write([]byte("line\n"))
|
||||
}
|
||||
|
||||
ts := newLogsTestServer(t, buf)
|
||||
|
||||
res, err := http.Get(ts.URL + "/setup/logs?limit=3")
|
||||
if err != nil {
|
||||
t.Fatalf("GET: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
var resp logsResponse
|
||||
if err := json.NewDecoder(res.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
|
||||
if len(resp.Entries) != 3 {
|
||||
t.Errorf("limit=3 should cap result, got %d entries", len(resp.Entries))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleGetLogs_MalformedSinceReturns400(t *testing.T) {
|
||||
ts := newLogsTestServer(t, logbuf.New(4))
|
||||
|
||||
res, err := http.Get(ts.URL + "/setup/logs?since=notanumber")
|
||||
if err != nil {
|
||||
t.Fatalf("GET: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
if res.StatusCode != http.StatusBadRequest {
|
||||
t.Errorf("expected 400, got %d", res.StatusCode)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleGetLogs_NoBufferAttached(t *testing.T) {
|
||||
ts := newLogsTestServer(t, nil)
|
||||
|
||||
res, err := http.Get(ts.URL + "/setup/logs")
|
||||
if err != nil {
|
||||
t.Fatalf("GET: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
if res.StatusCode != http.StatusOK {
|
||||
t.Fatalf("expected 200 even without buffer, got %d", res.StatusCode)
|
||||
}
|
||||
|
||||
var resp logsResponse
|
||||
if err := json.NewDecoder(res.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
|
||||
if len(resp.Entries) != 0 {
|
||||
t.Errorf("expected empty entries when no buffer, got %d", len(resp.Entries))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleGetLogs_DroppedReportedWhenLagging(t *testing.T) {
|
||||
buf := logbuf.New(3)
|
||||
for i := 0; i < 10; i++ {
|
||||
_, _ = buf.Write([]byte("line\n"))
|
||||
}
|
||||
|
||||
ts := newLogsTestServer(t, buf)
|
||||
|
||||
res, err := http.Get(ts.URL + "/setup/logs?since=2")
|
||||
if err != nil {
|
||||
t.Fatalf("GET: %v", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
var resp logsResponse
|
||||
if err := json.NewDecoder(res.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
|
||||
if resp.Dropped == 0 {
|
||||
t.Errorf("expected dropped > 0 when buffer evicted past `since`, got 0")
|
||||
}
|
||||
}
|
||||
@@ -21,6 +21,7 @@ import (
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/constants"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/datastore"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/health"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/logbuf"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/marge"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/proxy"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/setup"
|
||||
@@ -64,6 +65,7 @@ type Server struct {
|
||||
amazonService *amazon.Service
|
||||
peerObserver *peerObserver
|
||||
healthRegistry *health.Registry
|
||||
logBuf *logbuf.Buffer
|
||||
}
|
||||
|
||||
// RequestSnapshot represents an immutable snapshot of an HTTP request.
|
||||
@@ -147,6 +149,26 @@ func (s *Server) SetVersionInfo(version, commit, date, repoURL string) {
|
||||
s.RepoURL = repoURL
|
||||
}
|
||||
|
||||
// SetLogBuffer attaches a logbuf.Buffer to the server. When set,
|
||||
// HandleGetLogs returns its contents; when nil, the endpoint
|
||||
// reports an empty snapshot. Optional so that tests and
|
||||
// alternative composers (the standalone web binary, etc.) don't
|
||||
// have to construct a buffer they don't need.
|
||||
func (s *Server) SetLogBuffer(buf *logbuf.Buffer) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
s.logBuf = buf
|
||||
}
|
||||
|
||||
// LogBuffer returns the attached log buffer, or nil if none.
|
||||
func (s *Server) LogBuffer() *logbuf.Buffer {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
return s.logBuf
|
||||
}
|
||||
|
||||
// SetDiscoverySettings sets the discovery settings for the server.
|
||||
func (s *Server) SetDiscoverySettings(interval time.Duration, enabled bool) {
|
||||
s.mu.Lock()
|
||||
|
||||
@@ -39,6 +39,9 @@
|
||||
<button class="tab-btn" onclick="openTab(event, 'tab-health')">
|
||||
7. Health
|
||||
</button>
|
||||
<button class="tab-btn" onclick="openTab(event, 'tab-logs')">
|
||||
8. Logs
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<!-- Tab 0: Overview -->
|
||||
@@ -1508,6 +1511,26 @@
|
||||
<div id="health-generated-at" style="font-size: 0.8em; color: #888; margin-bottom: 10px;"></div>
|
||||
<div id="health-findings">Loading…</div>
|
||||
</div>
|
||||
|
||||
<!-- Tab 8: Logs -->
|
||||
<div id="tab-logs" class="tab-content">
|
||||
<div style="display: flex; justify-content: space-between; align-items: center; flex-wrap: wrap; gap: 10px;">
|
||||
<h2 style="margin: 0;">Service Logs</h2>
|
||||
<div style="display: flex; gap: 10px; align-items: center; flex-wrap: wrap;">
|
||||
<input type="text" id="logs-filter" placeholder="Filter (substring, case-insensitive)" style="padding: 4px 8px; min-width: 260px;"/>
|
||||
<label style="font-size: 0.9em;">
|
||||
<input type="checkbox" id="logs-follow" checked/> Follow tail
|
||||
</label>
|
||||
</div>
|
||||
</div>
|
||||
<p style="font-size: 0.9em; color: #555;">
|
||||
Live mirror of the service's stderr log. Stderr still
|
||||
receives every line — this is a read-only view for
|
||||
convenience.
|
||||
</p>
|
||||
<div id="logs-status" style="font-size: 0.8em; color: #888; margin-bottom: 6px;">Idle.</div>
|
||||
<pre id="logs-view" style="background: #111; color: #ddd; padding: 10px; border-radius: 4px; max-height: 60vh; overflow-y: auto; font-size: 0.8em; line-height: 1.35; margin: 0; white-space: pre-wrap; word-break: break-all;"></pre>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<script src="/web/js/script.js"></script>
|
||||
|
||||
@@ -511,6 +511,12 @@ function openTab(evt, tabId) {
|
||||
fetchHealth();
|
||||
}
|
||||
|
||||
if (tabId === "tab-logs") {
|
||||
startLogsPolling();
|
||||
} else {
|
||||
stopLogsPolling();
|
||||
}
|
||||
|
||||
if (evt) {
|
||||
evt.currentTarget.className += " active";
|
||||
let hash = tabId;
|
||||
@@ -4160,3 +4166,136 @@ async function runQuickFix(checkId, fixId, target, confirmMsg, button) {
|
||||
button.disabled = false;
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Logs tab
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const logsState = {
|
||||
timerId: null,
|
||||
nextSince: 0,
|
||||
entries: [],
|
||||
maxEntries: 5000, // client-side cap; UI lag is the bottleneck
|
||||
droppedTotal: 0,
|
||||
pollIntervalMs: 1500,
|
||||
followTail: true,
|
||||
initialised: false,
|
||||
};
|
||||
|
||||
function startLogsPolling() {
|
||||
initLogsTabOnce();
|
||||
// Reset window each time the tab opens so the user gets a
|
||||
// fresh snapshot rather than picking up stale state.
|
||||
logsState.entries = [];
|
||||
logsState.nextSince = 0;
|
||||
logsState.droppedTotal = 0;
|
||||
renderLogs();
|
||||
pollLogsOnce();
|
||||
if (logsState.timerId !== null) clearInterval(logsState.timerId);
|
||||
logsState.timerId = setInterval(pollLogsOnce, logsState.pollIntervalMs);
|
||||
}
|
||||
|
||||
function stopLogsPolling() {
|
||||
if (logsState.timerId !== null) {
|
||||
clearInterval(logsState.timerId);
|
||||
logsState.timerId = null;
|
||||
}
|
||||
}
|
||||
|
||||
function initLogsTabOnce() {
|
||||
if (logsState.initialised) return;
|
||||
logsState.initialised = true;
|
||||
|
||||
const filterEl = document.getElementById("logs-filter");
|
||||
if (filterEl) {
|
||||
filterEl.addEventListener("input", () => renderLogs());
|
||||
}
|
||||
|
||||
const followEl = document.getElementById("logs-follow");
|
||||
if (followEl) {
|
||||
followEl.addEventListener("change", () => {
|
||||
logsState.followTail = followEl.checked;
|
||||
if (logsState.followTail) scrollLogsToBottom();
|
||||
});
|
||||
}
|
||||
|
||||
const viewEl = document.getElementById("logs-view");
|
||||
if (viewEl) {
|
||||
// Disengage follow-tail when the user scrolls up; re-engage
|
||||
// when they're back at the bottom. tail -f muscle memory.
|
||||
viewEl.addEventListener("scroll", () => {
|
||||
const distanceFromBottom = viewEl.scrollHeight - viewEl.scrollTop - viewEl.clientHeight;
|
||||
const atBottom = distanceFromBottom < 8;
|
||||
if (atBottom !== logsState.followTail) {
|
||||
logsState.followTail = atBottom;
|
||||
const followEl = document.getElementById("logs-follow");
|
||||
if (followEl) followEl.checked = atBottom;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function pollLogsOnce() {
|
||||
if (typeof document !== "undefined" && document.hidden) return;
|
||||
|
||||
const statusEl = document.getElementById("logs-status");
|
||||
try {
|
||||
const url = `/setup/logs?since=${encodeURIComponent(logsState.nextSince)}`;
|
||||
const resp = await fetch(url);
|
||||
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
|
||||
const data = await resp.json();
|
||||
|
||||
if (Array.isArray(data.entries) && data.entries.length > 0) {
|
||||
logsState.entries.push(...data.entries);
|
||||
// Trim from the front when we exceed the client cap.
|
||||
if (logsState.entries.length > logsState.maxEntries) {
|
||||
logsState.entries.splice(0, logsState.entries.length - logsState.maxEntries);
|
||||
}
|
||||
}
|
||||
|
||||
if (typeof data.nextSince === "number") {
|
||||
logsState.nextSince = data.nextSince;
|
||||
}
|
||||
|
||||
if (typeof data.dropped === "number" && data.dropped > 0) {
|
||||
logsState.droppedTotal += data.dropped;
|
||||
}
|
||||
|
||||
renderLogs();
|
||||
if (statusEl) {
|
||||
const now = new Date().toLocaleTimeString();
|
||||
const droppedNote = logsState.droppedTotal > 0
|
||||
? ` · ${logsState.droppedTotal} dropped`
|
||||
: "";
|
||||
statusEl.textContent = `${logsState.entries.length} buffered${droppedNote} · last update ${now}`;
|
||||
}
|
||||
} catch (e) {
|
||||
if (statusEl) statusEl.textContent = `Polling failed: ${e.message || e}`;
|
||||
}
|
||||
}
|
||||
|
||||
function renderLogs() {
|
||||
const viewEl = document.getElementById("logs-view");
|
||||
if (!viewEl) return;
|
||||
|
||||
const filterEl = document.getElementById("logs-filter");
|
||||
const filter = filterEl ? filterEl.value.trim().toLowerCase() : "";
|
||||
|
||||
const lines = [];
|
||||
for (const entry of logsState.entries) {
|
||||
if (filter && entry.message.toLowerCase().indexOf(filter) === -1) continue;
|
||||
const ts = entry.time ? entry.time.replace("T", " ").replace("Z", "") : "";
|
||||
lines.push(`${ts} ${entry.message}`);
|
||||
}
|
||||
|
||||
viewEl.textContent = lines.join("\n");
|
||||
|
||||
if (logsState.followTail) {
|
||||
scrollLogsToBottom();
|
||||
}
|
||||
}
|
||||
|
||||
function scrollLogsToBottom() {
|
||||
const viewEl = document.getElementById("logs-view");
|
||||
if (viewEl) viewEl.scrollTop = viewEl.scrollHeight;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
// Package logbuf provides a bounded, in-memory ring buffer for log
|
||||
// lines. It implements io.Writer so it can be installed as a
|
||||
// second sink under stdlib log via log.SetOutput(io.MultiWriter(
|
||||
// os.Stderr, buf)). Each newline-terminated chunk written to the
|
||||
// buffer becomes one Entry with a monotonic sequence number and a
|
||||
// capture timestamp; the admin UI polls these entries and renders
|
||||
// them as a live trace of the running service.
|
||||
//
|
||||
// The buffer holds at most Capacity entries; older entries are
|
||||
// evicted FIFO when new ones arrive. Callers that fall behind can
|
||||
// detect this via the dropped count returned by Since.
|
||||
package logbuf
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Entry is a single buffered log line. Message excludes the
|
||||
// trailing newline. Seq is monotonic across the process lifetime
|
||||
// of the Buffer (it keeps counting up even past evictions, so
|
||||
// clients can use it as a high-water mark).
|
||||
type Entry struct {
|
||||
Seq uint64 `json:"seq"`
|
||||
Time time.Time `json:"time"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
// Buffer is a bounded, concurrent-safe in-memory log ring.
|
||||
type Buffer struct {
|
||||
mu sync.Mutex
|
||||
capacity int
|
||||
entries []Entry // ring; oldest first
|
||||
nextSeq uint64 // seq assigned to the next entry written
|
||||
partial bytes.Buffer
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
// New returns a Buffer that holds up to capacity entries. A
|
||||
// non-positive capacity is clamped to a small default (16) — a
|
||||
// zero-capacity buffer would silently drop every line and make
|
||||
// the feature look broken, which is worse than picking a number.
|
||||
func New(capacity int) *Buffer {
|
||||
if capacity <= 0 {
|
||||
capacity = 16
|
||||
}
|
||||
|
||||
return &Buffer{
|
||||
capacity: capacity,
|
||||
entries: make([]Entry, 0, capacity),
|
||||
nextSeq: 1,
|
||||
now: time.Now,
|
||||
}
|
||||
}
|
||||
|
||||
// Write implements io.Writer. Bytes are split on '\n'; each
|
||||
// complete line is appended as an Entry. A trailing partial line
|
||||
// (no terminating newline) is held until the next Write supplies
|
||||
// the rest. Returns len(p) on success (never reports a short
|
||||
// write — the buffer always accepts the bytes, even when older
|
||||
// entries get evicted).
|
||||
func (b *Buffer) Write(p []byte) (int, error) {
|
||||
if len(p) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
remaining := p
|
||||
for {
|
||||
idx := bytes.IndexByte(remaining, '\n')
|
||||
if idx < 0 {
|
||||
// No newline in the rest of p — stash it.
|
||||
b.partial.Write(remaining)
|
||||
break
|
||||
}
|
||||
|
||||
// We have a complete line: anything held in `partial` +
|
||||
// remaining[:idx]. The newline itself is dropped.
|
||||
var line string
|
||||
|
||||
if b.partial.Len() > 0 {
|
||||
b.partial.Write(remaining[:idx])
|
||||
line = b.partial.String()
|
||||
|
||||
b.partial.Reset()
|
||||
} else {
|
||||
line = string(remaining[:idx])
|
||||
}
|
||||
|
||||
b.append(line)
|
||||
|
||||
remaining = remaining[idx+1:]
|
||||
if len(remaining) == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return len(p), nil
|
||||
}
|
||||
|
||||
// append adds one Entry under the lock, evicting the oldest when
|
||||
// the buffer is at capacity.
|
||||
func (b *Buffer) append(message string) {
|
||||
entry := Entry{
|
||||
Seq: b.nextSeq,
|
||||
Time: b.now().UTC(),
|
||||
Message: message,
|
||||
}
|
||||
b.nextSeq++
|
||||
|
||||
if len(b.entries) < b.capacity {
|
||||
b.entries = append(b.entries, entry)
|
||||
return
|
||||
}
|
||||
|
||||
// At capacity: drop the oldest, slide left, append. A
|
||||
// circular index would avoid the copy, but at 2k entries
|
||||
// the copy is negligible and the slice API is simpler.
|
||||
copy(b.entries, b.entries[1:])
|
||||
b.entries[len(b.entries)-1] = entry
|
||||
}
|
||||
|
||||
// Snapshot returns a copy of every currently buffered entry, in
|
||||
// order of increasing Seq.
|
||||
func (b *Buffer) Snapshot() []Entry {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
out := make([]Entry, len(b.entries))
|
||||
copy(out, b.entries)
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
// Since returns all entries with Seq strictly greater than since,
|
||||
// up to limit. nextSince is the highest Seq among the returned
|
||||
// entries (or the input `since` when none match), suitable for
|
||||
// the caller's next poll. dropped is the count of entries the
|
||||
// caller missed — i.e. entries with Seq > since that have already
|
||||
// been evicted from the buffer. A limit <= 0 means "no limit".
|
||||
func (b *Buffer) Since(since uint64, limit int) (entries []Entry, nextSince, dropped uint64) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
nextSince = since
|
||||
|
||||
if len(b.entries) == 0 {
|
||||
return nil, nextSince, 0
|
||||
}
|
||||
|
||||
oldest := b.entries[0].Seq
|
||||
if since+1 < oldest {
|
||||
dropped = oldest - (since + 1)
|
||||
}
|
||||
|
||||
// Binary-walking the ring isn't worth it at 2k entries.
|
||||
out := make([]Entry, 0, len(b.entries))
|
||||
|
||||
for i := range b.entries {
|
||||
if b.entries[i].Seq <= since {
|
||||
continue
|
||||
}
|
||||
|
||||
out = append(out, b.entries[i])
|
||||
nextSince = b.entries[i].Seq
|
||||
|
||||
if limit > 0 && len(out) >= limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return out, nextSince, dropped
|
||||
}
|
||||
|
||||
// Capacity returns the configured maximum number of entries.
|
||||
func (b *Buffer) Capacity() int {
|
||||
return b.capacity
|
||||
}
|
||||
|
||||
// Len returns the current count of buffered entries.
|
||||
func (b *Buffer) Len() int {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
return len(b.entries)
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
package logbuf
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestBuffer_SingleLineWrite(t *testing.T) {
|
||||
b := New(8)
|
||||
|
||||
n, err := b.Write([]byte("hello\n"))
|
||||
if err != nil {
|
||||
t.Fatalf("Write: %v", err)
|
||||
}
|
||||
|
||||
if n != len("hello\n") {
|
||||
t.Errorf("expected %d bytes consumed, got %d", len("hello\n"), n)
|
||||
}
|
||||
|
||||
got := b.Snapshot()
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("expected 1 entry, got %d", len(got))
|
||||
}
|
||||
|
||||
if got[0].Message != "hello" {
|
||||
t.Errorf("expected message 'hello', got %q", got[0].Message)
|
||||
}
|
||||
|
||||
if got[0].Seq != 1 {
|
||||
t.Errorf("expected seq=1, got %d", got[0].Seq)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_MultiLineSingleWrite(t *testing.T) {
|
||||
b := New(8)
|
||||
_, _ = b.Write([]byte("one\ntwo\nthree\n"))
|
||||
|
||||
got := b.Snapshot()
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("expected 3 entries, got %d", len(got))
|
||||
}
|
||||
|
||||
want := []string{"one", "two", "three"}
|
||||
for i := range got {
|
||||
if got[i].Message != want[i] {
|
||||
t.Errorf("entry %d: want %q, got %q", i, want[i], got[i].Message)
|
||||
}
|
||||
|
||||
if got[i].Seq != uint64(i+1) {
|
||||
t.Errorf("entry %d: want seq=%d, got %d", i, i+1, got[i].Seq)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_PartialLineBuffered(t *testing.T) {
|
||||
b := New(8)
|
||||
|
||||
_, _ = b.Write([]byte("hel"))
|
||||
_, _ = b.Write([]byte("lo "))
|
||||
_, _ = b.Write([]byte("world\n"))
|
||||
|
||||
got := b.Snapshot()
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("expected 1 entry, got %d", len(got))
|
||||
}
|
||||
|
||||
if got[0].Message != "hello world" {
|
||||
t.Errorf("expected reassembled line, got %q", got[0].Message)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_PartialLineDoesNotLeakUntilNewline(t *testing.T) {
|
||||
b := New(8)
|
||||
_, _ = b.Write([]byte("no newline here"))
|
||||
|
||||
if b.Len() != 0 {
|
||||
t.Errorf("expected zero entries for an unterminated write, got %d", b.Len())
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_RingEvictionAtCapacity(t *testing.T) {
|
||||
const capacity = 3
|
||||
|
||||
b := New(capacity)
|
||||
for i := 1; i <= 5; i++ {
|
||||
_, _ = fmt.Fprintf(b, "line %d\n", i)
|
||||
}
|
||||
|
||||
got := b.Snapshot()
|
||||
if len(got) != capacity {
|
||||
t.Fatalf("expected %d entries after eviction, got %d", capacity, len(got))
|
||||
}
|
||||
|
||||
// Oldest surviving entry should be "line 3" with Seq=3.
|
||||
if got[0].Message != "line 3" || got[0].Seq != 3 {
|
||||
t.Errorf("oldest survivor: want line 3/seq 3, got %q/seq %d", got[0].Message, got[0].Seq)
|
||||
}
|
||||
|
||||
if got[capacity-1].Message != "line 5" || got[capacity-1].Seq != 5 {
|
||||
t.Errorf("newest entry: want line 5/seq 5, got %q/seq %d", got[capacity-1].Message, got[capacity-1].Seq)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_SinceFilters(t *testing.T) {
|
||||
b := New(8)
|
||||
for i := 1; i <= 5; i++ {
|
||||
_, _ = fmt.Fprintf(b, "line %d\n", i)
|
||||
}
|
||||
|
||||
entries, nextSince, dropped := b.Since(2, 0)
|
||||
if len(entries) != 3 {
|
||||
t.Fatalf("expected 3 entries since=2, got %d", len(entries))
|
||||
}
|
||||
|
||||
if entries[0].Seq != 3 {
|
||||
t.Errorf("first entry seq: want 3, got %d", entries[0].Seq)
|
||||
}
|
||||
|
||||
if nextSince != 5 {
|
||||
t.Errorf("nextSince: want 5, got %d", nextSince)
|
||||
}
|
||||
|
||||
if dropped != 0 {
|
||||
t.Errorf("dropped: want 0, got %d", dropped)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_SinceReportsDropped(t *testing.T) {
|
||||
b := New(3)
|
||||
for i := 1; i <= 10; i++ {
|
||||
_, _ = fmt.Fprintf(b, "line %d\n", i)
|
||||
}
|
||||
|
||||
// Capacity 3 → only seq 8,9,10 remain. Polling with since=2
|
||||
// means seq 3..7 (five entries) were evicted before we saw them.
|
||||
entries, nextSince, dropped := b.Since(2, 0)
|
||||
if len(entries) != 3 {
|
||||
t.Fatalf("expected 3 entries, got %d", len(entries))
|
||||
}
|
||||
|
||||
if dropped != 5 {
|
||||
t.Errorf("expected dropped=5, got %d", dropped)
|
||||
}
|
||||
|
||||
if nextSince != 10 {
|
||||
t.Errorf("nextSince: want 10, got %d", nextSince)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_SinceRespectsLimit(t *testing.T) {
|
||||
b := New(10)
|
||||
for i := 1; i <= 5; i++ {
|
||||
_, _ = fmt.Fprintf(b, "line %d\n", i)
|
||||
}
|
||||
|
||||
entries, nextSince, _ := b.Since(0, 2)
|
||||
if len(entries) != 2 {
|
||||
t.Fatalf("expected limit=2 to cap result, got %d", len(entries))
|
||||
}
|
||||
|
||||
if nextSince != 2 {
|
||||
t.Errorf("nextSince after limit: want 2, got %d", nextSince)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_SinceNoNewEntries(t *testing.T) {
|
||||
b := New(4)
|
||||
_, _ = b.Write([]byte("only\n"))
|
||||
|
||||
entries, nextSince, dropped := b.Since(5, 0)
|
||||
if len(entries) != 0 {
|
||||
t.Errorf("expected no entries, got %d", len(entries))
|
||||
}
|
||||
|
||||
if nextSince != 5 {
|
||||
t.Errorf("nextSince should pass through since when no matches: want 5, got %d", nextSince)
|
||||
}
|
||||
|
||||
if dropped != 0 {
|
||||
t.Errorf("dropped: want 0, got %d", dropped)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_TimestampMonotonic(t *testing.T) {
|
||||
b := New(4)
|
||||
// Inject a deterministic clock so the test isn't time-flaky.
|
||||
tick := time.Unix(1_700_000_000, 0)
|
||||
b.now = func() time.Time {
|
||||
t := tick
|
||||
tick = tick.Add(time.Millisecond)
|
||||
|
||||
return t
|
||||
}
|
||||
|
||||
_, _ = b.Write([]byte("a\nb\n"))
|
||||
|
||||
got := b.Snapshot()
|
||||
if !got[1].Time.After(got[0].Time) {
|
||||
t.Errorf("expected later entry to have a later timestamp, got %v vs %v", got[1].Time, got[0].Time)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_ConcurrentWrites(t *testing.T) {
|
||||
b := New(10000)
|
||||
|
||||
const writers = 8
|
||||
const perWriter = 500
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < writers; w++ {
|
||||
wg.Add(1)
|
||||
go func(id int) {
|
||||
defer wg.Done()
|
||||
for i := 0; i < perWriter; i++ {
|
||||
_, _ = fmt.Fprintf(b, "w%d-i%d\n", id, i)
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
got := b.Snapshot()
|
||||
if len(got) != writers*perWriter {
|
||||
t.Fatalf("expected %d entries, got %d", writers*perWriter, len(got))
|
||||
}
|
||||
|
||||
// Seq must be strictly increasing and contiguous from 1.
|
||||
for i := range got {
|
||||
if got[i].Seq != uint64(i+1) {
|
||||
t.Fatalf("seq gap at index %d: got %d", i, got[i].Seq)
|
||||
}
|
||||
|
||||
if !strings.HasPrefix(got[i].Message, "w") {
|
||||
t.Errorf("unexpected message: %q", got[i].Message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_ZeroCapacityClamped(t *testing.T) {
|
||||
b := New(0)
|
||||
if b.Capacity() == 0 {
|
||||
t.Errorf("zero capacity should be clamped to a positive default")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuffer_EmptyWriteNoOp(t *testing.T) {
|
||||
b := New(4)
|
||||
|
||||
n, err := b.Write(nil)
|
||||
if err != nil {
|
||||
t.Errorf("nil write: %v", err)
|
||||
}
|
||||
|
||||
if n != 0 {
|
||||
t.Errorf("nil write should consume 0 bytes, got %d", n)
|
||||
}
|
||||
|
||||
if b.Len() != 0 {
|
||||
t.Errorf("buffer should be empty, got %d entries", b.Len())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user