recording unverified resources, users, and invoke workflows when user register
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/Constants.java b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/Constants.java new file mode 100644 index 0000000..c1f3d5d --- /dev/null +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/Constants.java
@@ -0,0 +1,8 @@ +package org.apache.airavata.datalake.orchestrator; + +public class Constants { + + public static final String ERROR_CODE_INVALID_TYPE="ERR_0001"; + public static final String ERROR_CODE_INVALID_USERNAME="ERR_0002"; + public static final String ERROR_CODE_INVALID_PATH="ERR_0003"; +}
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/DataOrchestratorAPIRunner.java b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/DataOrchestratorAPIRunner.java index 7fcefc3..dde3708 100644 --- a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/DataOrchestratorAPIRunner.java +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/DataOrchestratorAPIRunner.java
@@ -61,24 +61,16 @@ LOGGER.info("Starting Data orchestrator API Server ..."); LOGGER.info("Loading configuration from file {} ...", configPath); - Configuration configuration = this.loadConfig(configPath); + Configuration configuration = Utils.loadConfig(configPath); LOGGER.info("Registering Orchestration even handler " + OrchestratorEventHandler.class.getName() + " ..."); orchestratorEventHandler.init(configuration); + LOGGER.info("Data orchestrator start accepting events ...."); orchestratorEventHandler.startProcessing(); } - private Configuration loadConfig(String filePath) { - LOGGER.info("File path " + filePath); - try (InputStream in = new FileInputStream(filePath)) { - Yaml yaml = new Yaml(); - return yaml.loadAs(in, Configuration.class); - } catch (Exception exception) { - LOGGER.error("Error loading config file", exception); - } - return null; - } + }
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/Utils.java b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/Utils.java index 420c042..79fbb45 100644 --- a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/Utils.java +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/Utils.java
@@ -1,5 +1,11 @@ package org.apache.airavata.datalake.orchestrator; +import org.yaml.snakeyaml.Yaml; + +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; import java.math.BigInteger; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; @@ -26,4 +32,11 @@ return hexString.toString(); } + + public static Configuration loadConfig(String filePath) throws IOException { + try (InputStream in = new FileInputStream(filePath)) { + Yaml yaml = new Yaml(); + return yaml.loadAs(in, Configuration.class); + } + } }
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/connectors/DRMSConnector.java b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/connectors/DRMSConnector.java index f943449..1f10703 100644 --- a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/connectors/DRMSConnector.java +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/connectors/DRMSConnector.java
@@ -194,6 +194,60 @@ } } + public Optional<GenericResource> createUnverifiedResource(String authToken, + String tenantId, + String resourceId, + String resourcePath, + String type, + String errorCode, + String errorDiscription, + String unverifiedUser) throws Exception { + + DRMSServiceAuthToken serviceAuthToken = DRMSServiceAuthToken.newBuilder() + .setAccessToken(authToken) + .setUserUnverified(true) + .setAuthCredentialType(AuthCredentialType.AGENT_ACCOUNT_CREDENTIAL) + .setAuthenticatedUser(AuthenticatedUser.newBuilder() + .setTenantId(tenantId) + .build()) + .build(); + if (unverifiedUser != null){ + serviceAuthToken = serviceAuthToken + .toBuilder() + .setAuthenticatedUser(AuthenticatedUser.newBuilder() + .setTenantId(tenantId) + .setUsername(unverifiedUser) + .build()).build(); + } + + GenericResource genericResource = GenericResource + .newBuilder() + .setResourceId(resourceId) + .setResourcePath(resourcePath) + .setType(type) + .putProperties("ERROR_CODE", errorCode) + .putProperties("ERROR_DISCRIPTION", errorDiscription) + .build(); + ResourceCreateRequest resourceCreateRequest = ResourceCreateRequest + .newBuilder() + .setAuthToken(serviceAuthToken) + .setResource(genericResource) + .build(); + + try { + ResourceCreateResponse resourceCreateResponse = resourceServiceBlockingStub.createUnverifiedResource(resourceCreateRequest); + return Optional.ofNullable(resourceCreateResponse.getResource()); + } catch (Exception ex) { + LOGGER.error("Error occurred while creating unverified resource {} in DRMS", resourcePath, ex); + return Optional.empty(); + } + } + + + + + + public void addResourceMetadata(String authToken, String tenantId, String resourceId,
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/async/OrchestratorEventHandler.java b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/async/OrchestratorEventHandler.java index 9b056a5..074d216 100644 --- a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/async/OrchestratorEventHandler.java +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/async/OrchestratorEventHandler.java
@@ -20,6 +20,7 @@ import org.apache.airavata.datalake.data.orchestrator.api.stub.notification.Notification; import org.apache.airavata.datalake.data.orchestrator.api.stub.notification.NotificationRegisterRequest; import org.apache.airavata.datalake.orchestrator.Configuration; +import org.apache.airavata.datalake.orchestrator.registry.persistance.entity.notification.NotificationEntity; import org.apache.airavata.dataorchestrator.clients.core.NotificationClient; import org.apache.airavata.dataorchestrator.messaging.consumer.MessageConsumer; import org.dozer.DozerBeanMapper; @@ -94,6 +95,23 @@ })); } + + + public void invokeMessageFlowForNotification(Notification notification) throws Exception { + try { + if (!eventCache.contains(notification.getResourcePath() + ":" + notification.getHostName())) { + eventCache.add(notification.getResourcePath() + ":" + notification.getHostName()); + this.executorService.submit(new OrchestratorEventProcessor( + configuration, notification, eventCache, notificationClient)); + } + }catch (Exception e) { + LOGGER.error("Failed to submit data orchestrator event to process on path {}", + notification.getResourcePath(), e); + } + } + + + public Configuration getConfiguration() { return configuration; }
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/async/OrchestratorEventProcessor.java b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/async/OrchestratorEventProcessor.java index f6a4e49..e8e0546 100644 --- a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/async/OrchestratorEventProcessor.java +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/async/OrchestratorEventProcessor.java
@@ -24,6 +24,7 @@ import org.apache.airavata.datalake.drms.storage.AnyStoragePreference; import org.apache.airavata.datalake.drms.storage.TransferMapping; import org.apache.airavata.datalake.orchestrator.Configuration; +import org.apache.airavata.datalake.orchestrator.Constants; import org.apache.airavata.datalake.orchestrator.Utils; import org.apache.airavata.datalake.orchestrator.connectors.CustosConnector; import org.apache.airavata.datalake.orchestrator.connectors.DRMSConnector; @@ -160,17 +161,16 @@ } } - private String verifyUser(String userName) throws Exception { + private Optional<String> verifyUser(String userName) throws Exception { if (custosConnector.findUserByUserName(userName).isEmpty()) { Optional<UserRepresentation> userByEmail = custosConnector.findUserByEmail(userName); if (userByEmail.isPresent()) { - return userByEmail.get().getUsername(); + return Optional.of(userByEmail.get().getUsername()); } else { - logger.error("No user {} by email or user name", userName); - throw new Exception("Could not find the user " + userName); + return Optional.empty(); } } else { - return userName; + return Optional.of(userName); } } @@ -198,6 +198,11 @@ notification.getResourcePath(), notification.getResourceType()); logger.error("Resource should be a Folder type"); + this.drmsConnector.createUnverifiedResource(notification.getAuthToken(), notification.getTenantId(), + notification.getNotificationId(), notification.getResourcePath(), notification.getResourceType(), + Constants.ERROR_CODE_INVALID_TYPE, " Invalid type :" + notification.getResourceType(), null); + throw new Exception("Resource should be a Folder type"); + } String removeBasePath = notification.getResourcePath().substring(notification.getBasePath().length()); @@ -205,11 +210,31 @@ if (splitted.length < 2) { logger.error("Invalid path. Need at least two folder levels from base. {}", removeBasePath); + this.drmsConnector.createUnverifiedResource(notification.getAuthToken(), notification.getTenantId(), + notification.getNotificationId(), notification.getResourcePath(), notification.getResourceType(), + Constants.ERROR_CODE_INVALID_PATH, " Invalid path ", null); throw new Exception("Invalid path. Need at least two folder levels from base"); } + String adminUser = null; + String owner = null; - String adminUser = verifyUser(splitted[0]); - String owner = verifyUser(splitted[1].split("_")[0]); + Optional<String> adminUserOp = verifyUser(splitted[0]); + Optional<String> ownerOp = verifyUser(splitted[1].split("_")[0]); + if (adminUserOp.isEmpty()) { + this.drmsConnector.createUnverifiedResource(notification.getAuthToken(), notification.getTenantId(), + notification.getNotificationId(), notification.getResourcePath(), notification.getResourceType(), + Constants.ERROR_CODE_INVALID_USERNAME, " User not verified ", splitted[0]); + logger.error("Invalid user. User should be verified users {} . {}", splitted[0], removeBasePath); + throw new Exception("Invalid user. User " + splitted[0] + " should be registered users"); + } + if (ownerOp.isEmpty()) { + this.drmsConnector.createUnverifiedResource(notification.getAuthToken(), notification.getTenantId(), + notification.getNotificationId(), notification.getResourcePath(), notification.getResourceType(), + Constants.ERROR_CODE_INVALID_USERNAME, " User not verified ", splitted[1]); + logger.error("Invalid user. User should be verified users {} . {}", splitted[1], removeBasePath); + throw new Exception("Invalid user. User " + splitted[1] + " should be registered users"); + } + Map<String, String> ownerRules = new HashMap<>(); ownerRules.put(adminUser, "VIEWER"); @@ -393,4 +418,6 @@ directoryMetadata.getResourcePath(), sourceStorageId, sourceStorageId, System.currentTimeMillis() - start); } } + + }
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/grpc/NotificationApiHandler.java b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/grpc/NotificationApiHandler.java index 61fbbd1..d6269c7 100644 --- a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/grpc/NotificationApiHandler.java +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/java/org/apache/airavata/datalake/orchestrator/handlers/grpc/NotificationApiHandler.java
@@ -17,29 +17,48 @@ package org.apache.airavata.datalake.orchestrator.handlers.grpc; +import com.google.protobuf.Empty; +import io.grpc.Status; import io.grpc.stub.StreamObserver; import org.apache.airavata.datalake.data.orchestrator.api.stub.notification.*; +import org.apache.airavata.datalake.orchestrator.Configuration; +import org.apache.airavata.datalake.orchestrator.Utils; +import org.apache.airavata.datalake.orchestrator.handlers.async.OrchestratorEventHandler; import org.apache.airavata.datalake.orchestrator.registry.persistance.entity.notification.NotificationEntity; import org.apache.airavata.datalake.orchestrator.registry.persistance.entity.notification.NotificationStatusEntity; import org.apache.airavata.datalake.orchestrator.registry.persistance.repository.NotificationEntityRepository; import org.apache.airavata.datalake.orchestrator.registry.persistance.repository.NotificationStatusEntityRepository; import org.dozer.DozerBeanMapper; import org.lognet.springboot.grpc.GRpcService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import java.util.ArrayList; import java.util.List; -import java.util.stream.Stream; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; @GRpcService public class NotificationApiHandler extends NotificationServiceGrpc.NotificationServiceImplBase { + private static final Logger LOGGER = LoggerFactory.getLogger(OrchestratorEventHandler.class); + @Autowired private NotificationEntityRepository notificationRepository; @Autowired private NotificationStatusEntityRepository notificationStatusRepository; + + @Autowired + private OrchestratorEventHandler orchestratorEventHandler; + + @org.springframework.beans.factory.annotation.Value("${config.path}") + private String configPath; + + @Override public void registerNotification(NotificationRegisterRequest request, StreamObserver<NotificationRegisterResponse> responseObserver) { DozerBeanMapper mapper = new DozerBeanMapper(); @@ -99,4 +118,40 @@ responseObserver.onNext(responseBuilder.build()); responseObserver.onCompleted(); } + + + @Override + public void invokeNotification(NotificationInvokeRequest request, StreamObserver<NotificationInvokeResponse> responseObserver) { + try { + Optional<NotificationEntity> notificationOP = notificationRepository.findById(request.getNotificationId()); + + if (notificationOP.isPresent()) { + NotificationEntity notificationEntity = notificationOP.get(); + + + Notification notification = Notification + .newBuilder() + .setNotificationId(UUID.randomUUID().toString()) + .setBasePath(notificationEntity.getBasePath()) + .setResourcePath(notificationEntity.getResourcePath()) + .setAuthToken(notificationEntity.getAuthToken()) + .setEventType(Notification.NotificationType.valueOf(notificationEntity.getResourceType())) + .setHostName(notificationEntity.getHostName()) + .setOccuredTime(notificationEntity.getOccuredTime()) + .setTenantId(notificationEntity.getTenantId()) + .setResourceType(notificationEntity.getResourceType()).build(); + + Configuration configuration = Utils.loadConfig(configPath); + orchestratorEventHandler.init(configuration); + orchestratorEventHandler.invokeMessageFlowForNotification(notification); + responseObserver.onNext(NotificationInvokeResponse.newBuilder().setStatus(true).build()); + responseObserver.onCompleted(); + } + } catch (Exception ex) { + String msg = "Notification invocation failed for id " + request.getNotificationId(); + LOGGER.error(msg, ex); + responseObserver.onError(Status.INTERNAL.withDescription(msg).asRuntimeException()); + } + + } }
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/resources/application.properties b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/resources/application.properties index 88dd260..606cc01 100644 --- a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/resources/application.properties +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-server/src/main/resources/application.properties
@@ -17,4 +17,4 @@ mft.host=localhost mft.port=6565 drms.host=localhost -drms.port=7070 \ No newline at end of file +drms.port=7070
diff --git a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-stub/src/main/proto/notification.proto b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-stub/src/main/proto/notification.proto index 2a019fb..bef56ed 100644 --- a/data-orchestrator/data-orchestrator-service/data-orchestrator-api-stub/src/main/proto/notification.proto +++ b/data-orchestrator/data-orchestrator-service/data-orchestrator-api-stub/src/main/proto/notification.proto
@@ -73,6 +73,14 @@ Notification notification = 1; } +message NotificationInvokeRequest { + string notificationId = 1; +} + +message NotificationInvokeResponse { + bool status = 1; +} + message NotificationRegisterResponse { } @@ -109,6 +117,12 @@ }; } + rpc invokeNotification (NotificationInvokeRequest) returns (NotificationInvokeResponse) { + option (google.api.http) = { + post: "/v1.0/api/dataorch/notification/invoke" + }; + } + rpc listNotifications (NotificationListRequest) returns (NotificationListResponse) { option (google.api.http) = { get: "/v1.0/api/dataorch/notifications"
diff --git a/data-resource-management-service/drms-rdbms-impl/drms-server/pom.xml b/data-resource-management-service/drms-rdbms-impl/drms-server/pom.xml index c8b3080..66c2ab3 100644 --- a/data-resource-management-service/drms-rdbms-impl/drms-server/pom.xml +++ b/data-resource-management-service/drms-rdbms-impl/drms-server/pom.xml
@@ -153,6 +153,17 @@ <artifactId>drms-core</artifactId> <version>${project.version}</version> </dependency> + <dependency> + <groupId>org.apache.airavata.data.lake</groupId> + <artifactId>data-orchestrator-api-stub</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.airavata.data.lake</groupId> + <artifactId>data-orchestrator-clients-core</artifactId> + <version>0.01-SNAPSHOT</version> + <scope>compile</scope> + </dependency> </dependencies>
diff --git a/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/handlers/ResourceServiceHandler.java b/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/handlers/ResourceServiceHandler.java index ddb7438..7b47784 100644 --- a/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/handlers/ResourceServiceHandler.java +++ b/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/handlers/ResourceServiceHandler.java
@@ -22,17 +22,22 @@ import com.google.protobuf.util.JsonFormat; import io.grpc.Status; import io.grpc.stub.StreamObserver; +import org.apache.airavata.datalake.data.orchestrator.api.stub.notification.NotificationInvokeRequest; +import org.apache.airavata.datalake.data.orchestrator.api.stub.notification.NotificationInvokeResponse; import org.apache.airavata.datalake.drms.AuthenticatedUser; import org.apache.airavata.datalake.drms.resource.GenericResource; import org.apache.airavata.datalake.drms.storage.*; +import org.apache.airavata.dataorchestrator.clients.core.NotificationClient; import org.apache.airavata.drms.api.persistance.mapper.ResourceMapper; import org.apache.airavata.drms.api.persistance.mapper.StorageMapper; import org.apache.airavata.drms.api.persistance.model.Resource; import org.apache.airavata.drms.api.persistance.model.ResourceProperty; import org.apache.airavata.drms.api.persistance.model.TransferMapping; +import org.apache.airavata.drms.api.persistance.model.UnverifiedResource; import org.apache.airavata.drms.api.persistance.repository.ResourcePropertyRepository; import org.apache.airavata.drms.api.persistance.repository.ResourceRepository; import org.apache.airavata.drms.api.persistance.repository.TransferMappingRepository; +import org.apache.airavata.drms.api.persistance.repository.UnverifiedResourceRepository; import org.apache.airavata.drms.api.utils.CustosUtils; import org.apache.airavata.drms.core.constants.SharingConstants; import org.apache.airavata.drms.core.constants.StorageConstants; @@ -42,12 +47,15 @@ import org.apache.custos.sharing.core.SearchCondition; import org.apache.custos.sharing.core.SearchCriteria; import org.apache.custos.sharing.management.client.SharingManagementClient; -import org.apache.custos.sharing.service.*; +import org.apache.custos.sharing.service.Entities; +import org.apache.custos.sharing.service.SearchRequest; import org.json.JSONObject; import org.lognet.springboot.grpc.GRpcService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.domain.Page; +import org.springframework.data.domain.PageRequest; import java.io.IOException; import java.util.*; @@ -71,6 +79,15 @@ @Autowired private TransferMappingRepository transferMappingRepository; + @Autowired + private UnverifiedResourceRepository unverifiedResourceRepository; + + @org.springframework.beans.factory.annotation.Value("${orch.host}") + private String orchHost; + + @org.springframework.beans.factory.annotation.Value("${orch.port}") + private int orchPort; + @Override public void fetchResource(ResourceFetchRequest request, StreamObserver<ResourceFetchResponse> responseObserver) { @@ -152,7 +169,7 @@ callUser.getUsername()); if (exEntity.isPresent()) { - Resource resource = ResourceMapper.map(request.getResource(),null, exEntity.get(), callUser); + Resource resource = ResourceMapper.map(request.getResource(), null, exEntity.get(), callUser); resource.setResourceType(type); resource.setParentResourceId(parentId); resourceRepository.save(resource); @@ -256,37 +273,37 @@ String entityId = request.getResource().getResourceId(); String name = request.getResource().getResourceName(); - Optional<Resource> exResource = resourceRepository.findById(entityId); - if (exResource.isPresent()) { - List<ResourceProperty> resourceProperties = resourcePropertyRepository. - findByPropertyKeyAndResourceId("owner",exResource.get().getId()); - if (parentId == null|| parentId.isEmpty()) - parentId = exResource.get().getParentResourceId(); - if (!resourceProperties.isEmpty()) { - Optional<Entity> exEntity = CustosUtils.mergeResourceEntity(custosClientProvider, callUser.getTenantId(), - parentId, type, entityId, - request.getResource().getResourceName(), request.getResource().getResourceName(), - resourceProperties.get(0).getPropertyValue()); + Optional<Resource> exResource = resourceRepository.findById(entityId); + if (exResource.isPresent()) { + List<ResourceProperty> resourceProperties = resourcePropertyRepository. + findByPropertyKeyAndResourceId("owner", exResource.get().getId()); + if (parentId == null || parentId.isEmpty()) + parentId = exResource.get().getParentResourceId(); + if (!resourceProperties.isEmpty()) { + Optional<Entity> exEntity = CustosUtils.mergeResourceEntity(custosClientProvider, callUser.getTenantId(), + parentId, type, entityId, + request.getResource().getResourceName(), request.getResource().getResourceName(), + resourceProperties.get(0).getPropertyValue()); - if (exEntity.isPresent()) { - Resource resource = ResourceMapper.map(request.getResource(), exResource.get(),exEntity.get(), callUser); - resource.setResourceType(type); - resource.setParentResourceId(parentId); + if (exEntity.isPresent()) { + Resource resource = ResourceMapper.map(request.getResource(), exResource.get(), exEntity.get(), callUser); + resource.setResourceType(type); + resource.setParentResourceId(parentId); - resourceRepository.save(resource); + resourceRepository.save(resource); - GenericResource genericResource = ResourceMapper.map(resource, exEntity.get()); + GenericResource genericResource = ResourceMapper.map(resource, exEntity.get()); - ResourceUpdateResponse response = ResourceUpdateResponse - .newBuilder() - .setResource(genericResource) - .build(); - responseObserver.onNext(response); - responseObserver.onCompleted(); - return; - } - } - } + ResourceUpdateResponse response = ResourceUpdateResponse + .newBuilder() + .setResource(genericResource) + .build(); + responseObserver.onNext(response); + responseObserver.onCompleted(); + return; + } + } + } //TODO: Error @@ -307,6 +324,8 @@ request, StreamObserver<ResourceSearchResponse> responseObserver) { AuthenticatedUser callUser = request.getAuthToken().getAuthenticatedUser(); + invokeUnVerifiedResourceRegistrationWorkflow(callUser.getUsername()); + List<ResourceSearchQuery> resourceSearchQueries = request.getQueriesList(); SearchRequest.Builder searchRequestBuilder = SearchRequest.newBuilder(); @@ -341,7 +360,7 @@ Optional<TransferMapping> transferMappingOptional = transferMappingRepository. findTransferMappingByScope(TransferScope.GLOBAL.name()); - if (transferMappingOptional.isPresent() && searchMap.isEmpty() && !type.equalsIgnoreCase("COLLECTION_GROUP")) { + if (transferMappingOptional.isPresent() && searchMap.isEmpty() && !type.equalsIgnoreCase("COLLECTION_GROUP")) { TransferMapping transferMapping = transferMappingOptional.get(); String sourceId = transferMapping.getSource().getId(); @@ -412,34 +431,34 @@ AuthenticatedUser callUser = request.getAuthToken().getAuthenticatedUser(); GenericResource resource = request.getParentResource(); List<GenericResource> childResources = request.getChildResourcesList(); - childResources.forEach(childResource-> { + childResources.forEach(childResource -> { - List<ResourceProperty> resourceProperties = resourcePropertyRepository. - findByPropertyKeyAndResourceId("resourceName",childResource.getResourceId()); - Optional<Resource> exRes = resourceRepository.findById(childResource.getResourceId()); - try{ - if(!resourceProperties.isEmpty() && exRes.isPresent()) { - CustosUtils.mergeResourceEntity(custosClientProvider, callUser.getTenantId(), - resource.getResourceId(), childResource.getType(), childResource.getResourceId(), - resourceProperties.get(0).getPropertyValue(), resourceProperties.get(0).getPropertyValue(), - callUser.getUsername()); - Resource chResource = exRes.get(); - chResource.setParentResourceId(resource.getResourceId()); - resourceRepository.save(chResource); + List<ResourceProperty> resourceProperties = resourcePropertyRepository. + findByPropertyKeyAndResourceId("resourceName", childResource.getResourceId()); + Optional<Resource> exRes = resourceRepository.findById(childResource.getResourceId()); + try { + if (!resourceProperties.isEmpty() && exRes.isPresent()) { + CustosUtils.mergeResourceEntity(custosClientProvider, callUser.getTenantId(), + resource.getResourceId(), childResource.getType(), childResource.getResourceId(), + resourceProperties.get(0).getPropertyValue(), resourceProperties.get(0).getPropertyValue(), + callUser.getUsername()); + Resource chResource = exRes.get(); + chResource.setParentResourceId(resource.getResourceId()); + resourceRepository.save(chResource); - } - } catch (IOException e) { - String msg = " Error occurred while adding child memberships " + e.getMessage(); - logger.error(" Error occurred while adding child memberships: Messages {} ", e.getMessage(), e); } - }); - OperationStatusResponse operationStatusResponse = OperationStatusResponse - .newBuilder(). - setStatus(true) - .build(); - responseObserver.onNext(operationStatusResponse); - responseObserver.onCompleted(); + } catch (IOException e) { + String msg = " Error occurred while adding child memberships " + e.getMessage(); + logger.error(" Error occurred while adding child memberships: Messages {} ", e.getMessage(), e); + } + }); + OperationStatusResponse operationStatusResponse = OperationStatusResponse + .newBuilder(). + setStatus(true) + .build(); + responseObserver.onNext(operationStatusResponse); + responseObserver.onCompleted(); } catch (Exception e) { String msg = " Error occurred while adding child memberships " + e.getMessage(); @@ -457,16 +476,16 @@ AuthenticatedUser callUser = request.getAuthToken().getAuthenticatedUser(); GenericResource resource = request.getParentResource(); List<GenericResource> childResources = request.getChildResourcesList(); - childResources.forEach(childResource-> { - List<ResourceProperty> resourceProperties = resourcePropertyRepository.findByPropertyKeyAndResourceId("resourceName",childResource.getResourceId()); - Optional<Resource> exRes = resourceRepository.findById(childResource.getResourceId()); + childResources.forEach(childResource -> { + List<ResourceProperty> resourceProperties = resourcePropertyRepository.findByPropertyKeyAndResourceId("resourceName", childResource.getResourceId()); + Optional<Resource> exRes = resourceRepository.findById(childResource.getResourceId()); try { - if(!resourceProperties.isEmpty() && exRes.isPresent()) { + if (!resourceProperties.isEmpty() && exRes.isPresent()) { CustosUtils.mergeResourceEntity(custosClientProvider, callUser.getTenantId(), "", childResource.getType(), childResource.getResourceId(), resourceProperties.get(0).getPropertyValue(), resourceProperties.get(0).getPropertyValue(), callUser.getUsername()); - Resource chResource = exRes.get(); + Resource chResource = exRes.get(); chResource.setParentResourceId(null); resourceRepository.save(chResource); } @@ -643,6 +662,92 @@ } + + @Override + public void fetchUnverifiedResources(ResourceSearchRequest request, StreamObserver<ResourceSearchResponse> responseObserver) { + try { + PageRequest pageRequest = PageRequest.of(request.getOffset(), request.getLimit()); + Page<UnverifiedResource> resources = unverifiedResourceRepository.findAll(pageRequest); + + ResourceSearchResponse.Builder response = ResourceSearchResponse.newBuilder(); + + Map<String, String> props = new HashMap<>(); + + + resources.forEach(resource -> { + Map<String, String> prop = new HashMap<>(); + prop.put("ERROR_CODE", resource.getErrorCode()); + prop.put("ERROR_DISCRIPTION", resource.getErrorDiscription()); + response.addResources(GenericResource.newBuilder() + .setResourceId(resource.getId()) + .setResourcePath(resource.getPath()) + .setType(resource.getType()) + .putAllProperties(prop).build()); + }); + + responseObserver.onNext(response.build()); + responseObserver.onCompleted(); + } catch (Exception ex) { + String msg = " Error occurred while fetching unverified resources " + ex.getMessage(); + logger.error(" Error occurred while fetching unverified resources {} ", ex.getMessage(), ex); + responseObserver.onError(Status.INTERNAL.withDescription(msg).asRuntimeException()); + } + + } + + @Override + public void createUnverifiedResource(ResourceCreateRequest request, StreamObserver<ResourceCreateResponse> responseObserver) { + try { + AuthenticatedUser callUser = request.getAuthToken().getAuthenticatedUser(); + String type = request.getResource().getType(); + String entityId = request.getResource().getResourceId(); + String path = request.getResource().getResourcePath(); + String tenantId = request.getAuthToken().getAuthenticatedUser().getTenantId(); + String errorCode = request.getResource().getPropertiesMap().get("ERROR_CODE"); + String errorDescription = request.getResource().getPropertiesMap().get("ERROR_DESCRIPTION"); + + + UnverifiedResource unverifiedResource = new UnverifiedResource(); + unverifiedResource.setId(entityId); + unverifiedResource.setPath(path); + unverifiedResource.setErrorCode(errorCode); + unverifiedResource.setErrorDiscription(errorDescription); + unverifiedResource.setTenantId(tenantId); + unverifiedResource.setType(type); + + if (!callUser.getUsername().isEmpty()) { + unverifiedResource.setUnverifiedAssociatedOwner(callUser.getUsername()); + } + + unverifiedResourceRepository.save(unverifiedResource); + + ResourceCreateResponse response = ResourceCreateResponse + .newBuilder() + .setResource(request.getResource()) + .build(); + responseObserver.onNext(response); + responseObserver.onCompleted(); + return; + + } catch (Exception ex) { + logger.error("Error occurred while creating unverified resource {}", request.getResource().getResourceId(), ex); + String msg = "Error occurred while creating unverified resource" + ex.getMessage(); + responseObserver.onError(Status.INTERNAL.withDescription(msg).asRuntimeException()); + } + } + + @Override + public void deleteUnverifiedResource(ResourceUpdateRequest request, StreamObserver<ResourceUpdateResponse> responseObserver) { + try { + unverifiedResourceRepository.deleteById(request.getResourceId()); + } catch (Exception ex) { + + logger.error("Error occurred while deleting unverified resource {}", request.getResource().getResourceId(), ex); + responseObserver.onError(Status.INTERNAL. + withDescription("Error occurred while deleting unverified resource {}" + request.getResource().getResourceId()).asRuntimeException()); + } + } + private Set<ResourceProperty> mergeProperties(Resource resource, Map<String, Object> values) { Set<ResourceProperty> exProperties = resource.getResourceProperty(); @@ -678,4 +783,30 @@ return newProperties; } + + private void invokeUnVerifiedResourceRegistrationWorkflow(String username) { + List<UnverifiedResource> unverifiedResources = unverifiedResourceRepository + .getUnverifiedResourceByUnverifiedAssociatedOwnerAndErrorCode(username, "ERR_0002"); + + NotificationClient notificationClient = new NotificationClient( + orchHost, orchPort); + + if (!unverifiedResources.isEmpty()) { + + for (UnverifiedResource unverifiedResource : unverifiedResources) { + + NotificationInvokeResponse response = notificationClient.get() + .invokeNotification(NotificationInvokeRequest + .newBuilder() + .setNotificationId(unverifiedResource.getId()) + .build()); + if (response.getStatus()) { + unverifiedResourceRepository.deleteById(unverifiedResource.getId()); + } + + } + } + + } + }
diff --git a/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/interceptors/Authenticator.java b/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/interceptors/Authenticator.java index aa3007f..beb9564 100644 --- a/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/interceptors/Authenticator.java +++ b/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/interceptors/Authenticator.java
@@ -120,7 +120,7 @@ String tenantId = drmsServiceAuthToken.getAuthenticatedUser().getTenantId(); Struct struct = identityManagementClient .getAgentToken(tenantId, agentClientId, agentClientSec, "client_credentials", ""); - if (struct.getFieldsMap().get("access_token").isInitialized()) { + if (struct.getFieldsMap().get("access_token").isInitialized() && ! drmsServiceAuthToken.getUserUnverified()) { UserRepresentation user = userManagementClient.getUser(username, tenantId); return Optional.ofNullable(AuthenticatedUser.newBuilder() .setUsername(user.getUsername()) @@ -129,6 +129,14 @@ .setEmailAddress(user.getEmail()) .setTenantId(tenantId) .build()); + } else { + AuthenticatedUser user = AuthenticatedUser.newBuilder() + .setTenantId(tenantId) + .build(); + if (username == null || username.isEmpty()) { + user = user.toBuilder().setUsername(username).build(); + } + return Optional.ofNullable(user); } } } else {
diff --git a/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/persistance/model/UnverifiedResource.java b/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/persistance/model/UnverifiedResource.java new file mode 100644 index 0000000..b7b44ae --- /dev/null +++ b/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/persistance/model/UnverifiedResource.java
@@ -0,0 +1,91 @@ +package org.apache.airavata.drms.api.persistance.model; + + +import org.springframework.data.jpa.domain.support.AuditingEntityListener; + +import javax.persistence.*; + +@Entity +@Table(name = "UNVERIFIED_RESOURCE") +@EntityListeners(AuditingEntityListener.class) +public class UnverifiedResource { + + @Id + @Column(name="ID") + private String id; + + @Column(name="TENANT_ID",nullable = false) + private String tenantId; + + @Column(name="RESOURCE_PATH") + private String path; + + @Column(name="ERROR_CODE") + private String errorCode; + + @Column(name="ERROR_DISCRIPTION") + private String errorDiscription; + + @Column(name="TYPE") + private String type; + + @Column(name="UNVERIFIED_OWNER") + private String unverifiedAssociatedOwner; + + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getTenantId() { + return tenantId; + } + + public void setTenantId(String tenantId) { + this.tenantId = tenantId; + } + + public String getPath() { + return path; + } + + public void setPath(String path) { + this.path = path; + } + + public String getErrorCode() { + return errorCode; + } + + public void setErrorCode(String errorCode) { + this.errorCode = errorCode; + } + + public String getErrorDiscription() { + return errorDiscription; + } + + public void setErrorDiscription(String errorDiscription) { + this.errorDiscription = errorDiscription; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public String getUnverifiedAssociatedOwner() { + return unverifiedAssociatedOwner; + } + + public void setUnverifiedAssociatedOwner(String unverifiedAssociatedOwner) { + this.unverifiedAssociatedOwner = unverifiedAssociatedOwner; + } +}
diff --git a/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/persistance/repository/UnverifiedResourceRepository.java b/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/persistance/repository/UnverifiedResourceRepository.java new file mode 100644 index 0000000..2ad5877 --- /dev/null +++ b/data-resource-management-service/drms-rdbms-impl/drms-server/src/main/java/org/apache/airavata/drms/api/persistance/repository/UnverifiedResourceRepository.java
@@ -0,0 +1,15 @@ +package org.apache.airavata.drms.api.persistance.repository; + +import org.apache.airavata.drms.api.persistance.model.Resource; +import org.apache.airavata.drms.api.persistance.model.UnverifiedResource; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Query; + +import java.util.List; + +public interface UnverifiedResourceRepository extends JpaRepository<UnverifiedResource, String> { + + + public List<UnverifiedResource> getUnverifiedResourceByUnverifiedAssociatedOwnerAndErrorCode(String username,String errorCode); + +}
diff --git a/data-resource-management-service/drms-stubs/src/main/proto/Common.proto b/data-resource-management-service/drms-stubs/src/main/proto/Common.proto index 604c1fd..6bcfeea 100644 --- a/data-resource-management-service/drms-stubs/src/main/proto/Common.proto +++ b/data-resource-management-service/drms-stubs/src/main/proto/Common.proto
@@ -35,6 +35,7 @@ string access_token = 1; AuthenticatedUser authenticated_user = 2; AuthCredentialType auth_credential_type =3; + bool user_unverified =4; } enum AuthCredentialType {
diff --git a/data-resource-management-service/drms-stubs/src/main/proto/resource/DRMSResource.proto b/data-resource-management-service/drms-stubs/src/main/proto/resource/DRMSResource.proto index c696cb0..b6eb64d 100644 --- a/data-resource-management-service/drms-stubs/src/main/proto/resource/DRMSResource.proto +++ b/data-resource-management-service/drms-stubs/src/main/proto/resource/DRMSResource.proto
@@ -36,4 +36,5 @@ string parent_resource_path = 7; string resource_name = 8; string parent_id = 9; + } \ No newline at end of file
diff --git a/data-resource-management-service/drms-stubs/src/main/proto/resource/DRMSResourceService.proto b/data-resource-management-service/drms-stubs/src/main/proto/resource/DRMSResourceService.proto index f74f21e..2d886bc 100644 --- a/data-resource-management-service/drms-stubs/src/main/proto/resource/DRMSResourceService.proto +++ b/data-resource-management-service/drms-stubs/src/main/proto/resource/DRMSResourceService.proto
@@ -86,6 +86,8 @@ repeated ResourceSearchQuery queries = 2; int32 depth = 4; string type = 5; + int32 offset=6; + int32 limit=7; } message ResourceSearchResponse { @@ -222,5 +224,23 @@ }; } + rpc fetchUnverifiedResources (ResourceSearchRequest) returns (ResourceSearchResponse) { + option (google.api.http) = { + get: "/v1.0/api/drms/resources/unverified" + }; + } + + rpc createUnverifiedResource (ResourceCreateRequest) returns (ResourceCreateResponse) { + option (google.api.http) = { + post: "/v1.0/api/drms/resource/unverified" + }; + } + + rpc deleteUnverifiedResource (ResourceUpdateRequest) returns (ResourceUpdateResponse) { + option (google.api.http) = { + delete: "/v1.0/api/drms/resource/unverified" + }; + } + } \ No newline at end of file