Refine action status report (#1677)

littlezhou · Apr 26, 2018 · 4950fb6 · 4950fb6
1 parent 996b7b8
commit 4950fb6
Show file tree

Hide file tree

Showing 8 changed files with 52 additions and 19 deletions.
diff --git a/conf/smart-default.xml b/conf/smart-default.xml
@@ -153,7 +153,7 @@
 
   <property>
     <name>smart.status.report.period</name>
-    <value>50</value>
+    <value>10</value>
     <description>
       The status report period for actions. The time unit is millisecond.
     </description>

diff --git a/smart-agent/src/main/java/org/smartdata/agent/SmartAgent.java b/smart-agent/src/main/java/org/smartdata/agent/SmartAgent.java
@@ -146,9 +146,9 @@ public void run() {
     cmdletExecutor = agentCmdletService.getCmdletExecutor();
 
     ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
-    StatusReportTask statusReportTask = new StatusReportTask(this, cmdletExecutor);
-    long reportPeriod = conf.getLong(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
+    int reportPeriod = conf.getInt(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
             SmartConfKeys.SMART_STATUS_REPORT_PERIOD_DEFAULT);
+    StatusReportTask statusReportTask = new StatusReportTask(this, cmdletExecutor, reportPeriod);
     executorService.scheduleAtFixedRate(
             statusReportTask, 1000, reportPeriod, TimeUnit.MILLISECONDS);
 

diff --git a/smart-common/src/main/java/org/smartdata/conf/SmartConfKeys.java b/smart-common/src/main/java/org/smartdata/conf/SmartConfKeys.java
@@ -157,7 +157,7 @@ public class SmartConfKeys {
       "60s,60;1hour,60;1day";
 
   public static final String SMART_STATUS_REPORT_PERIOD_KEY = "smart.status.report.period";
-  public static final long SMART_STATUS_REPORT_PERIOD_DEFAULT = 50;
+  public static final int SMART_STATUS_REPORT_PERIOD_DEFAULT = 10;
 
   //Tidb
   public static final String SMART_TIDB_ENABLED = "smart.tidb.enable";

diff --git a/smart-engine/src/main/java/org/smartdata/server/cluster/HazelcastWorker.java b/smart-engine/src/main/java/org/smartdata/server/cluster/HazelcastWorker.java
@@ -48,7 +48,7 @@ public class HazelcastWorker implements StatusReporter {
   private ITopic<StatusMessage> statusTopic;
   private CmdletExecutor cmdletExecutor;
   private CmdletFactory factory;
-  private long reportPeriod;
+  private int reportPeriod;
 
   public HazelcastWorker(SmartContext smartContext) {
     this.factory = new CmdletFactory(smartContext, this);
@@ -60,12 +60,12 @@ public HazelcastWorker(SmartContext smartContext) {
     this.masterMessages =
         instance.getTopic(HazelcastExecutorService.WORKER_TOPIC_PREFIX + instanceId);
     this.masterMessages.addMessageListener(new MasterMessageListener());
-    this.reportPeriod = smartContext.getConf().getLong(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
+    this.reportPeriod = smartContext.getConf().getInt(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
             SmartConfKeys.SMART_STATUS_REPORT_PERIOD_DEFAULT);
   }
 
   public void start() {
-    StatusReportTask statusReportTask = new StatusReportTask(this, cmdletExecutor);
+    StatusReportTask statusReportTask = new StatusReportTask(this, cmdletExecutor, reportPeriod);
     executorService.scheduleAtFixedRate(
             statusReportTask, 1000, reportPeriod, TimeUnit.MILLISECONDS);
   }

diff --git a/smart-engine/src/main/java/org/smartdata/server/engine/CmdletManager.java b/smart-engine/src/main/java/org/smartdata/server/engine/CmdletManager.java
@@ -136,7 +136,7 @@ public CmdletManager(ServerContext context) throws IOException {
         .getInt(SmartConfKeys.SMART_CMDLET_CACHE_BATCH,
         SmartConfKeys.SMART_CMDLET_CACHE_BATCH_DEFAULT);
 
-    long reportPeriod = context.getConf().getLong(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
+    int reportPeriod = context.getConf().getInt(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
             SmartConfKeys.SMART_STATUS_REPORT_PERIOD_DEFAULT);
     this.timeout =
             TIMEOUT_MULTIPLIER * reportPeriod < 30000 ? 30000 : TIMEOUT_MULTIPLIER * reportPeriod;

diff --git a/...t-engine/src/main/java/org/smartdata/server/engine/cmdlet/LocalCmdletExecutorService.java b/...t-engine/src/main/java/org/smartdata/server/engine/cmdlet/LocalCmdletExecutorService.java
@@ -55,9 +55,9 @@ public LocalCmdletExecutorService(SmartConf smartConf, CmdletManager cmdletManag
     this.cmdletExecutor = new CmdletExecutor(smartConf);
     this.executorService = Executors.newSingleThreadScheduledExecutor();
 
-    StatusReportTask statusReportTask = new StatusReportTask(this, cmdletExecutor);
-    long reportPeriod = smartConf.getLong(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
+    int reportPeriod = smartConf.getInt(SmartConfKeys.SMART_STATUS_REPORT_PERIOD_KEY,
             SmartConfKeys.SMART_STATUS_REPORT_PERIOD_DEFAULT);
+    StatusReportTask statusReportTask = new StatusReportTask(this, cmdletExecutor, reportPeriod);
     this.executorService.scheduleAtFixedRate(
         statusReportTask, 1000, reportPeriod, TimeUnit.MILLISECONDS);
 

diff --git a/smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/StatusReportTask.java b/smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/StatusReportTask.java
@@ -17,23 +17,56 @@
  */
 package org.smartdata.server.engine.cmdlet;
 
+import org.smartdata.protocol.message.ActionStatus;
 import org.smartdata.protocol.message.StatusReport;
 import org.smartdata.protocol.message.StatusReporter;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 public class StatusReportTask implements Runnable {
-  StatusReporter statusReporter;
-  CmdletExecutor cmdletExecutor;
+  private StatusReporter statusReporter;
+  private CmdletExecutor cmdletExecutor;
+  private long lastReportTime;
+  private int interval;
+  private Map<Long, ActionStatus> idToActionStatus;
+  public static final int TIME_MULTIPLIER = 50;
+  public static final double FINISHED_RATIO = 0.2;
 
-  public StatusReportTask(StatusReporter statusReporter, CmdletExecutor cmdletExecutor) {
+  public StatusReportTask(
+      StatusReporter statusReporter, CmdletExecutor cmdletExecutor, int period) {
     this.statusReporter = statusReporter;
     this.cmdletExecutor = cmdletExecutor;
+    this.lastReportTime = System.currentTimeMillis();
+    this.interval = TIME_MULTIPLIER * period;
+    this.idToActionStatus = new HashMap<>();
   }
 
   @Override
   public void run() {
     StatusReport statusReport = cmdletExecutor.getStatusReport();
-    if (statusReport != null && !statusReport.getActionStatuses().isEmpty()) {
-      statusReporter.report(statusReport);
+    if (statusReport != null) {
+      List<ActionStatus> actionStatuses = statusReport.getActionStatuses();
+      for (ActionStatus actionStatus : actionStatuses) {
+        idToActionStatus.put(actionStatus.getActionId(), actionStatus);
+      }
+      if (!idToActionStatus.values().isEmpty()) {
+        int finishedNum = 0;
+        for (ActionStatus actionStatus : idToActionStatus.values()) {
+          if (actionStatus.isFinished()) {
+            finishedNum++;
+          }
+        }
+        long currentTime = System.currentTimeMillis();
+        if (currentTime - lastReportTime >= interval
+            || (float) finishedNum / idToActionStatus.size() >= FINISHED_RATIO) {
+          statusReporter.report(new StatusReport(new ArrayList(idToActionStatus.values())));
+          idToActionStatus.clear();
+          lastReportTime = currentTime;
+        }
+      }
     }
   }
 }
diff --git a/smart-engine/src/test/java/org/smartdata/server/engine/cmdlet/TestCmdletExecutor.java b/smart-engine/src/test/java/org/smartdata/server/engine/cmdlet/TestCmdletExecutor.java
@@ -50,10 +50,10 @@ public void report(StatusMessage status) {
           }
         };
     CmdletExecutor executor = new CmdletExecutor(new SmartConf());
-    StatusReportTask statusReportTask = new StatusReportTask(reporter, executor);
+    StatusReportTask statusReportTask = new StatusReportTask(reporter, executor, 10);
     ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
     executorService.scheduleAtFixedRate(
-            statusReportTask, 1000, 1000, TimeUnit.MILLISECONDS);
+            statusReportTask, 100, 10, TimeUnit.MILLISECONDS);
     SmartAction action = new HelloAction();
     Map<String, String> args = new HashMap<>();
     args.put(HelloAction.PRINT_MESSAGE, "message");
@@ -91,10 +91,10 @@ public void report(StatusMessage status) {
           }
         };
     CmdletExecutor executor = new CmdletExecutor(new SmartConf());
-    StatusReportTask statusReportTask = new StatusReportTask(reporter, executor);
+    StatusReportTask statusReportTask = new StatusReportTask(reporter, executor, 10);
     ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
     executorService.scheduleAtFixedRate(
-            statusReportTask, 1000, 1000, TimeUnit.MILLISECONDS);
+            statusReportTask, 100, 10, TimeUnit.MILLISECONDS);
     SmartAction action = new HangingAction();
     action.setActionId(101);
     Cmdlet cmdlet = new Cmdlet(Arrays.asList(action));