diff --git a/Makefile b/Makefile index 824fa4a..aa2fc5d 100644 --- a/Makefile +++ b/Makefile @@ -33,7 +33,7 @@ install: build # Uninstall the binary uninstall: - rm -f $(BINDIR)/$(BINARY_PATH) + rm -f $(BINDIR)/$(BINARY_NAME) # Clean build artifacts clean: diff --git a/README.md b/README.md index c6f7817..66c0481 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@

LogWisp

- Go + Go License Documentation

@@ -137,7 +137,7 @@ synthetic generator writing JSON to stdout. - **Operating systems**: Linux (kernel 6.10+), FreeBSD (14.0+) - **Architecture**: amd64 -- **Go**: 1.26+ to build from source +- **Go**: 1.27.1+ to build from source Network sources and sinks bind and dial over IPv4 only. diff --git a/doc/installation.md b/doc/installation.md index 0f7b296..583edad 100644 --- a/doc/installation.md +++ b/doc/installation.md @@ -22,7 +22,7 @@ The Makefile works with both GNU make and BSD make. Targets: | `make` / `make build` | Build `bin/logwisp` with version metadata | | `make dev` | Build with the race detector enabled | | `make install` | Install the binary to `$(PREFIX)/bin` (default `/usr/local`) | -| `make uninstall` | Intended to remove the installed binary — currently broken: it expands to `$(BINDIR)/bin/logwisp` instead of `$(BINDIR)/logwisp`, so it removes nothing. Delete the binary by hand | +| `make uninstall` | Remove `$(BINDIR)/logwisp` | | `make clean` | Remove the built binary | | `make version` | Print the version, commit, and build time that would be embedded | diff --git a/doc/operations.md b/doc/operations.md index 892fd58..48aa92b 100644 --- a/doc/operations.md +++ b/doc/operations.md @@ -124,6 +124,9 @@ curl -s http://127.0.0.1:8080/status | jq . "tls": false, "active_clients": 3, "buffer_size": 1000, + "client_buffer_size": 256, + "max_connections": 32, + "write_timeout_ms": 5000, "uptime_seconds": 8130 }, "endpoints": { "stream": "/stream", "status": "/status" }, @@ -183,8 +186,10 @@ the filter stage logs several lines per entry evaluated. Raise `buffer_size` when `total_dropped_by_sink` is climbing but the sink itself is healthy — that is a burst-absorption problem. Raise `client_buffer_size` when -`dropped_writes` is climbing for network sinks; that is a slow-consumer problem, -and a bigger buffer only buys time. +`dropped_writes` is climbing for network sinks; that is a slow-consumer or +startup-replay burst problem, and a bigger buffer only buys time. The HTTP +status endpoint reports both queue bounds alongside the counters so an operator +can distinguish configuration from demand. ```toml [pipelines.plugin_sinks.config] diff --git a/doc/sinks.md b/doc/sinks.md index 7e3bbe8..e610d91 100644 --- a/doc/sinks.md +++ b/doc/sinks.md @@ -160,9 +160,10 @@ allow = ["viewer-01"] - HTTP/2 is negotiated via ALPN when TLS is enabled; plaintext is HTTP/1.1. **Status endpoint** returns service and version identity, host, port, TLS flag, -the compiled auth policy, active client count, buffer size, uptime, endpoint -paths, and the `total_processed` / `dropped_writes` / `rejected_clients` / -`auth_rejected` counters. +the compiled auth policy, active client count, sink and per-client buffer sizes, +connection limit, write timeout, uptime, endpoint paths, and the +`total_processed` / `dropped_writes` / `rejected_clients` / `auth_rejected` +counters. > Without an `auth` block both endpoints are unauthenticated, and the stream > response carries `Access-Control-Allow-Origin: *`, so any web origin can read diff --git a/internal/sink/http/http.go b/internal/sink/http/http.go index bd55828..ffccac8 100644 --- a/internal/sink/http/http.go +++ b/internal/sink/http/http.go @@ -452,14 +452,17 @@ func (h *HTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) { "version": version.Short(), "instance_id": h.id, "server": map[string]any{ - "type": "http", - "host": h.config.Host, - "port": h.config.Port, - "tls": h.tlsConfig != nil, - "auth": h.auth.Describe(), - "active_clients": h.activeClients.Load(), - "buffer_size": h.config.BufferSize, - "uptime_seconds": int(time.Since(h.startTime).Seconds()), + "type": "http", + "host": h.config.Host, + "port": h.config.Port, + "tls": h.tlsConfig != nil, + "auth": h.auth.Describe(), + "active_clients": h.activeClients.Load(), + "buffer_size": h.config.BufferSize, + "client_buffer_size": h.config.ClientBufferSize, + "max_connections": h.config.MaxConnections, + "write_timeout_ms": h.config.WriteTimeoutMS, + "uptime_seconds": int(time.Since(h.startTime).Seconds()), }, "endpoints": map[string]string{ "stream": h.config.StreamPath, @@ -481,12 +484,15 @@ func (h *HTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) { func (h *HTTPSink) GetStats() sink.SinkStats { lastProc, _ := h.lastProcessed.Load().(time.Time) details := map[string]any{ - "host": h.config.Host, - "port": h.config.Port, - "buffer_size": h.config.BufferSize, - "tls": h.tlsConfig != nil, - "dropped_writes": h.droppedWrites.Load(), - "rejected_clients": h.rejectedClients.Load(), + "host": h.config.Host, + "port": h.config.Port, + "buffer_size": h.config.BufferSize, + "client_buffer_size": h.config.ClientBufferSize, + "max_connections": h.config.MaxConnections, + "write_timeout_ms": h.config.WriteTimeoutMS, + "tls": h.tlsConfig != nil, + "dropped_writes": h.droppedWrites.Load(), + "rejected_clients": h.rejectedClients.Load(), "endpoints": map[string]string{ "stream": h.config.StreamPath, "status": h.config.StatusPath, diff --git a/internal/sink/http/http_test.go b/internal/sink/http/http_test.go new file mode 100644 index 0000000..b01b1c8 --- /dev/null +++ b/internal/sink/http/http_test.go @@ -0,0 +1,75 @@ +package http + +import ( + "encoding/json" + "net/http/httptest" + "testing" + "time" + + "logwisp/internal/session" + "logwisp/internal/sink" + + "github.com/lixenwraith/log" +) + +func TestStatusReportsQueueAndConnectionBounds(t *testing.T) { + manager := session.NewManager(time.Hour) + defer manager.Stop() + created, err := NewHTTPSinkPlugin( + "stream", + map[string]any{ + "host": "127.0.0.1", + "port": int64(8081), + "buffer_size": int64(4096), + "client_buffer_size": int64(512), + "max_connections": int64(32), + "write_timeout_ms": int64(5000), + }, + log.NewLogger(), + session.NewProxy(manager, "stream"), + ) + if err != nil { + t.Fatal(err) + } + httpSink, ok := created.(*HTTPSink) + if !ok { + t.Fatalf("sink type = %T", created) + } + + recorder := httptest.NewRecorder() + httpSink.handleStatus(recorder, httptest.NewRequest("GET", "/status", nil)) + if recorder.Code != 200 { + t.Fatalf("status code = %d", recorder.Code) + } + var response struct { + Server map[string]any `json:"server"` + } + if err := json.NewDecoder(recorder.Body).Decode(&response); err != nil { + t.Fatal(err) + } + for key, want := range map[string]float64{ + "buffer_size": 4096, + "client_buffer_size": 512, + "max_connections": 32, + "write_timeout_ms": 5000, + } { + if got := response.Server[key]; got != want { + t.Errorf("server.%s = %v, want %v", key, got, want) + } + } + + stats := httpSink.GetStats() + details := stats.Details + for key, want := range map[string]int64{ + "buffer_size": 4096, + "client_buffer_size": 512, + "max_connections": 32, + "write_timeout_ms": 5000, + } { + if got := details[key]; got != want { + t.Errorf("details[%q] = %v, want %v", key, got, want) + } + } + + var _ sink.Sink = httpSink +} diff --git a/internal/source/file/file.go b/internal/source/file/file.go index a63aeb9..76111cb 100644 --- a/internal/source/file/file.go +++ b/internal/source/file/file.go @@ -292,12 +292,21 @@ func (fs *FileSource) ensureWatcher(path string) { } } - fs.mu.Lock() - delete(fs.watchers, path) - fs.mu.Unlock() + fs.removeWatcher(path, w) }() } +// removeWatcher removes only the watcher that finished. A deleted file can be +// recreated before its old watcher observes stop; in that case ensureWatcher +// has already installed a replacement under the same path, which must survive. +func (fs *FileSource) removeWatcher(path string, watcher *fileWatcher) { + fs.mu.Lock() + if fs.watchers[path] == watcher { + delete(fs.watchers, path) + } + fs.mu.Unlock() +} + // cleanupWatchers stops and removes watchers for files that no longer exist. func (fs *FileSource) cleanupWatchers() { fs.mu.Lock() @@ -369,4 +378,3 @@ func globToRegex(glob string) string { regex = strings.ReplaceAll(regex, `\?`, `.`) return "^" + regex + "$" } - diff --git a/internal/source/file/file_test.go b/internal/source/file/file_test.go new file mode 100644 index 0000000..3f6b323 --- /dev/null +++ b/internal/source/file/file_test.go @@ -0,0 +1,46 @@ +package file + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "logwisp/internal/core" +) + +func TestStoppedWatcherReturnsNormally(t *testing.T) { + path := filepath.Join(t.TempDir(), "session.jsonl") + if err := os.WriteFile(path, nil, 0o600); err != nil { + t.Fatal(err) + } + watcher := newFileWatcher(path, true, true, func(_ core.LogEntry) {}, nil) + watcher.stop() + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := watcher.watch(ctx); err != nil { + t.Fatalf("stopped watcher returned error: %v", err) + } +} + +func TestRemoveWatcherPreservesReplacement(t *testing.T) { + oldWatcher := &fileWatcher{} + replacement := &fileWatcher{} + source := &FileSource{ + watchers: map[string]*fileWatcher{ + "session.jsonl": replacement, + }, + } + + source.removeWatcher("session.jsonl", oldWatcher) + if got := source.watchers["session.jsonl"]; got != replacement { + t.Fatalf("replacement watcher = %p, want %p", got, replacement) + } + + source.removeWatcher("session.jsonl", replacement) + if _, exists := source.watchers["session.jsonl"]; exists { + t.Fatal("finished watcher was not removed") + } +} diff --git a/internal/source/file/file_watcher.go b/internal/source/file/file_watcher.go index 8af71eb..8c71bf5 100644 --- a/internal/source/file/file_watcher.go +++ b/internal/source/file/file_watcher.go @@ -80,7 +80,7 @@ func (w *fileWatcher) watch(ctx context.Context) error { return ctx.Err() case <-ticker.C: if w.isStopped() { - return fmt.Errorf("watcher stopped") + return nil } if err := w.checkFile(); err != nil { // Log error but continue watching