Merge pull request #47 from factorysh/features/restart

Handle proper restarts

Author: athoune

application/application.go

@@ -1,10 +1,14 @@
 package application
 
 import (
+	"context"
 	"net/http"
 	"os"
+	"os/signal"
 	"path"
 	"path/filepath"
+	"syscall"
+	"time"
 
 	"github.com/docker/go-events"
 	"github.com/factorysh/microdensity/badge"
@@ -18,6 +22,7 @@ import (
 	"github.com/factorysh/microdensity/sessions"
 	"github.com/factorysh/microdensity/sink"
 	"github.com/factorysh/microdensity/storage"
+	"github.com/factorysh/microdensity/task"
 	"github.com/factorysh/microdensity/volumes"
 	"github.com/getsentry/sentry-go"
 	"github.com/go-chi/chi/v5"
@@ -39,6 +44,8 @@ type Application struct {
 	logger        *zap.Logger
 	queue         *queue.Queue
 	Sink          events.Sink
+	Server        *http.Server
+	Stopper       chan (os.Signal)
 }
 
 func New(cfg *conf.Conf) (*Application, error) {
@@ -114,6 +121,7 @@ func New(cfg *conf.Conf) (*Application, error) {
 		logger:        logger,
 		queue:         &q,
 		Sink:          &sink.VoidSink{},
+		Stopper:       make(chan os.Signal, 1),
 	}
 	// A good base middleware stack
 	r.Use(middleware.RequestID)
@@ -202,3 +210,90 @@ func (a *Application) ListServices() []string {
 
 	return list
 }
+
+// Run make the app listen and serve requests
+func (a *Application) Run(listen string) error {
+	// listen for stop/restart signals and sends them to the stopper channel
+	signal.Notify(a.Stopper, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	// setup the router
+	a.Server = &http.Server{
+		Addr:    listen,
+		Handler: a.Router,
+	}
+
+	// interrupted tasks becomes ready task and are added to queue
+	tasks, err := a.storage.All()
+	if err != nil {
+		return err
+	}
+
+	for _, t := range tasks {
+		if t.State == task.Ready || t.State == task.Interrupted {
+			t.State = task.Ready
+			parsedArgs, err := a.Services[t.Service].Validate(t.Args)
+			// non blocking error
+			if err != nil {
+				t.State = task.Failed
+				err := a.storage.Upsert(t)
+				if err != nil {
+					a.logger.Fatal("unable to save task", zap.Error(err))
+				}
+
+				a.logger.Error("error when validating task args", zap.String("task", t.Id.String()))
+				continue
+			}
+
+			err = a.addTask(t, parsedArgs.Environments)
+			// non blocking error
+			if err != nil {
+				a.logger.Error("error when adding task", zap.Error(err))
+			}
+		}
+	}
+
+	// start and serve
+	go func() {
+		if err := a.Server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			a.logger.Fatal("fatal error when running server", zap.Error(err))
+		}
+	}()
+
+	return nil
+}
+
+// Shutdown the server and put running tasks into interrupted state
+func (a *Application) Shutdown() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	// handle the last pending requests
+	// we don't want this error stop the shutdown flow
+	// just log it
+	err := a.Server.Shutdown(ctx)
+	if err != nil {
+		a.logger.Error("error on server shutdown", zap.Error(err))
+	}
+
+	tasks, err := a.storage.All()
+	if err != nil {
+		return err
+	}
+
+	for _, t := range tasks {
+		// running tasks becomes interrupted tasks
+		if t.State == task.Running {
+			// TODO: send a cancel request to docker ?
+			t.State = task.Interrupted
+			err := a.storage.Upsert(t)
+			// same here, non blocking error
+			if err != nil {
+				a.logger.Error("error when updating task status", zap.Error(err), zap.String("task id", t.Id.String()))
+			}
+		}
+	}
+
+	a.logger.Info("server shutdown")
+
+	return nil
+}

application/application_test.go

@@ -10,6 +10,7 @@ import (
 	"net/http/httptest"
 	"net/url"
 	"os"
+	"path"
 	"path/filepath"
 	"testing"
 	"time"
@@ -19,6 +20,8 @@ import (
 	"github.com/factorysh/microdensity/conf"
 	"github.com/factorysh/microdensity/mockup"
 	"github.com/factorysh/microdensity/service"
+	"github.com/factorysh/microdensity/storage"
+	"github.com/factorysh/microdensity/task"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -187,6 +190,70 @@ func TestApplication(t *testing.T) {
 
 }
 
+func TestApplicationStart(t *testing.T) {
+	// state 5 is interrupted
+	rawJSON := `{"id":"f79b5c4c-94b4-11ec-a442-00163e007d68","service":"waiter","project":"group%2Fproject","branch":"master","commit":"7e15b158cfc3e8f6bbe3e441a0cdb64bba135ef3","creation":"2022-02-23T15:29:11.082288364+01:00","Args":{"WAIT":10},"State":5}`
+
+	gitlab := httptest.NewServer(mockup.GitlabJWK(&key.PublicKey))
+	defer gitlab.Close()
+
+	cfg, cb, err := SpawnConfig(gitlab.URL)
+	defer cb()
+	assert.NoError(t, err)
+
+	a, err := New(cfg)
+	assert.NoError(t, err)
+
+	taskPath := path.Join(cfg.DataPath, "data", "wait", "group%2Fproject", "master", "f79b5c4c-94b4-11ec-a442-00163e007d68")
+	err = os.MkdirAll(taskPath, storage.DirMode)
+	assert.NoError(t, err)
+
+	err = os.WriteFile(path.Join(taskPath, "task.json"), []byte(rawJSON), 0644)
+	assert.NoError(t, err)
+
+	err = a.Run(":9090")
+	assert.NoError(t, err)
+
+	tasks, err := a.storage.All()
+	assert.NoError(t, err)
+	assert.Equal(t, task.Running, tasks[0].State)
+
+	err = a.Shutdown()
+	assert.NoError(t, err)
+}
+
+func TestApplicationStop(t *testing.T) {
+	// state 5 is interrupted
+	rawJSON := `{"id":"f79b5c4c-94b4-11ec-a442-00163e007d68","service":"waiter","project":"group%2Fproject","branch":"master","commit":"7e15b158cfc3e8f6bbe3e441a0cdb64bba135ef3","creation":"2022-02-23T15:29:11.082288364+01:00","Args":{"WAIT":10},"State":1}`
+
+	gitlab := httptest.NewServer(mockup.GitlabJWK(&key.PublicKey))
+	defer gitlab.Close()
+
+	cfg, cb, err := SpawnConfig(gitlab.URL)
+	defer cb()
+	assert.NoError(t, err)
+
+	a, err := New(cfg)
+	assert.NoError(t, err)
+
+	taskPath := path.Join(cfg.DataPath, "data", "wait", "group%2Fproject", "master", "f79b5c4c-94b4-11ec-a442-00163e007d68")
+	err = os.MkdirAll(taskPath, storage.DirMode)
+	assert.NoError(t, err)
+
+	err = os.WriteFile(path.Join(taskPath, "task.json"), []byte(rawJSON), 0644)
+	assert.NoError(t, err)
+
+	err = a.Run(":9090")
+	assert.NoError(t, err)
+
+	err = a.Shutdown()
+	assert.NoError(t, err)
+
+	tasks, err := a.storage.All()
+	assert.NoError(t, err)
+	assert.Equal(t, task.Interrupted, tasks[0].State)
+}
+
 func SpawnConfig(gitlabURL string) (*conf.Conf, func(), error) {
 	dataDir, err := ioutil.TempDir(os.TempDir(), "data")
 	if err != nil {

application/badge.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/factorysh/microdensity/badge"
 	_badge "github.com/factorysh/microdensity/badge"
+	"github.com/factorysh/microdensity/task"
 	"github.com/go-chi/chi/v5"
 	"go.uber.org/zap"
 )
@@ -35,6 +36,7 @@ func (a *Application) BadgeMyTaskHandler(latest bool) http.HandlerFunc {
 		t, err := a.storage.GetByCommit(service, project, branch, commit, latest)
 		if err != nil {
 			l.Warn("Task get error", zap.Error(err))
+			badge.WriteBadge(service, "not found", _badge.Colors.Default, w)
 			w.WriteHeader(http.StatusNotFound)
 			return
 		}
@@ -43,6 +45,12 @@ func (a *Application) BadgeMyTaskHandler(latest bool) http.HandlerFunc {
 		p := filepath.Join(a.storage.GetVolumePath(t), "/data", fmt.Sprintf("%s.badge", bdg))
 		_, err = os.Stat(p)
 
+		// if running return early
+		if t.State == task.Running {
+			badge.WriteBadge(service, t.State.String(), _badge.Colors.Get(t.State), w)
+			return
+		}
+
 		// if not found
 		if err != nil {
 			// fallback to status badge
@@ -51,7 +59,7 @@ func (a *Application) BadgeMyTaskHandler(latest bool) http.HandlerFunc {
 				badge.WriteBadge(service, t.State.String(), _badge.Colors.Get(t.State), w)
 				return
 			}
-			w.WriteHeader(http.StatusBadRequest)
+			w.WriteHeader(http.StatusInternalServerError)
 			return
 		}
 		l = l.With(zap.String("path", p))

application/task.go

@@ -85,29 +85,40 @@ func (a *Application) PostTaskHandler(w http.ResponseWriter, r *http.Request) {
 		Args:     args,
 		State:    task.Ready,
 	}
-	err = a.storage.Upsert(t)
+
+	err = a.addTask(t, parsedArgs.Environments)
 	if err != nil {
-		l.Warn("Queue error", zap.Error(err))
-		panic(err)
+		l.Error("error when adding task", zap.String("task", t.Id.String()), zap.Error(err))
+		return
 	}
-	err = a.storage.EnsureVolumesDir(t)
+	render.JSON(w, r, map[string]string{
+		"id": id.String(),
+	})
+}
+
+// addTask adds a task to a queue
+func (a *Application) addTask(t *task.Task, args map[string]string) error {
+	err := a.storage.EnsureVolumesDir(t)
 	if err != nil {
-		l.Warn("Volume creation", zap.Error(err))
-		panic(err)
+		return err
+	}
+
+	err = a.storage.Upsert(t)
+	if err != nil {
+		return err
 	}
-	err = a.queue.Put(t, parsedArgs.Environments)
+
+	err = a.queue.Put(t, args)
 	if err != nil {
-		l.Warn("Task prepare/put", zap.Error(err))
-		panic(err)
+		return err
 	}
+
 	err = a.storage.SetLatest(t)
 	if err != nil {
-		l.Warn("Task set latest", zap.Error(err))
-		panic(err)
+		return err
 	}
-	render.JSON(w, r, map[string]string{
-		"id": id.String(),
-	})
+
+	return err
 }
 
 // TaskHandler show a Task

demo/services/waiter/README.md

@@ -0,0 +1 @@
+# Demo service

demo/services/waiter/docker-compose.yml

@@ -0,0 +1,15 @@
+---
+
+services:
+  background:
+    image: busybox
+    command: sleep 50000
+  hello:
+    depends_on:
+      - background
+    image: busybox
+    command: >-
+      && sleep ${WAIT}
+      && echo '{\"color\": \"lime\", \"subject\":\"wait\", \"status\":\"${WAIT}\"}' > /data/demo.badge"
+    volumes:
+      - "./data:/data"

demo/services/waiter/meta.js

@@ -0,0 +1,13 @@
+function validate(params) {
+  if (!("WAIT" in params)) {
+    throw "WAIT argument is mandatory";
+  }
+  if (!Number.isInteger(params.WAIT)) {
+    throw `WAIT is only numbers : [${params.WAIT}]`;
+  }
+  return {
+    environments: {
+      WAIT: params.WAIT,
+    },
+  };
+}

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/docker/cli v20.10.12+incompatible
 	github.com/docker/compose/v2 v2.2.3
 	github.com/docker/docker v20.10.12+incompatible
+	github.com/docker/go-events v0.0.0-20190806004212-e31b211e4f1c
 	github.com/dop251/goja v0.0.0-20220214123719-b09a6bfa842f
 	github.com/getsentry/sentry-go v0.12.0
 	github.com/go-chi/chi/v5 v5.0.7
@@ -40,6 +41,7 @@ require (
 	github.com/containerd/console v1.0.3 // indirect
 	github.com/containerd/containerd v1.5.9 // indirect
 	github.com/containerd/continuity v0.1.0 // indirect
+	github.com/containerd/fifo v1.0.0 // indirect
 	github.com/containerd/typeurl v1.0.2 // indirect
 	github.com/distribution/distribution/v3 v3.0.0-20210316161203-a01c71e2477e // indirect
 	github.com/dlclark/regexp2 v1.4.1-0.20201116162257-a2a8dda75c91 // indirect
@@ -48,7 +50,6 @@ 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

main.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
-	"net/http"
 	"os"
 
 	"github.com/factorysh/microdensity/application"
@@ -52,6 +51,19 @@ func main() {
 		os.Exit(1)
 	}
 
-	l.Info("starting")
-	http.ListenAndServe(cfg.Listen, a.Router)
+	logger.Info("starting")
+	err = a.Run(cfg.Listen)
+	if err != nil {
+		l.Error("Run", zap.Error(err))
+		os.Exit(1)
+	}
+
+	<-a.Stopper
+
+	logger.Info("shutdown signal received")
+
+	err = a.Shutdown()
+	if err != nil {
+		l.Error("error on shutdown", zap.Error(err))
+	}
 }