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")
|
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{
|
runnerType := graphql.NewObject(graphql.ObjectConfig{
|
||||||
Name: "Runner",
|
Name: "Runner",
|
||||||
Description: "A runner available for use inside of a pipeline.",
|
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
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pollChan <- pipeline.Id
|
||||||
|
|
||||||
return pipeline, nil
|
return pipeline, nil
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
@@ -558,6 +560,72 @@ func createSchema(db database.Database) (graphql.Schema, error) {
|
|||||||
return runner, nil
|
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{
|
"setPipelineCloneCredential": &graphql.Field{
|
||||||
Type: pipelineType,
|
Type: pipelineType,
|
||||||
Description: "Set the CloneCredential used by a pipeline to clone the source repo",
|
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
|
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 {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -77,6 +77,49 @@ RETURNING id, name, url, poll_interval;`
|
|||||||
return pipeline, nil
|
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) {
|
func (db *Database) SetPipelineCloneCredential(pipelineId uuid.UUID, credentialId *uuid.UUID) (Pipeline, error) {
|
||||||
query := `
|
query := `
|
||||||
UPDATE pipelines
|
UPDATE pipelines
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ module git.ohea.xyz/cursorius/server
|
|||||||
go 1.19
|
go 1.19
|
||||||
|
|
||||||
require (
|
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/cursorius/runner-api/go/api/v2 v2.0.0-20230109074922-e20285fe6cf2
|
||||||
git.ohea.xyz/golang/config v0.0.0-20220915224621-b9debd233173
|
git.ohea.xyz/golang/config v0.0.0-20220915224621-b9debd233173
|
||||||
github.com/bufbuild/connect-go v1.4.1
|
github.com/bufbuild/connect-go v1.4.1
|
||||||
@@ -17,7 +17,7 @@ require (
|
|||||||
github.com/jhoonb/archivex v0.0.0-20201016144719-6a343cdae81d
|
github.com/jhoonb/archivex v0.0.0-20201016144719-6a343cdae81d
|
||||||
github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
|
github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
|
||||||
golang.org/x/net v0.2.0
|
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
|
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/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=
|
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=
|
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 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/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=
|
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.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
|
||||||
google.golang.org/protobuf v1.27.1/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.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.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/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/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=
|
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/pipeline_api"
|
||||||
"git.ohea.xyz/cursorius/server/runnermanager"
|
"git.ohea.xyz/cursorius/server/runnermanager"
|
||||||
"git.ohea.xyz/cursorius/server/webhook"
|
"git.ohea.xyz/cursorius/server/webhook"
|
||||||
|
|
||||||
|
"github.com/google/uuid"
|
||||||
"github.com/op/go-logging"
|
"github.com/op/go-logging"
|
||||||
"golang.org/x/net/http2"
|
"golang.org/x/net/http2"
|
||||||
"golang.org/x/net/http2/h2c"
|
"golang.org/x/net/http2/h2c"
|
||||||
@@ -23,13 +25,14 @@ func setupHTTPServer(
|
|||||||
conf config.PipelineConf,
|
conf config.PipelineConf,
|
||||||
db database.Database,
|
db database.Database,
|
||||||
runnerManagerChans runnermanager.RunnerManagerChans,
|
runnerManagerChans runnermanager.RunnerManagerChans,
|
||||||
|
pollChan chan uuid.UUID,
|
||||||
) error {
|
) error {
|
||||||
|
|
||||||
webhook.CreateWebhookHandler(db, conf, mux)
|
webhook.CreateWebhookHandler(db, conf, mux)
|
||||||
|
|
||||||
pipeline_api.CreateHandler(runnerManagerChans.Allocation, runnerManagerChans.Release, 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 {
|
if err != nil {
|
||||||
return fmt.Errorf("Could not create admin api handler: %w", err)
|
return fmt.Errorf("Could not create admin api handler: %w", err)
|
||||||
}
|
}
|
||||||
@@ -52,6 +55,7 @@ func Listen(
|
|||||||
conf config.PipelineConf,
|
conf config.PipelineConf,
|
||||||
db database.Database,
|
db database.Database,
|
||||||
runnerManagerChans runnermanager.RunnerManagerChans,
|
runnerManagerChans runnermanager.RunnerManagerChans,
|
||||||
|
pollChan chan uuid.UUID,
|
||||||
) error {
|
) error {
|
||||||
|
|
||||||
err := setupHTTPServer(
|
err := setupHTTPServer(
|
||||||
@@ -59,6 +63,7 @@ func Listen(
|
|||||||
conf,
|
conf,
|
||||||
db,
|
db,
|
||||||
runnerManagerChans,
|
runnerManagerChans,
|
||||||
|
pollChan,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("Could not setup http endpoints: %w", err)
|
return fmt.Errorf("Could not setup http endpoints: %w", err)
|
||||||
|
|||||||
@@ -46,7 +46,7 @@ func main() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
_ = poll.StartPolling(configData.Config.PipelineConf, db)
|
pollChan := poll.StartPolling(configData.Config.PipelineConf, db)
|
||||||
|
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
|
|
||||||
@@ -57,5 +57,6 @@ func main() {
|
|||||||
configData.Config.PipelineConf,
|
configData.Config.PipelineConf,
|
||||||
db,
|
db,
|
||||||
runnerManagerChans,
|
runnerManagerChans,
|
||||||
|
pollChan,
|
||||||
))
|
))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
apiv2 "git.ohea.xyz/cursorius/pipeline-api/go/api/v2"
|
apiv2 "git.ohea.xyz/cursorius/pipeline-api/go/api/v2"
|
||||||
"git.ohea.xyz/cursorius/pipeline-api/go/api/v2/apiv2connect"
|
"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
|
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(
|
func (s *ApiServer) GetRunner(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
req *connect.Request[apiv2.GetRunnerRequest],
|
req *connect.Request[apiv2.GetRunnerRequest],
|
||||||
) (*connect.Response[apiv2.GetRunnerResponse], error) {
|
) (*connect.Response[apiv2.GetRunnerResponse], error) {
|
||||||
|
|
||||||
|
var response runnermanager.RunnerAllocationResponse
|
||||||
|
var timeoutCtx *context.Context
|
||||||
|
var retryInterval int64 = 0
|
||||||
|
|
||||||
respChan := make(chan runnermanager.RunnerAllocationResponse)
|
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{
|
s.allocationCh <- runnermanager.RunnerAllocationRequest{
|
||||||
Tags: req.Msg.Tags,
|
Tags: req.Msg.Tags,
|
||||||
RespChan: respChan,
|
RespChan: respChan,
|
||||||
}
|
}
|
||||||
|
|
||||||
tagsStr := util.FormatTags(req.Msg.Tags)
|
response = <-respChan
|
||||||
|
if response.Err == nil {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
|
||||||
response := <-respChan
|
log.Infof("Could not get runner with tags \"%v\": %v", tagsStr, response.Err)
|
||||||
if response.Err != nil {
|
|
||||||
log.Errorf("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"))
|
return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("Could not get runner"))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -78,9 +120,7 @@ func (s *ApiServer) GetRunner(
|
|||||||
|
|
||||||
runnerUuid := uuid.New()
|
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{
|
res := connect.NewResponse(&apiv2.GetRunnerResponse{
|
||||||
Id: runnerUuid.String(),
|
Id: runnerUuid.String(),
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io/ioutil"
|
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -141,13 +140,7 @@ func ExecutePipeline(pe PipelineExecution, db database.Database, pipelineConf co
|
|||||||
return
|
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 {
|
if err != nil {
|
||||||
log.Errorf("could not update build output for run: %w", err)
|
log.Errorf("could not update build output for run: %w", err)
|
||||||
return
|
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
|
package poll
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/google/uuid"
|
|
||||||
"github.com/op/go-logging"
|
|
||||||
|
|
||||||
"git.ohea.xyz/cursorius/server/config"
|
"git.ohea.xyz/cursorius/server/config"
|
||||||
"git.ohea.xyz/cursorius/server/database"
|
"git.ohea.xyz/cursorius/server/database"
|
||||||
"git.ohea.xyz/cursorius/server/pipeline_executor"
|
"git.ohea.xyz/cursorius/server/pipeline_executor"
|
||||||
|
|
||||||
"github.com/go-git/go-git/v5"
|
"github.com/go-git/go-git/v5"
|
||||||
"github.com/go-git/go-git/v5/plumbing"
|
"github.com/go-git/go-git/v5/plumbing"
|
||||||
"github.com/go-git/go-git/v5/storage/memory"
|
"github.com/go-git/go-git/v5/storage/memory"
|
||||||
|
"github.com/google/uuid"
|
||||||
|
"github.com/op/go-logging"
|
||||||
)
|
)
|
||||||
|
|
||||||
var log = logging.MustGetLogger("cursorius-server")
|
var log = logging.MustGetLogger("cursorius-server")
|
||||||
@@ -26,10 +27,31 @@ type tag struct {
|
|||||||
commitHash string
|
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 {
|
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)
|
log.Infof("Polling repo %v", pipeline.Name)
|
||||||
|
} else {
|
||||||
|
firstScan = false
|
||||||
|
}
|
||||||
|
|
||||||
prevRefs, err := db.GetPipelineRefs(pipeline.Id)
|
prevRefs, err := db.GetPipelineRefs(pipeline.Id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -129,11 +151,17 @@ func launchPollJobs(conf config.PipelineConf, db database.Database, pollChan cha
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pipelineCancelations := make(map[uuid.UUID]context.CancelFunc)
|
||||||
|
|
||||||
for _, pipeline := range pipelines {
|
for _, pipeline := range pipelines {
|
||||||
if pipeline.PollInterval == 0 {
|
if pipeline.PollInterval == 0 {
|
||||||
continue
|
continue
|
||||||
} else {
|
} 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)
|
log.Errorf("Could not get pipeline with id \"%v\" from database: %v", err)
|
||||||
continue
|
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) {
|
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
|
// Write RunCommand message to client
|
||||||
serverToRunnerMsg := &runner_api.ServerToRunnerMsg{
|
serverToRunnerMsg := &runner_api.ServerToRunnerMsg{
|
||||||
Msg: &runner_api.ServerToRunnerMsg_RunCommandMsg{
|
Msg: &runner_api.ServerToRunnerMsg_RunCommandMsg{
|
||||||
@@ -97,6 +101,8 @@ func (r *Runner) sendProtoStruct(p protoreflect.ProtoMessage) error {
|
|||||||
|
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
|
||||||
|
log.Debugf("r.conn: %p", r.conn)
|
||||||
|
|
||||||
if err := r.conn.Write(ctx, websocket.MessageBinary, protoOut); err != nil {
|
if err := r.conn.Write(ctx, websocket.MessageBinary, protoOut); err != nil {
|
||||||
return fmt.Errorf("Could not send proto to websocket: %w", err)
|
return fmt.Errorf("Could not send proto to websocket: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,46 +19,6 @@ import (
|
|||||||
|
|
||||||
var log = logging.MustGetLogger("cursorius-server")
|
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) {
|
func (r *runnerManager) processRunnerAllocation(req RunnerAllocationRequest) {
|
||||||
tagsStr := util.FormatTags(req.Tags)
|
tagsStr := util.FormatTags(req.Tags)
|
||||||
log.Infof("Got request for runner with tags \"%v\"", tagsStr)
|
log.Infof("Got request for runner with tags \"%v\"", tagsStr)
|
||||||
@@ -92,6 +52,7 @@ runnerIter:
|
|||||||
|
|
||||||
runnersToRemove = append(runnersToRemove, i)
|
runnersToRemove = append(runnersToRemove, i)
|
||||||
foundRunner = true
|
foundRunner = true
|
||||||
|
log.Debugf("Runner %v has requested tags, allocating", runner.id)
|
||||||
req.RespChan <- RunnerAllocationResponse{
|
req.RespChan <- RunnerAllocationResponse{
|
||||||
Runner: &r.connectedRunners[i],
|
Runner: &r.connectedRunners[i],
|
||||||
Err: nil,
|
Err: nil,
|
||||||
@@ -103,7 +64,7 @@ runnerIter:
|
|||||||
// since we iterate, all the indexes will be in accending order
|
// since we iterate, all the indexes will be in accending order
|
||||||
for i, runnerInd := range runnersToRemove {
|
for i, runnerInd := range runnersToRemove {
|
||||||
r.connectedRunners[runnerInd-i] = r.connectedRunners[len(r.connectedRunners)-1]
|
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 {
|
if foundRunner {
|
||||||
@@ -115,7 +76,7 @@ runnerIter:
|
|||||||
errorMsg = "no connected runners"
|
errorMsg = "no connected runners"
|
||||||
}
|
}
|
||||||
req.RespChan <- RunnerAllocationResponse{
|
req.RespChan <- RunnerAllocationResponse{
|
||||||
Runner: &Runner{},
|
Runner: nil,
|
||||||
Err: fmt.Errorf("Could not allocate runner: %v", errorMsg),
|
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