diff --git a/CHANGELOG.md b/CHANGELOG.md index ea617b91d..2d2bdbb4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed +- `tt log -f`: fixed possible line loss/duplication on rename and hanging + after a watched log directory is removed. + ## [2.13.0] - 2026-05-21 This release adds cluster worker configuration management and fixes diff --git a/cli/cmd/log.go b/cli/cmd/log.go index e6e793c6f..5b6b489a1 100644 --- a/cli/cmd/log.go +++ b/cli/cmd/log.go @@ -6,7 +6,9 @@ import ( "fmt" "os" "os/signal" + "path/filepath" "sync" + "time" "github.com/spf13/cobra" "github.com/tarantool/tt/cli/cmd/internal" @@ -20,6 +22,8 @@ var logOpts struct { follow bool // Follow logs output. } +const logRootCheckInterval = 100 * time.Millisecond + // NewLogCmd creates log command. func NewLogCmd() *cobra.Command { logCmd := &cobra.Command{ @@ -60,10 +64,46 @@ func printLines(ctx context.Context, in <-chan string) error { } } +type logRootContext struct { + ctx context.Context + cancel context.CancelFunc +} + +func logRoot(inst running.InstanceCtx) string { + if inst.SingleApp { + return inst.LogDir + } + return filepath.Dir(inst.LogDir) +} + +func monitorLogRoot(ctx context.Context, root string, cancel context.CancelFunc) { + ticker := time.NewTicker(logRootCheckInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if _, err := os.Stat(root); errors.Is(err, os.ErrNotExist) { + cancel() + return + } + } + } +} + func follow(instances []running.InstanceCtx, n int) error { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) defer stop() + rootContexts := make(map[string]logRootContext) + defer func() { + for _, rootCtx := range rootContexts { + rootCtx.cancel() + } + }() + nextColor := tail.DefaultColorPicker() color := nextColor() const logLinesChannelCapacity = 64 @@ -72,9 +112,18 @@ func follow(instances []running.InstanceCtx, n int) error { // Wait group to wait for completion of all log reading routines to close the channel once. var wg sync.WaitGroup for _, inst := range instances { - if err := tail.Follow(ctx, logLines, + root := logRoot(inst) + rootCtx, ok := rootContexts[root] + if !ok { + rootCtx.ctx, rootCtx.cancel = context.WithCancel(ctx) + rootContexts[root] = rootCtx + go monitorLogRoot(rootCtx.ctx, root, rootCtx.cancel) + } + + err := tail.Follow(rootCtx.ctx, logLines, tail.NewLogFormatter(running.GetAppInstanceName(inst)+": ", color), - inst.Log, n, &wg); err != nil { + inst.Log, n, &wg) + if err != nil { if errors.Is(err, os.ErrNotExist) { continue } diff --git a/cli/cmd/log_test.go b/cli/cmd/log_test.go new file mode 100644 index 000000000..3131581e0 --- /dev/null +++ b/cli/cmd/log_test.go @@ -0,0 +1,51 @@ +package cmd + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/tarantool/tt/cli/running" +) + +func TestLogRoot(t *testing.T) { + t.Run("multi-instance application", func(t *testing.T) { + inst := running.InstanceCtx{ + LogDir: filepath.Join("app", "var", "log", "instance"), + } + expected := filepath.Join("app", "var", "log") + if actual := logRoot(inst); actual != expected { + t.Fatalf("Unexpected log root: got %q, want %q", actual, expected) + } + }) + + t.Run("single-instance application", func(t *testing.T) { + inst := running.InstanceCtx{ + LogDir: filepath.Join("var", "log"), + SingleApp: true, + } + if actual := logRoot(inst); actual != inst.LogDir { + t.Fatalf("Unexpected log root: got %q, want %q", actual, inst.LogDir) + } + }) +} + +func TestMonitorLogRoot(t *testing.T) { + root := t.TempDir() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go monitorLogRoot(ctx, root, cancel) + + if err := os.Remove(root); err != nil { + t.Fatalf("Failed to remove log root: %v", err) + } + + select { + case <-ctx.Done(): + case <-time.After(time.Second): + t.Fatal("Log root removal did not cancel the follow context") + } +} diff --git a/cli/tail/follow.go b/cli/tail/follow.go index 0ef24d740..1020cd975 100644 --- a/cli/tail/follow.go +++ b/cli/tail/follow.go @@ -10,7 +10,7 @@ import ( "time" "github.com/apex/log" - "github.com/nxadm/tail" + "github.com/tarantool/go-tail" ) const ( diff --git a/cli/tail/follow_test.go b/cli/tail/follow_test.go index a7756b0db..3eda867af 100644 --- a/cli/tail/follow_test.go +++ b/cli/tail/follow_test.go @@ -11,10 +11,13 @@ import ( ) const ( - linesPerStep = 3 - channelCapacity = 100 - logLineFormat = "%03d: line" - logNewLineFormat = "%03d: new line added" + linesPerStep = 3 + channelCapacity = 100 + lineReadTimeout = 5 * time.Second + reopenWatchDelay = 500 * time.Millisecond + watcherStartDelay = 100 * time.Millisecond + logLineFormat = "%03d: line" + logNewLineFormat = "%03d: new line added" ) // readWithTimeout Helper to read from channel with timeout. @@ -25,7 +28,10 @@ func readWithTimeout(t *testing.T, ch <-chan string, timeout time.Duration) (str defer timer.Stop() select { - case s := <-ch: + case s, ok := <-ch: + if !ok { + return "", fmt.Errorf("channel closed while waiting for data") + } return s, nil case <-timer.C: return "", fmt.Errorf("timeout waiting for data") @@ -60,13 +66,30 @@ func createTmpLogFile(t *testing.T, count int, line_fmt string) string { return f.Name() } +func appendSyncedLine(t *testing.T, path, line string) { + t.Helper() + + file, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + t.Fatalf("Failed to open log file for append: %v", err) + } + defer file.Close() + + if _, err := fmt.Fprintln(file, line); err != nil { + t.Fatalf("Failed to append log line: %v", err) + } + if err := file.Sync(); err != nil { + t.Fatalf("Failed to flush log line: %v", err) + } +} + func checksLinesInFile(t *testing.T, lines int, ch <-chan string, exp_fmt string) error { t.Helper() for i := range lines { n := i + 1 - line, err := readWithTimeout(t, ch, time.Second) + line, err := readWithTimeout(t, ch, lineReadTimeout) if err != nil { return fmt.Errorf("failed to read line %d: %w", n, err) } @@ -103,10 +126,6 @@ func TestFollow2_ReadExistingContent(t *testing.T) { } func TestFollow2_FollowNewContent(t *testing.T) { - if os.Getenv("CI") != "" { - t.Skip("Skipping flaky test on CI until issue #TNTP-3131 is fixed") - } - lf := createTmpLogFile(t, linesPerStep, logLineFormat) ctx, cancel := context.WithCancel(context.Background()) @@ -124,6 +143,10 @@ func TestFollow2_FollowNewContent(t *testing.T) { t.Fatalf("Failed to check lines in file: %v", err) } + // TailFile starts its filesystem watcher asynchronously. Keep that startup + // outside the append scenario this test exercises. + time.Sleep(watcherStartDelay) + // Append new content appendFile, err := os.OpenFile(lf, os.O_APPEND|os.O_WRONLY, 0o644) if err != nil { @@ -195,89 +218,82 @@ func TestFollow2_NonExistentFile(t *testing.T) { } } -func rotationTest(t *testing.T, use_delay bool) { +func TestFollow2_FileRotation(t *testing.T) { lf := createTmpLogFile(t, linesPerStep, logLineFormat) ctx, cancel := context.WithCancel(context.Background()) - defer cancel() f := tail.NewTailFollower(lf) outCh, err := f.Follow(ctx, linesPerStep) if err != nil { - t.Skipf("Failed to follow: %v", err) + cancel() + t.Fatalf("Failed to follow: %v", err) } + defer func() { + cancel() + f.Wait() + }() err = checksLinesInFile(t, linesPerStep, outCh, logLineFormat) if err != nil { - t.Skipf("Failed to check initial lines in file: %v", err) - return + t.Fatalf("Failed to check initial lines in file: %v", err) } - err = os.Rename(lf, lf+".bak") + // TailFile starts its filesystem watcher asynchronously. Keep that startup + // outside the write-plus-rename scenario this test exercises. + time.Sleep(watcherStartDelay) + + const readinessLine = "watcher is ready" + appendSyncedLine(t, lf, readinessLine) + line, err := readWithTimeout(t, outCh, lineReadTimeout) if err != nil { - t.Fatalf("Failed to rotate log file: %v", err) + t.Fatalf("Failed to read watcher readiness line: %v", err) } - - if use_delay { - time.Sleep(500 * time.Millisecond) // Add delay to avoid flaky fails. + if line != readinessLine { + t.Fatalf("Watcher readiness line mismatch: got %q, want %q", line, readinessLine) } - newFile, err := os.Create(lf) - if err != nil { - t.Fatalf("Failed to create new log file: %v", err) + const replacementLine = "line in replacement file" + replacementPath := lf + ".new" + if err := os.WriteFile(replacementPath, []byte(replacementLine+"\n"), 0o644); err != nil { + t.Fatalf("Failed to prepare replacement log file: %v", err) } - err = writeLogLines(t, newFile, linesPerStep, logNewLineFormat) - - newFile.Close() + const lineBeforeRotation = "line written immediately before rotation" + appendSyncedLine(t, lf, lineBeforeRotation) + if err := os.Rename(lf, lf+".bak"); err != nil { + t.Fatalf("Failed to rotate log file: %v", err) + } + line, err = readWithTimeout(t, outCh, lineReadTimeout) if err != nil { - t.Skipf("Failed to write new log lines after rotation: %v", err) - return + t.Fatalf("Failed to read line written before rotation: %v", err) + } + if line != lineBeforeRotation { + t.Fatalf("Line written before rotation mismatch: got %q, want %q", + line, lineBeforeRotation) } - newFile.Close() + // Let the asynchronous reopen path start watching for the replacement. + // The loss assertion above remains adjacent to the rename. + time.Sleep(reopenWatchDelay) + if err := os.Rename(replacementPath, lf); err != nil { + t.Fatalf("Failed to install replacement log file: %v", err) + } - err = checksLinesInFile(t, linesPerStep, outCh, logNewLineFormat) + line, err = readWithTimeout(t, outCh, lineReadTimeout) if err != nil { - t.Skipf("Failed to check appended lines in file: %v", err) - return + t.Fatalf("Failed to read line from replacement file: %v", err) + } + if line != replacementLine { + t.Fatalf("Replacement line mismatch: got %q, want %q", line, replacementLine) } cancel() f.Wait() -} - -// TestFollow2_FileRotation_Flaky tests the file rotation with flaky retries. -// It retries the test multiple times to handle potential flakiness in the tail library. -// This is a workaround for the issue #TNTP-3131, where the tail library -// does not handle file rotation correctly. -// - TODO: Need fix `tail` library, see #TNTP-3131 for more details. -func TestFollow2_FileRotation_Flaky(t *testing.T) { - if os.Getenv("CI") != "" { - t.Skip("Skipping flaky test on CI until issue #TNTP-3131 is fixed") - } - - const flakyRepeatCount = 3 - - test_pass := false - - for i := range flakyRepeatCount { - t.Run(fmt.Sprintf("Rotation-%d", i+1), func(t *testing.T) { - rotationTest(t, i > 0) - - test_pass = !t.Skipped() - }) - - if test_pass { - break - } - - t.Logf("FLAKY test %s failed, retrying flaky test iteration", t.Name()) - } - if !test_pass { - t.Fatalf("Test failed after all flaky iterations") + for line := range outCh { + t.Fatalf("Unexpected line after rotation: %q", line) } } diff --git a/cli/tail/tail.go b/cli/tail/tail.go index 80b01df06..22ba1af0a 100644 --- a/cli/tail/tail.go +++ b/cli/tail/tail.go @@ -9,13 +9,17 @@ import ( "os" "strings" "sync" + "time" "github.com/apex/log" "github.com/fatih/color" - "github.com/nxadm/tail" + "github.com/tarantool/go-tail" ) -const blockSize = 8192 +const ( + blockSize = 8192 + tailShutdownTimeout = time.Second +) // Reader is an interface for reading the last `lines` lines from a file. type Reader interface { @@ -154,6 +158,19 @@ func TailN(ctx context.Context, logFormatter LogFormatter, fileName string, return out, nil } +func stopTail(t *tail.Tail) error { + t.Kill(nil) + select { + case <-t.Dead(): + if err := t.Err(); err != nil { + return fmt.Errorf("failed to stop tailer for %q: %w", t.Filename, err) + } + return nil + case <-time.After(tailShutdownTimeout): + return fmt.Errorf("timeout stopping tailer for %q", t.Filename) + } +} + // Follow sends to the channel each new line from the file as it grows. func Follow(ctx context.Context, out chan<- string, logFormatter LogFormatter, fileName string, n int, wg *sync.WaitGroup, @@ -190,12 +207,13 @@ func Follow(ctx context.Context, out chan<- string, logFormatter LogFormatter, f for { select { case <-ctx.Done(): - t.Stop() - t.Wait() + if err := stopTail(t); err != nil { + log.Warn(err.Error()) + } return case line, more := <-t.Lines: if !more { - err := t.Stop() + err := stopTail(t) if err != nil { log.Error(err.Error()) } else { diff --git a/go.mod b/go.mod index dac40372f..082014101 100644 --- a/go.mod +++ b/go.mod @@ -24,7 +24,6 @@ require ( github.com/moby/moby/api v1.54.2 github.com/moby/moby/client v0.4.0 github.com/moby/term v0.5.2 - github.com/nxadm/tail v1.4.11 github.com/otiai10/copy v1.14.1 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 github.com/spf13/cobra v1.10.2 @@ -33,6 +32,7 @@ require ( github.com/tarantool/go-iproto v1.1.0 github.com/tarantool/go-prompt v1.0.1 github.com/tarantool/go-storage v1.5.0 + github.com/tarantool/go-tail v0.0.0-20260730123520-0e16f7e63f1b github.com/tarantool/go-tarantool v1.12.3 github.com/tarantool/go-tarantool/v2 v2.4.2 github.com/tarantool/go-xlog v0.0.0-20260707203858-fed522934686 diff --git a/go.sum b/go.sum index df05add30..3e4e3ac68 100644 --- a/go.sum +++ b/go.sum @@ -108,7 +108,6 @@ github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= -github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw= github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k= github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= github.com/go-ini/ini v1.67.0 h1:z6ZrTEZqSWOTyH2FlglNbNgARyHG8oLW9gMELqKr06A= @@ -262,8 +261,6 @@ github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7P github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= -github.com/nxadm/tail v1.4.11 h1:8feyoE3OzPrcshW5/MJ4sGESc5cqmGkGCWlco4l0bqY= -github.com/nxadm/tail v1.4.11/go.mod h1:OTaG3NK980DZzxbRq6lEuzgU+mug70nY11sMd4JXXHc= github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec= github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= @@ -363,6 +360,8 @@ github.com/tarantool/go-prompt v1.0.1 h1:88Yer6gCFylqGRrdWwikNFVbklRQsqKF7mycvGd github.com/tarantool/go-prompt v1.0.1/go.mod h1:9Vuvi60Bk+3yaXqgYaXNTpLbwPPaaEOeaUgpFW1jqTU= github.com/tarantool/go-storage v1.5.0 h1:WRi5eahOinBHh+Nnc5CmiaFNxYQCd5OoiUfA7n9pe30= github.com/tarantool/go-storage v1.5.0/go.mod h1:Aj8RoWXZGYOI7oWT3Utj9IE7FItvMSDf2smHxHJo96M= +github.com/tarantool/go-tail v0.0.0-20260730123520-0e16f7e63f1b h1:/a2HXn6bI6+EkOEPXJo6zEsc9T1Otgq56AINJcZWRqA= +github.com/tarantool/go-tail v0.0.0-20260730123520-0e16f7e63f1b/go.mod h1:NqLWssaRJ2w9myxdJWlc4WaZWHU2CLZDdEYfKlSnbEQ= github.com/tarantool/go-tarantool v1.12.3 h1:GXabowmrTSW225xFEjX4t+8PlccVDCeGB5OM1VLbBXE= github.com/tarantool/go-tarantool v1.12.3/go.mod h1:QRiXv0jnxwgxHtr9ZmifSr/eRba76gTUBgp69pDMX1U= github.com/tarantool/go-tarantool/v2 v2.4.2 h1:rkzYtFhLJLA9RDIhjzN93MJBN5PBxHW4+soq+RB90gE= @@ -532,7 +531,6 @@ golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= diff --git a/test/async_reader.py b/test/async_reader.py index 72ba7a772..82ea67357 100644 --- a/test/async_reader.py +++ b/test/async_reader.py @@ -111,6 +111,17 @@ def stdout_wait_for(self, expected: str, timeout: float = 5.0) -> tuple[list[str """ return self.__wait_for(self.__q_stdout, expected, timeout) + def stdout_wait_for_all( + self, + expected: list[str], + timeout: float = 5.0, + ) -> tuple[list[str], bool]: + """ + Expects all tokens in stdout in any order, collecting all lines. + Returns a list of lines and a flag if every token was found. + """ + return self.__wait_for_all(self.__q_stdout, expected, timeout) + def stderr_wait_for(self, expected: str, timeout: float = 5.0) -> tuple[list[str], bool]: """ Expects a token in stderr, collecting all lines. @@ -126,8 +137,8 @@ def pStop(self) -> None: """Stop process gracefully.""" if not self._isExited(): self.__stop_event.set() - self.__stdout_thread.join(timeout=5.0) - self.__stderr_thread.join(timeout=5.0) + self.__stdout_thread.join(timeout=5.0) + self.__stderr_thread.join(timeout=5.0) def pKill(self) -> None: """Force terminates the process.""" @@ -160,12 +171,20 @@ def __wait_for( q: _StreamQueue, expected: str, timeout: float = 10.0, + ) -> tuple[list[str], bool]: + return self.__wait_for_all(q, [expected], timeout) + + def __wait_for_all( + self, + q: _StreamQueue, + expected: list[str], + timeout: float, ) -> tuple[list[str], bool]: stop_time = time.monotonic() + timeout wait_close = False FINAL_TIMEOUT = 1.0 # After process exit, tiny wait to collect remaining lines. - marker_found = False + pending = expected.copy() lines: list[str] = [] while time.monotonic() < stop_time: @@ -175,8 +194,8 @@ def __wait_for( break lines.append(line) - if expected in line: - marker_found = True + pending = [marker for marker in pending if marker not in line] + if not pending: break except queue.Empty: @@ -188,7 +207,7 @@ def __wait_for( stop_time = next_timeout continue - return lines, marker_found + return lines, not pending def __stream_reader_thread_target( self, diff --git a/test/integration/log/test_log.py b/test/integration/log/test_log.py index b345c71a7..5b98cc8a6 100644 --- a/test/integration/log/test_log.py +++ b/test/integration/log/test_log.py @@ -1,11 +1,13 @@ import os import shutil +import signal import subprocess -import time +from contextlib import contextmanager import pytest +from async_reader import AsyncProcessReader -from utils import config_name, wait_for_lines_in_output +from utils import config_name @pytest.fixture(scope="function") @@ -31,6 +33,57 @@ def mock_env_dir(tmp_path): return tmp_path +@contextmanager +def managed_reader(cmd, cwd): + reader = AsyncProcessReader(cmd, cwd) + try: + yield reader + finally: + if reader.pWait(timeout=0) is None: + reader.pKill() + reader.pStop() + + +def wait_for_output(reader, expected, timeout=10): + lines, found = reader.stdout_wait_for_all(expected, timeout=timeout) + output = "".join(lines) + assert found, ( + f"Expected lines {expected} not found in stdout:\n{output}\n" + f"stderr:\n{''.join(reader.stderr)}" + ) + return output + + +def append_synced_line(path, line): + with open(path, "a") as log: + log.write(f"{line}\n") + log.flush() + os.fsync(log.fileno()) + + +def wait_for_watcher(reader, path, output_prefix): + output = "" + for attempt in range(5): + marker = f"watcher readiness probe {attempt}" + append_synced_line(path, marker) + lines, found = reader.stdout_wait_for( + f"{output_prefix}: {marker}", + timeout=2, + ) + output += "".join(lines) + if found: + return output + + raise AssertionError( + f"Watcher for {output_prefix} did not consume a readiness probe.\n" + f"stdout:\n{output}\nstderr:\n{''.join(reader.stderr)}", + ) + + +def log_path(root, app, instance): + return os.path.join(root, "ie", app, "var", "log", instance, "tt.log") + + def test_log_output_default_run(tt_cmd, mock_env_dir): cmd = [tt_cmd, "log"] process = subprocess.Popen( @@ -192,36 +245,38 @@ def test_log_no_inst(tt_cmd, mock_env_dir): def test_log_output_default_follow(tt_cmd, mock_env_dir): cmd = [tt_cmd, "log", "-f"] - process = subprocess.Popen( - cmd, - cwd=mock_env_dir, - stderr=subprocess.STDOUT, - stdout=subprocess.PIPE, - text=True, - ) - - output = wait_for_lines_in_output( - process.stdout, - [ - "app0:inst0: line 19", - "app1:inst2: line 19", - "app0:inst1: line 19", - "app1:inst1: line 19", - ], - ) - - with open(os.path.join(mock_env_dir, "ie", "app0", "var", "log", "inst0", "tt.log"), "w") as f: - f.writelines([f"line {i}\n" for i in range(20, 23)]) - - with open(os.path.join(mock_env_dir, "ie", "app1", "var", "log", "inst2", "tt.log"), "w") as f: - f.writelines([f"line {i}\n" for i in range(20, 23)]) - - output += wait_for_lines_in_output( - process.stdout, - ["app1:inst2: line 22", "app0:inst0: line 22"], - ) + with managed_reader(cmd, mock_env_dir) as reader: + output = wait_for_output( + reader, + [ + "app0:inst0: line 19", + "app1:inst2: line 19", + "app0:inst1: line 19", + "app1:inst1: line 19", + ], + ) + + app0_log = log_path(mock_env_dir, "app0", "inst0") + app1_log = log_path(mock_env_dir, "app1", "inst2") + + output += wait_for_watcher(reader, app0_log, "app0:inst0") + output += wait_for_watcher(reader, app1_log, "app1:inst2") + + with open(app0_log, "w") as log: + log.writelines([f"line {i}\n" for i in range(20, 23)]) + with open(app1_log, "w") as log: + log.writelines([f"line {i}\n" for i in range(20, 23)]) + + output += wait_for_output( + reader, + ["app1:inst2: line 22", "app0:inst0: line 22"], + ) + + reader.send_signal(signal.SIGINT) + reader.pWait(timeout=10) + reader.pStop() + output += "".join(reader.stdout) - process.terminate() for i in range(10, 23): assert f"app0:inst0: line {i}" in output assert f"app1:inst2: line {i}" in output @@ -233,30 +288,28 @@ def test_log_output_default_follow(tt_cmd, mock_env_dir): def test_log_output_default_follow_want_zero_last(tt_cmd, mock_env_dir): cmd = [tt_cmd, "log", "-f", "-n", "0"] - process = subprocess.Popen( - cmd, - cwd=mock_env_dir, - stderr=subprocess.STDOUT, - stdout=subprocess.PIPE, - text=True, - universal_newlines=True, - bufsize=1, - ) + with managed_reader(cmd, mock_env_dir) as reader: + app0_log = log_path(mock_env_dir, "app0", "inst0") + app1_log = log_path(mock_env_dir, "app1", "inst2") - time.sleep(1) + output = wait_for_watcher(reader, app0_log, "app0:inst0") + output += wait_for_watcher(reader, app1_log, "app1:inst2") - with open(os.path.join(mock_env_dir, "ie", "app0", "var", "log", "inst0", "tt.log"), "w") as f: - f.writelines([f"line {i}\n" for i in range(20, 23)]) + with open(app0_log, "w") as log: + log.writelines([f"line {i}\n" for i in range(20, 23)]) + with open(app1_log, "w") as log: + log.writelines([f"line {i}\n" for i in range(20, 23)]) - with open(os.path.join(mock_env_dir, "ie", "app1", "var", "log", "inst2", "tt.log"), "w") as f: - f.writelines([f"line {i}\n" for i in range(20, 23)]) + output += wait_for_output( + reader, + ["app1:inst2: line 22", "app0:inst0: line 22"], + ) - output = wait_for_lines_in_output( - process.stdout, - ["app1:inst2: line 22", "app0:inst0: line 22"], - ) + reader.send_signal(signal.SIGINT) + reader.pWait(timeout=10) + reader.pStop() + output += "".join(reader.stdout) - process.terminate() for i in range(20, 23): assert f"app0:inst0: line {i}" in output assert f"app1:inst2: line {i}" in output @@ -266,68 +319,119 @@ def test_log_output_default_follow_want_zero_last(tt_cmd, mock_env_dir): assert "app1:inst0" not in output -def test_log_dir_removed_after_follow(tt_cmd, mock_env_dir): - cmd = [tt_cmd, "log", "-f"] - process = subprocess.Popen( - cmd, - cwd=mock_env_dir, - stderr=subprocess.STDOUT, - stdout=subprocess.PIPE, - text=True, +def test_log_rotation_preserves_pending_lines(tt_cmd, mock_env_dir): + cmd = [tt_cmd, "log", "app0:inst0", "-f", "-n", "1"] + with managed_reader(cmd, mock_env_dir) as reader: + output = wait_for_output( + reader, + ["app0:inst0: line 19"], + ) + + app_log = log_path(mock_env_dir, "app0", "inst0") + output += wait_for_watcher(reader, app_log, "app0:inst0") + + append_synced_line(app_log, "line written immediately before rotation") + os.rename(app_log, app_log + ".bak") + + output += wait_for_output( + reader, + ["app0:inst0: line written immediately before rotation"], + ) + + # Establish the asynchronous reopen before writing replacement records. + # The pending-write assertion above remains adjacent to the rename. + with open(app_log, "w"): + pass + output += wait_for_watcher(reader, app_log, "app0:inst0") + + replacement_lines = [ + "first line in replacement file", + "last line in replacement file", + ] + for line in replacement_lines: + append_synced_line(app_log, line) + output += wait_for_output( + reader, + [f"app0:inst0: {replacement_lines[-1]}"], + ) + + reader.send_signal(signal.SIGINT) + reader.pWait(timeout=10) + reader.pStop() + output += "".join(reader.stdout) + + actual_lines = [ + line + for line in output.splitlines() + if line.startswith("app0:inst0: ") + and "watcher readiness probe" not in line + ] + expected_lines = [ + "app0:inst0: line 19", + "app0:inst0: line written immediately before rotation", + *(f"app0:inst0: {line}" for line in replacement_lines), + ] + assert actual_lines == expected_lines, ( + f"Unexpected log sequence: {actual_lines}\n" + f"stderr:\n{''.join(reader.stderr)}" ) - wait_for_lines_in_output( - process.stdout, - [ - "app0:inst0: line 19", - "app1:inst2: line 19", - "app0:inst1: line 19", - "app1:inst1: line 19", - ], - ) - var_dir = os.path.join(mock_env_dir, "ie") - assert os.path.exists(var_dir) - shutil.rmtree(var_dir) +def test_log_dir_removed_after_follow(tt_cmd, mock_env_dir): + cmd = [tt_cmd, "log", "-f"] + with managed_reader(cmd, mock_env_dir) as reader: + wait_for_output( + reader, + [ + "app0:inst0: line 19", + "app1:inst2: line 19", + "app0:inst1: line 19", + "app1:inst1: line 19", + ], + ) + + var_dir = os.path.join(mock_env_dir, "ie") + assert os.path.exists(var_dir) + shutil.rmtree(var_dir) - assert process.wait(2) == 0 - assert "Failed to detect creation of" in process.stdout.read() + assert reader.pWait(timeout=10) == 0 + reader.pStop() # There are two apps in this test: app0 and app1. After removing app0 dirs, # tt log -f is still able to monitor the app1 log files, so there should be no issue. def test_log_dir_partially_removed_after_follow(tt_cmd, mock_env_dir): cmd = [tt_cmd, "log", "-f"] - process = subprocess.Popen( - cmd, - cwd=mock_env_dir, - stderr=subprocess.STDOUT, - stdout=subprocess.PIPE, - text=True, - ) - - wait_for_lines_in_output( - process.stdout, - [ - "app0:inst0: line 19", - "app1:inst2: line 19", - "app0:inst1: line 19", - "app1:inst1: line 19", - ], - ) - - # Remove one app log dir. - var_dir = os.path.join(mock_env_dir, "ie", "app0", "var", "log") - assert os.path.exists(var_dir) - shutil.rmtree(var_dir) - - wait_for_lines_in_output(process.stdout, ["Failed to detect creation of"]) - assert process.poll() is None # Still running. - - # Remove app1 log dir. - var_dir = os.path.join(mock_env_dir, "ie", "app1") - assert os.path.exists(var_dir) - shutil.rmtree(var_dir) - - assert process.wait(2) == 0 - assert "Failed to detect creation of" in process.stdout.read() + with managed_reader(cmd, mock_env_dir) as reader: + wait_for_output( + reader, + [ + "app0:inst0: line 19", + "app1:inst2: line 19", + "app0:inst1: line 19", + "app1:inst1: line 19", + ], + ) + + app1_log = log_path(mock_env_dir, "app1", "inst0") + wait_for_watcher(reader, app1_log, "app1:inst0") + + # Remove one app log dir. + var_dir = os.path.join(mock_env_dir, "ie", "app0", "var", "log") + assert os.path.exists(var_dir) + shutil.rmtree(var_dir) + + assert reader.pWait(timeout=0) is None # Still running. + append_synced_line(app1_log, "still following after app0 removal") + wait_for_output( + reader, + ["app1:inst0: still following after app0 removal"], + ) + + # Remove app1 log dir. + var_dir = os.path.join(mock_env_dir, "ie", "app1") + assert os.path.exists(var_dir) + shutil.rmtree(var_dir) + + assert reader.pWait(timeout=10) == 0 + reader.pStop() diff --git a/test/integration/tcm/test_tcm_log.py b/test/integration/tcm/test_tcm_log.py index d9747ab2f..e8f711460 100644 --- a/test/integration/tcm/test_tcm_log.py +++ b/test/integration/tcm/test_tcm_log.py @@ -1,4 +1,3 @@ -import os import signal import sys import time @@ -158,16 +157,11 @@ def handle_updating_logs( lines, is_found = reader.stdout_wait_for(eof_marker, timeout=10) reader.send_signal(signal.SIGINT) - reader.pStop() - - stderr_lines = reader.stderr - assert "context canceled" in "".join(stderr_lines), ( - f"Expected message not found in stderr {stderr_lines}" - ) - if reader.pWait() is None: reader.pKill() + reader.pStop() + stderr_lines = reader.stderr if not is_found: print(f"Second marker not found in stdout:\n{''.join(lines)}") rest_lines = reader.stdout @@ -178,6 +172,10 @@ def handle_updating_logs( assert is_found, f"Expected end marker {eof_marker} not found." + assert "context canceled" in "".join(stderr_lines), ( + f"Expected message not found in stderr {stderr_lines}" + ) + assert reader.returncode == 1, ( f"Command failed with return code {reader.returncode} (expected 1)." ) @@ -203,10 +201,6 @@ def final_checks( check_output("".join(stdout_lines), update_testdata, expected_file) -@pytest.mark.skipif( - condition=os.getenv("CI") is not None, - reason="Skip on CI runs until issue #TNTP-3131 is fixed.", -) @pytest.mark.parametrize("delay_time", (0.1, 0.01, 0)) @pytest.mark.parametrize("lines, options", TEST_CASES) @pytest.mark.parametrize("mode", TEST_DATA_MODES) @@ -236,11 +230,6 @@ def test_log_follow( @pytest.mark.slow -@pytest.mark.skipif( - condition=os.getenv("CI") is not None, - reason="Skip on CI runs until issue #TNTP-3131 is fixed.", -) -@pytest.mark.flaky(reruns=5) # See notes below about issues with `tail` package. @pytest.mark.parametrize("delay_time", (0.1, 0.01, 0)) @pytest.mark.parametrize("lines, options", TEST_CASES) @pytest.mark.parametrize("mode", TEST_DATA_MODES) @@ -264,13 +253,11 @@ def test_log_rotate( tmp_log.rename(tmp_log.with_suffix(".bak")) assert not tmp_log.exists(), "Temporary log file should be deleted." - # TODO: Need fix `tail` library, see #TNTP-3131 for more details. - # See `tail` opened issues, since 2024: - # - https://github.com/nxadm/tail/issues/72 - # - https://github.com/nxadm/tail/pull/73 - _, found = reader.stderr_wait_for("tryReopenTailer", timeout=0.5) - if not found: - time.sleep(1) + + # Create the replacement separately so the asynchronous reopen path can + # attach its watcher before new records are written. + tmp_log.touch() + time.sleep(0.5) new_lines, cnt_lines = handle_updating_logs(reader, tmp_log, mode, delay_time, is_append=True) stdout_lines.extend(new_lines)