Merge pull request #25 from factorysh/events

Events

Author: athoune

application/application.go

@@ -6,6 +6,7 @@ import (
 	"path"
 	"path/filepath"
 
+	"github.com/docker/go-events"
 	"github.com/factorysh/microdensity/badge"
 	"github.com/factorysh/microdensity/conf"
 	"github.com/factorysh/microdensity/middlewares/jwt"
@@ -15,6 +16,7 @@ import (
 	"github.com/factorysh/microdensity/run"
 	"github.com/factorysh/microdensity/service"
 	"github.com/factorysh/microdensity/sessions"
+	"github.com/factorysh/microdensity/sink"
 	"github.com/factorysh/microdensity/storage"
 	"github.com/factorysh/microdensity/volumes"
 	"github.com/getsentry/sentry-go"
@@ -36,6 +38,7 @@ type Application struct {
 	volumes       *volumes.Volumes
 	logger        *zap.Logger
 	queue         *queue.Queue
+	Sink          events.Sink
 }
 
 func New(cfg *conf.Conf) (*Application, error) {
@@ -110,6 +113,7 @@ func New(cfg *conf.Conf) (*Application, error) {
 		volumes:       v,
 		logger:        logger,
 		queue:         &q,
+		Sink:          &sink.VoidSink{},
 	}
 	// A good base middleware stack
 	r.Use(middleware.RequestID)

event/event.go

@@ -0,0 +1,12 @@
+package event
+
+import (
+	"github.com/factorysh/microdensity/task"
+	"github.com/google/uuid"
+)
+
+type Event struct {
+	Id    uuid.UUID  `json:"id"`
+	State task.State `json:"state"`
+	Error error      `json:"error,omit_empty"`
+}

go.mod

@@ -48,6 +48,7 @@ require (
 	github.com/docker/docker-credential-helpers v0.6.4 // indirect
 	github.com/docker/go v1.5.1-1.0.20160303222718-d30aec9fd63c // indirect
 	github.com/docker/go-connections v0.4.0 // indirect
+	github.com/docker/go-events v0.0.0-20190806004212-e31b211e4f1c // indirect
 	github.com/docker/go-metrics v0.0.1 // indirect
 	github.com/docker/go-units v0.4.0 // indirect
 	github.com/fvbommel/sortorder v1.0.1 // indirect

queue/queue.go

@@ -4,7 +4,10 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/docker/go-events"
+	"github.com/factorysh/microdensity/event"
 	"github.com/factorysh/microdensity/run"
+	"github.com/factorysh/microdensity/sink"
 	"github.com/factorysh/microdensity/storage"
 	"github.com/factorysh/microdensity/task"
 	"github.com/oleiade/lane"
@@ -34,6 +37,7 @@ type Queue struct {
 	BatchEnded chan bool
 	logger     *zap.Logger
 	working    bool
+	Sink       events.Sink
 }
 
 // NewQueue inits a new queue struct
@@ -50,6 +54,7 @@ func NewQueue(s storage.Storage, runner *run.Runner) Queue {
 		runner:     runner,
 		storage:    s,
 		logger:     logger,
+		Sink:       &sink.VoidSink{},
 	}
 }
 
@@ -76,6 +81,10 @@ func (q *Queue) Put(item *task.Task, env map[string]string) error {
 	queueAdded.Inc()
 	queueSize.Inc()
 	q.logger.Info("queue add", zap.Any("task", item))
+	q.Sink.Write(event.Event{
+		Id:    item.Id,
+		State: item.State,
+	})
 
 	if !q.working {
 		q.logger.Info("Start queue")
@@ -126,6 +135,11 @@ func (q *Queue) DequeueWhile() {
 		} else {
 			t.State = task.Failed
 		}
+		q.Sink.Write(event.Event{
+			Id:    t.Id,
+			State: t.State,
+			Error: err,
+		})
 
 		err = q.storage.Upsert(t)
 		// FIXME: handle err

queue/queue_test.go

@@ -1,17 +1,36 @@
 package queue
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
+	"sync"
 	"testing"
 
+	"github.com/docker/go-events"
 	"github.com/factorysh/microdensity/run"
 	"github.com/factorysh/microdensity/storage"
 	"github.com/factorysh/microdensity/task"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 )
 
+var _ events.Sink = (*DummyEventLogger)(nil)
+
+type DummyEventLogger struct {
+	Cpt *sync.WaitGroup
+}
+
+func (d *DummyEventLogger) Write(evt events.Event) error {
+	d.Cpt.Done()
+	fmt.Println("evt", evt)
+	return nil
+}
+
+func (d *DummyEventLogger) Close() error {
+	return nil
+}
+
 func TestDeq(t *testing.T) {
 	dir, err := ioutil.TempDir(os.TempDir(), "data-")
 	assert.NoError(t, err)
@@ -22,44 +41,42 @@ func TestDeq(t *testing.T) {
 	r, err := run.NewRunner("../demo/services", "/tmp/microdensity/volumes", []string{})
 	assert.NoError(t, err)
 	que := NewQueue(store, r)
+	snk := &DummyEventLogger{
+		Cpt: &sync.WaitGroup{},
+	}
+	que.Sink = snk
+
+	dummySink := &DummyEventLogger{}
+	dummySink.Cpt = &sync.WaitGroup{}
 
 	tsk1 := &task.Task{
 		Id:      uuid.New(),
 		Service: "demo",
 		Project: "beuha",
 	}
-	err = r.Prepare(tsk1, nil)
-	assert.NoError(t, err)
 
 	tsk2 := &task.Task{
 		Id:      uuid.New(),
 		Service: "demo",
 		Project: "alice",
 	}
-	err = r.Prepare(tsk2, nil)
-	assert.NoError(t, err)
 
 	tsk3 := &task.Task{
 		Id:      uuid.New(),
 		Project: "another",
 		Service: "demo",
 	}
-	err = r.Prepare(tsk3, nil)
-	assert.NoError(t, err)
-
-	tsk4 := &task.Task{
-		Id:      uuid.New(),
-		Project: "notprepared",
-		Service: "demo",
-	}
-	assert.NoError(t, err)
 
+	snk.Cpt.Add(4)
 	// FIXME: asserts on state status
-	que.Put(tsk1, nil)
-	que.Put(tsk2, nil)
-	que.Put(tsk3, nil)
-	que.Put(tsk4, nil)
+	err = que.Put(tsk1, nil)
+	assert.NoError(t, err)
+	err = que.Put(tsk2, nil)
+	assert.NoError(t, err)
+	err = que.Put(tsk3, nil)
+	assert.NoError(t, err)
 
 	<-que.BatchEnded
+	//snk.Cpt.Wait()
 
 }

run/run.go

@@ -46,6 +46,8 @@ func NewRunner(servicesDir string, volumesRoot string, hosts []string) (*Runner,
 }
 
 // Prepare the run
+// Prepare is synchronous, in order to raise an error in the REST endpoint.
+// Prepare checks volumes stuff.
 func (r *Runner) Prepare(t *task.Task, env map[string]string) error {
 	if t.Id == uuid.Nil {
 		return fmt.Errorf("task requires an ID to be prepared")

sink/sink.go

@@ -0,0 +1,16 @@
+package sink
+
+import "github.com/docker/go-events"
+
+var _ events.Sink = (*VoidSink)(nil)
+
+type VoidSink struct {
+}
+
+func (v *VoidSink) Write(events.Event) error {
+	return nil
+}
+
+func (v *VoidSink) Close() error {
+	return nil
+}