Cleanup docker build output before saving (#23)

This currently contains the logic for restarting updated jobs,
but nothing exercises this logic. The logic for starting polling
for a newly created pipeline is implemented.

admin_api/admin_api.go

@@ -13,7 +13,7 @@ import (
 
 var log = logging.MustGetLogger("cursorius-server")
 
-func createSchema(db database.Database) (graphql.Schema, error) {
+func createSchema(db database.Database, pollChan chan uuid.UUID) (graphql.Schema, error) {
 	runnerType := graphql.NewObject(graphql.ObjectConfig{
 		Name:        "Runner",
 		Description: "A runner available for use inside of a pipeline.",
@@ -454,6 +454,8 @@ func createSchema(db database.Database) (graphql.Schema, error) {
 						return nil, err
 					}
 
+					pollChan <- pipeline.Id
+
 					return pipeline, nil
 				},
 			},
@@ -558,6 +560,72 @@ func createSchema(db database.Database) (graphql.Schema, error) {
 					return runner, nil
 				},
 			},
+			"updatePipeline": &graphql.Field{
+				Type:        pipelineType,
+				Description: "Create a new pipeline",
+				Args: graphql.FieldConfigArgument{
+					"pipelineId": &graphql.ArgumentConfig{
+						Type: graphql.NewNonNull(graphql.String),
+					},
+					"name": &graphql.ArgumentConfig{
+						Type: graphql.String,
+					},
+					"url": &graphql.ArgumentConfig{
+						Type: graphql.String,
+					},
+					"pollInterval": &graphql.ArgumentConfig{
+						Type: graphql.Int,
+					},
+					"cloneCredentialId": &graphql.ArgumentConfig{
+						Type: graphql.String,
+					},
+				},
+				Resolve: func(params graphql.ResolveParams) (interface{}, error) {
+					pipelineId, err := uuid.Parse(params.Args["pipelineId"].(string))
+					if err != nil {
+						return nil, err
+					}
+
+					var name *string
+					var url *string
+					var interval *int
+
+					if nameVal, ok := params.Args["name"]; ok {
+						nameVal := nameVal.(string)
+						name = &nameVal
+					} else {
+						name = nil
+					}
+
+					if urlVal, ok := params.Args["url"]; ok {
+						urlVal := urlVal.(string)
+						url = &urlVal
+					} else {
+						url = nil
+					}
+
+					if intervalVal, ok := params.Args["pollInterval"]; ok {
+						intervalVal := intervalVal.(int)
+						interval = &intervalVal
+					} else {
+						interval = nil
+					}
+
+					pipeline, err := db.UpdatePipeline(
+						pipelineId,
+						name,
+						url,
+						interval,
+					)
+					if err != nil {
+						return nil, err
+					}
+
+					pollChan <- pipeline.Id
+
+					return pipeline, nil
+				},
+			},
 			"setPipelineCloneCredential": &graphql.Field{
 				Type:        pipelineType,
 				Description: "Set the CloneCredential used by a pipeline to clone the source repo",
@@ -681,9 +749,9 @@ func createSchema(db database.Database) (graphql.Schema, error) {
 	return schema, nil
 }
 
-func CreateHandler(db database.Database, mux *http.ServeMux) error {
+func CreateHandler(db database.Database, pollChan chan uuid.UUID, mux *http.ServeMux) error {
 
-	schema, err := createSchema(db)
+	schema, err := createSchema(db, pollChan)
 	if err != nil {
 		return err
 	}

database/func.go

@@ -77,6 +77,49 @@ RETURNING id, name, url, poll_interval;`
 	return pipeline, nil
 }
 
+func (db *Database) UpdatePipeline(pipelineId uuid.UUID, name *string, url *string, pollInterval *int) (Pipeline, error) {
+	query := `
+UPDATE pipelines
+SET name=$1, url=$2, poll_interval=$3
+WHERE id=$4
+RETURNING name, url, poll_interval, clone_credential;`
+
+	pipeline, err := db.GetPipelineById(pipelineId)
+	if err != nil {
+		return pipeline, err
+	}
+
+	var nameNew string
+	var urlNew string
+	var pollIntervalNew int
+
+	if name != nil {
+		nameNew = *name
+	} else {
+		nameNew = pipeline.Name
+	}
+	if url != nil {
+		urlNew = *url
+	} else {
+		urlNew = pipeline.Url
+	}
+	if pollInterval != nil {
+		pollIntervalNew = *pollInterval
+	} else {
+		pollIntervalNew = pipeline.PollInterval
+	}
+
+	err = db.Conn.QueryRow(context.Background(),
+		query, nameNew, urlNew, pollIntervalNew, pipelineId).Scan(
+		&pipeline.Name, &pipeline.Url, &pipeline.PollInterval, &pipeline.CloneCredential,
+	)
+	if err != nil {
+		return pipeline, fmt.Errorf("Could not add credential to pipeline: %w", err)
+	}
+
+	return pipeline, err
+}
+
 func (db *Database) SetPipelineCloneCredential(pipelineId uuid.UUID, credentialId *uuid.UUID) (Pipeline, error) {
 	query := `
 UPDATE pipelines

go.mod

@@ -3,7 +3,7 @@ module git.ohea.xyz/cursorius/server
 go 1.19
 
 require (
-	git.ohea.xyz/cursorius/pipeline-api/go/api/v2 v2.0.0-20230109075652-ead0aeff2eb9
+	git.ohea.xyz/cursorius/pipeline-api/go/api/v2 v2.0.0-20230405234139-34d8875b72f4
 	git.ohea.xyz/cursorius/runner-api/go/api/v2 v2.0.0-20230109074922-e20285fe6cf2
 	git.ohea.xyz/golang/config v0.0.0-20220915224621-b9debd233173
 	github.com/bufbuild/connect-go v1.4.1
@@ -17,7 +17,7 @@ require (
 	github.com/jhoonb/archivex v0.0.0-20201016144719-6a343cdae81d
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	golang.org/x/net v0.2.0
-	google.golang.org/protobuf v1.28.1
+	google.golang.org/protobuf v1.30.0
 	nhooyr.io/websocket v1.8.7
 )
 

go.sum

@@ -78,8 +78,8 @@ contrib.go.opencensus.io/exporter/stackdriver v0.13.5/go.mod h1:aXENhDJ1Y4lIg4EU
 contrib.go.opencensus.io/integrations/ocsql v0.1.4/go.mod h1:8DsSdjz3F+APR+0z0WkU1aRorQCFfRxvqjUUPMbF3fE=
 contrib.go.opencensus.io/resource v0.1.1/go.mod h1:F361eGI91LCmW1I/Saf+rX0+OFcigGlFvXwEGEnkRLA=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
-git.ohea.xyz/cursorius/pipeline-api/go/api/v2 v2.0.0-20230109075652-ead0aeff2eb9 h1:8p7Kw3B7dbi2zdgG+Me9ETRWrJzoNVjcase4YqXfGbs=
+git.ohea.xyz/cursorius/pipeline-api/go/api/v2 v2.0.0-20230405234139-34d8875b72f4 h1:kKQQEg1nmWnqiNOqtUHteEuacyfy0NdxyDj6HPjbA2c=
-git.ohea.xyz/cursorius/pipeline-api/go/api/v2 v2.0.0-20230109075652-ead0aeff2eb9/go.mod h1:D7GGcFIH421mo6KuRaXXXmlXPwWeEsemTZG/BOZA/4o=
+git.ohea.xyz/cursorius/pipeline-api/go/api/v2 v2.0.0-20230405234139-34d8875b72f4/go.mod h1:D7GGcFIH421mo6KuRaXXXmlXPwWeEsemTZG/BOZA/4o=
 git.ohea.xyz/cursorius/runner-api/go/api/v2 v2.0.0-20230109074922-e20285fe6cf2 h1:G1XQEqhj1LZPQbH7avzvT7QL9Wfbb4CXMm0nLL39eDc=
 git.ohea.xyz/cursorius/runner-api/go/api/v2 v2.0.0-20230109074922-e20285fe6cf2/go.mod h1:F9y5Ck4Wchsaj5amSX2eDRUlQ/iYP1VNLFduvjNwmLc=
 git.ohea.xyz/cursorius/webhooks/v6 v6.0.2-0.20221224221147-a2bdbf1756ed h1:gsK15m4Npow74+R6OfZKwwAg1sl7QWQCRXOeE2QLUco=
@@ -2129,8 +2129,9 @@ google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp0
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
-google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
 google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
+google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc/go.mod h1:m7x9LTH6d71AHyAX77c9yqWCCa3UKHcVEj9y7hAtKDk=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

listen/listen.go

@@ -10,6 +10,8 @@ import (
 	"git.ohea.xyz/cursorius/server/pipeline_api"
 	"git.ohea.xyz/cursorius/server/runnermanager"
 	"git.ohea.xyz/cursorius/server/webhook"
+
+	"github.com/google/uuid"
 	"github.com/op/go-logging"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
@@ -23,13 +25,14 @@ func setupHTTPServer(
 	conf config.PipelineConf,
 	db database.Database,
 	runnerManagerChans runnermanager.RunnerManagerChans,
+	pollChan chan uuid.UUID,
 ) error {
 
 	webhook.CreateWebhookHandler(db, conf, mux)
 
 	pipeline_api.CreateHandler(runnerManagerChans.Allocation, runnerManagerChans.Release, mux)
 
-	err := admin_api.CreateHandler(db, mux)
+	err := admin_api.CreateHandler(db, pollChan, mux)
 	if err != nil {
 		return fmt.Errorf("Could not create admin api handler: %w", err)
 	}
@@ -52,6 +55,7 @@ func Listen(
 	conf config.PipelineConf,
 	db database.Database,
 	runnerManagerChans runnermanager.RunnerManagerChans,
+	pollChan chan uuid.UUID,
 ) error {
 
 	err := setupHTTPServer(
@@ -59,6 +63,7 @@ func Listen(
 		conf,
 		db,
 		runnerManagerChans,
+		pollChan,
 	)
 	if err != nil {
 		return fmt.Errorf("Could not setup http endpoints: %w", err)

main.go

@@ -46,7 +46,7 @@ func main() {
 		return
 	}
 
-	_ = poll.StartPolling(configData.Config.PipelineConf, db)
+	pollChan := poll.StartPolling(configData.Config.PipelineConf, db)
 
 	mux := http.NewServeMux()
 
@@ -57,5 +57,6 @@ func main() {
 		configData.Config.PipelineConf,
 		db,
 		runnerManagerChans,
+		pollChan,
 	))
 }

pipeline_api/pipeline_api.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"sync"
+	"time"
 
 	apiv2 "git.ohea.xyz/cursorius/pipeline-api/go/api/v2"
 	"git.ohea.xyz/cursorius/pipeline-api/go/api/v2/apiv2connect"
@@ -55,22 +56,63 @@ func (s *ApiServer) GetRunnerFromMap(u uuid.UUID) (*RunnerWrapper, bool) {
 	return runner, ok
 }
 
+func (s *ApiServer) AddRunnerToMap(u uuid.UUID, runner *runnermanager.Runner) {
+	s.allocatedRunnersMutex.Lock()
+	defer s.allocatedRunnersMutex.Unlock()
+	s.allocatedRunners[u] = &RunnerWrapper{runner: runner}
+}
+
 func (s *ApiServer) GetRunner(
 	ctx context.Context,
 	req *connect.Request[apiv2.GetRunnerRequest],
 ) (*connect.Response[apiv2.GetRunnerResponse], error) {
 
+	var response runnermanager.RunnerAllocationResponse
+	var timeoutCtx *context.Context
+	var retryInterval int64 = 0
+
 	respChan := make(chan runnermanager.RunnerAllocationResponse)
-	s.allocationCh <- runnermanager.RunnerAllocationRequest{
-		Tags:     req.Msg.Tags,
-		RespChan: respChan,
-	}
 
 	tagsStr := util.FormatTags(req.Msg.Tags)
 
-	response := <-respChan
+	if req.Msg.Options != nil {
-	if response.Err != nil {
+		if req.Msg.Options.Timeout != 0 {
-		log.Errorf("Could not get runner with tags \"%v\": %v", tagsStr, response.Err)
+			ctx, cancel := context.WithTimeout(context.Background(), time.Duration(req.Msg.Options.Timeout)*time.Second)
+			timeoutCtx = &ctx
+			defer cancel()
+		}
+
+		retryInterval = req.Msg.Options.RetryInterval
+	}
+
+	for {
+		s.allocationCh <- runnermanager.RunnerAllocationRequest{
+			Tags:     req.Msg.Tags,
+			RespChan: respChan,
+		}
+
+		response = <-respChan
+		if response.Err == nil {
+			break
+		}
+
+		log.Infof("Could not get runner with tags \"%v\": %v", tagsStr, response.Err)
+
+		// If no timeout is specified, skip after one attempt
+		if timeoutCtx == nil {
+			break
+		}
+
+		// If timeout is expired, stop trying to allocate runner
+		if (*timeoutCtx).Err() != nil {
+			break
+		}
+
+		log.Infof("Sleeping for %v seconds before retry...", retryInterval)
+		time.Sleep(time.Duration(retryInterval) * time.Second)
+	}
+
+	if response.Runner == nil {
 		return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("Could not get runner"))
 	}
 
@@ -78,9 +120,7 @@ func (s *ApiServer) GetRunner(
 
 	runnerUuid := uuid.New()
 
-	s.allocatedRunnersMutex.Lock()
+	s.AddRunnerToMap(runnerUuid, response.Runner)
-	s.allocatedRunners[runnerUuid] = &RunnerWrapper{runner: response.Runner}
-	s.allocatedRunnersMutex.Unlock()
 
 	res := connect.NewResponse(&apiv2.GetRunnerResponse{
 		Id: runnerUuid.String(),

pipeline_executor/pipeline_executor.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
@@ -141,13 +140,7 @@ func ExecutePipeline(pe PipelineExecution, db database.Database, pipelineConf co
 		return
 	}
 
-	response, err := ioutil.ReadAll(buildResponse.Body)
+	err = db.UpdateRunBuildOutput(pe.Run.Id, cleanupBuildOutput(buildResponse.Body))
-	if err != nil {
-		log.Errorf("could no read build response: %w", err)
-		return
-	}
-
-	err = db.UpdateRunBuildOutput(pe.Run.Id, string(response))
 	if err != nil {
 		log.Errorf("could not update build output for run: %w", err)
 		return

pipeline_executor/util.go

@@ -0,0 +1,25 @@
+package pipeline_executor
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+)
+
+func cleanupBuildOutput(input io.ReadCloser) string {
+	output := ""
+
+	scanner := bufio.NewScanner(input)
+	for scanner.Scan() {
+		var log map[string]any
+		json.Unmarshal(scanner.Bytes(), &log)
+
+		if logVar, ok := log["stream"]; ok {
+			if log, ok := logVar.(string); ok {
+				output += log
+			}
+		}
+	}
+
+	return output
+}

poll/poll.go

@@ -1,17 +1,18 @@
 package poll
 
 import (
+	"context"
 	"time"
 
-	"github.com/google/uuid"
-	"github.com/op/go-logging"
-
 	"git.ohea.xyz/cursorius/server/config"
 	"git.ohea.xyz/cursorius/server/database"
 	"git.ohea.xyz/cursorius/server/pipeline_executor"
+
 	"github.com/go-git/go-git/v5"
 	"github.com/go-git/go-git/v5/plumbing"
 	"github.com/go-git/go-git/v5/storage/memory"
+	"github.com/google/uuid"
+	"github.com/op/go-logging"
 )
 
 var log = logging.MustGetLogger("cursorius-server")
@@ -26,10 +27,31 @@ type tag struct {
 	commitHash string
 }
 
-func pollJob(pipeline database.Pipeline, pipelineConf config.PipelineConf, db database.Database) {
+func pollJob(ctx context.Context, pipeline database.Pipeline, pipelineConf config.PipelineConf, db database.Database) {
+	firstScan := true
 	for {
-		time.Sleep(time.Duration(pipeline.PollInterval) * time.Second)
+		// Don't sleep on first scan to ease testing
-		log.Infof("Polling repo %v", pipeline.Name)
+		// TODO: this should be replaced with a script that mocks a webhook
+		if !firstScan {
+
+			ctx, cancel := context.WithTimeout(ctx, time.Duration(pipeline.PollInterval)*time.Second)
+
+			select {
+			case <-ctx.Done():
+				switch ctx.Err() {
+				case context.Canceled:
+					log.Infof("Polling for pipeline %v canceled, stopping", pipeline.Name)
+					cancel()
+					return
+				}
+			}
+
+			cancel()
+
+			log.Infof("Polling repo %v", pipeline.Name)
+		} else {
+			firstScan = false
+		}
 
 		prevRefs, err := db.GetPipelineRefs(pipeline.Id)
 		if err != nil {
@@ -129,11 +151,17 @@ func launchPollJobs(conf config.PipelineConf, db database.Database, pollChan cha
 		return
 	}
 
+	pipelineCancelations := make(map[uuid.UUID]context.CancelFunc)
+
 	for _, pipeline := range pipelines {
 		if pipeline.PollInterval == 0 {
 			continue
 		} else {
-			go pollJob(pipeline, conf, db)
+			ctx, cancel := context.WithCancel(context.Background())
+			pipelineCancelations[pipeline.Id] = cancel
+
+			log.Infof("Starting polling for pipeline %v with id %v", pipeline.Name, pipeline.Id)
+			go pollJob(ctx, pipeline, conf, db)
 		}
 	}
 
@@ -144,8 +172,17 @@ func launchPollJobs(conf config.PipelineConf, db database.Database, pollChan cha
 			log.Errorf("Could not get pipeline with id \"%v\" from database: %v", err)
 			continue
 		}
-		// TODO: stop existing polling process for given uuid
+
-		go pollJob(pipeline, conf, db)
+		// Cancel existing polling job if it exists
+		if cancelFunc, ok := pipelineCancelations[pipeline.Id]; ok {
+			cancelFunc()
+		}
+
+		// Start new polling job
+		log.Infof("Starting polling for pipeline %v with id %v", pipeline.Name, pipeline.Id)
+		ctx, cancel := context.WithCancel(context.Background())
+		pipelineCancelations[pipeline.Id] = cancel
+		go pollJob(ctx, pipeline, conf, db)
 	}
 
 }

runnermanager/runner.go

@@ -45,6 +45,10 @@ func (r *Runner) Id() uuid.UUID {
 
 func (r *Runner) RunCommand(cmd string, args []string) (returnCode int64, stdout string, stderr string, err error) {
 
+	if r.conn == nil {
+		return 0, "", "", fmt.Errorf("runner with id %v has nil conn, THIS IS A BUG", r.id)
+	}
+
 	// Write RunCommand message to client
 	serverToRunnerMsg := &runner_api.ServerToRunnerMsg{
 		Msg: &runner_api.ServerToRunnerMsg_RunCommandMsg{
@@ -97,6 +101,8 @@ func (r *Runner) sendProtoStruct(p protoreflect.ProtoMessage) error {
 
 	ctx := context.Background()
 
+	log.Debugf("r.conn: %p", r.conn)
+
 	if err := r.conn.Write(ctx, websocket.MessageBinary, protoOut); err != nil {
 		return fmt.Errorf("Could not send proto to websocket: %w", err)
 	}

runnermanager/runnermanager.go

@@ -19,46 +19,6 @@ import (
 
 var log = logging.MustGetLogger("cursorius-server")
 
-type RunnerManagerChans struct {
-	Allocation   chan RunnerAllocationRequest
-	Release      chan RunnerReleaseRequest
-	Registration chan RunnerRegistrationRequest
-}
-
-type runnerManager struct {
-	chans               RunnerManagerChans
-	connectedRunners    []Runner
-	numConnectedRunners uint64
-	configuredRunners   map[string]config.Runner
-	db                  database.Database
-}
-
-type RunnerAllocationRequest struct {
-	Tags     []string
-	RespChan chan RunnerAllocationResponse
-}
-
-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) processRunnerAllocation(req RunnerAllocationRequest) {
 	tagsStr := util.FormatTags(req.Tags)
 	log.Infof("Got request for runner with tags \"%v\"", tagsStr)
@@ -92,6 +52,7 @@ runnerIter:
 
 			runnersToRemove = append(runnersToRemove, i)
 			foundRunner = true
+			log.Debugf("Runner %v has requested tags, allocating", runner.id)
 			req.RespChan <- RunnerAllocationResponse{
 				Runner: &r.connectedRunners[i],
 				Err:    nil,
@@ -103,7 +64,7 @@ runnerIter:
 	// 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]
-		r.connectedRunners = r.connectedRunners[0 : len(r.connectedRunners)-2]
+		r.connectedRunners = r.connectedRunners[0 : len(r.connectedRunners)-1]
 	}
 
 	if foundRunner {
@@ -115,7 +76,7 @@ runnerIter:
 		errorMsg = "no connected runners"
 	}
 	req.RespChan <- RunnerAllocationResponse{
-		Runner: &Runner{},
+		Runner: nil,
 		Err:    fmt.Errorf("Could not allocate runner: %v", errorMsg),
 	}
 

runnermanager/types.go

@@ -0,0 +1,49 @@
+package runnermanager
+
+import (
+	"nhooyr.io/websocket"
+
+	"git.ohea.xyz/cursorius/server/config"
+	"git.ohea.xyz/cursorius/server/database"
+)
+
+type RunnerManagerChans struct {
+	Allocation   chan RunnerAllocationRequest
+	Release      chan RunnerReleaseRequest
+	Registration chan RunnerRegistrationRequest
+}
+
+type runnerManager struct {
+	chans               RunnerManagerChans
+	connectedRunners    []Runner
+	numConnectedRunners uint64
+	configuredRunners   map[string]config.Runner
+	db                  database.Database
+}
+
+type RunnerAllocationRequest struct {
+	Tags       []string
+	RespChan   chan RunnerAllocationResponse
+	CancelChan chan string
+}
+
+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
+}