Compare commits
7 Commits
20c664f0ed
...
v0.2.0
| Author | SHA1 | Date | |
|---|---|---|---|
| e1382e50ea | |||
| 7f44e5ed41 | |||
| bcc53dfbe0 | |||
| bbf96498aa | |||
| 954966db58 | |||
| ed7df18f83 | |||
| a8e9a68f0e |
+71
-3
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
|
||||
@@ -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-20230109075652-ead0aeff2eb9/go.mod h1:D7GGcFIH421mo6KuRaXXXmlXPwWeEsemTZG/BOZA/4o=
|
||||
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-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=
|
||||
|
||||
+6
-1
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
))
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
tagsStr := util.FormatTags(req.Msg.Tags)
|
||||
|
||||
if req.Msg.Options != nil {
|
||||
if req.Msg.Options.Timeout != 0 {
|
||||
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,
|
||||
}
|
||||
|
||||
tagsStr := util.FormatTags(req.Msg.Tags)
|
||||
response = <-respChan
|
||||
if response.Err == nil {
|
||||
break
|
||||
}
|
||||
|
||||
response := <-respChan
|
||||
if response.Err != nil {
|
||||
log.Errorf("Could not get runner with tags \"%v\": %v", tagsStr, response.Err)
|
||||
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.allocatedRunners[runnerUuid] = &RunnerWrapper{runner: response.Runner}
|
||||
s.allocatedRunnersMutex.Unlock()
|
||||
s.AddRunnerToMap(runnerUuid, response.Runner)
|
||||
|
||||
res := connect.NewResponse(&apiv2.GetRunnerResponse{
|
||||
Id: runnerUuid.String(),
|
||||
|
||||
@@ -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)
|
||||
if err != nil {
|
||||
log.Errorf("could no read build response: %w", err)
|
||||
return
|
||||
}
|
||||
|
||||
err = db.UpdateRunBuildOutput(pe.Run.Id, string(response))
|
||||
err = db.UpdateRunBuildOutput(pe.Run.Id, cleanupBuildOutput(buildResponse.Body))
|
||||
if err != nil {
|
||||
log.Errorf("could not update build output for run: %w", err)
|
||||
return
|
||||
|
||||
@@ -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
|
||||
}
|
||||
+45
-8
@@ -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
|
||||
// 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)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user