Repository: allegro/marathon-consul
Pull request #140 (2016-11-16 15:09:41)

Drop not supported events

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

Coverage decreased (-0.3%) to 82.442% when pulling **4ccbc91164a30e299e44e561ab3575a872bd09fc on fixes/#139/drop_not_supported_events** into **a7150d0fbe7138f31552b8dc2f65124c8494c70f on master**.

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

Coverage decreased (-0.02%) to 82.704% when pulling **a0d41bbe99dd29cb80ffb008e72eb51fa8fbfd7b on fixes/#139/drop_not_supported_events** into **a7150d0fbe7138f31552b8dc2f65124c8494c70f on master**.

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

Coverage decreased (-0.3%) to 82.442% when pulling **3b56a96e985e8b6766b680880cf9be4ee0ad81d9 on fixes/#139/drop_not_supported_events** into **a7150d0fbe7138f31552b8dc2f65124c8494c70f on master**.

janisz: Results of synthetic benchmark with 400MB of real events.

```
$ git co master && go test -bench=BenchmarkWebHandler -benchtime=10s ./web -test.benchmem  > old.txt && git co fixes/\#139/drop_not_supported_events  && go test -bench=BenchmarkWebHandler -benchtime=10s ./web -test.benchmem  > new.txt && benchcmp old.txt new.txt

benchmark                 old ns/op     new ns/op     delta
BenchmarkWebHandler-4     808854977     690828083     -14.59%

benchmark                 old allocs     new allocs     delta
BenchmarkWebHandler-4     2628254        1900092        -27.71%

benchmark                 old bytes     new bytes     delta
BenchmarkWebHandler-4     374338713     276159748     -26.23%

```

<details>
Benchmark does only test change in handling events. Does not include speed up with less events pushed to event queue and processing them. More details could be find here https://gist.github.com/janisz/9db4d50fc09f2eba81781af8dbc26a03
```go
package web

import (
	"bytes"
	"net/http/httptest"
	"testing"
	"os"
	"bufio"
	"net/http"
	"io"
)

func BenchmarkWebHandler(b *testing.B) {

	// given
	queue := make(chan event)
	handler := newWebHandler(queue)
	f, err := os.Open("events.jsons")
	if err != nil {
		b.Fatal(err)
	}
	eventsReader := bufio.NewReader(f)

	go func() {
		for {
			<-queue
		}
	}()

	events := make([]*http.Request, 0, 3634)
	for {
		event, _, err := eventsReader.ReadLine()

		if err == io.EOF {
			break
		}
		if err != nil {
			b.Fatal(err)
		}
		req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer(event))
		events = append(events, req)
	}

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		for _, event := range events {
			handler.Handle(httptest.NewRecorder(), event)
		}
	}
}
```
</details>
coveralls: 
[![Coverage Status](https://coveralls.io/builds/8857339/badge)](https://coveralls.io/builds/8857339)

Coverage decreased (-0.2%) to 82.526% when pulling **78a69d377045d83857f45f50007f89a28a27ce86 on fixes/#139/drop_not_supported_events** into **a7150d0fbe7138f31552b8dc2f65124c8494c70f on master**.

kamilchm: It would be good to replace it with implementation of https://github.com/mesosphere/marathon/issues/4637 and use in the future.
coveralls: 
[![Coverage Status](https://coveralls.io/builds/8873586/badge)](https://coveralls.io/builds/8873586)

Coverage decreased (-0.2%) to 82.526% when pulling **a5bf5120d54178fb4404f2837cf63c515ea41caa on fixes/#139/drop_not_supported_events** into **a7150d0fbe7138f31552b8dc2f65124c8494c70f on master**.


Patch:
diff --git a/.travis.yml b/.travis.yml
index 27869ad..e1d3bd5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -12,7 +12,7 @@ before_install:
 script:
   - make deps
   - gometalinter --install
-  - gometalinter ./... --deadline 120s --vendor -D dupl -D gotype -D errcheck -D gas -D golint -E gofmt
+  - gometalinter ./... --deadline 720s --vendor -D dupl -D gotype -D errcheck -D gas -D golint -E gofmt
   - go test -coverprofile=apps.coverprofile ./apps
   - go test -coverprofile=config.coverprofile ./config
   - go test -coverprofile=consul.coverprofile ./consul
diff --git a/README.md b/README.md
index ed2450f..6d16de7 100644
--- a/README.md
+++ b/README.md
@@ -191,6 +191,7 @@ consul-tag                  | `marathon`      | Common tag name added to every s
 consul-timeout              | `3s`            | Time limit for requests made by the Consul HTTP client. A Timeout of zero means no timeout
 consul-token                |                 | The Consul ACL token
 events-queue-size           | `1000`          | Size of events queue
+event-max-size              | `4096`          | Maximum size of event to process (bytes)
 listen                      | `:4000`         | Accept connections at this address
 log-file                    |                 | Save logs to file (e.g.: `/var/log/marathon-consul.log`). If empty logs are published to STDERR
 log-format                  | `text`          |  Log format: JSON, text
diff --git a/config/config.go b/config/config.go
index 996004c..5e45819 100644
--- a/config/config.go
+++ b/config/config.go
@@ -78,6 +78,7 @@ func (config *Config) parseFlags() {
 	flag.StringVar(&config.Web.Listen, "listen", ":4000", "Accept connections at this address")
 	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)")
 
 	// 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 04c2b18..b15a141 100644
--- a/debian/config.json
+++ b/debian/config.json
@@ -21,7 +21,8 @@
   "Web": {
     "Listen": ":4000",
     "QueueSize": 1000,
-    "WorkersCount": 10
+    "WorkersCount": 10,
+    "MaxEventSize": 4096
   },
   "Sync": {
     "Enabled": true,
diff --git a/events/unhealthy_task_kill.go b/events/unhealthy_task_kill.go
deleted file mode 100644
index 771bde1..0000000
--- a/events/unhealthy_task_kill.go
+++ /dev/null
@@ -1,20 +0,0 @@
-package events
-
-import (
-	"encoding/json"
-
-	"github.com/allegro/marathon-consul/apps"
-)
-
-type UnhealthyTaskKilled struct {
-	Timestamp string      `json:"timestamp"`
-	ID        apps.TaskID `json:"taskId"`
-	AppID     apps.AppID  `json:"appId"`
-	Version   string      `json:"version"`
-}
-
-func ParseUnhealthyTaskKilled(event []byte) (*UnhealthyTaskKilled, error) {
-	task := &UnhealthyTaskKilled{}
-	err := json.Unmarshal(event, task)
-	return task, err
-}
diff --git a/web/config.go b/web/config.go
index 93ebc3e..acdcb38 100644
--- a/web/config.go
+++ b/web/config.go
@@ -4,4 +4,5 @@ type Config struct {
 	Listen       string
 	QueueSize    int
 	WorkersCount int
+	MaxEventSize int64
 }
diff --git a/web/event_handler.go b/web/event_handler.go
index fd4d613..9918ee5 100644
--- a/web/event_handler.go
+++ b/web/event_handler.go
@@ -76,8 +76,9 @@ func (fh *eventHandler) handleEvent(eventType string, body []byte) error {
 	case "health_status_changed_event":
 		return fh.handleHealthStatusEvent(body)
 	default:
-		log.WithField("EventType", eventType).Debug("Not handled event type")
-		return nil
+		err := fmt.Errorf("Unsuported event type: %s", eventType)
+		log.WithError(err).WithField("EventType", eventType).Error("This should never happen. Not handled event type")
+		return err
 	}
 }
 
diff --git a/web/web.go b/web/web.go
index 0e9e5e0..967ee20 100644
--- a/web/web.go
+++ b/web/web.go
@@ -18,7 +18,7 @@ func NewHandler(config Config, marathon marathon.Marathoner, serviceOperations s
 		handler := newEventHandler(i, serviceOperations, marathon, eventQueue)
 		stopChannels[i] = handler.start()
 	}
-	return newWebHandler(eventQueue).Handle, stop(stopChannels)
+	return newWebHandler(eventQueue, config.MaxEventSize).Handle, stop(stopChannels)
 }
 
 func stop(channels []chan<- stopEvent) Stop {
diff --git a/web/web_handler.go b/web/web_handler.go
index 4439019..788e44d 100644
--- a/web/web_handler.go
+++ b/web/web_handler.go
@@ -4,55 +4,79 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
-
 	"time"
 
-	log "github.com/Sirupsen/logrus"
 	"github.com/allegro/marathon-consul/events"
 	"github.com/allegro/marathon-consul/metrics"
+
+	log "github.com/Sirupsen/logrus"
 )
 
 type EventHandler struct {
-	eventQueue chan event
+	eventQueue   chan event
+	maxEventSize int64
 }
 
-func newWebHandler(eventQueue chan event) *EventHandler {
-	return &EventHandler{eventQueue: eventQueue}
+func newWebHandler(eventQueue chan 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,
+	}
 }
 
+const (
+	statusUpdateEventType        = "status_update_event"
+	healthStatusChangedEventType = "health_status_changed_event"
+)
+
+// 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() {
-		body, err := ioutil.ReadAll(r.Body)
+		limitedBody := http.MaxBytesReader(w, r.Body, h.maxEventSize)
+		defer limitedBody.Close()
+		body, err := ioutil.ReadAll(limitedBody)
 		if err != nil {
-			log.WithError(err).Debug("Malformed request")
-			handleBadRequest(err, w)
+			drop(err, w)
 			return
 		}
-		log.WithField("Body", string(body)).Debug("Received request")
 
 		eventType, err := events.EventType(body)
 		if err != nil {
-			handleBadRequest(err, w)
+			drop(err, w)
 			return
 		}
 
 		log.WithField("EventType", eventType).Debug("Received event")
 		metrics.Mark("events.requests." + eventType)
 
+		if eventType != statusUpdateEventType && eventType != healthStatusChangedEventType {
+			drop(fmt.Errorf("%s is not supported", eventType), w)
+			return
+		}
+
 		h.eventQueue <- event{eventType: eventType, body: body, timestamp: time.Now()}
-		accepted(w)
-	})
-}
+		accept(w)
 
-func handleBadRequest(err error, w http.ResponseWriter) {
-	metrics.Mark("events.response.error.400")
-	w.WriteHeader(http.StatusBadRequest)
-	log.WithError(err).Debug("Returning 400 due to malformed request")
-	fmt.Fprintln(w, err.Error())
+	})
 }
 
-func accepted(w http.ResponseWriter) {
-	metrics.Mark("events.response.202")
+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 2ef4506..2759575 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -105,6 +105,7 @@ func TestConfig_ShouldBeMergedWithFileDefaultsAndFlags(t *testing.T) {
 			Listen:       ":4000",
 			QueueSize:    1000,
 			WorkersCount: 10,
+			MaxEventSize: 4096,
 		},
 		Sync: sync.Config{
 			Interval: 15 * time.Minute,
diff --git a/events/unhealthy_task_kill_test.go b/events/unhealthy_task_kill_test.go
deleted file mode 100644
index 926d9e6..0000000
--- a/events/unhealthy_task_kill_test.go
+++ /dev/null
@@ -1,27 +0,0 @@
-package events
-
-import (
-	"encoding/json"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-var unhealthyTaskKill = &UnhealthyTaskKilled{
-	AppID: "/my-app",
-	ID:    "my-app_0-1396592784349",
-}
-
-func TestUnhealthTaskKillParseTask(t *testing.T) {
-	t.Parallel()
-
-	jsonified, err := json.Marshal(unhealthyTaskKill)
-	assert.Nil(t, err)
-
-	service, err := ParseUnhealthyTaskKilled(jsonified)
-	assert.Nil(t, err)
-
-	assert.Equal(t, unhealthyTaskKill.Timestamp, service.Timestamp)
-	assert.Equal(t, unhealthyTaskKill.ID, service.ID)
-	assert.Equal(t, unhealthyTaskKill.AppID, service.AppID)
-}
diff --git a/web/event_handler_test.go b/web/event_handler_test.go
index 93075c5..501346a 100644
--- a/web/event_handler_test.go
+++ b/web/event_handler_test.go
@@ -5,7 +5,6 @@ import (
 	"errors"
 	"net/http"
 	"net/http/httptest"
-	"strings"
 	"testing"
 
 	"github.com/allegro/marathon-consul/apps"
@@ -15,12 +14,14 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestWebHandler_NotHandleUnknownEventType(t *testing.T) {
+const maxEventSize = 4096
+
+func TestWebHandler_DropUnknownEventType(t *testing.T) {
 	t.Parallel()
 
 	// given
 	queue := make(chan event)
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	stopChan := newEventHandler(0, nil, nil, queue).start()
 	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(`{"eventType":"test_event"}`)))
 
@@ -30,14 +31,14 @@ func TestWebHandler_NotHandleUnknownEventType(t *testing.T) {
 	stopChan <- stopEvent{}
 
 	// then
-	assertAccepted(t, recorder)
+	assertDropped(t, recorder)
 }
 
-func TestWebHandler_HandleRadderError(t *testing.T) {
+func TestWebHandler_HandleReadderError(t *testing.T) {
 	t.Parallel()
 
 	// given
-	handler := newWebHandler(nil)
+	handler := newWebHandler(nil, maxEventSize)
 	req, _ := http.NewRequest("POST", "/events", BadReader{})
 
 	// when
@@ -45,47 +46,46 @@ func TestWebHandler_HandleRadderError(t *testing.T) {
 	handler.Handle(recorder, req)
 
 	// then
-	assert.Equal(t, 400, recorder.Code)
+	assert.Equal(t, 200, recorder.Code)
 	assert.Equal(t, "Some error\n", recorder.Body.String())
 }
 
-func TestWebHandler_HandleEmptyBody(t *testing.T) {
+func TestWebHandler_DropBigBody(t *testing.T) {
 	t.Parallel()
 
 	// given
-	handler := newWebHandler(nil)
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte{}))
+	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, 400, recorder.Code)
-	assert.Equal(t, "unexpected end of JSON input\n", recorder.Body.String())
+	assert.Equal(t, 200, recorder.Code)
+	assert.Equal(t, "http: request body too large\n", recorder.Body.String())
 }
 
-func TestWebHandler_NotHandleMalformedEventType(t *testing.T) {
+func TestWebHandler_DropEmptyBody(t *testing.T) {
 	t.Parallel()
 
 	// given
-	handler := newWebHandler(nil)
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(`{eventType:"test_event"}`)))
+	handler := newWebHandler(nil, maxEventSize)
+	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte{}))
 
 	// when
 	recorder := httptest.NewRecorder()
 	handler.Handle(recorder, req)
 
 	// then
-	assert.Equal(t, 400, recorder.Code)
-	assert.Equal(t, "invalid character 'e' looking for beginning of object key string\n", recorder.Body.String())
+	assertDropped(t, recorder)
 }
 
-func TestWebHandler_HandleMalformedEventType(t *testing.T) {
+func TestWebHandler_DropMalformedEventType(t *testing.T) {
 	t.Parallel()
 
 	// given
-	handler := newWebHandler(nil)
+	handler := newWebHandler(nil, maxEventSize)
 	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(`{eventType:"test_event"}`)))
 
 	// when
@@ -93,15 +93,14 @@ func TestWebHandler_HandleMalformedEventType(t *testing.T) {
 	handler.Handle(recorder, req)
 
 	// then
-	assert.Equal(t, 400, recorder.Code)
-	assert.Equal(t, "invalid character 'e' looking for beginning of object key string\n", recorder.Body.String())
+	assertDropped(t, recorder)
 }
 
-func TestWebHandler_NotHandleInvalidEventType(t *testing.T) {
+func TestWebHandler_DropInvalidEventType(t *testing.T) {
 	t.Parallel()
 
 	// given
-	handler := newWebHandler(nil)
+	handler := newWebHandler(nil, maxEventSize)
 	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(`{"eventType":[1,2]}`)))
 
 	// when
@@ -109,18 +108,16 @@ func TestWebHandler_NotHandleInvalidEventType(t *testing.T) {
 	handler.Handle(recorder, req)
 
 	// then
-	assert.Equal(t, 400, recorder.Code)
-
-	assert.True(t, strings.HasPrefix(recorder.Body.String(), "json: cannot unmarshal array"))
+	assertDropped(t, recorder)
 
 }
 
-func TestWebHandler_HandleAppInvalidBody(t *testing.T) {
+func TestWebHandler_DropAppInvalidBody(t *testing.T) {
 	t.Parallel()
 
 	// given
 	queue := make(chan event)
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	stopChan := newEventHandler(0, nil, nil, queue).start()
 	body := `{"type":  "app_terminated_event", "appID": 123}`
 	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
@@ -131,8 +128,7 @@ func TestWebHandler_HandleAppInvalidBody(t *testing.T) {
 	stopChan <- stopEvent{}
 
 	// then
-	assert.Equal(t, 400, recorder.Code)
-	assert.Equal(t, "no event\n", recorder.Body.String())
+	assertDropped(t, recorder)
 }
 
 func TestWebHandler_NotHandleStatusEventWithInvalidBody(t *testing.T) {
@@ -140,7 +136,7 @@ func TestWebHandler_NotHandleStatusEventWithInvalidBody(t *testing.T) {
 
 	// given
 	queue := make(chan event)
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	stopChan := newEventHandler(0, nil, nil, queue).start()
 	body := `{
 	  "slaveId":"85e59460-a99e-4f16-b91f-145e0ea595bd-S0",
@@ -172,7 +168,7 @@ func TestWebHandler_NotHandleStatusEventAboutStartingTask(t *testing.T) {
 	services := consul.NewConsulStub()
 	queue := make(chan event)
 	stopChan := newEventHandler(0, services, nil, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	ignoredTaskStatuses := []string{"TASK_STAGING", "TASK_STARTING", "TASK_RUNNING", "unknown"}
 	for _, taskStatus := range ignoredTaskStatuses {
 		body := `{
@@ -215,7 +211,7 @@ func TestWebHandler_HandleStatusEventAboutDeadTask(t *testing.T) {
 		}
 		queue := make(chan event)
 		stopChan := newEventHandler(0, service, nil, queue).start()
-		handler := newWebHandler(queue)
+		handler := newWebHandler(queue, maxEventSize)
 		body := `{
 		  "slaveId":"85e59460-a99e-4f16-b91f-145e0ea595bd-S0",
 		  "taskId":"` + app.Tasks[1].ID.String() + `",
@@ -256,7 +252,7 @@ func TestWebHandler_HandleStatusEventAboutDeadTaskErrOnDeregistration(t *testing
 	service.FailDeregisterByTaskForID("task.1")
 	queue := make(chan event)
 	stopChan := newEventHandler(0, service, nil, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	body := `{
 	  "slaveId":"85e59460-a99e-4f16-b91f-145e0ea595bd-S0",
 	  "taskId":"task.1",
@@ -291,7 +287,7 @@ func TestWebHandler_NotHandleStatusEventAboutNonConsulAppsDeadTask(t *testing.T)
 	service := consul.NewConsulStub()
 	queue := make(chan event)
 	stopChan := newEventHandler(0, service, nil, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	taskStatuses := []string{"TASK_FINISHED", "TASK_FAILED", "TASK_KILLED", "TASK_LOST"}
 	for _, taskStatus := range taskStatuses {
 		body := `{
@@ -329,7 +325,7 @@ func TestWebHandler_NotHandleHealthStatusEventWhenAppHasNotConsulLabel(t *testin
 	marathon := marathon.MarathonerStubForApps(app)
 	queue := make(chan event)
 	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	body := healthStatusChangeEventForTask("test_app.1")
 	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
@@ -399,7 +395,7 @@ func TestWebHandler_NotHandleHealthStatusEventForTaskWithNotAllHealthChecksPasse
 	marathon := marathon.MarathonerStubForApps(app)
 	queue := make(chan event)
 	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	body := healthStatusChangeEventForTask("test_app.1")
 	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
@@ -422,7 +418,7 @@ func TestWebHandler_NotHandleHealthStatusEventForTaskWithNoHealthCheck(t *testin
 	marathon := marathon.MarathonerStubForApps(app)
 	queue := make(chan event)
 	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	body := healthStatusChangeEventForTask("/test/app.0")
 	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
@@ -442,7 +438,7 @@ func TestWebHandler_NotHandleHealthStatusEventWhenTaskIsNotAlive(t *testing.T) {
 	// given
 	queue := make(chan event)
 	stopChan := newEventHandler(0, nil, nil, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	body := `{
 	  "appId":"/test/app",
 	  "taskId":"test_app.1",
@@ -468,7 +464,7 @@ func TestWebHandler_NotHandleHealthStatusEventWhenBodyIsInvalid(t *testing.T) {
 	// given
 	queue := make(chan event)
 	stopChan := newEventHandler(0, nil, nil, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	body := `{
 	  "appId":"/test/app",
 	  "taskId":"test_app.1",
@@ -496,7 +492,7 @@ func TestWebHandler_HandleHealthStatusEventReturn202WhenMarathonReturnedError(t
 	marathon := marathon.MarathonerStubForApps(app)
 	queue := make(chan event)
 	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	body := `{
 	  "appId":"unknown",
 	  "taskId":"unknown.1",
@@ -525,7 +521,7 @@ func TestWebHandler_HandleHealthStatusEventWhenTaskIsNotInMarathon(t *testing.T)
 	marathon := marathon.MarathonerStubForApps(app)
 	queue := make(chan event)
 	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue)
+	handler := newWebHandler(queue, maxEventSize)
 	body := healthStatusChangeEventForTask("unknown.1")
 	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
@@ -539,19 +535,9 @@ func TestWebHandler_HandleHealthStatusEventWhenTaskIsNotInMarathon(t *testing.T)
 	assert.True(t, marathon.Interactions())
 }
 
-func newConsulStubWithApplicationsTasksRegistered(applications ...*apps.App) *consul.Stub {
-	service := consul.NewConsulStub()
-	for _, app := range applications {
-		for _, task := range app.Tasks {
-			service.Register(&task, app)
-		}
-	}
-	return service
-}
-
 type BadReader struct{}
 
-func (r BadReader) Read(p []byte) (n int, err error) {
+func (r BadReader) Read(p []byte) (int, error) {
 	return 0, errors.New("Some error")
 }
 
@@ -570,3 +556,7 @@ func assertAccepted(t *testing.T, recorder *httptest.ResponseRecorder) {
 	assert.Equal(t, 202, recorder.Code)
 	assert.Equal(t, "OK\n", recorder.Body.String())
 }
+
+func assertDropped(t *testing.T, recorder *httptest.ResponseRecorder) {
+	assert.Equal(t, 200, recorder.Code)
+}