Compare commits
33 Commits
4a4b6fd185
...
v0.1.0
| Author | SHA1 | Date | |
|---|---|---|---|
| a2acb99689 | |||
| 191b73fe41 | |||
| 3ca1481632 | |||
| c0e33fa52a | |||
| 63529b7174 | |||
| 7e7c49c2e7 | |||
| 712a7b1429 | |||
| 3ae27bffc5 | |||
| 5373a37bee | |||
| 6fee5aa268 | |||
| b475631df6 | |||
| fbf918d627 | |||
| 4069e1b0e1 | |||
| 708fbca91a | |||
| 77a5514578 | |||
| 77a8d0840a | |||
| 62b4e8f17e | |||
| 620c20f717 | |||
| 0979a2379e | |||
| 85ebd856eb | |||
| 6b103d074e | |||
| 664fe8fd09 | |||
| edafd5108a | |||
| c0f6186eac | |||
| bfd05b6a8a | |||
| 6d2936393b | |||
| e4043ae3be | |||
| 7a665aa348 | |||
| d870335d25 | |||
| 4bda3c7a3b | |||
| 724757b23c | |||
| 1e0526a599 | |||
| c1d4f3cc16 |
+502
-16
@@ -8,9 +8,143 @@ import (
|
||||
"github.com/google/uuid"
|
||||
"github.com/graphql-go/graphql"
|
||||
"github.com/graphql-go/handler"
|
||||
"github.com/op/go-logging"
|
||||
)
|
||||
|
||||
var log = logging.MustGetLogger("cursorius-server")
|
||||
|
||||
func createSchema(db database.Database) (graphql.Schema, error) {
|
||||
runnerType := graphql.NewObject(graphql.ObjectConfig{
|
||||
Name: "Runner",
|
||||
Description: "A runner available for use inside of a pipeline.",
|
||||
Fields: graphql.Fields{
|
||||
"id": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The id of the runner.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if runner, ok := p.Source.(database.Runner); ok {
|
||||
return runner.Id, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"name": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The name of the runner.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if runner, ok := p.Source.(database.Runner); ok {
|
||||
return runner.Name, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"token": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The token.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if runner, ok := p.Source.(database.Runner); ok {
|
||||
return runner.Token, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
secretType := graphql.NewObject(graphql.ObjectConfig{
|
||||
Name: "Secret",
|
||||
Description: "A secret available for use inside of a pipeline.",
|
||||
Fields: graphql.Fields{
|
||||
"id": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The id of the secret.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if secret, ok := p.Source.(database.Secret); ok {
|
||||
return secret.Id, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"name": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The name of the secret.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if secret, ok := p.Source.(database.Secret); ok {
|
||||
return secret.Name, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"secret": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The secret.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if secret, ok := p.Source.(database.Secret); ok {
|
||||
return secret.Secret, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
cloneCredentialType := graphql.NewObject(graphql.ObjectConfig{
|
||||
Name: "CloneCredential",
|
||||
Description: "A credential for authenticating with the pipeline source host.",
|
||||
Fields: graphql.Fields{
|
||||
"id": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The id of the credential.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if credential, ok := p.Source.(database.CloneCredential); ok {
|
||||
return credential.Id, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"name": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The name of the credential.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if credential, ok := p.Source.(database.CloneCredential); ok {
|
||||
return credential.Name, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"type": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The credential type.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if credential, ok := p.Source.(database.CloneCredential); ok {
|
||||
return credential.Type, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"username": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The username to user with the credential.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if credential, ok := p.Source.(database.CloneCredential); ok {
|
||||
return credential.Username, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"secret": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The secret for the credential.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if credential, ok := p.Source.(database.CloneCredential); ok {
|
||||
return credential.Secret, nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
webhookType := graphql.NewObject(graphql.ObjectConfig{
|
||||
Name: "Webhook",
|
||||
Description: "A webhook for triggering pipelines",
|
||||
@@ -48,6 +182,74 @@ 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
|
||||
},
|
||||
},
|
||||
"inProgress": &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
|
||||
},
|
||||
},
|
||||
"buildOutput": &graphql.Field{
|
||||
Type: graphql.String,
|
||||
Description: "Logs of the top level container build for the run.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if run, ok := p.Source.(database.Run); ok {
|
||||
return string(run.BuildOutput), 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 string(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 string(run.Stderr), nil
|
||||
}
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
pipelineType := graphql.NewObject(graphql.ObjectConfig{
|
||||
Name: "Pipeline",
|
||||
Description: "A pipeline for running ci jobs",
|
||||
@@ -92,23 +294,26 @@ func createSchema(db database.Database) (graphql.Schema, error) {
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"createWebhook": &graphql.Field{
|
||||
Type: webhookType,
|
||||
Description: "Create a new webhook",
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"type": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
webhook, err := db.CreateWebhook(
|
||||
database.WebhookSender(params.Args["type"].(string)),
|
||||
params.Source.(database.Pipeline).Id,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
"cloneCredential": &graphql.Field{
|
||||
Type: cloneCredentialType,
|
||||
Description: "The configured credential for cloning the pipeline source.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if pipeline, ok := p.Source.(database.Pipeline); ok {
|
||||
if pipeline.CloneCredential != nil {
|
||||
return db.GetCloneCredentialById(*pipeline.CloneCredential)
|
||||
}
|
||||
}
|
||||
return webhook, nil
|
||||
return nil, nil
|
||||
},
|
||||
},
|
||||
"secrets": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.NewList(graphql.NewNonNull(secretType))),
|
||||
Description: "The list of secrets for the pipeline.",
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
if pipeline, ok := p.Source.(database.Pipeline); ok {
|
||||
return db.GetSecretsForPipeline(pipeline.Id)
|
||||
}
|
||||
return []database.Secret{}, nil
|
||||
},
|
||||
},
|
||||
"webhooks": &graphql.Field{
|
||||
@@ -121,6 +326,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
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
@@ -143,6 +358,50 @@ 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()
|
||||
},
|
||||
},
|
||||
"CloneCredential": &graphql.Field{
|
||||
Type: cloneCredentialType,
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"id": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
Description: "The id of the requested credential.",
|
||||
},
|
||||
},
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
id, err := uuid.Parse(p.Args["id"].(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return db.GetCloneCredentialById(id)
|
||||
},
|
||||
},
|
||||
"CloneCredentials": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.NewList(cloneCredentialType)),
|
||||
Args: graphql.FieldConfigArgument{},
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
return db.GetCredentials()
|
||||
},
|
||||
},
|
||||
"Secrets": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.NewList(secretType)),
|
||||
Args: graphql.FieldConfigArgument{},
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
return db.GetSecrets()
|
||||
},
|
||||
},
|
||||
"Runners": &graphql.Field{
|
||||
Type: graphql.NewNonNull(graphql.NewList(runnerType)),
|
||||
Args: graphql.FieldConfigArgument{},
|
||||
Resolve: func(p graphql.ResolveParams) (interface{}, error) {
|
||||
return db.GetRunners()
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
@@ -162,6 +421,9 @@ func createSchema(db database.Database) (graphql.Schema, error) {
|
||||
"pollInterval": &graphql.ArgumentConfig{
|
||||
Type: graphql.Int,
|
||||
},
|
||||
"cloneCredentialId": &graphql.ArgumentConfig{
|
||||
Type: graphql.String,
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
var interval int
|
||||
@@ -170,14 +432,238 @@ func createSchema(db database.Database) (graphql.Schema, error) {
|
||||
} else {
|
||||
interval = 0
|
||||
}
|
||||
|
||||
var credential *uuid.UUID
|
||||
if credentialVal, ok := params.Args["cloneCredentialId"]; ok {
|
||||
id, err := uuid.Parse(credentialVal.(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
credential = &id
|
||||
} else {
|
||||
credential = nil
|
||||
}
|
||||
|
||||
pipeline, err := db.CreatePipeline(
|
||||
params.Args["name"].(string),
|
||||
params.Args["url"].(string),
|
||||
interval,
|
||||
credential,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return pipeline, nil
|
||||
},
|
||||
},
|
||||
"createWebhook": &graphql.Field{
|
||||
Type: webhookType,
|
||||
Description: "Create a new webhook",
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"type": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
"pipelineId": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
|
||||
id, err := uuid.Parse(params.Args["id"].(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
webhook, err := db.CreateWebhook(
|
||||
database.WebhookSender(params.Args["type"].(string)),
|
||||
id,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return webhook, nil
|
||||
},
|
||||
},
|
||||
"createCloneCredential": &graphql.Field{
|
||||
Type: cloneCredentialType,
|
||||
Description: "Create a new CloneCredential",
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"name": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
"type": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
"username": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
"secret": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
|
||||
credential, err := db.CreateCredential(
|
||||
params.Args["name"].(string),
|
||||
database.CloneCredentialType(params.Args["type"].(string)),
|
||||
params.Args["username"].(string),
|
||||
params.Args["secret"].(string),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return credential, nil
|
||||
},
|
||||
},
|
||||
"createSecret": &graphql.Field{
|
||||
Type: secretType,
|
||||
Description: "Create a new secret",
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"name": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
"secret": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
|
||||
secret, err := db.CreateSecret(
|
||||
params.Args["name"].(string),
|
||||
params.Args["secret"].(string),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return secret, nil
|
||||
},
|
||||
},
|
||||
"createRunner": &graphql.Field{
|
||||
Type: runnerType,
|
||||
Description: "Create a new runner",
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"name": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
|
||||
runner, err := db.CreateRunner(
|
||||
params.Args["name"].(string),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return runner, nil
|
||||
},
|
||||
},
|
||||
"setPipelineCloneCredential": &graphql.Field{
|
||||
Type: pipelineType,
|
||||
Description: "Set the CloneCredential used by a pipeline to clone the source repo",
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"cloneCredentialId": &graphql.ArgumentConfig{
|
||||
Type: graphql.String,
|
||||
},
|
||||
"pipelineId": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
|
||||
pipelineId, err := uuid.Parse(params.Args["pipelineId"].(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if cloneCredentialIdVal, ok := params.Args["cloneCredentialId"]; ok {
|
||||
cloneCredentialId, err := uuid.Parse(cloneCredentialIdVal.(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pipeline, err := db.SetPipelineCloneCredential(pipelineId, &cloneCredentialId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pipeline, nil
|
||||
} else {
|
||||
pipeline, err := db.SetPipelineCloneCredential(pipelineId, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pipeline, nil
|
||||
}
|
||||
|
||||
},
|
||||
},
|
||||
"addSecretToPipeline": &graphql.Field{
|
||||
Type: pipelineType,
|
||||
Description: "Allow a secret to be accessed by a pipeline.",
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"secretId": &graphql.ArgumentConfig{
|
||||
Type: graphql.String,
|
||||
},
|
||||
"pipelineId": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
|
||||
secretId, err := uuid.Parse(params.Args["secretId"].(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pipelineId, err := uuid.Parse(params.Args["pipelineId"].(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = db.AssignSecretToPipeline(pipelineId, secretId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pipeline, err := db.GetPipelineById(pipelineId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pipeline, nil
|
||||
},
|
||||
},
|
||||
"removeSecretFromPipeline": &graphql.Field{
|
||||
Type: pipelineType,
|
||||
Description: "Remove a pipeline's access to a secret.",
|
||||
Args: graphql.FieldConfigArgument{
|
||||
"secretId": &graphql.ArgumentConfig{
|
||||
Type: graphql.String,
|
||||
},
|
||||
"pipelineId": &graphql.ArgumentConfig{
|
||||
Type: graphql.NewNonNull(graphql.String),
|
||||
},
|
||||
},
|
||||
Resolve: func(params graphql.ResolveParams) (interface{}, error) {
|
||||
|
||||
secretId, err := uuid.Parse(params.Args["secretId"].(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pipelineId, err := uuid.Parse(params.Args["pipelineId"].(string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = db.RemoveSecretFromPipeline(pipelineId, secretId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
pipeline, err := db.GetPipelineById(pipelineId)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return pipeline, nil
|
||||
},
|
||||
},
|
||||
|
||||
@@ -6,24 +6,6 @@ import (
|
||||
"git.ohea.xyz/golang/config"
|
||||
)
|
||||
|
||||
type WebhookSender string
|
||||
|
||||
const (
|
||||
Gitea WebhookSender = "gitea"
|
||||
)
|
||||
|
||||
type Webhook struct {
|
||||
Sender WebhookSender
|
||||
Secret string
|
||||
}
|
||||
|
||||
type Job struct {
|
||||
URL string
|
||||
Webhook *Webhook
|
||||
Cron *string
|
||||
PollInterval uint64
|
||||
}
|
||||
|
||||
type Runner struct {
|
||||
Secret string
|
||||
}
|
||||
@@ -60,7 +42,6 @@ type Config struct {
|
||||
Port int
|
||||
DBConfig DBConfig
|
||||
PipelineConf PipelineConf
|
||||
Jobs map[string]Job
|
||||
Runners map[string]Runner
|
||||
}
|
||||
|
||||
@@ -88,7 +69,6 @@ func GetConfig() (config.Config[Config], error) {
|
||||
Source: "/opt/cursorius/working",
|
||||
},
|
||||
},
|
||||
Jobs: make(map[string]Job),
|
||||
Runners: make(map[string]Runner),
|
||||
},
|
||||
}
|
||||
|
||||
+68
-16
@@ -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) {
|
||||
@@ -40,9 +40,15 @@ func LaunchDB(conf config.DBConfig) (Database, error) {
|
||||
db := Database{}
|
||||
|
||||
var err error
|
||||
log.Infof("Connecting to database with URL \"%v\"", dbURLNoPasswd)
|
||||
db.Conn, err = pgxpool.New(context.Background(), dbURL)
|
||||
if err != nil {
|
||||
return db, fmt.Errorf("could not create database pool: %w", err)
|
||||
}
|
||||
|
||||
// sleep until we can sucessfully acquire a connection
|
||||
for i := 0; i < 10; i++ {
|
||||
log.Infof("Connecting to database with URL \"%v\" (attempt %v)", dbURLNoPasswd, i)
|
||||
db.Conn, err = pgx.Connect(context.Background(), dbURL)
|
||||
_, err = db.Conn.Acquire(context.Background())
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
@@ -80,19 +86,52 @@ 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 clone_credentials (
|
||||
id UUID PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
type TEXT NOT NULL,
|
||||
username TEXT NOT NULL,
|
||||
secret TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE pipelines (
|
||||
id UUID PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
url TEXT NOT NULL,
|
||||
poll_interval INTEGER
|
||||
id UUID PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
url TEXT NOT NULL,
|
||||
poll_interval INTEGER,
|
||||
clone_credential UUID DEFAULT NULL,
|
||||
|
||||
CONSTRAINT fk_clone_credential
|
||||
FOREIGN KEY(clone_credential)
|
||||
REFERENCES clone_credentials(id)
|
||||
);
|
||||
|
||||
CREATE TABLE secrets (
|
||||
id UUID PRIMARY KEY,
|
||||
name TEXT NOT NULL UNIQUE,
|
||||
secret TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE pipeline_secret_mappings (
|
||||
pipeline UUID NOT NULL,
|
||||
secret UUID NOT NULL,
|
||||
|
||||
CONSTRAINT fk_pipeline
|
||||
FOREIGN KEY(pipeline)
|
||||
REFERENCES pipelines(id),
|
||||
|
||||
CONSTRAINT fk_secret
|
||||
FOREIGN KEY(secret)
|
||||
REFERENCES secrets(id)
|
||||
);
|
||||
|
||||
CREATE TABLE webhooks (
|
||||
@@ -106,16 +145,14 @@ CREATE TABLE webhooks (
|
||||
REFERENCES pipelines(id)
|
||||
);
|
||||
|
||||
CREATE TABLE runners (
|
||||
id UUID PRIMARY KEY,
|
||||
name TEXT,
|
||||
secret TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE runs (
|
||||
id UUID PRIMARY KEY,
|
||||
pipeline UUID,
|
||||
result BOOLEAN NOT NULL,
|
||||
in_progress BOOLEAN DEFAULT NULL,
|
||||
build_output TEXT DEFAULT NULL,
|
||||
result BIGINT DEFAULT NULL,
|
||||
stdout TEXT DEFAULT NULL,
|
||||
stderr TEXT DEFAULT NULL,
|
||||
|
||||
CONSTRAINT fk_pipeline
|
||||
FOREIGN KEY(pipeline)
|
||||
@@ -137,6 +174,21 @@ CREATE TABLE command_executions (
|
||||
REFERENCES runs(id)
|
||||
);
|
||||
|
||||
CREATE TABLE runners (
|
||||
id UUID PRIMARY KEY,
|
||||
name TEXT NOT NULL UNIQUE,
|
||||
token TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE pipeline_refs (
|
||||
name TEXT PRIMARY KEY NOT NULL,
|
||||
pipeline_id UUID NOT NULL,
|
||||
hash TEXT NOT NULL,
|
||||
|
||||
CONSTRAINT fk_pipeline_id
|
||||
FOREIGN KEY(pipeline_id)
|
||||
REFERENCES pipelines(id)
|
||||
);
|
||||
`
|
||||
|
||||
_, err := conn.Exec(context.Background(), createTablesQuery)
|
||||
|
||||
+523
-18
@@ -3,11 +3,42 @@ package database
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"regexp"
|
||||
|
||||
"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, clone_credential
|
||||
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, &pipeline.CloneCredential); 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 +48,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,29 +56,71 @@ 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, credential *uuid.UUID) (Pipeline, error) {
|
||||
query := `
|
||||
INSERT INTO pipelines (id, name, url, poll_interval)
|
||||
VALUES (uuid_generate_v4(), $1, $2, $3)
|
||||
INSERT INTO pipelines (id, name, url, poll_interval, clone_credential)
|
||||
VALUES (uuid_generate_v4(), $1, $2, $3, $4)
|
||||
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, credential).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)
|
||||
pipeline.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
|
||||
|
||||
return pipeline, nil
|
||||
}
|
||||
|
||||
func (db *Database) SetPipelineCloneCredential(pipelineId uuid.UUID, credentialId *uuid.UUID) (Pipeline, error) {
|
||||
query := `
|
||||
UPDATE pipelines
|
||||
SET clone_credential=$1
|
||||
WHERE id=$2
|
||||
RETURNING name, url, poll_interval, clone_credential;`
|
||||
|
||||
pipeline := Pipeline{
|
||||
Id: pipelineId,
|
||||
}
|
||||
|
||||
err := db.Conn.QueryRow(context.Background(),
|
||||
query, credentialId, 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) RemovePipelineCredential(pipelineId uuid.UUID) (Pipeline, error) {
|
||||
query := `
|
||||
UPDATE pipelines
|
||||
SET credential=null
|
||||
WHERE id=$1
|
||||
RETURNING name, url, poll_interval, clone_credential;`
|
||||
|
||||
pipeline := Pipeline{
|
||||
Id: pipelineId,
|
||||
}
|
||||
|
||||
err := db.Conn.QueryRow(context.Background(),
|
||||
query, 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) GetWebhooksForPipeline(id uuid.UUID) ([]Webhook, error) {
|
||||
query := `
|
||||
SELECT id, server_type, secret
|
||||
@@ -58,7 +131,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 +152,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 +162,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 +170,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 +179,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 +193,437 @@ RETURNING id, server_type, secret, pipeline;`
|
||||
|
||||
return webhook, nil
|
||||
}
|
||||
|
||||
func (db *Database) CreateCredential(name string, credentialtype CloneCredentialType, username string, secret string) (CloneCredential, error) {
|
||||
query := `
|
||||
INSERT INTO clone_credentials (id, name, type, username, secret)
|
||||
VALUES(uuid_generate_v4(), $1, $2, $3, $4)
|
||||
RETURNING id, name, type, username, secret;`
|
||||
|
||||
credential := CloneCredential{}
|
||||
var idStr string
|
||||
err := db.Conn.QueryRow(
|
||||
context.Background(),
|
||||
query,
|
||||
name,
|
||||
string(credentialtype),
|
||||
username,
|
||||
secret,
|
||||
).Scan(&idStr, &credential.Name, &credential.Type, &credential.Username, &credential.Secret)
|
||||
if err != nil {
|
||||
return credential, err
|
||||
}
|
||||
|
||||
id, err := uuid.Parse(idStr)
|
||||
if err != nil {
|
||||
return credential, err
|
||||
}
|
||||
|
||||
credential.Id = id
|
||||
|
||||
return credential, nil
|
||||
}
|
||||
|
||||
func (db *Database) GetCloneCredentialById(id uuid.UUID) (CloneCredential, error) {
|
||||
query := `
|
||||
SELECT name, type, username, secret
|
||||
FROM clone_credentials
|
||||
WHERE id=$1;`
|
||||
|
||||
log.Debugf("requested credential with id %v", id)
|
||||
|
||||
credential := CloneCredential{
|
||||
Id: id,
|
||||
}
|
||||
|
||||
err := db.Conn.QueryRow(context.Background(), query, id).Scan(&credential.Name, &credential.Type, &credential.Username, &credential.Secret)
|
||||
if err != nil {
|
||||
return credential, fmt.Errorf("Could not query database for credential with id %v: %w", id.String(), err)
|
||||
}
|
||||
|
||||
return credential, nil
|
||||
}
|
||||
|
||||
func (db *Database) GetCredentials() ([]CloneCredential, error) {
|
||||
query := `
|
||||
SELECT id, name, type, username, secret
|
||||
FROM clone_credentials;`
|
||||
|
||||
credentials := make([]CloneCredential, 0)
|
||||
|
||||
rows, err := db.Conn.Query(context.Background(), query)
|
||||
if err != nil {
|
||||
return credentials, fmt.Errorf("Could not query database for credentials: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var credential CloneCredential
|
||||
var idStr string
|
||||
if err := rows.Scan(&idStr, &credential.Name, &credential.Type, &credential.Username, &credential.Secret); err != nil {
|
||||
return credentials, err
|
||||
}
|
||||
|
||||
credential.Id, err = uuid.Parse(idStr)
|
||||
if err != nil {
|
||||
return credentials, err
|
||||
}
|
||||
credentials = append(credentials, credential)
|
||||
}
|
||||
|
||||
return credentials, 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) UpdateRunBuildOutput(runId uuid.UUID, buildResult string) error {
|
||||
query := `
|
||||
UPDATE runs
|
||||
SET build_output=$1
|
||||
WHERE id=$2;`
|
||||
|
||||
_, err := db.Conn.Exec(context.Background(),
|
||||
query, buildResult, runId)
|
||||
|
||||
return err
|
||||
|
||||
}
|
||||
|
||||
func (db *Database) UpdateRunResult(r Run) error {
|
||||
query := `
|
||||
UPDATE runs
|
||||
SET in_progress=$1, result=$2, stdout=$3, stderr=$4
|
||||
WHERE id=$5;`
|
||||
|
||||
// TODO: does r.Result need a pointer derefrence?
|
||||
_, err := db.Conn.Exec(context.Background(),
|
||||
query, r.InProgress, r.Result, r.Stdout, r.Stderr, r.Id)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (db *Database) GetRunsForPipeline(pipelineId uuid.UUID) ([]Run, error) {
|
||||
query := `
|
||||
SELECT id, in_progress, result, build_output, 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.BuildOutput,
|
||||
&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
|
||||
}
|
||||
|
||||
func (db *Database) GetPipelineRefs(pipelineId uuid.UUID) (map[string]string, error) {
|
||||
query := `
|
||||
SELECT name, hash
|
||||
FROM pipeline_refs
|
||||
WHERE pipeline_id=$1;`
|
||||
|
||||
refsMap := make(map[string]string)
|
||||
|
||||
refs, err := db.Conn.Query(context.Background(), query, pipelineId)
|
||||
if err != nil {
|
||||
return refsMap, fmt.Errorf("Could not get pipeline refs for pipeline with id \"%v\": %w", pipelineId, err)
|
||||
}
|
||||
defer refs.Close()
|
||||
|
||||
for refs.Next() {
|
||||
var name string
|
||||
var hash string
|
||||
if err := refs.Scan(
|
||||
&name,
|
||||
&hash,
|
||||
); err != nil {
|
||||
return refsMap, err
|
||||
}
|
||||
|
||||
refsMap[name] = hash
|
||||
}
|
||||
|
||||
return refsMap, nil
|
||||
}
|
||||
|
||||
func (db *Database) UpdatePipelineRefs(pipelineId uuid.UUID, refsMap map[string]string) error {
|
||||
|
||||
query := `
|
||||
INSERT INTO pipeline_refs(name, pipeline_id, hash)
|
||||
VALUES($1, $2, $3)
|
||||
ON CONFLICT (name)
|
||||
DO
|
||||
UPDATE SET hash=$3;`
|
||||
|
||||
for name, hash := range refsMap {
|
||||
_, err := db.Conn.Exec(context.Background(), query, name, pipelineId, hash)
|
||||
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (db *Database) GetSecrets() ([]Secret, error) {
|
||||
query := `
|
||||
SELECT id, name, secret
|
||||
FROM secrets;`
|
||||
|
||||
secrets := make([]Secret, 0)
|
||||
|
||||
rows, err := db.Conn.Query(context.Background(), query)
|
||||
if err != nil {
|
||||
return secrets, fmt.Errorf("Could not query database for secrets: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var secret Secret
|
||||
var idStr string
|
||||
if err := rows.Scan(&idStr, &secret.Name, &secret.Secret); err != nil {
|
||||
return secrets, err
|
||||
}
|
||||
|
||||
secret.Id, err = uuid.Parse(idStr)
|
||||
if err != nil {
|
||||
return secrets, err
|
||||
}
|
||||
secrets = append(secrets, secret)
|
||||
}
|
||||
|
||||
return secrets, nil
|
||||
}
|
||||
|
||||
func (db *Database) GetSecretById(id uuid.UUID) (Secret, error) {
|
||||
query := `
|
||||
SELECT id, name, secret
|
||||
FROM secrets
|
||||
WHERE id=$1;`
|
||||
|
||||
secret := Secret{
|
||||
Id: id,
|
||||
}
|
||||
|
||||
err := db.Conn.QueryRow(context.Background(), query, id).Scan(&secret.Name, &secret.Secret)
|
||||
if err != nil {
|
||||
return secret, fmt.Errorf("Could not query database for secret with id %v: %w", id.String(), err)
|
||||
}
|
||||
|
||||
return secret, nil
|
||||
}
|
||||
|
||||
func (db *Database) CreateSecret(name string, secret string) (Secret, error) {
|
||||
s := Secret{}
|
||||
|
||||
// validate that the secret is only A-Z or underscores and less than 256 characters
|
||||
if len(name) > 256 {
|
||||
return s, fmt.Errorf("secret name must be 256 characters or less")
|
||||
}
|
||||
|
||||
validName := regexp.MustCompile(`[A-Z0-9_]+$`)
|
||||
if !validName.MatchString(name) {
|
||||
return s, fmt.Errorf("secren name must be made up of only uppercase letters, numbers, and underscores")
|
||||
}
|
||||
|
||||
query := `
|
||||
INSERT INTO secrets (id, name, secret)
|
||||
VALUES (uuid_generate_v4(), $1, $2)
|
||||
RETURNING id, name, secret;`
|
||||
|
||||
var idStr string
|
||||
err := db.Conn.QueryRow(context.Background(), query, name, secret).Scan(&idStr, &s.Name, &s.Secret)
|
||||
if err != nil {
|
||||
return s, fmt.Errorf("Could not create secret: %w", err)
|
||||
}
|
||||
|
||||
s.Id, err = uuid.Parse(idStr)
|
||||
if err != nil {
|
||||
return s, fmt.Errorf("Could not parse UUID generated by DB: %w", err)
|
||||
}
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (db *Database) AssignSecretToPipeline(pipelineId uuid.UUID, secretId uuid.UUID) error {
|
||||
query := `
|
||||
INSERT INTO pipeline_secret_mappings (pipeline, secret)
|
||||
VALUES ($1, $2);`
|
||||
|
||||
_, err := db.Conn.Exec(context.Background(), query, pipelineId, secretId)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (db *Database) RemoveSecretFromPipeline(pipelineId uuid.UUID, secretId uuid.UUID) error {
|
||||
// TODO: implement this
|
||||
return fmt.Errorf("Not implemented")
|
||||
}
|
||||
|
||||
func (db *Database) GetSecretsForPipeline(pipelineId uuid.UUID) ([]Secret, error) {
|
||||
query := `
|
||||
SELECT
|
||||
secrets.id, secrets.name, secrets.secret
|
||||
FROM
|
||||
secrets INNER JOIN pipeline_secret_mappings
|
||||
ON secrets.id = pipeline_secret_mappings.secret
|
||||
WHERE
|
||||
pipeline_secret_mappings.pipeline=$1
|
||||
;`
|
||||
|
||||
secrets := make([]Secret, 0)
|
||||
|
||||
rows, err := db.Conn.Query(context.Background(), query, pipelineId)
|
||||
if err != nil {
|
||||
return secrets, fmt.Errorf("Could not get secrets for pipeline with id \"%v\": %w", pipelineId, err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var secret Secret
|
||||
var idStr string
|
||||
if err := rows.Scan(
|
||||
&idStr,
|
||||
&secret.Name,
|
||||
&secret.Secret,
|
||||
); err != nil {
|
||||
return secrets, err
|
||||
}
|
||||
|
||||
secret.Id, err = uuid.Parse(idStr)
|
||||
if err != nil {
|
||||
return secrets, err
|
||||
}
|
||||
secrets = append(secrets, secret)
|
||||
}
|
||||
|
||||
return secrets, nil
|
||||
}
|
||||
|
||||
func (db *Database) GetRunners() ([]Runner, error) {
|
||||
query := `
|
||||
SELECT id, name, token
|
||||
FROM runners;`
|
||||
|
||||
runners := make([]Runner, 0)
|
||||
|
||||
rows, err := db.Conn.Query(context.Background(), query)
|
||||
if err != nil {
|
||||
return runners, fmt.Errorf("Could not query database for runners: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var runner Runner
|
||||
var idStr string
|
||||
if err := rows.Scan(&idStr, &runner.Name, &runner.Token); err != nil {
|
||||
return runners, err
|
||||
}
|
||||
|
||||
runner.Id, err = uuid.Parse(idStr)
|
||||
if err != nil {
|
||||
return runners, err
|
||||
}
|
||||
runners = append(runners, runner)
|
||||
}
|
||||
|
||||
return runners, nil
|
||||
}
|
||||
|
||||
func (db *Database) GetRunnerById(id uuid.UUID) (Runner, error) {
|
||||
query := `
|
||||
SELECT name, token
|
||||
FROM runners
|
||||
WHERE id=$1;`
|
||||
|
||||
runner := Runner{
|
||||
Id: id,
|
||||
}
|
||||
|
||||
err := db.Conn.QueryRow(context.Background(), query, id).Scan(&runner.Name, &runner.Token)
|
||||
if err != nil {
|
||||
return runner, fmt.Errorf("Could not query database for runner with id %v: %w", id.String(), err)
|
||||
}
|
||||
|
||||
return runner, nil
|
||||
}
|
||||
|
||||
func (db *Database) CreateRunner(name string) (Runner, error) {
|
||||
s := Runner{}
|
||||
|
||||
// validate that the runner name is only A-Z or underscores and less than 256 characters
|
||||
if len(name) > 256 {
|
||||
return s, fmt.Errorf("runner name must be 256 characters or less")
|
||||
}
|
||||
|
||||
validName := regexp.MustCompile(`[A-Z0-9_]+$`)
|
||||
if !validName.MatchString(name) {
|
||||
return s, fmt.Errorf("runner name must be made up of only uppercase letters, numbers, and underscores")
|
||||
}
|
||||
|
||||
query := `
|
||||
INSERT INTO runners (id, name, token)
|
||||
VALUES
|
||||
(
|
||||
uuid_generate_v4(),
|
||||
$1,
|
||||
(
|
||||
SELECT md5(random()::text)
|
||||
)
|
||||
)
|
||||
RETURNING id, name, token;`
|
||||
|
||||
var idStr string
|
||||
err := db.Conn.QueryRow(context.Background(), query, name).Scan(&idStr, &s.Name, &s.Token)
|
||||
if err != nil {
|
||||
return s, fmt.Errorf("Could not create runner: %w", err)
|
||||
}
|
||||
|
||||
s.Id, err = uuid.Parse(idStr)
|
||||
if err != nil {
|
||||
return s, fmt.Errorf("Could not parse UUID generated by DB: %w", err)
|
||||
}
|
||||
|
||||
return s, nil
|
||||
}
|
||||
|
||||
+45
-14
@@ -6,11 +6,38 @@ import (
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type CloneCredentialType string
|
||||
|
||||
const (
|
||||
USER_PASS CloneCredentialType = "USER_PASS"
|
||||
SSH_KEY CloneCredentialType = "SSH_KEY"
|
||||
)
|
||||
|
||||
type CloneCredential struct {
|
||||
Id uuid.UUID
|
||||
Name string
|
||||
Type CloneCredentialType
|
||||
Username string
|
||||
Secret string
|
||||
}
|
||||
|
||||
type Pipeline struct {
|
||||
Id uuid.UUID
|
||||
Name string
|
||||
Url string
|
||||
PollInterval int
|
||||
Id uuid.UUID
|
||||
Name string
|
||||
Url string
|
||||
PollInterval int
|
||||
CloneCredential *uuid.UUID
|
||||
}
|
||||
|
||||
type Secret struct {
|
||||
Id uuid.UUID
|
||||
Name string
|
||||
Secret string
|
||||
}
|
||||
|
||||
type PipelineSecretMapping struct {
|
||||
Pipeline uuid.UUID
|
||||
Secret uuid.UUID
|
||||
}
|
||||
|
||||
type WebhookSender string
|
||||
@@ -26,25 +53,29 @@ type Webhook struct {
|
||||
Pipeline uuid.UUID
|
||||
}
|
||||
|
||||
type Runner struct {
|
||||
Id uuid.UUID
|
||||
Name string
|
||||
Secret string
|
||||
}
|
||||
|
||||
type Run struct {
|
||||
Id uuid.UUID
|
||||
Pipeline uuid.UUID
|
||||
Result bool
|
||||
Id uuid.UUID
|
||||
Pipeline uuid.UUID
|
||||
InProgress bool
|
||||
Result *int64
|
||||
BuildOutput []byte
|
||||
Stdout []byte
|
||||
Stderr []byte
|
||||
}
|
||||
|
||||
type CommandExecution struct {
|
||||
Id uuid.UUID
|
||||
RunId uuid.UUID
|
||||
Command string
|
||||
Command []string
|
||||
ReturnCode int
|
||||
Stdout string
|
||||
Stderr string
|
||||
StartTime time.Time
|
||||
EndTime time.Time
|
||||
}
|
||||
|
||||
type Runner struct {
|
||||
Id uuid.UUID
|
||||
Name string
|
||||
Token string
|
||||
}
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
FROM nginx:latest
|
||||
COPY graphiql.html /usr/share/nginx/html/index.html
|
||||
COPY graphiql.conf /etc/nginx/conf.d/default.conf
|
||||
@@ -4,5 +4,5 @@
|
||||
set -e
|
||||
|
||||
cd /build/server
|
||||
go build .
|
||||
go build -buildvcs=false .
|
||||
./server
|
||||
|
||||
@@ -22,6 +22,14 @@ services:
|
||||
- POSTGRES_DB=cursorius
|
||||
networks:
|
||||
- cursorius
|
||||
graphiql:
|
||||
build:
|
||||
dockerfile: Dockerfile.graphiql
|
||||
ports:
|
||||
- "0.0.0.0:45421:80"
|
||||
networks:
|
||||
- cursorius
|
||||
|
||||
networks:
|
||||
cursorius:
|
||||
external: true
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
upstream backend {
|
||||
server cursorius-server:45420;
|
||||
}
|
||||
|
||||
server {
|
||||
listen 80;
|
||||
listen [::]:80;
|
||||
server_name localhost;
|
||||
|
||||
#access_log /var/log/nginx/host.access.log main;
|
||||
|
||||
location / {
|
||||
root /usr/share/nginx/html;
|
||||
index index.html index.htm;
|
||||
}
|
||||
|
||||
location /graphql {
|
||||
proxy_pass http://backend/graphql;
|
||||
}
|
||||
|
||||
#error_page 404 /404.html;
|
||||
|
||||
# redirect server error pages to the static page /50x.html
|
||||
#
|
||||
error_page 500 502 503 504 /50x.html;
|
||||
location = /50x.html {
|
||||
root /usr/share/nginx/html;
|
||||
}
|
||||
|
||||
# proxy the PHP scripts to Apache listening on 127.0.0.1:80
|
||||
#
|
||||
#location ~ \.php$ {
|
||||
# proxy_pass http://127.0.0.1;
|
||||
#}
|
||||
|
||||
# pass the PHP scripts to FastCGI server listening on 127.0.0.1:9000
|
||||
#
|
||||
#location ~ \.php$ {
|
||||
# root html;
|
||||
# fastcgi_pass 127.0.0.1:9000;
|
||||
# fastcgi_index index.php;
|
||||
# fastcgi_param SCRIPT_FILENAME /scripts$fastcgi_script_name;
|
||||
# include fastcgi_params;
|
||||
#}
|
||||
|
||||
# deny access to .htaccess files, if Apache's document root
|
||||
# concurs with nginx's one
|
||||
#
|
||||
#location ~ /\.ht {
|
||||
# deny all;
|
||||
#}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
<!--
|
||||
* Copyright (c) 2021 GraphQL Contributors
|
||||
* All rights reserved.
|
||||
*
|
||||
* This source code is licensed under the license found in the
|
||||
* LICENSE file in the root directory of this source tree.
|
||||
-->
|
||||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<title>GraphiQL</title>
|
||||
<style>
|
||||
body {
|
||||
height: 100%;
|
||||
margin: 0;
|
||||
width: 100%;
|
||||
overflow: hidden;
|
||||
}
|
||||
|
||||
#graphiql {
|
||||
height: 100vh;
|
||||
}
|
||||
</style>
|
||||
|
||||
<!--
|
||||
This GraphiQL example depends on Promise and fetch, which are available in
|
||||
modern browsers, but can be "polyfilled" for older browsers.
|
||||
GraphiQL itself depends on React DOM.
|
||||
If you do not want to rely on a CDN, you can host these files locally or
|
||||
include them directly in your favored resource bundler.
|
||||
-->
|
||||
<script
|
||||
src="https://unpkg.com/react@17/umd/react.development.js"
|
||||
integrity="sha512-Vf2xGDzpqUOEIKO+X2rgTLWPY+65++WPwCHkX2nFMu9IcstumPsf/uKKRd5prX3wOu8Q0GBylRpsDB26R6ExOg=="
|
||||
crossorigin="anonymous"
|
||||
></script>
|
||||
<script
|
||||
src="https://unpkg.com/react-dom@17/umd/react-dom.development.js"
|
||||
integrity="sha512-Wr9OKCTtq1anK0hq5bY3X/AvDI5EflDSAh0mE9gma+4hl+kXdTJPKZ3TwLMBcrgUeoY0s3dq9JjhCQc7vddtFg=="
|
||||
crossorigin="anonymous"
|
||||
></script>
|
||||
|
||||
<!--
|
||||
These two files can be found in the npm module, however you may wish to
|
||||
copy them directly into your environment, or perhaps include them in your
|
||||
favored resource bundler.
|
||||
-->
|
||||
<link rel="stylesheet" href="https://unpkg.com/graphiql/graphiql.min.css" />
|
||||
</head>
|
||||
|
||||
<body>
|
||||
<div id="graphiql">Loading...</div>
|
||||
<script
|
||||
src="https://unpkg.com/graphiql/graphiql.min.js"
|
||||
type="application/javascript"
|
||||
></script>
|
||||
<script>
|
||||
ReactDOM.render(
|
||||
React.createElement(GraphiQL, {
|
||||
fetcher: GraphiQL.createFetcher({
|
||||
//url: 'https://swapi-graphql.netlify.app/.netlify/functions/index',
|
||||
url: 'http://127.0.0.1:45421/graphql',
|
||||
}),
|
||||
defaultEditorToolsVisibility: true,
|
||||
}),
|
||||
document.getElementById('graphiql'),
|
||||
);
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
+49
-9
@@ -8,9 +8,13 @@ mkdir -p _working/jobs
|
||||
base_default_compose_files="docker/docker-compose.yml"
|
||||
default_compose_files="$base_default_compose_files"
|
||||
|
||||
if [ -f "docker/docker-compose.override.yml" ]
|
||||
override_compose="docker/docker-compose.override.yml"
|
||||
gitea_compose="docker/gitea-override.yml"
|
||||
|
||||
|
||||
if [ -f "$override_compose" ]
|
||||
then
|
||||
default_compose_files+=" docker/docker-compose.override.yml"
|
||||
default_compose_files+=" $override_compose"
|
||||
else
|
||||
default_compose_files="docker/docker-compose.yml"
|
||||
fi
|
||||
@@ -20,9 +24,9 @@ function stop_containers {
|
||||
if [ "$current_containers" == "default" ]
|
||||
then
|
||||
compose_files="$default_compose_files"
|
||||
elif [ "$current_containers" == "webhook" ]
|
||||
elif [ "$current_containers" == "gitea" ]
|
||||
then
|
||||
compose_files="$default_compose_files docker/webhook-override.yml"
|
||||
compose_files="$default_compose_files $gitea_compose"
|
||||
fi
|
||||
compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
|
||||
docker compose $compose_file_flags down
|
||||
@@ -33,14 +37,42 @@ function show_logs {
|
||||
if [ "$current_containers" == "default" ]
|
||||
then
|
||||
compose_files="$default_compose_files"
|
||||
elif [ "$current_containers" == "webhook" ]
|
||||
elif [ "$current_containers" == "gitea" ]
|
||||
then
|
||||
compose_files="$default_compose_files docker/webhook-override.yml"
|
||||
compose_files="$default_compose_files $gitea_compose"
|
||||
fi
|
||||
compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
|
||||
docker compose $compose_file_flags logs -f
|
||||
}
|
||||
|
||||
function show_ps {
|
||||
current_containers="$(cat _working/current_containers)"
|
||||
if [ "$current_containers" == "default" ]
|
||||
then
|
||||
compose_files="$default_compose_files"
|
||||
elif [ "$current_containers" == "gitea" ]
|
||||
then
|
||||
compose_files="$default_compose_files $gitea_compose"
|
||||
fi
|
||||
compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
|
||||
docker compose $compose_file_flags ps
|
||||
}
|
||||
|
||||
function runprev {
|
||||
current_containers="$(cat _working/current_containers)"
|
||||
if [ "$current_containers" == "default" ]
|
||||
then
|
||||
compose_files="$default_compose_files"
|
||||
elif [ "$current_containers" == "gitea" ]
|
||||
then
|
||||
compose_files="$default_compose_files $gitea_compose"
|
||||
fi
|
||||
compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
|
||||
docker compose $compose_file_flags up --build -d
|
||||
docker compose $compose_file_flags logs -f
|
||||
}
|
||||
|
||||
|
||||
case $1 in
|
||||
"default")
|
||||
echo "default" > _working/current_containers
|
||||
@@ -48,17 +80,25 @@ case $1 in
|
||||
compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
|
||||
docker compose $compose_file_flags up --build -d
|
||||
docker compose $compose_file_flags logs -f;;
|
||||
"webhook")
|
||||
echo "webhook" > _working/current_containers
|
||||
"gitea")
|
||||
echo "gitea" > _working/current_containers
|
||||
stop_containers
|
||||
compose_files="$default_compose_files docker/webhook-override.yml"
|
||||
compose_files="$default_compose_files $gitea_compose"
|
||||
compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
|
||||
docker compose $compose_file_flags up --build -d
|
||||
docker compose $compose_file_flags logs -f;;
|
||||
"dbshell")
|
||||
compose_files="$default_compose_files $gitea_compose"
|
||||
compose_file_flags=$(echo "$compose_files" | tr ' ' '\n' | xargs -I'{}' echo "-f {} " | tr -d '\n')
|
||||
docker compose $compose_file_flags exec cursorius-db psql --user=cursorius;;
|
||||
"stop")
|
||||
stop_containers;;
|
||||
"logs")
|
||||
show_logs;;
|
||||
"ps")
|
||||
show_ps;;
|
||||
"runprev")
|
||||
runprev;;
|
||||
*) echo "ERROR: Unknown param \"$1\"" 2>&1
|
||||
exit 255;;
|
||||
esac
|
||||
|
||||
@@ -14,6 +14,7 @@ require (
|
||||
github.com/graphql-go/graphql v0.8.0
|
||||
github.com/graphql-go/handler v0.2.3
|
||||
github.com/jackc/pgx/v5 v5.2.0
|
||||
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
|
||||
@@ -37,6 +38,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 +55,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
|
||||
|
||||
@@ -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=
|
||||
@@ -856,6 +858,8 @@ github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOl
|
||||
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
|
||||
github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
|
||||
github.com/jessevdk/go-flags v1.5.0/go.mod h1:Fw0T6WPc1dYxT4mKEZRfG5kJhaTDP9pj1c2EWnYs/m4=
|
||||
github.com/jhoonb/archivex v0.0.0-20201016144719-6a343cdae81d h1:q7n+5taxmM+9T2Q7Ydo7YN90FkoDuR5bbzByZwkQqPo=
|
||||
github.com/jhoonb/archivex v0.0.0-20201016144719-6a343cdae81d/go.mod h1:GN1Mg/uXQ6qwXA0HypnUO3xlcQJS9/y68EsHNeuuRa4=
|
||||
github.com/jhump/protoreflect v1.6.1/go.mod h1:RZQ/lnuN+zqeRVpQigTwO6o0AJUkxbnSnpuG7toUTG4=
|
||||
github.com/jhump/protoreflect v1.8.2/go.mod h1:7GcYQDdMU/O/BBrl/cX6PNHpXh6cenjd8pneu5yW7Tg=
|
||||
github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
|
||||
@@ -1443,6 +1447,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 +1675,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=
|
||||
|
||||
+3
-3
@@ -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,
|
||||
|
||||
@@ -40,13 +40,13 @@ func main() {
|
||||
return
|
||||
}
|
||||
|
||||
getRunnerCh, registerCh, err := runnermanager.StartRunnerManager(configData.Config.Runners)
|
||||
getRunnerCh, registerCh, err := runnermanager.StartRunnerManager(configData.Config.Runners, db)
|
||||
if err != nil {
|
||||
log.Errorf("Could not start runner: %v", err)
|
||||
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,
|
||||
|
||||
@@ -64,11 +64,13 @@ func (s *ApiServer) GetRunner(
|
||||
}
|
||||
|
||||
var runnerTagsStr strings.Builder
|
||||
fmt.Fprintf(&runnerTagsStr, "[%v", req.Msg.Tags[0])
|
||||
for _, tag := range req.Msg.Tags[1:] {
|
||||
fmt.Fprintf(&runnerTagsStr, ", %v", tag)
|
||||
if len(req.Msg.Tags) > 0 {
|
||||
fmt.Fprintf(&runnerTagsStr, "[%v", req.Msg.Tags[0])
|
||||
for _, tag := range req.Msg.Tags[1:] {
|
||||
fmt.Fprintf(&runnerTagsStr, ", %v", tag)
|
||||
}
|
||||
fmt.Fprintf(&runnerTagsStr, "]")
|
||||
}
|
||||
fmt.Fprintf(&runnerTagsStr, "]")
|
||||
|
||||
response := <-respChan
|
||||
if response.Err != nil {
|
||||
|
||||
@@ -4,81 +4,162 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/jhoonb/archivex"
|
||||
|
||||
"git.ohea.xyz/cursorius/server/config"
|
||||
"github.com/docker/docker/api/types"
|
||||
"github.com/docker/docker/api/types/container"
|
||||
"github.com/docker/docker/api/types/mount"
|
||||
"github.com/docker/docker/client"
|
||||
"github.com/docker/docker/pkg/stdcopy"
|
||||
"github.com/go-git/go-git/v5"
|
||||
"github.com/go-git/go-git/v5/plumbing/transport"
|
||||
"github.com/go-git/go-git/v5/plumbing/transport/http"
|
||||
"github.com/go-git/go-git/v5/plumbing/transport/ssh"
|
||||
"github.com/op/go-logging"
|
||||
|
||||
"git.ohea.xyz/cursorius/server/config"
|
||||
"git.ohea.xyz/cursorius/server/database"
|
||||
)
|
||||
|
||||
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())
|
||||
cloneFolder := filepath.Join(jobFolder, "repo")
|
||||
|
||||
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)
|
||||
err = os.MkdirAll(cloneFolder, 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)
|
||||
// TODO: should I use go-git here instead of shelling out to raw git?
|
||||
cloneCmd := exec.Command("git", "clone", pe.Job.URL, jobFolder)
|
||||
output, err := cloneCmd.CombinedOutput()
|
||||
log.Infof("Cloning source from URL %v", pe.Pipeline.Url)
|
||||
|
||||
var auth transport.AuthMethod
|
||||
|
||||
if pe.Pipeline.CloneCredential != nil {
|
||||
credential, err := db.GetCloneCredentialById(*pe.Pipeline.CloneCredential)
|
||||
if err != nil {
|
||||
log.Errorf("could not get credenital from db: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
switch credential.Type {
|
||||
case "USER_PASS":
|
||||
log.Debugf("job %v configured to use credential %v", pe.Pipeline.Name, credential.Name)
|
||||
auth = transport.AuthMethod(&http.BasicAuth{
|
||||
Username: credential.Username,
|
||||
Password: credential.Secret,
|
||||
})
|
||||
case "SSH_KEY":
|
||||
publicKeys, err := ssh.NewPublicKeys(credential.Username, []byte(credential.Secret), "")
|
||||
if err != nil {
|
||||
log.Errorf("could not parse credential %v", credential.Name)
|
||||
return
|
||||
}
|
||||
auth = transport.AuthMethod(publicKeys)
|
||||
default:
|
||||
log.Errorf("unsupported credential type %v", credential.Type)
|
||||
return
|
||||
}
|
||||
} else {
|
||||
auth = nil
|
||||
}
|
||||
|
||||
_, err = git.PlainClone(cloneFolder, false, &git.CloneOptions{
|
||||
URL: pe.Pipeline.Url,
|
||||
Auth: auth,
|
||||
})
|
||||
if err != nil {
|
||||
log.Debugf("%s", output)
|
||||
return fmt.Errorf("could not clone source: %w", err)
|
||||
log.Errorf("could not clone repo: %v", 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")
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
imageName := "git.ohea.xyz/cursorius/pipeline-api/cursorius-pipeline:v2"
|
||||
log.Info("Building container")
|
||||
|
||||
log.Infof("Pulling image %v", imageName)
|
||||
pullOutput, err := cli.ImagePull(ctx, imageName, types.ImagePullOptions{})
|
||||
tarFile := filepath.Join(jobFolder, "archive.tar")
|
||||
tar := new(archivex.TarFile)
|
||||
err = tar.Create(tarFile)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not pull image %v: %w", imageName, err)
|
||||
log.Errorf("could not create tarfile: %w", err)
|
||||
return
|
||||
}
|
||||
|
||||
buf, err := io.ReadAll(pullOutput)
|
||||
err = tar.AddAll(cloneFolder, false)
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not read from io.ReadCloser:, %w", err)
|
||||
log.Errorf("could not add repo to tarfile: %w", err)
|
||||
return
|
||||
}
|
||||
log.Infof("%s", buf)
|
||||
|
||||
err = pullOutput.Close()
|
||||
err = tar.Close()
|
||||
if err != nil {
|
||||
return fmt.Errorf("could not close io.ReadCloser: %w", err)
|
||||
log.Errorf("could not close tarfile: %w", err)
|
||||
return
|
||||
}
|
||||
log.Info("Image pulled sucessfully")
|
||||
|
||||
dockerBuildContext, err := os.Open(tarFile)
|
||||
defer dockerBuildContext.Close()
|
||||
|
||||
imageName := fmt.Sprintf("%v-%v:latest", pe.Pipeline.Id.String(), pe.Run.Id.String())
|
||||
|
||||
buildResponse, err := cli.ImageBuild(context.Background(), dockerBuildContext, types.ImageBuildOptions{
|
||||
Tags: []string{imageName},
|
||||
Dockerfile: ".cursorius/Dockerfile",
|
||||
})
|
||||
if err != nil {
|
||||
log.Errorf("could not build container: %w", err)
|
||||
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))
|
||||
if err != nil {
|
||||
log.Errorf("could not update build output for run: %w", err)
|
||||
return
|
||||
}
|
||||
|
||||
err = buildResponse.Body.Close()
|
||||
if err != nil {
|
||||
log.Errorf("Could not close build response body: %w", err)
|
||||
return
|
||||
}
|
||||
|
||||
log.Info("Image built sucessfully")
|
||||
|
||||
hostConfig := container.HostConfig{}
|
||||
|
||||
@@ -87,10 +168,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,
|
||||
@@ -108,42 +191,71 @@ func ExecutePipeline(pe PipelineExecution, pipelineConf config.PipelineConf) err
|
||||
)
|
||||
}
|
||||
|
||||
env := make([]string, 0)
|
||||
|
||||
// set cursorius environment variables
|
||||
env = append(env, []string{
|
||||
fmt.Sprintf("CURSORIUS_RUN_ID=%v", pe.Run.Id),
|
||||
"CURSORIUS_SRC_DIR=/cursorius/src",
|
||||
fmt.Sprintf("CURSORIUS_SERVER_URL=%v", pipelineConf.AccessURL),
|
||||
}...)
|
||||
|
||||
// load secrets into environment
|
||||
secrets, err := db.GetSecretsForPipeline(pe.Pipeline.Id)
|
||||
if err != nil {
|
||||
log.Errorf("Could not get secrets for pipeline", err)
|
||||
return
|
||||
}
|
||||
|
||||
for _, secret := range secrets {
|
||||
// the env name is validated to be just uppercase letters, numbers, and underscores on ingestion
|
||||
env = append(env, fmt.Sprintf("%v=%v", strings.ToUpper(secret.Name), secret.Secret))
|
||||
}
|
||||
|
||||
resp, err := cli.ContainerCreate(ctx,
|
||||
&container.Config{
|
||||
Image: imageName,
|
||||
Cmd: []string{"/launcher.sh"},
|
||||
Tty: false,
|
||||
Env: []string{
|
||||
"CURSORIUS_SRC_DIR=/cursorius/src",
|
||||
fmt.Sprintf("CUROSRIUS_SERVER_URL=%v", pipelineConf.AccessURL),
|
||||
},
|
||||
Env: env,
|
||||
},
|
||||
// TODO: fix running the runner in docker (add VolumesFrom to HostConfig)
|
||||
&hostConfig,
|
||||
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 +263,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
|
||||
}
|
||||
|
||||
+76
-33
@@ -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,22 @@ type tag struct {
|
||||
commitHash string
|
||||
}
|
||||
|
||||
func pollJob(repoName string, jobConfig config.Job, pipelineConf config.PipelineConf) {
|
||||
prevCommits := make(map[string]string)
|
||||
func pollJob(pipeline database.Pipeline, pipelineConf config.PipelineConf, db database.Database) {
|
||||
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)
|
||||
|
||||
prevRefs, err := db.GetPipelineRefs(pipeline.Id)
|
||||
if err != nil {
|
||||
log.Errorf("Could not get pipeline refs from db: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
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,24 +49,24 @@ 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)
|
||||
prevRef, ok := prevCommits[branch.Name().String()]
|
||||
log.Debugf("Processing branch %v from repo %v (id: %v)", branch.Name().String(), pipeline.Name, pipeline.Id)
|
||||
prevRef, ok := prevRefs[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())
|
||||
prevCommits[branch.Name().String()] = 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())
|
||||
prevRefs[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())
|
||||
prevCommits[branch.Name().String()] = 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())
|
||||
prevRefs[branch.Name().String()] = branch.Hash().String()
|
||||
refsToRunFor = append(refsToRunFor, branch.Name().String())
|
||||
}
|
||||
return nil
|
||||
@@ -67,48 +74,84 @@ 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)
|
||||
prevRef, ok := prevCommits[tag.Name().String()]
|
||||
log.Debugf("Processing tag %v from repo %v (id: %v)", tag.Name().String(), pipeline.Name, pipeline.Id)
|
||||
prevRef, ok := prevRefs[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())
|
||||
prevCommits[tag.Name().String()] = 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())
|
||||
prevRefs[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())
|
||||
prevCommits[tag.Name().String()] = 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())
|
||||
prevRefs[tag.Name().String()] = tag.Hash().String()
|
||||
refsToRunFor = append(refsToRunFor, tag.Name().String())
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
for _, ref := range refsToRunFor {
|
||||
log.Debugf("Dispatching job for ref %v in repo %v", ref, repoName)
|
||||
err = db.UpdatePipelineRefs(pipeline.Id, prevRefs)
|
||||
if err != nil {
|
||||
log.Errorf("Could not update pipeline refs: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
pe := pipeline_executor.PipelineExecution{
|
||||
Name: repoName,
|
||||
Job: jobConfig,
|
||||
Ref: ref,
|
||||
for _, ref := range refsToRunFor {
|
||||
log.Debugf("Dispatching job for ref %v in repo %v (id: %v)", ref, pipeline.Name, pipeline.Id)
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"google.golang.org/protobuf/reflect/protoreflect"
|
||||
"nhooyr.io/websocket"
|
||||
@@ -17,14 +18,14 @@ type RunnerData struct {
|
||||
}
|
||||
|
||||
type Runner struct {
|
||||
id string
|
||||
id uuid.UUID
|
||||
tags []string
|
||||
conn *websocket.Conn
|
||||
receiveChan chan []byte
|
||||
running bool
|
||||
}
|
||||
|
||||
func (r *Runner) Id() string {
|
||||
func (r *Runner) Id() uuid.UUID {
|
||||
return r.id
|
||||
}
|
||||
|
||||
|
||||
@@ -6,11 +6,13 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/op/go-logging"
|
||||
"google.golang.org/protobuf/proto"
|
||||
"nhooyr.io/websocket"
|
||||
|
||||
"git.ohea.xyz/cursorius/server/config"
|
||||
"git.ohea.xyz/cursorius/server/database"
|
||||
|
||||
runner_api "git.ohea.xyz/cursorius/runner-api/go/api/v2"
|
||||
)
|
||||
@@ -30,6 +32,7 @@ type runnerManager struct {
|
||||
connectedRunners []Runner
|
||||
numConnectedRunners uint64
|
||||
configuredRunners map[string]config.Runner
|
||||
db database.Database
|
||||
}
|
||||
|
||||
type GetRunnerRequest struct {
|
||||
@@ -49,11 +52,13 @@ type runnerJob struct {
|
||||
|
||||
func (r *runnerManager) processRequest(req GetRunnerRequest) {
|
||||
var runnerTagsStr strings.Builder
|
||||
fmt.Fprintf(&runnerTagsStr, "[%v", req.Tags[0])
|
||||
for _, tag := range req.Tags[1:] {
|
||||
fmt.Fprintf(&runnerTagsStr, ", %v", tag)
|
||||
if len(req.Tags) > 0 {
|
||||
fmt.Fprintf(&runnerTagsStr, "[%v", req.Tags[0])
|
||||
for _, tag := range req.Tags[1:] {
|
||||
fmt.Fprintf(&runnerTagsStr, ", %v", tag)
|
||||
}
|
||||
fmt.Fprintf(&runnerTagsStr, "]")
|
||||
}
|
||||
fmt.Fprintf(&runnerTagsStr, "]")
|
||||
log.Infof("Got request for runner with tags \"%v\"", runnerTagsStr.String())
|
||||
|
||||
log.Debugf("Finding runner with tags %v", runnerTagsStr.String())
|
||||
@@ -126,50 +131,60 @@ runnerIter:
|
||||
|
||||
func (r *runnerManager) processRegistration(reg RunnerRegistration) {
|
||||
log.Debugf("New runner appeared with id: %v and secret: %v", reg.Id, reg.Secret)
|
||||
if configuredRunner, doesExist := r.configuredRunners[reg.Id]; doesExist {
|
||||
if configuredRunner.Secret == reg.Secret {
|
||||
log.Infof("Registering runner \"%v\" with tags %v", reg.Id, reg.Tags)
|
||||
runner := Runner{
|
||||
id: reg.Id,
|
||||
tags: reg.Tags,
|
||||
conn: reg.conn,
|
||||
receiveChan: make(chan []byte),
|
||||
running: false,
|
||||
}
|
||||
r.connectedRunners = append(r.connectedRunners, runner)
|
||||
// start goroutine to call Read function on websocket connection
|
||||
// this is required to keep the connection functioning
|
||||
go func() {
|
||||
defer log.Noticef("Deregistered runner with id: %v", runner.id)
|
||||
defer close(runner.receiveChan)
|
||||
for {
|
||||
msgType, data, err := reg.conn.Read(context.Background())
|
||||
if err != nil {
|
||||
// TODO: this is still racy, since a runner could be allocated between the
|
||||
// connection returning an err and the channel closing
|
||||
// This should probably be handled by sending erroring, but not 100% sure
|
||||
log.Errorf("Could not read from connection: %v", err)
|
||||
return
|
||||
}
|
||||
if msgType != websocket.MessageBinary {
|
||||
close(runner.receiveChan)
|
||||
log.Errorf("Got binary data from connection")
|
||||
return
|
||||
}
|
||||
|
||||
runner.receiveChan <- data
|
||||
|
||||
}
|
||||
}()
|
||||
|
||||
} else {
|
||||
log.Errorf("Disconnecting runner with id: %v and invalid secret: %v", reg.Id, reg.Secret)
|
||||
reg.conn.Close(websocket.StatusNormalClosure, "registration invalid")
|
||||
}
|
||||
} else {
|
||||
log.Errorf("Disconnecting runner with invalid id: %v", reg.Id)
|
||||
// Get runner with give id from database
|
||||
runnerId, err := uuid.Parse(reg.Id)
|
||||
if err != nil {
|
||||
log.Errorf("Disconnecting runner with id: %v, could not parse as UUID: %v", reg.Id, err)
|
||||
reg.conn.Close(websocket.StatusNormalClosure, "registration invalid")
|
||||
return
|
||||
}
|
||||
dbRunner, err := r.db.GetRunnerById(runnerId)
|
||||
if err != nil {
|
||||
log.Errorf("Disconnecting runner with id: %v, could not find runner in DB: %v", runnerId, err)
|
||||
reg.conn.Close(websocket.StatusNormalClosure, "registration invalid")
|
||||
return
|
||||
}
|
||||
|
||||
if reg.Secret != dbRunner.Token {
|
||||
log.Errorf("Disconnecting runner with id: %v, invalid secret", runnerId)
|
||||
reg.conn.Close(websocket.StatusNormalClosure, "registration invalid")
|
||||
return
|
||||
}
|
||||
|
||||
log.Infof("Registering runner \"%v\" with tags %v", reg.Id, reg.Tags)
|
||||
runner := Runner{
|
||||
id: runnerId,
|
||||
tags: reg.Tags,
|
||||
conn: reg.conn,
|
||||
receiveChan: make(chan []byte),
|
||||
running: false,
|
||||
}
|
||||
r.connectedRunners = append(r.connectedRunners, runner)
|
||||
// start goroutine to call Read function on websocket connection
|
||||
// this is required to keep the connection functioning
|
||||
go func() {
|
||||
defer log.Noticef("Deregistered runner with id: %v", runner.id)
|
||||
defer close(runner.receiveChan)
|
||||
for {
|
||||
msgType, data, err := reg.conn.Read(context.Background())
|
||||
if err != nil {
|
||||
// TODO: this is still racy, since a runner could be allocated between the
|
||||
// connection returning an err and the channel closing
|
||||
// This should probably be handled by sending erroring, but not 100% sure
|
||||
log.Errorf("Could not read from connection: %v", err)
|
||||
return
|
||||
}
|
||||
if msgType != websocket.MessageBinary {
|
||||
close(runner.receiveChan)
|
||||
log.Errorf("Got binary data from connection")
|
||||
return
|
||||
}
|
||||
|
||||
runner.receiveChan <- data
|
||||
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func runRunnerManager(r runnerManager) {
|
||||
@@ -184,12 +199,13 @@ func runRunnerManager(r runnerManager) {
|
||||
}
|
||||
}
|
||||
|
||||
func StartRunnerManager(configuredRunners map[string]config.Runner) (chan GetRunnerRequest, chan RunnerRegistration, error) {
|
||||
func StartRunnerManager(configuredRunners map[string]config.Runner, db database.Database) (chan GetRunnerRequest, chan RunnerRegistration, error) {
|
||||
scheduler := runnerManager{
|
||||
getRunnerCh: make(chan GetRunnerRequest),
|
||||
registerCh: make(chan RunnerRegistration),
|
||||
connectedRunners: make([]Runner, 0),
|
||||
configuredRunners: configuredRunners,
|
||||
db: db,
|
||||
}
|
||||
|
||||
go runRunnerManager(scheduler)
|
||||
|
||||
+75
-57
@@ -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)
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user