You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by di...@apache.org on 2021/09/01 23:32:46 UTC
[airavata-mft] branch develop updated: Fixing stale connection
issue on unclosed resource and secret service clients
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
The following commit(s) were added to refs/heads/develop by this push:
new e9fd099 Fixing stale connection issue on unclosed resource and secret service clients
e9fd099 is described below
commit e9fd099a3e666ad139039ac9680649cf0b4968c9
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Wed Sep 1 19:32:38 2021 -0400
Fixing stale connection issue on unclosed resource and secret service clients
---
.../apache/airavata/mft/agent/rpc/RPCParser.java | 2 +-
.../mft/transport/scp/SCPMetadataCollector.java | 113 +++++++++++++--------
.../airavata/mft/transport/scp/SCPReceiver.java | 19 ++--
.../airavata/mft/transport/scp/SCPSender.java | 19 ++--
4 files changed, 96 insertions(+), 57 deletions(-)
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/rpc/RPCParser.java b/agent/src/main/java/org/apache/airavata/mft/agent/rpc/RPCParser.java
index e83b0f3..f5c47e4 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/rpc/RPCParser.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/rpc/RPCParser.java
@@ -62,7 +62,7 @@ public class RPCParser {
public String resolveRPCRequest(SyncRPCRequest request) throws Exception {
// TODO implement using the reflection
ObjectMapper mapper = new ObjectMapper();
- logger.info("Accepting sync request {}", request.getRequestId());
+ logger.info("Accepting sync request {} for method {}", request.getRequestId(), request.getMethod());
switch (request.getMethod()) {
case "getFileResourceMetadata":
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPMetadataCollector.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPMetadataCollector.java
index 1468479..9a12990 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPMetadataCollector.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPMetadataCollector.java
@@ -120,27 +120,37 @@ public class SCPMetadataCollector implements MetadataCollector {
public FileResourceMetadata getFileResourceMetadata(AuthToken authZToken, String resourceId, String credentialToken) throws Exception {
checkInitialized();
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- GenericResource scpResource = resourceClient.get()
- .getGenericResource(GenericResourceGetRequest.newBuilder()
- .setAuthzToken(authZToken).setResourceId(resourceId).build());
- SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder()
- .setAuthzToken(authZToken).setSecretId(credentialToken).build());
+ GenericResource scpResource;
+ SCPSecret scpSecret;
+ try (ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort)) {
+ scpResource = resourceClient.get()
+ .getGenericResource(GenericResourceGetRequest.newBuilder()
+ .setAuthzToken(authZToken).setResourceId(resourceId).build());
+ }
+
+ try (SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort)) {
+ scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder()
+ .setAuthzToken(authZToken).setSecretId(credentialToken).build());
+ }
return getFileResourceMetadata(authZToken,scpResource, scpSecret);
}
@Override
public FileResourceMetadata getFileResourceMetadata(AuthToken authZToken, String parentResourceId, String childResourcePath, String credentialToken) throws Exception {
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- GenericResource resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
- .setAuthzToken(authZToken)
- .setResourceId(parentResourceId).build());
- SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+ GenericResource resource;
+ SCPSecret scpSecret;
+ try (ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort)) {
+ resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
+ .setAuthzToken(authZToken)
+ .setResourceId(parentResourceId).build());
+ }
+
+ try (SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort)) {
+ scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+ }
boolean isChildPath = false;
if (childResourcePath != null && !"".equals(childResourcePath)) {
@@ -148,9 +158,9 @@ public class SCPMetadataCollector implements MetadataCollector {
}
String resourcePath = null;
- switch (resource.getResourceCase()){
+ switch (resource.getResourceCase()) {
case FILE:
- if (isChildPath){
+ if (isChildPath) {
throw new Exception("A child path can not be associated with a file parent");
}
resourcePath = resource.getFile().getResourcePath();
@@ -170,9 +180,9 @@ public class SCPMetadataCollector implements MetadataCollector {
}
GenericResource scpResource2 = GenericResource.newBuilder()
- .setFile(FileResource.newBuilder()
- .setResourcePath(resourcePath).build())
- .setScpStorage(resource.getScpStorage()).build();
+ .setFile(FileResource.newBuilder()
+ .setResourcePath(resourcePath).build())
+ .setScpStorage(resource.getScpStorage()).build();
return getFileResourceMetadata(authZToken, scpResource2, scpSecret);
}
@@ -224,25 +234,34 @@ public class SCPMetadataCollector implements MetadataCollector {
@Override
public DirectoryResourceMetadata getDirectoryResourceMetadata(AuthToken authZToken, String resourceId, String credentialToken) throws Exception {
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- GenericResource scpPResource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
- .setAuthzToken(authZToken)
- .setResourceId(resourceId).build());
+ GenericResource resource;
+ SCPSecret scpSecret;
+ try (ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort)) {
+ resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
+ .setAuthzToken(authZToken)
+ .setResourceId(resourceId).build());
+ }
- SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+ try (SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort)) {
+ scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+ }
- return getDirectoryResourceMetadata(authZToken,scpPResource, scpSecret);
+ return getDirectoryResourceMetadata(authZToken, resource, scpSecret);
}
@Override
public DirectoryResourceMetadata getDirectoryResourceMetadata(AuthToken authZToken, String parentResourceId, String childResourcePath, String credentialToken) throws Exception {
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- GenericResource resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
- .setAuthzToken(authZToken).setResourceId(parentResourceId).build());
- SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+ GenericResource resource;
+ SCPSecret scpSecret;
+ try (ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort)) {
+ resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
+ .setAuthzToken(authZToken).setResourceId(parentResourceId).build());
+ }
+
+ try (SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort)) {
+ scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder().setSecretId(credentialToken).build());
+ }
boolean isChildPath = false;
if (childResourcePath != null && !"".equals(childResourcePath)) {
@@ -283,34 +302,44 @@ public class SCPMetadataCollector implements MetadataCollector {
public Boolean isAvailable(AuthToken authZToken, String resourceId, String credentialToken) throws Exception {
checkInitialized();
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- GenericResource scpResource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
- .setAuthzToken(authZToken).setResourceId(resourceId).build());
- return isAvailable(authZToken, scpResource, credentialToken);
+ GenericResource resource;
+ try (ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort)) {
+ resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
+ .setAuthzToken(authZToken).setResourceId(resourceId).build());
+ }
+
+ return isAvailable(authZToken, resource, credentialToken);
}
@Override
public Boolean isAvailable(AuthToken authToken, String parentResourceId, String resourcePath, String credentialToken) throws Exception {
checkInitialized();
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- GenericResource scpResource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
- .setAuthzToken(authToken).setResourceId(parentResourceId).build());
- validateParent(scpResource, resourcePath);
+ GenericResource resource;
+ try (ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort)) {
+ resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
+ .setAuthzToken(authToken).setResourceId(parentResourceId).build());
+ }
+
+ validateParent(resource, resourcePath);
GenericResource targetScpResource = GenericResource.newBuilder()
.setFile(FileResource.newBuilder().setResourcePath(resourcePath).build())
- .setScpStorage(scpResource.getScpStorage()).build();
+ .setScpStorage(resource.getScpStorage()).build();
return isAvailable(authToken, targetScpResource, credentialToken);
}
public Boolean isAvailable(AuthToken authToken, GenericResource scpResource, String credentialToken) throws Exception {
- SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder()
- .setAuthzToken(authToken).setSecretId(credentialToken).build());
+
+ SCPSecret scpSecret;
+
+ try (SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort)) {
+ scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder()
+ .setAuthzToken(authToken).setSecretId(credentialToken).build());
+ }
try (SSHClient sshClient = getSSHClient(scpResource, scpSecret)) {
logger.info("Checking the availability of file {}", scpResource.getFile().getResourcePath());
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
index dab47d2..b70aa34 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPReceiver.java
@@ -89,19 +89,24 @@ public class SCPReceiver implements Connector {
ConnectorContext context) throws Exception {
checkInitialized();
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- GenericResource resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
- .setAuthzToken(authToken).setResourceId(resourceId).build());
+ GenericResource resource;
+ try (ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort)) {
+ resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
+ .setAuthzToken(authToken).setResourceId(resourceId).build());
+ }
if (resource.getStorageCase() != GenericResource.StorageCase.SCPSTORAGE) {
logger.error("Invalid storage type {} specified for resource {}", resource.getStorageCase(), resourceId);
throw new Exception("Invalid storage type specified for resource " + resourceId);
}
- SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder()
- .setAuthzToken(authToken)
- .setSecretId(credentialToken).build());
+ SCPSecret scpSecret;
+
+ try (SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort)) {
+ scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder()
+ .setAuthzToken(authToken)
+ .setSecretId(credentialToken).build());
+ }
SCPStorage scpStorage = resource.getScpStorage();
logger.info("Creating a ssh session for {}@{}:{}", scpSecret.getUser(), scpStorage.getHost(), scpStorage.getPort());
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
index ce33646..f17a532 100644
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPSender.java
@@ -88,19 +88,24 @@ public class SCPSender implements Connector {
public void startStream(AuthToken authToken, String resourceId, String childResourcePath, String credentialToken, ConnectorContext context) throws Exception {
checkInitialized();
- ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort);
- GenericResource resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
- .setAuthzToken(authToken).setResourceId(resourceId).build());
+ GenericResource resource;
+ try (ResourceServiceClient resourceClient = ResourceServiceClientBuilder.buildClient(resourceServiceHost, resourceServicePort)) {
+ resource = resourceClient.get().getGenericResource(GenericResourceGetRequest.newBuilder()
+ .setAuthzToken(authToken).setResourceId(resourceId).build());
+ }
if (resource.getStorageCase() != GenericResource.StorageCase.SCPSTORAGE) {
logger.error("Invalid storage type {} specified for resource {}", resource.getStorageCase(), resourceId);
throw new Exception("Invalid storage type specified for resource " + resourceId);
}
- SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort);
- SCPSecret scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder()
- .setAuthzToken(authToken)
- .setSecretId(credentialToken).build());
+ SCPSecret scpSecret;
+
+ try (SecretServiceClient secretClient = SecretServiceClientBuilder.buildClient(secretServiceHost, secretServicePort)) {
+ scpSecret = secretClient.scp().getSCPSecret(SCPSecretGetRequest.newBuilder()
+ .setAuthzToken(authToken)
+ .setSecretId(credentialToken).build());
+ }
SCPStorage scpStorage = resource.getScpStorage();
logger.info("Creating a ssh session for {}@{}:{}", scpSecret.getUser(), scpStorage.getHost(), scpStorage.getPort());