Repository: allegro/marathon-consul
Pull request #254 (2017-07-31 14:05:05)

Remove web hamdler. (Breaking change)
This change removes the support for HTTP Callbacs.
This mechanism is deprecated and will be removed in
futre Marathon releases.

Breaking changes:

* Remove `sse-enabled` and `web-enabled` flags

Fixes: #253

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

Coverage increased (+0.7%) to 90.119% when pulling **58b844260f03dc10b6ab4ae2852c2bd6c1916842 on remove_http_callback** into **b1c1ec5b1e30a40a7a531d220cfa49036ef5831c on master**.

tomez: Are these files still in use?

events/web_events.go,
events/web_events_test.go.
coveralls: 
[![Coverage Status](https://coveralls.io/builds/12635783/badge)](https://coveralls.io/builds/12635783)

Coverage increased (+0.7%) to 90.06% when pulling **bc8ad8a29ba16a161dafa281893fd4f01f70ea7a on remove_http_callback** into **b1c1ec5b1e30a40a7a531d220cfa49036ef5831c on master**.


Patch:
diff --git a/README.md b/README.md
index 1984ee8..c9f4f95 100644
--- a/README.md
+++ b/README.md
@@ -98,23 +98,15 @@ your cert store might be different depending on your system.
 docker run '/etc/ssl/certs/ca-certificates.crt:/etc/ssl/certs/ca-certificates.crt' -P  allegro/marathon-consul
 ```
 
-## Setting up `marathon-consul` after installation
-
-The Marathon [event bus](https://mesosphere.github.io/marathon/docs/event-bus.html) should point to [`/events`](#endpoints). You can set up the event subscription with a call similar to this one:
-```
-curl -X POST 'http://marathon.service.consul:8080/v2/eventSubscriptions?callbackUrl=http://marathon-consul.service.consul:4000/events'
-```
-The event subscription should be set to `localhost` to reduce network traffic.
-
 ## Usage
 
 ### Marathon masters
 
 - marathon-consul should be installed on all Marathon masters
 
-### Mesos slaves
+### Mesos agents
 
-- Consul Agents should be available on every Mesos slave.
+- Consul Agents should be available on every Mesos agent.
 - Tasks will be registered at the Mesos slave they run on.
 
 ### Tagging tasks
@@ -236,15 +228,12 @@ sync-enabled                | `true`          | Enable Marathon-consul scheduled
 sync-force                  | `false`         | Force leadership-independent Marathon-consul sync (run always)
 sync-interval               | `15m0s`         | Marathon-consul sync interval
 workers-pool-size           | `10`            | Number of concurrent workers processing events
-sse-enabled                 | `false`         | Enable marathon-consul SSE on this node
-web-enabled                 | `true`          | Enable marathon-consul Web callbacks on this node
 
 ### Endpoints
 
 Endpoint  | Description
 ----------|------------------------------------------------------------------------------------
 `/health` | healthcheck - returns `OK`
-`/events` | event sink - returns `OK` if all keys are set in an event, error message otherwise
 
 ## Advanced usage
 
@@ -322,7 +311,6 @@ reregister all healthy services managed by marathon-consul to the new format. Un
 
 ## SSE Support
 
-In future callback interface between marathon and marathon-consul will be replaced by SSE.
 While using SSE please consider:
 - SSE is using Web module config for queues, event sizes, in the future will be moved to sse module,
 - SSE is using marathon-leader config for determining current leader, when this value match leader returned by marathon (/v2/leader endpoint)
@@ -330,6 +318,12 @@ then SSE is started on this instance,
 - when enabled SSE is spawning its own own set of workers and separated dispatcher,
 - be advised to disable marathon callback subscription when enabling SSE, otherwise it might result in doubling registers and deregisers.
 
+## HTTP callbacks support
+
+Marathon-Consul does not support HTTP callbacks.
+Marathon [deprecated support for HTTP callbacks in 1.4](https://github.com/mesosphere/marathon/blob/master/changelog.md#deprecate-event-callback-subscriptions).
+This mechanism is no longer available starting from [Marathon 1.5](https://github.com/mesosphere/marathon/blob/master/changelog.md#event-subscribers-has-been-removed).
+
 ## Known limitations
 
 The following section describes known limitations in `marathon-consul`.
diff --git a/config/config.go b/config/config.go
index 4a81344..c526a94 100644
--- a/config/config.go
+++ b/config/config.go
@@ -85,10 +85,8 @@ func (config *Config) parseFlags() {
 	flag.IntVar(&config.Web.QueueSize, "events-queue-size", 1000, "Size of events queue")
 	flag.IntVar(&config.Web.WorkersCount, "workers-pool-size", 10, "Number of concurrent workers processing events")
 	flag.Int64Var(&config.Web.MaxEventSize, "event-max-size", 4096, "Maximum size of event to process (bytes)")
-	flag.BoolVar(&config.Web.Enabled, "web-enabled", true, "Enable web events (callbacks).")
 
 	// SSE
-	flag.BoolVar(&config.SSE.Enabled, "sse-enabled", false, "Enable sse event stream.")
 
 	// Sync
 	flag.BoolVar(&config.Sync.Enabled, "sync-enabled", true, "Enable Marathon-consul scheduled sync")
diff --git a/debian/config.json b/debian/config.json
index 5ee8673..62abc2a 100644
--- a/debian/config.json
+++ b/debian/config.json
@@ -24,11 +24,9 @@
     "Listen": ":4000",
     "QueueSize": 1000,
     "WorkersCount": 10,
-    "MaxEventSize": 4096,
-    "Enabled": true
+    "MaxEventSize": 4096
   },
   "SSE": {
-    "Enabled": false
   },
   "Sync": {
     "Enabled": true,
diff --git a/events/web_events.go b/events/web_events.go
deleted file mode 100644
index 5ab3ce3..0000000
--- a/events/web_events.go
+++ /dev/null
@@ -1,26 +0,0 @@
-package events
-
-import (
-	"encoding/json"
-	"errors"
-
-	"github.com/allegro/marathon-consul/time"
-)
-
-type WebEvent struct {
-	Type      string         `json:"eventType"`
-	Timestamp time.Timestamp `json:"timestamp"`
-}
-
-func ParseEvent(jsonBlob []byte) (WebEvent, error) {
-	webEvent := WebEvent{}
-	err := json.Unmarshal(jsonBlob, &webEvent)
-	if err != nil {
-		return WebEvent{}, err
-	} else if webEvent.Type == "" {
-		return WebEvent{}, errors.New("Missing event type")
-	} else if webEvent.Timestamp.Missing() {
-		return WebEvent{}, errors.New("Missing timestamp")
-	}
-	return webEvent, nil
-}
diff --git a/main.go b/main.go
index e3270f0..edf548b 100644
--- a/main.go
+++ b/main.go
@@ -43,19 +43,13 @@ func main() {
 
 	sync.New(config.Sync, remote, consulInstance, consulInstance.AddAgentsFromApps).StartSyncServicesJob()
 
-	if config.SSE.Enabled {
+	go func() {
 		stopSSE, err := sse.NewHandler(config.SSE, config.Web, remote, consulInstance)
 		if err != nil {
 			log.WithError(err).Fatal("Cannot instantiate SSE handler")
 		}
 		defer stopSSE()
-	}
-
-	if config.Web.Enabled {
-		handler, stop := web.NewHandler(config.Web, remote, consulInstance)
-		defer stop()
-		http.HandleFunc("/events", handler)
-	}
+	}()
 
 	http.HandleFunc("/health", web.HealthHandler)
 
diff --git a/sse/config.go b/sse/config.go
index 3de3ab2..c19e323 100644
--- a/sse/config.go
+++ b/sse/config.go
@@ -1,7 +1,6 @@
 package sse
 
 type Config struct {
-	Enabled      bool
 	Retries      int
 	RetryBackoff int
 }
diff --git a/web/config.go b/web/config.go
index ac7cadb..acdcb38 100644
--- a/web/config.go
+++ b/web/config.go
@@ -5,5 +5,4 @@ type Config struct {
 	QueueSize    int
 	WorkersCount int
 	MaxEventSize int64
-	Enabled      bool
 }
diff --git a/web/web.go b/web/web.go
deleted file mode 100644
index 9d9908c..0000000
--- a/web/web.go
+++ /dev/null
@@ -1,31 +0,0 @@
-package web
-
-import (
-	"net/http"
-
-	"github.com/allegro/marathon-consul/events"
-	"github.com/allegro/marathon-consul/marathon"
-	"github.com/allegro/marathon-consul/service"
-)
-
-type Stop func()
-type Handler func(w http.ResponseWriter, r *http.Request)
-
-func NewHandler(config Config, marathon marathon.Marathoner, serviceOperations service.ServiceRegistry) (Handler, Stop) {
-
-	stopChannels := make([]chan<- events.StopEvent, config.WorkersCount)
-	eventQueue := make(chan events.Event, config.QueueSize)
-	for i := 0; i < config.WorkersCount; i++ {
-		handler := events.NewEventHandler(i, serviceOperations, marathon, eventQueue)
-		stopChannels[i] = handler.Start()
-	}
-	return newWebHandler(eventQueue, config.MaxEventSize).Handle, stop(stopChannels)
-}
-
-func stop(channels []chan<- events.StopEvent) Stop {
-	return func() {
-		for _, channel := range channels {
-			channel <- events.StopEvent{}
-		}
-	}
-}
diff --git a/web/web_handler.go b/web/web_handler.go
deleted file mode 100644
index 07ff4f2..0000000
--- a/web/web_handler.go
+++ /dev/null
@@ -1,82 +0,0 @@
-package web
-
-import (
-	"errors"
-	"fmt"
-	"io/ioutil"
-	"net/http"
-	"time"
-
-	log "github.com/Sirupsen/logrus"
-	"github.com/allegro/marathon-consul/events"
-	"github.com/allegro/marathon-consul/metrics"
-)
-
-type EventHandler struct {
-	eventQueue   chan events.Event
-	maxEventSize int64
-}
-
-func newWebHandler(eventQueue chan events.Event, maxEventSize int64) *EventHandler {
-	if maxEventSize < 1000 {
-		log.WithField("maxEventSize", maxEventSize).Warning("Max event size is too small. Switching to 1000")
-		maxEventSize = 1000
-	}
-	return &EventHandler{
-		eventQueue:   eventQueue,
-		maxEventSize: maxEventSize,
-	}
-}
-
-// Handle is responsible for accepting events and passing them to event queue
-// for async processing. It always returns 2xx even if requests are malformed
-// to prevent marathon from suspending subscription.
-// Processed events must be smaller than maxEventSize and must contain
-// supported event type.
-func (h *EventHandler) Handle(w http.ResponseWriter, r *http.Request) {
-	metrics.Time("events.response", func() {
-		limitedBody := http.MaxBytesReader(w, r.Body, h.maxEventSize)
-		defer limitedBody.Close()
-		body, err := ioutil.ReadAll(limitedBody)
-		if err != nil {
-			drop(err, w)
-			return
-		}
-
-		e, err := events.ParseEvent(body)
-		if err != nil {
-			drop(err, w)
-			return
-		}
-
-		metrics.Mark("events.requests." + e.Type)
-		log.WithFields(log.Fields{"EventType": e.Type, "OriginalTimestamp": e.Timestamp.String()}).Debug("Received event")
-
-		if e.Type != events.StatusUpdateEventType && e.Type != events.HealthStatusChangedEventType {
-			drop(fmt.Errorf("%s is not supported", e.Type), w)
-			return
-		}
-
-		select {
-		case h.eventQueue <- events.Event{EventType: e.Type, Body: body, Timestamp: time.Now()}:
-			accept(w)
-		default:
-			metrics.Mark("events.queue.drop")
-			drop(errors.New("Event queue full"), w)
-		}
-
-	})
-}
-
-func accept(w http.ResponseWriter) {
-	metrics.Mark("events.response.accept")
-	w.WriteHeader(http.StatusAccepted)
-	fmt.Fprintln(w, "OK")
-}
-
-func drop(err error, w http.ResponseWriter) {
-	log.WithError(err).Debug("Malformed request")
-	metrics.Mark("events.response.drop")
-	w.WriteHeader(http.StatusOK)
-	fmt.Fprintln(w, err.Error())
-}


Test patch:
diff --git a/config/config_test.go b/config/config_test.go
index 2dfc13d..d19fdeb 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -111,11 +111,8 @@ func TestConfig_ShouldBeMergedWithFileDefaultsAndFlags(t *testing.T) {
 			QueueSize:    1000,
 			WorkersCount: 10,
 			MaxEventSize: 4096,
-			Enabled:      true,
-		},
-		SSE: sse.Config{
-			Enabled: false,
 		},
+		SSE: sse.Config{},
 		Sync: sync.Config{
 			Interval: timeutil.Interval{Duration: 15 * time.Minute},
 			Enabled:  true,
diff --git a/events/web_events_test.go b/events/web_events_test.go
deleted file mode 100644
index 8eaba2b..0000000
--- a/events/web_events_test.go
+++ /dev/null
@@ -1,44 +0,0 @@
-package events
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestWebEventTypeValid(t *testing.T) {
-	t.Parallel()
-	out, err := ParseEvent([]byte(`{"eventType":"test","timestamp": "2014-03-01T23:29:30.158Z"}`))
-	assert.Nil(t, err)
-	assert.Equal(t, "2014-03-01T23:29:30.158Z", out.Timestamp.String())
-}
-
-func TestWebEventTypeInvalid(t *testing.T) {
-	t.Parallel()
-
-	out, err := ParseEvent([]byte(`{}`))
-	assert.Error(t, err)
-	assert.Equal(t, out, WebEvent{})
-}
-
-func TestTimestampInvalid(t *testing.T) {
-	t.Parallel()
-	out, err := ParseEvent([]byte(`{"eventType":"test","timestamp": "invalid"}`))
-	assert.Error(t, err)
-	assert.Equal(t, out, WebEvent{})
-}
-
-func TestMissingTimestamp(t *testing.T) {
-	t.Parallel()
-	out, err := ParseEvent([]byte(`{"eventType":"test"}`))
-	assert.EqualError(t, err, "Missing timestamp")
-	assert.Equal(t, out, WebEvent{})
-}
-
-func TestWebEventTypeInvalidJson(t *testing.T) {
-	t.Parallel()
-
-	out, err := ParseEvent([]byte(`not a json`))
-	assert.Error(t, err)
-	assert.Equal(t, out, WebEvent{})
-}
diff --git a/web/web_handler_test.go b/web/web_handler_test.go
deleted file mode 100644
index 8433f68..0000000
--- a/web/web_handler_test.go
+++ /dev/null
@@ -1,188 +0,0 @@
-package web
-
-import (
-	"bytes"
-	"errors"
-	"net/http"
-	"net/http/httptest"
-	"testing"
-
-	"github.com/allegro/marathon-consul/events"
-	"github.com/stretchr/testify/assert"
-)
-
-const maxEventSize = 4096
-
-type BadReader struct{}
-
-func (r BadReader) Read(p []byte) (int, error) {
-	return 0, errors.New("Some error")
-}
-
-func TestWebHandler_Send202AcceptedWhenEventsPassesValidation(t *testing.T) {
-	t.Parallel()
-
-	// given
-	body := []byte(`{
-		  "eventType":"status_update_event",
-		  "timestamp":"2015-12-07T09:33:40.898Z"
-		}`)
-
-	queue := make(chan events.Event, 1)
-	handler := newWebHandler(queue, maxEventSize)
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer(body))
-	recorder := httptest.NewRecorder()
-
-	// when
-	handler.Handle(recorder, req)
-
-	// then
-	assertAccepted(t, recorder)
-}
-
-func TestWebHandler_HandleReaderError(t *testing.T) {
-	t.Parallel()
-
-	// given
-	handler := newWebHandler(nil, maxEventSize)
-	req, _ := http.NewRequest("POST", "/events", BadReader{})
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-
-	// then
-	assert.Equal(t, 200, recorder.Code)
-	assert.Equal(t, "Some error\n", recorder.Body.String())
-}
-
-func TestWebHandler_DropBigBody(t *testing.T) {
-	t.Parallel()
-
-	// given
-	handler := newWebHandler(nil, maxEventSize)
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer(make([]byte, 4097)))
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-
-	// then
-	assert.Equal(t, 200, recorder.Code)
-	assert.Equal(t, "http: request body too large\n", recorder.Body.String())
-}
-
-func TestWebHandler_DropEmptyBody(t *testing.T) {
-	t.Parallel()
-
-	// given
-	handler := newWebHandler(nil, maxEventSize)
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte{}))
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-
-	// then
-	assertDropped(t, recorder)
-}
-
-func TestWebHandler_DropAppInvalidBody(t *testing.T) {
-	t.Parallel()
-
-	// given
-	handler := newWebHandler(nil, maxEventSize)
-	body := `{"type":  "app_terminated_event", "appID": 123}`
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-
-	// then
-	assertDropped(t, recorder)
-}
-
-func TestWebHandler_DropMalformedEventType(t *testing.T) {
-	t.Parallel()
-
-	// given
-	handler := newWebHandler(nil, maxEventSize)
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(`{eventType:"test_event"}`)))
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-
-	// then
-	assertDropped(t, recorder)
-}
-
-func TestWebHandler_DropInvalidEventType(t *testing.T) {
-	t.Parallel()
-
-	// given
-	handler := newWebHandler(nil, maxEventSize)
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(`{"eventType":[1,2]}`)))
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-
-	// then
-	assertDropped(t, recorder)
-}
-
-func TestWebHandler_DropUnknownEventType(t *testing.T) {
-	t.Parallel()
-
-	// given
-	handler := newWebHandler(nil, maxEventSize)
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(`{"eventType":"test_event"}`)))
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-
-	// then
-	assertDropped(t, recorder)
-}
-
-func TestWebHandler_DropWhenEventQueueIsFull(t *testing.T) {
-	t.Parallel()
-
-	// given
-	body := []byte(`{
-                  "eventType":"status_update_event",
-                  "timestamp":"2015-12-07T09:33:40.898Z"
-                }`)
-
-	queue := make(chan events.Event, 1)
-	handler := newWebHandler(queue, maxEventSize)
-	req1, _ := http.NewRequest("POST", "/events", bytes.NewBuffer(body))
-	req2, _ := http.NewRequest("POST", "/events", bytes.NewBuffer(body))
-	recorder1 := httptest.NewRecorder()
-	recorder2 := httptest.NewRecorder()
-
-	// when
-	handler.Handle(recorder1, req1)
-	handler.Handle(recorder2, req2)
-
-	// then
-	assertAccepted(t, recorder1)
-	assertQueueFull(t, recorder2)
-}
-
-func assertAccepted(t *testing.T, recorder *httptest.ResponseRecorder) {
-	assert.Equal(t, 202, recorder.Code)
-	assert.Equal(t, "OK\n", recorder.Body.String())
-}
-
-func assertQueueFull(t *testing.T, recorder *httptest.ResponseRecorder) {
-	assert.Equal(t, 200, recorder.Code)
-	assert.Equal(t, "Event queue full\n", recorder.Body.String())
-}
-
-func assertDropped(t *testing.T, recorder *httptest.ResponseRecorder) {
-	assert.Equal(t, 200, recorder.Code)
-}