cryostatio · andrewazores · Jan 31, 2025 · Jan 31, 2025 · Jan 31, 2025 · Jan 31, 2025
diff --git a/src/main/java/io/cryostat/ConfigProperties.java b/src/main/java/io/cryostat/ConfigProperties.java
@@ -32,6 +32,7 @@ public class ConfigProperties {
     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 CONNECTIONS_UPLOAD_TIMEOUT = "cryostat.connections.upload-timeout";
 
     public static final String REPORTS_SIDECAR_URL = "quarkus.rest-client.reports.url";
     public static final String REPORTS_MEMORY_CACHE_ENABLED =

diff --git a/src/main/java/io/cryostat/recordings/ActiveRecordingsDownload.java b/src/main/java/io/cryostat/recordings/ActiveRecordingsDownload.java
@@ -49,6 +49,9 @@ public class ActiveRecordingsDownload {
     @Named(Producers.BASE64_URL)
     Base64 base64Url;
 
+    @ConfigProperty(name = ConfigProperties.CONNECTIONS_FAILED_TIMEOUT)
+    Duration connectionFailedTimeout;
+
     @ConfigProperty(name = ConfigProperties.STORAGE_TRANSIENT_ARCHIVES_ENABLED)
     boolean transientArchivesEnabled;
 
@@ -66,7 +69,9 @@ public RestResponse<InputStream> handleActiveDownload(@RestPath long id) throws
                             HttpHeaders.CONTENT_DISPOSITION,
                             String.format("attachment; filename=\"%s.jfr\"", recording.name))
                     .header(HttpHeaders.CONTENT_TYPE, HttpMimeType.OCTET_STREAM.mime())
-                    .entity(recordingHelper.getActiveInputStream(recording))
+                    .entity(
+                            recordingHelper.getActiveInputStream(
+                                    recording, connectionFailedTimeout))
                     .build();
         }
 

diff --git a/src/main/java/io/cryostat/recordings/LongRunningRequestGenerator.java b/src/main/java/io/cryostat/recordings/LongRunningRequestGenerator.java
@@ -62,6 +62,9 @@ public class LongRunningRequestGenerator {
     @ConfigProperty(name = ConfigProperties.CONNECTIONS_FAILED_TIMEOUT)
     Duration timeout;
 
+    @ConfigProperty(name = ConfigProperties.CONNECTIONS_UPLOAD_TIMEOUT)
+    Duration uploadFailedTimeout;
+
     public LongRunningRequestGenerator() {}
 
     @ConsumeEvent(value = ARCHIVE_ADDRESS, blocking = true)
@@ -88,7 +91,10 @@ public void onMessage(ArchiveRequest request) {
     public void onMessage(GrafanaArchiveUploadRequest request) {
         try {
             logger.trace("Job ID: " + request.getId() + " submitted.");
-            recordingHelper.uploadToJFRDatasource(request.getPair()).await().atMost(timeout);
+            recordingHelper
+                    .uploadToJFRDatasource(request.getPair())
+                    .await()
+                    .atMost(uploadFailedTimeout);
             logger.trace("Grafana upload complete, firing notification");
             bus.publish(
                     MessagingServer.class.getName(),
@@ -108,7 +114,7 @@ public void onMessage(GrafanaActiveUploadRequest request) {
             recordingHelper
                     .uploadToJFRDatasource(request.getTargetId(), request.getRemoteId())
                     .await()
-                    .atMost(timeout);
+                    .atMost(uploadFailedTimeout);
             logger.trace("Grafana upload complete, firing notification");
             bus.publish(
                     MessagingServer.class.getName(),

diff --git a/src/main/java/io/cryostat/recordings/RecordingHelper.java b/src/main/java/io/cryostat/recordings/RecordingHelper.java
@@ -176,6 +176,9 @@ public class RecordingHelper {
     @ConfigProperty(name = ConfigProperties.CONNECTIONS_FAILED_TIMEOUT)
     Duration connectionFailedTimeout;
 
+    @ConfigProperty(name = ConfigProperties.CONNECTIONS_UPLOAD_TIMEOUT)
+    Duration uploadFailedTimeout;
+
     @ConfigProperty(name = ConfigProperties.GRAFANA_DATASOURCE_URL)
     Optional<String> grafanaDatasourceURLProperty;
 
@@ -475,7 +478,7 @@ public Uni<ActiveRecording> createSnapshot(Target target) {
                     desc = updatedDescriptor.get();
 
                     try (InputStream snapshot =
-                            remoteRecordingStreamFactory.open(connection, target, desc)) {
+                            remoteRecordingStreamFactory.openDirect(connection, target, desc)) {
                         if (!snapshotIsReadable(target, snapshot)) {
                             safeCloseRecording(connection, desc);
                             throw new SnapshotCreationException(
@@ -822,7 +825,7 @@ public ArchivedRecording archiveRecording(
         String multipartId = null;
         List<Pair<Integer, String>> parts = new ArrayList<>();
         long accum = 0;
-        try (var stream = getActiveInputStream(recording);
+        try (var stream = getActiveInputStream(recording, uploadFailedTimeout);
                 var ch = Channels.newChannel(stream)) {
             ByteBuffer buf = ByteBuffer.allocate(20 * MIB);
             CreateMultipartUploadRequest.Builder builder =
@@ -996,14 +999,16 @@ public Pair<String, String> decodedKey(String encodedKey) {
         return Pair.of(parts[0], parts[1]);
     }
 
-    public InputStream getActiveInputStream(ActiveRecording recording) throws Exception {
-        return remoteRecordingStreamFactory.open(recording);
+    public InputStream getActiveInputStream(ActiveRecording recording, Duration timeout)
+            throws Exception {
+        return getActiveInputStream(recording.target.id, recording.remoteId, timeout);
     }
 
-    public InputStream getActiveInputStream(long targetId, long remoteId) throws Exception {
+    public InputStream getActiveInputStream(long targetId, long remoteId, Duration timeout)
+            throws Exception {
         var target = Target.getTargetById(targetId);
         var recording = target.getRecordingById(remoteId);
-        var stream = remoteRecordingStreamFactory.open(recording);
+        var stream = remoteRecordingStreamFactory.open(recording, timeout);
         return stream;
     }
 
@@ -1323,7 +1328,7 @@ Optional<Path> getRecordingCopyPath(
                             try {
                                 Path tempFile = fs.createTempFile(null, null);
                                 try (var stream =
-                                        remoteRecordingStreamFactory.open(
+                                        remoteRecordingStreamFactory.openDirect(
                                                 connection, target, descriptor)) {
                                     fs.copy(stream, tempFile, StandardCopyOption.REPLACE_EXISTING);
                                 }

diff --git a/src/main/java/io/cryostat/recordings/RemoteRecordingInputStreamFactory.java b/src/main/java/io/cryostat/recordings/RemoteRecordingInputStreamFactory.java
@@ -16,6 +16,7 @@
 package io.cryostat.recordings;
 
 import java.io.InputStream;
+import java.time.Duration;
 
 import org.openjdk.jmc.flightrecorder.configuration.IRecordingDescriptor;
 
@@ -33,18 +34,19 @@ public class RemoteRecordingInputStreamFactory {
     @Inject TargetConnectionManager connectionManager;
     @Inject RecordingHelper recordingHelper;
 
-    public ProgressInputStream open(ActiveRecording recording) throws Exception {
+    public ProgressInputStream open(ActiveRecording recording, Duration timeout) throws Exception {
         return connectionManager.executeConnectedTask(
                 recording.target,
                 conn -> {
                     IRecordingDescriptor desc =
                             recordingHelper.getDescriptor(conn, recording).orElseThrow();
-                    return open(conn, recording.target, desc);
-                });
+                    return openDirect(conn, recording.target, desc);
+                },
+                timeout);
     }
 
-    public ProgressInputStream open(JFRConnection conn, Target target, IRecordingDescriptor desc)
-            throws Exception {
+    public ProgressInputStream openDirect(
+            JFRConnection conn, Target target, IRecordingDescriptor desc) throws Exception {
         InputStream bareStream = conn.getService().openStream(desc, false);
         return new ProgressInputStream(
                 bareStream, n -> connectionManager.markConnectionInUse(target));

diff --git a/src/main/java/io/cryostat/reports/ReportsServiceImpl.java b/src/main/java/io/cryostat/reports/ReportsServiceImpl.java
@@ -17,6 +17,7 @@
 
 import java.io.BufferedInputStream;
 import java.io.InputStream;
+import java.time.Duration;
 import java.util.Map;
 import java.util.function.Predicate;
 
@@ -41,6 +42,9 @@ class ReportsServiceImpl implements ReportsService {
 
     private static final String NO_SIDECAR_URL = "http://localhost/";
 
+    @ConfigProperty(name = ConfigProperties.CONNECTIONS_UPLOAD_TIMEOUT)
+    Duration uploadFailedTimeout;
+
     @ConfigProperty(name = ConfigProperties.REPORTS_SIDECAR_URL)
     String sidecarUri;
 
@@ -55,7 +59,7 @@ public Uni<Map<String, AnalysisResult>> reportFor(
             ActiveRecording recording, Predicate<IRule> predicate) {
         InputStream stream;
         try {
-            stream = helper.getActiveInputStream(recording);
+            stream = helper.getActiveInputStream(recording, uploadFailedTimeout);
         } catch (Exception e) {
             throw new ReportGenerationException(e);
         }

diff --git a/src/main/java/io/cryostat/targets/TargetConnectionManager.java b/src/main/java/io/cryostat/targets/TargetConnectionManager.java
@@ -204,7 +204,11 @@ public <T> Uni<T> executeConnectedTaskUni(Target target, ConnectedTask<T> task)
     }
 
     public <T> T executeConnectedTask(Target target, ConnectedTask<T> task) {
-        return executeConnectedTaskUni(target, task).await().atMost(failedTimeout);
+        return executeConnectedTask(target, task, failedTimeout);
+    }
+
+    public <T> T executeConnectedTask(Target target, ConnectedTask<T> task, Duration timeout) {
+        return executeConnectedTaskUni(target, task).await().atMost(timeout);
     }
 
     public <T> Uni<T> executeDirect(

diff --git a/src/main/resources/application-test.properties b/src/main/resources/application-test.properties
@@ -6,6 +6,7 @@ cryostat.discovery.docker.enabled=true
 cryostat.discovery.kubernetes.enabled=true
 
 quarkus.test.env.JAVA_OPTS_APPEND=-Dquarkus.http.host=0.0.0.0 -Djava.util.logging.manager=org.jboss.logmanager.LogManager -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=9091 -Dcom.sun.management.jmxremote.rmi.port=9091 -Djava.rmi.server.hostname=127.0.0.1 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.local.only=false
+quarkus.http.test-timeout=60s
 
 quarkus.datasource.devservices.enabled=true
 quarkus.datasource.devservices.image-name=quay.io/cryostat/cryostat-db

diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
@@ -28,7 +28,8 @@ quarkus.test.integration-test-profile=test
 cryostat.connections.max-open=0
 cryostat.connections.ttl=10s
 cryostat.connections.failed-backoff=2s
-cryostat.connections.failed-timeout=10s
+cryostat.connections.failed-timeout=30s
+cryostat.connections.upload-timeout=30m
 quarkus.rest-client.reports.url=http://localhost/
 quarkus.rest-client.reports.verify-host=true
 quarkus.cache.enabled=true