diff --git a/admin_api/admin_api.go b/admin_api/admin_api.go index e025480..37b2982 100644 --- a/admin_api/admin_api.go +++ b/admin_api/admin_api.go @@ -51,6 +51,64 @@ func createSchema(db database.Database) (graphql.Schema, error) { }, }) + runType := graphql.NewObject(graphql.ObjectConfig{ + Name: "Run", + Description: "A pipeline run", + Fields: graphql.Fields{ + "id": &graphql.Field{ + Type: graphql.NewNonNull(graphql.String), + Description: "The id of the run.", + Resolve: func(p graphql.ResolveParams) (interface{}, error) { + if run, ok := p.Source.(database.Run); ok { + return run.Id, nil + } + return nil, nil + }, + }, + "progress": &graphql.Field{ + Type: graphql.Boolean, + Description: "The progress status of the run.", + Resolve: func(p graphql.ResolveParams) (interface{}, error) { + if run, ok := p.Source.(database.Run); ok { + return run.InProgress, nil + } + return nil, nil + }, + }, + "result": &graphql.Field{ + // TODO: handle bigint properly here + Type: graphql.Float, + Description: "The result of the run.", + Resolve: func(p graphql.ResolveParams) (interface{}, error) { + if run, ok := p.Source.(database.Run); ok { + return run.Result, nil + } + return nil, nil + }, + }, + "stdout": &graphql.Field{ + Type: graphql.String, + Description: "The stdout used to validate the run.", + Resolve: func(p graphql.ResolveParams) (interface{}, error) { + if run, ok := p.Source.(database.Run); ok { + return run.Stdout, nil + } + return nil, nil + }, + }, + "stderr": &graphql.Field{ + Type: graphql.String, + Description: "The stderr used to validate the run.", + Resolve: func(p graphql.ResolveParams) (interface{}, error) { + if run, ok := p.Source.(database.Run); ok { + return run.Stderr, nil + } + return nil, nil + }, + }, + }, + }) + pipelineType := graphql.NewObject(graphql.ObjectConfig{ Name: "Pipeline", Description: "A pipeline for running ci jobs", @@ -105,6 +163,16 @@ func createSchema(db database.Database) (graphql.Schema, error) { return []database.Webhook{}, nil }, }, + "runs": &graphql.Field{ + Type: graphql.NewNonNull(graphql.NewList(graphql.NewNonNull(runType))), + Description: "The list of runs for the pipeline.", + Resolve: func(p graphql.ResolveParams) (interface{}, error) { + if pipeline, ok := p.Source.(database.Pipeline); ok { + return db.GetRunsForPipeline(pipeline.Id) + } + return []database.Webhook{}, nil + }, + }, }, }) @@ -127,6 +195,13 @@ func createSchema(db database.Database) (graphql.Schema, error) { return db.GetPipelineById(id) }, }, + "Pipelines": &graphql.Field{ + Type: graphql.NewNonNull(graphql.NewList(pipelineType)), + Args: graphql.FieldConfigArgument{}, + Resolve: func(p graphql.ResolveParams) (interface{}, error) { + return db.GetPipelines() + }, + }, }, }) diff --git a/database/db.go b/database/db.go index 7df2cac..8ffba24 100644 --- a/database/db.go +++ b/database/db.go @@ -8,14 +8,14 @@ import ( "context" "fmt" - "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" "github.com/op/go-logging" ) var log = logging.MustGetLogger("cursorius-server") type Database struct { - Conn *pgx.Conn + Conn *pgxpool.Pool } func LaunchDB(conf config.DBConfig) (Database, error) { @@ -41,8 +41,9 @@ func LaunchDB(conf config.DBConfig) (Database, error) { var err error for i := 0; i < 10; i++ { + // TODO: retry logic is broken with pgxpool log.Infof("Connecting to database with URL \"%v\" (attempt %v)", dbURLNoPasswd, i) - db.Conn, err = pgx.Connect(context.Background(), dbURL) + db.Conn, err = pgxpool.New(context.Background(), dbURL) if err == nil { break } @@ -80,12 +81,13 @@ SELECT EXISTS ( return db, nil } -func initDB(conn *pgx.Conn) error { +func initDB(conn *pgxpool.Pool) error { createTablesQuery := ` CREATE EXTENSION IF NOT EXISTS "uuid-ossp"; CREATE TABLE version ( version INT NOT NULL + ); CREATE TABLE pipelines ( @@ -115,7 +117,10 @@ CREATE TABLE runners ( CREATE TABLE runs ( id UUID PRIMARY KEY, pipeline UUID, - result BOOLEAN NOT NULL, + in_progress BOOLEAN DEFAULT NULL, + result BIGINT DEFAULT NULL, + stdout TEXT DEFAULT NULL, + stderr TEXT DEFAULT NULL, CONSTRAINT fk_pipeline FOREIGN KEY(pipeline) diff --git a/database/func.go b/database/func.go index e928418..a28a39c 100644 --- a/database/func.go +++ b/database/func.go @@ -7,7 +7,37 @@ import ( "github.com/google/uuid" ) -func (d *Database) GetPipelineById(id uuid.UUID) (Pipeline, error) { +func (db *Database) GetPipelines() ([]Pipeline, error) { + query := ` +SELECT id, name, url, poll_interval +FROM pipelines;` + + pipelines := make([]Pipeline, 0) + + rows, err := db.Conn.Query(context.Background(), query) + if err != nil { + return pipelines, fmt.Errorf("Could not query database for pipelines: %w", err) + } + defer rows.Close() + + for rows.Next() { + var pipeline Pipeline + var idStr string + if err := rows.Scan(&idStr, &pipeline.Name, &pipeline.Url, &pipeline.PollInterval); err != nil { + return pipelines, err + } + + pipeline.Id, err = uuid.Parse(idStr) + if err != nil { + return pipelines, err + } + pipelines = append(pipelines, pipeline) + } + + return pipelines, nil +} + +func (db *Database) GetPipelineById(id uuid.UUID) (Pipeline, error) { query := ` SELECT name, url, poll_interval FROM pipelines @@ -17,7 +47,7 @@ WHERE id=$1;` Id: id, } - err := d.Conn.QueryRow(context.Background(), query, id).Scan(&pipeline.Name, &pipeline.Url, &pipeline.PollInterval) + err := db.Conn.QueryRow(context.Background(), query, id).Scan(&pipeline.Name, &pipeline.Url, &pipeline.PollInterval) if err != nil { return pipeline, fmt.Errorf("Could not query database for pipeline with id %v: %w", id.String(), err) } @@ -25,7 +55,7 @@ WHERE id=$1;` return pipeline, nil } -func (d *Database) CreatePipeline(name string, url string, pollInterval int) (Pipeline, error) { +func (db *Database) CreatePipeline(name string, url string, pollInterval int) (Pipeline, error) { query := ` INSERT INTO pipelines (id, name, url, poll_interval) VALUES (uuid_generate_v4(), $1, $2, $3) @@ -33,14 +63,14 @@ RETURNING id, name, url, poll_interval;` pipeline := Pipeline{} var idStr string - err := d.Conn.QueryRow(context.Background(), query, name, url, pollInterval).Scan(&idStr, &pipeline.Name, &pipeline.Url, &pipeline.PollInterval) + err := db.Conn.QueryRow(context.Background(), query, name, url, pollInterval).Scan(&idStr, &pipeline.Name, &pipeline.Url, &pipeline.PollInterval) if err != nil { - return pipeline, err + return pipeline, fmt.Errorf("Could not create pipeline: %w", err) } id, err := uuid.Parse(idStr) if err != nil { - return pipeline, err + return pipeline, fmt.Errorf("Could not parse UUID generated by DB: %w", err) } pipeline.Id = id @@ -58,7 +88,7 @@ WHERE pipeline=$1;` rows, err := db.Conn.Query(context.Background(), query, id) if err != nil { - log.Fatal(err) + return webhooks, fmt.Errorf("Could not get webhooks for pipeline with id \"%v\": %w", id, err) } defer rows.Close() @@ -79,7 +109,7 @@ WHERE pipeline=$1;` return webhooks, nil } -func (d *Database) GetWebhookById(id uuid.UUID) (Webhook, error) { +func (db *Database) GetWebhookById(id uuid.UUID) (Webhook, error) { query := ` SELECT server_type, secret, pipeline FROM webhooks @@ -89,7 +119,7 @@ WHERE id=$1;` Id: id, } - err := d.Conn.QueryRow(context.Background(), query, id).Scan(&webhook.ServerType, &webhook.Secret, &webhook.Pipeline) + err := db.Conn.QueryRow(context.Background(), query, id).Scan(&webhook.ServerType, &webhook.Secret, &webhook.Pipeline) if err != nil { return webhook, fmt.Errorf("Could not query database for webhook with id %v: %w", id.String(), err) } @@ -97,9 +127,7 @@ WHERE id=$1;` return webhook, nil } -func (d *Database) CreateWebhook(serverType WebhookSender, pipelineId uuid.UUID) (Webhook, error) { - - //WITH secret_val as (select substr(md5(random()::text), 0, 50)), +func (db *Database) CreateWebhook(serverType WebhookSender, pipelineId uuid.UUID) (Webhook, error) { query := ` INSERT INTO webhooks (id, server_type, secret, pipeline) @@ -108,7 +136,7 @@ RETURNING id, server_type, secret, pipeline;` webhook := Webhook{} var idStr string - err := d.Conn.QueryRow(context.Background(), query, string(serverType), pipelineId).Scan(&idStr, &webhook.ServerType, &webhook.Secret, &webhook.Pipeline) + err := db.Conn.QueryRow(context.Background(), query, string(serverType), pipelineId).Scan(&idStr, &webhook.ServerType, &webhook.Secret, &webhook.Pipeline) if err != nil { return webhook, err } @@ -122,3 +150,74 @@ RETURNING id, server_type, secret, pipeline;` return webhook, nil } + +func (db *Database) CreateRun(pipelineId uuid.UUID) (Run, error) { + query := ` +INSERT INTO runs (id, pipeline, in_progress) +VALUES(uuid_generate_v4(), $1, true) +RETURNING id, pipeline, in_progress;` + + run := Run{} + var idStr string + err := db.Conn.QueryRow(context.Background(), query, pipelineId).Scan(&idStr, &run.Pipeline, &run.InProgress) + if err != nil { + return run, err + } + + run.Id, err = uuid.Parse(idStr) + if err != nil { + return run, err + } + + return run, nil +} + +func (db *Database) UpdateRunResult(r Run) error { + query := ` +UPDATE runs +SET in_progress=$1, result=$2, stdout=$3, stderr=$4 +WHERE id=$3;` + + // TODO: does r.Result need a pointer derefrence? + _, err := db.Conn.Exec(context.Background(), + query, r.InProgress, r.Result, r.Stdout, r.Stderr) + + return err +} + +func (db *Database) GetRunsForPipeline(pipelineId uuid.UUID) ([]Run, error) { + query := ` +SELECT id, in_progress, result, stdout, stderr +FROM runs +WHERE pipeline=$1;` + + runs := make([]Run, 0) + + rows, err := db.Conn.Query(context.Background(), query, pipelineId) + if err != nil { + return runs, fmt.Errorf("Could not get runs for pipeline with id \"%v\": %w", pipelineId, err) + } + defer rows.Close() + + for rows.Next() { + var run Run + var idStr string + if err := rows.Scan( + &idStr, + &run.InProgress, + &run.Result, + &run.Stdout, + &run.Stderr, + ); err != nil { + return runs, err + } + + run.Id, err = uuid.Parse(idStr) + if err != nil { + return runs, err + } + runs = append(runs, run) + } + + return runs, nil +} diff --git a/database/types.go b/database/types.go index 6766c8c..387f6d6 100644 --- a/database/types.go +++ b/database/types.go @@ -33,15 +33,18 @@ type Runner struct { } type Run struct { - Id uuid.UUID - Pipeline uuid.UUID - Result bool + Id uuid.UUID + Pipeline uuid.UUID + InProgress bool + Result *int64 + Stdout []byte + Stderr []byte } type CommandExecution struct { Id uuid.UUID RunId uuid.UUID - Command string + Command []string ReturnCode int Stdout string Stderr string diff --git a/go.mod b/go.mod index 25d6425..9fd25e0 100644 --- a/go.mod +++ b/go.mod @@ -37,6 +37,7 @@ require ( github.com/imdario/mergo v0.3.13 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b // indirect + github.com/jackc/puddle/v2 v2.1.2 // indirect github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kevinburke/ssh_config v1.2.0 // indirect @@ -53,8 +54,10 @@ require ( github.com/sirupsen/logrus v1.9.0 // indirect github.com/stretchr/testify v1.8.1 // indirect github.com/xanzy/ssh-agent v0.3.2 // indirect + go.uber.org/atomic v1.10.0 // indirect golang.org/x/crypto v0.2.1-0.20221112162523-6fad3dfc1891 // indirect golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect + golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 // indirect golang.org/x/sys v0.2.0 // indirect golang.org/x/text v0.4.0 // indirect golang.org/x/tools v0.1.12 // indirect diff --git a/go.sum b/go.sum index 27aa077..1bc8c09 100644 --- a/go.sum +++ b/go.sum @@ -849,6 +849,8 @@ github.com/jackc/puddle v1.1.0/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dv github.com/jackc/puddle v1.1.1/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk= github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk= github.com/jackc/puddle v1.2.1/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk= +github.com/jackc/puddle/v2 v2.1.2 h1:0f7vaaXINONKTsxYDn4otOAiJanX/BMeAtY//BXqzlg= +github.com/jackc/puddle/v2 v2.1.2/go.mod h1:2lpufsF5mRHO6SuZkm0fNYxM6SWHfvyFj62KwNzgels= github.com/jarcoal/httpmock v0.0.0-20180424175123-9c70cfe4a1da/go.mod h1:ks+b9deReOc7jgqp+e7LuFiCBH6Rm5hL32cLcEAArb4= github.com/jarcoal/httpmock v1.0.5/go.mod h1:ATjnClrvW/3tijVmpL/va5Z3aAyGvqU3gCT8nX0Txik= github.com/jaytaylor/html2text v0.0.0-20211105163654-bc68cce691ba/go.mod h1:CVKlgaMiht+LXvHG173ujK6JUhZXKb2u/BQtjPDIvyk= @@ -1443,6 +1445,7 @@ go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ= go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= @@ -1670,6 +1673,8 @@ golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20220513210516-0976fa681c29/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 h1:ZrnxWX62AgTKOSagEqxvb3ffipvEDX2pl7E1TdqLqIc= +golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= diff --git a/listen/listen.go b/listen/listen.go index 0aa9841..ebcb063 100644 --- a/listen/listen.go +++ b/listen/listen.go @@ -20,13 +20,13 @@ var log = logging.MustGetLogger("cursorius-server") func setupHTTPServer( mux *http.ServeMux, - conf config.Config, + conf config.PipelineConf, db database.Database, registerCh chan runnermanager.RunnerRegistration, getRunnerCh chan runnermanager.GetRunnerRequest, ) error { - webhook.CreateWebhookHandler(conf, mux) + webhook.CreateWebhookHandler(db, conf, mux) pipeline_api.CreateHandler(getRunnerCh, mux) @@ -50,7 +50,7 @@ func Listen( mux *http.ServeMux, address string, port int, - conf config.Config, + conf config.PipelineConf, db database.Database, registerCh chan runnermanager.RunnerRegistration, getRunnerCh chan runnermanager.GetRunnerRequest, diff --git a/main.go b/main.go index 20d7e34..97153ca 100644 --- a/main.go +++ b/main.go @@ -46,7 +46,7 @@ func main() { return } - poll.StartPolling(configData.Config) + _ = poll.StartPolling(configData.Config.PipelineConf, db) mux := http.NewServeMux() @@ -54,7 +54,7 @@ func main() { mux, configData.Config.Address, configData.Config.Port, - configData.Config, + configData.Config.PipelineConf, db, registerCh, getRunnerCh, diff --git a/pipeline_executor/pipeline_executor.go b/pipeline_executor/pipeline_executor.go index 3889ede..27b8698 100644 --- a/pipeline_executor/pipeline_executor.go +++ b/pipeline_executor/pipeline_executor.go @@ -10,6 +10,7 @@ import ( "path/filepath" "git.ohea.xyz/cursorius/server/config" + "git.ohea.xyz/cursorius/server/database" "github.com/docker/docker/api/types" "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/mount" @@ -21,40 +22,44 @@ import ( var log = logging.MustGetLogger("cursorius-server") type PipelineExecution struct { - Name string - Job config.Job - Ref string + Pipeline database.Pipeline + Ref string + Run database.Run } -func ExecutePipeline(pe PipelineExecution, pipelineConf config.PipelineConf) error { - jobFolder := filepath.Join(pipelineConf.WorkingDir, pe.Name) +func ExecutePipeline(pe PipelineExecution, db database.Database, pipelineConf config.PipelineConf) { + jobFolder := filepath.Join(pipelineConf.WorkingDir, pe.Pipeline.Id.String(), pe.Run.Id.String()) - log.Debugf("Job %v configured with URL \"%v\"", pe.Name, pe.Job.URL) + log.Debugf("Job %v configured with URL \"%v\"", pe.Pipeline.Name, pe.Pipeline.Url) - log.Debugf("Job %v configured with folder \"%v\"", pe.Name, jobFolder) + log.Debugf("Job %v configured with folder \"%v\"", pe.Pipeline.Name, jobFolder) err := os.RemoveAll(jobFolder) if err != nil { - return fmt.Errorf("could not delete existing folder %v", jobFolder) + log.Errorf("could not delete existing folder %v", jobFolder) + return } err = os.MkdirAll(jobFolder, 0755) if err != nil { - return fmt.Errorf("could not create working directory for job %v: %w", pe.Name, err) + log.Errorf("could not create working directory for job %v: %w", pe.Pipeline.Name, err) + return } - log.Infof("Cloning source from URL %v", pe.Job.URL) + log.Infof("Cloning source from URL %v", pe.Pipeline.Url) // TODO: should I use go-git here instead of shelling out to raw git? - cloneCmd := exec.Command("git", "clone", pe.Job.URL, jobFolder) + cloneCmd := exec.Command("git", "clone", pe.Pipeline.Url, jobFolder) output, err := cloneCmd.CombinedOutput() if err != nil { log.Debugf("%s", output) - return fmt.Errorf("could not clone source: %w", err) + log.Errorf("could not clone source: %w", err) + return } cli, err := client.NewClientWithOpts(client.FromEnv) if err != nil { - return fmt.Errorf("Could not create docker client: %w", err) + log.Errorf("Could not create docker client: %w", err) + return } log.Info("Source cloned successfully") @@ -65,18 +70,21 @@ func ExecutePipeline(pe PipelineExecution, pipelineConf config.PipelineConf) err log.Infof("Pulling image %v", imageName) pullOutput, err := cli.ImagePull(ctx, imageName, types.ImagePullOptions{}) if err != nil { - return fmt.Errorf("could not pull image %v: %w", imageName, err) + log.Errorf("could not pull image %v: %w", imageName, err) + return } buf, err := io.ReadAll(pullOutput) if err != nil { - return fmt.Errorf("could not read from io.ReadCloser:, %w", err) + log.Errorf("could not read from io.ReadCloser:, %w", err) + return } log.Infof("%s", buf) err = pullOutput.Close() if err != nil { - return fmt.Errorf("could not close io.ReadCloser: %w", err) + log.Errorf("could not close io.ReadCloser: %w", err) + return } log.Info("Image pulled sucessfully") @@ -87,10 +95,12 @@ func ExecutePipeline(pe PipelineExecution, pipelineConf config.PipelineConf) err } if pipelineConf.MountConf.Type == config.Bind { + sourceDir := filepath.Join(pipelineConf.MountConf.Source, pe.Pipeline.Id.String(), pe.Run.Id.String()) + hostConfig.Mounts = append(hostConfig.Mounts, mount.Mount{ Type: mount.TypeBind, - Source: fmt.Sprintf("%v/%v", pipelineConf.MountConf.Source, pe.Name), + Source: sourceDir, Target: "/cursorius/src", ReadOnly: false, Consistency: mount.ConsistencyDefault, @@ -111,9 +121,9 @@ func ExecutePipeline(pe PipelineExecution, pipelineConf config.PipelineConf) err resp, err := cli.ContainerCreate(ctx, &container.Config{ Image: imageName, - Cmd: []string{"/launcher.sh"}, Tty: false, Env: []string{ + fmt.Sprintf("RUNID=%v", pe.Run.Id), "CURSORIUS_SRC_DIR=/cursorius/src", fmt.Sprintf("CUROSRIUS_SERVER_URL=%v", pipelineConf.AccessURL), }, @@ -123,27 +133,39 @@ func ExecutePipeline(pe PipelineExecution, pipelineConf config.PipelineConf) err nil, nil, "", ) if err != nil { - return fmt.Errorf("could not create container: %w", err) + log.Errorf("could not create container: %w", err) + return } log.Info("Launching container") if err := cli.ContainerStart(ctx, resp.ID, types.ContainerStartOptions{}); err != nil { - return fmt.Errorf("could not start container: %w", err) + log.Errorf("could not start container: %v", err) + return } statusCh, errCh := cli.ContainerWait(ctx, resp.ID, container.WaitConditionNotRunning) select { case err := <-errCh: if err != nil { - return fmt.Errorf("container returned error: %w", err) + log.Errorf("container returned error: %v", err) + return + } + case okBody := <-statusCh: + if okBody.Error != nil { + log.Errorf("Could not wait on container: %v", err) + return + } else { + log.Debugf("Container finished running with return code: %v", okBody.StatusCode) + pe.Run.Result = &okBody.StatusCode } - case retCode := <-statusCh: - log.Debugf("Container finished running with return code: %v", retCode) } + pe.Run.InProgress = false + out, err := cli.ContainerLogs(ctx, resp.ID, types.ContainerLogsOptions{ShowStdout: true, ShowStderr: true}) if err != nil { - return fmt.Errorf("could not get container logs: %w", err) + log.Errorf("could not get container logs: %w", err) + return } var stdOut bytes.Buffer @@ -151,8 +173,10 @@ func ExecutePipeline(pe PipelineExecution, pipelineConf config.PipelineConf) err stdcopy.StdCopy(&stdOut, &stdErr, out) - log.Debugf("%s", stdOut.Bytes()) - log.Debugf("%s", stdErr.Bytes()) + pe.Run.Stdout = stdOut.Bytes() + pe.Run.Stderr = stdErr.Bytes() - return nil + db.UpdateRunResult(pe.Run) + + return } diff --git a/poll/poll.go b/poll/poll.go index d6e3f5f..367d5b7 100644 --- a/poll/poll.go +++ b/poll/poll.go @@ -3,9 +3,11 @@ package poll import ( "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" @@ -24,17 +26,17 @@ type tag struct { commitHash string } -func pollJob(repoName string, jobConfig config.Job, pipelineConf config.PipelineConf) { +func pollJob(pipeline database.Pipeline, pipelineConf config.PipelineConf, db database.Database) { prevCommits := make(map[string]string) for { - time.Sleep(time.Duration(jobConfig.PollInterval) * time.Second) - log.Infof("Polling repo %v", repoName) + time.Sleep(time.Duration(pipeline.PollInterval) * time.Second) + log.Infof("Polling repo %v", pipeline.Name) repo, err := git.Clone(memory.NewStorage(), nil, &git.CloneOptions{ - URL: jobConfig.URL, + URL: pipeline.Url, }) if err != nil { - log.Errorf("Could not clone repo %v from url %v: %v", repoName, jobConfig.URL, err) + log.Errorf("Could not clone repo %v from url %v: %v", pipeline.Name, pipeline.Url, err) continue } refsToRunFor := []string{} @@ -42,23 +44,23 @@ func pollJob(repoName string, jobConfig config.Job, pipelineConf config.Pipeline // get branches branches, err := repo.Branches() if err != nil { - log.Errorf("Could not enumerate branches in repo %v: %v", repoName, err) + log.Errorf("Could not enumerate branches in repo %v: %v", pipeline.Name, err) continue } branches.ForEach(func(branch *plumbing.Reference) error { - log.Debugf("Processing branch %v from repo %v", branch.Name().String(), repoName) + log.Debugf("Processing branch %v from repo %v (id: %v)", branch.Name().String(), pipeline.Name, pipeline.Id) prevRef, ok := prevCommits[branch.Name().String()] if ok { if branch.Hash().String() != prevRef { - log.Debugf("Queuing job for branch %v in repo %v with hash %v", branch.Name().String(), repoName, branch.Hash().String()) + log.Debugf("Queuing job for branch %v in repo %v (id: %v) with hash %v", branch.Name().String(), pipeline.Name, pipeline.Id, branch.Hash().String()) prevCommits[branch.Name().String()] = branch.Hash().String() refsToRunFor = append(refsToRunFor, branch.Name().String()) } else { - log.Debugf("Branch %v in repo %v has hash %v, which matches the previously seen hash of %v", branch.Name().String(), repoName, branch.Hash().String(), prevRef) + log.Debugf("Branch %v in repo %v (id: %v) has hash %v, which matches the previously seen hash of %v", branch.Name().String(), pipeline.Name, pipeline.Id, branch.Hash().String(), prevRef) } } else { - log.Debugf("Queuing job for newly discovered branch %v in repo %v with hash %v", branch.Name().String(), repoName, branch.Hash().String()) + log.Debugf("Queuing job for newly discovered branch %v in repo %v (id: %v) with hash %v", branch.Name().String(), pipeline.Name, pipeline.Id, branch.Hash().String()) prevCommits[branch.Name().String()] = branch.Hash().String() refsToRunFor = append(refsToRunFor, branch.Name().String()) } @@ -67,22 +69,22 @@ func pollJob(repoName string, jobConfig config.Job, pipelineConf config.Pipeline tags, err := repo.Tags() if err != nil { - log.Errorf("Could not enumerate tags in repo %v: %v", repoName, err) + log.Errorf("Could not enumerate tags in repo %v: %v", pipeline.Name, err) continue } tags.ForEach(func(tag *plumbing.Reference) error { - log.Debugf("Processing tag %v from repo %v", tag.Name().String(), repoName) + log.Debugf("Processing tag %v from repo %v (id: %v)", tag.Name().String(), pipeline.Name, pipeline.Id) prevRef, ok := prevCommits[tag.Name().String()] if ok { if tag.Hash().String() != prevRef { - log.Debugf("Queuing job for tag %v in repo %v with hash %v", tag.Name().String(), repoName, tag.Hash().String()) + log.Debugf("Queuing job for tag %v in repo %v (id: %v) with hash %v", tag.Name().String(), pipeline.Name, pipeline.Id, tag.Hash().String()) prevCommits[tag.Name().String()] = tag.Hash().String() refsToRunFor = append(refsToRunFor, tag.Name().String()) } else { - log.Debugf("Tag %v in repo %v has hash %v, which matches the previously seen hash of %v", tag.Name().String(), repoName, tag.Hash().String(), prevRef) + log.Debugf("Tag %v in repo %v (id: %v) has hash %v, which matches the previously seen hash of %v", tag.Name().String(), pipeline.Name, pipeline.Id, tag.Hash().String(), prevRef) } } else { - log.Debugf("Queuing job for newly discovered tag %v in repo %v with hash %v", tag.Name().String(), repoName, tag.Hash().String()) + log.Debugf("Queuing job for newly discovered tag %v in repo %v (id: %v) with hash %v", tag.Name().String(), pipeline.Name, pipeline.Id, tag.Hash().String()) prevCommits[tag.Name().String()] = tag.Hash().String() refsToRunFor = append(refsToRunFor, tag.Name().String()) } @@ -90,25 +92,55 @@ func pollJob(repoName string, jobConfig config.Job, pipelineConf config.Pipeline }) for _, ref := range refsToRunFor { - log.Debugf("Dispatching job for ref %v in repo %v", ref, repoName) + log.Debugf("Dispatching job for ref %v in repo %v (id: %v)", ref, pipeline.Name, pipeline.Id) - pe := pipeline_executor.PipelineExecution{ - Name: repoName, - Job: jobConfig, - Ref: ref, + run, err := db.CreateRun(pipeline.Id) + if err != nil { + log.Errorf("Could not create run for pipeline with id \"%v\": ", pipeline.Id, err) + continue } - pipeline_executor.ExecutePipeline(pe, pipelineConf) + pe := pipeline_executor.PipelineExecution{ + Pipeline: pipeline, + Ref: ref, + Run: run, + } + + go pipeline_executor.ExecutePipeline(pe, db, pipelineConf) } } } -func StartPolling(conf config.Config) { - for jobName, job := range conf.Jobs { - if job.PollInterval == 0 { +func launchPollJobs(conf config.PipelineConf, db database.Database, pollChan chan uuid.UUID) { + pipelines, err := db.GetPipelines() + if err != nil { + log.Errorf("Could not get pipelines from database: %w", err) + return + } + + for _, pipeline := range pipelines { + if pipeline.PollInterval == 0 { continue } else { - go pollJob(jobName, job, conf.PipelineConf) + go pollJob(pipeline, conf, db) } } + + for { + jobUUID := <-pollChan + pipeline, err := db.GetPipelineById(jobUUID) + if err != nil { + 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) + } + +} + +func StartPolling(conf config.PipelineConf, db database.Database) chan uuid.UUID { + pollChan := make(chan uuid.UUID) + go launchPollJobs(conf, db, pollChan) + return pollChan } diff --git a/webhook/webhook.go b/webhook/webhook.go index 935caf0..fdd6027 100644 --- a/webhook/webhook.go +++ b/webhook/webhook.go @@ -5,79 +5,97 @@ import ( "strings" "git.ohea.xyz/cursorius/server/config" + "git.ohea.xyz/cursorius/server/database" "git.ohea.xyz/cursorius/server/pipeline_executor" "github.com/go-playground/webhooks/v6/gitea" + "github.com/google/uuid" "github.com/op/go-logging" ) var log = logging.MustGetLogger("cursorius-server") -func CreateWebhookHandler(conf config.Config, mux *http.ServeMux) { - mux.HandleFunc("/webhook/", func(w http.ResponseWriter, r *http.Request) { - switch r.Method { - case "POST": - splitUrl := strings.Split(r.URL.Path, "/") - if len(splitUrl) != 3 { - log.Errorf("Webhook recieved with invalid url \"%v\", ignoring...", r.URL) - return - } +func webhookHandler(w http.ResponseWriter, r *http.Request, db database.Database, conf config.PipelineConf) { + switch r.Method { + case "POST": + splitUrl := strings.Split(r.URL.Path, "/") + if len(splitUrl) != 4 { + log.Errorf("Webhook recieved with invalid url \"%v\", ignoring...", r.URL) + return + } - // get URL path after /webhook/ - // TODO: verify that this handles all valid URL formats - webhookJobName := splitUrl[2] + // get URL path after /webhook/ + // TODO: verify that this handles all valid URL formats + pipelineUUIDStr := splitUrl[2] + webhookUUIDStr := splitUrl[3] - for jobName, jobConfig := range conf.Jobs { - if webhookJobName == jobName { - if jobConfig.Webhook == nil { - log.Errorf("Matching job does not have webhook configuration, ignoring....") + pipelineUUID, err := uuid.Parse(pipelineUUIDStr) + if err != nil { + log.Errorf("Could not parse pipeline UUID: %v", err) + return + } + webhookUUID, err := uuid.Parse(webhookUUIDStr) + if err != nil { + log.Errorf("Could not parse webhook UUID: %v", err) + return + } + + pipeline, err := db.GetPipelineById(pipelineUUID) + if err != nil { + log.Errorf("Could not get webhooks for pipeline with UUID \"%v\": %v", pipelineUUID, err) + return + } + + webhooks, err := db.GetWebhooksForPipeline(pipelineUUID) + if err != nil { + log.Errorf("Could not get webhooks for pipeline with UUID \"%v\": %v", webhookUUID, err) + return + } + + if len(webhooks) < 1 { + log.Errorf("No webhooks configured for pipeline with UUID \"%v\"", webhookUUID) + return + } + + for _, webhook := range webhooks { + if webhook.Id == webhookUUID { + switch webhook.ServerType { + case database.Gitea: + hook, err := gitea.New(gitea.Options.Secret(webhook.Secret)) + if err != nil { + log.Errorf("Could not create Gitea webhook handler: %v", err) return } - switch jobConfig.Webhook.Sender { - case config.Gitea: - hook, err := gitea.New(gitea.Options.Secret(jobConfig.Webhook.Secret)) - if err != nil { - log.Errorf("Could not create Gitea webhook handler: %v", err) - return + payload, err := hook.Parse(r, gitea.PushEvent) + if err != nil { + if err == gitea.ErrEventNotFound { + log.Warning("Got webhook \"%v\" for unexpected event type, ignoring...", webhookUUID) + break } - payload, err := hook.Parse(r, gitea.PushEvent) - if err != nil { - if err == gitea.ErrEventNotFound { - log.Info("Got webhook for unexpected event type, ignoring...") - break - } - log.Errorf("Could not parse webhook: %v", err) - return + } + + switch payload.(type) { + case gitea.PushPayload: + pushPayload := payload.(gitea.PushPayload) + + pe := pipeline_executor.PipelineExecution{ + Pipeline: pipeline, + Ref: pushPayload.Ref, } - log.Infof("Got webhook with payload %v", payload) - - switch payload.(type) { - case gitea.PushPayload: - pushPayload := payload.(gitea.PushPayload) - - pe := pipeline_executor.PipelineExecution{ - Name: webhookJobName, - Job: jobConfig, - Ref: pushPayload.Ref, - } - - pipeline_executor.ExecutePipeline(pe, conf.PipelineConf) - } - - return - default: - log.Errorf("Job configured with unknown webhook sender \"%v\", igonring...", jobConfig.Webhook.Sender) - return - + go pipeline_executor.ExecutePipeline(pe, db, conf) } } } - - log.Errorf("Not job configured with name \"%v\", required by webhook with url \"%v\", ignoring...", - webhookJobName, r.URL) - - default: - log.Errorf("Got request with method \"%v\", ignoring...", r.Method) } - }) + log.Errorf("No webhook found with ID \"%v\"", webhookUUID) + + default: + log.Errorf("Got request with method \"%v\", ignoring...", r.Method) + } +} + +func CreateWebhookHandler(db database.Database, conf config.PipelineConf, mux *http.ServeMux) { + mux.HandleFunc("/webhook/", func(w http.ResponseWriter, r *http.Request) { + webhookHandler(w, r, db, conf) + }) }