path: root/integrationtests/djournal_test.go

//go:build linux

package integrationtests

import (
	"bufio"
	"context"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"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 (
	journalTestSpec = "journal:test.service"
)

func TestDJournalWithServer(t *testing.T) {
	if !config.Env("DTAIL_INTEGRATION_TEST_RUN_MODE") {
		t.Log("Skipping")
		return
	}

	cleanupTmpFiles(t)
	testLogger := NewTestLogger("TestDJournalWithServer")
	defer testLogger.WriteLogFile()

	t.Run("DCatBoundedJournal", func(t *testing.T) {
		testDJournalDCatBounded(t, testLogger)
	})
	t.Run("DTailFollowJournal", func(t *testing.T) {
		testDJournalDTailFollow(t, testLogger)
	})
	t.Run("DGrepJournalRegex", func(t *testing.T) {
		testDJournalDGrepRegex(t, testLogger)
	})
	t.Run("MissingJournalCapability", func(t *testing.T) {
		testDJournalMissingCapability(t, testLogger)
	})
}

func testDJournalDCatBounded(t *testing.T, logger *TestLogger) {
	env := newJournalTestEnv(t)
	ctx, cancel, address := startDJournalServer(t, logger, env.configFile, env.ServerEnv())
	defer cancel()

	outFile := "djournal_dcat.stdout.tmp"
	_, err := runCommand(ctx, t, outFile, "../dcat",
		"--plain",
		"--cfg", "none",
		"--logLevel", "error",
		"--servers", address,
		"--files", journalTestSpec,
		"--trustAllHosts",
		"--noColor",
	)
	if err != nil {
		t.Fatalf("dcat journal failed: %v", err)
	}

	got := readTestFile(t, outFile)
	if got != journalBoundedOutput() {
		t.Fatalf("dcat journal output mismatch\ngot:\n%swant:\n%s", got, journalBoundedOutput())
	}
	assertJournalctlArgs(t, env.argsFile, "-u test.service")
}

func testDJournalDTailFollow(t *testing.T, logger *TestLogger) {
	env := newJournalTestEnv(t)
	ctx, cancel, address := startDJournalServer(t, logger, env.configFile, env.ServerEnv())
	defer cancel()

	clientCtx, clientCancel := context.WithTimeout(ctx, 10*time.Second)
	defer clientCancel()
	cmd, stdoutCh, stderrCh, cmdErrCh := startDJournalCommand(clientCtx, t, nil, "../dtail",
		"--plain",
		"--cfg", "none",
		"--logLevel", "error",
		"--servers", address,
		"--files", journalTestSpec,
		"--trustAllHosts",
		"--noColor",
	)

	got := readJournalFollowLines(clientCtx, t, stdoutCh, stderrCh, 3)
	for i, line := range got {
		want := fmt.Sprintf("journal follow %d", i+1)
		if line != want {
			t.Fatalf("follow line %d = %q, want %q; all lines: %v", i, line, want, got)
		}
	}

	if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
		t.Fatalf("signal dtail: %v", err)
	}
	select {
	case err := <-cmdErrCh:
		if err != nil {
			t.Fatalf("dtail did not terminate cleanly after SIGTERM: %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("dtail did not terminate after SIGTERM")
	}
	waitForFileContent(t, env.termFile, "term", 2*time.Second)
}

func testDJournalDGrepRegex(t *testing.T, logger *TestLogger) {
	env := newJournalTestEnv(t)
	ctx, cancel, address := startDJournalServer(t, logger, env.configFile, env.ServerEnv())
	defer cancel()

	outFile := "djournal_dgrep.stdout.tmp"
	_, err := runCommand(ctx, t, outFile, "../dgrep",
		"--plain",
		"--cfg", "none",
		"--logLevel", "error",
		"--regex", "match-[0-9]+",
		"--servers", address,
		"--files", journalTestSpec,
		"--trustAllHosts",
		"--noColor",
	)
	if err != nil {
		t.Fatalf("dgrep journal failed: %v", err)
	}

	got := readTestFile(t, outFile)
	want := "journal beta match-123\njournal gamma match-456\n"
	if got != want {
		t.Fatalf("dgrep journal output mismatch\ngot:\n%swant:\n%s", got, want)
	}
}

func testDJournalMissingCapability(t *testing.T, logger *TestLogger) {
	env := newJournalTestEnv(t)
	noJournalPath := t.TempDir()
	serverEnv := env.ServerEnv()
	serverEnv["PATH"] = noJournalPath
	ctx, cancel, address := startDJournalServer(t, logger, env.configFile, serverEnv)
	defer cancel()

	runCtx, runCancel := context.WithTimeout(ctx, 4*time.Second)
	defer runCancel()

	outFile := "djournal_missing_capability.stdout.tmp"
	started := time.Now()
	exitCode, err := runCommand(runCtx, t, outFile, "../dcat",
		"--plain",
		"--cfg", "none",
		"--logLevel", "error",
		"--servers", address,
		"--files", journalTestSpec,
		"--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())
	}
}

type journalTestEnv struct {
	argsFile   string
	configFile string
	path       string
	termFile   string
}

func newJournalTestEnv(t *testing.T) journalTestEnv {
	t.Helper()

	tmpDir := t.TempDir()
	mock := journaltest.InstallMock(t, journaltest.Scenario{
		Default: journaltest.Invocation{
			Lines: []string{
				"journal alpha",
				"journal beta match-123",
				"journal gamma match-456",
			},
			FollowLines: []string{
				"journal follow 1",
				"journal follow 2",
				"journal follow 3",
			},
			InterLineDelay: 50 * time.Millisecond,
		},
	})

	configFile := filepath.Join(tmpDir, "dtail.json")
	configContent := fmt.Sprintf(`{
  "Server": {
    "HostKeyFile": %q,
    "Permissions": {
      "Default": [
        "readfiles:^journal:test\\.service$"
      ]
    }
  }
}
`, filepath.Join(tmpDir, "ssh_host_key"))
	if err := os.WriteFile(configFile, []byte(configContent), 0600); err != nil {
		t.Fatalf("write journal test config: %v", err)
	}

	return journalTestEnv{
		argsFile:   mock.ArgsFile,
		configFile: configFile,
		path:       mock.Path,
		termFile:   mock.TermFile,
	}
}

func (e journalTestEnv) ServerEnv() map[string]string {
	return map[string]string{
		"PATH": e.path,
	}
}

func journalBoundedOutput() string {
	return "journal alpha\njournal beta match-123\njournal gamma match-456\n"
}

func startDJournalServer(t *testing.T, logger *TestLogger, configFile string, env map[string]string) (context.Context, context.CancelFunc, string) {
	t.Helper()

	port := getUniquePortNumber()
	bindAddress := "localhost"
	ctx, cancel := context.WithCancel(context.Background())
	ctx = WithTestLogger(ctx, logger)
	t.Cleanup(cancel)

	_, _, _, err := startCommandWithEnv(ctx, t, "", "../dserver", env,
		"--cfg", configFile,
		"--logger", "stdout",
		"--logLevel", "error",
		"--bindAddress", bindAddress,
		"--port", fmt.Sprintf("%d", port),
	)
	if err != nil {
		cancel()
		t.Fatalf("start dserver: %v", err)
	}
	if err := waitForServerReady(ctx, bindAddress, port); err != nil {
		cancel()
		t.Fatalf("wait for dserver: %v", err)
	}

	return ctx, cancel, fmt.Sprintf("%s:%d", bindAddress, port)
}

func startDJournalCommand(ctx context.Context, t *testing.T, env map[string]string,
	cmdStr string, args ...string) (*exec.Cmd, <-chan string, <-chan string, <-chan error) {

	t.Helper()
	cmd := exec.CommandContext(ctx, cmdStr, args...)
	cmd.Env = os.Environ()
	for key, value := range env {
		cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", key, value))
	}

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		t.Fatalf("open stdout pipe: %v", err)
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		t.Fatalf("open stderr pipe: %v", err)
	}
	if err := cmd.Start(); err != nil {
		t.Fatalf("start %s: %v", cmdStr, err)
	}

	stdoutCh := scanLines(stdout)
	stderrCh := scanLines(stderr)
	errCh := make(chan error, 1)
	go func() {
		errCh <- cmd.Wait()
	}()
	t.Cleanup(func() {
		if cmd.ProcessState == nil {
			_ = cmd.Process.Kill()
		}
	})

	return cmd, stdoutCh, stderrCh, errCh
}

func scanLines(r io.Reader) <-chan string {
	ch := make(chan string, 100)
	go func() {
		defer close(ch)
		scanner := bufio.NewScanner(r)
		for scanner.Scan() {
			ch <- scanner.Text()
		}
	}()
	return ch
}

func readJournalFollowLines(ctx context.Context, t *testing.T, stdoutCh, stderrCh <-chan string, count int) []string {
	t.Helper()

	lines := make([]string, 0, count)
	for len(lines) < count {
		select {
		case line, ok := <-stdoutCh:
			if !ok {
				t.Fatalf("dtail stdout closed after %d/%d journal follow lines", len(lines), count)
			}
			if strings.Contains(line, "journal follow ") {
				lines = append(lines, line)
			}
		case line, ok := <-stderrCh:
			if !ok {
				stderrCh = nil
				continue
			}
			if line != "" {
				t.Log("dtail stderr:", line)
			}
		case <-ctx.Done():
			t.Fatalf("timed out waiting for journal follow lines: %v", ctx.Err())
		}
	}
	return lines
}

func readTestFile(t *testing.T, path string) string {
	t.Helper()

	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatalf("read %s: %v", path, err)
	}
	return string(data)
}

func assertJournalctlArgs(t *testing.T, path, want string) {
	t.Helper()

	got := strings.TrimSpace(readTestFile(t, path))
	if got != want {
		t.Fatalf("journalctl args = %q, want %q", got, want)
	}
}

func waitForFileContent(t *testing.T, path, want string, timeout time.Duration) {
	t.Helper()

	deadline := time.Now().Add(timeout)
	for {
		data, err := os.ReadFile(path)
		if err == nil && string(data) == want {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("file %s did not become %q before timeout; last read: %q, %v", path, want, string(data), err)
		}
		time.Sleep(20 * time.Millisecond)
	}
}