From cd6a5b423523817ff1194de968bd777be68fe86f Mon Sep 17 00:00:00 2001 From: Mikhail Elhimov Date: Wed, 6 May 2026 12:38:16 +0300 Subject: [PATCH 1/2] tail: fix line loss/duplication if file is renamed Closes TNTP-3131 --- CHANGELOG.md | 2 + cli/tail/follow.go | 2 +- cli/tail/follow_test.go | 43 +----------- cli/tail/tail.go | 2 +- go.mod | 3 +- go.sum | 6 +- test/integration/log/test_log.py | 2 +- test/integration/tcm/test_tcm_log.py | 100 ++++++++++++++++++++++----- test/utils.py | 7 ++ 9 files changed, 97 insertions(+), 70 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea617b91d..2c1a267b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed +- Possible line loss/duplication when the log file is renamed. + ## [2.13.0] - 2026-05-21 This release adds cluster worker configuration management and fixes 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..4dee87466 100644 --- a/cli/tail/follow_test.go +++ b/cli/tail/follow_test.go @@ -103,10 +103,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()) @@ -195,7 +191,7 @@ 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()) @@ -219,10 +215,6 @@ func rotationTest(t *testing.T, use_delay bool) { t.Fatalf("Failed to rotate log file: %v", err) } - if use_delay { - time.Sleep(500 * time.Millisecond) // Add delay to avoid flaky fails. - } - newFile, err := os.Create(lf) if err != nil { t.Fatalf("Failed to create new log file: %v", err) @@ -248,36 +240,3 @@ func rotationTest(t *testing.T, use_delay bool) { 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") - } -} diff --git a/cli/tail/tail.go b/cli/tail/tail.go index 80b01df06..89b358fe7 100644 --- a/cli/tail/tail.go +++ b/cli/tail/tail.go @@ -12,7 +12,7 @@ import ( "github.com/apex/log" "github.com/fatih/color" - "github.com/nxadm/tail" + "github.com/tarantool/go-tail" ) const blockSize = 8192 diff --git a/go.mod b/go.mod index dac40372f..6dcfafdae 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 v1.4.12 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 @@ -170,7 +170,6 @@ require ( go.opentelemetry.io/otel v1.42.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0 // indirect - go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 // indirect go.opentelemetry.io/otel/metric v1.42.0 // indirect go.opentelemetry.io/otel/sdk v1.42.0 // indirect go.opentelemetry.io/otel/trace v1.42.0 // indirect diff --git a/go.sum b/go.sum index df05add30..ed5b1480b 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 v1.4.12 h1:bVIO9XbroPSF7/n0Mdp6dwxN1umNtuE6VtZGUpH7YqE= +github.com/tarantool/go-tail v1.4.12/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/integration/log/test_log.py b/test/integration/log/test_log.py index b345c71a7..3e517bd4f 100644 --- a/test/integration/log/test_log.py +++ b/test/integration/log/test_log.py @@ -290,7 +290,7 @@ def test_log_dir_removed_after_follow(tt_cmd, mock_env_dir): assert os.path.exists(var_dir) shutil.rmtree(var_dir) - assert process.wait(2) == 0 + assert process.wait(5) == 0 assert "Failed to detect creation of" in process.stdout.read() diff --git a/test/integration/tcm/test_tcm_log.py b/test/integration/tcm/test_tcm_log.py index d9747ab2f..95af2d8cb 100644 --- a/test/integration/tcm/test_tcm_log.py +++ b/test/integration/tcm/test_tcm_log.py @@ -1,5 +1,5 @@ -import os import signal +import subprocess import sys import time from pathlib import Path @@ -8,7 +8,9 @@ import pytest from async_reader import AsyncProcessReader -from utils import run_command_and_get_output +import utils + +# from utils import run_command_and_get_output DATA_DIR = Path(__file__).parent / "testdata" EXPECTED_DATA = DATA_DIR / "expected" @@ -86,7 +88,7 @@ def test_log_n_lines( if lines > 0: cmd.extend(("-n", str(lines))) cmd.extend(options) - rc, output = run_command_and_get_output(cmd, cwd=DATA_DIR / mode) + rc, output = utils.run_command_and_get_output(cmd, cwd=DATA_DIR / mode) assert rc == 0, f"Command failed with return code {rc}." expected_file = expected_file_name(mode, lines, options) @@ -124,6 +126,35 @@ def handle_static_logs( return tmp_log, reader, stdout_lines +def handle_static_logs2( + tt_cmd: Path, + tmp_path: Path, + mode: str, + lines: int, + options: list[str], +) -> tuple[Path, AsyncProcessReader, list[str]]: + tmp_log, eof_marker = prepare_first_log(tmp_path, mode) + + cmd = [tt_cmd, "tcm", "log", "--follow"] + if lines > 0: + cmd.append(f"--lines={lines + 1}") # +1 to include the extra EOF marker line. + cmd.extend(options) + + p = subprocess.Popen( + cmd, + cwd=tmp_path, + stderr=subprocess.PIPE, + stdout=subprocess.PIPE, + text=True, + encoding="utf-8", + ) + + assert utils.wait_for_string(10, p.stdout, eof_marker), \ + f"Expected end marker {eof_marker} not found." + + return tmp_log, p + + def prepare_second_log( tmp_log: Path, mode: str, @@ -155,7 +186,9 @@ def handle_updating_logs( ) -> tuple[list[str], int]: eof_marker, write_cnt = prepare_second_log(tmp_log, mode, delay_time, is_append) + # print(f"@@@@@@@@@ {time.time()} handle_updating_logs") lines, is_found = reader.stdout_wait_for(eof_marker, timeout=10) + # print(f"@@@@@@@@@ {time.time()} handle_updating_logs2") reader.send_signal(signal.SIGINT) reader.pStop() @@ -203,10 +236,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) @@ -235,12 +264,52 @@ def test_log_follow( ) +def handle_updating_logs2( + p, + tmp_log: Path, + mode: str, + delay_time: float, + is_append: bool, +) -> tuple[list[str], int]: + eof_marker, write_cnt = prepare_second_log(tmp_log, mode, delay_time, is_append) + + # print(f"@@@@@@@@@ {time.time()} handle_updating_logs") + assert utils.wait_for_string(10, p.stdout, eof_marker), \ + f"Expected end marker {eof_marker} not found." + lines, is_found = reader.stdout_wait_for(eof_marker, timeout=10) + # print(f"@@@@@@@@@ {time.time()} handle_updating_logs2") + + 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() + + if not is_found: + print(f"Second marker not found in stdout:\n{''.join(lines)}") + rest_lines = reader.stdout + if rest_lines: + print(f"Rest output =>\n{''.join(rest_lines)}") + if stderr_lines: + print(f"Got stderr:\n{''.join(stderr_lines)}", file=sys.stderr) + + assert is_found, f"Expected end marker {eof_marker} not found." + + assert reader.returncode == 1, ( + f"Command failed with return code {reader.returncode} (expected 1)." + ) + return lines, write_cnt + + @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 +333,6 @@ 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) new_lines, cnt_lines = handle_updating_logs(reader, tmp_log, mode, delay_time, is_append=True) stdout_lines.extend(new_lines) diff --git a/test/utils.py b/test/utils.py index b9403744e..e6c9d585d 100644 --- a/test/utils.py +++ b/test/utils.py @@ -730,3 +730,10 @@ def update_dict_leaves(d, other): d[k] = v elif v is not None: d[k] = v + + +def wait_for_string(timeout, stream, s, interval=0.1): + def check_next_line(): + return s in stream.readline() + + return wait_event(timeout, check_next_line, interval) From c68469693cabed8549c459763e0970ce74c38cde Mon Sep 17 00:00:00 2001 From: Eugene Blikh Date: Thu, 30 Jul 2026 15:48:15 +0300 Subject: [PATCH 2/2] tail: harden rotation and directory removal handling --- CHANGELOG.md | 3 +- cli/cmd/log.go | 53 ++++- cli/cmd/log_test.go | 51 +++++ cli/tail/follow_test.go | 105 +++++++-- cli/tail/tail.go | 26 ++- go.mod | 3 +- go.sum | 4 +- test/async_reader.py | 31 ++- test/integration/log/test_log.py | 314 ++++++++++++++++++--------- test/integration/tcm/test_tcm_log.py | 101 ++------- test/utils.py | 7 - 11 files changed, 458 insertions(+), 240 deletions(-) create mode 100644 cli/cmd/log_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c1a267b2..2d2bdbb4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed -- Possible line loss/duplication when the log file is renamed. +- `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 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_test.go b/cli/tail/follow_test.go index 4dee87466..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) } @@ -120,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,48 +222,78 @@ 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") - if err != nil { - t.Fatalf("Failed to rotate log file: %v", err) - } + // TailFile starts its filesystem watcher asynchronously. Keep that startup + // outside the write-plus-rename scenario this test exercises. + time.Sleep(watcherStartDelay) - newFile, err := os.Create(lf) + const readinessLine = "watcher is ready" + appendSyncedLine(t, lf, readinessLine) + line, err := readWithTimeout(t, outCh, lineReadTimeout) if err != nil { - t.Fatalf("Failed to create new log file: %v", err) + t.Fatalf("Failed to read watcher readiness line: %v", err) + } + if line != readinessLine { + t.Fatalf("Watcher readiness line mismatch: got %q, want %q", line, readinessLine) } - err = writeLogLines(t, newFile, linesPerStep, logNewLineFormat) + 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) + } - 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() + + 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 89b358fe7..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/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 6dcfafdae..082014101 100644 --- a/go.mod +++ b/go.mod @@ -32,7 +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 v1.4.12 + 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 @@ -170,6 +170,7 @@ require ( go.opentelemetry.io/otel v1.42.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.40.0 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 // indirect go.opentelemetry.io/otel/metric v1.42.0 // indirect go.opentelemetry.io/otel/sdk v1.42.0 // indirect go.opentelemetry.io/otel/trace v1.42.0 // indirect diff --git a/go.sum b/go.sum index ed5b1480b..3e4e3ac68 100644 --- a/go.sum +++ b/go.sum @@ -360,8 +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 v1.4.12 h1:bVIO9XbroPSF7/n0Mdp6dwxN1umNtuE6VtZGUpH7YqE= -github.com/tarantool/go-tail v1.4.12/go.mod h1:NqLWssaRJ2w9myxdJWlc4WaZWHU2CLZDdEYfKlSnbEQ= +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= 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 3e517bd4f..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(5) == 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 95af2d8cb..e8f711460 100644 --- a/test/integration/tcm/test_tcm_log.py +++ b/test/integration/tcm/test_tcm_log.py @@ -1,5 +1,4 @@ import signal -import subprocess import sys import time from pathlib import Path @@ -8,9 +7,7 @@ import pytest from async_reader import AsyncProcessReader -import utils - -# from utils import run_command_and_get_output +from utils import run_command_and_get_output DATA_DIR = Path(__file__).parent / "testdata" EXPECTED_DATA = DATA_DIR / "expected" @@ -88,7 +85,7 @@ def test_log_n_lines( if lines > 0: cmd.extend(("-n", str(lines))) cmd.extend(options) - rc, output = utils.run_command_and_get_output(cmd, cwd=DATA_DIR / mode) + rc, output = run_command_and_get_output(cmd, cwd=DATA_DIR / mode) assert rc == 0, f"Command failed with return code {rc}." expected_file = expected_file_name(mode, lines, options) @@ -126,35 +123,6 @@ def handle_static_logs( return tmp_log, reader, stdout_lines -def handle_static_logs2( - tt_cmd: Path, - tmp_path: Path, - mode: str, - lines: int, - options: list[str], -) -> tuple[Path, AsyncProcessReader, list[str]]: - tmp_log, eof_marker = prepare_first_log(tmp_path, mode) - - cmd = [tt_cmd, "tcm", "log", "--follow"] - if lines > 0: - cmd.append(f"--lines={lines + 1}") # +1 to include the extra EOF marker line. - cmd.extend(options) - - p = subprocess.Popen( - cmd, - cwd=tmp_path, - stderr=subprocess.PIPE, - stdout=subprocess.PIPE, - text=True, - encoding="utf-8", - ) - - assert utils.wait_for_string(10, p.stdout, eof_marker), \ - f"Expected end marker {eof_marker} not found." - - return tmp_log, p - - def prepare_second_log( tmp_log: Path, mode: str, @@ -186,21 +154,14 @@ def handle_updating_logs( ) -> tuple[list[str], int]: eof_marker, write_cnt = prepare_second_log(tmp_log, mode, delay_time, is_append) - # print(f"@@@@@@@@@ {time.time()} handle_updating_logs") lines, is_found = reader.stdout_wait_for(eof_marker, timeout=10) - # print(f"@@@@@@@@@ {time.time()} handle_updating_logs2") 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 @@ -211,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)." ) @@ -264,51 +229,6 @@ def test_log_follow( ) -def handle_updating_logs2( - p, - tmp_log: Path, - mode: str, - delay_time: float, - is_append: bool, -) -> tuple[list[str], int]: - eof_marker, write_cnt = prepare_second_log(tmp_log, mode, delay_time, is_append) - - # print(f"@@@@@@@@@ {time.time()} handle_updating_logs") - assert utils.wait_for_string(10, p.stdout, eof_marker), \ - f"Expected end marker {eof_marker} not found." - lines, is_found = reader.stdout_wait_for(eof_marker, timeout=10) - # print(f"@@@@@@@@@ {time.time()} handle_updating_logs2") - - 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() - - if not is_found: - print(f"Second marker not found in stdout:\n{''.join(lines)}") - rest_lines = reader.stdout - if rest_lines: - print(f"Rest output =>\n{''.join(rest_lines)}") - if stderr_lines: - print(f"Got stderr:\n{''.join(stderr_lines)}", file=sys.stderr) - - assert is_found, f"Expected end marker {eof_marker} not found." - - assert reader.returncode == 1, ( - f"Command failed with return code {reader.returncode} (expected 1)." - ) - return lines, write_cnt - - @pytest.mark.slow @pytest.mark.parametrize("delay_time", (0.1, 0.01, 0)) @pytest.mark.parametrize("lines, options", TEST_CASES) @@ -334,6 +254,11 @@ def test_log_rotate( tmp_log.rename(tmp_log.with_suffix(".bak")) assert not tmp_log.exists(), "Temporary log file should be deleted." + # 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) final_checks( diff --git a/test/utils.py b/test/utils.py index e6c9d585d..b9403744e 100644 --- a/test/utils.py +++ b/test/utils.py @@ -730,10 +730,3 @@ def update_dict_leaves(d, other): d[k] = v elif v is not None: d[k] = v - - -def wait_for_string(timeout, stream, s, interval=0.1): - def check_next_line(): - return s in stream.readline() - - return wait_event(timeout, check_next_line, interval)