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 2020/03/09 00:37:50 UTC
[airavata-mft] branch master updated: Migrating to fully consul
lifecycle model
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-mft.git
The following commit(s) were added to refs/heads/master by this push:
new eb256f5 Migrating to fully consul lifecycle model
eb256f5 is described below
commit eb256f501501b3693febb700ed1fc1de1d346ef4
Author: Dimuthu Wannipurage <di...@gmail.com>
AuthorDate: Sun Mar 8 20:37:37 2020 -0400
Migrating to fully consul lifecycle model
---
.../admin/{MFTAdmin.java => MFTConsulClient.java} | 57 +++++---
.../org/apache/airavata/mft/agent/MFTAgent.java | 23 ++-
.../airavata/mft/agent/TransportMediator.java | 3 +-
.../org/apache/airavata/mft/api/AppConfig.java | 6 +-
.../airavata/mft/api/handler/MFTApiHandler.java | 25 ++--
.../airavata/mft/controller/MFTController.java | 71 +++------
.../apache/airavata/mft/core/ConnectorContext.java | 10 ++
.../airavata/mft/transport/scp/SCPReceiver.java | 158 +++++++++++----------
.../airavata/mft/transport/scp/SCPSender.java | 154 +++++++++++---------
.../mft/transport/scp/SCPTransportUtil.java | 8 +-
10 files changed, 267 insertions(+), 248 deletions(-)
diff --git a/admin/src/main/java/org/apache/airavata/mft/admin/MFTAdmin.java b/admin/src/main/java/org/apache/airavata/mft/admin/MFTConsulClient.java
similarity index 76%
rename from admin/src/main/java/org/apache/airavata/mft/admin/MFTAdmin.java
rename to admin/src/main/java/org/apache/airavata/mft/admin/MFTConsulClient.java
index ae5e780..f3f2c14 100644
--- a/admin/src/main/java/org/apache/airavata/mft/admin/MFTAdmin.java
+++ b/admin/src/main/java/org/apache/airavata/mft/admin/MFTConsulClient.java
@@ -18,6 +18,7 @@
package org.apache.airavata.mft.admin;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.orbitz.consul.Consul;
import com.orbitz.consul.ConsulException;
@@ -27,10 +28,13 @@ import org.apache.airavata.mft.admin.models.AgentInfo;
import org.apache.airavata.mft.admin.models.TransferCommand;
import org.apache.airavata.mft.admin.models.TransferRequest;
import org.apache.airavata.mft.admin.models.TransferState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
/*
mft/agents/messages/{agent-id} -> message
@@ -39,7 +43,9 @@ import java.util.stream.Collectors;
mft/transfer/state/{transfer-id} -> transfer state
*/
-public class MFTAdmin {
+public class MFTConsulClient {
+
+ private static final Logger logger = LoggerFactory.getLogger(MFTConsulClient.class);
private Consul client = Consul.builder().build();
private KeyValueClient kvClient = client.keyValueClient();
@@ -58,7 +64,7 @@ public class MFTAdmin {
public void commandTransferToAgent(String agentId, TransferCommand transferCommand) throws MFTAdminException {
try {
- updateTransferState(transferCommand.getTransferId(), new TransferState()
+ submitTransferState(transferCommand.getTransferId(), new TransferState()
.setState("INITIALIZING")
.setPercentage(0)
.setUpdateTimeMils(System.currentTimeMillis())
@@ -119,36 +125,55 @@ public class MFTAdmin {
}
}
- public TransferState getTransferState(String transferId) throws MFTAdminException {
+ public Optional<TransferState> getTransferState(String transferId) throws MFTAdminException {
+
try {
- Optional<Value> value = kvClient.getValue("mft/transfer/state/" + transferId);
- if (value.isPresent()) {
- if (value.get().getValueAsString().isPresent()) {
- String asStr = value.get().getValueAsString().get();
- return mapper.readValue(asStr, TransferState.class);
+ List<TransferState> states = getTransferStates(transferId);
+
+ Optional<TransferState> lastStatusOp = states.stream().min((o1, o2) -> {
+ if (o1.getUpdateTimeMils() == o2.getUpdateTimeMils()) {
+ return 0;
+ } else {
+ return o1.getUpdateTimeMils() - o2.getUpdateTimeMils() < 0 ? 1 : -1;
}
- }
- return new TransferState().setPercentage(0).setUpdateTimeMils(System.currentTimeMillis()).setState("UNKNOWN");
+ });
+
+ return lastStatusOp;
} catch (ConsulException e) {
- if (e.getCode() == 404) {
- return new TransferState().setPercentage(0).setUpdateTimeMils(System.currentTimeMillis()).setState("UNKNOWN");
- }
throw new MFTAdminException("Error in fetching transfer status " + transferId, e);
} catch (Exception e) {
throw new MFTAdminException("Error in fetching transfer status " + transferId, e);
}
}
- public void updateTransferState(String transferId, TransferState transferState) throws MFTAdminException {
+ public void submitTransferState(String transferId, TransferState transferState) throws MFTAdminException {
try {
- String asStr = mapper.writeValueAsString(transferState);
+ List<TransferState> allStates = getTransferStates(transferId);
+ System.out.println(allStates);
+ allStates.add(transferState);
+ String asStr = mapper.writeValueAsString(allStates);
kvClient.putValue("mft/transfer/state/" + transferId, asStr);
- } catch (JsonProcessingException e) {
+
+ logger.info("Saved transfer status " + asStr);
+
+ } catch (Exception e) {
throw new MFTAdminException("Error in serializing transfer status", e);
}
}
+ public List<TransferState> getTransferStates(String transferId) throws IOException {
+ Optional<Value> valueOp = kvClient.getValue("mft/transfer/state/" + transferId);
+ List<TransferState> allStates;
+ if (valueOp.isPresent()) {
+ String prevStates = valueOp.get().getValueAsString().get();
+ allStates = new ArrayList<>(Arrays.asList(mapper.readValue(prevStates, TransferState[].class)));
+ } else {
+ allStates = new ArrayList<>();
+ }
+ return allStates;
+ }
+
public List<AgentInfo> getLiveAgentInfos() throws MFTAdminException {
List<String> liveAgentIds = getLiveAgentIds();
return liveAgentIds.stream().map(id -> getAgentInfo(id).get()).collect(Collectors.toList());
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java b/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
index 25ba577..3c14483 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/MFTAgent.java
@@ -26,7 +26,7 @@ import com.orbitz.consul.cache.KVCache;
import com.orbitz.consul.model.kv.Value;
import com.orbitz.consul.model.session.ImmutableSession;
import com.orbitz.consul.model.session.SessionCreatedResponse;
-import org.apache.airavata.mft.admin.MFTAdmin;
+import org.apache.airavata.mft.admin.MFTConsulClient;
import org.apache.airavata.mft.admin.MFTAdminException;
import org.apache.airavata.mft.admin.models.AgentInfo;
import org.apache.airavata.mft.admin.models.TransferCommand;
@@ -36,12 +36,6 @@ import org.apache.airavata.mft.core.MetadataCollectorResolver;
import org.apache.airavata.mft.core.ResourceMetadata;
import org.apache.airavata.mft.core.api.Connector;
import org.apache.airavata.mft.core.api.MetadataCollector;
-import org.apache.airavata.mft.transport.local.LocalMetadataCollector;
-import org.apache.airavata.mft.transport.local.LocalReceiver;
-import org.apache.airavata.mft.transport.local.LocalSender;
-import org.apache.airavata.mft.transport.scp.SCPMetadataCollector;
-import org.apache.airavata.mft.transport.scp.SCPReceiver;
-import org.apache.airavata.mft.transport.scp.SCPSender;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,7 +47,6 @@ import org.springframework.context.annotation.PropertySource;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.Optional;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -90,13 +83,13 @@ public class MFTAgent implements CommandLineRunner {
private ObjectMapper mapper = new ObjectMapper();
- private MFTAdmin admin;
+ private MFTConsulClient mftConsulClient;
public void init() {
client = Consul.builder().build();
kvClient = client.keyValueClient();
messageCache = KVCache.newCache(kvClient, "mft/agents/messages/" + agentId );
- admin = new MFTAdmin();
+ mftConsulClient = new MFTConsulClient();
}
private void acceptRequests() {
@@ -110,7 +103,7 @@ public class MFTAgent implements CommandLineRunner {
try {
request = mapper.readValue(v, TransferCommand.class);
logger.info("Received request " + request.getTransferId());
- admin.updateTransferState(request.getTransferId(), new TransferState()
+ mftConsulClient.submitTransferState(request.getTransferId(), new TransferState()
.setState("STARTING")
.setPercentage(0)
.setUpdateTimeMils(System.currentTimeMillis())
@@ -127,7 +120,7 @@ public class MFTAgent implements CommandLineRunner {
MetadataCollector metadataCollector = metadataCollectorOp.orElseThrow(() -> new Exception("Could not find a metadata collector for input"));
ResourceMetadata metadata = metadataCollector.getGetResourceMetadata(request.getSourceId(), request.getSourceToken());
logger.debug("File size " + metadata.getResourceSize());
- admin.updateTransferState(request.getTransferId(), new TransferState()
+ mftConsulClient.submitTransferState(request.getTransferId(), new TransferState()
.setState("STARTED")
.setPercentage(0)
.setUpdateTimeMils(System.currentTimeMillis())
@@ -135,7 +128,7 @@ public class MFTAgent implements CommandLineRunner {
String transferId = mediator.transfer(request.getTransferId(), inConnector, outConnector, metadata, (id, st) -> {
try {
- admin.updateTransferState(id, st);
+ mftConsulClient.submitTransferState(id, st);
} catch (MFTAdminException e) {
logger.error("Failed while updating transfer state", e);
}
@@ -148,7 +141,7 @@ public class MFTAgent implements CommandLineRunner {
try {
logger.error("Error in submitting transfer {}", request.getTransferId(), e);
- admin.updateTransferState(request.getTransferId(), new TransferState()
+ mftConsulClient.submitTransferState(request.getTransferId(), new TransferState()
.setState("FAILED")
.setPercentage(0)
.setUpdateTimeMils(System.currentTimeMillis())
@@ -217,7 +210,7 @@ public class MFTAgent implements CommandLineRunner {
}
}, sessionRenewSeconds, sessionRenewSeconds, TimeUnit.SECONDS);
- this.admin.registerAgent(new AgentInfo()
+ this.mftConsulClient.registerAgent(new AgentInfo()
.setId(agentId)
.setHost(agentHost)
.setUser(agentUser)
diff --git a/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java b/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
index e524491..7cd5329 100644
--- a/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
+++ b/agent/src/main/java/org/apache/airavata/mft/agent/TransportMediator.java
@@ -52,6 +52,7 @@ public class TransportMediator {
ConnectorContext context = new ConnectorContext();
context.setMetadata(metadata);
context.setStreamBuffer(streamBuffer);
+ context.setTransferId(transferId);
TransferTask recvTask = new TransferTask(inConnector, context);
TransferTask sendTask = new TransferTask(outConnector, context);
@@ -109,7 +110,7 @@ public class TransportMediator {
logger.info("Transfer Speed " + (metadata.getResourceSize() * 1.0 / time) / (1024 * 1024) + " MB/s");
logger.info("Transfer " + transferId + " completed");
} catch (Exception e) {
- e.printStackTrace();
+ logger.error("Transfer {} failed", transferId, e);
} finally {
inConnector.destroy();
outConnector.destroy();
diff --git a/api/service/src/main/java/org/apache/airavata/mft/api/AppConfig.java b/api/service/src/main/java/org/apache/airavata/mft/api/AppConfig.java
index 964ccc1..f10c3a7 100644
--- a/api/service/src/main/java/org/apache/airavata/mft/api/AppConfig.java
+++ b/api/service/src/main/java/org/apache/airavata/mft/api/AppConfig.java
@@ -17,7 +17,7 @@
package org.apache.airavata.mft.api;
-import org.apache.airavata.mft.admin.MFTAdmin;
+import org.apache.airavata.mft.admin.MFTConsulClient;
import org.apache.airavata.mft.admin.models.TransferRequest;
import org.apache.airavata.mft.api.service.TransferApiRequest;
import org.dozer.DozerBeanMapper;
@@ -28,8 +28,8 @@ import org.springframework.context.annotation.Configuration;
@Configuration
public class AppConfig {
@Bean
- public MFTAdmin mftAdmin() {
- return new MFTAdmin();
+ public MFTConsulClient mftConsulClient() {
+ return new MFTConsulClient();
}
@Bean
diff --git a/api/service/src/main/java/org/apache/airavata/mft/api/handler/MFTApiHandler.java b/api/service/src/main/java/org/apache/airavata/mft/api/handler/MFTApiHandler.java
index 1146df6..2510370 100644
--- a/api/service/src/main/java/org/apache/airavata/mft/api/handler/MFTApiHandler.java
+++ b/api/service/src/main/java/org/apache/airavata/mft/api/handler/MFTApiHandler.java
@@ -18,12 +18,10 @@
package org.apache.airavata.mft.api.handler;
import io.grpc.stub.StreamObserver;
-import org.apache.airavata.mft.admin.MFTAdmin;
+import org.apache.airavata.mft.admin.MFTConsulClient;
import org.apache.airavata.mft.admin.models.TransferRequest;
import org.apache.airavata.mft.admin.models.TransferState;
import org.apache.airavata.mft.api.service.*;
-import org.apache.airavata.mft.api.db.entities.TransferStatusEntity;
-import org.apache.airavata.mft.api.db.repositories.TransferStatusRepository;
import org.apache.airavata.mft.core.MetadataCollectorResolver;
import org.apache.airavata.mft.core.api.MetadataCollector;
import org.dozer.DozerBeanMapper;
@@ -41,10 +39,7 @@ public class MFTApiHandler extends MFTApiServiceGrpc.MFTApiServiceImplBase {
private static final Logger logger = LoggerFactory.getLogger(MFTApiHandler.class);
@Autowired
- private TransferStatusRepository statusRepository;
-
- @Autowired
- private MFTAdmin mftAdmin;
+ private MFTConsulClient mftConsulClient;
@Autowired
private DozerBeanMapper dozerBeanMapper;
@@ -55,12 +50,13 @@ public class MFTApiHandler extends MFTApiServiceGrpc.MFTApiServiceImplBase {
TransferRequest transferRequest = dozerBeanMapper.map(request, TransferRequest.class);
Optional.ofNullable(request.getTargetAgentsMap()).ifPresent(transferRequest::setTargetAgents); // Custom mapping
- String transferId = mftAdmin.submitTransfer(transferRequest);
+ String transferId = mftConsulClient.submitTransfer(transferRequest);
logger.info("Submitted the transfer request {}", transferId);
- mftAdmin.updateTransferState(transferId, new TransferState()
+ mftConsulClient.submitTransferState(transferId, new TransferState()
.setUpdateTimeMils(System.currentTimeMillis())
- .setState("RECEIVED").setPercentage(0));
+ .setState("RECEIVED").setPercentage(0)
+ .setDescription("Received transfer job " + transferId));
responseObserver.onNext(TransferApiResponse.newBuilder().setTransferId(transferId).build());
responseObserver.onCompleted();
@@ -73,7 +69,7 @@ public class MFTApiHandler extends MFTApiServiceGrpc.MFTApiServiceImplBase {
@Override
public void getTransferStates(TransferStateApiRequest request, StreamObserver<TransferStateApiResponse> responseObserver) {
try {
- List<TransferStatusEntity> states = statusRepository.getByTransferId(request.getTransferId());
+ List<TransferState> states = mftConsulClient.getTransferStates(request.getTransferId());
states.forEach(st -> {
TransferStateApiResponse s = dozerBeanMapper.map(st, TransferStateApiResponse.newBuilder().getClass()).build();
responseObserver.onNext(s);
@@ -88,11 +84,10 @@ public class MFTApiHandler extends MFTApiServiceGrpc.MFTApiServiceImplBase {
@Override
public void getTransferState(TransferStateApiRequest request, StreamObserver<TransferStateApiResponse> responseObserver) {
try {
- List<TransferStatusEntity> states = statusRepository.getByTransferId(request.getTransferId());
+ Optional<TransferState> stateOp = mftConsulClient.getTransferState(request.getTransferId());
- Optional<TransferStatusEntity> firstElement = states.stream().findFirst();
- if (firstElement.isPresent()) {
- TransferStateApiResponse s = dozerBeanMapper.map(firstElement.get(),
+ if (stateOp.isPresent()) {
+ TransferStateApiResponse s = dozerBeanMapper.map(stateOp.get(),
TransferStateApiResponse.newBuilder().getClass()).build();
responseObserver.onNext(s);
} else {
diff --git a/controller/src/main/java/org/apache/airavata/mft/controller/MFTController.java b/controller/src/main/java/org/apache/airavata/mft/controller/MFTController.java
index 29f7a94..6a1532d 100644
--- a/controller/src/main/java/org/apache/airavata/mft/controller/MFTController.java
+++ b/controller/src/main/java/org/apache/airavata/mft/controller/MFTController.java
@@ -17,25 +17,20 @@
package org.apache.airavata.mft.controller;
+import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.orbitz.consul.Consul;
import com.orbitz.consul.KeyValueClient;
import com.orbitz.consul.cache.ConsulCache;
import com.orbitz.consul.cache.KVCache;
import com.orbitz.consul.model.kv.Value;
-import org.apache.airavata.mft.admin.MFTAdmin;
+import org.apache.airavata.mft.admin.MFTConsulClient;
import org.apache.airavata.mft.admin.MFTAdminException;
import org.apache.airavata.mft.admin.models.TransferCommand;
-import org.apache.airavata.mft.admin.models.TransferState;
-import org.apache.airavata.mft.api.db.entities.TransferEntity;
-import org.apache.airavata.mft.api.db.entities.TransferStatusEntity;
-import org.apache.airavata.mft.api.db.repositories.TransferRepository;
import org.apache.airavata.mft.admin.models.TransferRequest;
-import org.apache.airavata.mft.api.db.repositories.TransferStatusRepository;
import org.dozer.DozerBeanMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -66,23 +61,18 @@ public class MFTController implements CommandLineRunner {
private ConsulCache.Listener<String, Value> messageCacheListener;
private ConsulCache.Listener<String, Value> stateCacheListener;
- private MFTAdmin admin;
+ private MFTConsulClient mftConsulClient;
private ObjectMapper jsonMapper = new ObjectMapper();
private DozerBeanMapper dozerBeanMapper = new DozerBeanMapper();
- @Autowired
- private TransferRepository transferRepository;
-
- @Autowired
- private TransferStatusRepository statusRepository;
public void init() {
client = Consul.builder().build();
kvClient = client.keyValueClient();
messageCache = KVCache.newCache(kvClient, "mft/controller/messages");
stateCache = KVCache.newCache(kvClient, "mft/transfer/state");
- admin = new MFTAdmin();
+ mftConsulClient = new MFTConsulClient();
}
private void acceptRequests() {
@@ -96,7 +86,7 @@ public class MFTController implements CommandLineRunner {
TransferRequest transferRequest = jsonMapper.readValue(v, TransferRequest.class);
TransferCommand transferCommand = saveAndCreateTransferCommand(transferId, transferRequest);
String selectedAgent = selectAgent(transferRequest);
- admin.commandTransferToAgent(selectedAgent, transferCommand);
+ mftConsulClient.commandTransferToAgent(selectedAgent, transferCommand);
} catch (Exception e) {
logger.error("Failed to process the request", e);
} finally {
@@ -116,26 +106,14 @@ public class MFTController implements CommandLineRunner {
try {
if (value.getValueAsString().isPresent()) {
String asStr = value.getValueAsString().get();
- logger.info("Received state {}", asStr);
- TransferState transferState = jsonMapper.readValue(asStr, TransferState.class);
- String transferId = key.substring(key.lastIndexOf("/") + 1);
- Optional<TransferEntity> transferEntity = transferRepository.findById(transferId);
- if (transferEntity.isPresent()) {
- TransferStatusEntity ety = new TransferStatusEntity()
- .setPercentage(transferState.getPercentage())
- .setState(transferState.getState())
- .setUpdateTimeMils(transferState.getUpdateTimeMils())
- .setDescription(Optional.ofNullable(transferState.getDescription()).orElse(transferState.getState()))
- .setTransfer(transferEntity.get());
- statusRepository.save(ety);
- logger.info("Saved state for transfer {}", transferId);
- }
+
+ logger.info("Received state Key {} val {}", key, asStr);
}
} catch (Exception e) {
logger.error("Error while processing the state message", e);
} finally {
- logger.info("Deleting key " + value.getKey());
- kvClient.deleteKey(value.getKey()); // Due to bug in consul https://github.com/hashicorp/consul/issues/571
+ //logger.info("Deleting key " + value.getKey());
+ //kvClient.deleteKey(value.getKey()); // Due to bug in consul https://github.com/hashicorp/consul/issues/571
}
});
};
@@ -143,10 +121,12 @@ public class MFTController implements CommandLineRunner {
stateCache.start();
}
- private TransferCommand saveAndCreateTransferCommand(String transferId, TransferRequest transferRequest) {
- TransferEntity transferEntity = new TransferEntity();
- transferEntity.setTransferId(transferId);
- transferEntity.setSourceId(transferRequest.getSourceId())
+ private TransferCommand saveAndCreateTransferCommand(String transferId, TransferRequest transferRequest) throws JsonProcessingException {
+
+ kvClient.putValue("mft/transfer/processed/" +transferId, jsonMapper.writeValueAsString(transferRequest));
+
+ TransferCommand transferCommand = new TransferCommand();
+ transferCommand.setSourceId(transferRequest.getSourceId())
.setSourceToken(transferRequest.getSourceToken())
.setSourceType(transferRequest.getSourceType())
.setSourceResourceBackend(transferRequest.getSourceResourceBackend())
@@ -156,30 +136,13 @@ public class MFTController implements CommandLineRunner {
.setDestinationType(transferRequest.getDestinationType())
.setDestResourceBackend(transferRequest.getDestResourceBackend())
.setDestCredentialBackend(transferRequest.getDestCredentialBackend())
- .setAffinityTransfer(transferRequest.isAffinityTransfer());
-
- TransferEntity savedEntity = transferRepository.save(transferEntity);
-
-
- TransferCommand transferCommand = new TransferCommand();
- transferCommand.setSourceId(savedEntity.getSourceId())
- .setSourceToken(savedEntity.getSourceToken())
- .setSourceType(savedEntity.getSourceType())
- .setSourceResourceBackend(savedEntity.getSourceResourceBackend())
- .setSourceCredentialBackend(savedEntity.getSourceCredentialBackend())
- .setDestinationId(savedEntity.getDestinationId())
- .setDestinationToken(savedEntity.getDestinationToken())
- .setDestinationType(savedEntity.getDestinationType())
- .setDestResourceBackend(savedEntity.getDestResourceBackend())
- .setDestCredentialBackend(savedEntity.getDestCredentialBackend())
- .setTransferId(savedEntity.getTransferId());
-
+ .setTransferId(transferId);
return transferCommand;
}
private String selectAgent(TransferRequest transferRequest) throws ControllerException, MFTAdminException {
- List<String> liveAgentIds = admin.getLiveAgentIds();
+ List<String> liveAgentIds = mftConsulClient.getLiveAgentIds();
if (liveAgentIds.isEmpty()) {
logger.error("Live agents are not available. Skipping for now");
throw new ControllerException("Live agents are not available. Skipping for now");
diff --git a/core/src/main/java/org/apache/airavata/mft/core/ConnectorContext.java b/core/src/main/java/org/apache/airavata/mft/core/ConnectorContext.java
index f8d3dd9..2634649 100644
--- a/core/src/main/java/org/apache/airavata/mft/core/ConnectorContext.java
+++ b/core/src/main/java/org/apache/airavata/mft/core/ConnectorContext.java
@@ -21,6 +21,7 @@ public class ConnectorContext {
private CircularStreamingBuffer streamBuffer;
private ResourceMetadata metadata;
+ private String transferId;
public CircularStreamingBuffer getStreamBuffer() {
return streamBuffer;
@@ -37,4 +38,13 @@ public class ConnectorContext {
public void setMetadata(ResourceMetadata metadata) {
this.metadata = metadata;
}
+
+ public String getTransferId() {
+ return transferId;
+ }
+
+ public ConnectorContext setTransferId(String transferId) {
+ this.transferId = transferId;
+ return this;
+ }
}
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 f5db8e5..18b5ee0 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
@@ -28,10 +28,15 @@ import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.service.SCPSecret;
import org.apache.airavata.mft.secret.service.SCPSecretGetRequest;
import org.apache.airavata.mft.secret.service.SecretServiceGrpc;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.*;
public class SCPReceiver implements Connector {
+
+ private static final Logger logger = LoggerFactory.getLogger(SCPReceiver.class);
+
private Session session;
private SCPResource scpResource;
@@ -58,11 +63,7 @@ public class SCPReceiver implements Connector {
}
public void destroy() {
- try {
- this.session.disconnect();
- } catch (Exception e) {
- e.printStackTrace();
- }
+
}
public void startStream(ConnectorContext context) throws Exception {
@@ -72,98 +73,107 @@ public class SCPReceiver implements Connector {
}
transferRemoteToStream(session, this.scpResource.getResourcePath(), context.getStreamBuffer());
+ logger.info("SCP Receive completed. Transfer {}", context.getTransferId());
}
private void transferRemoteToStream(Session session, String from, CircularStreamingBuffer streamBuffer) throws Exception {
- OutputStream outputStream = streamBuffer.getOutputStream();
-
- System.out.println("Starting scp receive");
- // exec 'scp -f rfile' remotely
- String command = "scp -f " + from;
- com.jcraft.jsch.Channel channel = session.openChannel("exec");
- ((ChannelExec) channel).setCommand(command);
-
- // get I/O streams for remote scp
- OutputStream out = channel.getOutputStream();
- InputStream in = channel.getInputStream();
+ try {
+ OutputStream outputStream = streamBuffer.getOutputStream();
- channel.connect();
+ logger.info("Starting scp receive");
+ // exec 'scp -f rfile' remotely
+ String command = "scp -f " + from;
+ com.jcraft.jsch.Channel channel = session.openChannel("exec");
+ ((ChannelExec) channel).setCommand(command);
- byte[] buf = new byte[1024];
+ // get I/O streams for remote scp
+ OutputStream out = channel.getOutputStream();
+ InputStream in = channel.getInputStream();
- // send '\0'
- buf[0] = 0;
- out.write(buf, 0, 1);
- out.flush();
+ channel.connect();
- while (true) {
- int c = checkAck(in);
- if (c != 'C') {
- break;
- }
+ byte[] buf = new byte[1024];
- // read '0644 '
- in.read(buf, 0, 5);
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
- long filesize = 0L;
while (true) {
- if (in.read(buf, 0, 1) < 0) {
- // error
+ int c = checkAck(in);
+ if (c != 'C') {
break;
}
- if (buf[0] == ' ') break;
- filesize = filesize * 10L + (long) (buf[0] - '0');
- }
- String file = null;
- for (int i = 0; ; i++) {
- in.read(buf, i, 1);
- if (buf[i] == (byte) 0x0a) {
- file = new String(buf, 0, i);
- break;
+ // read '0644 '
+ in.read(buf, 0, 5);
+
+ long filesize = 0L;
+ while (true) {
+ if (in.read(buf, 0, 1) < 0) {
+ // error
+ break;
+ }
+ if (buf[0] == ' ') break;
+ filesize = filesize * 10L + (long) (buf[0] - '0');
}
- }
- System.out.println("file-size=" + filesize + ", file=" + file);
- // send '\0'
- buf[0] = 0;
- out.write(buf, 0, 1);
- out.flush();
+ String file = null;
+ for (int i = 0; ; i++) {
+ in.read(buf, i, 1);
+ if (buf[i] == (byte) 0x0a) {
+ file = new String(buf, 0, i);
+ break;
+ }
+ }
- // read a content of lfile
- int bufSize;
- while (true) {
- if (buf.length < filesize) bufSize = buf.length;
- else bufSize = (int) filesize;
- bufSize = in.read(buf, 0, bufSize);
- if (bufSize < 0) {
- // error
- break;
+ logger.info("file-size=" + filesize + ", file=" + file);
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
+
+ // read a content of lfile
+ int bufSize;
+ while (true) {
+ if (buf.length < filesize) bufSize = buf.length;
+ else bufSize = (int) filesize;
+ bufSize = in.read(buf, 0, bufSize);
+ if (bufSize < 0) {
+ // error
+ break;
+ }
+ //System.out.println("Read " + bufSize);
+ outputStream.write(buf, 0, bufSize);
+ outputStream.flush();
+
+ filesize -= bufSize;
+ if (filesize == 0L) break;
}
- //System.out.println("Read " + bufSize);
- outputStream.write(buf, 0, bufSize);
- outputStream.flush();
- filesize -= bufSize;
- if (filesize == 0L) break;
- }
+ if (checkAck(in) != 0) {
+ throw new IOException("Error code found in ack " + (checkAck(in)));
+ }
- if (checkAck(in) != 0) {
- throw new IOException("Error code found in ack " + (checkAck(in)));
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
+ outputStream.close();
}
- // send '\0'
- buf[0] = 0;
- out.write(buf, 0, 1);
- out.flush();
- outputStream.close();
- }
-
- channel.disconnect();
- session.disconnect();
+ channel.disconnect();
+ session.disconnect();
- System.out.println("Completed scp receive");
+ logger.info("Completed scp receive");
+ } finally {
+ try {
+ session.disconnect();
+ } catch (Exception e) {
+ logger.warn("Session disconnection failed", e);
+ }
+ }
}
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 acf9cbe..d909238 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
@@ -31,13 +31,19 @@ import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.service.SCPSecret;
import org.apache.airavata.mft.secret.service.SCPSecretGetRequest;
import org.apache.airavata.mft.secret.service.SecretServiceGrpc;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.*;
public class SCPSender implements Connector {
+
+ private static final Logger logger = LoggerFactory.getLogger(SCPSender.class);
+
private Session session;
private SCPResource scpResource;
- public void init(String resourceId, String credentialToken) throws IOException {
+
+ public void init(String resourceId, String credentialToken) throws Exception {
ResourceServiceGrpc.ResourceServiceBlockingStub resourceClient = ResourceServiceClient.buildClient("localhost", 7002);
this.scpResource = resourceClient.getSCPResource(SCPResourceGetRequest.newBuilder().setResourceId(resourceId).build());
@@ -50,6 +56,12 @@ public class SCPSender implements Connector {
writer.write(scpSecret.getPrivateKey());
writer.close();
+ logger.info("Creating a ssh session for {}@{}:{} with key {} and passphrase {}",
+ scpSecret.getUser(), scpResource.getScpStorage().getHost(),
+ scpResource.getScpStorage().getPort(),
+ privateKeyFile.getPath(),
+ scpSecret.getPassphrase());
+
this.session = SCPTransportUtil.createSession(scpSecret.getUser(), scpResource.getScpStorage().getHost(),
scpResource.getScpStorage().getPort(),
privateKeyFile.getPath(),
@@ -58,11 +70,11 @@ public class SCPSender implements Connector {
public void destroy() {
- try {
- this.session.disconnect();
- } catch (Exception e) {
- e.printStackTrace();
- }
+ //try {
+ // this.session.disconnect();
+ //} catch (Exception e) {
+ // logger.error("Errored while disconnecting session", e);
+ //}
}
public void startStream(ConnectorContext context) throws Exception {
@@ -72,91 +84,99 @@ public class SCPSender implements Connector {
}
try {
copyLocalToRemote(this.session, this.scpResource.getResourcePath(), context.getStreamBuffer(), context.getMetadata().getResourceSize());
+ logger.info("SCP send to transfer {} completed", context.getTransferId());
} catch (Exception e) {
- e.printStackTrace();
+ logger.error("Errored while streaming to remote scp server. Transfer {}", context.getTransferId() , e);
throw e;
}
}
private void copyLocalToRemote(Session session, String to, CircularStreamingBuffer streamBuffer, long fileSize) throws JSchException, IOException {
- System.out.println("Starting scp send");
- InputStream inputStream = streamBuffer.getInputStream();
+ try {
+ logger.info("Starting scp send for remote server");
+ InputStream inputStream = streamBuffer.getInputStream();
- boolean ptimestamp = true;
+ boolean ptimestamp = true;
- // exec 'scp -t rfile' remotely
- String command = "scp " + (ptimestamp ? "-p" : "") + " -t " + to;
- com.jcraft.jsch.Channel channel = session.openChannel("exec");
- ((ChannelExec) channel).setCommand(command);
+ // exec 'scp -t rfile' remotely
+ String command = "scp " + (ptimestamp ? "-p" : "") + " -t " + to;
+ com.jcraft.jsch.Channel channel = session.openChannel("exec");
+ ((ChannelExec) channel).setCommand(command);
- // get I/O streams for remote scp
- OutputStream out = channel.getOutputStream();
- InputStream in = channel.getInputStream();
+ // get I/O streams for remote scp
+ OutputStream out = channel.getOutputStream();
+ InputStream in = channel.getInputStream();
- channel.connect();
+ channel.connect();
- if (checkAck(in) != 0) {
- throw new IOException("Error code found in ack " + (checkAck(in)));
- }
-
- if (ptimestamp) {
- command = "T" + (System.currentTimeMillis() / 1000) + " 0";
- // The access time should be sent here,
- // but it is not accessible with JavaAPI ;-<
- command += (" " + (System.currentTimeMillis() / 1000) + " 0\n");
- out.write(command.getBytes());
- out.flush();
if (checkAck(in) != 0) {
throw new IOException("Error code found in ack " + (checkAck(in)));
}
- }
- // send "C0644 filesize filename", where filename should not include '/'
- command = "C0644 " + fileSize + " ";
- if (to.lastIndexOf('/') > 0) {
- command += to.substring(to.lastIndexOf('/') + 1);
- } else {
- command += to;
- }
+ if (ptimestamp) {
+ command = "T" + (System.currentTimeMillis() / 1000) + " 0";
+ // The access time should be sent here,
+ // but it is not accessible with JavaAPI ;-<
+ command += (" " + (System.currentTimeMillis() / 1000) + " 0\n");
+ out.write(command.getBytes());
+ out.flush();
+ if (checkAck(in) != 0) {
+ throw new IOException("Error code found in ack " + (checkAck(in)));
+ }
+ }
- command += "\n";
- out.write(command.getBytes());
- out.flush();
+ // send "C0644 filesize filename", where filename should not include '/'
+ command = "C0644 " + fileSize + " ";
+ if (to.lastIndexOf('/') > 0) {
+ command += to.substring(to.lastIndexOf('/') + 1);
+ } else {
+ command += to;
+ }
- if (checkAck(in) != 0) {
- throw new IOException("Error code found in ack " + (checkAck(in)));
- }
+ command += "\n";
+ out.write(command.getBytes());
+ out.flush();
- // send a content of lfile
- byte[] buf = new byte[1024];
- long totalWritten = 0;
- while (true) {
- int len = inputStream.read(buf, 0, buf.length);
- if (len == -1) {
- break;
- } else {
- out.write(buf, 0, len); //out.flush();
- totalWritten += len;
- //System.out.println("Write " + totalWritten);
- if (totalWritten == fileSize) {
+ if (checkAck(in) != 0) {
+ throw new IOException("Error code found in ack " + (checkAck(in)));
+ }
+
+ // send a content of lfile
+ byte[] buf = new byte[1024];
+ long totalWritten = 0;
+ while (true) {
+ int len = inputStream.read(buf, 0, buf.length);
+ if (len == -1) {
break;
+ } else {
+ out.write(buf, 0, len); //out.flush();
+ totalWritten += len;
+ //System.out.println("Write " + totalWritten);
+ if (totalWritten == fileSize) {
+ break;
+ }
}
}
- }
- // send '\0'
- buf[0] = 0;
- out.write(buf, 0, 1);
- out.flush();
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
- if (checkAck(in) != 0) {
- throw new IOException("Error code found in ack " + (checkAck(in)));
+ if (checkAck(in) != 0) {
+ throw new IOException("Error code found in ack " + (checkAck(in)));
+ }
+ out.close();
+ logger.info("Completed scp send for remote server");
+ channel.disconnect();
+
+ } finally {
+ try {
+ session.disconnect();
+ } catch (Exception e) {
+ logger.warn("Session disconnection failed", e);
+ }
}
- out.close();
-
- channel.disconnect();
- session.disconnect();
- System.out.println("Done sending");
}
public int checkAck(InputStream in) throws IOException {
diff --git a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
index 5174cb6..9f3c3bb 100755
--- a/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
+++ b/transport/scp-transport/src/main/java/org/apache/airavata/mft/transport/scp/SCPTransportUtil.java
@@ -26,12 +26,15 @@ import org.apache.airavata.mft.secret.client.SecretServiceClient;
import org.apache.airavata.mft.secret.service.SCPSecret;
import org.apache.airavata.mft.secret.service.SCPSecretCreateRequest;
import org.apache.airavata.mft.secret.service.SecretServiceGrpc;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Properties;
public class SCPTransportUtil {
// TODO replace with an API call to the registry
+ private static final Logger logger = LoggerFactory.getLogger(SCPTransportUtil.class);
public static void main(String args[]) {
ResourceServiceGrpc.ResourceServiceBlockingStub resourceClient = ResourceServiceClient.buildClient("localhost", 7002);
@@ -79,7 +82,7 @@ public class SCPTransportUtil {
return null;
}
- public static Session createSession(String user, String host, int port, String keyFilePath, String keyPassword) {
+ public static Session createSession(String user, String host, int port, String keyFilePath, String keyPassword) throws Exception {
try {
JSch jsch = new JSch();
@@ -100,8 +103,7 @@ public class SCPTransportUtil {
return session;
} catch (JSchException e) {
- e.printStackTrace();
- return null;
+ throw new Exception("Failed to create a ssh session for " + user + "@" + host + ":" + port, e);
}
}
}