diff --git a/go.mod b/go.mod index a2626e3..16166a2 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.25.4 require ( github.com/cilium/ebpf v0.22.0 + github.com/docker/docker v28.3.3+incompatible github.com/prometheus/client_golang v1.23.2 github.com/spf13/cobra v1.10.2 github.com/spf13/viper v1.21.0 @@ -29,7 +30,6 @@ require ( github.com/cpuguy83/go-md2man/v2 v2.0.6 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/distribution/reference v0.6.0 // indirect - github.com/docker/docker v28.3.3+incompatible // indirect github.com/docker/go-connections v0.6.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/ebitengine/purego v0.8.4 // indirect diff --git a/internal/integration/start_runtime_test.go b/internal/integration/start_runtime_test.go new file mode 100644 index 0000000..e7f79ce --- /dev/null +++ b/internal/integration/start_runtime_test.go @@ -0,0 +1,193 @@ +// Copyright 2026 Optiqor contributors +// SPDX-License-Identifier: Apache-2.0 + +//go:build integration + +package integration + +import ( + "bytes" + "context" + "encoding/json" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "syscall" + "testing" + "time" +) + +func TestStartCommandServesHealthAndStopsCleanly(t *testing.T) { + if os.Geteuid() != 0 { + t.Skip("skip start runtime integration test: requires root") + } + + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + + repoRoot := integrationRepoRoot(t) + binPath := buildKernoBinary(t, ctx, repoRoot) + configPath := writeRuntimeConfig(t) + addr := reserveLoopbackAddr(t) + healthURL := "http://" + addr + "/healthz" + + cmd := exec.CommandContext( + ctx, + binPath, + "--config", configPath, + "--log-format", "json", + "start", + "--prometheus-addr", addr, + ) + cmd.Dir = repoRoot + + var stdout bytes.Buffer + var stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + + if err := cmd.Start(); err != nil { + t.Fatalf("start kerno runtime command: %v", err) + } + + waitCh := make(chan error, 1) + go func() { + waitCh <- cmd.Wait() + }() + + t.Cleanup(func() { + if cmd.Process == nil { + return + } + + select { + case <-waitCh: + return + default: + } + + _ = cmd.Process.Signal(syscall.SIGKILL) + select { + case <-waitCh: + case <-time.After(5 * time.Second): + } + }) + + body := waitForHealthz(t, ctx, healthURL, waitCh, &stdout, &stderr) + if got := body["status"]; got != "ok" { + t.Fatalf("healthz status = %v, want ok (body=%v)", got, body) + } + if _, ok := body["programs_loaded"].(float64); !ok { + t.Fatalf("healthz missing numeric programs_loaded (body=%v)", body) + } + if _, ok := body["programs_total"].(float64); !ok { + t.Fatalf("healthz missing numeric programs_total (body=%v)", body) + } + + if err := cmd.Process.Signal(syscall.SIGTERM); err != nil { + t.Fatalf("send SIGTERM to kerno runtime command: %v", err) + } + + select { + case err := <-waitCh: + if err != nil { + t.Fatalf("kerno runtime command exited with error: %v\nstdout:\n%s\nstderr:\n%s", err, stdout.String(), stderr.String()) + } + case <-time.After(10 * time.Second): + t.Fatalf("timed out waiting for kerno runtime command to stop\nstdout:\n%s\nstderr:\n%s", stdout.String(), stderr.String()) + } +} + +func integrationRepoRoot(t *testing.T) string { + t.Helper() + + wd, err := os.Getwd() + if err != nil { + t.Fatalf("get working directory: %v", err) + } + + dir := wd + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + return dir + } + parent := filepath.Dir(dir) + if parent == dir { + t.Fatal("could not find go.mod in any parent directory") + } + dir = parent + } +} + +func buildKernoBinary(t *testing.T, ctx context.Context, repoRoot string) string { + t.Helper() + + binPath := filepath.Join(t.TempDir(), "kerno") + buildCmd := exec.CommandContext(ctx, "go", "build", "-o", binPath, "./cmd/kerno") + buildCmd.Dir = repoRoot + + output, err := buildCmd.CombinedOutput() + if err != nil { + t.Fatalf("build kerno binary for runtime integration test: %v\n%s", err, output) + } + + return binPath +} + +func writeRuntimeConfig(t *testing.T) string { + t.Helper() + + configPath := filepath.Join(t.TempDir(), "config.yaml") + const configYAML = "prometheus:\n enabled: true\n" + + if err := os.WriteFile(configPath, []byte(configYAML), 0o644); err != nil { + t.Fatalf("write runtime integration config: %v", err) + } + + return configPath +} + +func reserveLoopbackAddr(t *testing.T) string { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("reserve loopback address: %v", err) + } + defer ln.Close() + + return ln.Addr().String() +} + +func waitForHealthz(t *testing.T, ctx context.Context, url string, waitCh <-chan error, stdout, stderr *bytes.Buffer) map[string]any { + t.Helper() + + client := &http.Client{Timeout: 2 * time.Second} + + for { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + t.Fatalf("create healthz request: %v", err) + } + + resp, err := client.Do(req) + if err == nil { + var body map[string]any + decodeErr := json.NewDecoder(resp.Body).Decode(&body) + closeErr := resp.Body.Close() + if resp.StatusCode == http.StatusOK && decodeErr == nil && closeErr == nil { + return body + } + } + + select { + case waitErr := <-waitCh: + t.Fatalf("kerno runtime command exited before healthz became ready: %v\nstdout:\n%s\nstderr:\n%s", waitErr, stdout.String(), stderr.String()) + case <-ctx.Done(): + t.Fatalf("timed out waiting for %s: %v\nstdout:\n%s\nstderr:\n%s", url, ctx.Err(), stdout.String(), stderr.String()) + case <-time.After(250 * time.Millisecond): + } + } +} diff --git a/internal/integration/systemd_service_test.go b/internal/integration/systemd_service_test.go new file mode 100644 index 0000000..379914c --- /dev/null +++ b/internal/integration/systemd_service_test.go @@ -0,0 +1,177 @@ +// Copyright 2026 Optiqor contributors +// SPDX-License-Identifier: Apache-2.0 + +//go:build integration + +package integration + +import ( + "context" + "encoding/json" + "fmt" + "io" + "path/filepath" + "strings" + "testing" + "time" + + dockercontainer "github.com/docker/docker/api/types/container" + "github.com/testcontainers/testcontainers-go" + tcexec "github.com/testcontainers/testcontainers-go/exec" + "github.com/testcontainers/testcontainers-go/wait" +) + +const systemdRuntimeImage = "docker.io/jrei/systemd-ubuntu@sha256:c746f3cb801d8c070ec230a0eb3414d9b537f5d4b646a27a776cde3870d6f951" + +func TestSystemdUnitStartsAndServesHealthz(t *testing.T) { + testcontainers.SkipIfProviderIsNotHealthy(t) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + + repoRoot := integrationRepoRoot(t) + binPath := buildKernoBinary(t, ctx, repoRoot) + unitPath := filepath.Join(repoRoot, "deploy/systemd/kerno.service") + configPath := filepath.Join(repoRoot, "deploy/systemd/kerno.yaml") + + container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ + ContainerRequest: testcontainers.ContainerRequest{ + Image: systemdRuntimeImage, + WaitingFor: wait.ForExec([]string{ + "bash", "-lc", "ps -p 1 -o comm= | grep -qx systemd", + }).WithStartupTimeout(45 * time.Second), + HostConfigModifier: func(hc *dockercontainer.HostConfig) { + hc.Privileged = true + hc.CgroupnsMode = "host" + hc.Binds = append(hc.Binds, "/sys/fs/cgroup:/sys/fs/cgroup:rw") + }, + }, + Started: true, + }) + if err != nil { + t.Skipf("skip systemd service integration test: start systemd container: %v", err) + } + + t.Cleanup(func() { + terminateCtx, terminateCancel := context.WithTimeout(context.Background(), 15*time.Second) + defer terminateCancel() + _ = container.Terminate(terminateCtx) + }) + + t.Cleanup(func() { + if !t.Failed() { + return + } + + status, _, _ := execInContainer(ctx, container, []string{"systemctl", "status", "kerno", "--no-pager", "-l"}) + journal, _, _ := execInContainer(ctx, container, []string{"journalctl", "-u", "kerno", "--no-pager", "-n", "200"}) + t.Logf("systemctl status kerno:\n%s", status) + t.Logf("journalctl -u kerno:\n%s", journal) + }) + + mustExecInContainer(t, ctx, container, []string{"mkdir", "-p", "/etc/kerno"}) + + if err := container.CopyFileToContainer(ctx, binPath, "/usr/bin/kerno", 0o755); err != nil { + t.Fatalf("copy kerno binary into systemd container: %v", err) + } + if err := container.CopyFileToContainer(ctx, unitPath, "/etc/systemd/system/kerno.service", 0o644); err != nil { + t.Fatalf("copy systemd unit into systemd container: %v", err) + } + if err := container.CopyFileToContainer(ctx, configPath, "/etc/kerno/config.yaml", 0o644); err != nil { + t.Fatalf("copy systemd config into systemd container: %v", err) + } + + mustExecInContainer(t, ctx, container, []string{"systemctl", "daemon-reload"}) + mustExecInContainer(t, ctx, container, []string{"systemctl", "start", "kerno"}) + + waitForSystemdState(t, ctx, container, "kerno", "active") + + body := waitForContainerHealthz(t, ctx, container, "http://127.0.0.1:9090/healthz") + if got := body["status"]; got != "ok" { + t.Fatalf("healthz status = %v, want ok (body=%v)", got, body) + } + if _, ok := body["programs_loaded"].(float64); !ok { + t.Fatalf("healthz missing numeric programs_loaded (body=%v)", body) + } + if _, ok := body["programs_total"].(float64); !ok { + t.Fatalf("healthz missing numeric programs_total (body=%v)", body) + } + + mustExecInContainer(t, ctx, container, []string{"systemctl", "stop", "kerno"}) + waitForSystemdState(t, ctx, container, "kerno", "inactive") +} + +func execInContainer(ctx context.Context, container testcontainers.Container, cmd []string) (string, int, error) { + exitCode, reader, err := container.Exec(ctx, cmd, tcexec.Multiplexed()) + if err != nil { + return "", exitCode, err + } + + output, readErr := io.ReadAll(reader) + if readErr != nil { + return "", exitCode, readErr + } + + return string(output), exitCode, nil +} + +func mustExecInContainer(t *testing.T, ctx context.Context, container testcontainers.Container, cmd []string) string { + t.Helper() + + output, exitCode, err := execInContainer(ctx, container, cmd) + if err != nil { + t.Fatalf("exec %q in container: %v", strings.Join(cmd, " "), err) + } + if exitCode != 0 { + t.Fatalf("exec %q in container exited %d\noutput:\n%s", strings.Join(cmd, " "), exitCode, output) + } + + return output +} + +func waitForSystemdState(t *testing.T, ctx context.Context, container testcontainers.Container, unit string, want string) { + t.Helper() + + deadline := time.Now().Add(20 * time.Second) + cmd := []string{"systemctl", "is-active", unit} + + for time.Now().Before(deadline) { + output, _, err := execInContainer(ctx, container, cmd) + if err == nil && strings.TrimSpace(output) == want { + return + } + + time.Sleep(250 * time.Millisecond) + } + + output, _, _ := execInContainer(ctx, container, cmd) + t.Fatalf("timed out waiting for %s to become %s (last output=%q)", unit, want, strings.TrimSpace(output)) +} + +func waitForContainerHealthz(t *testing.T, ctx context.Context, container testcontainers.Container, url string) map[string]any { + t.Helper() + + python := fmt.Sprintf( + `import json, urllib.request; print(json.dumps(json.load(urllib.request.urlopen(%q))))`, + url, + ) + + deadline := time.Now().Add(20 * time.Second) + cmd := []string{"python3", "-c", python} + + for time.Now().Before(deadline) { + output, exitCode, err := execInContainer(ctx, container, cmd) + if err == nil && exitCode == 0 { + var body map[string]any + if jsonErr := json.Unmarshal([]byte(output), &body); jsonErr == nil { + return body + } + } + + time.Sleep(250 * time.Millisecond) + } + + output, _, _ := execInContainer(ctx, container, cmd) + t.Fatalf("timed out waiting for %s (last output=%q)", url, output) + return nil +}