Can reverse sha-ed job type

app.go

@@ -69,7 +69,8 @@ func NewApp(cfg *config.Config) App {
 	})
 
 	router.GET("/api/jobs/:jobid/logs", func(c *gin.Context) {
-		stream, err := docker.Logs(c.Param("jobid"))
+		// Check ID exists
+		_, err := docker.GetContainer(c.Param("jobid"))
 		if err != nil {
 			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 			return
@@ -81,6 +82,16 @@ func NewApp(cfg *config.Config) App {
 		}
 		defer ws.Close(websocket.StatusInternalError, "the sky is falling")
 
+		for !docker.HasStarted(c.Param("jobid")) {
+			time.Sleep(500 * time.Millisecond)
+		}
+
+		stream, err := docker.Logs(c.Param("jobid"))
+		if err != nil {
+			ws.Close(websocket.StatusNormalClosure, "")
+			log.Fatal("Something goes wrong during log access:", err)
+		}
+
 		r, w := io.Pipe()
 
 		ctx, cancel := context.WithTimeout(c.Request.Context(), time.Second*10)
@@ -119,6 +130,7 @@ func (app *App) Start() {
 		Handler: app.router,
 	}
 
+	log.Printf("Ready, listening on %s\n", app.cfg.Bind)
 	if err := app.srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
 		log.Fatalf("listen: %s\n", err)
 	}

engine/docker/docker.go

@@ -16,17 +16,17 @@ func newCli() (*client.Client, error) {
 	return client.NewClientWithOpts(client.FromEnv)
 }
 
-func Ps() ([]types.Container, error) {
+func Ps(opts types.ContainerListOptions) ([]types.Container, error) {
 	cli, err := newCli()
 	if err != nil {
 		return nil, err
 	}
 
-	return cli.ContainerList(context.Background(), types.ContainerListOptions{})
+	return cli.ContainerList(context.Background(), opts)
 }
 
 func PsName() (ret map[string]string, err error) {
-	containers, err := Ps()
+	containers, err := Ps(types.ContainerListOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -41,6 +41,24 @@ func PsName() (ret map[string]string, err error) {
 	return ret, err
 }
 
+func GetContainer(containerID string) (ctr types.ContainerJSON, err error) {
+	cli, err := newCli()
+	if err != nil {
+		return types.ContainerJSON{}, err
+	}
+
+	return cli.ContainerInspect(context.Background(), containerID)
+}
+
+func HasStarted(containerID string) bool {
+	ctr, err := GetContainer(containerID)
+	if err != nil {
+		return false
+	}
+
+	return ctr.State.Status == "running" || ctr.State.Status == "paused" || ctr.State.Status == "restarting" || ctr.State.Status == "removing" || ctr.State.Status == "exited" || ctr.State.Status == "dead"
+}
+
 func Create(jobtype string, image string, cmd []string, mounts []mount.Mount) (string, error) {
 	cli, err := newCli()
 	if err != nil {

engine/queue.go

@@ -3,11 +3,36 @@ package engine
 import (
 	"crypto/sha256"
 	"fmt"
+	"log"
 	"strings"
+
+	"github.com/nemunaire/minifaas/jobs"
 )
 
+const CTR_NAME_PREFIX = "minifaas"
+
 func GenContainerPrefix(jobtype string) string {
-	return fmt.Sprintf("minifaas-%x-", sha256.Sum224([]byte(jobtype)))
+	return fmt.Sprintf("%s-%x-", CTR_NAME_PREFIX, sha256.Sum224([]byte(jobtype)))
+}
+
+func ParseContainerName(name string) (jobtype, id string, err error) {
+	if !strings.HasPrefix(strings.TrimPrefix(name, "/"), CTR_NAME_PREFIX+"-") {
+		return "", "", fmt.Errorf("This is not a %s job: starting with %q", CTR_NAME_PREFIX, name)
+	}
+
+	tmp := strings.Split(name, "-")
+	if len(tmp) < 3 {
+		return "", "", fmt.Errorf("This is not a %s job: %q didn't has at least 3 args", CTR_NAME_PREFIX, name)
+	}
+
+	jobtype = jobs.GetJobType(tmp[1])
+	if jobtype == "" {
+		return "", "", fmt.Errorf("This is not a %s job: unknown job type %q", CTR_NAME_PREFIX, tmp[1])
+	}
+
+	id = strings.Join(tmp[2:], "-")
+
+	return
 }
 
 func FilterRunningContainers(jobtype string, ctrs map[string]string) (ret []string) {
@@ -25,7 +50,10 @@ func FilterRunningContainers(jobtype string, ctrs map[string]string) (ret []stri
 func CountRunningContainers(jobtype string, ctrs map[string]string) (n int) {
 	prefix := GenContainerPrefix(jobtype)
 
+	log.Println(ctrs)
+
 	for cname, _ := range ctrs {
+		log.Println(strings.TrimPrefix(cname, "/"), prefix)
 		if jobtype == "" || strings.HasPrefix(strings.TrimPrefix(cname, "/"), prefix) {
 			n += 1
 		}

jobs.go

@@ -1,30 +1,17 @@
 package main
 
 import (
-	"fmt"
-
 	"github.com/docker/docker/api/types/mount"
 
 	"github.com/nemunaire/minifaas/engine/docker"
+	"github.com/nemunaire/minifaas/jobs"
 )
 
-type Job struct {
-	Image     string
-	Cmd       []string
-	DataMount bool
-}
-
-var jobs = map[string]Job{
-	"counter": {
-		Image:     "alpine",
-		Cmd:       []string{"sh", "-c", "touch /data/work_done; for i in `seq 10`; do echo $i; sleep 0.5; done"},
-		DataMount: true,
-	},
-}
-
 func RunJob(jobtype string) (string, error) {
+	job := jobs.GetJob(jobtype)
+
 	var mnts []mount.Mount
-	if jobs[jobtype].DataMount {
+	if job.DataMount {
 		myVolume, err := docker.CreateVolumeDir("/data", false)
 		if err != nil {
 			return "", err
@@ -33,29 +20,21 @@ func RunJob(jobtype string) (string, error) {
 	}
 
 	// Check if the image is here
-	hasimg, err := docker.HasImage(jobs[jobtype].Image)
+	hasimg, err := docker.HasImage(job.Image)
 	if err != nil {
 		return "", err
 	}
 	if !hasimg {
-		err = docker.PullImage(jobs[jobtype].Image)
+		err = docker.PullImage(job.Image)
 		if err != nil {
 			return "", err
 		}
 	}
 
-	n, err := CountRunningContainers(jobtype)
-	if err != nil {
-		return "", err
-	}
-	if n > 0 {
-		return "", fmt.Errorf("Please wait, there is already a similar job running.")
-	}
-
-	return docker.Run(
+	return docker.Create(
 		jobtype,
-		jobs[jobtype].Image,
-		jobs[jobtype].Cmd,
+		job.Image,
+		job.Cmd,
 		mnts,
 	)
 }

jobs/jobs.go

@@ -0,0 +1,41 @@
+package jobs
+
+import (
+	"crypto/sha256"
+	"fmt"
+)
+
+type Job struct {
+	Image     string
+	Cmd       []string
+	DataMount bool
+}
+
+var jobs = map[string]Job{
+	"counter": {
+		Image:     "alpine",
+		Cmd:       []string{"sh", "-c", "touch /data/work_done; for i in `seq 10`; do echo $i; sleep 0.5; done"},
+		DataMount: true,
+	},
+}
+
+var revJobs = map[string]string{}
+
+func init() {
+	for jobtype := range jobs {
+		revJobs[fmt.Sprintf("%x", sha256.Sum224([]byte(jobtype)))] = jobtype
+	}
+}
+
+func GetJobType(hashJobType string) (jobtype string) {
+	var ok bool
+	if jobtype, ok = revJobs[hashJobType]; !ok {
+		jobtype = ""
+	}
+
+	return
+}
+
+func GetJob(jobtype string) (job Job) {
+	return jobs[jobtype]
+}

main.go

@@ -18,11 +18,12 @@ func main() {
 	a := NewApp(cfg)
 	go a.Start()
 
+	go runQueue()
+
 	quit := make(chan os.Signal)
 	signal.Notify(quit, os.Interrupt, syscall.SIGTERM)
 	<-quit
 	log.Println("Stopping the service...")
 	a.Stop()
 	log.Println("Stopped")
-
 }

queue.go

@@ -1,6 +1,12 @@
 package main
 
 import (
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/docker/docker/api/types"
+
 	"github.com/nemunaire/minifaas/engine"
 	"github.com/nemunaire/minifaas/engine/docker"
 )
@@ -22,3 +28,50 @@ func CountRunningContainers(jobtype string) (n int, err error) {
 
 	return engine.CountRunningContainers(jobtype, ctrs), nil
 }
+
+func GetJobTypeFromNames(names []string) (jobtype string, err error) {
+	for _, name := range names {
+		if jobtype, _, err = engine.ParseContainerName(name); err == nil {
+			// We found it, jobtype is already defined, just return
+			return
+		}
+	}
+
+	// Not returned before == not found
+	return "", fmt.Errorf("This is not a minifass job")
+}
+
+func runQueue() {
+	for {
+		time.Sleep(time.Second)
+		//log.Println("queue: tick")
+
+		ctrs, err := docker.Ps(types.ContainerListOptions{All: true})
+		if err != nil {
+			log.Println("queue: run:", err.Error())
+			continue
+		}
+
+		for _, ctr := range ctrs {
+			if ctr.State == "created" {
+				jobtype, err := GetJobTypeFromNames(ctr.Names)
+				if err != nil {
+					log.Println(err.Error())
+					continue
+				}
+
+				n, err := CountRunningContainers(jobtype)
+				if err != nil {
+					continue
+				}
+				if n > 0 {
+					log.Printf("Waiting slot for container %s (jobtype=%s)", ctr.ID, jobtype)
+					continue
+				}
+
+				log.Printf("Starting container %s (jobtype=%s) on available slot", ctr.ID, jobtype)
+				docker.Start(ctr.ID)
+			}
+		}
+	}
+}