Repository: allegro/marathon-consul
Pull request #156 (2016-12-29 14:12:34)

Handle Task HealthChange without TaskID.
Marathon moved from tasks to instances to support pods.
We didn't want to add pods support right now but we need
to handle events that are not backward compatible. To do this
I introduced `TaskID()` that returns task id from instanceId if
task id was not present in event. This is shortcut just to enable
marathon-consul to be used with Marathon 1.4. We need to rethink
if/how we want to support pods and changed Marathon API.
Fixes: #148

coveralls: 
[![Coverage Status](https://coveralls.io/builds/9459718/badge)](https://coveralls.io/builds/9459718)

Coverage decreased (-0.2%) to 91.467% when pulling **d03fa9bcea5f35cd93f309d622b91f5126ff93a6 on fixes/148/handle_new_events** into **bf8030bb7bec9b87a8fe358f812f954d13eb20b7 on master**.


Patch:
diff --git a/consul/consul.go b/consul/consul.go
index b5688ae..6924267 100644
--- a/consul/consul.go
+++ b/consul/consul.go
@@ -323,9 +323,13 @@ func (c *Consul) marathonToConsulChecks(task *apps.Task, healthChecks []apps.Hea
 		}
 
 		switch check.Protocol {
-		case "HTTP", "HTTPS":
+		case "HTTP", "HTTPS", "MESOS_HTTP", "MESOS_HTTPS":
 			if parsedURL, err := url.ParseRequestURI(check.Path); err == nil {
-				parsedURL.Scheme = strings.ToLower(check.Protocol)
+				if check.Protocol == "HTTP" || check.Protocol == "MESOS_HTTP" {
+					parsedURL.Scheme = "http"
+				} else {
+					parsedURL.Scheme = "https"
+				}
 				parsedURL.Host = fmt.Sprintf("%s:%d", serviceAddress, port)
 				consulCheck.HTTP = parsedURL.String()
 				checks = append(checks, &consulCheck)
@@ -335,7 +339,7 @@ func (c *Consul) marathonToConsulChecks(task *apps.Task, healthChecks []apps.Hea
 					WithField("Address", serviceAddress).
 					Warn(fmt.Sprintf("Could not parse provided path: %s", check.Path))
 			}
-		case "TCP":
+		case "TCP", "MESOS_TCP":
 			consulCheck.TCP = fmt.Sprintf("%s:%d", serviceAddress, port)
 			checks = append(checks, &consulCheck)
 		case "COMMAND":
diff --git a/events/task_health_change.go b/events/task_health_change.go
index f085e8b..6f79243 100644
--- a/events/task_health_change.go
+++ b/events/task_health_change.go
@@ -2,21 +2,47 @@ package events
 
 import (
 	"encoding/json"
+	"errors"
+	"regexp"
 
 	"github.com/allegro/marathon-consul/apps"
 )
 
 type TaskHealthChange struct {
-	Timestamp  string      `json:"timestamp"`
+	Timestamp string `json:"timestamp"`
+	// Prefer TaskID() instead of ID
 	ID         apps.TaskID `json:"id"`
-	TaskStatus string      `json:"taskStatus"`
+	InstanceID string      `json:"instanceId"`
 	AppID      apps.AppID  `json:"appId"`
 	Version    string      `json:"version"`
 	Alive      bool        `json:"alive"`
 }
 
+// Regular expression to extract runSpecId from instanceId
+// See: https://github.com/mesosphere/marathon/blob/v1.4.0-RC4/src/main/scala/mesosphere/marathon/core/instance/Instance.scala#L244
+var instanceIdRegex = regexp.MustCompile(`^(.+)\.(instance-|marathon-)([^\.]+)$`)
+
+func (t TaskHealthChange) TaskID() apps.TaskID {
+	if t.ID != "" {
+		return t.ID
+	}
+	return apps.TaskID(instanceIdRegex.ReplaceAllString(t.InstanceID, "$1.$3"))
+}
+
 func ParseTaskHealthChange(event []byte) (*TaskHealthChange, error) {
 	task := &TaskHealthChange{}
 	err := json.Unmarshal(event, task)
-	return task, err
+
+	if err != nil {
+		return nil, err
+	}
+
+	// Marathon 1.4 changes this event so it does not contain TaskID.
+	// We need to validate this event if it contains required fields.
+	// See: https://phabricator.mesosphere.com/D218#10153
+	if task.ID == "" && task.InstanceID == "" {
+		return nil, errors.New("Missing task ID")
+	}
+
+	return task, nil
 }
diff --git a/web/event_handler.go b/web/event_handler.go
index 12a3b66..6d1e549 100644
--- a/web/event_handler.go
+++ b/web/event_handler.go
@@ -81,10 +81,10 @@ func (fh *eventHandler) handleEvent(eventType string, body []byte) error {
 	body = replaceTaskIDWithID(body)
 
 	switch eventType {
-	case "status_update_event":
+	case statusUpdateEventType:
 		return fh.handleStatusEvent(body)
-	case "health_status_changed_event":
-		return fh.handleHealthStatusEvent(body)
+	case healthStatusChangedEventType:
+		return fh.handleHealthyTask(body)
 	default:
 		err := fmt.Errorf("Unsuported event type: %s", eventType)
 		log.WithError(err).WithField("EventType", eventType).Error("This should never happen. Not handled event type")
@@ -92,42 +92,39 @@ func (fh *eventHandler) handleEvent(eventType string, body []byte) error {
 	}
 }
 
-func (fh *eventHandler) handleHealthStatusEvent(body []byte) error {
+func (fh *eventHandler) handleHealthyTask(body []byte) error {
 	taskHealthChange, err := events.ParseTaskHealthChange(body)
 	if err != nil {
 		log.WithError(err).Error("Body generated error")
 		return err
 	}
 
-	log.WithFields(
-		log.Fields{
-			"Id":         taskHealthChange.ID,
-			"TaskStatus": taskHealthChange.TaskStatus,
-		}).Info("Got HealthStatusEvent")
+	appID := taskHealthChange.AppID
+	taskID := taskHealthChange.TaskID()
+	log.WithField("Id", taskID).Info("Got HealthStatusEvent")
 
 	if !taskHealthChange.Alive {
-		log.WithField("Id", taskHealthChange.ID).Debug("Task is not alive. Not registering")
+		log.WithField("Id", taskID).Debug("Task is not alive. Not registering")
 		return nil
 	}
 
-	appID := taskHealthChange.AppID
 	app, err := fh.marathon.App(appID)
 	if err != nil {
-		log.WithField("Id", taskHealthChange.ID).WithError(err).Error("There was a problem obtaining app info")
+		log.WithField("Id", taskID).WithError(err).Error("There was a problem obtaining app info")
 		return err
 	}
 
 	if !app.IsConsulApp() {
 		err = fmt.Errorf("%s is not consul app. Missing consul label", app.ID)
-		log.WithField("Id", taskHealthChange.ID).WithError(err).Debug("Skipping app registration in Consul")
+		log.WithField("Id", taskID).WithError(err).Debug("Skipping app registration in Consul")
 		return nil
 	}
 
 	tasks := app.Tasks
 
-	task, err := findTaskByID(taskHealthChange.ID, tasks)
+	task, err := findTaskByID(taskID, tasks)
 	if err != nil {
-		log.WithField("Id", taskHealthChange.ID).WithError(err).Error("Task not found")
+		log.WithField("Id", taskID).WithError(err).Error("Task not found")
 		return err
 	}
 


Test patch:
diff --git a/consul/consul_test.go b/consul/consul_test.go
index 0c17f55..5feb247 100644
--- a/consul/consul_test.go
+++ b/consul/consul_test.go
@@ -829,7 +829,7 @@ func TestMarathonTaskToConsulServiceMapping_IgnoredHealthcheckTypes(t *testing.T
 		HealthChecks: []apps.HealthCheck{
 			{
 				Path:                   "/api/health?with=query",
-				Protocol:               "HTTP",
+				Protocol:               "MESOS_HTTP",
 				PortIndex:              0,
 				IntervalSeconds:        60,
 				TimeoutSeconds:         20,
diff --git a/events/task_health_change_test.go b/events/task_health_change_test.go
index 0915748..284c23d 100644
--- a/events/task_health_change_test.go
+++ b/events/task_health_change_test.go
@@ -8,12 +8,11 @@ import (
 )
 
 var testHealthChange = &TaskHealthChange{
-	Timestamp:  "2014-03-01T23:29:30.158Z",
-	ID:         "my-app_0-1396592784349",
-	Alive:      true,
-	TaskStatus: "TASK_RUNNING",
-	AppID:      "/my-app",
-	Version:    "2014-04-04T06:26:23.051Z",
+	Timestamp: "2014-03-01T23:29:30.158Z",
+	ID:        "my-app_0-1396592784349",
+	Alive:     true,
+	AppID:     "/my-app",
+	Version:   "2014-04-04T06:26:23.051Z",
 }
 
 func TestHealthChangeParseTask(t *testing.T) {
@@ -28,7 +27,37 @@ func TestHealthChangeParseTask(t *testing.T) {
 	assert.Equal(t, testHealthChange.Timestamp, service.Timestamp)
 	assert.Equal(t, testHealthChange.ID, service.ID)
 	assert.Equal(t, testHealthChange.Alive, service.Alive)
-	assert.Equal(t, testHealthChange.TaskStatus, service.TaskStatus)
 	assert.Equal(t, testHealthChange.AppID, service.AppID)
 	assert.Equal(t, testHealthChange.Version, service.Version)
 }
+
+func TestHealthChangeParseTaskWithoutData(t *testing.T) {
+	t.Parallel()
+
+	event, err := ParseTaskHealthChange([]byte("{}"))
+	assert.Nil(t, event)
+	assert.EqualError(t, err, "Missing task ID")
+}
+
+func TestHealthChangeParseTaskWithBrokenJson(t *testing.T) {
+	t.Parallel()
+
+	event, err := ParseTaskHealthChange([]byte("not a Json"))
+	assert.Nil(t, event)
+	assert.Error(t, err)
+}
+
+func TestInstanceIDToTaskID(t *testing.T) {
+	t.Parallel()
+
+	ids := []string{
+		"python1.marathon-0bf4660a-cdc0-11e6-87df-0242ee53bf4b",
+		"python1.instance-0bf4660a-cdc0-11e6-87df-0242ee53bf4b",
+		"python1.0bf4660a-cdc0-11e6-87df-0242ee53bf4b",
+	}
+
+	for _, id := range ids {
+		instance := TaskHealthChange{InstanceID: id}
+		assert.Equal(t, "python1.0bf4660a-cdc0-11e6-87df-0242ee53bf4b", instance.TaskID().String())
+	}
+}