diff options
Diffstat (limited to 'integrationtests/djournal_extended_test.go')
| -rw-r--r-- | integrationtests/djournal_extended_test.go | 1028 |
1 files changed, 1028 insertions, 0 deletions
diff --git a/integrationtests/djournal_extended_test.go b/integrationtests/djournal_extended_test.go new file mode 100644 index 0000000..47f6d7f --- /dev/null +++ b/integrationtests/djournal_extended_test.go @@ -0,0 +1,1028 @@ +//go:build linux + +package integrationtests + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + osuser "os/user" + "path/filepath" + "regexp" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/mimecast/dtail/internal/config" + journaltest "github.com/mimecast/dtail/internal/io/journal/testhelper" + "github.com/mimecast/dtail/internal/protocol" +) + +const ( + journalExtendedSpec = "journal:extended.service" + mixedSourceMinLinesPerSource = 2 + mixedSourceMinSwitches = 2 +) + +func TestDJournalExtendedWithServer(t *testing.T) { + if !config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") { + t.Log("Skipping") + return + } + + cleanupTmpFiles(t) + testLogger := NewTestLogger("TestDJournalExtendedWithServer") + defer testLogger.WriteLogFile() + + t.Run("MixedFileAndJournalFollow", func(t *testing.T) { + testDJournalExtendedMixedSources(t, testLogger) + }) + t.Run("PermissionRegexDenialAndSuccess", func(t *testing.T) { + testDJournalExtendedPermissions(t, testLogger) + }) + t.Run("MissingCapabilityDoesNotHang", func(t *testing.T) { + testDJournalExtendedMissingCapability(t, testLogger) + }) + t.Run("RetryFirstFailureThenSuccess", func(t *testing.T) { + testDJournalExtendedRetry(t, testLogger) + }) + t.Run("CleanShutdownMidFollow", func(t *testing.T) { + testDJournalExtendedCleanShutdown(t, testLogger) + }) + t.Run("TurboCompatibility", func(t *testing.T) { + testDJournalExtendedTurboCompatibility(t, testLogger) + }) + t.Run("DMapJournal", func(t *testing.T) { + testDJournalExtendedDMap(t, testLogger) + }) +} + +func testDJournalExtendedMixedSources(t *testing.T, logger *TestLogger) { + for _, mode := range dJournalExtendedServerModes() { + mode := mode + t.Run(mode.name, func(t *testing.T) { + testDJournalExtendedMixedSourcesForMode(t, logger, mode) + }) + } +} + +func testDJournalExtendedMixedSourcesForMode(t *testing.T, logger *TestLogger, + mode dJournalExtendedServerMode) { + + tmpDir := t.TempDir() + regularFile := filepath.Join(tmpDir, "mixed.log") + if err := os.WriteFile(regularFile, nil, 0o600); err != nil { + t.Fatalf("create mixed regular file: %v", err) + } + + env := newDJournalExtendedEnv(t, journaltest.Scenario{ + Units: map[string]journaltest.Invocation{ + "extended.service": { + FollowLines: []string{ + "journal mixed 1", + "journal mixed 2", + "journal mixed 3", + }, + InterLineDelay: 40 * time.Millisecond, + }, + }, + }, []string{ + permissionForPath(regularFile), + "readfiles:^journal:extended\\.service$", + }, map[string]any{ + "MaxConcurrentTails": 2, + }) + + server := startDJournalExtendedServerForMode(t, logger, env, "debug", mode) + clientCtx, clientCancel := context.WithTimeout(server.ctx, 15*time.Second) + defer clientCancel() + cmd, stdoutCh, stderrCh, cmdErrCh := startDJournalCommand(clientCtx, t, nil, "../dtail", + "--plain", + "--cfg", "none", + "--logLevel", "error", + "--servers", server.address, + "--files", strings.Join([]string{regularFile, journalExtendedSpec}, ","), + "--trustAllHosts", + "--noColor", + ) + + server.logs.waitContains(t, "Start reading|"+regularFile, 3*time.Second) + appendLinesAfterDelay(clientCtx, t, regularFile, []string{ + "file mixed 1", + "file mixed 2", + "file mixed 3", + "file mixed 4", + "file mixed 5", + }, 500*time.Millisecond) + + lines := readMixedSourceLines(clientCtx, t, stdoutCh, stderrCh, server.logs) + assertMixedSourcesInterleaved(t, lines) + stopProcessAndWait(t, cmd, cmdErrCh, "dtail mixed sources") +} + +func testDJournalExtendedPermissions(t *testing.T, logger *TestLogger) { + scenario := journaltest.Scenario{ + Units: map[string]journaltest.Invocation{ + "extended.service": { + Lines: []string{"journal permission allowed"}, + }, + }, + } + + denyEnv := newDJournalExtendedEnv(t, scenario, []string{ + "readfiles:^journal:other\\.service$", + }, nil) + denyServer := startDJournalExtendedServer(t, logger, denyEnv.configFile, denyEnv.mock.Env(), "debug", false) + denyOut := "djournal_extended_permission_deny.stdout.tmp" + // A permission-denied dcat exits cleanly after receiving the server WARN, so + // its error is not meaningful here; the assertion is on the server logs. + _, _ = runCommand(denyServer.ctx, t, denyOut, "../dcat", + "--cfg", "none", + "--logLevel", "error", + "--servers", denyServer.address, + "--files", journalExtendedSpec, + "--trustAllHosts", + "--noColor", + ) + denyOutput := readTestFile(t, denyOut) + // Poll the server logs rather than reading them synchronously: the server + // runs with the (buffered) stdout logger, so the "No permission" diagnostic + // can lag the client's exit by up to the logger's idle-flush interval. + // waitContains still fails if the line never appears, so the assertion is + // not weakened. + denyServer.logs.waitContains(t, "No permission to read file", 5*time.Second) + if strings.Contains(denyOutput, "journal permission allowed") { + t.Fatalf("permission denial leaked journal output:\n%s", denyOutput) + } + if got := strings.TrimSpace(denyEnv.mock.Args(t)); got != "" { + t.Fatalf("permission denial invoked journalctl unexpectedly; args:\n%s", got) + } + + allowEnv := newDJournalExtendedEnv(t, scenario, []string{ + "readfiles:^journal:extended\\.service$", + }, nil) + allowServer := startDJournalExtendedServer(t, logger, allowEnv.configFile, allowEnv.mock.Env(), "error", false) + allowOut := "djournal_extended_permission_allow.stdout.tmp" + _, err := runCommand(allowServer.ctx, t, allowOut, "../dcat", + "--plain", + "--cfg", "none", + "--logLevel", "error", + "--servers", allowServer.address, + "--files", journalExtendedSpec, + "--trustAllHosts", + "--noColor", + ) + if err != nil { + t.Fatalf("dcat journal with matching permission failed: %v", err) + } + if got, want := readTestFile(t, allowOut), "journal permission allowed\n"; got != want { + t.Fatalf("permission allow output mismatch\ngot:\n%swant:\n%s", got, want) + } +} + +func testDJournalExtendedMissingCapability(t *testing.T, logger *TestLogger) { + env := newDJournalExtendedEnv(t, journaltest.Scenario{}, []string{ + "readfiles:^journal:extended\\.service$", + }, nil) + serverEnv := env.mock.Env() + serverEnv["PATH"] = t.TempDir() + server := startDJournalExtendedServer(t, logger, env.configFile, serverEnv, "error", false) + + runCtx, runCancel := context.WithTimeout(server.ctx, 4*time.Second) + defer runCancel() + + outFile := "djournal_extended_missing_capability.stdout.tmp" + started := time.Now() + exitCode, err := runCommand(runCtx, t, outFile, "../dcat", + "--plain", + "--cfg", "none", + "--logLevel", "error", + "--servers", server.address, + "--files", journalExtendedSpec, + "--trustAllHosts", + "--noColor", + ) + if err == nil || exitCode == 0 { + t.Fatalf("dcat journal unexpectedly succeeded without %s", protocol.CapabilityJournalV1) + } + if elapsed := time.Since(started); elapsed > 3*time.Second { + t.Fatalf("missing capability path took too long: %s", elapsed) + } + got := readTestFile(t, outFile) + if !strings.Contains(got, "journal file targets require server capability "+protocol.CapabilityJournalV1) { + t.Fatalf("missing capability output does not explain journal support failure:\n%s", got) + } + if runCtx.Err() != nil { + t.Fatalf("missing capability test hit timeout instead of returning an error: %v", runCtx.Err()) + } +} + +func testDJournalExtendedRetry(t *testing.T, logger *TestLogger) { + for _, mode := range dJournalExtendedServerModes() { + mode := mode + t.Run(mode.name, func(t *testing.T) { + testDJournalExtendedRetryForMode(t, logger, mode) + }) + } +} + +func testDJournalExtendedRetryForMode(t *testing.T, logger *TestLogger, + mode dJournalExtendedServerMode) { + + const retryInterval = 300 * time.Millisecond + env := newDJournalExtendedEnv(t, journaltest.Scenario{ + Units: map[string]journaltest.Invocation{ + "extended.service": { + FailFirst: 1, + FollowLines: []string{"journal retry success"}, + }, + }, + }, []string{ + "readfiles:^journal:extended\\.service$", + }, map[string]any{ + "ReadRetryIntervalMs": int(retryInterval / time.Millisecond), + }) + server := startDJournalExtendedServerForMode(t, logger, env, "debug", mode) + + clientCtx, clientCancel := context.WithTimeout(server.ctx, 10*time.Second) + defer clientCancel() + started := time.Now() + cmd, stdoutCh, stderrCh, cmdErrCh := startDJournalCommand(clientCtx, t, nil, "../dtail", + "--plain", + "--cfg", "none", + "--logLevel", "error", + "--servers", server.address, + "--files", journalExtendedSpec, + "--trustAllHosts", + "--noColor", + ) + lines := readMatchingLines(clientCtx, t, stdoutCh, stderrCh, 1, func(line string) bool { + return strings.Contains(line, "journal retry success") + }) + if elapsed := time.Since(started); elapsed < retryInterval { + t.Fatalf("journal retry returned too quickly: elapsed %s, want at least %s; lines=%v", elapsed, retryInterval, lines) + } + if got := len(nonEmptyLines(env.mock.Args(t))); got < 2 { + t.Fatalf("journalctl invocations = %d, want at least 2; args:\n%s", got, env.mock.Args(t)) + } + // Stop the follow client first so the server read goroutine returns and logs + // its post-read "File processing complete" marker; the turbo-state assertion + // keys on that marker, which is only emitted once the read has completed. + stopProcessAndWait(t, cmd, cmdErrCh, "dtail retry") + assertJournalTurboLogState(t, server.logs, mode) +} + +func testDJournalExtendedCleanShutdown(t *testing.T, logger *TestLogger) { + for _, mode := range dJournalExtendedServerModes() { + mode := mode + t.Run(mode.name, func(t *testing.T) { + testDJournalExtendedCleanShutdownForMode(t, logger, mode) + }) + } +} + +func testDJournalExtendedCleanShutdownForMode(t *testing.T, logger *TestLogger, + mode dJournalExtendedServerMode) { + + env := newDJournalExtendedEnv(t, journaltest.Scenario{ + Units: map[string]journaltest.Invocation{ + "extended.service": { + FollowLines: []string{"journal shutdown stream"}, + InterLineDelay: 50 * time.Millisecond, + }, + }, + }, []string{ + "readfiles:^journal:extended\\.service$", + }, map[string]any{ + "MaxConcurrentTails": 1, + }) + server := startDJournalExtendedServerForMode(t, logger, env, "debug", mode) + + firstCtx, firstCancel := context.WithTimeout(server.ctx, 10*time.Second) + defer firstCancel() + cmd, stdoutCh, stderrCh, cmdErrCh := startDJournalCommand(firstCtx, t, nil, "../dtail", + "--plain", + "--cfg", "none", + "--logLevel", "error", + "--servers", server.address, + "--files", journalExtendedSpec, + "--trustAllHosts", + "--noColor", + ) + readMatchingLines(firstCtx, t, stdoutCh, stderrCh, 1, func(line string) bool { + return strings.Contains(line, "journal shutdown stream") + }) + firstPID := readMockPID(t, env.mock.PIDFile) + stopProcessAndWait(t, cmd, cmdErrCh, "dtail shutdown") + env.mock.WaitForTerm(t, 2*time.Second) + waitForProcessExit(t, firstPID, 2*time.Second) + server.logs.waitContains(t, "File processing complete", 3*time.Second) + server.logs.waitContains(t, "Command finished", 3*time.Second) + // Assert turbo state only after the read has completed: the helper keys on + // the post-read "File processing complete" marker (emitted strictly after the + // turbo decision), which is available now that the follow client has stopped. + assertJournalTurboLogState(t, server.logs, mode) + + secondCtx, secondCancel := context.WithTimeout(server.ctx, 10*time.Second) + defer secondCancel() + secondCmd, secondStdout, secondStderr, secondErrCh := startDJournalCommand(secondCtx, t, nil, "../dtail", + "--plain", + "--cfg", "none", + "--logLevel", "error", + "--servers", server.address, + "--files", journalExtendedSpec, + "--trustAllHosts", + "--noColor", + ) + readMatchingLines(secondCtx, t, secondStdout, secondStderr, 1, func(line string) bool { + return strings.Contains(line, "journal shutdown stream") + }) + stopProcessAndWait(t, secondCmd, secondErrCh, "dtail shutdown second") +} + +func testDJournalExtendedTurboCompatibility(t *testing.T, logger *TestLogger) { + run := func(t *testing.T, name string, serverEnv map[string]string) (string, *safeLineLog) { + t.Helper() + + env := newDJournalExtendedEnv(t, journaltest.Scenario{ + Units: map[string]journaltest.Invocation{ + "extended.service": { + Lines: []string{ + "journal turbo alpha", + "journal turbo beta", + }, + }, + }, + }, []string{ + "readfiles:^journal:extended\\.service$", + }, nil) + for key, value := range env.mock.Env() { + serverEnv[key] = value + } + serverEnv["DTAIL_HOSTNAME_OVERRIDE"] = "integrationtest" + server := startDJournalExtendedServer(t, logger, env.configFile, serverEnv, "debug", true) + + outFile := fmt.Sprintf("djournal_extended_turbo_%s.stdout.tmp", name) + _, err := runCommand(server.ctx, t, outFile, "../dcat", + "--plain", + "--cfg", "none", + "--logLevel", "error", + "--servers", server.address, + "--files", journalExtendedSpec, + "--trustAllHosts", + "--noColor", + ) + if err != nil { + t.Fatalf("dcat turbo %s failed: %v", name, err) + } + return readTestFile(t, outFile), server.logs + } + + // The direct read path is now the one and only runtime path, so this is a + // single run. It preserves the real regression coverage: the exact journal + // payload and proof the optimized read path ran. + enabledOutput, enabledLogs := run(t, "enabled", map[string]string{}) + if want := "journal turbo alpha\njournal turbo beta\n"; enabledOutput != want { + t.Fatalf("turbo compatibility output mismatch\ngot:\n%swant:\n%s\nlogs:\n%s", + enabledOutput, want, enabledLogs.String()) + } + enabledLogs.waitContains(t, "Using turbo mode for reading", 3*time.Second) +} + +func testDJournalExtendedDMap(t *testing.T, logger *TestLogger) { + var expectedCSV string + for _, mode := range dJournalExtendedServerModes() { + mode := mode + t.Run(mode.name, func(t *testing.T) { + csv := testDJournalExtendedDMapForMode(t, logger, mode) + if expectedCSV == "" { + expectedCSV = csv + return + } + if csv != expectedCSV { + t.Fatalf("dmap journal output mismatch with prior turbo mode\ngot:\n%swant:\n%s", csv, expectedCSV) + } + }) + } +} + +func testDJournalExtendedDMapForMode(t *testing.T, logger *TestLogger, + mode dJournalExtendedServerMode) string { + + env := newDJournalExtendedEnv(t, journaltest.Scenario{ + Units: map[string]journaltest.Invocation{ + "extended.service": { + Lines: []string{ + "INFO|1002-071143|1|stats.go:56|8|13|7|0.21|471h0m21s|MAPREDUCE:STATS|currentConnections=0|lifetimeConnections=1", + "INFO|1002-071143|1|stats.go:56|8|13|7|0.21|471h0m21s|MAPREDUCE:STATS|currentConnections=0|lifetimeConnections=1", + "INFO|1002-071143|1|stats.go:56|8|13|7|0.21|471h0m21s|MAPREDUCE:STATS|currentConnections=0|lifetimeConnections=1", + }, + }, + }, + }, []string{ + "readfiles:^journal:extended\\.service$", + }, nil) + server := startDJournalExtendedServerForMode(t, logger, env, "debug", mode) + + csvFile := filepath.Join(t.TempDir(), fmt.Sprintf("djournal_extended_dmap_%s.csv", mode.name)) + queryFile := fmt.Sprintf("%s.query", csvFile) + query := fmt.Sprintf("from STATS select count($line),$hostname group by $hostname outfile %s", csvFile) + cleanupFiles(t, csvFile, queryFile) + + runCtx, runCancel := context.WithTimeout(server.ctx, 30*time.Second) + defer runCancel() + outFile := fmt.Sprintf("djournal_extended_dmap_%s.stdout.tmp", mode.name) + args := []string{ + "--cfg", "none", + "--logLevel", "error", + "--noColor", + "--query", query, + "--servers", server.address, + "--trustAllHosts", + "--files", journalExtendedSpec, + } + if mode.expectTurbo { + cmd, stdoutCh, stderrCh, cmdErrCh := startDJournalCommand(runCtx, t, nil, "../dmap", args...) + assertDMapTurboLogState(t, server.logs, mode) + got := waitFileContains(runCtx, t, csvFile, "3,integrationtest") + drainCommandOutput(t, stdoutCh, stderrCh) + stopProcessAndWait(t, cmd, cmdErrCh, "dmap journal turbo") + if !strings.Contains(got, "count($line),$hostname") { + t.Fatalf("dmap journal csv missing header:\n%s", got) + } + if err := verifyQueryFile(t, queryFile, query); err != nil { + t.Fatal(err) + } + return got + } + + _, err := runCommand(runCtx, t, outFile, "../dmap", args...) + if err != nil { + t.Fatalf("dmap journal failed: %v\nserver logs:\n%s", err, server.logs.String()) + } + + got := readTestFile(t, csvFile) + if !strings.Contains(got, "count($line),$hostname") || !strings.Contains(got, "3,integrationtest") { + t.Fatalf("dmap journal csv mismatch:\n%s", got) + } + if err := verifyQueryFile(t, queryFile, query); err != nil { + t.Fatal(err) + } + assertDMapTurboLogState(t, server.logs, mode) + return got +} + +type dJournalExtendedServerMode struct { + name string + env map[string]string + dropIntegrationEnv bool + expectTurbo bool +} + +func dJournalExtendedServerModes() []dJournalExtendedServerMode { + // Turbo is now the one and only runtime path, so the former two-mode + // comparison collapses to a single turbo-enabled mode. All per-mode + // golden/CSV assertions are preserved; only the redundant disabled iteration + // (and its byte-identity comparison) is dropped. + return []dJournalExtendedServerMode{ + { + name: "turbo-enabled", + env: map[string]string{ + "DTAIL_HOSTNAME_OVERRIDE": "integrationtest", + }, + dropIntegrationEnv: true, + expectTurbo: true, + }, + } +} + +type dJournalExtendedEnv struct { + configFile string + mock *journaltest.Mock +} + +func newDJournalExtendedEnv(t *testing.T, scenario journaltest.Scenario, + permissions []string, serverFields map[string]any) dJournalExtendedEnv { + + t.Helper() + + tmpDir := t.TempDir() + mock := journaltest.InstallMock(t, scenario) + cwd, err := os.Getwd() + if err != nil { + t.Fatalf("get test working directory: %v", err) + } + cacheDirAbs, err := os.MkdirTemp(cwd, "djournal-key-cache-") + if err != nil { + t.Fatalf("create journal extended key cache: %v", err) + } + t.Cleanup(func() { + os.RemoveAll(cacheDirAbs) + }) + if err := os.WriteFile(filepath.Join(cacheDirAbs, currentUsername(t)+".authorized_keys"), + readIntegrationPublicKey(t), 0o600); err != nil { + t.Fatalf("write journal extended authorized_keys cache: %v", err) + } + server := map[string]any{ + "HostKeyFile": filepath.Join(tmpDir, "ssh_host_key"), + "Permissions": map[string]any{ + "Default": permissions, + }, + } + for key, value := range serverFields { + server[key] = value + } + + content, err := json.MarshalIndent(map[string]any{ + "Common": map[string]any{ + "CacheDir": filepath.Base(cacheDirAbs), + }, + "Server": server, + }, "", " ") + if err != nil { + t.Fatalf("marshal journal extended config: %v", err) + } + + configFile := filepath.Join(tmpDir, "dtail.json") + if err := os.WriteFile(configFile, append(content, '\n'), 0o600); err != nil { + t.Fatalf("write journal extended config: %v", err) + } + return dJournalExtendedEnv{ + configFile: configFile, + mock: mock, + } +} + +func currentUsername(t *testing.T) string { + t.Helper() + + user, err := osuser.Current() + if err != nil { + t.Fatalf("get current user: %v", err) + } + return user.Username +} + +func readIntegrationPublicKey(t *testing.T) []byte { + t.Helper() + + for _, path := range []string{"id_rsa.pub", "../id_rsa.pub"} { + data, err := os.ReadFile(path) + if err == nil { + return data + } + if !os.IsNotExist(err) { + t.Fatalf("read integration public key %s: %v", path, err) + } + } + t.Fatal("integration public key id_rsa.pub not found") + return nil +} + +type dJournalExtendedServer struct { + ctx context.Context + address string + logs *safeLineLog +} + +func startDJournalExtendedServer(t *testing.T, logger *TestLogger, configFile string, + env map[string]string, logLevel string, dropIntegrationEnv bool) dJournalExtendedServer { + + t.Helper() + + port := getUniquePortNumber() + bindAddress := "localhost" + ctx, cancel := context.WithCancel(context.Background()) + ctx = WithTestLogger(ctx, logger) + logs := &safeLineLog{} + cmd := exec.CommandContext(ctx, "../dserver", + "--cfg", configFile, + "--logger", "stdout", + "--logLevel", logLevel, + "--bindAddress", bindAddress, + "--port", fmt.Sprintf("%d", port), + ) + cmd.Env = mergedCommandEnv(env, dropIntegrationEnv) + stdout, err := cmd.StdoutPipe() + if err != nil { + cancel() + t.Fatalf("open dserver stdout: %v", err) + } + stderr, err := cmd.StderrPipe() + if err != nil { + cancel() + t.Fatalf("open dserver stderr: %v", err) + } + if err := cmd.Start(); err != nil { + cancel() + t.Fatalf("start dserver: %v", err) + } + go scanToLog(stdout, logs) + go scanToLog(stderr, logs) + + errCh := make(chan error, 1) + go func() { + errCh <- cmd.Wait() + }() + t.Cleanup(func() { + cancel() + select { + case <-errCh: + case <-time.After(3 * time.Second): + if cmd.Process != nil { + _ = cmd.Process.Kill() + } + <-errCh + } + }) + + if err := waitForServerReady(ctx, bindAddress, port); err != nil { + cancel() + t.Fatalf("wait for dserver: %v\nlogs:\n%s", err, logs.String()) + } + + return dJournalExtendedServer{ + ctx: ctx, + address: fmt.Sprintf("%s:%d", bindAddress, port), + logs: logs, + } +} + +func startDJournalExtendedServerForMode(t *testing.T, logger *TestLogger, + env dJournalExtendedEnv, logLevel string, mode dJournalExtendedServerMode) dJournalExtendedServer { + + t.Helper() + + serverEnv := env.mock.Env() + for key, value := range mode.env { + serverEnv[key] = value + } + return startDJournalExtendedServer(t, logger, env.configFile, serverEnv, logLevel, mode.dropIntegrationEnv) +} + +type safeLineLog struct { + mu sync.Mutex + lines []string +} + +func (l *safeLineLog) add(line string) { + l.mu.Lock() + defer l.mu.Unlock() + l.lines = append(l.lines, line) +} + +func (l *safeLineLog) contains(substr string) bool { + l.mu.Lock() + defer l.mu.Unlock() + for _, line := range l.lines { + if strings.Contains(line, substr) { + return true + } + } + return false +} + +func (l *safeLineLog) String() string { + l.mu.Lock() + defer l.mu.Unlock() + return strings.Join(l.lines, "\n") +} + +func (l *safeLineLog) waitContains(t *testing.T, substr string, timeout time.Duration) { + t.Helper() + + deadline := time.Now().Add(timeout) + for { + if l.contains(substr) { + return + } + if time.Now().After(deadline) { + t.Fatalf("server logs did not contain %q before timeout; logs:\n%s", substr, l.String()) + } + time.Sleep(20 * time.Millisecond) + } +} + +func scanToLog(r io.Reader, logs *safeLineLog) { + scanner := bufio.NewScanner(r) + for scanner.Scan() { + logs.add(scanner.Text()) + } +} + +func mergedCommandEnv(overrides map[string]string, dropIntegrationEnv bool) []string { + skip := make(map[string]struct{}, len(overrides)+2) + for key := range overrides { + skip[key] = struct{}{} + } + if dropIntegrationEnv { + skip["DTAIL_INTEGRATION_TEST_RUN_MODE"] = struct{}{} + skip["DTAIL_TURBOBOOST_DISABLE"] = struct{}{} + } + + env := make([]string, 0, len(os.Environ())+len(overrides)) + for _, item := range os.Environ() { + key, _, found := strings.Cut(item, "=") + if found { + if _, ok := skip[key]; ok { + continue + } + } + env = append(env, item) + } + for key, value := range overrides { + env = append(env, fmt.Sprintf("%s=%s", key, value)) + } + return env +} + +func permissionForPath(path string) string { + return "readfiles:^" + regexp.QuoteMeta(path) + "$" +} + +func appendLinesAfterDelay(ctx context.Context, t *testing.T, path string, + lines []string, delay time.Duration) { + + t.Helper() + + go func() { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return + case <-timer.C: + } + + file, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0) + if err != nil { + t.Errorf("open %s for append: %v", path, err) + return + } + defer file.Close() + for _, line := range lines { + if _, err := fmt.Fprintln(file, line); err != nil { + t.Errorf("append to %s: %v", path, err) + return + } + timer := time.NewTimer(delay) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + } + }() +} + +func readMatchingLines(ctx context.Context, t *testing.T, stdoutCh, stderrCh <-chan string, + count int, match func(string) bool) []string { + + t.Helper() + + lines := make([]string, 0, count) + for len(lines) < count { + select { + case line, ok := <-stdoutCh: + if !ok { + t.Fatalf("stdout closed after %d/%d matching lines: %v", len(lines), count, lines) + } + if match(line) { + lines = append(lines, line) + } + case line, ok := <-stderrCh: + if !ok { + stderrCh = nil + continue + } + if line != "" { + t.Log("client stderr:", line) + } + case <-ctx.Done(): + t.Fatalf("timed out waiting for matching lines: %v; got %v", ctx.Err(), lines) + } + } + return lines +} + +func readMixedSourceLines(ctx context.Context, t *testing.T, + stdoutCh, stderrCh <-chan string, logs *safeLineLog) []string { + + t.Helper() + + var lines []string + for { + if mixedSourcesInterleaved(lines) { + return lines + } + + select { + case line, ok := <-stdoutCh: + if !ok { + t.Fatalf("stdout closed before mixed sources appeared: %v", lines) + } + if strings.Contains(line, "mixed") { + lines = append(lines, line) + } + case line, ok := <-stderrCh: + if !ok { + stderrCh = nil + continue + } + if line != "" { + t.Log("client stderr:", line) + } + case <-ctx.Done(): + t.Fatalf("timed out waiting for mixed source output: %v; got %v\nserver logs:\n%s", + ctx.Err(), lines, logs.String()) + } + } +} + +func waitFileContains(ctx context.Context, t *testing.T, path, substr string) string { + t.Helper() + + ticker := time.NewTicker(50 * time.Millisecond) + defer ticker.Stop() + var last string + for { + content, err := os.ReadFile(path) + if err == nil { + last = string(content) + if strings.Contains(last, substr) { + return last + } + } else if !os.IsNotExist(err) { + t.Fatalf("read %s: %v", path, err) + } + + select { + case <-ticker.C: + case <-ctx.Done(): + t.Fatalf("timed out waiting for %s to contain %q: %v; last content:\n%s", + path, substr, ctx.Err(), last) + } + } +} + +func drainCommandOutput(t *testing.T, stdoutCh, stderrCh <-chan string) { + t.Helper() + + for { + select { + case line, ok := <-stdoutCh: + if ok && line != "" { + t.Log("client stdout:", line) + } + case line, ok := <-stderrCh: + if ok && line != "" { + t.Log("client stderr:", line) + } + default: + return + } + } +} + +func assertJournalTurboLogState(t *testing.T, logs *safeLineLog, mode dJournalExtendedServerMode) { + t.Helper() + + // Turbo is now the one and only runtime path, so only the positive assertion + // remains: the former negative (turbo-absent) branch is dead and was deleted. + if mode.expectTurbo { + logs.waitContains(t, "Using turbo mode for reading", 3*time.Second) + } +} + +func assertDMapTurboLogState(t *testing.T, logs *safeLineLog, mode dJournalExtendedServerMode) { + t.Helper() + + // Turbo is now the one and only runtime path, so only the positive assertion + // remains: the former negative (turbo-absent) branch is dead and was deleted. + if mode.expectTurbo { + logs.waitContains(t, "Creating turbo aggregate for MapReduce", 3*time.Second) + logs.waitContains(t, "Using turbo aggregate processor for MapReduce", 3*time.Second) + } +} + +func assertMixedSourcesInterleaved(t *testing.T, lines []string) { + t.Helper() + + if mixedSourcesInterleaved(lines) { + return + } + t.Fatalf("mixed source output was not sufficiently interleaved: file lines=%d journal lines=%d switches=%d lines=%v", + countLinePrefix(lines, "file mixed "), countLinePrefix(lines, "journal mixed "), + sourceSwitchCount(lines), lines) +} + +func mixedSourcesInterleaved(lines []string) bool { + return countLinePrefix(lines, "file mixed ") >= mixedSourceMinLinesPerSource && + countLinePrefix(lines, "journal mixed ") >= mixedSourceMinLinesPerSource && + sourceSwitchCount(lines) >= mixedSourceMinSwitches +} + +func countLinePrefix(lines []string, prefix string) int { + count := 0 + for _, line := range lines { + if strings.HasPrefix(line, prefix) { + count++ + } + } + return count +} + +func sourceSwitchCount(lines []string) int { + lastSource := "" + switches := 0 + for _, line := range lines { + source := "" + switch { + case strings.HasPrefix(line, "file mixed "): + source = "file" + case strings.HasPrefix(line, "journal mixed "): + source = "journal" + } + if source == "" { + continue + } + if lastSource != "" && source != lastSource { + switches++ + } + lastSource = source + } + return switches +} + +func stopProcessAndWait(t *testing.T, cmd *exec.Cmd, cmdErrCh <-chan error, name string) { + t.Helper() + + if cmd.ProcessState == nil { + if err := cmd.Process.Signal(syscall.SIGTERM); err != nil && !errors.Is(err, os.ErrProcessDone) { + t.Fatalf("signal %s: %v", name, err) + } + } + select { + case err := <-cmdErrCh: + if err != nil { + t.Fatalf("%s did not terminate cleanly after SIGTERM: %v", name, err) + } + case <-time.After(5 * time.Second): + t.Fatalf("%s did not terminate after SIGTERM", name) + } +} + +func nonEmptyLines(value string) []string { + var lines []string + for _, line := range strings.Split(value, "\n") { + if strings.TrimSpace(line) != "" { + lines = append(lines, line) + } + } + return lines +} + +func readMockPID(t *testing.T, path string) int { + t.Helper() + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read mock pid file: %v", err) + } + var pid int + if _, err := fmt.Sscanf(strings.TrimSpace(string(data)), "%d", &pid); err != nil { + t.Fatalf("parse mock pid: %v", err) + } + return pid +} + +func waitForProcessExit(t *testing.T, pid int, timeout time.Duration) { + t.Helper() + + deadline := time.Now().Add(timeout) + for { + if !processExists(pid) { + return + } + if time.Now().After(deadline) { + t.Fatalf("process %d still exists after %s", pid, timeout) + } + time.Sleep(20 * time.Millisecond) + } +} + +func processExists(pid int) bool { + if pid <= 0 { + return false + } + err := syscall.Kill(pid, 0) + return err == nil || errors.Is(err, syscall.EPERM) +} |