diff --git a/.vscode/settings.json b/.vscode/settings.json index a162b86..4a87807 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,5 +1,8 @@ { "editor.aiStats.enabled": true, "go.lintTool": "golangci-lint", - "go.lintOnSave": "package" + "go.lintOnSave": "package", + "git.scanRepositories": [ + "common" + ] } \ No newline at end of file diff --git a/Makefile b/Makefile index df72ae1..d0ba914 100644 --- a/Makefile +++ b/Makefile @@ -103,6 +103,12 @@ image: vendor coverage: ##@ Calculate test coverage percentage from coverage.out @go tool cover -func=$(REPORTS_DIR)/coverage.out | grep total | awk '{print $$3}' +debug-setup: ##@ Set up local debug environment + ##@ Generates go.work (Go version taken from go.mod) and symlinks the common module for local debugging + @GO_VERSION=$$(grep -m1 '^go ' go.mod | awk '{print $$2}') && \ + printf 'go %s\n\nuse (\n\t.\n\t/opt/shared/common\n)\n' "$$GO_VERSION" > go.work + ln -sfn /opt/shared/common common + ##@ ##@ Misc commands ##@ diff --git a/go.mod b/go.mod index 2594d92..7e9a03d 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/fatih/color v1.18.0 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 github.com/gorilla/mux v1.8.1 - github.com/k8shell-io/common v0.40.0 + github.com/k8shell-io/common v0.45.0 github.com/k8shell-io/k8shell-go v0.2.3 github.com/pkg/sftp v1.13.10 github.com/rs/zerolog v1.34.0 diff --git a/go.sum b/go.sum index 6c8664a..1a4bd87 100644 --- a/go.sum +++ b/go.sum @@ -42,18 +42,10 @@ github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= -github.com/k8shell-io/common v0.36.0 h1:fkMH1XfYRLzDxqhIq5/luHusWWPGnCGJUXSTEIhEDzI= -github.com/k8shell-io/common v0.36.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= -github.com/k8shell-io/common v0.37.0 h1:whq66WosIJECKErUKZF1RQep7tdpOfI6GtP4hXREpsQ= -github.com/k8shell-io/common v0.37.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= -github.com/k8shell-io/common v0.39.0 h1:hfrKZYX2lBonornGrfK35rSY/+5o2sK+SakgUPKDL24= -github.com/k8shell-io/common v0.39.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= -github.com/k8shell-io/common v0.40.0 h1:MhQPVI5oe+JSdRJhWrtvCTJf0MaJDjM58KR7Nq3+lsM= -github.com/k8shell-io/common v0.40.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= -github.com/k8shell-io/k8shell-go v0.2.1 h1:6n88ijXkzP39//lIy4ai3XqtpSUXzoa/dVaWogHQYf4= -github.com/k8shell-io/k8shell-go v0.2.1/go.mod h1:j1JHgUIKIbaiRaitx6Pzw37ahqS4Hu9OcM4uvJ7BP4g= -github.com/k8shell-io/k8shell-go v0.2.2 h1:rwLOeIfyq1+l2Jyv0ak/lXZS7x6xbA5ye0yMsRkTonw= -github.com/k8shell-io/k8shell-go v0.2.2/go.mod h1:ZShnaWs7zxUlNwAkIn4lJodFqaB+PB8O+gn2EIscxq8= +github.com/k8shell-io/common v0.44.0 h1:cry62gDIXW8oyRWCzc2M+Hx62mpnarGdKZRTZ9TV8J8= +github.com/k8shell-io/common v0.44.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= +github.com/k8shell-io/common v0.45.0 h1:1f3QUIU1DhQ/HpBZVO/5BDcKYjf/EO4RKdOXZk33FzI= +github.com/k8shell-io/common v0.45.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= github.com/k8shell-io/k8shell-go v0.2.3 h1:gL7dXDYN4EhWdQvvnY4B2On9Tpb1sZ7G5lO8RtI/nr4= github.com/k8shell-io/k8shell-go v0.2.3/go.mod h1:wWb5gq693qqb48/p5iYrosLG4uNeGOr5dQJiOClIbE8= github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8= diff --git a/internal/grpc/acquire.go b/internal/grpc/acquire.go index d6a9889..ddfc1f5 100644 --- a/internal/grpc/acquire.go +++ b/internal/grpc/acquire.go @@ -105,6 +105,65 @@ func (s *ShellHandler) AcquireSession(ctx context.Context, req *k8shelldv1.Acqui }, nil } +// ListSessions implements SshServiceServer.ListSessions. +// It returns the set of live PTY sessions that AcquireSession would currently +// accept: no client attached and no unexpired lock held, along with the OS +// user each session runs as. +func (s *ShellHandler) ListSessions(_ context.Context, _ *k8shelldv1.ListSessionsRequest) (*k8shelldv1.ListSessionsResponse, error) { + if !s.grpcApi.allowSessionDetach { + return nil, status.Errorf(codes.PermissionDenied, "session attachment is not enabled on this server") + } + + now := time.Now() + locked := make(map[string]bool) + s.grpcApi.SessionLockStore.Range(func(_, v any) bool { + lk := v.(*sessionLock) + if now.Before(lk.expiresAt) { + locked[lk.sessionId] = true + } + return true + }) + + resp := &k8shelldv1.ListSessionsResponse{} + s.grpcApi.SessionStore.Range(func(_, value any) bool { + session, ok := value.(*SessionData) + if !ok || session.ptyDone == nil || !session.Deleted.IsZero() { + return true + } + select { + case <-session.ptyDone: + return true + default: + } + + session.mu.Lock() + attached := session.attachedSender != nil + detachedAt := session.DetachedAt + session.mu.Unlock() + + if attached || locked[session.Id] { + return true + } + + var detachedAtStr string + if !detachedAt.IsZero() { + detachedAtStr = detachedAt.Format(timeFormat) + } + + resp.Sessions = append(resp.Sessions, &k8shelldv1.AcquirableSession{ + SessionId: session.Id, + Owner: session.user.Username, + CmdShell: session.CmdShell, + Pid: int32(session.Pid), + Created: session.Created.Format(timeFormat), + DetachedAt: detachedAtStr, + }) + return true + }) + + return resp, nil +} + // cleanupExpiredLocks removes session locks that have passed their TTL. func (a *GRPCService) cleanupExpiredLocks() { now := time.Now() diff --git a/internal/grpc/ssh.go b/internal/grpc/ssh.go index ef69302..57b7875 100644 --- a/internal/grpc/ssh.go +++ b/internal/grpc/ssh.go @@ -49,6 +49,10 @@ func (s *SshServiceServer) AcquireSession(ctx context.Context, req *k8shelldv1.A return s.shell.AcquireSession(ctx, req) } +func (s *SshServiceServer) ListSessions(ctx context.Context, req *k8shelldv1.ListSessionsRequest) (*k8shelldv1.ListSessionsResponse, error) { + return s.shell.ListSessions(ctx, req) +} + func (s *SshServiceServer) Exec(stream grpc.BidiStreamingServer[k8shelldv1.ExecRequest, k8shelldv1.ExecResponse]) error { return s.exec.Exec(stream) } diff --git a/internal/grpc/system.go b/internal/grpc/system.go index a8391ec..3fb7813 100644 --- a/internal/grpc/system.go +++ b/internal/grpc/system.go @@ -19,6 +19,14 @@ import ( "google.golang.org/grpc/status" ) +// logStreamPollInterval is how often GetLogsStream polls the in-memory log +// store for new entries while following (mirrors the REST /logs handler). +const logStreamPollInterval = 100 * time.Millisecond + +// defaultLogPageLimit is the page size GetLogsPage falls back to when the +// caller doesn't specify one (mirrors the REST /logs handler's default). +const defaultLogPageLimit = 100 + // SystemServiceServer is the gRPC server for the system service type SystemServiceServer struct { grpcApi *GRPCService @@ -124,3 +132,101 @@ func (s *SystemServiceServer) SystemInfo(ctx context.Context, return k8shelld.SystemInfoToProto(&systemInfo), nil } + +// GetLogsStream streams k8shelld daemon logs (the same logs shown by +// `kbox logs`). With Follow=false it sends the currently buffered entries +// and closes the stream; with Follow=true it keeps streaming new entries as +// they are produced until the client cancels. +func (s *SystemServiceServer) GetLogsStream(req *k8shelldv1.SystemLogsStreamRequest, + stream k8shelldv1.SystemService_GetLogsStreamServer) error { + + component := req.GetComponent() + level := k8shelld.LogLevelFromProto(req.GetLevel()) + follow := req.GetFollow() + + send := func(entries []logger.LogEntry) error { + for _, entry := range entries { + if sendErr := stream.Send(logEntryToProto(entry)); sendErr != nil { + return status.Errorf(codes.Canceled, "client canceled") + } + } + return nil + } + + var backlog []logger.LogEntry + var sinceID int64 + if n := req.GetLastN(); n > 0 { + backlog, _ = logger.GetLogsBefore(0, int(n), component, level) + if len(backlog) > 0 { + sinceID = backlog[len(backlog)-1].ID + } + } else { + backlog, sinceID = logger.GetLogsSince(0, component, level) + } + if err := send(backlog); err != nil { + return err + } + + if !follow { + return nil + } + + ctx := stream.Context() + for { + select { + case <-ctx.Done(): + return status.Errorf(codes.Canceled, "client canceled") + default: + entries, newSinceID := logger.GetLogsSince(sinceID, component, level) + if err := send(entries); err != nil { + return err + } + sinceID = newSinceID + + time.Sleep(logStreamPollInterval) + } + } +} + +// GetLogsPage returns one page of k8shelld daemon logs strictly older than +// the requested BeforeId, for "load more" / infinite-scroll style backward +// pagination independent of GetLogsStream's live tail. +func (s *SystemServiceServer) GetLogsPage(ctx context.Context, + req *k8shelldv1.GetLogsPageRequest) (*k8shelldv1.GetLogsPageResponse, error) { + + if req.GetBeforeId() < 0 { + return nil, status.Errorf(codes.InvalidArgument, "before_id must not be negative") + } + if req.GetLimit() < 0 { + return nil, status.Errorf(codes.InvalidArgument, "limit must not be negative") + } + + limit := int(req.GetLimit()) + if limit == 0 { + limit = defaultLogPageLimit + } + + component := req.GetComponent() + level := k8shelld.LogLevelFromProto(req.GetLevel()) + + entries, hasMore := logger.GetLogsBefore(req.GetBeforeId(), limit, component, level) + + resp := &k8shelldv1.GetLogsPageResponse{ + Entries: make([]*k8shelldv1.SystemLogsStreamResponse, 0, len(entries)), + HasMore: hasMore, + } + for _, entry := range entries { + resp.Entries = append(resp.Entries, logEntryToProto(entry)) + } + return resp, nil +} + +func logEntryToProto(entry logger.LogEntry) *k8shelldv1.SystemLogsStreamResponse { + return &k8shelldv1.SystemLogsStreamResponse{ + Id: entry.ID, + Time: entry.Timestamp, + Component: entry.Component, + Level: entry.Level, + Message: entry.Message, + } +} diff --git a/internal/logger/logger.go b/internal/logger/logger.go index 430df28..0d58ad0 100644 --- a/internal/logger/logger.go +++ b/internal/logger/logger.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "io" + "sort" "sync" clogger "github.com/k8shell-io/common/pkg/logger" @@ -17,19 +18,26 @@ const LOGSTORE_CAPACITY = 10000 // Shared memory log store var logStore = &MemoryLogStore{ - entries: make([]logEntry, 0, LOGSTORE_CAPACITY), + entries: make([]LogEntry, 0, LOGSTORE_CAPACITY), cap: LOGSTORE_CAPACITY, } // MemoryLogStore is an in-memory log store that implements io.Writer type MemoryLogStore struct { mu sync.Mutex - entries []logEntry + entries []LogEntry cap int + nextID int64 } -// logEntry represents a single log entry -type logEntry struct { +// LogEntry represents a single log entry. +// +// ID is a per-store, monotonically increasing sequence number assigned on +// write (starting at 1). Unlike a slice index it never shifts as older +// entries are evicted from the buffer, so it's safe to hold onto as a +// pagination cursor across calls. +type LogEntry struct { + ID int64 `json:"id"` Timestamp string `json:"time"` Component string `json:"component"` Level string `json:"level"` @@ -45,7 +53,7 @@ func NewLogger(component string) *zerolog.Logger { // Write implements the io.Writer interface for MemoryLogStore func (s *MemoryLogStore) Write(p []byte) (int, error) { - var entry logEntry + var entry LogEntry if err := json.Unmarshal(p, &entry); err != nil { return 0, fmt.Errorf("failed to unmarshal log entry: %w", err) @@ -54,6 +62,9 @@ func (s *MemoryLogStore) Write(p []byte) (int, error) { s.mu.Lock() defer s.mu.Unlock() + s.nextID++ + entry.ID = s.nextID + if len(s.entries) >= s.cap { s.entries = s.entries[len(s.entries)-s.cap:] } @@ -82,28 +93,67 @@ func InitLogLevel(level string) error { return nil } -// GetLogsSince returns new log entries from the given offset. -func GetLogsSince(offset int, component, level string) ([]logEntry, int) { +// GetLogsSince returns log entries with ID > sinceID (in ID order), along +// with the highest entry ID currently in the store — pass that back in as +// sinceID on the next call to continue tailing without gaps or repeats, +// regardless of how many entries have since been evicted from the buffer. +// sinceID <= 0 returns every entry currently buffered. +func GetLogsSince(sinceID int64, component, level string) ([]LogEntry, int64) { logStore.mu.Lock() defer logStore.mu.Unlock() - if offset >= len(logStore.entries) { - return nil, len(logStore.entries) + lastID := sinceID + if n := len(logStore.entries); n > 0 && logStore.entries[n-1].ID > lastID { + lastID = logStore.entries[n-1].ID } - if offset < 0 { - offset = len(logStore.entries) + offset - if offset < 0 { - offset = 0 + var logs []LogEntry + for _, entry := range logStore.entries { + if entry.ID <= sinceID { + continue + } + if (component == "" || entry.Component == component) && (level == "" || entry.Level == level) { + logs = append(logs, entry) } } + return logs, lastID +} - var logs []logEntry - for i := offset; i < len(logStore.entries); i++ { +// GetLogsBefore returns up to limit log entries with ID < beforeID, oldest +// matching entry first — the "before" half of cursor-based pagination: +// pass the ID of the oldest entry from the previous page back in as +// beforeID to load the page before it. beforeID <= 0 starts from the most +// recent entry (there is no valid entry ID 0, since IDs are assigned +// starting at 1, so it doubles as the "no cursor yet" sentinel for the +// first page). The returned bool reports whether older entries remain +// unscanned in the buffer, i.e. whether a further "load more" call could +// return anything. +func GetLogsBefore(beforeID int64, limit int, component, level string) ([]LogEntry, bool) { + logStore.mu.Lock() + defer logStore.mu.Unlock() + + end := len(logStore.entries) + if beforeID > 0 { + end = sort.Search(end, func(i int) bool { + return logStore.entries[i].ID >= beforeID + }) + } + + var logs []LogEntry + hasMore := false + for i := end - 1; i >= 0; i-- { + if len(logs) == limit { + hasMore = true + break + } entry := logStore.entries[i] if (component == "" || entry.Component == component) && (level == "" || entry.Level == level) { logs = append(logs, entry) } } - return logs, len(logStore.entries) + + for i, j := 0, len(logs)-1; i < j; i, j = i+1, j-1 { + logs[i], logs[j] = logs[j], logs[i] + } + return logs, hasMore } diff --git a/internal/logger/logger_test.go b/internal/logger/logger_test.go index 4d1cb9d..df16245 100644 --- a/internal/logger/logger_test.go +++ b/internal/logger/logger_test.go @@ -24,11 +24,11 @@ func TestMemoryLogStore_Initialization(t *testing.T) { func TestMemoryLogStore_Write(t *testing.T) { // Create a test store store := &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } - logData := logEntry{ + logData := LogEntry{ Timestamp: "2025-12-18T10:00:00Z", Component: "test-component", Level: "info", @@ -63,7 +63,7 @@ func TestMemoryLogStore_Write(t *testing.T) { func TestMemoryLogStore_Write_InvalidJSON(t *testing.T) { store := &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } @@ -77,13 +77,13 @@ func TestMemoryLogStore_Write_InvalidJSON(t *testing.T) { func TestMemoryLogStore_CapacityLimit(t *testing.T) { capacity := 5 store := &MemoryLogStore{ - entries: make([]logEntry, 0, capacity), + entries: make([]LogEntry, 0, capacity), cap: capacity, } // Write more entries than capacity for i := 0; i < 10; i++ { - logData := logEntry{ + logData := LogEntry{ Timestamp: "2025-12-18T10:00:00Z", Component: "test", Level: "info", @@ -129,12 +129,12 @@ func TestGetLogsSince_Basic(t *testing.T) { // Create test store logStore = &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } // Add some test entries - entries := []logEntry{ + entries := []LogEntry{ {Timestamp: "2025-12-18T10:00:00Z", Component: "comp1", Level: "info", Message: "msg1"}, {Timestamp: "2025-12-18T10:01:00Z", Component: "comp2", Level: "debug", Message: "msg2"}, {Timestamp: "2025-12-18T10:02:00Z", Component: "comp1", Level: "error", Message: "msg3"}, @@ -166,13 +166,13 @@ func TestGetLogsSince_WithOffset(t *testing.T) { // Create test store logStore = &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } // Add test entries for i := 0; i < 5; i++ { - entry := logEntry{ + entry := LogEntry{ Timestamp: "2025-12-18T10:00:00Z", Component: "test", Level: "info", @@ -200,20 +200,20 @@ func TestGetLogsSince_WithOffset(t *testing.T) { } } -func TestGetLogsSince_NegativeOffset(t *testing.T) { +func TestGetLogsBefore_MostRecent(t *testing.T) { // Save and restore original logStore originalStore := logStore defer func() { logStore = originalStore }() // Create test store logStore = &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } // Add test entries for i := 0; i < 5; i++ { - entry := logEntry{ + entry := LogEntry{ Timestamp: "2025-12-18T10:00:00Z", Component: "test", Level: "info", @@ -226,19 +226,107 @@ func TestGetLogsSince_NegativeOffset(t *testing.T) { } } - // -2 should get last 2 entries - logs, newOffset := GetLogsSince(-2, "", "") + // beforeID <= 0 should get the last 2 entries, oldest-first + logs, hasMore := GetLogsBefore(0, 2, "", "") if len(logs) != 2 { - t.Errorf("expected 2 logs with offset -2, got %d", len(logs)) + t.Fatalf("expected 2 logs, got %d", len(logs)) } - if newOffset != 5 { - t.Errorf("expected newOffset 5, got %d", newOffset) + if !hasMore { + t.Error("expected hasMore=true, since 3 older entries remain") } - if logs[0].Message != "D" { - t.Errorf("expected first log message 'D', got '%s'", logs[0].Message) + if logs[0].Message != "D" || logs[1].Message != "E" { + t.Errorf("expected [D, E], got [%s, %s]", logs[0].Message, logs[1].Message) + } +} + +func TestGetLogsBefore_Pagination(t *testing.T) { + // Save and restore original logStore + originalStore := logStore + defer func() { logStore = originalStore }() + + // Create test store + logStore = &MemoryLogStore{ + entries: make([]LogEntry, 0, 10), + cap: 10, + } + + // Add 5 entries: A..E, IDs 1..5 + for i := 0; i < 5; i++ { + entry := LogEntry{ + Timestamp: "2025-12-18T10:00:00Z", + Component: "test", + Level: "info", + Message: string(rune('A' + i)), + } + data, _ := json.Marshal(entry) + _, err := logStore.Write(data) + if err != nil { + t.Fatalf("failed to write log entry: %v", err) + } + } + + // Page 1: most recent 2 -> [D, E], more remain (A, B, C) + page1, hasMore1 := GetLogsBefore(0, 2, "", "") + if len(page1) != 2 || page1[0].Message != "D" || page1[1].Message != "E" || !hasMore1 { + t.Fatalf("unexpected page1: %+v hasMore=%v", page1, hasMore1) + } + + // Page 2: before the oldest entry of page1 -> [B, C], A remains + page2, hasMore2 := GetLogsBefore(page1[0].ID, 2, "", "") + if len(page2) != 2 || page2[0].Message != "B" || page2[1].Message != "C" || !hasMore2 { + t.Fatalf("unexpected page2: %+v hasMore=%v", page2, hasMore2) + } + + // Page 3: before the oldest entry of page2 -> [A], nothing left + page3, hasMore3 := GetLogsBefore(page2[0].ID, 2, "", "") + if len(page3) != 1 || page3[0].Message != "A" || hasMore3 { + t.Fatalf("unexpected page3: %+v hasMore=%v", page3, hasMore3) + } +} + +// TestGetLogsBefore_StableAcrossEviction verifies that a cursor obtained +// before older entries are evicted from the buffer still lands in the +// right place afterwards — the whole point of using a stable sequence id +// instead of a slice index, which would shift under eviction. +func TestGetLogsBefore_StableAcrossEviction(t *testing.T) { + // Save and restore original logStore + originalStore := logStore + defer func() { logStore = originalStore }() + + capacity := 5 + logStore = &MemoryLogStore{ + entries: make([]LogEntry, 0, capacity), + cap: capacity, + } + + // Write 10 entries (IDs 1..10); with capacity 5 the store retains the + // last 6 (see TestMemoryLogStore_CapacityLimit), i.e. messages E..J + // with IDs 5..10. + for i := 0; i < 10; i++ { + entry := LogEntry{ + Timestamp: "2025-12-18T10:00:00Z", + Component: "test", + Level: "info", + Message: string(rune('A' + i)), + } + data, _ := json.Marshal(entry) + if _, err := logStore.Write(data); err != nil { + t.Fatalf("Write failed at iteration %d: %v", i, err) + } + } + + // Cursor id 8 ("H") predates eviction of "A"-"D"; paging before it must + // still resolve correctly against what remains in the buffer: entries + // with ID < 8 that are still present are E(5), F(6), G(7). + logs, hasMore := GetLogsBefore(8, 10, "", "") + if len(logs) != 3 || logs[0].Message != "E" || logs[1].Message != "F" || logs[2].Message != "G" { + t.Fatalf("expected [E, F, G], got %+v", logs) + } + if hasMore { + t.Error("expected hasMore=false, buffer has no entries older than E") } } @@ -249,11 +337,11 @@ func TestGetLogsSince_ComponentFilter(t *testing.T) { // Create test store logStore = &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } - entries := []logEntry{ + entries := []LogEntry{ {Timestamp: "2025-12-18T10:00:00Z", Component: "comp1", Level: "info", Message: "msg1"}, {Timestamp: "2025-12-18T10:01:00Z", Component: "comp2", Level: "info", Message: "msg2"}, {Timestamp: "2025-12-18T10:02:00Z", Component: "comp1", Level: "info", Message: "msg3"}, @@ -292,11 +380,11 @@ func TestGetLogsSince_LevelFilter(t *testing.T) { // Create test store logStore = &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } - entries := []logEntry{ + entries := []LogEntry{ {Timestamp: "2025-12-18T10:00:00Z", Component: "comp1", Level: "info", Message: "msg1"}, {Timestamp: "2025-12-18T10:01:00Z", Component: "comp1", Level: "error", Message: "msg2"}, {Timestamp: "2025-12-18T10:02:00Z", Component: "comp1", Level: "info", Message: "msg3"}, @@ -333,11 +421,11 @@ func TestGetLogsSince_CombinedFilters(t *testing.T) { // Create test store logStore = &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } - entries := []logEntry{ + entries := []LogEntry{ {Timestamp: "2025-12-18T10:00:00Z", Component: "comp1", Level: "info", Message: "msg1"}, {Timestamp: "2025-12-18T10:01:00Z", Component: "comp2", Level: "error", Message: "msg2"}, {Timestamp: "2025-12-18T10:02:00Z", Component: "comp1", Level: "error", Message: "msg3"}, @@ -363,20 +451,20 @@ func TestGetLogsSince_CombinedFilters(t *testing.T) { } } -func TestGetLogsSince_OffsetBeyondLength(t *testing.T) { +func TestGetLogsSince_SinceIDBeyondLatest(t *testing.T) { // Save and restore original logStore originalStore := logStore defer func() { logStore = originalStore }() // Create test store logStore = &MemoryLogStore{ - entries: make([]logEntry, 0, 10), + entries: make([]LogEntry, 0, 10), cap: 10, } - // Add 3 entries + // Add 3 entries (IDs 1..3) for i := 0; i < 3; i++ { - entry := logEntry{ + entry := LogEntry{ Timestamp: "2025-12-18T10:00:00Z", Component: "test", Level: "info", @@ -389,14 +477,16 @@ func TestGetLogsSince_OffsetBeyondLength(t *testing.T) { } } - logs, newOffset := GetLogsSince(10, "", "") + // A sinceID beyond the latest entry id has nothing new to return, and + // the cursor doesn't move backward. + logs, lastID := GetLogsSince(10, "", "") if logs != nil { - t.Errorf("expected nil logs for offset beyond length, got %d logs", len(logs)) + t.Errorf("expected nil logs for sinceID beyond the latest entry, got %d logs", len(logs)) } - if newOffset != 3 { - t.Errorf("expected newOffset 3, got %d", newOffset) + if lastID != 10 { + t.Errorf("expected lastID to stay at 10, got %d", lastID) } } diff --git a/internal/server/restapi.go b/internal/server/restapi.go index eed1185..afe6d42 100644 --- a/internal/server/restapi.go +++ b/internal/server/restapi.go @@ -449,26 +449,59 @@ func (a *RESTService) GetSplash(w http.ResponseWriter, r *http.Request) { } } +// defaultLogPageSize is the page size used for cursor-based pagination +// (the "before" query parameter) when the caller doesn't specify a 'limit'. +const defaultLogPageSize = 100 + +// GetLogs serves k8shelld daemon logs (the same logs shown by `kbox logs`). +// +// Query parameters: +// - component, level: optional filters +// - follow=true: keep the connection open and stream new entries as they +// arrive, seeded with the last 'limit' entries (or everything currently +// buffered if 'limit' is unset) +// - before=: cursor-based pagination — returns up to 'limit' entries +// older than the given entry id, oldest-first, for "load more" style +// paging that stays correct even as the in-memory buffer evicts old +// entries underneath it. Omit (or 0) for the most recent page. The +// response carries an 'X-Log-Has-More' header (true/false) indicating +// whether an older page exists; the next 'before' cursor is simply the +// 'id' of the oldest entry in the response body. +// - limit (alias: lastN, kept for backward compatibility): page size for +// 'before' pagination, or entry count for the initial/follow backlog. +// Omitted with no 'before' returns everything currently buffered. func (a *RESTService) GetLogs(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/x-ndjson") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") - component := r.URL.Query().Get("component") - level := r.URL.Query().Get("level") - follow := r.URL.Query().Get("follow") == "true" - lastN := r.URL.Query().Get("lastN") + q := r.URL.Query() + component := q.Get("component") + level := q.Get("level") + follow := q.Get("follow") == "true" - var err error - var n int - if lastN != "" { - n, err = strconv.Atoi(lastN) + limitStr := q.Get("limit") + if limitStr == "" { + limitStr = q.Get("lastN") + } + var limit int + if limitStr != "" { + n, err := strconv.Atoi(limitStr) if err != nil || n < 0 { - http.Error(w, "Invalid 'lastN' parameter", http.StatusBadRequest) + http.Error(w, "Invalid 'limit' parameter", http.StatusBadRequest) return } - } else { - n = 0 + limit = n + } + + var beforeID int64 + if beforeStr := q.Get("before"); beforeStr != "" { + id, err := strconv.ParseInt(beforeStr, 10, 64) + if err != nil || id < 0 { + http.Error(w, "Invalid 'before' parameter", http.StatusBadRequest) + return + } + beforeID = id } flusher, ok := w.(http.Flusher) @@ -477,37 +510,62 @@ func (a *RESTService) GetLogs(w http.ResponseWriter, r *http.Request) { return } - offset := 0 - if n > 0 { - offset = -n + writeEntries := func(entries []logger.LogEntry) { + for _, entry := range entries { + b, err := json.Marshal(entry) + if err != nil { + a.logger.Error().Err(err).Msg("failed to encode log entry") + continue + } + _, _ = fmt.Fprintln(w, string(b)) + } } - for { - select { - case <-r.Context().Done(): - return - default: - entries, newOffset := logger.GetLogsSince(offset, component, level) - - for _, entry := range entries { - b, err := json.Marshal(entry) - if err != nil { - a.logger.Error().Err(err).Msg("failed to encode log entry") - continue - } - _, _ = fmt.Fprintln(w, string(b)) + if follow { + var backlog []logger.LogEntry + var sinceID int64 + if limit > 0 { + backlog, _ = logger.GetLogsBefore(0, limit, component, level) + if len(backlog) > 0 { + sinceID = backlog[len(backlog)-1].ID } + } else { + backlog, sinceID = logger.GetLogsSince(0, component, level) + } + writeEntries(backlog) + flusher.Flush() - flusher.Flush() - offset = newOffset - - if !follow { + for { + select { + case <-r.Context().Done(): return + default: + entries, newSinceID := logger.GetLogsSince(sinceID, component, level) + if len(entries) > 0 { + writeEntries(entries) + flusher.Flush() + } + sinceID = newSinceID + time.Sleep(100 * time.Millisecond) } + } + } - time.Sleep(100 * time.Millisecond) + var entries []logger.LogEntry + if beforeID > 0 || limit > 0 { + pageLimit := limit + if pageLimit == 0 { + pageLimit = defaultLogPageSize } + var hasMore bool + entries, hasMore = logger.GetLogsBefore(beforeID, pageLimit, component, level) + w.Header().Set("X-Log-Has-More", strconv.FormatBool(hasMore)) + } else { + entries, _ = logger.GetLogsSince(0, component, level) } + + writeEntries(entries) + flusher.Flush() } func (a *RESTService) ValidateK8shelldFile(w http.ResponseWriter, r *http.Request) {