Repository: allegro/marathon-consul
Pull request #154 (2016-12-26 20:12:38)

Nonblocking events
In addition to adding nonblocking handling of events i added:

* `make format` and `make check` that runs same metalinter command as travis.ci
* bumped Consul version to 0.7.2 to fix random segfaults on Mac
* fixed race condition detected by `go test -race` in `MarathonerStub`
* fixed race condition in `even_hander_test` that caused tests to fail randomly (2nd commit)

Bit more about test fixes. I moved `web_handler` test suite to `web_handler_test` and it tests only `web_handler`, the event queue is stubbed. `event_handler` tests on the other hand no longer start `web_handler`, instead events that should be processed are sent directly to the event queue. This fixes a non-trivial race condition in tests, where success of test depended on wether request will be processed by `web_handler` and sent to the queue before test reaches synchronisation point at blocking send to `event_handler` `stopChan`. 

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

Coverage decreased (-1.01%) to 81.403% when pulling **9933ee606bdeff1e7ed5985eae27cbe4b56d0682 on nonblocking_events** into **df6a8a7346d26b1d24d5b68312c00f5d14b33679 on master**.

adamdubiel: @janisz i added requested changes: split change into commits and `goimports` everywhere

Patch:
diff --git a/Makefile b/Makefile
index 2b4ece6..693fb41 100644
--- a/Makefile
+++ b/Makefile
@@ -1,8 +1,11 @@
+SHELL := /bin/bash
+
 PACKAGES=$(shell go list ./... | grep -v /vendor/)
 TESTARGS?=-race
 CURRENT_DIR = $(shell pwd)
 SOURCEDIR = $(CURRENT_DIR)
 SOURCES := $(shell find $(SOURCEDIR) -name '*.go')
+APP_SOURCES := $(shell find $(SOURCEDIR) -name '*.go' -not -path '$(SOURCEDIR)/vendor/*')
 VERSION=$(shell cat .goxc.json | python -c "import json,sys;obj=json.load(sys.stdin);print obj['PackageVersion'];")
 TEMPDIR := $(shell mktemp -d)
 LD_FLAGS=-ldflags '-w' -ldflags "-X main.VERSION=$(VERSION)"
@@ -27,6 +30,16 @@ test: deps $(SOURCES)
 	PATH=$(CURRENT_DIR)/bin:$(PATH) go test $(PACKAGES) $(TESTARGS)
 	go vet $(PACKAGES)
 
+gometalint-exists:
+	@which gometalinter > /dev/null || \
+	(echo >&2 "gometalinter must be installed on the system. See https://github.com/alecthomas/gometalinter")
+
+check: gometalint-exists $(SOURCES)
+	gometalinter . --deadline  720s --vendor -D dupl -D gotype -D errcheck -D gas -D golint -E gofmt
+
+format:
+	goimports -w -l $(APP_SOURCES)
+
 FPM-exists:
 	@fpm -v || \
 	(echo >&2 "FPM must be installed on the system. See https://github.com/jordansissel/fpm"; false)
diff --git a/consul/consul_stub.go b/consul/consul_stub.go
index 681e0b1..877eb53 100644
--- a/consul/consul_stub.go
+++ b/consul/consul_stub.go
@@ -157,8 +157,8 @@ func (c *Stub) servicesMatchingTask(taskID apps.TaskID) []*consulapi.AgentServic
 	return matching
 }
 
-func (c *Stub) RegisteredTaskIDs() []apps.TaskID {
-	services, _ := c.GetAllServices()
+func (c *Stub) RegisteredTaskIDs(serviceName string) []apps.TaskID {
+	services, _ := c.GetServices(serviceName)
 	taskIds := []apps.TaskID{}
 	for _, s := range services {
 		taskID, _ := s.TaskId()
diff --git a/install_consul.sh b/install_consul.sh
index 88aecae..4dbda83 100755
--- a/install_consul.sh
+++ b/install_consul.sh
@@ -20,7 +20,7 @@ hash consul 2>/dev/null || {
         os="windows"
         arch="386"
     fi
-    version="0.7.0"
+    version="0.7.2"
     archive="consul_${version}_${os}_${arch}.zip"
 
     mkdir -p bin
diff --git a/marathon/marathon_stub.go b/marathon/marathon_stub.go
index 7eb1e28..3b9204a 100644
--- a/marathon/marathon_stub.go
+++ b/marathon/marathon_stub.go
@@ -2,25 +2,27 @@ package marathon
 
 import (
 	"errors"
+	"sync"
 
 	"github.com/allegro/marathon-consul/apps"
 )
 
 type MarathonerStub struct {
-	AppsStub     []*apps.App
-	AppStub      map[apps.AppID]*apps.App
-	TasksStub    map[apps.AppID][]*apps.Task
-	leader       string
-	interactions bool
+	AppsStub       []*apps.App
+	AppStub        map[apps.AppID]*apps.App
+	TasksStub      map[apps.AppID][]*apps.Task
+	leader         string
+	interactionsMu sync.RWMutex
+	interactions   bool
 }
 
 func (m *MarathonerStub) ConsulApps() ([]*apps.App, error) {
-	m.interactions = true
+	m.noteInteraction()
 	return m.AppsStub, nil
 }
 
 func (m *MarathonerStub) App(id apps.AppID) (*apps.App, error) {
-	m.interactions = true
+	m.noteInteraction()
 	if app, ok := m.AppStub[id]; ok {
 		return app, nil
 	}
@@ -28,7 +30,7 @@ func (m *MarathonerStub) App(id apps.AppID) (*apps.App, error) {
 }
 
 func (m *MarathonerStub) Tasks(appID apps.AppID) ([]*apps.Task, error) {
-	m.interactions = true
+	m.noteInteraction()
 	if app, ok := m.TasksStub[appID]; ok {
 		return app, nil
 	}
@@ -36,14 +38,22 @@ func (m *MarathonerStub) Tasks(appID apps.AppID) ([]*apps.Task, error) {
 }
 
 func (m *MarathonerStub) Leader() (string, error) {
-	m.interactions = true
+	m.noteInteraction()
 	return m.leader, nil
 }
 
-func (m MarathonerStub) Interactions() bool {
+func (m *MarathonerStub) Interactions() bool {
+	m.interactionsMu.RLock()
+	defer m.interactionsMu.RUnlock()
 	return m.interactions
 }
 
+func (m *MarathonerStub) noteInteraction() {
+	m.interactionsMu.Lock()
+	defer m.interactionsMu.Unlock()
+	m.interactions = true
+}
+
 func MarathonerStubWithLeaderForApps(leader string, args ...*apps.App) *MarathonerStub {
 	stub := MarathonerStubForApps(args...)
 	stub.leader = leader
diff --git a/web/event_handler.go b/web/event_handler.go
index af5ad54..12a3b66 100644
--- a/web/event_handler.go
+++ b/web/event_handler.go
@@ -55,7 +55,17 @@ func (fh *eventHandler) start() chan<- stopEvent {
 			select {
 			case e = <-fh.eventQueue:
 				metrics.Mark(fmt.Sprintf("events.handler.%d", fh.id))
-				metrics.UpdateGauge("events.queue.len", int64(len(fh.eventQueue)))
+
+				queueLength := int64(len(fh.eventQueue))
+				metrics.UpdateGauge("events.queue.len", queueLength)
+				queueCapacity := int64(cap(fh.eventQueue))
+
+				utilization := int64(0)
+				if queueCapacity > 0 {
+					utilization = 100 * (queueLength / queueCapacity)
+				}
+				metrics.UpdateGauge("events.queue.util", utilization)
+
 				metrics.UpdateGauge("events.queue.delay_ns", time.Since(e.timestamp).Nanoseconds())
 				metrics.Time("events.processing."+e.eventType, process)
 			case <-quitChan:
diff --git a/web/web_handler.go b/web/web_handler.go
index b4b01b5..9eed4ef 100644
--- a/web/web_handler.go
+++ b/web/web_handler.go
@@ -1,15 +1,15 @@
 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"
-
-	log "github.com/Sirupsen/logrus"
 )
 
 type EventHandler struct {
@@ -64,8 +64,13 @@ func (h *EventHandler) Handle(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		h.eventQueue <- event{eventType: e.Type, body: body, timestamp: time.Now()}
-		accept(w)
+		select {
+		case h.eventQueue <- event{eventType: e.Type, body: body, timestamp: time.Now()}:
+			accept(w)
+		default:
+			metrics.Mark("events.queue.drop")
+			drop(errors.New("Event queue full"), w)
+		}
 
 	})
 }


Test patch:
diff --git a/consul/consul_stub_test.go b/consul/consul_stub_test.go
index 935916a..2dbed1b 100644
--- a/consul/consul_stub_test.go
+++ b/consul/consul_stub_test.go
@@ -39,7 +39,7 @@ func TestConsulStub(t *testing.T) {
 	// when
 	err = consul.DeregisterByTask(app.Tasks[1].ID)
 	services, _ = consul.GetAllServices()
-	taskIds := consul.RegisteredTaskIDs()
+	taskIds := consul.RegisteredTaskIDs("test")
 
 	// then
 	assert.NoError(t, err)
@@ -70,7 +70,7 @@ func TestConsulStub(t *testing.T) {
 
 	// then
 	assert.NoError(t, err)
-	assert.Len(t, consul.RegisteredTaskIDs(), 1)
+	assert.Len(t, consul.RegisteredTaskIDs("test"), 1)
 
 	// when
 	app = utils.ConsulApp("other", 2)
diff --git a/time/time_test.go b/time/time_test.go
index db1f52f..04a5ab5 100644
--- a/time/time_test.go
+++ b/time/time_test.go
@@ -2,9 +2,10 @@ package time
 
 import (
 	"encoding/json"
-	"github.com/stretchr/testify/assert"
 	"testing"
 	gotime "time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestTimestampParsing(t *testing.T) {
diff --git a/web/event_handler_test.go b/web/event_handler_test.go
index 501346a..c6bb5eb 100644
--- a/web/event_handler_test.go
+++ b/web/event_handler_test.go
@@ -1,144 +1,42 @@
 package web
 
 import (
-	"bytes"
 	"errors"
-	"net/http"
-	"net/http/httptest"
+	"strconv"
 	"testing"
+	"time"
 
 	"github.com/allegro/marathon-consul/apps"
 	"github.com/allegro/marathon-consul/consul"
 	"github.com/allegro/marathon-consul/marathon"
+	"github.com/allegro/marathon-consul/service"
 	. "github.com/allegro/marathon-consul/utils"
 	"github.com/stretchr/testify/assert"
 )
 
-const maxEventSize = 4096
-
-func TestWebHandler_DropUnknownEventType(t *testing.T) {
-	t.Parallel()
-
-	// given
-	queue := make(chan event)
-	handler := newWebHandler(queue, maxEventSize)
-	stopChan := newEventHandler(0, nil, nil, queue).start()
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(`{"eventType":"test_event"}`)))
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
-
-	// then
-	assertDropped(t, recorder)
-}
-
-func TestWebHandler_HandleReadderError(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_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)
-
+type handlerStubs struct {
+	serviceRegistry service.ServiceRegistry
+	marathon        marathon.Marathoner
 }
 
-func TestWebHandler_DropAppInvalidBody(t *testing.T) {
-	t.Parallel()
-
-	// given
+// Creates eventHandler and returns nonbuffered event queue that has to be used to send events to handler and
+// function that can be used as a synchronization point to wait until previous event has been processed.
+// Under the hood synchronization function simply sends a stop signal to the handlers stopChan.
+func testEventHandler(stubs handlerStubs) (chan<- event, func()) {
 	queue := make(chan event)
-	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)))
-
-	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	awaitChan := newEventHandler(0, stubs.serviceRegistry, stubs.marathon, queue).start()
 
-	// then
-	assertDropped(t, recorder)
+	return queue, func() { awaitChan <- stopEvent{} }
 }
 
-func TestWebHandler_NotHandleStatusEventWithInvalidBody(t *testing.T) {
+func TestEventHandler_NotHandleStatusEventWithInvalidBody(t *testing.T) {
 	t.Parallel()
 
 	// given
-	queue := make(chan event)
-	handler := newWebHandler(queue, maxEventSize)
-	stopChan := newEventHandler(0, nil, nil, queue).start()
-	body := `{
+	serviceRegistry := consul.NewConsulStub()
+	queue, awaitFunc := testEventHandler(handlerStubs{serviceRegistry: serviceRegistry})
+
+	body := []byte(`{
 	  "slaveId":"85e59460-a99e-4f16-b91f-145e0ea595bd-S0",
 	  "taskId":"python_simple.4a7e99d0-9cc1-11e5-b4d8-0a0027000004",
 	  "taskStatus":"TASK_KILLED",
@@ -149,29 +47,26 @@ func TestWebHandler_NotHandleStatusEventWithInvalidBody(t *testing.T) {
 	  "version":"2015-12-07T09:02:48.981Z",
 	  "eventType":"status_update_event",
 	  "timestamp":"2015-12-07T09:02:49.934Z"
-	}`
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
+	}`)
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "status_update_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
+	assert.Empty(t, serviceRegistry.RegisteredTaskIDs("test_app"))
 }
 
-func TestWebHandler_NotHandleStatusEventAboutStartingTask(t *testing.T) {
+func TestEventHandler_NotHandleStatusEventAboutStartingTask(t *testing.T) {
 	t.Parallel()
 
 	// given
-	services := consul.NewConsulStub()
-	queue := make(chan event)
-	stopChan := newEventHandler(0, services, nil, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
+	serviceRegistry := consul.NewConsulStub()
+	queue, awaitFunc := testEventHandler(handlerStubs{serviceRegistry: serviceRegistry})
+
 	ignoredTaskStatuses := []string{"TASK_STAGING", "TASK_STARTING", "TASK_RUNNING", "unknown"}
 	for _, taskStatus := range ignoredTaskStatuses {
-		body := `{
+		body := []byte(`{
 		  "slaveId":"85e59460-a99e-4f16-b91f-145e0ea595bd-S0",
 		  "taskId":"python_simple.4a7e99d0-9cc1-11e5-b4d8-0a0027000004",
 		  "taskStatus":"` + taskStatus + `",
@@ -184,40 +79,42 @@ func TestWebHandler_NotHandleStatusEventAboutStartingTask(t *testing.T) {
 		  "version":"2015-12-07T09:02:48.981Z",
 		  "eventType":"status_update_event",
 		  "timestamp":"2015-12-07T09:02:49.934Z"
-		}`
-		req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
+		}`)
 
 		// when
-		recorder := httptest.NewRecorder()
-		handler.Handle(recorder, req)
-		stopChan <- stopEvent{}
+		queue <- event{eventType: "status_update_event", timestamp: time.Now(), body: body}
+		awaitFunc()
 
 		// then
-		assert.Equal(t, 202, recorder.Code)
-		assert.Equal(t, "OK\n", recorder.Body.String())
-		assert.Empty(t, services.RegisteredTaskIDs())
+		assert.Empty(t, serviceRegistry.RegisteredTaskIDs("test_app"))
 	}
 }
 
-func TestWebHandler_HandleStatusEventAboutDeadTask(t *testing.T) {
+func TestEventHandler_HandleStatusEventAboutDeadTask(t *testing.T) {
 	t.Parallel()
+
+	// given
+	serviceRegistry := consul.NewConsulStub()
+
+	queue, awaitFunc := testEventHandler(handlerStubs{serviceRegistry: serviceRegistry})
+
 	taskStatuses := []string{"TASK_FINISHED", "TASK_FAILED", "TASK_KILLED", "TASK_LOST"}
-	for _, taskStatus := range taskStatuses {
+	for index, taskStatus := range taskStatuses {
 		// given
-		app := ConsulApp("/test/app", 3)
-		service := consul.NewConsulStub()
+		appId := "/test/app" + strconv.Itoa(index)
+		serviceName := "test.app" + strconv.Itoa(index)
+
+		app := ConsulApp(appId, 3)
 		for _, task := range app.Tasks {
-			service.Register(&task, app)
+			serviceRegistry.Register(&task, app)
 		}
-		queue := make(chan event)
-		stopChan := newEventHandler(0, service, nil, queue).start()
-		handler := newWebHandler(queue, maxEventSize)
-		body := `{
+
+		body := []byte(`{
 		  "slaveId":"85e59460-a99e-4f16-b91f-145e0ea595bd-S0",
 		  "taskId":"` + app.Tasks[1].ID.String() + `",
 		  "taskStatus":"` + taskStatus + `",
 		  "message":"Command terminated with signal Terminated",
-		  "appId":"/test/app",
+		  "appId":"` + appId + `",
 		  "host":"localhost",
 		  "ports":[
 			31372
@@ -225,18 +122,14 @@ func TestWebHandler_HandleStatusEventAboutDeadTask(t *testing.T) {
 		  "version":"2015-12-07T09:02:48.981Z",
 		  "eventType":"status_update_event",
 		  "timestamp":"2015-12-07T09:33:40.898Z"
-		}`
-		req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
+		}`)
 
 		// when
-		recorder := httptest.NewRecorder()
-		handler.Handle(recorder, req)
-		stopChan <- stopEvent{}
-		taskIds := service.RegisteredTaskIDs()
+		queue <- event{eventType: "status_update_event", timestamp: time.Now(), body: body}
+		awaitFunc()
 
 		// then
-		assert.Equal(t, 202, recorder.Code)
-		assert.Equal(t, "OK\n", recorder.Body.String())
+		taskIds := serviceRegistry.RegisteredTaskIDs(serviceName)
 		assert.Len(t, taskIds, 2)
 		assert.NotContains(t, taskIds, app.Tasks[1].ID)
 		assert.Contains(t, taskIds, app.Tasks[0].ID)
@@ -244,16 +137,16 @@ func TestWebHandler_HandleStatusEventAboutDeadTask(t *testing.T) {
 	}
 }
 
-func TestWebHandler_HandleStatusEventAboutDeadTaskErrOnDeregistration(t *testing.T) {
+func TestEventHandler_HandleStatusEventAboutDeadTaskErrOnDeregistration(t *testing.T) {
 	t.Parallel()
 
 	// given
-	service := consul.NewConsulStub()
-	service.FailDeregisterByTaskForID("task.1")
-	queue := make(chan event)
-	stopChan := newEventHandler(0, service, nil, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
-	body := `{
+	serviceRegistry := consul.NewConsulStub()
+	serviceRegistry.FailDeregisterByTaskForID("task.1")
+
+	queue, awaitFunc := testEventHandler(handlerStubs{serviceRegistry: serviceRegistry})
+
+	body := []byte(`{
 	  "slaveId":"85e59460-a99e-4f16-b91f-145e0ea595bd-S0",
 	  "taskId":"task.1",
 	  "taskStatus":"TASK_KILLED",
@@ -266,31 +159,29 @@ func TestWebHandler_HandleStatusEventAboutDeadTaskErrOnDeregistration(t *testing
 	  "version":"2015-12-07T09:02:48.981Z",
 	  "eventType":"status_update_event",
 	  "timestamp":"2015-12-07T09:33:40.898Z"
-	}`
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
+	}`)
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "status_update_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
-	assert.Empty(t, service.RegisteredTaskIDs())
+	assert.Empty(t, serviceRegistry.RegisteredTaskIDs("test_app"))
 }
 
-func TestWebHandler_NotHandleStatusEventAboutNonConsulAppsDeadTask(t *testing.T) {
+func TestEventHandler_NotHandleStatusEventAboutNonConsulAppsDeadTask(t *testing.T) {
 	t.Parallel()
 
 	// given
 	app := NonConsulApp("/test/app", 3)
-	service := consul.NewConsulStub()
-	queue := make(chan event)
-	stopChan := newEventHandler(0, service, nil, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
+	marathon := marathon.MarathonerStubForApps(app)
+	serviceRegistry := consul.NewConsulStub()
+
+	queue, awaitFunc := testEventHandler(handlerStubs{serviceRegistry: serviceRegistry, marathon: marathon})
+
 	taskStatuses := []string{"TASK_FINISHED", "TASK_FAILED", "TASK_KILLED", "TASK_LOST"}
 	for _, taskStatus := range taskStatuses {
-		body := `{
+		body := []byte(`{
 		  "slaveId":"85e59460-a99e-4f16-b91f-145e0ea595bd-S0",
 		  "taskId":"` + app.Tasks[1].ID.String() + `",
 		  "taskStatus":"` + taskStatus + `",
@@ -303,235 +194,214 @@ func TestWebHandler_NotHandleStatusEventAboutNonConsulAppsDeadTask(t *testing.T)
 		  "version":"2015-12-07T09:02:48.981Z",
 		  "eventType":"status_update_event",
 		  "timestamp":"2015-12-07T09:33:40.898Z"
-		}`
-		req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
+		}`)
 
 		// when
-		recorder := httptest.NewRecorder()
-		handler.Handle(recorder, req)
-		stopChan <- stopEvent{}
+		queue <- event{eventType: "status_update_event", timestamp: time.Now(), body: body}
+		awaitFunc()
 
 		// then
-		assert.Equal(t, 202, recorder.Code)
-		assert.Equal(t, "OK\n", recorder.Body.String())
+		assert.False(t, marathon.Interactions())
 	}
 }
 
-func TestWebHandler_NotHandleHealthStatusEventWhenAppHasNotConsulLabel(t *testing.T) {
+func TestEventHandler_NotHandleHealthStatusEventWhenAppHasNotConsulLabel(t *testing.T) {
 	t.Parallel()
 
 	// given
 	app := NonConsulApp("/test/app", 3)
 	marathon := marathon.MarathonerStubForApps(app)
-	queue := make(chan event)
-	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
+
+	queue, awaitFunc := testEventHandler(handlerStubs{marathon: marathon})
+
 	body := healthStatusChangeEventForTask("test_app.1")
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
 	assert.True(t, marathon.Interactions())
 }
 
-func TestWebHandler_HandleHealthStatusEvent(t *testing.T) {
+func TestEventHandler_HandleHealthStatusEvent(t *testing.T) {
 	t.Parallel()
 
 	// given
 	app := ConsulApp("/test/app", 3)
+
 	marathon := marathon.MarathonerStubForApps(app)
-	service := consul.NewConsulStub()
-	handle, stop := NewHandler(Config{WorkersCount: 1}, marathon, service)
+	serviceRegistry := consul.NewConsulStub()
+
+	queue, awaitFunc := testEventHandler(handlerStubs{serviceRegistry: serviceRegistry, marathon: marathon})
 	body := healthStatusChangeEventForTask("test_app.1")
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
 	// when
-	recorder := httptest.NewRecorder()
-	handle(recorder, req)
-	stop()
-	taskIds := service.RegisteredTaskIDs()
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
+	taskIds := serviceRegistry.RegisteredTaskIDs("test.app")
 	assert.Len(t, taskIds, 1)
 	assert.Contains(t, taskIds, app.Tasks[1].ID)
 	assert.True(t, marathon.Interactions())
 }
 
-func TestWebHandler_HandleHealthStatusEventWithErrorsOnRegistration(t *testing.T) {
+func TestEventHandler_HandleHealthStatusEventWithErrorsOnRegistration(t *testing.T) {
 	t.Parallel()
 
 	// given
 	app := ConsulApp("/test/app", 3)
+
 	marathon := marathon.MarathonerStubForApps(app)
-	service := consul.NewConsulStub()
-	service.FailRegisterForID(app.Tasks[1].ID)
-	handle, stop := NewHandler(Config{WorkersCount: 1}, marathon, service)
+	serviceRegistry := consul.NewConsulStub()
+	serviceRegistry.FailRegisterForID(app.Tasks[1].ID)
+
+	queue, awaitFunc := testEventHandler(handlerStubs{serviceRegistry: serviceRegistry, marathon: marathon})
 	body := healthStatusChangeEventForTask("test_app.1")
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
 	// when
-	recorder := httptest.NewRecorder()
-	handle(recorder, req)
-	stop()
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
-	assert.Empty(t, service.RegisteredTaskIDs())
+	assert.Empty(t, serviceRegistry.RegisteredTaskIDs("test.app"))
 	assert.True(t, marathon.Interactions())
 }
 
-func TestWebHandler_NotHandleHealthStatusEventForTaskWithNotAllHealthChecksPassed(t *testing.T) {
+func TestEventHandler_NotHandleHealthStatusEventForTaskWithNotAllHealthChecksPassed(t *testing.T) {
 	t.Parallel()
 
 	// given
 	app := ConsulApp("/test/app", 3)
 	app.Tasks[1].HealthCheckResults = []apps.HealthCheckResult{{Alive: true}, {Alive: false}}
 	marathon := marathon.MarathonerStubForApps(app)
-	queue := make(chan event)
-	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
+
+	queue, awaitFunc := testEventHandler(handlerStubs{marathon: marathon})
 	body := healthStatusChangeEventForTask("test_app.1")
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
 	assert.True(t, marathon.Interactions())
 }
 
-func TestWebHandler_NotHandleHealthStatusEventForTaskWithNoHealthCheck(t *testing.T) {
+func TestEventHandler_NotHandleHealthStatusEventForTaskWithNoHealthCheck(t *testing.T) {
 	t.Parallel()
 
 	// given
 	app := ConsulApp("/test/app", 1)
 	app.Tasks[0].HealthCheckResults = []apps.HealthCheckResult{}
 	marathon := marathon.MarathonerStubForApps(app)
-	queue := make(chan event)
-	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
+
+	queue, awaitFunc := testEventHandler(handlerStubs{marathon: marathon})
+
 	body := healthStatusChangeEventForTask("/test/app.0")
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
 	assert.True(t, marathon.Interactions())
 }
 
-func TestWebHandler_NotHandleHealthStatusEventWhenTaskIsNotAlive(t *testing.T) {
+func TestEventHandler_NotHandleHealthStatusEventWhenTaskIsNotAlive(t *testing.T) {
 	t.Parallel()
 
 	// given
-	queue := make(chan event)
-	stopChan := newEventHandler(0, nil, nil, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
-	body := `{
+	app := ConsulApp("/test/app", 1)
+	marathon := marathon.MarathonerStubForApps(app)
+
+	queue, awaitFunc := testEventHandler(handlerStubs{marathon: marathon})
+
+	body := []byte(`{
 	  "appId":"/test/app",
 	  "taskId":"test_app.1",
 	  "version":"2015-12-07T09:02:48.981Z",
 	  "alive":false,
 	  "eventType":"health_status_changed_event",
 	  "timestamp":"2015-12-07T09:33:50.069Z"
-	}`
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
+	}`)
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
+	assert.False(t, marathon.Interactions())
 }
 
-func TestWebHandler_NotHandleHealthStatusEventWhenBodyIsInvalid(t *testing.T) {
+func TestEventHandler_NotHandleHealthStatusEventWhenBodyIsInvalid(t *testing.T) {
 	t.Parallel()
 
 	// given
-	queue := make(chan event)
-	stopChan := newEventHandler(0, nil, nil, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
-	body := `{
+	app := ConsulApp("/test/app", 1)
+	marathon := marathon.MarathonerStubForApps(app)
+
+	queue, awaitFunc := testEventHandler(handlerStubs{marathon: marathon})
+
+	body := []byte(`{
 	  "appId":"/test/app",
 	  "taskId":"test_app.1",
 	  "version":123,
 	  "alive":false,
 	  "eventType":"health_status_changed_event",
 	  "timestamp":"2015-12-07T09:33:50.069Z"
-	}`
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
+	}`)
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
+	assert.False(t, marathon.Interactions())
 }
 
-func TestWebHandler_HandleHealthStatusEventReturn202WhenMarathonReturnedError(t *testing.T) {
+func TestEventHandler_HandleHealthStatusEventReturn202WhenMarathonReturnedError(t *testing.T) {
 	t.Parallel()
 
 	// given
 	app := ConsulApp("/test/app", 3)
 	marathon := marathon.MarathonerStubForApps(app)
-	queue := make(chan event)
-	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
-	body := `{
+
+	queue, awaitFunc := testEventHandler(handlerStubs{marathon: marathon})
+
+	body := []byte(`{
 	  "appId":"unknown",
 	  "taskId":"unknown.1",
 	  "version":"2015-12-07T09:02:48.981Z",
 	  "alive":true,
 	  "eventType":"health_status_changed_event",
 	  "timestamp":"2015-12-07T09:33:50.069Z"
-	}`
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
+	}`)
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
 	assert.True(t, marathon.Interactions())
 }
 
-func TestWebHandler_HandleHealthStatusEventWhenTaskIsNotInMarathon(t *testing.T) {
+func TestEventHandler_HandleHealthStatusEventWhenTaskIsNotInMarathon(t *testing.T) {
 	t.Parallel()
 
 	// given
 	app := ConsulApp("/test/app", 3)
 	marathon := marathon.MarathonerStubForApps(app)
-	queue := make(chan event)
-	stopChan := newEventHandler(0, nil, marathon, queue).start()
-	handler := newWebHandler(queue, maxEventSize)
+
+	queue, awaitFunc := testEventHandler(handlerStubs{marathon: marathon})
+
 	body := healthStatusChangeEventForTask("unknown.1")
-	req, _ := http.NewRequest("POST", "/events", bytes.NewBuffer([]byte(body)))
 
 	// when
-	recorder := httptest.NewRecorder()
-	handler.Handle(recorder, req)
-	stopChan <- stopEvent{}
+	queue <- event{eventType: "health_status_changed_event", timestamp: time.Now(), body: body}
+	awaitFunc()
 
 	// then
-	assertAccepted(t, recorder)
 	assert.True(t, marathon.Interactions())
 }
 
@@ -541,22 +411,13 @@ func (r BadReader) Read(p []byte) (int, error) {
 	return 0, errors.New("Some error")
 }
 
-func healthStatusChangeEventForTask(taskID string) string {
-	return `{
+func healthStatusChangeEventForTask(taskID string) []byte {
+	return []byte(`{
 	  "appId":"/test/app",
 	  "taskId":"` + taskID + `",
 	  "version":"2015-12-07T09:02:48.981Z",
 	  "alive":true,
 	  "eventType":"health_status_changed_event",
 	  "timestamp":"2015-12-07T09:33:50.069Z"
-	}`
-}
-
-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)
+	}`)
 }
diff --git a/web/web_handler_test.go b/web/web_handler_test.go
new file mode 100644
index 0000000..1fa8b92
--- /dev/null
+++ b/web/web_handler_test.go
@@ -0,0 +1,180 @@
+package web
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const maxEventSize = 4096
+
+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 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 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)
+}