From db3fdea4a13b8f2e68ae392833905835f54b645b Mon Sep 17 00:00:00 2001 From: LTbinglingfeng Date: Mon, 15 Jun 2026 15:50:03 +0800 Subject: [PATCH] Disambiguate zero-offset log cursors --- internal/api/handlers/management/logs.go | 65 ++++++++++++++-- internal/api/handlers/management/logs_test.go | 74 +++++++++++++++++++ 2 files changed, 134 insertions(+), 5 deletions(-) diff --git a/internal/api/handlers/management/logs.go b/internal/api/handlers/management/logs.go index 72e95c6f..3dfc635a 100644 --- a/internal/api/handlers/management/logs.go +++ b/internal/api/handlers/management/logs.go @@ -527,6 +527,7 @@ type logCursor struct { Offset int64 `json:"offset"` Size int64 `json:"size"` ModTime int64 `json:"modTime"` + ModTimeUnixNano int64 `json:"modTimeUnixNano,omitempty"` LatestTimestamp int64 `json:"latestTimestamp"` Fingerprint string `json:"fingerprint"` } @@ -808,6 +809,7 @@ func locateLogCursorFile(files []string, cursor logCursor) (int, bool, error) { for i := range files { nameToIndex[filepath.Base(files[i])] = i } + deferEmptyMainMatch := false if index, ok := nameToIndex[cursor.File]; ok { matches, truncated, errMatch := logFileMatchesCursor(files[index], cursor) if errMatch != nil { @@ -817,17 +819,24 @@ func locateLogCursorFile(files []string, cursor logCursor) (int, bool, error) { return 0, false, errMatch } if matches && !truncated { - return index, true, nil + if shouldDeferEmptyMainCursorToRotated(files, cursor) { + deferEmptyMainMatch = true + } else { + return index, true, nil + } } } - if cursor.File != defaultLogFileName || cursor.Offset == 0 { + if cursor.File != defaultLogFileName || (cursor.Offset == 0 && cursor.Size == 0 && !deferEmptyMainMatch) { return 0, false, nil } - for i := range files { + for i := len(files) - 1; i >= 0; i-- { if filepath.Base(files[i]) == defaultLogFileName { continue } + if cursor.Offset == 0 && cursor.Size == 0 && !logFileChangedAfterCursor(files[i], cursor) { + continue + } matches, truncated, errMatch := logFileMatchesCursor(files[i], cursor) if errMatch != nil { if errors.Is(errMatch, os.ErrNotExist) { @@ -845,6 +854,29 @@ func locateLogCursorFile(files []string, cursor logCursor) (int, bool, error) { return 0, false, nil } +func shouldDeferEmptyMainCursorToRotated(files []string, cursor logCursor) bool { + if cursor.File != defaultLogFileName || cursor.Offset != 0 || cursor.Size != 0 { + return false + } + for i := range files { + if filepath.Base(files[i]) == defaultLogFileName { + continue + } + if logFileChangedAfterCursor(files[i], cursor) { + return true + } + } + return false +} + +func logFileChangedAfterCursor(path string, cursor logCursor) bool { + info, errStat := os.Stat(path) + if errStat != nil || info.IsDir() || info.Size() == 0 { + return false + } + return info.ModTime().UnixNano() > cursorModTimeUnixNano(cursor) +} + func logFileMatchesCursor(path string, cursor logCursor) (bool, bool, error) { info, errStat := os.Stat(path) if errStat != nil { @@ -856,7 +888,11 @@ func logFileMatchesCursor(path string, cursor logCursor) (bool, bool, error) { if info.Size() < cursor.Offset { return false, true, nil } - fingerprint, errFingerprint := logFileFingerprint(path, cursor.Offset) + boundary := cursorFingerprintBoundary(cursor) + if info.Size() < boundary { + return false, true, nil + } + fingerprint, errFingerprint := logFileFingerprint(path, boundary) if errFingerprint != nil { return false, false, errFingerprint } @@ -953,7 +989,11 @@ func newLogCursor(path string, offset, latest int64) (string, error) { if offset < 0 || offset > info.Size() { return "", fmt.Errorf("invalid cursor offset") } - fingerprint, errFingerprint := logFileFingerprint(path, offset) + fingerprintCursor := logCursor{ + Offset: offset, + Size: info.Size(), + } + fingerprint, errFingerprint := logFileFingerprint(path, cursorFingerprintBoundary(fingerprintCursor)) if errFingerprint != nil { return "", errFingerprint } @@ -963,11 +1003,26 @@ func newLogCursor(path string, offset, latest int64) (string, error) { Offset: offset, Size: info.Size(), ModTime: info.ModTime().Unix(), + ModTimeUnixNano: info.ModTime().UnixNano(), LatestTimestamp: latest, Fingerprint: fingerprint, }) } +func cursorFingerprintBoundary(cursor logCursor) int64 { + if cursor.Offset == 0 && cursor.Size > 0 { + return cursor.Size + } + return cursor.Offset +} + +func cursorModTimeUnixNano(cursor logCursor) int64 { + if cursor.ModTimeUnixNano > 0 { + return cursor.ModTimeUnixNano + } + return cursor.ModTime * int64(time.Second) +} + func logFileFingerprint(path string, boundary int64) (string, error) { if boundary < 0 { return "", fmt.Errorf("invalid fingerprint boundary") diff --git a/internal/api/handlers/management/logs_test.go b/internal/api/handlers/management/logs_test.go index 1f1056e5..eb38176c 100644 --- a/internal/api/handlers/management/logs_test.go +++ b/internal/api/handlers/management/logs_test.go @@ -382,6 +382,80 @@ func TestGetLogsCursorReadsRotatedFileWhenNewMainIsSmaller(t *testing.T) { } } +func TestGetLogsZeroOffsetCursorWithPartialLineReadsAcrossRotation(t *testing.T) { + dir := t.TempDir() + writeMainLog(t, dir, "partial") + initial := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?limit=1") + if initial.NextCursor == "" { + t.Fatal("initial next-cursor is empty") + } + cursor, errCursor := decodeLogCursor(initial.NextCursor) + if errCursor != nil { + t.Fatalf("decode initial cursor: %v", errCursor) + } + if cursor.Offset != 0 || cursor.Size == 0 { + t.Fatalf("cursor offset/size = %d/%d, want zero offset with partial size", cursor.Offset, cursor.Size) + } + + appendMainLog(t, dir, " complete\n") + if err := os.Rename(filepath.Join(dir, defaultLogFileName), filepath.Join(dir, defaultLogFileName+".1")); err != nil { + t.Fatalf("rotate main log: %v", err) + } + writeMainLog(t, dir, "new\n") + + resp := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?cursor="+url.QueryEscape(initial.NextCursor)+"&limit=10") + wantLines := []string{"partial complete", "new"} + if !reflect.DeepEqual(resp.Lines, wantLines) { + t.Fatalf("lines = %#v, want %#v", resp.Lines, wantLines) + } + if resp.CursorReset { + t.Fatal("cursor-reset = true, want false") + } +} + +func TestGetLogsZeroOffsetCursorWithEmptyFileReadsAcrossRotation(t *testing.T) { + dir := t.TempDir() + writeMainLog(t, dir, "") + initial := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?limit=1") + if initial.NextCursor == "" { + t.Fatal("initial next-cursor is empty") + } + cursor, errCursor := decodeLogCursor(initial.NextCursor) + if errCursor != nil { + t.Fatalf("decode initial cursor: %v", errCursor) + } + if cursor.Offset != 0 || cursor.Size != 0 { + t.Fatalf("cursor offset/size = %d/%d, want empty zero offset", cursor.Offset, cursor.Size) + } + + appendMainLog(t, dir, "first\n") + mainPath := filepath.Join(dir, defaultLogFileName) + nextModTime := time.Unix(0, cursorModTimeUnixNano(cursor)+int64(time.Second)) + if err := os.Chtimes(mainPath, nextModTime, nextModTime); err != nil { + t.Fatalf("update main log mtime: %v", err) + } + if err := os.Rename(mainPath, filepath.Join(dir, defaultLogFileName+".1")); err != nil { + t.Fatalf("rotate main log: %v", err) + } + writeMainLog(t, dir, "second\n") + + resp := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?cursor="+url.QueryEscape(initial.NextCursor)+"&limit=1") + if !reflect.DeepEqual(resp.Lines, []string{"first"}) { + t.Fatalf("lines = %#v, want first rotated line", resp.Lines) + } + if resp.CursorReset { + t.Fatal("cursor-reset = true, want false") + } + + next := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?cursor="+url.QueryEscape(resp.NextCursor)+"&limit=1") + if !reflect.DeepEqual(next.Lines, []string{"second"}) { + t.Fatalf("next lines = %#v, want second main line", next.Lines) + } + if next.CursorReset { + t.Fatal("next cursor-reset = true, want false") + } +} + func TestGetLogsInvalidCursorResetsToTail(t *testing.T) { dir := t.TempDir() lines := []string{