admin343/browser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: bound management terminal logs

Browse Source
This commit is contained in:
npc0-hue
2026-08-07 10:43:13 +08:00
parent 4b0233cf6f
commit 87e903998e
10 changed files with 229 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/api/log_events_handlers.go
+40 -20
View File
@@ -4,6 +4,7 @@ import (
"encoding/json"
"fmt"
"net/http"
"sort"
"strconv"
"strings"
"time"
@@ -15,7 +16,7 @@ import (
const (
defaultLogEventHistoryLimit = 100
maxLogEventHistoryLimit = 500
maxLogEventHistoryLimit = 10000
logEventHeartbeatInterval = 15 * time.Second
)
@@ -25,7 +26,7 @@ const (
// @Tags logs
// @Produce text/event-stream
// @Param id path string true "Server instance ID"
// @Param historyLimit query int false "Recent entries per stream to replay before live events"
// @Param historyLimit query int false "Total recent entries to replay across this server's streams"
// @Success 200 {string} string "event-stream"
// @Failure 401 {object} dto.ErrorResponse
// @Failure 403 {object} dto.ErrorResponse
@@ -61,10 +62,16 @@ func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(stream)); err != nil {
return
}
if historyLimit > 0 {
if err := h.writeLogEventHistory(w, stream, historyLimit); err != nil {
_ = writeSSEJSON(w, "error", "", map[string]string{"message": err.Error()})
flusher.Flush()
}
if historyLimit > 0 {
history, err := h.loadLogEventHistory(streams, historyLimit)
if err != nil {
_ = writeSSEJSON(w, "error", "", map[string]string{"message": err.Error()})
flusher.Flush()
return
}
for _, event := range history {
if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil {
return
}
}
@@ -127,22 +134,35 @@ func (h *coreHandlers) openLogEventSubscription(r *http.Request) (domain.ServerI
return instance, streams, subscription, err
}
func (h *coreHandlers) writeLogEventHistory(w http.ResponseWriter, stream domain.LogStream, limit int) error {
afterSeq := uint64(0)
if stream.LatestSeq > uint64(limit) {
afterSeq = stream.LatestSeq - uint64(limit)
}
cursor, err := h.core.QueryLogStream(domain.LogStreamCursorQuery{LogStreamID: stream.ID, AfterSeq: afterSeq, Limit: limit})
if err != nil {
return err
}
for _, entry := range cursor.Entries {
event := domain.LogStreamEvent{ServerInstanceID: stream.ServerInstanceID, Stream: stream, Entry: entry, LatestSeq: cursor.LatestSeq}
if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil {
return err
func (h *coreHandlers) loadLogEventHistory(streams []domain.LogStream, limit int) ([]domain.LogStreamEvent, error) {
history := make([]domain.LogStreamEvent, 0, limit)
for _, stream := range streams {
afterSeq := uint64(0)
if stream.LatestSeq > uint64(limit) {
afterSeq = stream.LatestSeq - uint64(limit)
}
cursor, err := h.core.QueryLogStream(domain.LogStreamCursorQuery{LogStreamID: stream.ID, AfterSeq: afterSeq, Limit: limit})
if err != nil {
return nil, err
}
for _, entry := range cursor.Entries {
history = append(history, domain.LogStreamEvent{ServerInstanceID: stream.ServerInstanceID, Stream: stream, Entry: entry, LatestSeq: cursor.LatestSeq})
}
}
return nil
sort.SliceStable(history, func(i, j int) bool {
left, right := history[i], history[j]
if !left.Entry.Timestamp.Equal(right.Entry.Timestamp) {
return left.Entry.Timestamp.Before(right.Entry.Timestamp)
}
if left.Entry.Seq != right.Entry.Seq {
return left.Entry.Seq < right.Entry.Seq
}
return left.Stream.ID < right.Stream.ID
})
if len(history) > limit {
history = history[len(history)-limit:]
}
return history, nil
}
func parseLogEventHistoryLimit(value string) int {
platform/api/log_ingest_handlers_test.go
+35 -3
View File
@@ -64,6 +64,34 @@ func TestLogEventsSSEReplaysHistory(t *testing.T) {
}
}
func TestLogEventsSSEUsesServerWideNewestHistory(t *testing.T) {
router := newTestRouter()
hello := createLogIngestAPIFixtures(t, router)
postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{
ID: "log-2", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "stderr",
StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default",
})
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequestForStream(t, hello.SessionToken, "log-1", "stdout", 1, 2, 0)), http.StatusOK)
assertStatus(t, performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", validLogBatchRequestForStream(t, hello.SessionToken, "log-2", "stderr", 1, 2, 10)), http.StatusOK)
server := httptest.NewServer(router)
defer server.Close()
client := server.Client()
client.Timeout = 2 * time.Second
response, err := client.Get(server.URL + "/api/v1/server-instances/server-1/logs/events?historyLimit=2")
if err != nil {
t.Fatalf("open log event stream: %v", err)
}
defer response.Body.Close()
body := readSSEUntil(t, response, "event: ready")
if strings.Contains(body, `"streamId":"log-1"`) {
t.Fatalf("expected no history entries from older stream, got:\n%s", body)
}
if strings.Count(body, `"streamId":"log-2"`) != 2 || !strings.Contains(body, `"seq":2`) {
t.Fatalf("expected newest two entries from stream log-2, got:\n%s", body)
}
}
func TestLogIngestAPIDuplicateAndErrors(t *testing.T) {
router := newTestRouter()
hello := createLogIngestAPIFixtures(t, router)
@@ -121,11 +149,15 @@ func readSSEUntil(t *testing.T, response *http.Response, marker string) string {
}
func validLogBatchRequest(t *testing.T, sessionToken string, firstSeq uint64, lastSeq uint64) dto.LogBatchIngestRequest {
return validLogBatchRequestForStream(t, sessionToken, "log-1", "stdout", firstSeq, lastSeq, 0)
}
func validLogBatchRequestForStream(t *testing.T, sessionToken string, streamID string, streamKey string, firstSeq uint64, lastSeq uint64, timestampOffset int) dto.LogBatchIngestRequest {
t.Helper()
entries := make([]dto.LogEntryBody, 0, lastSeq-firstSeq+1)
domainEntries := make([]domain.LogEntry, 0, lastSeq-firstSeq+1)
for seq := firstSeq; seq <= lastSeq; seq++ {
entry := dto.LogEntryBody{Seq: seq, Timestamp: time.Date(2026, 7, 3, 12, 0, int(seq), 0, time.UTC), Level: "info", Line: "line"}
entry := dto.LogEntryBody{Seq: seq, Timestamp: time.Date(2026, 7, 3, 12, 0, timestampOffset+int(seq), 0, time.UTC), Level: "info", Line: "line"}
entries = append(entries, entry)
domainEntries = append(domainEntries, domain.LogEntry{Seq: entry.Seq, Timestamp: entry.Timestamp, Level: entry.Level, Line: entry.Line})
}
@@ -136,9 +168,9 @@ func validLogBatchRequest(t *testing.T, sessionToken string, firstSeq uint64, la
return dto.LogBatchIngestRequest{
RunEndpointID: "run-local",
SessionToken: sessionToken,
LogStreamID: "log-1",
LogStreamID: streamID,
ServerInstanceID: "server-1",
StreamKey: "stdout",
StreamKey: streamKey,
Source: domain.LogStreamSourceProcess,
FirstSeq: firstSeq,
LastSeq: lastSeq,