diff --git a/.github/scripts/with-capture-docker-logs.sh b/.github/scripts/with-capture-docker-logs.sh index 7482bf71..4e75dd5f 100755 --- a/.github/scripts/with-capture-docker-logs.sh +++ b/.github/scripts/with-capture-docker-logs.sh @@ -18,7 +18,8 @@ log_event() { # Named pipes so we can capture PIDs of both sides of each pipeline IMAGE_FIFO=$(mktemp -u) CONTAINER_FIFO=$(mktemp -u) -mkfifo "$IMAGE_FIFO" "$CONTAINER_FIFO" +NETWORK_FIFO=$(mktemp -u) +mkfifo "$IMAGE_FIFO" "$CONTAINER_FIFO" "$NETWORK_FIFO" # --- Image pull logging --- docker events \ @@ -32,6 +33,18 @@ while read -r line; do done < "$IMAGE_FIFO" & IMAGE_READER_PID=$! +# --- Network lifecycle logging --- +docker events \ + --filter 'type=network' \ + --format '{{.Time}} NETWORK {{.Action}} name={{.Actor.Attributes.name}} id={{.Actor.ID}} container={{.Actor.Attributes.container}}' \ + > "$NETWORK_FIFO" & +NETWORK_EVENTS_PID=$! + +while read -r line; do + log_event "$line" +done < "$NETWORK_FIFO" & +NETWORK_READER_PID=$! + # --- Container lifecycle logging --- docker events \ --filter 'type=container' \ @@ -82,10 +95,12 @@ CONTAINER_READER_PID=$! cleanup() { log_event "Shutting down docker log collector" kill "$IMAGE_EVENTS_PID" "$IMAGE_READER_PID" \ + "$NETWORK_EVENTS_PID" "$NETWORK_READER_PID" \ "$CONTAINER_EVENTS_PID" "$CONTAINER_READER_PID" 2>/dev/null || true pkill -f "docker logs -f" 2>/dev/null || true - rm -f "$IMAGE_FIFO" "$CONTAINER_FIFO" + rm -f "$IMAGE_FIFO" "$CONTAINER_FIFO" "$NETWORK_FIFO" wait "$IMAGE_EVENTS_PID" "$IMAGE_READER_PID" \ + "$NETWORK_EVENTS_PID" "$NETWORK_READER_PID" \ "$CONTAINER_EVENTS_PID" "$CONTAINER_READER_PID" 2>/dev/null || true } trap cleanup EXIT diff --git a/pom.xml b/pom.xml index 711a143a..97cf9aec 100644 --- a/pom.xml +++ b/pom.xml @@ -22,7 +22,7 @@ 3.13.0 3.6.0 25.0.2 - 2.0.37 + 2.1.0-dev.STX-221.4 4.0.1 false @@ -167,6 +167,11 @@ ${cloudevents.version} + + jakarta.ws.rs + jakarta.ws.rs-api + + org.graalvm.sdk @@ -586,4 +591,4 @@ - \ No newline at end of file + diff --git a/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java b/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java index 23c3b653..b65191a2 100644 --- a/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java +++ b/src/test/java/com/streamx/cli/commands/local/run/RunCommandIT.java @@ -16,6 +16,7 @@ import io.quarkus.test.junit.QuarkusTest; import java.nio.file.Paths; import java.time.Duration; +import java.util.List; import java.util.UUID; import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterEach; @@ -26,17 +27,21 @@ @DisabledIfDockerUnavailable public class RunCommandIT extends CliBaseIT { - private static final String PREFIX = - "sx-run-" + UUID.randomUUID().toString().substring(0, 4) + "-"; + private static String freshPrefix() { + return "sx-run-" + UUID.randomUUID().toString().substring(0, 4) + "-"; + } private static final String WEB_SERVER_SINK_IMAGE = "ghcr.io/streamx-com/streamx-blueprints/web-server-sink:3.0.7-jvm"; private static final String BLOCKER_IMAGE = "alpine:3.20"; + private static final int SIGTERM_EXIT_CODE = 128 + 15; + @BeforeEach void isolateRunFromConcurrentInstances() { - System.setProperty("streamx.container.startup-timeout-seconds", "180"); + System.setProperty("streamx.container.startup-timeout-seconds", + MeshTestSupport.CONTAINER_STARTUP_TIMEOUT_SECONDS); System.setProperty("streamx.runner.pulsar.broker-port", String.valueOf(MeshTestSupport.freePort())); System.setProperty("streamx.runner.pulsar.http-port", @@ -51,6 +56,7 @@ void stopMeshAndResetRunnerState() { } catch (Exception ignored) { // best-effort cleanup } + awaitAsyncCommands(); System.clearProperty("streamx.runner.mesh-name-prefix"); System.clearProperty("streamx.container.startup-timeout-seconds"); System.clearProperty("streamx.runner.pulsar.broker-port"); @@ -60,7 +66,7 @@ void stopMeshAndResetRunnerState() { @Test void shouldWarnWhenEnvVariableIsUndefined() throws Exception { - System.setProperty("streamx.runner.mesh-name-prefix", PREFIX); + System.setProperty("streamx.runner.mesh-name-prefix", freshPrefix()); exec("settings", "set", "config.image.interpolated", WEB_SERVER_SINK_IMAGE); exec("settings", "unset", "STREAMX_OWNER_SERVICE_NAME"); clearEnv("STREAMX_OWNER_SERVICE_NAME"); @@ -73,18 +79,14 @@ void shouldWarnWhenEnvVariableIsUndefined() throws Exception { AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath); try { - Awaitility.await() - .atMost(Duration.ofMinutes(3)) - .pollInterval(Duration.ofSeconds(1)) - .until(() -> handle.getStderr() - .contains("Environment variable 'STREAMX_OWNER_SERVICE_NAME'")); + awaitStderrContains(handle, "Environment variable 'STREAMX_OWNER_SERVICE_NAME'"); assertThat(handle.getStderr()) .contains("WARNING:") .contains("STREAMX_OWNER_SERVICE_NAME"); } finally { - if (handle.thread().isAlive()) { - handle.interruptAndJoin(Duration.ofSeconds(30).toMillis()); + if (handle.isAlive()) { + handle.stopAndJoin(Duration.ofSeconds(30).toMillis()); } } } @@ -101,9 +103,10 @@ void shouldFailWhenMeshFileDoesNotExist() throws Exception { @Test void shouldReportContainerFailureWhenItsHostPortIsAlreadyTaken() throws Exception { - System.setProperty("streamx.runner.mesh-name-prefix", PREFIX); + String prefix = freshPrefix(); + System.setProperty("streamx.runner.mesh-name-prefix", prefix); exec("settings", "set", "config.image.interpolated", WEB_SERVER_SINK_IMAGE); - exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", PREFIX + "test-owner"); + exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", prefix + "test-owner"); String meshPath = Paths.get("target/test-classes/mesh-interpolated.yaml") .toAbsolutePath() @@ -114,27 +117,32 @@ void shouldReportContainerFailureWhenItsHostPortIsAlreadyTaken() throws Exceptio String blockerId = startPortBlocker(blockedPort); System.setProperty("test.proxy.host-port", String.valueOf(blockedPort)); - AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath); + AsyncProcessHandle handle = execAsync("local", "run", "--verbose", "-f=" + meshPath); try { - Awaitility.await() - .atMost(Duration.ofMinutes(3)) - .pollInterval(Duration.ofSeconds(1)) - .untilAsserted(() -> { - assertThat(handle.getStdout()) - .as("the user must be told which container failed") - .contains("rest-ingestion.proxy failed"); - assertThat(handle.getStderr()) - .as("the run must be reported as failed") - .contains(msg.somethingWentWrong().strip()); - }); + awaitStderrContains(handle, msg.somethingWentWrong().strip(), + String.valueOf(blockedPort)); } finally { - if (handle.thread().isAlive()) { - handle.interruptAndJoin(Duration.ofSeconds(30).toMillis()); + if (handle.isAlive()) { + handle.stopAndJoin(Duration.ofSeconds(30).toMillis()); } removePortBlocker(blockerId); } } + private static void awaitNoContainersWithPrefix(String prefix) throws Exception { + try (DockerClient docker = DockerClientFactory.create()) { + Awaitility.await() + .atMost(Duration.ofSeconds(60)) + .pollInterval(Duration.ofSeconds(1)) + .untilAsserted(() -> assertThat(docker.listContainersCmd() + .withShowAll(true) + .withNameFilter(List.of(prefix)) + .exec()) + .as("mesh containers with prefix %s must be removed after the run stops", prefix) + .isEmpty()); + } + } + private static String startPortBlocker(int hostPort) throws Exception { try (DockerClient docker = DockerClientFactory.create()) { docker.pullImageCmd(BLOCKER_IMAGE).start().awaitCompletion(); @@ -162,9 +170,10 @@ private static void removePortBlocker(String containerId) { @Test void shouldFailWhenSystemPropertyIsUndefined() throws Exception { - System.setProperty("streamx.runner.mesh-name-prefix", PREFIX); + String prefix = freshPrefix(); + System.setProperty("streamx.runner.mesh-name-prefix", prefix); exec("settings", "unset", "config.image.interpolated"); - exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", PREFIX + "test-owner"); + exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", prefix + "test-owner"); String meshPath = Paths.get("target/test-classes/mesh-interpolated.yaml") .toAbsolutePath() @@ -177,7 +186,7 @@ void shouldFailWhenSystemPropertyIsUndefined() throws Exception { Awaitility.await() .atMost(Duration.ofMinutes(3)) .pollInterval(Duration.ofSeconds(1)) - .until(() -> !handle.thread().isAlive()); + .until(() -> !handle.isAlive()); ProcessResult result = handle.toResult(); assertThat(result.exitCode()).isNotEqualTo(0); @@ -185,8 +194,8 @@ void shouldFailWhenSystemPropertyIsUndefined() throws Exception { .contains("Property 'config.image.interpolated'") .contains("is not set"); } finally { - if (handle.thread().isAlive()) { - handle.interruptAndJoin(Duration.ofSeconds(30).toMillis()); + if (handle.isAlive()) { + handle.stopAndJoin(Duration.ofSeconds(30).toMillis()); } } } @@ -219,27 +228,26 @@ void shouldBridgeRunnerSettingToSystemPropertyForLocalRun() throws Exception { AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath); try { - Awaitility.await() - .atMost(Duration.ofMinutes(3)) - .pollInterval(Duration.ofSeconds(1)) - .until(() -> handle.getStdout().contains("STREAMX IS READY!")); + awaitStdoutContains(handle, "STREAMX IS READY!"); assertThat(handle.getStdout()) .as("runner should use the prefix from streamxHome settings via the bridge") .contains(bridgedPrefix); } finally { - if (handle.thread().isAlive()) { - handle.interruptAndJoin(Duration.ofSeconds(30).toMillis()); + if (handle.isAlive()) { + handle.stopAndJoin(Duration.ofSeconds(30).toMillis()); } + exec("settings", "unset", "streamx.runner.mesh-name-prefix"); System.clearProperty("streamx.runner.mesh-name-prefix"); } } @Test void shouldSucceedWhenInterpolationValuesAreDefined() throws Exception { - System.setProperty("streamx.runner.mesh-name-prefix", PREFIX); + String prefix = freshPrefix(); + System.setProperty("streamx.runner.mesh-name-prefix", prefix); exec("settings", "set", "config.image.interpolated", WEB_SERVER_SINK_IMAGE); - exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", PREFIX + "test-owner"); + exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", prefix + "test-owner"); String meshPath = Paths.get("target/test-classes/mesh-interpolated.yaml") .toAbsolutePath() @@ -249,33 +257,28 @@ void shouldSucceedWhenInterpolationValuesAreDefined() throws Exception { AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath); try { - Awaitility.await() - .atMost(Duration.ofMinutes(3)) - .pollInterval(Duration.ofSeconds(1)) - .until(() -> handle.getStdout().contains("STREAMX IS READY!")); + awaitStdoutContains(handle, "STREAMX IS READY!"); Thread.sleep(Duration.ofSeconds(5)); - assertThat(handle.thread().isAlive()).isTrue(); + assertThat(handle.isAlive()).isTrue(); - handle.interruptAndJoin(Duration.ofSeconds(30).toMillis()); - assertThat(handle.thread().isAlive()).isFalse(); + handle.stopAndJoin(Duration.ofSeconds(30).toMillis()); + assertThat(handle.isAlive()).isFalse(); ProcessResult result = handle.toResult(); - result.assertSuccess(); - assertThat(result.stdout()).contains("Stopping mesh..."); - assertThat(result.stderr()).doesNotContain("Exception"); + assertThat(result.exitCode()).isIn(0, SIGTERM_EXIT_CODE); + awaitNoContainersWithPrefix(prefix); } finally { - if (handle.thread().isAlive()) { - handle.interruptAndJoin(Duration.ofSeconds(30).toMillis()); + if (handle.isAlive()) { + handle.stopAndJoin(Duration.ofSeconds(30).toMillis()); } } } @Test void shouldStartMeshSecondTimeAfterPreviousStopped() throws Exception { - System.setProperty("streamx.runner.mesh-name-prefix", PREFIX); exec("settings", "set", "config.image.interpolated", WEB_SERVER_SINK_IMAGE); - exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", PREFIX + "test-owner"); + exec("settings", "set", "STREAMX_OWNER_SERVICE_NAME", freshPrefix() + "test-owner"); String meshPath = Paths.get("target/test-classes/mesh-interpolated.yaml") .toAbsolutePath() @@ -286,19 +289,22 @@ void shouldStartMeshSecondTimeAfterPreviousStopped() throws Exception { runUntilReadyThenStop(meshPath); } - private void runUntilReadyThenStop(String meshPath) throws InterruptedException { + private void runUntilReadyThenStop(String meshPath) throws Exception { + System.setProperty("streamx.runner.mesh-name-prefix", freshPrefix()); + System.setProperty("streamx.runner.pulsar.broker-port", + String.valueOf(MeshTestSupport.freePort())); + System.setProperty("streamx.runner.pulsar.http-port", + String.valueOf(MeshTestSupport.freePort())); + System.setProperty("test.proxy.host-port", String.valueOf(MeshTestSupport.freePort())); AsyncProcessHandle handle = execAsync("local", "run", "-f=" + meshPath); try { - Awaitility.await() - .atMost(Duration.ofMinutes(3)) - .pollInterval(Duration.ofSeconds(1)) - .until(() -> handle.getStdout().contains("STREAMX IS READY!")); + awaitStdoutContains(handle, "STREAMX IS READY!"); assertThat(handle.getStderr()) .doesNotContain("MissingReflectionRegistrationError"); } finally { - if (handle.thread().isAlive()) { - handle.interruptAndJoin(Duration.ofSeconds(30).toMillis()); + if (handle.isAlive()) { + handle.stopAndJoin(Duration.ofSeconds(30).toMillis()); } } } diff --git a/src/test/java/com/streamx/cli/test/CliBaseIT.java b/src/test/java/com/streamx/cli/test/CliBaseIT.java index ba49fc2f..ae69b2a0 100644 --- a/src/test/java/com/streamx/cli/test/CliBaseIT.java +++ b/src/test/java/com/streamx/cli/test/CliBaseIT.java @@ -1,5 +1,7 @@ package com.streamx.cli.test; +import static org.assertj.core.api.Assertions.assertThat; + import com.streamx.cli.commands.StreamxCommand; import com.streamx.cli.framework.AbstractCommand; import io.quarkus.arc.Arc; @@ -11,14 +13,25 @@ import java.io.InputStream; import java.io.OutputStream; import java.io.PrintStream; +import java.io.UncheckedIOException; +import java.lang.management.ManagementFactory; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.time.Duration; import java.util.ArrayList; +import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Stream; +import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; @@ -39,6 +52,14 @@ public abstract class CliBaseIT { private Process process; private final Map envVars = new HashMap<>(); + private static final long ASYNC_COMMAND_SHUTDOWN_TIMEOUT_MILLIS = 60_000; + + private static final Path BUILD_OUTPUT_DIR = Path.of("target"); + + private static final List ASYNC_COMMANDS = new ArrayList<>(); + + private static final AtomicInteger ASYNC_COMMAND_COUNTER = new AtomicInteger(); + private static boolean isNative() { return "true".equals(System.getProperty("native.image")); } @@ -56,17 +77,15 @@ static void ensureBuilt() { } protected void setEnv(String key, String value) { - if (isNative()) { - envVars.put(key, value); - } else { + envVars.put(key, value); + if (!isNative()) { System.setProperty(key, value); } } protected void clearEnv(String key) { - if (isNative()) { - envVars.remove(key); - } else { + envVars.remove(key); + if (!isNative()) { System.clearProperty(key); } } @@ -231,9 +250,13 @@ private ProcessResult execSubprocess( return new ProcessResult(process.exitValue(), stdout, stderr); } - private record StreamCapture(Thread thread, ByteArrayOutputStream buffer) { + record StreamCapture(Thread thread, ByteArrayOutputStream buffer) { String join() throws InterruptedException { thread.join(); + return content(); + } + + String content() { return buffer.toString(StandardCharsets.UTF_8); } } @@ -271,81 +294,188 @@ public void assertExitCode(int expected) { } public record AsyncProcessHandle( - Thread thread, - ByteArrayOutputStream stdout, - ByteArrayOutputStream stderr, - AtomicInteger exitCode + Process process, + StreamCapture stdoutCapture, + StreamCapture stderrCapture ) { public String getStdout() { - return stdout.toString(StandardCharsets.UTF_8); + return stdoutCapture.content(); } public String getStderr() { - return stderr.toString(StandardCharsets.UTF_8); + return stderrCapture.content(); + } + + public boolean isAlive() { + return process.isAlive(); + } + + public void stopAndJoin(long timeoutMillis) throws InterruptedException { + process.destroy(); + if (!process.waitFor(timeoutMillis, TimeUnit.MILLISECONDS)) { + process.destroyForcibly(); + process.waitFor(timeoutMillis, TimeUnit.MILLISECONDS); + } + joinCaptures(timeoutMillis); } - public void interruptAndJoin(long timeoutMillis) throws InterruptedException { - thread.interrupt(); - thread.join(timeoutMillis); + public ProcessResult toResult() throws InterruptedException { + if (!process.isAlive()) { + joinCaptures(ASYNC_COMMAND_SHUTDOWN_TIMEOUT_MILLIS); + } + return new ProcessResult( + process.isAlive() ? -1 : process.exitValue(), getStdout(), getStderr()); } - public ProcessResult toResult() { - return new ProcessResult(exitCode.get(), getStdout(), getStderr()); + private void joinCaptures(long timeoutMillis) throws InterruptedException { + stdoutCapture.thread().join(timeoutMillis); + stderrCapture.thread().join(timeoutMillis); } } - protected AsyncProcessHandle execAsync(String... args) { - ByteArrayOutputStream out = new ByteArrayOutputStream(); - ByteArrayOutputStream err = new ByteArrayOutputStream(); - AtomicInteger exitCode = new AtomicInteger(-1); + protected AsyncProcessHandle execAsync(String... args) throws IOException { + awaitAsyncCommands(); - PrintStream originalOut = System.out; - PrintStream originalErr = System.err; + List command = cliLaunchCommand(); + command.addAll(List.of(args)); - PrintStream teeOut = new PrintStream(new TeeOutputStream(out, originalOut), true); - PrintStream teeErr = new PrintStream(new TeeOutputStream(err, originalErr), true); + ProcessBuilder pb = new ProcessBuilder(command); + pb.environment().put("STREAMX_HOME", streamxHome.toAbsolutePath().toString()); + pb.environment().putAll(envVars); + Process process = pb.start(); + + AsyncProcessHandle handle = new AsyncProcessHandle( + process, + captureAndForward(process.getInputStream(), System.out), + captureAndForward(process.getErrorStream(), System.err)); + ASYNC_COMMANDS.add(handle); + return handle; + } - Thread thread = Thread.ofVirtual().start(() -> { - System.setOut(teeOut); - System.setErr(teeErr); - System.setProperty("STREAMX_HOME", streamxHome.toAbsolutePath().toString()); + protected void awaitAsyncCommands() { + for (AsyncProcessHandle handle : ASYNC_COMMANDS) { try { - exitCode.set(createCommandLine().execute(args)); - } finally { - System.clearProperty("STREAMX_HOME"); - System.setOut(originalOut); - System.setErr(originalErr); + handle.stopAndJoin(ASYNC_COMMAND_SHUTDOWN_TIMEOUT_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; } - }); + } + ASYNC_COMMANDS.clear(); + } - return new AsyncProcessHandle(thread, out, err, exitCode); + protected void awaitStdoutContains(AsyncProcessHandle handle, String... expected) { + awaitOutputContains(handle, AsyncProcessHandle::getStdout, expected); } - private static class TeeOutputStream extends OutputStream { - private final OutputStream buffer; - private final OutputStream console; + protected void awaitStderrContains(AsyncProcessHandle handle, String... expected) { + awaitOutputContains(handle, AsyncProcessHandle::getStderr, expected); + } - TeeOutputStream(OutputStream buffer, OutputStream console) { - this.buffer = buffer; - this.console = console; + private static void awaitOutputContains( + AsyncProcessHandle handle, + Function output, + String... expected + ) { + try { + Awaitility.await() + .atMost(Duration.ofMinutes(3)) + .pollInterval(Duration.ofSeconds(1)) + .failFast(() -> { + if (handle.isAlive()) { + return false; + } + try { + handle.joinCaptures(Duration.ofSeconds(5).toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return !containsAll(output.apply(handle), expected); + }) + .untilAsserted(() -> assertThat(output.apply(handle)).contains(expected)); + } catch (RuntimeException e) { + throw new AssertionError( + ("Expected the CLI output to contain %s. CLI process alive: %s" + + "%n--- captured stdout ---%n%s%n--- captured stderr ---%n%s%s") + .formatted(List.of(expected), handle.isAlive(), + handle.getStdout(), handle.getStderr(), errorLogContent(handle)), + e); } + } - @Override - public void write(int b) throws IOException { - buffer.write(b); - console.write(b); + private static String errorLogContent(AsyncProcessHandle handle) { + Matcher matcher = Pattern.compile("Error details saved to: (\\S+)") + .matcher(handle.getStderr()); + if (!matcher.find()) { + return ""; + } + Path errorLog = Path.of(matcher.group(1)); + try { + return "%n--- %s ---%n%s".formatted(errorLog, Files.readString(errorLog)); + } catch (IOException e) { + return "%n--- %s (unreadable: %s) ---".formatted(errorLog, e); } + } - @Override - public void write(byte[] b, int off, int len) throws IOException { - buffer.write(b, off, len); - console.write(b, off, len); + private static boolean containsAll(String output, String... expected) { + for (String part : expected) { + if (!output.contains(part)) { + return false; + } } + return true; + } - @Override - public void flush() throws IOException { - buffer.flush(); - console.flush(); + private static List cliLaunchCommand() { + List command = new ArrayList<>(); + if (isNative()) { + command.addAll(BuildExecutableOnce.getExecutablePath()); + command.addAll(forwardedSystemProperties()); + return command; + } + command.add(ProcessHandle.current().info().command().orElseThrow()); + command.addAll(tracingAgentArguments()); + command.addAll(forwardedSystemProperties()); + command.add("-jar"); + command.add(packagedCliJar().toString()); + return command; + } + + private static Path packagedCliJar() { + try (Stream files = Files.list(BUILD_OUTPUT_DIR)) { + return files + .filter(path -> path.getFileName().toString().endsWith("-runner.jar")) + .max(Comparator.comparing(CliBaseIT::lastModified)) + .orElseThrow(() -> new IllegalStateException( + "Packaged CLI (*-runner.jar) not found in target; run mvn package first")); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static FileTime lastModified(Path path) { + try { + return Files.getLastModifiedTime(path); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static List tracingAgentArguments() { + return ManagementFactory.getRuntimeMXBean().getInputArguments().stream() + .filter(argument -> argument.startsWith("-agentlib:native-image-agent")) + .map(argument -> argument.replace("config-merge-dir=", "config-output-dir=") + + "-async-" + ASYNC_COMMAND_COUNTER.incrementAndGet()) + .toList(); + } + + private static List forwardedSystemProperties() { + List arguments = new ArrayList<>(); + for (String name : System.getProperties().stringPropertyNames()) { + if (name.startsWith("streamx.") || name.startsWith("test.")) { + arguments.add("-D" + name + "=" + System.getProperty(name)); + } } + return arguments; } } diff --git a/src/test/java/com/streamx/cli/test/MeshTestSupport.java b/src/test/java/com/streamx/cli/test/MeshTestSupport.java index aa2b646f..9c1012b3 100644 --- a/src/test/java/com/streamx/cli/test/MeshTestSupport.java +++ b/src/test/java/com/streamx/cli/test/MeshTestSupport.java @@ -5,11 +5,15 @@ import java.io.IOException; import java.io.PrintStream; import java.net.ServerSocket; +import java.nio.file.FileAlreadyExistsException; +import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.util.UUID; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -22,6 +26,19 @@ */ public final class MeshTestSupport { + private static final Path PORT_RESERVATIONS = + Paths.get(System.getProperty("java.io.tmpdir"), "streamx-cli-test-ports"); + + /** Docker allocates its own published ports from 32768 upward, so we allocate below it. */ + private static final int PORT_RANGE_START = 20000; + + private static final int PORT_RANGE_SIZE = 12000; + + private static final AtomicInteger NEXT_PORT_OFFSET = + new AtomicInteger(ThreadLocalRandom.current().nextInt(PORT_RANGE_SIZE)); + + public static final String CONTAINER_STARTUP_TIMEOUT_SECONDS = "180"; + private static volatile MeshManager activeMeshManager; private static volatile int activeProxyPort; private static volatile int activePulsarHttpPort; @@ -31,11 +48,41 @@ public final class MeshTestSupport { private MeshTestSupport() { } + /** + * The port reservation machinery is needed to fix parallel tests flakiness for commands which + * run docker containers like `streamx local run`. + * The naive new ServerSocket(0) doesn't work reliably in this case. + */ public static int freePort() { - try (ServerSocket s = new ServerSocket(0)) { - return s.getLocalPort(); + for (int attempt = 0; attempt < PORT_RANGE_SIZE; attempt++) { + int candidate = PORT_RANGE_START + + Math.floorMod(NEXT_PORT_OFFSET.getAndIncrement(), PORT_RANGE_SIZE); + if (isFree(candidate) && reserve(candidate)) { + return candidate; + } + } + throw new IllegalStateException("Could not reserve a free port"); + } + + private static boolean isFree(int port) { + try (ServerSocket probe = new ServerSocket(port)) { + return probe.getLocalPort() == port; + } catch (IOException alreadyInUse) { + return false; + } + } + + private static boolean reserve(int port) { + try { + Path lock = PORT_RESERVATIONS.resolve(port + ".lock"); + Files.createDirectories(PORT_RESERVATIONS); + Files.createFile(lock); + lock.toFile().deleteOnExit(); + return true; + } catch (FileAlreadyExistsException takenByAnotherJvm) { + return false; } catch (IOException e) { - throw new RuntimeException("Failed to find a free port", e); + throw new RuntimeException("Failed to reserve port " + port, e); } } @@ -62,7 +109,8 @@ public static void startMesh(String meshYamlPath) { "http://localhost:" + activeProxyPort); System.setProperty("test.proxy.host-port", String.valueOf(activeProxyPort)); - System.setProperty("streamx.container.startup-timeout-seconds", "180"); + System.setProperty("streamx.container.startup-timeout-seconds", + CONTAINER_STARTUP_TIMEOUT_SECONDS); capturedToken = null; tokenLatch = new CountDownLatch(1);