[INFINITY-1312] Set isHealthy field on TaskStatus (#832)

* [SDK] Set `isHealthy` field on TaskStatus Also change TASK_ERROR to TASK_FAILED * Extend marathon deployment readiness wait
mesosphere · Apr 12, 2017 · 8ec86de · 8ec86de
1 parent 5d957cd
commit 8ec86de
Show file tree

Hide file tree

Showing 14 changed files with 224 additions and 15 deletions.
diff --git a/frameworks/cassandra/universe/marathon.json.mustache b/frameworks/cassandra/universe/marathon.json.mustache
@@ -145,6 +145,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/frameworks/elastic/universe/marathon.json.mustache b/frameworks/elastic/universe/marathon.json.mustache
@@ -86,6 +86,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/frameworks/hdfs/universe/marathon.json.mustache b/frameworks/hdfs/universe/marathon.json.mustache
@@ -117,6 +117,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/frameworks/helloworld/universe/marathon.json.mustache b/frameworks/helloworld/universe/marathon.json.mustache
@@ -55,6 +55,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/frameworks/kafka/universe/marathon.json.mustache b/frameworks/kafka/universe/marathon.json.mustache
@@ -156,6 +156,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/frameworks/prototype/universe/marathon.json.mustache b/frameworks/prototype/universe/marathon.json.mustache
@@ -29,6 +29,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/frameworks/proxylite/universe/marathon.json.mustache b/frameworks/proxylite/universe/marathon.json.mustache
@@ -30,6 +30,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/frameworks/spark/universe/marathon.json.mustache b/frameworks/spark/universe/marathon.json.mustache
@@ -44,6 +44,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/frameworks/template/universe/marathon.json.mustache b/frameworks/template/universe/marathon.json.mustache
@@ -51,6 +51,26 @@
     "minimumHealthCapacity": 0,
     "maximumOverCapacity": 0
   },
+  "healthChecks": [
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/deploy",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    },
+    {
+      "protocol": "HTTP",
+      "path": "/v1/plans/recovery",
+      "gracePeriodSeconds": 900,
+      "intervalSeconds": 30,
+      "portIndex": 0,
+      "timeoutSeconds": 30,
+      "maxConsecutiveFailures": 0
+    }
+  ],
   "portDefinitions": [
     {
       "port": 0,

diff --git a/sdk/common/src/main/java/com/mesosphere/sdk/offer/CommonTaskUtils.java b/sdk/common/src/main/java/com/mesosphere/sdk/offer/CommonTaskUtils.java
@@ -360,16 +360,17 @@ public static Environment.Builder fromMapToEnvironment(Map<String, String> envir
     }
 
     /**
-     * Invokes {@link #sendStatus(ExecutorDriver, TaskState, TaskID, SlaveID, ExecutorID, String, byte[])} with a null
-     * {@code data} value.
+     * Invokes {@link #sendStatus(ExecutorDriver, TaskState, TaskID, SlaveID, ExecutorID, String, boolean)} with null
+     * {@code labels} and {@code data} values.
      */
     public static void sendStatus(ExecutorDriver driver,
                                   TaskState state,
                                   TaskID taskID,
                                   SlaveID slaveID,
                                   ExecutorID executorID,
-                                  String message) {
-        sendStatus(driver, state, taskID, slaveID, executorID, message, null, null);
+                                  String message,
+                                  boolean isHealthy) {
+        sendStatus(driver, state, taskID, slaveID, executorID, message, isHealthy, null, null);
     }
 
     /**
@@ -381,6 +382,7 @@ public static void sendStatus(ExecutorDriver driver,
                                   SlaveID slaveID,
                                   ExecutorID executorID,
                                   String message,
+                                  boolean isHealthy,
                                   Labels labels,
                                   byte[] data) {
         final TaskStatus.Builder builder = TaskStatus.newBuilder();
@@ -391,6 +393,7 @@ public static void sendStatus(ExecutorDriver driver,
         builder.setSlaveId(slaveID);
         builder.setExecutorId(executorID);
         builder.setSource(TaskStatus.Source.SOURCE_EXECUTOR);
+        builder.setHealthy(isHealthy);
 
         if (data != null) {
             builder.setData(ByteString.copyFrom(data));

diff --git a/sdk/executor/src/main/java/com/mesosphere/sdk/executor/CustomExecutor.java b/sdk/executor/src/main/java/com/mesosphere/sdk/executor/CustomExecutor.java
@@ -82,8 +82,10 @@ public void launchTask(final ExecutorDriver driver, final Protos.TaskInfo task)
                     task.getTaskId(),
                     task.getSlaveId(),
                     task.getExecutor().getExecutorId(),
-                    String.format("Exception launching task %s",
-                            t.getMessage()));
+                    String.format(
+                            "Exception launching task %s",
+                            t.getMessage()),
+                    false);
         }
     }
 

diff --git a/sdk/executor/src/main/java/com/mesosphere/sdk/executor/HealthCheckHandler.java b/sdk/executor/src/main/java/com/mesosphere/sdk/executor/HealthCheckHandler.java
@@ -4,7 +4,6 @@
 import com.google.protobuf.TextFormat;
 import com.mesosphere.sdk.offer.CommonTaskUtils;
 import com.mesosphere.sdk.offer.Constants;
-
 import org.apache.mesos.ExecutorDriver;
 import org.apache.mesos.Protos;
 import org.slf4j.Logger;
@@ -191,9 +190,26 @@ public void run() {
 
         private void handleHealthCheck() {
             if (healthCheckStats.getConsecutiveFailures() >= healthCheck.getConsecutiveFailures()) {
+                CommonTaskUtils.sendStatus(
+                        executorDriver,
+                        Protos.TaskState.TASK_FAILED,
+                        taskInfo.getTaskId(),
+                        taskInfo.getSlaveId(),
+                        taskInfo.getExecutor().getExecutorId(),
+                        "Health check failed ",
+                        false);
                 throw new HealthCheckRuntimeException(
                         "Health check exceeded its maximum consecutive failures.",
                         healthCheckStats);
+            } else if (healthCheckStats.getConsecutiveSuccesses() == 1) {
+                CommonTaskUtils.sendStatus(
+                        executorDriver,
+                        Protos.TaskState.TASK_RUNNING,
+                        taskInfo.getTaskId(),
+                        taskInfo.getSlaveId(),
+                        taskInfo.getExecutor().getExecutorId(),
+                        "Health check passed",
+                        true);
             }
         }
 
@@ -211,6 +227,7 @@ private void handleReadinessCheck() {
                         taskInfo.getSlaveId(),
                         taskInfo.getExecutor().getExecutorId(),
                         "Readiness check passed",
+                        true,
                         labels,
                         null);
                 throw new HealthCheckRuntimeException("Readiness check passed.", healthCheckStats);