support percent display show when dataX execution

tis-assemble/src/main/java/com/qlangtech/tis/exec/impl/DefaultChainContext.java

@@ -34,18 +34,21 @@
 import com.qlangtech.tis.fullbuild.phasestatus.PhaseStatusCollection;
 import com.qlangtech.tis.fullbuild.servlet.IRebindableMDC;
 import com.qlangtech.tis.job.common.JobCommon;
+import com.qlangtech.tis.manage.common.DagTaskUtils;
 import com.qlangtech.tis.offline.DataxUtils;
 import com.qlangtech.tis.order.center.IAppSourcePipelineController;
 import com.qlangtech.tis.order.center.IParamContext;
 import com.qlangtech.tis.order.center.IndexSwapTaskflowLauncher;
 import com.qlangtech.tis.plugin.StoreResourceType;
 import com.qlangtech.tis.sql.parser.TabPartitions;
+import com.qlangtech.tis.workflow.pojo.WorkFlowBuildHistory;
 import org.apache.commons.lang.StringUtils;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.function.Supplier;
 
 /**
@@ -368,20 +371,30 @@ public void setAppSourcePipelineController(IAppSourcePipelineController appSourc
 
  @Override
  public PhaseStatusCollection loadPhaseStatusFromLatest() {
+ return loadPhaseStatusFromLatest(this.getIndexName());
+ }
 
- // String appName = context.getIndexName();
+ /**
+ * 取得pipeline最近一次成功的执行状态
+ * TODO 还需要支持workflow的执行状态
+ *
+ * @param appName
+ * @return
+ */
+ public static PhaseStatusCollection loadPhaseStatusFromLatest(String appName) {
+ //FIXME 这段代码在执行workflow时候应该会不兼容，届时再调整
+ Optional<WorkFlowBuildHistory> latestWFSuccessTask = DagTaskUtils.getLatestWFSuccessTaskId(appName);
 
- Integer taskId = this.getTaskId();
+ return latestWFSuccessTask.map((history) -> IndexSwapTaskflowLauncher.loadPhaseStatusFromLocal(history.getId())).orElse(null);
 
-// Optional<WorkFlowBuildHistory> latestWFSuccessTask = DagTaskUtils.getLatestWFSuccessTaskId(appName);
 // if (!latestWFSuccessTask.isPresent()) {
 // return null;
 // }
 // WorkFlowBuildHistory h = latestWFSuccessTask.get();
- PhaseStatusCollection phaseStatusCollection = IndexSwapTaskflowLauncher.loadPhaseStatusFromLocal(taskId);
- if (phaseStatusCollection == null) {
- return null;
- }
- return phaseStatusCollection;
+// PhaseStatusCollection phaseStatusCollection = IndexSwapTaskflowLauncher.loadPhaseStatusFromLocal(h.getId());
+// if (phaseStatusCollection == null) {
+// return null;
+// }
+// return phaseStatusCollection;
  }
 }

tis-hadoop-rpc/log-collector.proto

@@ -16,13 +16,27 @@ service LogCollector {
  rpc BuildPhraseStatus (PBuildPhaseStatusParam) returns (stream PPhaseStatusCollection) {
  }
  // 分布式环境下，启动一个同步任务先需要向assemble节点注册所有子任务
- rpc InitTask (PPhaseStatusCollection) returns (Empty) {}
+ rpc InitTask (PPhaseStatusCollection) returns (Empty) {
+ }
+ /**
+ *取得最近一次成功的同步任务状态
+ */
+ rpc LoadPhaseStatusFromLatest (PSynResTarget) returns (PPhaseStatusCollection) {
+ }
 }
 
 message PBuildPhaseStatusParam {
  uint64 taskid = 1;
 }
 
+message PSynResTarget {
+ string name = 1;
+ /**
+ * dataX pipeline OR transform workflow
+ */
+ bool pipeline = 2;
+}
+
 message PPhaseStatusCollection {
  PDumpPhaseStatus dumpPhase = 1;
  PJoinPhaseStatus joinPhase = 2;
@@ -54,14 +68,14 @@ message PMonotorTarget {
  uint32 taskid = 2;
  PExecuteState.LogType logtype = 3;
 
-// enum LogType {
-// INCR_DEPLOY_STATUS_CHANGE = 0;
-// // 近一段时间内增量监听的各个tag的多少值
-// MQ_TAGS_STATUS = 1;
-// FULL = 2;
-// INCR = 3;
-// INCR_SEND = 4;
-// }
+ // enum LogType {
+ // INCR_DEPLOY_STATUS_CHANGE = 0;
+ // // 近一段时间内增量监听的各个tag的多少值
+ // MQ_TAGS_STATUS = 1;
+ // FULL = 2;
+ // INCR = 3;
+ // INCR_SEND = 4;
+ // }
 }
 message PExecuteState {
 

tis-hadoop-rpc/src/main/java/com/qlangtech/tis/rpc/server/IncrStatusClient.java

@@ -30,6 +30,7 @@
 import com.qlangtech.tis.realtime.yarn.rpc.LaunchReportInfo;
 import com.qlangtech.tis.realtime.yarn.rpc.MasterJob;
 import com.qlangtech.tis.realtime.yarn.rpc.PingResult;
+import com.qlangtech.tis.realtime.yarn.rpc.SynResTarget;
 import com.qlangtech.tis.realtime.yarn.rpc.TopicInfo;
 import com.qlangtech.tis.realtime.yarn.rpc.UpdateCounterMap;
 import com.qlangtech.tis.rpc.grpc.log.LogCollectorClient;
@@ -163,6 +164,11 @@ public void initSynJob(PhaseStatusCollection buildStatus) {
  logCollectorBlockingStub.initTask(LogCollectorClient.convertPP(buildStatus));
  }
 
+ @Override
+ public PhaseStatusCollection loadPhaseStatusFromLatest(SynResTarget resTarget) {
+ return null;
+ }
+
  @Override
  public void reportBuildIndexStatus(BuildSharedPhaseStatus buildStatus) {
  com.qlangtech.tis.rpc.grpc.log.common.BuildSharedPhaseStatus.Builder

tis-manage-pojo/src/main/java/com/qlangtech/tis/fullbuild/phasestatus/PhaseStatusCollection.java

@@ -60,16 +60,16 @@ public boolean isFaild() {
 
  public void flushStatus2Local() {
  if (executePhaseRange.contains(FullbuildPhase.FullDump)) {
- dumpPhase.writeStatus2Local();
+ dumpPhase.intervalWriteStatus2Local();
  }
  if (executePhaseRange.contains(FullbuildPhase.JOIN)) {
- joinPhase.writeStatus2Local();
+ joinPhase.intervalWriteStatus2Local();
  }
  if (executePhaseRange.contains(FullbuildPhase.BUILD)) {
- buildPhase.writeStatus2Local();
+ buildPhase.intervalWriteStatus2Local();
  }
  if (executePhaseRange.contains(FullbuildPhase.IndexBackFlow)) {
- indexBackFlowPhaseStatus.writeStatus2Local();
+ indexBackFlowPhaseStatus.intervalWriteStatus2Local();
  }
  }
 

tis-manage-pojo/src/main/java/com/qlangtech/tis/fullbuild/phasestatus/impl/BasicPhaseStatus.java

@@ -1,26 +1,28 @@
 /**
- *  Licensed to the Apache Software Foundation (ASF) under one
- *  or more contributor license agreements. See the NOTICE file
- *  distributed with this work for additional information
- *  regarding copyright ownership. The ASF licenses this file
- *  to you under the Apache License, Version 2.0 (the
- *  "License"); you may not use this file except in compliance
- *  with the License. You may obtain a copy of the License at
- *
- *  http:https://www.apache.org/licenses/LICENSE-2.0
- *
- *  Unless required by applicable law or agreed to in writing, software
- *  distributed under the License is distributed on an "AS IS" BASIS,
- *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *  See the License for the specific language governing permissions and
- *  limitations under the License.
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
  */
 package com.qlangtech.tis.fullbuild.phasestatus.impl;
 
+import com.google.common.util.concurrent.RateLimiter;
 import com.qlangtech.tis.assemble.FullbuildPhase;
 import com.qlangtech.tis.fullbuild.phasestatus.IChildProcessStatus;
 import com.qlangtech.tis.fullbuild.phasestatus.IPhaseStatus;
 import com.qlangtech.tis.manage.common.Config;
+
 import java.io.File;
 import java.util.Collection;
 import java.util.Optional;
@@ -34,9 +36,13 @@ public abstract class BasicPhaseStatus<T extends IChildProcessStatus> implements
  private final int taskid;
 
  // 是否有将状态写入到本地磁盘
- private boolean hasFlush2Local = false;
+ private transient boolean hasFlush2Local = false;
 
  public static transient IFlush2Local statusWriter = null;
+ /**
+ * 10秒执行一次速度
+ */
+ private transient final RateLimiter intervalWriteLimit = RateLimiter.create(0.1);
 
  public static File getFullBuildPhaseLocalFile(int taskid, FullbuildPhase phase) {
  return new File(Config.getMetaCfgDir(), "df-logs/" + taskid + "/" + phase.getName());
@@ -114,38 +120,54 @@ public final boolean isComplete() {
  }
  for (T ts : children) {
  if (ts.isFaild()) {
- return writeStatus2Local();
+ return finalWriteStatus2Local();
  }
  if (!ts.isComplete()) {
  return false;
  }
  }
- return writeStatus2Local();
+ return finalWriteStatus2Local();
+ }
+
+ private boolean finalWriteStatus2Local() {
+ boolean result = false;
+ if (!this.hasFlush2Local) {
+ result = this.writeStatus2Local();
+ this.hasFlush2Local = true;
+ }
+ return result;
+ }
+
+ public boolean intervalWriteStatus2Local() {
+ if (this.intervalWriteLimit.tryAcquire()) {
+ this.writeStatus2Local();
+ return true;
+ }
+ return false;
  }
 
  /**
  * 将本阶段的状态写入本地文件系统
  *
  * @return
  */
- public boolean writeStatus2Local() {
- if (!this.hasFlush2Local) {
- if (statusWriter != null) {
- synchronized (this) {
- File localFile = getFullBuildPhaseLocalFile(this.taskid, this.getPhase());
- this.hasFlush2Local = true;
- try {
- if (!localFile.exists()) {
- // 写入本地文件系统中，后续查看状态可以直接从本地文件中拿到
- statusWriter.write(localFile, this);
- }
- } catch (Exception e) {
- this.hasFlush2Local = false;
- throw new RuntimeException(e);
+ private boolean writeStatus2Local() {
+
+ if (statusWriter != null) {
+ synchronized (this) {
+ File localFile = getFullBuildPhaseLocalFile(this.taskid, this.getPhase());
+
+ try {
+ if (!localFile.exists()) {
+ // 写入本地文件系统中，后续查看状态可以直接从本地文件中拿到
+ statusWriter.write(localFile, this);
  }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
  }
  }
  }
+
  return true;
  }
 

tis-manage-pojo/src/main/java/com/qlangtech/tis/realtime/yarn/rpc/IncrStatusUmbilicalProtocol.java

@@ -75,4 +75,12 @@ public interface IncrStatusUmbilicalProtocol {
  * @param buildStatus
  */
  public void initSynJob(PhaseStatusCollection buildStatus);
+
+ /**
+ * 取得最近一次成功的同步任务状态
+ *
+ * @param resTarget
+ * @return
+ */
+ public PhaseStatusCollection loadPhaseStatusFromLatest(SynResTarget resTarget);
 }

tis-manage-pojo/src/main/java/com/qlangtech/tis/realtime/yarn/rpc/SynResTarget.java

@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.qlangtech.tis.realtime.yarn.rpc;
+
+/**
+ * @author: 百岁（[email protected]）
+ * @create: 2024-04-15 10:51
+ **/
+public class SynResTarget {
+ private final String name;
+ /**
+ * dataX pipeline OR transform workflow
+ */
+ private final boolean pipeline;
+
+ public static SynResTarget pipeline(String name) {
+ return new SynResTarget(name, true);
+ }
+
+ /**
+ * transform workflow 处理管道
+ *
+ * @param name
+ * @return
+ */
+ public static SynResTarget transform(String name) {
+ return new SynResTarget(name, false);
+ }
+
+ private SynResTarget(String name, boolean pipeline) {
+ this.name = name;
+ this.pipeline = pipeline;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public boolean isPipeline() {
+ return pipeline;
+ }
+}

tis-manage-pojo/src/main/java/com/qlangtech/tis/realtime/yarn/rpc/impl/AdapterStatusUmbilicalProtocol.java

@@ -60,6 +60,11 @@ public void reportDumpTableStatus(DumpPhaseStatus.TableDumpStatus tableDumpStatu
  throw new NotImplementedException();
  }
 
+ @Override
+ public PhaseStatusCollection loadPhaseStatusFromLatest(SynResTarget resTarget) {
+ throw new NotImplementedException();
+ }
+
  @Override
  public void reportBuildIndexStatus(BuildSharedPhaseStatus buildStatus) {
  throw new NotImplementedException();