Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion docs/user-manual/modules/ROOT/pages/camel-jbang-mcp.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -812,7 +812,8 @@ help validate and scaffold that workflow, following the rules of the `rest-opena
NOTE: Runtime tools require a running Camel application started via `camel run`. They communicate with the
application through the file-based IPC protocol in `~/.camel/`. All tools accept an optional `nameOrPid`
parameter; when omitted, the server auto-discovers the running Camel process (this works when exactly one
process is running).
process is running). The MCP server itself is never one of them: it is not listed, not selected when no name is
given, and a call that names its pid is refused.

==== Process Discovery

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -62,13 +62,16 @@ public class FileCliConnectorTransport extends ServiceSupport implements CliConn
private CliSnapshotProducer snapshots;
private Runnable shutdown;
private ScheduledExecutorService executor;
private volatile Thread pollThread;
private ScheduledFuture<?> scheduledFuture;
private int delay = 1000;
private long counter;
private final Map<File, PendingAction> pending = new HashMap<>();

// an action file this young that cannot be read may still be written: the Camel CLI before 4.24 writes it in place
private static final long UNREADABLE_GRACE_MS = 2000;
// how long a stop waits for the poll in flight before it removes the files
private static final long STOP_POLL_TIMEOUT_MS = 5000;

private record PendingAction(File output, boolean replyAlways, CompletableFuture<CliActionOutcome> outcome) {
}
Expand Down Expand Up @@ -108,6 +111,7 @@ protected void doStart() throws Exception {
String threadName = ThreadHelper.resolveThreadName(null, "LocalCliConnector");
Thread thread = new Thread(r, threadName);
thread.setDaemon(true);
pollThread = thread;
return thread;
});

Expand Down Expand Up @@ -436,6 +440,7 @@ protected void doStop() throws Exception {
if (scheduledFuture != null) {
scheduledFuture.cancel(false);
}
awaitPoll();
synchronized (this) {
pending.forEach((request, action) -> {
action.outcome().cancel(true);
Expand Down Expand Up @@ -480,6 +485,29 @@ protected void doStop() throws Exception {
}
}

/**
* Waits a little for the poll in flight: a status written after the files are removed would stay behind, and the
* Camel CLI would list this process as a running integration after its Camel has stopped (CAMEL-26396). Not when
* the poll itself stops the transport, which would wait for itself.
*/
private void awaitPoll() {
if (executor == null) {
return;
}
// the poll is cancelled, so the thread ends once the poll in flight is done
executor.shutdown();
if (Thread.currentThread() == pollThread) {
return;
}
try {
if (!executor.awaitTermination(STOP_POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
LOG.debug("The poll of the Camel CLI connector has not finished within {} millis", STOP_POLL_TIMEOUT_MS);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}

private static String getPid() {
return String.valueOf(ProcessHandle.current().pid());
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.camel.cli.connector;

import java.io.File;
import java.nio.file.Path;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.util.json.JsonObject;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.api.parallel.Isolated;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

/**
* CAMEL-26396: a stop waits for the status write in flight, so a stopped integration leaves no status file behind. The
* Camel CLI lists a live process with a status file as a running integration: the MCP server, whose scratch contexts
* ran with the connector, listed itself and took itself for the integration to evaluate in.
* <p>
* Isolated because it changes {@code user.home}, and every connector in this JVM writes files named after the same pid.
*/
@Isolated
class FileCliConnectorTransportStopTest {

private static final String PID = String.valueOf(ProcessHandle.current().pid());

@TempDir
Path home;

private String oldHome;
private DefaultCamelContext context;

@BeforeEach
void useHome() {
oldHome = System.getProperty("user.home");
System.setProperty("user.home", home.toString());
context = new DefaultCamelContext();
}

@AfterEach
void restoreHome() {
context.stop();
System.setProperty("user.home", oldHome);
}

@Test
void aStatusWriteInFlightDoesNotOutliveTheStop() throws Exception {
SlowSnapshots snapshots = new SlowSnapshots();
FileCliConnectorTransport transport = new FileCliConnectorTransport();
transport.configure(context, (action, output) -> false, snapshots, () -> {
});
transport.start();
File status = home.resolve(".camel").resolve(PID + "-status.json").toFile();
// the first poll collects the status, which takes until the test lets it go on
assertThat(snapshots.collecting.await(10, TimeUnit.SECONDS)).isTrue();

Thread stopper = new Thread(transport::stop, "test-stop");
stopper.start();
// the stop is done, as it was at once before the fix, or it waits for the poll
await().atMost(10, TimeUnit.SECONDS).until(() -> switch (stopper.getState()) {
case TERMINATED, WAITING, TIMED_WAITING -> true;
default -> false;
});
snapshots.collected.countDown();

stopper.join(10_000);
assertThat(stopper.isAlive()).isFalse();
// the poll has written the status, before the stop removed the files or after
snapshots.poll.join(10_000);
assertThat(snapshots.poll.isAlive()).isFalse();
assertThat(status).doesNotExist();
}

/**
* Snapshots whose status takes until the test counts down {@link #collected}.
*/
private static final class SlowSnapshots implements CliSnapshotProducer {

final CountDownLatch collecting = new CountDownLatch(1);
final CountDownLatch collected = new CountDownLatch(1);
volatile Thread poll;

@Override
public JsonObject status() throws Exception {
poll = Thread.currentThread();
collecting.countDown();
collected.await(10, TimeUnit.SECONDS);
JsonObject context = new JsonObject();
context.put("name", "demo");
context.put("phase", 4);
JsonObject root = new JsonObject();
root.put("context", context);
return root;
}

@Override
public JsonObject trace() {
return null;
}

@Override
public JsonObject debug() {
return null;
}

@Override
public JsonObject messageHistory() {
return null;
}

@Override
public JsonObject errors() {
return null;
}

@Override
public JsonObject receive() {
return null;
}

@Override
public JsonObject activity() {
return null;
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.camel.Expression;
import org.apache.camel.Predicate;
import org.apache.camel.catalog.LanguageValidationResult;
import org.apache.camel.dsl.jbang.core.common.ScratchContextHelper;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.main.download.DependencyDownloaderClassLoader;
import org.apache.camel.main.download.MavenDependencyDownloader;
Expand Down Expand Up @@ -193,7 +194,8 @@ private static void evaluateLocally(
*/
private static boolean evaluateWith(
ClassLoader loader, String lang, String expression, String body, boolean predicate, JsonObject result) {
try (DefaultCamelContext context = new DefaultCamelContext(false)) {
// without the CLI connector, which would make this process, such as the MCP server, a running integration
try (DefaultCamelContext context = ScratchContextHelper.withoutCliConnector(new DefaultCamelContext(false))) {
if (loader != null) {
// before the start: the language resolver reads the class loader as the context comes up
context.setApplicationContextClassLoader(loader);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -570,6 +570,10 @@ private static String readTail(Path output) {
public static String control(ToolContext ctx, String action) {
ctx.requireProcess();
long pid = ctx.pid();
if (pid == ProcessHandle.current().pid()) {
// the tools run in this process: it cannot kill itself (the JDK refuses), nor answer an action it waits for
throw ToolContext.thisProcess(pid);
}
return switch (action) {
case "stop" -> ctx.stopApplication();
case "kill" -> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -222,12 +222,13 @@ private CamelCatalog loadCatalog() {
* Selects the process a tool call names ({@code name} is an integration name or pid), or keeps the current
* selection when the call names none.
*
* @throws ToolExecutionException when the name matches no running process
* @throws ToolExecutionException when the name matches no running process, or is the pid of this process
*/
public void selectProcess(String nameOrPid) {
if (nameOrPid == null || nameOrPid.isBlank()) {
return;
}
refuseThisProcess(nameOrPid);
RuntimeHelper.ProcessInfo p = RuntimeHelper.findProcess(nameOrPid);
if (p == null) {
throw new ToolExecutionException(
Expand All @@ -237,6 +238,27 @@ public void selectProcess(String nameOrPid) {
selectProcess(p.pid());
}

/**
* Refuses the pid of this process, which runs the tools: a status file of its own comes from a CamelContext it ran
* for itself, never from an integration, and it would wait for itself to answer an action (CAMEL-26396).
*
* @throws ToolExecutionException when the name or pid is the pid of this process
*/
static void refuseThisProcess(String nameOrPid) {
if (RuntimeHelper.isThisProcess(nameOrPid)) {
throw thisProcess(ProcessHandle.current().pid());
}
}

/**
* The error for a tool call on the pid of this process (see {@link #refuseThisProcess(String)}).
*/
static ToolExecutionException thisProcess(long pid) {
return new ToolExecutionException(
"pid " + pid + " is the process that runs these tools (the MCP server, camel ask or the TUI), not a"
+ " Camel integration; list_processes shows the integrations");
}

public String readStatus(String section) {
requireProcess();
return RuntimeHelper.readStatusSection(pid, section);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,7 @@ private static void registerProcessTools() {
if (name == null || name.isBlank()) {
throw new ToolExecutionException("name or PID is required");
}
ToolContext.refuseThisProcess(name);
RuntimeHelper.ProcessInfo found = ctx.findProcess(name);
if (found == null) {
List<RuntimeHelper.ProcessInfo> processes = ctx.discoverProcesses();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -60,10 +60,12 @@
import org.apache.camel.dsl.jbang.core.common.RuntimeCompletionCandidates;
import org.apache.camel.dsl.jbang.core.common.RuntimeType;
import org.apache.camel.dsl.jbang.core.common.RuntimeTypeConverter;
import org.apache.camel.dsl.jbang.core.common.ScratchContextHelper;
import org.apache.camel.dsl.jbang.core.common.VersionHelper;
import org.apache.camel.dsl.jbang.core.model.VersionListDTO;
import org.apache.camel.main.KameletMain;
import org.apache.camel.main.download.MavenDependencyDownloader;
import org.apache.camel.spi.CliConnectorFactory;
import org.apache.camel.tooling.maven.RepositoryResolver;
import org.apache.camel.tooling.model.ReleaseModel;
import org.apache.camel.util.StringHelper;
Expand Down Expand Up @@ -400,6 +402,9 @@ Stream<CamelAndRuntimeVersions> downloadReleases() {
mavenResolver.fresh());
}
KameletMain main = new KameletMain(CAMEL_INSTANCE_TYPE);
// only to download: without the CLI connector, which would list this process (such as the MCP server, whose
// camel_version_list runs this) as a running integration
main.bind(CliConnectorFactory.FACTORY, ScratchContextHelper.disabledCliConnectorFactory());
try {
main.setFresh(mavenResolver.fresh());
main.setDownload(mavenResolver.download());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -61,10 +61,13 @@ private RuntimeHelper() {
/**
* Discovers all running Camel processes by scanning {@code ~/.camel/} for {@code {pid}-status.json} files and
* verifying each PID is still alive.
* <p>
* This process is never one of them: it runs the tools (camel ask, the TUI, the MCP server), and a status file of
* its own comes from a CamelContext it ran for itself, not from an integration (CAMEL-26396).
*/
public static List<ProcessInfo> discoverProcesses() {
List<ProcessInfo> result = new ArrayList<>();
for (ProcessFinder.CamelProcess p : ProcessFinder.findAll()) {
for (ProcessFinder.CamelProcess p : others(ProcessFinder.findAll())) {
result.add(new ProcessInfo(p.pid(), p.name(), p.contextName()));
}
return result;
Expand All @@ -73,20 +76,37 @@ public static List<ProcessInfo> discoverProcesses() {
/**
* Finds a single Camel process matching the given name or PID. Returns {@code null} if no match is found or if
* multiple processes match. When {@code nameOrPid} is {@code null} and exactly one process is running, returns that
* process.
* process. This process never matches, also not by its pid (see {@link #discoverProcesses()}).
*
* @param nameOrPid a process name (with optional wildcard), a numeric PID string, or {@code null} for auto-detect
*/
public static ProcessInfo findProcess(String nameOrPid) {
List<ProcessFinder.CamelProcess> found = nameOrPid != null && !nameOrPid.isBlank()
? ProcessFinder.find(nameOrPid) : ProcessFinder.findAll();
List<ProcessFinder.CamelProcess> found = others(nameOrPid != null && !nameOrPid.isBlank()
? ProcessFinder.find(nameOrPid) : ProcessFinder.findAll());
if (found.size() == 1) {
ProcessFinder.CamelProcess p = found.get(0);
return new ProcessInfo(p.pid(), p.name(), p.contextName());
}
return null;
}

/**
* Whether the name or pid is the pid of this process, which {@link #findProcess(String)} never finds.
*
* @since 4.24
*/
public static boolean isThisProcess(String nameOrPid) {
if (nameOrPid == null || !nameOrPid.matches("\\d{1,18}")) {
return false;
}
return Long.parseLong(nameOrPid) == ProcessHandle.current().pid();
}

private static List<ProcessFinder.CamelProcess> others(List<ProcessFinder.CamelProcess> processes) {
long self = ProcessHandle.current().pid();
return processes.stream().filter(p -> p.pid() != self).toList();
}

/**
* Reads the full status JSON for the given process.
*
Expand Down
Loading
Loading