fix(connections): use Unis to enable internal connection retry behaviour

src/main/java/io/cryostat/ConfigProperties.java

@@ -20,6 +20,11 @@ public class ConfigProperties {
     public static final String AWS_OBJECT_EXPIRATION_LABELS =
             "storage.buckets.archives.expiration-label";
 
+    public static final String CONNECTIONS_MAX_OPEN = "cryostat.connections.max-open";
+    public static final String CONNECTIONS_TTL = "cryostat.connections.ttl";
+    public static final String CONNECTIONS_FAILED_BACKOFF = "cryostat.connections.failed-backoff";
+    public static final String CONNECTIONS_FAILED_TIMEOUT = "cryostat.connections.failed-timeout";
+
     public static final String REPORTS_SIDECAR_URL = "cryostat.services.reports.url";
     public static final String MEMORY_CACHE_ENABLED =
             "cryostat.services.reports.memory-cache.enabled";

src/main/java/io/cryostat/ExceptionMappers.java

@@ -20,6 +20,7 @@
 import io.cryostat.targets.TargetConnectionManager;
 
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.smallrye.mutiny.TimeoutException;
 import jakarta.inject.Inject;
 import jakarta.persistence.NoResultException;
 import org.hibernate.exception.ConstraintViolationException;
@@ -83,4 +84,10 @@ public RestResponse<Void> mapFlightRecorderException(
         }
         return RestResponse.status(HttpResponseStatus.BAD_GATEWAY.code());
     }
+
+    @ServerExceptionMapper
+    public RestResponse<Void> mapMutinyTimeoutException(TimeoutException ex) {
+        logger.warn(ex);
+        return RestResponse.status(HttpResponseStatus.GATEWAY_TIMEOUT.code());
+    }
 }

src/main/java/io/cryostat/discovery/CustomDiscovery.java

@@ -18,6 +18,7 @@
 import java.net.MalformedURLException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Map;
 import java.util.Optional;
@@ -133,8 +134,13 @@ Response doV2Create(
 
             try {
                 target.jvmId =
-                        connectionManager.executeDirect(
-                                target, credential, conn -> conn.getJvmId());
+                        connectionManager
+                                .executeDirect(
+                                        target,
+                                        credential,
+                                        conn -> conn.getJvmIdentifier().getHash())
+                                .await()
+                                .atMost(Duration.ofSeconds(10));
             } catch (Exception e) {
                 logger.error("Target connection failed", e);
                 return Response.status(Response.Status.BAD_REQUEST.getStatusCode())

src/main/java/io/cryostat/recordings/RecordingHelper.java

@@ -25,6 +25,7 @@
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
+import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.temporal.ChronoUnit;
@@ -34,8 +35,6 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -70,7 +69,6 @@
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
-import jakarta.persistence.EntityManager;
 import jakarta.ws.rs.BadRequestException;
 import jakarta.ws.rs.ServerErrorException;
 import jakarta.ws.rs.core.Response;
@@ -99,7 +97,6 @@
 import software.amazon.awssdk.services.s3.model.Tag;
 import software.amazon.awssdk.services.s3.model.Tagging;
 import software.amazon.awssdk.services.s3.model.UploadPartRequest;
-import software.amazon.awssdk.services.s3.presigner.S3Presigner;
 
 @ApplicationScoped
 public class RecordingHelper {
@@ -110,18 +107,13 @@ public class RecordingHelper {
             Pattern.compile("^template=([\\w]+)(?:,type=([\\w]+))?$");
     public static final String DATASOURCE_FILENAME = "cryostat-analysis.jfr";
 
-    private static final long httpTimeoutSeconds = 5; // TODO: configurable client timeout
-
     @Inject Logger logger;
-    @Inject EntityManager entityManager;
     @Inject TargetConnectionManager connectionManager;
     @Inject RecordingOptionsBuilderFactory recordingOptionsBuilderFactory;
     @Inject EventOptionsBuilder.Factory eventOptionsBuilderFactory;
-    @Inject ScheduledExecutorService scheduler;
     @Inject EventBus bus;
 
     @Inject Clock clock;
-    @Inject S3Presigner presigner;
 
     @Inject
     @Named(Producers.BASE64_URL)
@@ -139,6 +131,9 @@ public class RecordingHelper {
     @ConfigProperty(name = ConfigProperties.AWS_OBJECT_EXPIRATION_LABELS)
     String objectExpirationLabel;
 
+    @ConfigProperty(name = ConfigProperties.CONNECTIONS_FAILED_TIMEOUT)
+    Duration connectionFailedTimeout;
+
     public ActiveRecording startRecording(
             Target target,
             IConstrainedMap<String> recordingOptions,
@@ -787,7 +782,7 @@ public Response uploadToJFRDatasource(long targetEntityId, long remoteId, URL up
                     webClient
                             .postAbs(uploadUrl.toURI().resolve("/load").normalize().toString())
                             .addQueryParam("overwrite", "true")
-                            .timeout(TimeUnit.SECONDS.toMillis(httpTimeoutSeconds))
+                            .timeout(connectionFailedTimeout.toMillis())
                             .sendMultipartForm(form);
             return asyncRequest
                     .onItem()

src/main/java/io/cryostat/recordings/Recordings.java

@@ -64,6 +64,7 @@
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.quarkus.runtime.StartupEvent;
 import io.smallrye.common.annotation.Blocking;
+import io.smallrye.mutiny.Uni;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.handler.HttpException;
 import io.vertx.mutiny.core.eventbus.EventBus;
@@ -507,65 +508,63 @@ public Response patchV1(@RestPath URI connectUrl, @RestPath String recordingName
 
     @POST
     @Transactional
-    @Blocking
     @Path("/api/v1/targets/{connectUrl}/snapshot")
     @RolesAllowed("write")
-    public Response createSnapshotV1(@RestPath URI connectUrl) throws Exception {
+    public Uni<Response> createSnapshotV1(@RestPath URI connectUrl) throws Exception {
         Target target = Target.getTargetByConnectUrl(connectUrl);
-        try {
-            ActiveRecording recording =
-                    connectionManager.executeConnectedTask(
-                            target,
-                            connection -> recordingHelper.createSnapshot(target, connection));
-            return Response.status(Response.Status.OK).entity(recording.name).build();
-        } catch (SnapshotCreationException sce) {
-            return Response.status(Response.Status.ACCEPTED).build();
-        }
+        return connectionManager
+                .executeConnectedTaskUni(
+                        target, connection -> recordingHelper.createSnapshot(target, connection))
+                .onItem()
+                .transform(
+                        recording ->
+                                Response.status(Response.Status.OK).entity(recording.name).build())
+                .onFailure(SnapshotCreationException.class)
+                .recoverWithItem(Response.status(Response.Status.ACCEPTED).build());
     }
 
     @POST
     @Transactional
-    @Blocking
     @Path("/api/v2/targets/{connectUrl}/snapshot")
     @RolesAllowed("write")
-    public Response createSnapshotV2(@RestPath URI connectUrl) throws Exception {
+    public Uni<Response> createSnapshotV2(@RestPath URI connectUrl) throws Exception {
         Target target = Target.getTargetByConnectUrl(connectUrl);
-        try {
-            ActiveRecording recording =
-                    connectionManager.executeConnectedTask(
-                            target,
-                            connection -> recordingHelper.createSnapshot(target, connection));
-            return Response.status(Response.Status.CREATED)
-                    .entity(
-                            V2Response.json(
-                                    Response.Status.CREATED,
-                                    recordingHelper.toExternalForm(recording)))
-                    .build();
-        } catch (SnapshotCreationException sce) {
-            return Response.status(Response.Status.ACCEPTED)
-                    .entity(V2Response.json(Response.Status.ACCEPTED, null))
-                    .build();
-        }
+        return connectionManager
+                .executeConnectedTaskUni(
+                        target, connection -> recordingHelper.createSnapshot(target, connection))
+                .onItem()
+                .transform(
+                        recording ->
+                                Response.status(Response.Status.CREATED)
+                                        .entity(
+                                                V2Response.json(
+                                                        Response.Status.CREATED,
+                                                        recordingHelper.toExternalForm(recording)))
+                                        .build())
+                .onFailure(SnapshotCreationException.class)
+                .recoverWithItem(
+                        Response.status(Response.Status.ACCEPTED)
+                                .entity(V2Response.json(Response.Status.ACCEPTED, null))
+                                .build());
     }
 
     @POST
     @Transactional
-    @Blocking
     @Path("/api/v3/targets/{id}/snapshot")
     @RolesAllowed("write")
-    public Response createSnapshot(@RestPath long id) throws Exception {
+    public Uni<Response> createSnapshot(@RestPath long id) throws Exception {
         Target target = Target.find("id", id).singleResult();
-        try {
-            ActiveRecording recording =
-                    connectionManager.executeConnectedTask(
-                            target,
-                            connection -> recordingHelper.createSnapshot(target, connection));
-            return Response.status(Response.Status.OK)
-                    .entity(recordingHelper.toExternalForm(recording))
-                    .build();
-        } catch (SnapshotCreationException sce) {
-            return Response.status(Response.Status.ACCEPTED).build();
-        }
+        return connectionManager
+                .executeConnectedTaskUni(
+                        target, connection -> recordingHelper.createSnapshot(target, connection))
+                .onItem()
+                .transform(
+                        recording ->
+                                Response.status(Response.Status.OK)
+                                        .entity(recordingHelper.toExternalForm(recording))
+                                        .build())
+                .onFailure(SnapshotCreationException.class)
+                .recoverWithItem(Response.status(Response.Status.ACCEPTED).build());
     }
 
     @POST

src/main/java/io/cryostat/targets/AgentConnection.java

@@ -164,6 +164,7 @@ public MBeanMetrics getMBeanMetrics()
         MemoryMetrics memory = new MemoryMetrics(Map.of());
         ThreadMetrics thread = new ThreadMetrics(Map.of());
         OperatingSystemMetrics operatingSystem = new OperatingSystemMetrics(Map.of());
-        return new MBeanMetrics(runtime, memory, thread, operatingSystem, getJvmId());
+        return new MBeanMetrics(
+                runtime, memory, thread, operatingSystem, getJvmIdentifier().getHash());
     }
 }

src/main/java/io/cryostat/targets/Target.java

@@ -18,6 +18,7 @@
 import java.net.URI;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -228,7 +229,13 @@ void prePersist(Target target) throws JvmIdException {
             }
             try {
                 target.jvmId =
-                        connectionManager.executeConnectedTask(target, conn -> conn.getJvmId());
+                        connectionManager
+                                .executeDirect(
+                                        target,
+                                        Optional.empty(),
+                                        conn -> conn.getJvmIdentifier().getHash())
+                                .await()
+                                .atMost(Duration.ofSeconds(10));
             } catch (Exception e) {
                 // TODO tolerate this in the condition that the connection failed because of JMX
                 // auth. In that instance then persist the entity with a null jvmId, but listen for