Skip to content

Commit

Permalink
fix: saveFlow3
Browse files Browse the repository at this point in the history
  • Loading branch information
@qweqnotallow
qweqnotallow committed Sep 18, 2024
1 parent 9a2c0f0 commit d6c4c87
Show file tree
Hide file tree
Showing 7 changed files with 37 additions and 26 deletions.
2 changes: 1 addition & 1 deletion ...rc/main/java/com/webank/wedatasphere/dss/orchestrator/server/job/OrchestratorCopyJob.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ private void doOrchestratorCopy(String userName,
if (dssLabels.get(0).getValue().get(EnvDSSLabel.DSS_ENV_LABEL_KEY).equals("dev")) {
DSSFlow flowByID = orchestratorCopyEnv.getFlowService().getFlow(flowId);
if (flowByID != null) {
orchestratorCopyEnv.getFlowService().saveFlowMetaData(flowId, flowByID.getFlowJson(), orchestratorId, appId);
orchestratorCopyEnv.getFlowService().saveFlowMetaData(flowId, flowByID.getFlowJson(), orchestratorId);
}
if (targetProject != null && targetProject.getAssociateGit() && flowByID != null) {
orchestratorCopyEnv.getFlowService().updateTOSaveStatus(targetProject.getId(), flowByID.getId(), orchestratorId);
Expand Down
3 changes: 2 additions & 1 deletion .../webank/wedatasphere/dss/orchestrator/server/restful/DSSFrameworkOrchestratorRestful.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -301,9 +301,10 @@ public Message rollbackOrchestrator(HttpServletRequest request, @RequestBody Rol
LabelRouteVO labels = rollbackOrchestratorRequest.getLabels();
try {
LOGGER.info("user {} begin to rollbackOrchestrator, params:{}", username, rollbackOrchestratorRequest);
DSSOrchestratorVersion latestVersionById = orchestratorMapper.getLatestOrchestratorVersionById(orchestratorId);
OrchestratorRollBackGitVo rollbackOrchestrator = orchestratorService.rollbackOrchestrator(username, projectId, projectName, orchestratorId, version, labels, workspace);
try {
orchestratorService.rollbackOrchestratorGit(rollbackOrchestrator, username, projectId, projectName, orchestratorId, labels, workspace);
orchestratorService.rollbackOrchestratorGit(rollbackOrchestrator, username, projectId, projectName, orchestratorId, labels, workspace, latestVersionById.getAppId());
} catch (Exception e) {
return Message.ok("回滚版本成功,git回滚失败，请重新保存并提交工作流").data("newVersion", rollbackOrchestrator.getVersion());
}
Expand Down
2 changes: 1 addition & 1 deletion ...ain/java/com/webank/wedatasphere/dss/orchestrator/server/service/OrchestratorService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ OrchestratorRollBackGitVo rollbackOrchestrator(String username, Long projectId,
Long orchestratorId, String version, LabelRouteVO labels, Workspace workspace) throws Exception;

void rollbackOrchestratorGit(OrchestratorRollBackGitVo rollBackGitVo, String userName, Long projectId, String projectName,
Long orchestratorId, LabelRouteVO labels, Workspace workspace) throws Exception;
Long orchestratorId, LabelRouteVO labels, Workspace workspace, Long latestAppId) throws Exception;
//**** new method
void isExistSameNameBeforeCreate(Long workspaceId, Long projectId, String orchestratorName) throws DSSFrameworkErrorException;

Expand Down
5 changes: 2 additions & 3 deletions ...com/webank/wedatasphere/dss/orchestrator/server/service/impl/OrchestratorServiceImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -437,7 +437,7 @@ public OrchestratorRollBackGitVo rollbackOrchestrator(String userName, Long proj

@Override
public void rollbackOrchestratorGit(OrchestratorRollBackGitVo rollBackGitVo, String userName, Long projectId, String projectName,
Long orchestratorId, LabelRouteVO labels, Workspace workspace) throws Exception {
Long orchestratorId, LabelRouteVO labels, Workspace workspace, Long latestAppId) throws Exception {
if (rollBackGitVo == null) {
return;
}
Expand All @@ -451,8 +451,7 @@ public void rollbackOrchestratorGit(OrchestratorRollBackGitVo rollBackGitVo, Str
orchestratorMapper.updateOrchestratorBmlVersion(orchestratorId, null, null);
}
if (labels.getRoute().equals("dev")) {
DSSOrchestratorVersion oldOrcVersion = rollBackGitVo.getOldOrcVersion();
flowService.saveFlowMetaData(dssFlow.getId(), dssFlow.getFlowJson(), orchestratorId, oldOrcVersion.getAppId());
flowService.saveFlowMetaData(dssFlow.getId(), dssFlow.getFlowJson(), orchestratorId);
}

}
Expand Down
2 changes: 1 addition & 1 deletion ...om/webank/wedatasphere/dss/orchestrator/publish/impl/ImportDSSOrchestratorPluginImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -230,7 +230,7 @@ public DSSOrchestratorVersion importOrchestratorNew(RequestImportOrchestrator re
Long versionOrchestratorId = dssOrchestratorVersion.getOrchestratorId();
DSSFlow flowByID = dssFlowService.getFlow(appId);
if (flowByID != null) {
dssFlowService.saveFlowMetaData(appId, flowByID.getFlowJson(), versionOrchestratorId, oldAppId);
dssFlowService.saveFlowMetaData(appId, flowByID.getFlowJson(), versionOrchestratorId);
}
DSSProject projectInfo = DSSFlowEditLockManager.getProjectInfo(projectId);
if (projectInfo.getAssociateGit()) {
Expand Down
2 changes: 1 addition & 1 deletion ...low-server/src/main/java/com/webank/wedatasphere/dss/workflow/service/DSSFlowService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ String saveFlow(Long flowId,

void updateTOSaveStatus(Long projectId, Long flowID, Long orchestratorId) throws Exception;

void saveFlowMetaData(Long flowID, String jsonFlow, Long orchestratorId, Long oldFLowId);
void saveFlowMetaData(Long flowID, String jsonFlow, Long orchestratorId);

DSSFlow copyRootFlow(Long rootFlowId, String userName, Workspace workspace,
String projectName, String version, String contextIdStr,
Expand Down
47 changes: 29 additions & 18 deletions ...r/src/main/java/com/webank/wedatasphere/dss/workflow/service/impl/DSSFlowServiceImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -340,7 +340,7 @@ public String saveFlow(Long flowID,
Long orchestratorId = dssOrchestratorVersion != null ? dssOrchestratorVersion.getOrchestratorId() : null;
try {
if (orchestratorId != null) {
saveFlowMetaData(flowID, jsonFlow, orchestratorId, null);
saveFlowMetaData(flowID, jsonFlow, orchestratorId);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
Expand Down Expand Up @@ -426,16 +426,16 @@ private Long getRootFlowId(Long flowId) {
}

@Override
public void saveFlowMetaData(Long flowID, String jsonFlow, Long orchestratorId, Long oldFLowId) {
// flowId发生变化时，清空旧的flowId
if (oldFLowId != null) {
List<NodeContentDO> contentListByOrchestratorId = nodeContentMapper.getContentListByOrchestratorId(orchestratorId, oldFLowId);
if (CollectionUtils.isNotEmpty(contentListByOrchestratorId)) {
List<Long> collect = contentListByOrchestratorId.stream().map(NodeContentDO::getId).collect(Collectors.toList());
nodeContentUIMapper.deleteNodeContentUIByContentList(collect);
}
nodeContentMapper.deleteNodeContentByOrchestratorId(orchestratorId, oldFLowId);
}
public void saveFlowMetaData(Long flowID, String jsonFlow, Long orchestratorId) {
// // flowId发生变化时，清空旧的flowId
// if (oldFLowId != null) {
// List<NodeContentDO> contentListByOrchestratorId = nodeContentMapper.getContentListByOrchestratorId(orchestratorId, oldFLowId);
// if (CollectionUtils.isNotEmpty(contentListByOrchestratorId)) {
// List<Long> collect = contentListByOrchestratorId.stream().map(NodeContentDO::getId).collect(Collectors.toList());
// nodeContentUIMapper.deleteNodeContentUIByContentList(collect);
// }
// nodeContentMapper.deleteNodeContentByOrchestratorId(orchestratorId, oldFLowId);
// }
// 解析 jsonflow
// 解析 proxyUser
try {
Expand Down Expand Up @@ -770,9 +770,13 @@ public DSSFlow copyRootFlow(Long rootFlowId, String userName, Workspace workspac
String description, List<DSSLabel> dssLabels, String nodeSuffix,
String newFlowName, Long newProjectId) throws DSSErrorException, IOException {
DSSFlow dssFlow = flowMapper.selectFlowByID(rootFlowId);
DSSFlow rootFlowWithSubFlows = copyFlowAndSetSubFlowInDB(dssFlow, userName, description, nodeSuffix, newFlowName, newProjectId);
Sender orcSender = DSSSenderServiceFactory.getOrCreateServiceInstance().getOrcSender(dssLabels);
OrchestratorVo orchestratorVo = RpcAskUtils.processAskException(orcSender.ask(new RequestQuertByAppIdOrchestrator(dssFlow.getId())),
OrchestratorVo.class, RequestQueryByIdOrchestrator.class);
Long orchestratorId = orchestratorVo.getDssOrchestratorInfo().getId();
DSSFlow rootFlowWithSubFlows = copyFlowAndSetSubFlowInDB(dssFlow, userName, description, nodeSuffix, newFlowName, newProjectId, orchestratorId, orcSender);
updateFlowJson(userName, projectName, rootFlowWithSubFlows, version, null,
contextIdStr, workspace, dssLabels, nodeSuffix);
contextIdStr, workspace, dssLabels, nodeSuffix, orchestratorId);
DSSFlow copyFlow = flowMapper.selectFlowByID(rootFlowWithSubFlows.getId());
copyFlow.setFlowIdParamConfTemplateIdTuples(rootFlowWithSubFlows.getFlowIdParamConfTemplateIdTuples());
return copyFlow;
Expand Down Expand Up @@ -821,7 +825,8 @@ public boolean checkIsExistSameFlow(String jsonFlow) {
* @return
*/
private DSSFlow copyFlowAndSetSubFlowInDB(DSSFlow dssFlow, String userName, String description,
String subFlowNameSuffix, String newFlowName, Long newProjectId) {
String subFlowNameSuffix, String newFlowName, Long newProjectId,
Long orchestratorId, Sender orcSender) {
DSSFlow cyFlow = new DSSFlow();
BeanUtils.copyProperties(dssFlow, cyFlow, "children", "flowVersions");
//封装flow信息
Expand All @@ -843,14 +848,17 @@ private DSSFlow copyFlowAndSetSubFlowInDB(DSSFlow dssFlow, String userName, Stri
}
cyFlow.setId(null);
flowMapper.insertFlow(cyFlow);
nodeContentMapper.updateFlowId(cyFlow.getId(), dssFlow.getId());
//
OrchestratorVo orchestratorVo = RpcAskUtils.processAskException(orcSender.ask(new RequestQueryByIdOrchestrator(orchestratorId, null)),
OrchestratorVo.class, RequestQueryByIdOrchestrator.class);
nodeContentMapper.updateFlowId(cyFlow.getId(), orchestratorVo.getDssOrchestratorVersion().getAppId());
List<Long> subFlowIDs = flowMapper.selectSubFlowIDByParentFlowID(dssFlow.getId());
for (Long subFlowID : subFlowIDs) {
DSSFlow subDSSFlow = flowMapper.selectFlowByID(subFlowID);
if (dssFlow.getChildren() == null) {
dssFlow.setChildren(new ArrayList<DSSFlow>());
}
DSSFlow copySubFlow = copyFlowAndSetSubFlowInDB(subDSSFlow, userName, description, subFlowNameSuffix, null, newProjectId);
DSSFlow copySubFlow = copyFlowAndSetSubFlowInDB(subDSSFlow, userName, description, subFlowNameSuffix, null, newProjectId, orchestratorId, orcSender);
persistenceFlowRelation(copySubFlow.getId(), cyFlow.getId());
cyFlow.addChildren(copySubFlow);
}
Expand All @@ -859,7 +867,8 @@ private DSSFlow copyFlowAndSetSubFlowInDB(DSSFlow dssFlow, String userName, Stri

private void updateFlowJson(String userName, String projectName, DSSFlow rootFlow,
String version, Long parentFlowId, String contextIdStr,
Workspace workspace, List<DSSLabel> dssLabels, String nodeSuffix) throws DSSErrorException, IOException {
Workspace workspace, List<DSSLabel> dssLabels, String nodeSuffix,
Long orchestratorId) throws DSSErrorException, IOException {
String flowJson = bmlService.readTextFromBML(userName, rootFlow.getResourceId(), rootFlow.getBmlVersion());
//如果包含subflow,需要一同导入subflow内容，并更新parrentflow的json内容
// TODO: 2020/7/31 优化update方法里面的saveContent
Expand All @@ -877,12 +886,14 @@ private void updateFlowJson(String userName, String projectName, DSSFlow rootFlo
//上传节点的资源或调用appconn的copyRef
updateFlowJson = updateWorkFlowNodeJson(userName, projectName, updateFlowJson, rootFlow,
version, workspace, dssLabels);
// 更新对应节点的FlowJson
saveFlowMetaData(rootFlow.getId(), updateFlowJson, orchestratorId);
List<? extends DSSFlow> subFlows = rootFlow.getChildren();
List<String[]> templateIds = new ArrayList<>();
if (subFlows != null) {
for (DSSFlow subflow : subFlows) {
updateFlowJson(userName, projectName, subflow, version, rootFlow.getId(),
contextIdStr, workspace, dssLabels, nodeSuffix);
contextIdStr, workspace, dssLabels, nodeSuffix, orchestratorId);
templateIds.addAll(subflow.getFlowIdParamConfTemplateIdTuples());
}
}
Expand Down

0 comments on commit d6c4c87

Please sign in to comment.