Adapt to create workflow when have not scheduler.

Author: zqburde

dss-framework/dss-framework-orchestrator-server/src/main/java/com/webank/wedatasphere/dss/orchestrator/server/service/impl/OrchestratorFrameworkServiceImpl.java

@@ -16,6 +16,7 @@
 
 package com.webank.wedatasphere.dss.orchestrator.server.service.impl;
 
+import com.webank.wedatasphere.dss.appconn.core.AppConn;
 import com.webank.wedatasphere.dss.appconn.scheduler.SchedulerAppConn;
 import com.webank.wedatasphere.dss.appconn.scheduler.structure.orchestration.OrchestrationCreationOperation;
 import com.webank.wedatasphere.dss.appconn.scheduler.structure.orchestration.OrchestrationDeletionOperation;
@@ -41,6 +42,7 @@
 import com.webank.wedatasphere.dss.orchestrator.core.type.DSSOrchestratorRelationManager;
 import com.webank.wedatasphere.dss.orchestrator.core.utils.OrchestratorUtils;
 import com.webank.wedatasphere.dss.orchestrator.db.dao.OrchestratorMapper;
+import com.webank.wedatasphere.dss.orchestrator.loader.LinkedAppConnResolver;
 import com.webank.wedatasphere.dss.orchestrator.loader.OrchestratorManager;
 import com.webank.wedatasphere.dss.orchestrator.server.entity.request.OrchestratorCreateRequest;
 import com.webank.wedatasphere.dss.orchestrator.server.entity.request.OrchestratorDeleteRequest;
@@ -64,6 +66,7 @@
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.Optional;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -79,7 +82,8 @@ public class OrchestratorFrameworkServiceImpl implements OrchestratorFrameworkSe
     private OrchestratorService newOrchestratorService;
     @Autowired
     private OrchestratorManager orchestratorManager;
-
+    @Autowired
+    private LinkedAppConnResolver linkedAppConnResolver;
     /**
      * 1.拿到的dss orchestrator的appconn
      * 2.然后创建
@@ -145,7 +149,7 @@ private <K extends StructureRequestRef, V extends ResponseRef> V tryOrchestratio
         Long refProjectId, refOrchestrationId;
         if (askProjectSender) {
             ProjectRefIdResponse projectRefIdResponse = (ProjectRefIdResponse) DSSSenderServiceFactory.getOrCreateServiceInstance().getProjectServerSender()
-                    .ask(new ProjectRefIdRequest(orchestrationPair.getValue().getId(), dssOrchestrator.getProjectId()));
+                    .ask(new ProjectRefIdRequest(Optional.ofNullable(orchestrationPair).map(ImmutablePair::getValue).map(AppInstance::getId).orElse(null), dssOrchestrator.getProjectId()));
             refProjectId = projectRefIdResponse.getRefProjectId();
             refOrchestrationId = null;
         } else {
@@ -244,11 +248,16 @@ protected ImmutablePair<OrchestrationService, AppInstance> getOrchestrationServi
             dssOrchestratorInfo.setLinkedAppConnNames(dssOrchestrator.getLinkedAppConn().stream().map(appConn -> appConn.getAppDesc().getAppName()).collect(Collectors.toList()));
         }
         SchedulerAppConn appConn = dssOrchestrator.getSchedulerAppConn();
-        if (appConn == null) {
-            throw new ExternalOperationWarnException(50322, "DSSOrchestrator " + dssOrchestrator.getName() + " has no SchedulerAppConn.");
+//        if (appConn == null) {
+//            throw new ExternalOperationWarnException(50322, "DSSOrchestrator " + dssOrchestrator.getName() + " has no SchedulerAppConn.");
+//        }
+        if (appConn != null) {
+            AppInstance appInstance = appConn.getAppDesc().getAppInstances().get(0);
+            return new ImmutablePair<>(appConn.getOrCreateStructureStandard().getOrchestrationService(appInstance), appInstance);
+        } else {
+            return new ImmutablePair<>(null, null);
         }
-        AppInstance appInstance = appConn.getAppDesc().getAppInstances().get(0);
-        return new ImmutablePair<>(appConn.getOrCreateStructureStandard().getOrchestrationService(appInstance), appInstance);
+
     }
 
     /**

dss-framework/dss-framework-orchestrator-server/src/main/java/com/webank/wedatasphere/dss/orchestrator/server/service/impl/OrchestratorServiceImpl.java

@@ -16,6 +16,7 @@
 
 package com.webank.wedatasphere.dss.orchestrator.server.service.impl;
 
+import com.webank.wedatasphere.dss.appconn.core.AppConn;
 import com.webank.wedatasphere.dss.common.constant.project.ProjectUserPrivEnum;
 import com.webank.wedatasphere.dss.common.exception.DSSErrorException;
 import com.webank.wedatasphere.dss.common.label.DSSLabel;
@@ -51,6 +52,7 @@
 import com.webank.wedatasphere.dss.standard.app.development.service.RefQueryService;
 import com.webank.wedatasphere.dss.standard.app.development.standard.DevelopmentIntegrationStandard;
 import com.webank.wedatasphere.dss.standard.app.sso.Workspace;
+import com.webank.wedatasphere.dss.standard.common.desc.AppDesc;
 import com.webank.wedatasphere.dss.standard.common.desc.AppInstance;
 import com.webank.wedatasphere.dss.standard.common.entity.ref.ResponseRef;
 import com.webank.wedatasphere.dss.standard.common.exception.operation.ExternalOperationWarnException;
@@ -110,7 +112,9 @@ public OrchestratorVo createOrchestrator(String userName,
                     Map<String, Object> dssJobContent = MapUtils.newCommonMapBuilder()
                             .put(OrchestratorRefConstant.DSS_ORCHESTRATOR_INFO_KEY, dssOrchestratorInfo)
                             .put(OrchestratorRefConstant.ORCHESTRATOR_VERSION_KEY, version)
-                            .put(OrchestratorRefConstant.ORCHESTRATION_SCHEDULER_APP_CONN, dssOrchestrator.getSchedulerAppConn().getAppDesc().getAppName()).build();
+                            .put(OrchestratorRefConstant.ORCHESTRATION_SCHEDULER_APP_CONN, Optional.ofNullable(dssOrchestrator)
+                                    .map(DSSOrchestrator::getSchedulerAppConn).map(AppConn::getAppDesc).map(AppDesc::getAppName)
+                                    .map(Object::toString).orElse("NULL")).build();
                     DSSJobContentRequestRef requestRef = (DSSJobContentRequestRef) developmentRequestRef;
                     requestRef.setDSSJobContent(dssJobContent);
                     return ((RefCreationOperation) developmentOperation).createRef(requestRef);

dss-orchestrator/dss-orchestrator-loader/src/main/java/com/webank/wedatasphere/dss/orchestrator/loader/DefaultOrchestratorLoader.java

@@ -63,17 +63,19 @@ protected DSSOrchestratorContext createOrchestratorContext() {
         appConnList.stream().filter(relation.isLinkedAppConn()).forEach(dssOrchestrator::addLinkedAppConn);
         AppConn appConn = AppConnManager.getAppConnManager().getAppConn(relation.getBindingAppConnName());
         dssOrchestrator.setAppConn(appConn);
-        List<SchedulerAppConn> schedulerAppConns = AppConnManager.getAppConnManager().listAppConns(SchedulerAppConn.class);
-        SchedulerAppConn schedulerAppConn;
-        if(StringUtils.isBlank(relation.getBindingSchedulerAppConnName())) {
-            schedulerAppConn = schedulerAppConns.get(0);
-        } else {
-            schedulerAppConn = schedulerAppConns.stream().filter(appConn1 -> appConn1.getAppDesc().getAppName().equals(relation.getBindingSchedulerAppConnName()))
-                    .findAny().orElseThrow(() -> new ExternalOperationFailedException(50032, "cannot find the matched SchedulerAppConn with name " + relation.getBindingSchedulerAppConnName()));
+        if (appConnList.stream().anyMatch(t -> t instanceof SchedulerAppConn)) {
+            List<SchedulerAppConn> schedulerAppConns = AppConnManager.getAppConnManager().listAppConns(SchedulerAppConn.class);
+            SchedulerAppConn schedulerAppConn;
+            if(StringUtils.isBlank(relation.getBindingSchedulerAppConnName())) {
+                schedulerAppConn = schedulerAppConns.get(0);
+            } else {
+                schedulerAppConn = schedulerAppConns.stream().filter(appConn1 -> appConn1.getAppDesc().getAppName().equals(relation.getBindingSchedulerAppConnName()))
+                        .findAny().orElseThrow(() -> new ExternalOperationFailedException(50032, "cannot find the matched SchedulerAppConn with name " + relation.getBindingSchedulerAppConnName()));
+            }
+            dssOrchestrator.setSchedulerAppConn(schedulerAppConn);
+            LOGGER.info("Load dss orchestrator: {}, with the binding AppConn: {} and binding SchedulerAppConn {}.",
+                    typeName, relation.getBindingAppConnName(), schedulerAppConn.getAppDesc().getAppName());
         }
-        dssOrchestrator.setSchedulerAppConn(schedulerAppConn);
-        LOGGER.info("Load dss orchestrator: {}, with the binding AppConn: {} and binding SchedulerAppConn {}.",
-                typeName, relation.getBindingAppConnName(), schedulerAppConn.getAppDesc().getAppName());
         dssLabels.forEach(dssOrchestrator::addLinkedDssLabels);
         return dssOrchestrator;
     }