Reorganize docker configuration

This removes an existing unlocked shared access to runner.running.
This also sets us up for better management of the runners.

docker/cursorius/Dockerfile.dev

@@ -4,4 +4,4 @@ MAINTAINER restitux <restitux@ohea.xyz>
 RUN apt-get update && apt-get install -y \
     ca-certificates \
     && rm -rf /var/lib/apt/lists/*
-ENTRYPOINT ["/build/server/docker/build-and-run.sh"]
+ENTRYPOINT ["/build/server/docker/cursorius/build-and-run.sh"]

docker/docker-compose.yml

@@ -2,15 +2,15 @@ version: "3.3"
 services:
   cursorius-server:
     build:
-      context: ..
-      dockerfile: docker/Dockerfile.dev
+      context: ".."
+      dockerfile: "docker/cursorius/Dockerfile.dev"
     ports:
       - "0.0.0.0:45420:45420"
     networks:
       - cursorius
     volumes:
       - "..:/build/server"
-      - "../server.toml:/root/.config/cursorius/server.toml"
+      - "./server.toml:/root/.config/cursorius/server.toml"
       - "/var/run/docker.sock:/var/run/docker.sock"
       - "../_working/go:/go"
       - "../_working/jobs:/cursorius/jobs"
@@ -20,15 +20,30 @@ services:
       - POSTGRES_USER=cursorius
       - POSTGRES_PASSWORD=cursorius
       - POSTGRES_DB=cursorius
+    volumes:
+      - "../_working/postgres:/var/lib/postgresql/data"
     networks:
       - cursorius
   graphiql:
     build:
-      dockerfile: Dockerfile.graphiql
+      context: "graphiql"
+      dockerfile: "Dockerfile.graphiql"
     ports:
       - "0.0.0.0:45421:80"
     networks:
       - cursorius
+  gitea:
+    image: gitea/gitea:latest
+    profiles: ["gitea"]
+    environment:
+      - GITEA__webhook__ALLOWED_HOST_LIST=cursorius-server, external
+    ports:
+      - "127.0.0.1:2222:22"
+      - "127.0.0.1:3000:3000"
+    networks:
+      - cursorius
+    volumes:
+      - "../_working/gitea:/data"
 
 networks:
   cursorius:

docker/gitea-override.yml

@@ -1,24 +0,0 @@
-version: "3.3"
-services:
-  cursorius-server:
-    networks:
-      - gitea
-  gitea:
-    image: gitea/gitea:latest
-    environment:
-      - GITEA__webhook__ALLOWED_HOST_LIST=cursorius-server, external
-    ports:
-      - "127.0.0.1:2222:22"
-      - "127.0.0.1:3000:3000"
-    networks:
-      - gitea
-    volumes:
-      - gitea-data:/data
-
-
-volumes:
-  gitea-data:
-
-networks:
-  gitea:
-    external: false

docker/run-dev.sh

@@ -2,104 +2,4 @@
 
 set -e
 
-mkdir -p _working/go
-mkdir -p _working/jobs
-
-base_default_compose_files="docker/docker-compose.yml"
-default_compose_files="$base_default_compose_files"
-
-override_compose="docker/docker-compose.override.yml"
-gitea_compose="docker/gitea-override.yml"
-
-
-if [ -f "$override_compose" ]
-then
-    default_compose_files+=" $override_compose"
-else
-    default_compose_files="docker/docker-compose.yml"
-fi
-
-function stop_containers {
-    current_containers="$(cat _working/current_containers)"
-    if [ "$current_containers" == "default" ]
-    then
-        compose_files="$default_compose_files"
-    elif [ "$current_containers" == "gitea" ]
-    then
-        compose_files="$default_compose_files $gitea_compose"
-    fi
-    compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
-    docker compose $compose_file_flags down
-}
-
-function show_logs {
-    current_containers="$(cat _working/current_containers)"
-    if [ "$current_containers" == "default" ]
-    then
-        compose_files="$default_compose_files"
-    elif [ "$current_containers" == "gitea" ]
-    then
-        compose_files="$default_compose_files $gitea_compose"
-    fi
-    compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
-    docker compose $compose_file_flags logs -f
-}
-
-function show_ps {
-    current_containers="$(cat _working/current_containers)"
-    if [ "$current_containers" == "default" ]
-    then
-        compose_files="$default_compose_files"
-    elif [ "$current_containers" == "gitea" ]
-    then
-        compose_files="$default_compose_files $gitea_compose"
-    fi
-    compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
-    docker compose $compose_file_flags ps
-}
-
-function runprev {
-     current_containers="$(cat _working/current_containers)"
-    if [ "$current_containers" == "default" ]
-    then
-        compose_files="$default_compose_files"
-    elif [ "$current_containers" == "gitea" ]
-    then
-        compose_files="$default_compose_files $gitea_compose"
-    fi
-    compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
-    docker compose $compose_file_flags up --build -d
-    docker compose $compose_file_flags logs -f
-}
-
-
-case $1 in
-    "default")
-        echo "default" > _working/current_containers
-        compose_files="$default_compose_files"
-        compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
-        docker compose $compose_file_flags up --build -d
-        docker compose $compose_file_flags logs -f;;
-    "gitea")
-        echo "gitea" > _working/current_containers
-        stop_containers
-        compose_files="$default_compose_files $gitea_compose"
-        compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
-        docker compose $compose_file_flags up --build -d
-        docker compose $compose_file_flags logs -f;;
-    "dbshell")
-        compose_files="$default_compose_files $gitea_compose"
-        compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
-        docker compose $compose_file_flags exec cursorius-db psql --user=cursorius;;
-    "stop")
-        stop_containers;;
-    "logs")
-        show_logs;;
-    "ps")
-        show_ps;;
-    "runprev")
-        runprev;;
-    *) echo "ERROR: Unknown param \"$1\"" 2>&1
-       exit 255;;
-esac
-
+go run docker/run.go "$@"

docker/run.go

@@ -0,0 +1,93 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+)
+
+func panicError(errorString string, params ...any) {
+	fmt.Fprintf(os.Stderr, fmt.Sprintf("ERROR: %v\n", errorString), params...)
+	os.Exit(1)
+}
+
+func run(name string, arg ...string) {
+	cmd := exec.Command(name, arg...)
+	if err := cmd.Run(); err != nil {
+		panicError("could not run command %v: %v", name, err)
+	}
+}
+
+func runAttach(name string, arg ...string) {
+	cmd := exec.Command(name, arg...)
+	cmd.Stdin = os.Stdin
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	if err := cmd.Run(); err != nil {
+		panicError("could not run command %v: %v", name, err)
+	}
+}
+
+func runCompose(args []string) {
+	runAttach("docker", append([]string{"compose"}, args...)...)
+}
+
+func createDirs() {
+	run("mkdir", "-p", "_working/go")
+	run("mkdir", "-p", "_working/jobs")
+}
+
+func currentContainers() string {
+	bytes, err := os.ReadFile("_working/current_containers")
+	if err != nil {
+		panicError("could not read current containers: %v", err)
+	}
+	return string(bytes)
+}
+
+func composeFlags() []string {
+	containers := currentContainers()
+	flags := []string{"-f", "docker/docker-compose.yml"}
+	switch containers {
+	case "gitea":
+		flags = append(flags, "--profile", "gitea")
+	}
+	return flags
+}
+
+func runContainers(containers string) {
+	err := os.WriteFile("_working/current_containers", []byte(containers), 0633)
+	if err != nil {
+		panicError("could not write current_containers file: %v", err)
+	}
+	runCompose(append(composeFlags(), "up", "--build", "-d"))
+	runCompose(append(composeFlags(), "logs", "-f"))
+}
+
+func main() {
+	if len(os.Args) < 2 {
+		panicError("not enough arguments passed")
+	}
+
+	createDirs()
+
+	switch os.Args[1] {
+	case "default", "gitea":
+		runContainers(os.Args[1])
+	case "runprev":
+		runContainers(currentContainers())
+	case "stop":
+		runCompose(append(composeFlags(), "down"))
+	case "dbshell":
+		runCompose(append(composeFlags(), "exec", "cursorius-db", "psql", "--user=cursorius"))
+	case "logs":
+		runCompose(append(composeFlags(), "logs", "-f"))
+	case "ps":
+		runCompose(append(composeFlags(), "ps"))
+	case "help":
+		fmt.Println("commands: default, gitea, runprev, stop, dbshell, logs, ps, help")
+	default:
+		panicError("Unknown subcommand: %v", os.Args[1])
+
+	}
+}

listen/listen.go

@@ -22,13 +22,12 @@ func setupHTTPServer(
 	mux *http.ServeMux,
 	conf config.PipelineConf,
 	db database.Database,
-	registerCh chan runnermanager.RunnerRegistration,
-	getRunnerCh chan runnermanager.GetRunnerRequest,
+	runnerManagerChans runnermanager.RunnerManagerChans,
 ) error {
 
 	webhook.CreateWebhookHandler(db, conf, mux)
 
-	pipeline_api.CreateHandler(getRunnerCh, mux)
+	pipeline_api.CreateHandler(runnerManagerChans.Allocation, runnerManagerChans.Release, mux)
 
 	err := admin_api.CreateHandler(db, mux)
 	if err != nil {
@@ -41,7 +40,7 @@ func setupHTTPServer(
 			log.Errorf("Could not upgrade runner connection to websocket: %v", err)
 			return
 		}
-		go runnermanager.RegisterRunner(conn, registerCh)
+		go runnermanager.RegisterRunner(conn, runnerManagerChans.Registration)
 	})
 	return nil
 }
@@ -52,16 +51,14 @@ func Listen(
 	port int,
 	conf config.PipelineConf,
 	db database.Database,
-	registerCh chan runnermanager.RunnerRegistration,
-	getRunnerCh chan runnermanager.GetRunnerRequest,
+	runnerManagerChans runnermanager.RunnerManagerChans,
 ) error {
 
 	err := setupHTTPServer(
 		mux,
 		conf,
 		db,
-		registerCh,
-		getRunnerCh,
+		runnerManagerChans,
 	)
 	if err != nil {
 		return fmt.Errorf("Could not setup http endpoints: %w", err)

main.go

@@ -40,7 +40,7 @@ func main() {
 		return
 	}
 
-	getRunnerCh, registerCh, err := runnermanager.StartRunnerManager(configData.Config.Runners, db)
+	runnerManagerChans, err := runnermanager.StartRunnerManager(configData.Config.Runners, db)
 	if err != nil {
 		log.Errorf("Could not start runner: %v", err)
 		return
@@ -56,7 +56,6 @@ func main() {
 		configData.Config.Port,
 		configData.Config.PipelineConf,
 		db,
-		registerCh,
-		getRunnerCh,
+		runnerManagerChans,
 	))
 }

pipeline_api/pipeline_api.go

@@ -4,12 +4,12 @@ import (
 	"context"
 	"fmt"
 	"net/http"
-	"strings"
 	"sync"
 
 	apiv2 "git.ohea.xyz/cursorius/pipeline-api/go/api/v2"
 	"git.ohea.xyz/cursorius/pipeline-api/go/api/v2/apiv2connect"
 	"git.ohea.xyz/cursorius/server/runnermanager"
+	"git.ohea.xyz/cursorius/server/util"
 	"github.com/bufbuild/connect-go"
 	"github.com/google/uuid"
 	"github.com/op/go-logging"
@@ -18,7 +18,8 @@ import (
 var log = logging.MustGetLogger("cursorius-server")
 
 type ApiServer struct {
-	getRunnerCh           chan runnermanager.GetRunnerRequest
+	allocationCh          chan runnermanager.RunnerAllocationRequest
+	releaseCh             chan runnermanager.RunnerReleaseRequest
 	allocatedRunners      map[uuid.UUID]*RunnerWrapper
 	allocatedRunnersMutex sync.RWMutex
 }
@@ -34,15 +35,17 @@ func (r *RunnerWrapper) RunCommand(cmd string, args []string) (int64, string, st
 
 	return_code, stdout, stderr, err := r.runner.RunCommand(cmd, args)
 
-	// TODO: run command by sending websocket packet
-	// TODO: get stdout and stderr response
 	return return_code, stdout, stderr, err
 }
 
-func (r *RunnerWrapper) Release() {
+func (r *RunnerWrapper) Release(releaseCh chan runnermanager.RunnerReleaseRequest) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
-	r.runner.Release()
+
+	releaseCh <- runnermanager.RunnerReleaseRequest{
+		Runner: r.runner,
+	}
+	r.runner = nil
 }
 
 func (s *ApiServer) GetRunnerFromMap(u uuid.UUID) (*RunnerWrapper, bool) {
@@ -57,28 +60,21 @@ func (s *ApiServer) GetRunner(
 	req *connect.Request[apiv2.GetRunnerRequest],
 ) (*connect.Response[apiv2.GetRunnerResponse], error) {
 
-	respChan := make(chan runnermanager.GetRunnerResponse)
-	s.getRunnerCh <- runnermanager.GetRunnerRequest{
+	respChan := make(chan runnermanager.RunnerAllocationResponse)
+	s.allocationCh <- runnermanager.RunnerAllocationRequest{
 		Tags:     req.Msg.Tags,
 		RespChan: respChan,
 	}
 
-	var runnerTagsStr strings.Builder
-	if len(req.Msg.Tags) > 0 {
-		fmt.Fprintf(&runnerTagsStr, "[%v", req.Msg.Tags[0])
-		for _, tag := range req.Msg.Tags[1:] {
-			fmt.Fprintf(&runnerTagsStr, ", %v", tag)
-		}
-		fmt.Fprintf(&runnerTagsStr, "]")
-	}
+	tagsStr := util.FormatTags(req.Msg.Tags)
 
 	response := <-respChan
 	if response.Err != nil {
-		log.Errorf("Could not get runner with tags \"%v\": %v", runnerTagsStr.String(), response.Err)
+		log.Errorf("Could not get runner with tags \"%v\": %v", tagsStr, response.Err)
 		return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("Could not get runner"))
 	}
 
-	log.Infof("Got runner with tags: %v", runnerTagsStr.String())
+	log.Infof("Got runner with tags: %v", tagsStr)
 
 	runnerUuid := uuid.New()
 
@@ -107,7 +103,7 @@ func (s *ApiServer) ReleaseRunner(
 	s.allocatedRunnersMutex.Lock()
 	runner := s.allocatedRunners[uuid]
 	delete(s.allocatedRunners, uuid)
-	runner.Release()
+	runner.Release(s.releaseCh)
 	s.allocatedRunnersMutex.Unlock()
 
 	res := connect.NewResponse(&apiv2.ReleaseRunnerResponse{})
@@ -146,9 +142,10 @@ func (s *ApiServer) RunCommand(
 	return res, nil
 }
 
-func CreateHandler(getRunnerCh chan runnermanager.GetRunnerRequest, mux *http.ServeMux) {
+func CreateHandler(allocationCh chan runnermanager.RunnerAllocationRequest, releaseCh chan runnermanager.RunnerReleaseRequest, mux *http.ServeMux) {
 	api_server := &ApiServer{
-		getRunnerCh:      getRunnerCh,
+		allocationCh:     allocationCh,
+		releaseCh:        releaseCh,
 		allocatedRunners: make(map[uuid.UUID]*RunnerWrapper),
 	}
 	path, handler := apiv2connect.NewGetRunnerServiceHandler(api_server)

runnermanager/runner.go

@@ -22,17 +22,27 @@ type Runner struct {
 	tags        []string
 	conn        *websocket.Conn
 	receiveChan chan []byte
-	running     bool
+}
+
+func (r *Runner) HasTags(requestedTags []string) bool {
+tagIter:
+	for _, requestedTag := range requestedTags {
+		for _, posessedTag := range r.tags {
+			// if we find the tag, move on to search for the next one
+			if posessedTag == requestedTag {
+				continue tagIter
+			}
+		}
+		// if we don't find the tag
+		return false
+	}
+	return true
 }
 
 func (r *Runner) Id() uuid.UUID {
 	return r.id
 }
 
-func (r *Runner) Release() {
-	r.running = false
-}
-
 func (r *Runner) RunCommand(cmd string, args []string) (returnCode int64, stdout string, stderr string, err error) {
 
 	// Write RunCommand message to client

runnermanager/runnermanager.go

@@ -3,7 +3,6 @@ package runnermanager
 import (
 	"context"
 	"fmt"
-	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -13,66 +12,64 @@ import (
 
 	"git.ohea.xyz/cursorius/server/config"
 	"git.ohea.xyz/cursorius/server/database"
+	"git.ohea.xyz/cursorius/server/util"
 
 	runner_api "git.ohea.xyz/cursorius/runner-api/go/api/v2"
 )
 
 var log = logging.MustGetLogger("cursorius-server")
 
-type RunnerRegistration struct {
-	Secret string
-	Id     string
-	Tags   []string
-	conn   *websocket.Conn
+type RunnerManagerChans struct {
+	Allocation   chan RunnerAllocationRequest
+	Release      chan RunnerReleaseRequest
+	Registration chan RunnerRegistrationRequest
 }
 
 type runnerManager struct {
-	getRunnerCh         chan GetRunnerRequest
-	registerCh          chan RunnerRegistration
+	chans               RunnerManagerChans
 	connectedRunners    []Runner
 	numConnectedRunners uint64
 	configuredRunners   map[string]config.Runner
 	db                  database.Database
 }
 
-type GetRunnerRequest struct {
+type RunnerAllocationRequest struct {
 	Tags     []string
-	RespChan chan GetRunnerResponse
+	RespChan chan RunnerAllocationResponse
 }
 
-type GetRunnerResponse struct {
+type RunnerAllocationResponse struct {
 	Runner *Runner
 	Err    error
 }
 
+type RunnerReleaseRequest struct {
+	Runner *Runner
+}
+
+type RunnerRegistrationRequest struct {
+	Secret string
+	Id     string
+	Tags   []string
+	conn   *websocket.Conn
+}
+
 type runnerJob struct {
 	Id  string
 	URL string
 }
 
-func (r *runnerManager) processRequest(req GetRunnerRequest) {
-	var runnerTagsStr strings.Builder
-	if len(req.Tags) > 0 {
-		fmt.Fprintf(&runnerTagsStr, "[%v", req.Tags[0])
-		for _, tag := range req.Tags[1:] {
-			fmt.Fprintf(&runnerTagsStr, ", %v", tag)
-		}
-		fmt.Fprintf(&runnerTagsStr, "]")
-	}
-	log.Infof("Got request for runner with tags \"%v\"", runnerTagsStr.String())
+func (r *runnerManager) processRunnerAllocation(req RunnerAllocationRequest) {
+	tagsStr := util.FormatTags(req.Tags)
+	log.Infof("Got request for runner with tags \"%v\"", tagsStr)
 
-	log.Debugf("Finding runner with tags %v", runnerTagsStr.String())
+	log.Debugf("Finding runner with tags %v", tagsStr)
 
 	foundRunner := false
 
 	runnersToRemove := []int{}
 runnerIter:
 	for i, runner := range r.connectedRunners {
-		// don't allocate runner that is already occupied
-		if runner.running {
-			log.Debugf("Skipping runner %v, as runner is activly running another job", runner.id)
-			continue
-		}
 		// don't allocate runner with closed receiveChan (is defunct)
 		// there should never be messages to read on an inactive runner,
 		// so we aren't losing any data here
@@ -89,25 +86,20 @@ runnerIter:
 		default:
 			log.Debugf("Checking runner %v for requested tags", runner.id)
 
-		tagIter:
-			for _, requestedTag := range req.Tags {
-				for _, posessedTag := range runner.tags {
-					if requestedTag == posessedTag {
-						continue tagIter
-					}
-				}
+			if !runner.HasTags(req.Tags) {
 				continue runnerIter
 			}
 
-			r.connectedRunners[i].running = true
+			runnersToRemove = append(runnersToRemove, i)
 			foundRunner = true
-			req.RespChan <- GetRunnerResponse{
+			req.RespChan <- RunnerAllocationResponse{
 				Runner: &r.connectedRunners[i],
 				Err:    nil,
 			}
 		}
 
 	}
+	// remove allocated runner plus defunct runners
 	// since we iterate, all the indexes will be in accending order
 	for i, runnerInd := range runnersToRemove {
 		r.connectedRunners[runnerInd-i] = r.connectedRunners[len(r.connectedRunners)-1]
@@ -122,43 +114,42 @@ runnerIter:
 	if len(r.connectedRunners) == 0 {
 		errorMsg = "no connected runners"
 	}
-	req.RespChan <- GetRunnerResponse{
+	req.RespChan <- RunnerAllocationResponse{
 		Runner: &Runner{},
 		Err:    fmt.Errorf("Could not allocate runner: %v", errorMsg),
 	}
 
 }
 
-func (r *runnerManager) processRegistration(reg RunnerRegistration) {
-	log.Debugf("New runner appeared with id: %v and secret: %v", reg.Id, reg.Secret)
+func (r *runnerManager) processRunnerRegistration(req RunnerRegistrationRequest) {
+	log.Debugf("New runner appeared with id: %v and secret: %v", req.Id, req.Secret)
 
 	// Get runner with give id from database
-	runnerId, err := uuid.Parse(reg.Id)
+	runnerId, err := uuid.Parse(req.Id)
 	if err != nil {
-		log.Errorf("Disconnecting runner with id: %v, could not parse as UUID: %v", reg.Id, err)
-		reg.conn.Close(websocket.StatusNormalClosure, "registration invalid")
+		log.Errorf("Disconnecting runner with id: %v, could not parse as UUID: %v", req.Id, err)
+		req.conn.Close(websocket.StatusNormalClosure, "registration invalid")
 		return
 	}
 	dbRunner, err := r.db.GetRunnerById(runnerId)
 	if err != nil {
 		log.Errorf("Disconnecting runner with id: %v, could not find runner in DB: %v", runnerId, err)
-		reg.conn.Close(websocket.StatusNormalClosure, "registration invalid")
+		req.conn.Close(websocket.StatusNormalClosure, "registration invalid")
 		return
 	}
 
-	if reg.Secret != dbRunner.Token {
+	if req.Secret != dbRunner.Token {
 		log.Errorf("Disconnecting runner with id: %v, invalid secret", runnerId)
-		reg.conn.Close(websocket.StatusNormalClosure, "registration invalid")
+		req.conn.Close(websocket.StatusNormalClosure, "registration invalid")
 		return
 	}
 
-	log.Infof("Registering runner \"%v\" with tags %v", reg.Id, reg.Tags)
+	log.Infof("Registering runner \"%v\" with tags %v", req.Id, req.Tags)
 	runner := Runner{
 		id:          runnerId,
-		tags:        reg.Tags,
-		conn:        reg.conn,
+		tags:        req.Tags,
+		conn:        req.conn,
 		receiveChan: make(chan []byte),
-		running:     false,
 	}
 	r.connectedRunners = append(r.connectedRunners, runner)
 	// start goroutine to call Read function on websocket connection
@@ -167,7 +158,7 @@ func (r *runnerManager) processRegistration(reg RunnerRegistration) {
 		defer log.Noticef("Deregistered runner with id: %v", runner.id)
 		defer close(runner.receiveChan)
 		for {
-			msgType, data, err := reg.conn.Read(context.Background())
+			msgType, data, err := req.conn.Read(context.Background())
 			if err != nil {
 				// TODO: this is still racy, since a runner could be allocated between the
 				// connection returning an err and the channel closing
@@ -187,22 +178,30 @@ func (r *runnerManager) processRegistration(reg RunnerRegistration) {
 	}()
 }
 
+func (r *runnerManager) processRunnerRelease(req RunnerReleaseRequest) {
+	r.connectedRunners = append(r.connectedRunners, *req.Runner)
+}
+
 func runRunnerManager(r runnerManager) {
 	for {
 		select {
-		case request := <-r.getRunnerCh:
-			r.processRequest(request)
-
-		case registration := <-r.registerCh:
-			r.processRegistration(registration)
+		case request := <-r.chans.Allocation:
+			r.processRunnerAllocation(request)
+		case release := <-r.chans.Release:
+			r.processRunnerRelease(release)
+		case registration := <-r.chans.Registration:
+			r.processRunnerRegistration(registration)
 		}
 	}
 }
 
-func StartRunnerManager(configuredRunners map[string]config.Runner, db database.Database) (chan GetRunnerRequest, chan RunnerRegistration, error) {
+func StartRunnerManager(configuredRunners map[string]config.Runner, db database.Database) (RunnerManagerChans, error) {
 	scheduler := runnerManager{
-		getRunnerCh:       make(chan GetRunnerRequest),
-		registerCh:        make(chan RunnerRegistration),
+		chans: RunnerManagerChans{
+			Allocation:   make(chan RunnerAllocationRequest),
+			Release:      make(chan RunnerReleaseRequest),
+			Registration: make(chan RunnerRegistrationRequest),
+		},
 		connectedRunners:  make([]Runner, 0),
 		configuredRunners: configuredRunners,
 		db:                db,
@@ -210,14 +209,14 @@ func StartRunnerManager(configuredRunners map[string]config.Runner, db database.
 
 	go runRunnerManager(scheduler)
 
-	return scheduler.getRunnerCh, scheduler.registerCh, nil
+	return scheduler.chans, nil
 }
 
-func RegisterRunner(conn *websocket.Conn, registerCh chan RunnerRegistration) {
+func RegisterRunner(conn *websocket.Conn, registerCh chan RunnerRegistrationRequest) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
 	defer cancel()
 
-	var registration RunnerRegistration
+	var registration RunnerRegistrationRequest
 	registration.conn = conn
 
 	typ, r, err := conn.Read(ctx)

util/util.go

@@ -0,0 +1,18 @@
+package util
+
+import (
+	"fmt"
+	"strings"
+)
+
+func FormatTags(tags []string) string {
+	var tagsStr strings.Builder
+	if len(tags) > 0 {
+		fmt.Fprintf(&tagsStr, "[%v", tags[0])
+		for _, tag := range tags[1:] {
+			fmt.Fprintf(&tagsStr, ", %v", tag)
+		}
+		fmt.Fprintf(&tagsStr, "]")
+	}
+	return tagsStr.String()
+}