support to execute bml2linkis node type.

Author: wushengyeyouya

dss-orchestrator/orchestrators/dss-workflow/dss-linkis-node-execution/src/main/java/com/webank/wedatasphere/dss/linkis/node/execution/parser/BML2LinkisJobParser.java

@@ -0,0 +1,55 @@
+package com.webank.wedatasphere.dss.linkis.node.execution.parser;
+
+import com.google.gson.reflect.TypeToken;
+import com.webank.wedatasphere.dss.linkis.node.execution.entity.BMLResource;
+import com.webank.wedatasphere.dss.linkis.node.execution.exception.LinkisJobExecutionErrorException;
+import com.webank.wedatasphere.dss.linkis.node.execution.job.AppConnLinkisJob;
+import com.webank.wedatasphere.dss.linkis.node.execution.job.Job;
+import com.webank.wedatasphere.dss.linkis.node.execution.utils.LinkisJobExecutionUtils;
+
+import java.util.Map;
+
+public class BML2LinkisJobParser extends CodeParser {
+
+    @Override
+    public void parseJob(Job job) throws Exception {
+        if(!(job instanceof AppConnLinkisJob)) {
+            return;
+        }
+        AppConnLinkisJob appConnLinkisJob = (AppConnLinkisJob) job;
+        String runType = appConnLinkisJob.getRunType();
+        // 只处理包含 bml2linkis 的 AppConnLinkisJob，例如：linkis.appconn.<appconnName>.bml2linkis
+        if(!runType.toLowerCase().contains("bml2linkis")) {
+            return;
+        }
+        job.getLogObj().info(String.format("AppConn %s try to generate Linkis jobContent from code %s.", runType,
+                job.getCode()));
+        Map<String, Object> script = LinkisJobExecutionUtils.gson.fromJson(job.getCode(), new TypeToken<Map<String, Object>>() {}.getType());
+        if(!script.containsKey("resourceId") || !script.containsKey("version") || !script.containsKey("fileName")) {
+            job.getLogObj().error("the code error, resourceId, version or fileName is not exists.");
+            throw new LinkisJobExecutionErrorException(90100, "cannot recognize fileName from jobContent.");
+        }
+        BMLResource bmlResource = new BMLResource();
+        bmlResource.setResourceId((String) script.get("resourceId"));
+        bmlResource.setVersion((String) script.get("version"));
+        // fileName 的格式为 ${resourceId}.${engineType}.${runType}
+        bmlResource.setFileName((String) script.get("fileName"));
+        getAndSetCode(bmlResource, appConnLinkisJob);
+        String[] fileNameArray = bmlResource.getFileName().split("\\.");
+        if(fileNameArray.length < 3) {
+            job.getLogObj().error(String.format("cannot recognize fileName %s, the fileName format must be ${resourceId}.${engineType}.${runType}", bmlResource.getFileName()));
+            throw new LinkisJobExecutionErrorException(90100, "cannot recognize fileName from jobContent.");
+        }
+        String realEngineType = fileNameArray[fileNameArray.length - 2];
+        String realRunType = fileNameArray[fileNameArray.length - 1];
+        setEngineType(job, realEngineType, realRunType);
+    }
+
+    protected void setEngineType(Job job, String realEngineType, String realRunType) {
+        job.getLogObj().warn(String.format("switch job from engineType %s with runType %s to engineType %s with runType %s",
+                job.getEngineType(), job.getRunType(), realEngineType, realRunType));
+        job.setEngineType(realEngineType);
+        job.setRunType(realRunType);
+    }
+
+}

dss-orchestrator/orchestrators/dss-workflow/dss-linkis-node-execution/src/main/java/com/webank/wedatasphere/dss/linkis/node/execution/parser/CodeParser.java

@@ -22,20 +22,17 @@
 import com.webank.wedatasphere.dss.linkis.node.execution.exception.LinkisJobExecutionErrorException;
 import com.webank.wedatasphere.dss.linkis.node.execution.job.CommonLinkisJob;
 import com.webank.wedatasphere.dss.linkis.node.execution.job.Job;
+import com.webank.wedatasphere.dss.linkis.node.execution.job.LinkisJob;
 import com.webank.wedatasphere.dss.linkis.node.execution.service.LinkisURLService;
 import com.webank.wedatasphere.dss.linkis.node.execution.utils.LinkisJobExecutionUtils;
 import org.apache.linkis.filesystem.WorkspaceClientFactory;
 import org.apache.linkis.filesystem.request.WorkspaceClient;
 import org.apache.linkis.filesystem.response.ScriptFromBMLResponse;
 
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
-import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import org.apache.commons.lang3.StringUtils;
 
 
 public class CodeParser implements JobParser {
@@ -44,9 +41,10 @@ public class CodeParser implements JobParser {
     private volatile WorkspaceClient client1X = null;
     private volatile WorkspaceClient client0X = null;
     private final Object clientLocker = new Object();
+
     @Override
     public void parseJob(Job job) throws Exception{
-        if (! ( job instanceof CommonLinkisJob) ) {
+        if (! (job instanceof CommonLinkisJob) ) {
             return ;
         }
         CommonLinkisJob linkisAppConnJob  = (CommonLinkisJob) job;
@@ -66,8 +64,11 @@ public void parseJob(Job job) throws Exception{
         if(null == scriptResource) {
             throw  new LinkisJobExecutionErrorException(90102,"Failed to get script resource");
         }
+        getAndSetCode(scriptResource, linkisAppConnJob);
+    }
 
-        Map<String, Object> executionParams = getExecutionParams(scriptResource, linkisAppConnJob);
+    protected void getAndSetCode(BMLResource bmlResource,  LinkisJob linkisAppConnJob) {
+        Map<String, Object> executionParams = getExecutionParams(bmlResource, linkisAppConnJob);
         if (executionParams.get("executionCode") != null) {
             String executionCode = (String) executionParams.get("executionCode");
             linkisAppConnJob.getLogObj().info("************************************SUBMIT CODE************************************");
@@ -81,9 +82,12 @@ public void parseJob(Job job) throws Exception{
                 linkisAppConnJob.getParams().putAll( (Map<String, Object>)executionParams.get("params"));
             }
         }
+        dealExecutionParams(linkisAppConnJob, executionParams);
     }
 
-    private Map<String, Object> getExecutionParams(BMLResource bmlResource,  CommonLinkisJob linkisAppConnJob) {
+    protected void dealExecutionParams(LinkisJob linkisAppConnJob, Map<String, Object> executionParams) {}
+
+    protected Map<String, Object> getExecutionParams(BMLResource bmlResource,  LinkisJob linkisAppConnJob) {
         Map<String, Object> map = new HashMap<>();
         ScriptFromBMLResponse response = getOrCreateWorkSpaceClient(linkisAppConnJob).requestOpenScriptFromBML(bmlResource.getResourceId(), bmlResource.getVersion(), bmlResource.getFileName());
         linkisAppConnJob.getLogObj().info("Get execution code from workspace client,bml resource id "+bmlResource.getResourceId()+", version is "+bmlResource.getVersion());
@@ -92,7 +96,7 @@ private Map<String, Object> getExecutionParams(BMLResource bmlResource,  CommonL
         return map;
     }
 
-    private WorkspaceClient getOrCreateWorkSpaceClient(CommonLinkisJob linkisAppConnJob) {
+    private WorkspaceClient getOrCreateWorkSpaceClient(LinkisJob linkisAppConnJob) {
         Map<String, String> props = linkisAppConnJob.getJobProps();
         if(LinkisJobExecutionConfiguration.isLinkis1_X(props)) {
             if (null == client1X) {
@@ -119,131 +123,4 @@ private WorkspaceClient getOrCreateWorkSpaceClient(CommonLinkisJob linkisAppConn
         }
     }
 
-    private   ArrayList<String> getResourceNames(String code){
-        ArrayList<String> bmlResourceNames = new ArrayList<String>();
-        Matcher mb = pb.matcher(code);
-        while (mb.find()) {
-            bmlResourceNames.add(mb.group().trim());
-        }
-        return bmlResourceNames;
-    }
-
-
-    /**
-     * 1.Find the project file used in the script
-     * 2.Find the node file used in the script
-     * 3.Recursively find the flow file used in the script
-     * 4.Replace file name with prefixed name
-     * @param resourceNames
-     * @param linkisAppConnJob
-     * @return
-     */
-    private  ArrayList<BMLResource> getResourcesByNames(ArrayList<String> resourceNames, CommonLinkisJob linkisAppConnJob) {
-
-        ArrayList<BMLResource> bmlResourceArrayList = new ArrayList<>();
-
-        String jobName = linkisAppConnJob.getJobName();
-        String flowName = linkisAppConnJob.getSource().get("flowName");
-        String projectName = linkisAppConnJob.getSource().get("projectName");
-
-
-        List<BMLResource> projectResourceList = linkisAppConnJob.getProjectResourceList();
-
-
-        List<BMLResource> jobResourceList = linkisAppConnJob.getJobResourceList();
-        for (String resourceName : resourceNames) {
-            String[] resourceNameSplit = resourceName.split("://");
-            String prefix = resourceNameSplit[0].toLowerCase();
-            String fileName = resourceNameSplit[1];
-            BMLResource resource = null;
-            String afterFileName = fileName;
-            switch (prefix) {
-                case "project":
-                    resource = findResource(projectResourceList, fileName);
-                    afterFileName = LinkisJobExecutionConfiguration.PROJECT_PREFIX + "_" + projectName + "_" + fileName;
-                    break;
-                case "flow":
-                    resource = findFlowResource(linkisAppConnJob, fileName, flowName);
-                    break;
-                case "node":
-                    resource = findResource(jobResourceList, fileName);
-                    afterFileName = LinkisJobExecutionConfiguration.JOB_PREFIX + "_" + jobName + "_" + fileName;
-                    break;
-                default:
-            }
-            if (null == resource) {
-                linkisAppConnJob.getLogObj().error("Failed to find the " + prefix + " resource file of " + fileName);
-                throw new RuntimeException("Failed to find the " + prefix + " resource file of " + fileName);
-            }
-            if (!afterFileName.equals(fileName)) {
-                resource.setFileName(afterFileName);
-            }
-            bmlResourceArrayList.add(resource);
-        }
-        return bmlResourceArrayList;
-    }
-
-
-    /**
-     * Recursively find the flow file used in the script
-     * Recursive exit condition is top-level flow
-     *
-     */
-    private  BMLResource findFlowResource(CommonLinkisJob linkisAppConnJob, String fileName, String flowName) {
-
-        String fullFlowName = "";
-        Map<String, List<BMLResource>> fLowNameAndResources = linkisAppConnJob.getFlowNameAndResources();
-        if (fLowNameAndResources == null){
-            return null;
-        }
-        Optional<Map.Entry<String, List<BMLResource>>> first = fLowNameAndResources.entrySet().stream().filter(fLowNameAndResource -> fLowNameAndResource.getKey().endsWith(flowName + LinkisJobExecutionConfiguration.RESOURCES_NAME)).findFirst();
-
-        if(first.isPresent()){
-            fullFlowName = first.get().getKey();
-            BMLResource resource = findResource(first.get().getValue(), fileName);
-            if (resource != null) {
-                resource.setFileName(flowName + "_" + fileName);
-                return resource;
-            }
-        }
-
-        String firstFlow = "flow." + flowName + LinkisJobExecutionConfiguration.RESOURCES_NAME;
-        if (firstFlow.equals(fullFlowName)) {
-            return null;
-        }
-        //getParentFlowName:flow.flows1.test.resources  return:flows1
-        String parentFlowName = StringUtils.substringAfterLast(StringUtils.substringBefore(fullFlowName, "." + flowName
-                + LinkisJobExecutionConfiguration.RESOURCES_NAME), ".");
-        if (StringUtils.isEmpty(parentFlowName)) {
-            return null;
-        }
-
-        return findFlowResource(linkisAppConnJob, fileName, parentFlowName);
-    }
-
-
-    private  String replaceCodeResourceNames(String code, ArrayList<String> resourceNameList, ArrayList<BMLResource> resourceList){
-        if(resourceList.size() != resourceNameList.size()){
-            throw new RuntimeException("Failed to parsed resource file");
-        }
-
-        String[] names = resourceNameList.toArray(new String[]{});
-
-        String[] afterNames = new String[resourceList.size()];
-        for (int i=0 ; i < afterNames.length ; i++){
-            afterNames[i] = resourceList.get(i).getFileName();
-        }
-        return StringUtils.replaceEach(code, names, afterNames);
-    }
-
-    private   BMLResource findResource(List<BMLResource> resourceArrayList, String fileName){
-        if(resourceArrayList != null && !resourceArrayList.isEmpty()) {
-            for(BMLResource resource : resourceArrayList){
-                if(resource.getFileName().equals(fileName)){
-                    return resource;
-                }
-            }
-        }
-        return null;
-    }
 }