github/drone
5
0
Fork 0
mirror of https://github.com/harness/drone.git synced 2025-05-19 02:20:03 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1981 from bradrydzewski/feature/steps

Browse Source 
Store build steps and logs as structured data [breaking change]
This commit is contained in:
Brad Rydzewski 2017-04-05 18:31:46 +09:00 committed by GitHub
commit e2e88ab52f
44 changed files with 2244 additions and 641 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -1,6 +1,9 @@
drone/drone drone/drone
*.sqlite *.sqlite
*_gen.go *_gen.go
!store/datastore/sql/sqlite/sql_gen.go
!store/datastore/sql/mysql/sql_gen.go
!store/datastore/sql/postgres/sql_gen.go
#*.css #*.css
*.txt *.txt
*.zip *.zip

94
drone/agent/agent.go
View File

@ -2,10 +2,13 @@ package agent
import ( import (
"context" "context"
"encoding/json"
"io" "io"
"io/ioutil"
"log" "log"
"math" "math"
"net/url" "net/url"
"strconv"
"sync" "sync"
"time" "time"
@ -189,9 +192,9 @@ func run(ctx context.Context, client rpc.Peer, filter rpc.Filter) error {
state := rpc.State{} state := rpc.State{}
state.Started = time.Now().Unix() state.Started = time.Now().Unix()
err = client.Update(context.Background(), work.ID, state) err = client.Init(context.Background(), work.ID, state)
if err != nil { if err != nil {
log.Printf("pipeline: error updating pipeline status: %s: %s", work.ID, err) log.Printf("pipeline: error signaling pipeline init: %s: %s", work.ID, err)
} }
var uploads sync.WaitGroup var uploads sync.WaitGroup
@ -201,9 +204,31 @@ func run(ctx context.Context, client rpc.Peer, filter rpc.Filter) error {
return rerr return rerr
} }
uploads.Add(1) uploads.Add(1)
writer := rpc.NewLineWriter(client, work.ID, proc.Alias)
rlimit := io.LimitReader(part, maxLogsUpload) var secrets []string
io.Copy(writer, rlimit) for _, secret := range work.Config.Secrets {
if secret.Mask {
secrets = append(secrets, secret.Value)
}
}
limitedPart := io.LimitReader(part, maxLogsUpload)
logstream := rpc.NewLineWriter(client, work.ID, proc.Alias, secrets...)
io.Copy(logstream, limitedPart)
file := &rpc.File{}
file.Mime = "application/json+logs"
file.Proc = proc.Alias
file.Name = "logs.json"
file.Data, _ = json.Marshal(logstream.Lines())
file.Size = len(file.Data)
file.Time = time.Now().Unix()
if serr := client.Upload(context.Background(), work.ID, file); serr != nil {
log.Printf("pipeline: cannot upload logs: %s: %s: %s", work.ID, file.Mime, serr)
} else {
log.Printf("pipeline: finish uploading logs: %s: step %s: %s", file.Mime, work.ID, proc.Alias)
}
defer func() { defer func() {
log.Printf("pipeline: finish uploading logs: %s: step %s", work.ID, proc.Alias) log.Printf("pipeline: finish uploading logs: %s: step %s", work.ID, proc.Alias)
@ -214,10 +239,54 @@ func run(ctx context.Context, client rpc.Peer, filter rpc.Filter) error {
if rerr != nil { if rerr != nil {
return nil return nil
} }
rlimit = io.LimitReader(part, maxFileUpload) // TODO should be configurable
mime := part.Header().Get("Content-Type") limitedPart = io.LimitReader(part, maxFileUpload)
if serr := client.Upload(context.Background(), work.ID, mime, rlimit); serr != nil { file = &rpc.File{}
log.Printf("pipeline: cannot upload artifact: %s: %s: %s", work.ID, mime, serr) file.Mime = part.Header().Get("Content-Type")
file.Proc = proc.Alias
file.Name = part.FileName()
file.Data, _ = ioutil.ReadAll(limitedPart)
file.Size = len(file.Data)
file.Time = time.Now().Unix()
if serr := client.Upload(context.Background(), work.ID, file); serr != nil {
log.Printf("pipeline: cannot upload artifact: %s: %s: %s", work.ID, file.Mime, serr)
} else {
log.Printf("pipeline: finish uploading artifact: %s: step %s: %s", file.Mime, work.ID, proc.Alias)
}
return nil
})
defaultTracer := pipeline.TraceFunc(func(state *pipeline.State) error {
procState := rpc.State{
Proc: state.Pipeline.Step.Alias,
Exited: state.Process.Exited,
ExitCode: state.Process.ExitCode,
Started: time.Now().Unix(), // TODO do not do this
Finished: time.Now().Unix(),
}
defer func() {
if uerr := client.Update(context.Background(), work.ID, procState); uerr != nil {
log.Printf("Pipeine: error updating pipeline step status: %s: %s: %s", work.ID, procState.Proc, uerr)
}
}()
if state.Process.Exited {
return nil
}
if state.Pipeline.Step.Environment == nil {
state.Pipeline.Step.Environment = map[string]string{}
}
state.Pipeline.Step.Environment["CI_BUILD_STATUS"] = "success"
state.Pipeline.Step.Environment["CI_BUILD_STARTED"] = strconv.FormatInt(state.Pipeline.Time, 10)
state.Pipeline.Step.Environment["CI_BUILD_FINISHED"] = strconv.FormatInt(time.Now().Unix(), 10)
state.Pipeline.Step.Environment["CI_JOB_STATUS"] = "success"
state.Pipeline.Step.Environment["CI_JOB_STARTED"] = strconv.FormatInt(state.Pipeline.Time, 10)
state.Pipeline.Step.Environment["CI_JOB_FINISHED"] = strconv.FormatInt(time.Now().Unix(), 10)
if state.Pipeline.Error != nil {
state.Pipeline.Step.Environment["CI_BUILD_STATUS"] = "failure"
state.Pipeline.Step.Environment["CI_JOB_STATUS"] = "failure"
} }
return nil return nil
}) })
@ -225,7 +294,7 @@ func run(ctx context.Context, client rpc.Peer, filter rpc.Filter) error {
err = pipeline.New(work.Config, err = pipeline.New(work.Config,
pipeline.WithContext(ctx), pipeline.WithContext(ctx),
pipeline.WithLogger(defaultLogger), pipeline.WithLogger(defaultLogger),
pipeline.WithTracer(pipeline.DefaultTracer), pipeline.WithTracer(defaultTracer),
pipeline.WithEngine(engine), pipeline.WithEngine(engine),
).Run() ).Run()
@ -247,9 +316,10 @@ func run(ctx context.Context, client rpc.Peer, filter rpc.Filter) error {
log.Printf("pipeline: execution complete: %s", work.ID) log.Printf("pipeline: execution complete: %s", work.ID)
uploads.Wait() uploads.Wait()
err = client.Update(context.Background(), work.ID, state)
err = client.Done(context.Background(), work.ID, state)
if err != nil { if err != nil {
log.Printf("Pipeine: error updating pipeline status: %s: %s", work.ID, err) log.Printf("Pipeine: error signaling pipeline done: %s: %s", work.ID, err)
} }
return nil return nil

3
model/build.go
View File

@ -31,7 +31,8 @@ type Build struct {
Verified bool `json:"verified" meddler:"build_verified"` // deprecate Verified bool `json:"verified" meddler:"build_verified"` // deprecate
Reviewer string `json:"reviewed_by" meddler:"build_reviewer"` Reviewer string `json:"reviewed_by" meddler:"build_reviewer"`
Reviewed int64 `json:"reviewed_at" meddler:"build_reviewed"` Reviewed int64 `json:"reviewed_at" meddler:"build_reviewed"`
Jobs []*Job `json:"jobs,omitempty" meddler:"-"` // Jobs []*Job `json:"jobs,omitempty" meddler:"-"`
Procs []*Proc `json:"procs,omitempty" meddler:"-"`
} }
type BuildGroup struct { type BuildGroup struct {

21
model/event.go
View File

@ -15,24 +15,5 @@ type Event struct {
Type EventType `json:"type"` Type EventType `json:"type"`
Repo Repo `json:"repo"` Repo Repo `json:"repo"`
Build Build `json:"build"` Build Build `json:"build"`
Job Job `json:"job"` Proc Proc `json:"proc"`
}
// NewEvent creates a new Event for the build, using copies of
// the build data to avoid possible mutation or race conditions.
func NewEvent(t EventType, r *Repo, b *Build, j *Job) *Event {
return &Event{
Type: t,
Repo: *r,
Build: *b,
Job: *j,
}
}
func NewBuildEvent(t EventType, r *Repo, b *Build) *Event {
return &Event{
Type: t,
Repo: *r,
Build: *b,
}
} }

23
model/file.go Normal file
View File

@ -0,0 +1,23 @@
package model
import "io"
// FileStore persists pipeline artifacts to storage.
type FileStore interface {
FileList(*Build) ([]*File, error)
FileFind(*Proc, string) (*File, error)
FileRead(*Proc, string) (io.ReadCloser, error)
FileCreate(*File, io.Reader) error
}
// File represents a pipeline artifact.
type File struct {
ID int64 `json:"id" meddler:"file_id,pk"`
BuildID int64 `json:"build_id" meddler:"file_build_id"`
ProcID int64 `json:"proc_id" meddler:"file_proc_id"`
Name string `json:"name" meddler:"file_name"`
Size int `json:"size" meddler:"file_size"`
Mime string `json:"mime" meddler:"file_mime"`
Time int64 `json:"time" meddler:"file_time"`
// Data []byte `json:"data" meddler:"file_data"`
}

30
model/job.go
View File

@ -1,17 +1,17 @@
package model package model
// swagger:model job // // swagger:model job
type Job struct { // type Job struct {
ID int64 `json:"id" meddler:"job_id,pk"` // ID int64 `json:"id" meddler:"job_id,pk"`
BuildID int64 `json:"-" meddler:"job_build_id"` // BuildID int64 `json:"-" meddler:"job_build_id"`
NodeID int64 `json:"-" meddler:"job_node_id"` // NodeID int64 `json:"-" meddler:"job_node_id"`
Number int `json:"number" meddler:"job_number"` // Number int `json:"number" meddler:"job_number"`
Error string `json:"error" meddler:"job_error"` // Error string `json:"error" meddler:"job_error"`
Status string `json:"status" meddler:"job_status"` // Status string `json:"status" meddler:"job_status"`
ExitCode int `json:"exit_code" meddler:"job_exit_code"` // ExitCode int `json:"exit_code" meddler:"job_exit_code"`
Enqueued int64 `json:"enqueued_at" meddler:"job_enqueued"` // Enqueued int64 `json:"enqueued_at" meddler:"job_enqueued"`
Started int64 `json:"started_at" meddler:"job_started"` // Started int64 `json:"started_at" meddler:"job_started"`
Finished int64 `json:"finished_at" meddler:"job_finished"` // Finished int64 `json:"finished_at" meddler:"job_finished"`
//
Environment map[string]string `json:"environment" meddler:"job_environment,json"` // Environment map[string]string `json:"environment" meddler:"job_environment,json"`
} // }

8
model/key.go
View File

@ -1,8 +0,0 @@
package model
type Key struct {
ID int64 `json:"-" meddler:"key_id,pk"`
RepoID int64 `json:"-" meddler:"key_repo_id"`
Public string `json:"public" meddler:"key_public"`
Private string `json:"private" meddler:"key_private"`
}

7
model/log.go
View File

@ -1,7 +0,0 @@
package model
type Log struct {
ID int64 `meddler:"log_id,pk"`
JobID int64 `meddler:"log_job_id"`
Data []byte `meddler:"log_data"`
}

60
model/proc.go Normal file
View File

@ -0,0 +1,60 @@
package model
// ProcStore persists process information to storage.
type ProcStore interface {
ProcLoad(int64) (*Proc, error)
ProcFind(*Build, int) (*Proc, error)
ProcChild(*Build, int, string) (*Proc, error)
ProcList(*Build) ([]*Proc, error)
ProcCreate([]*Proc) error
ProcUpdate(*Proc) error
ProcClear(*Build) error
}
// Proc represents a process in the build pipeline.
// swagger:model proc
type Proc struct {
ID int64 `json:"id" meddler:"proc_id,pk"`
BuildID int64 `json:"build_id" meddler:"proc_build_id"`
PID int `json:"pid" meddler:"proc_pid"`
PPID int `json:"ppid" meddler:"proc_ppid"`
PGID int `json:"pgid" meddler:"proc_pgid"`
Name string `json:"name" meddler:"proc_name"`
State string `json:"state" meddler:"proc_state"`
Error string `json:"error,omitempty" meddler:"proc_error"`
ExitCode int `json:"exit_code" meddler:"proc_exit_code"`
Started int64 `json:"start_time,omitempty" meddler:"proc_started"`
Stopped int64 `json:"end_time,omitempty" meddler:"proc_stopped"`
Machine string `json:"machine,omitempty" meddler:"proc_machine"`
Platform string `json:"platform,omitempty" meddler:"proc_platform"`
Environ map[string]string `json:"environ,omitempty" meddler:"proc_environ,json"`
Children []*Proc `json:"children,omitempty" meddler:"-"`
}
// Running returns true if the process state is pending or running.
func (p *Proc) Running() bool {
return p.State == StatusPending || p.State == StatusRunning
}
// Failing returns true if the process state is failed, killed or error.
func (p *Proc) Failing() bool {
return p.State == StatusError || p.State == StatusKilled || p.State == StatusFailure
}
// Tree creates a process tree from a flat process list.
func Tree(procs []*Proc) []*Proc {
var (
nodes []*Proc
parent *Proc
)
for _, proc := range procs {
if proc.PPID == 0 {
nodes = append(nodes, proc)
parent = proc
continue
} else {
parent.Children = append(parent.Children, proc)
}
}
return nodes
}

19
model/work.go
View File

@ -1,19 +0,0 @@
package model
// Work represents an item for work to be
// processed by a worker.
type Work struct {
Signed bool `json:"signed"`
Verified bool `json:"verified"`
Yaml string `json:"config"`
YamlEnc string `json:"secret"`
Repo *Repo `json:"repo"`
Build *Build `json:"build"`
BuildLast *Build `json:"build_last"`
Job *Job `json:"job"`
Netrc *Netrc `json:"netrc"`
Keys *Key `json:"keys"`
System *System `json:"system"`
Secrets []*Secret `json:"secrets"`
User *User `json:"user"`
}

2
router/router.go
View File

@ -104,7 +104,7 @@ func Load(middleware ...gin.HandlerFunc) http.Handler {
repo.GET("", server.GetRepo) repo.GET("", server.GetRepo)
repo.GET("/builds", server.GetBuilds) repo.GET("/builds", server.GetBuilds)
repo.GET("/builds/:number", server.GetBuild) repo.GET("/builds/:number", server.GetBuild)
repo.GET("/logs/:number/:job", server.GetBuildLogs) repo.GET("/logs/:number/:ppid/:proc", server.GetBuildLogs)
repo.POST("/sign", session.MustPush, server.Sign) repo.POST("/sign", session.MustPush, server.Sign)
repo.GET("/secrets", session.MustPush, server.GetSecrets) repo.GET("/secrets", session.MustPush, server.GetSecrets)

228
server/build.go
View File

@ -1,7 +1,6 @@
package server package server
import ( import (
"bufio"
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
@ -51,14 +50,10 @@ func GetBuild(c *gin.Context) {
c.AbortWithError(http.StatusInternalServerError, err) c.AbortWithError(http.StatusInternalServerError, err)
return return
} }
jobs, _ := store.GetJobList(c, build) procs, _ := store.FromContext(c).ProcList(build)
build.Procs = model.Tree(procs)
out := struct { c.JSON(http.StatusOK, build)
*model.Build
Jobs []*model.Job `json:"jobs"`
}{build, jobs}
c.JSON(http.StatusOK, &out)
} }
func GetBuildLast(c *gin.Context) { func GetBuildLast(c *gin.Context) {
@ -70,27 +65,20 @@ func GetBuildLast(c *gin.Context) {
c.String(http.StatusInternalServerError, err.Error()) c.String(http.StatusInternalServerError, err.Error())
return return
} }
jobs, _ := store.GetJobList(c, build)
out := struct { procs, _ := store.FromContext(c).ProcList(build)
*model.Build build.Procs = model.Tree(procs)
Jobs []*model.Job `json:"jobs"` c.JSON(http.StatusOK, build)
}{build, jobs}
c.JSON(http.StatusOK, &out)
} }
func GetBuildLogs(c *gin.Context) { func GetBuildLogs(c *gin.Context) {
repo := session.Repo(c) repo := session.Repo(c)
// the user may specify to stream the full logs,
// or partial logs, capped at 2MB.
full, _ := strconv.ParseBool(c.DefaultQuery("full", "false"))
// parse the build number and job sequence number from // parse the build number and job sequence number from
// the repquest parameter. // the repquest parameter.
num, _ := strconv.Atoi(c.Params.ByName("number")) num, _ := strconv.Atoi(c.Params.ByName("number"))
seq, _ := strconv.Atoi(c.Params.ByName("job")) ppid, _ := strconv.Atoi(c.Params.ByName("ppid"))
name := c.Params.ByName("proc")
build, err := store.GetBuildNumber(c, repo, num) build, err := store.GetBuildNumber(c, repo, num)
if err != nil { if err != nil {
@ -98,25 +86,22 @@ func GetBuildLogs(c *gin.Context) {
return return
} }
job, err := store.GetJobNumber(c, build, seq) proc, err := store.FromContext(c).ProcChild(build, ppid, name)
if err != nil { if err != nil {
c.AbortWithError(404, err) c.AbortWithError(404, err)
return return
} }
r, err := store.ReadLog(c, job) rc, err := store.FromContext(c).LogFind(proc)
if err != nil { if err != nil {
c.AbortWithError(404, err) c.AbortWithError(404, err)
return return
} }
defer r.Close() defer rc.Close()
if full {
// TODO implement limited streaming to avoid crashing the browser
}
c.Header("Content-Type", "application/json") c.Header("Content-Type", "application/json")
copyLogs(c.Writer, r) io.Copy(c.Writer, rc)
} }
func DeleteBuild(c *gin.Context) { func DeleteBuild(c *gin.Context) {
@ -133,26 +118,27 @@ func DeleteBuild(c *gin.Context) {
return return
} }
job, err := store.GetJobNumber(c, build, seq) proc, err := store.FromContext(c).ProcFind(build, seq)
if err != nil { if err != nil {
c.AbortWithError(404, err) c.AbortWithError(404, err)
return return
} }
if job.Status != model.StatusRunning { if proc.State != model.StatusRunning {
c.String(400, "Cannot cancel a non-running build") c.String(400, "Cannot cancel a non-running build")
return return
} }
job.Status = model.StatusKilled proc.State = model.StatusKilled
job.Finished = time.Now().Unix() proc.Stopped = time.Now().Unix()
if job.Started == 0 { if proc.Started == 0 {
job.Started = job.Finished proc.Started = proc.Stopped
} }
job.ExitCode = 137 proc.ExitCode = 137
store.UpdateBuildJob(c, build, job) // TODO cancel child procs
store.FromContext(c).ProcUpdate(proc)
config.queue.Error(context.Background(), fmt.Sprint(job.ID), queue.ErrCancel) config.queue.Error(context.Background(), fmt.Sprint(proc.ID), queue.ErrCancel)
c.String(204, "") c.String(204, "")
} }
@ -243,15 +229,37 @@ func PostApproval(c *gin.Context) {
return return
} }
var pcounter = len(items)
for _, item := range items { for _, item := range items {
build.Jobs = append(build.Jobs, item.Job) build.Procs = append(build.Procs, item.Proc)
store.CreateJob(c, item.Job) item.Proc.BuildID = build.ID
// TODO err
for _, stage := range item.Config.Stages {
var gid int
for _, step := range stage.Steps {
pcounter++
if gid == 0 {
gid = pcounter
}
proc := &model.Proc{
BuildID: build.ID,
Name: step.Alias,
PID: pcounter,
PPID: item.Proc.PID,
PGID: gid,
State: model.StatusPending,
}
build.Procs = append(build.Procs, proc)
}
}
} }
store.FromContext(c).ProcCreate(build.Procs)
// //
// publish topic // publish topic
// //
buildCopy := *build
buildCopy.Procs = model.Tree(buildCopy.Procs)
message := pubsub.Message{ message := pubsub.Message{
Labels: map[string]string{ Labels: map[string]string{
"repo": repo.FullName, "repo": repo.FullName,
@ -261,17 +269,18 @@ func PostApproval(c *gin.Context) {
message.Data, _ = json.Marshal(model.Event{ message.Data, _ = json.Marshal(model.Event{
Type: model.Enqueued, Type: model.Enqueued,
Repo: *repo, Repo: *repo,
Build: *build, Build: buildCopy,
}) })
// TODO remove global reference // TODO remove global reference
config.pubsub.Publish(c, "topic/events", message) config.pubsub.Publish(c, "topic/events", message)
// //
// end publish topic // end publish topic
// //
for _, item := range items { for _, item := range items {
task := new(queue.Task) task := new(queue.Task)
task.ID = fmt.Sprint(item.Job.ID) task.ID = fmt.Sprint(item.Proc.ID)
task.Labels = map[string]string{} task.Labels = map[string]string{}
task.Labels["platform"] = item.Platform task.Labels["platform"] = item.Platform
for k, v := range item.Labels { for k, v := range item.Labels {
@ -279,7 +288,7 @@ func PostApproval(c *gin.Context) {
} }
task.Data, _ = json.Marshal(rpc.Pipeline{ task.Data, _ = json.Marshal(rpc.Pipeline{
ID: fmt.Sprint(item.Job.ID), ID: fmt.Sprint(item.Proc.ID),
Config: item.Config, Config: item.Config,
Timeout: b.Repo.Timeout, Timeout: b.Repo.Timeout,
}) })
@ -336,23 +345,6 @@ func GetBuildQueue(c *gin.Context) {
c.JSON(200, out) c.JSON(200, out)
} }
// copyLogs copies the stream from the source to the destination in valid JSON
// format. This converts the logs, which are per-line JSON objects, to a
// proper JSON array.
func copyLogs(dest io.Writer, src io.Reader) error {
io.WriteString(dest, "[")
scanner := bufio.NewScanner(src)
for scanner.Scan() {
io.WriteString(dest, scanner.Text())
io.WriteString(dest, ",\n")
}
io.WriteString(dest, "{}]")
return nil
}
// //
// //
// //
@ -411,13 +403,6 @@ func PostBuild(c *gin.Context) {
return return
} }
jobs, err := store.GetJobList(c, build)
if err != nil {
logrus.Errorf("failure to get build %d jobs. %s", build.Number, err)
c.AbortWithError(404, err)
return
}
// must not restart a running build // must not restart a running build
if build.Status == model.StatusPending || build.Status == model.StatusRunning { if build.Status == model.StatusPending || build.Status == model.StatusRunning {
c.String(409, "Cannot re-start a started build") c.String(409, "Cannot re-start a started build")
@ -430,11 +415,12 @@ func PostBuild(c *gin.Context) {
build.ID = 0 build.ID = 0
build.Number = 0 build.Number = 0
build.Parent = num build.Parent = num
for _, job := range jobs { build.Status = model.StatusPending
job.ID = 0 build.Started = 0
job.NodeID = 0 build.Finished = 0
} build.Enqueued = time.Now().UTC().Unix()
err := store.CreateBuild(c, build, jobs...) build.Error = ""
err = store.CreateBuild(c, build)
if err != nil { if err != nil {
c.String(500, err.Error()) c.String(500, err.Error())
return return
@ -448,6 +434,26 @@ func PostBuild(c *gin.Context) {
build.Event = event build.Event = event
} }
build.Deploy = c.DefaultQuery("deploy_to", build.Deploy) build.Deploy = c.DefaultQuery("deploy_to", build.Deploy)
} else {
// todo move this to database tier
// and wrap inside a transaction
build.Status = model.StatusPending
build.Started = 0
build.Finished = 0
build.Enqueued = time.Now().UTC().Unix()
build.Error = ""
err = store.FromContext(c).ProcClear(build)
if err != nil {
c.AbortWithStatus(500)
return
}
err = store.UpdateBuild(c, build)
if err != nil {
c.AbortWithStatus(500)
return
}
} }
// Read query string parameters into buildParams, exclude reserved params // Read query string parameters into buildParams, exclude reserved params
@ -462,35 +468,6 @@ func PostBuild(c *gin.Context) {
} }
} }
// todo move this to database tier
// and wrap inside a transaction
build.Status = model.StatusPending
build.Started = 0
build.Finished = 0
build.Enqueued = time.Now().UTC().Unix()
build.Error = ""
for _, job := range jobs {
for k, v := range buildParams {
job.Environment[k] = v
}
job.Error = ""
job.Status = model.StatusPending
job.Started = 0
job.Finished = 0
job.ExitCode = 0
job.NodeID = 0
job.Enqueued = build.Enqueued
store.UpdateJob(c, job)
}
err = store.UpdateBuild(c, build)
if err != nil {
c.AbortWithStatus(500)
return
}
c.JSON(202, build)
// get the previous build so that we can send // get the previous build so that we can send
// on status change notifications // on status change notifications
last, _ := store.GetBuildLastBefore(c, repo, build.Branch, build.ID) last, _ := store.GetBuildLastBefore(c, repo, build.Branch, build.ID)
@ -508,25 +485,60 @@ func PostBuild(c *gin.Context) {
Link: httputil.GetURL(c.Request), Link: httputil.GetURL(c.Request),
Yaml: string(raw), Yaml: string(raw),
} }
// TODO inject environment varibles !!!!!! buildParams
items, err := b.Build() items, err := b.Build()
if err != nil { if err != nil {
build.Status = model.StatusError build.Status = model.StatusError
build.Started = time.Now().Unix() build.Started = time.Now().Unix()
build.Finished = build.Started build.Finished = build.Started
build.Error = err.Error() build.Error = err.Error()
c.JSON(500, build)
return return
} }
for i, item := range items { var pcounter = len(items)
// TODO prevent possible index out of bounds for _, item := range items {
item.Job.ID = jobs[i].ID build.Procs = append(build.Procs, item.Proc)
build.Jobs = append(build.Jobs, item.Job) item.Proc.BuildID = build.ID
store.UpdateJob(c, item.Job)
for _, stage := range item.Config.Stages {
var gid int
for _, step := range stage.Steps {
pcounter++
if gid == 0 {
gid = pcounter
}
proc := &model.Proc{
BuildID: build.ID,
Name: step.Alias,
PID: pcounter,
PPID: item.Proc.PID,
PGID: gid,
State: model.StatusPending,
}
build.Procs = append(build.Procs, proc)
}
}
} }
err = store.FromContext(c).ProcCreate(build.Procs)
if err != nil {
logrus.Errorf("cannot restart %s#%d: %s", repo.FullName, build.Number, err)
build.Status = model.StatusError
build.Started = time.Now().Unix()
build.Finished = build.Started
build.Error = err.Error()
c.JSON(500, build)
return
}
c.JSON(202, build)
// //
// publish topic // publish topic
// //
buildCopy := *build
buildCopy.Procs = model.Tree(buildCopy.Procs)
message := pubsub.Message{ message := pubsub.Message{
Labels: map[string]string{ Labels: map[string]string{
"repo": repo.FullName, "repo": repo.FullName,
@ -536,7 +548,7 @@ func PostBuild(c *gin.Context) {
message.Data, _ = json.Marshal(model.Event{ message.Data, _ = json.Marshal(model.Event{
Type: model.Enqueued, Type: model.Enqueued,
Repo: *repo, Repo: *repo,
Build: *build, Build: buildCopy,
}) })
// TODO remove global reference // TODO remove global reference
config.pubsub.Publish(c, "topic/events", message) config.pubsub.Publish(c, "topic/events", message)
@ -546,7 +558,7 @@ func PostBuild(c *gin.Context) {
for _, item := range items { for _, item := range items {
task := new(queue.Task) task := new(queue.Task)
task.ID = fmt.Sprint(item.Job.ID) task.ID = fmt.Sprint(item.Proc.ID)
task.Labels = map[string]string{} task.Labels = map[string]string{}
task.Labels["platform"] = item.Platform task.Labels["platform"] = item.Platform
for k, v := range item.Labels { for k, v := range item.Labels {
@ -554,7 +566,7 @@ func PostBuild(c *gin.Context) {
} }
task.Data, _ = json.Marshal(rpc.Pipeline{ task.Data, _ = json.Marshal(rpc.Pipeline{
ID: fmt.Sprint(item.Job.ID), ID: fmt.Sprint(item.Proc.ID),
Config: item.Config, Config: item.Config,
Timeout: b.Repo.Timeout, Timeout: b.Repo.Timeout,
}) })

71
server/hook.go
View File

@ -123,7 +123,7 @@ func PostHook(c *gin.Context) {
// if the remote has a refresh token, the current access token // if the remote has a refresh token, the current access token
// may be stale. Therefore, we should refresh prior to dispatching // may be stale. Therefore, we should refresh prior to dispatching
// the job. // the build.
if refresher, ok := remote_.(remote.Refresher); ok { if refresher, ok := remote_.(remote.Refresher); ok {
ok, _ := refresher.Refresh(user) ok, _ := refresher.Refresh(user)
if ok { if ok {
@ -221,7 +221,7 @@ func PostHook(c *gin.Context) {
build.RepoID = repo.ID build.RepoID = repo.ID
build.Verified = true build.Verified = true
if err := store.CreateBuild(c, build, build.Jobs...); err != nil { if err := store.CreateBuild(c, build, build.Procs...); err != nil {
logrus.Errorf("failure to save commit for %s. %s", repo.FullName, err) logrus.Errorf("failure to save commit for %s. %s", repo.FullName, err)
c.AbortWithError(500, err) c.AbortWithError(500, err)
return return
@ -268,10 +268,34 @@ func PostHook(c *gin.Context) {
return return
} }
var pcounter = len(items)
for _, item := range items { for _, item := range items {
build.Jobs = append(build.Jobs, item.Job) build.Procs = append(build.Procs, item.Proc)
store.CreateJob(c, item.Job) item.Proc.BuildID = build.ID
// TODO err
for _, stage := range item.Config.Stages {
var gid int
for _, step := range stage.Steps {
pcounter++
if gid == 0 {
gid = pcounter
}
proc := &model.Proc{
BuildID: build.ID,
Name: step.Alias,
PID: pcounter,
PPID: item.Proc.PID,
PGID: gid,
State: model.StatusPending,
}
build.Procs = append(build.Procs, proc)
}
}
}
err = store.FromContext(c).ProcCreate(build.Procs)
if err != nil {
logrus.Errorf("error persisting procs %s/%d: %s", repo.FullName, build.Number, err)
} }
// //
@ -283,10 +307,12 @@ func PostHook(c *gin.Context) {
"private": strconv.FormatBool(repo.IsPrivate), "private": strconv.FormatBool(repo.IsPrivate),
}, },
} }
buildCopy := *build
buildCopy.Procs = model.Tree(buildCopy.Procs)
message.Data, _ = json.Marshal(model.Event{ message.Data, _ = json.Marshal(model.Event{
Type: model.Enqueued, Type: model.Enqueued,
Repo: *repo, Repo: *repo,
Build: *build, Build: buildCopy,
}) })
// TODO remove global reference // TODO remove global reference
config.pubsub.Publish(c, "topic/events", message) config.pubsub.Publish(c, "topic/events", message)
@ -296,7 +322,7 @@ func PostHook(c *gin.Context) {
for _, item := range items { for _, item := range items {
task := new(queue.Task) task := new(queue.Task)
task.ID = fmt.Sprint(item.Job.ID) task.ID = fmt.Sprint(item.Proc.ID)
task.Labels = map[string]string{} task.Labels = map[string]string{}
task.Labels["platform"] = item.Platform task.Labels["platform"] = item.Platform
for k, v := range item.Labels { for k, v := range item.Labels {
@ -304,7 +330,7 @@ func PostHook(c *gin.Context) {
} }
task.Data, _ = json.Marshal(rpc.Pipeline{ task.Data, _ = json.Marshal(rpc.Pipeline{
ID: fmt.Sprint(item.Job.ID), ID: fmt.Sprint(item.Proc.ID),
Config: item.Config, Config: item.Config,
Timeout: b.Repo.Timeout, Timeout: b.Repo.Timeout,
}) })
@ -315,7 +341,7 @@ func PostHook(c *gin.Context) {
} }
// return the metadata from the cli context. // return the metadata from the cli context.
func metadataFromStruct(repo *model.Repo, build, last *model.Build, job *model.Job, link string) frontend.Metadata { func metadataFromStruct(repo *model.Repo, build, last *model.Build, proc *model.Proc, link string) frontend.Metadata {
return frontend.Metadata{ return frontend.Metadata{
Repo: frontend.Repo{ Repo: frontend.Repo{
Name: repo.Name, Name: repo.Name,
@ -368,8 +394,8 @@ func metadataFromStruct(repo *model.Repo, build, last *model.Build, job *model.J
}, },
}, },
Job: frontend.Job{ Job: frontend.Job{
Number: job.Number, Number: proc.PID,
Matrix: job.Environment, Matrix: proc.Environ,
}, },
Sys: frontend.System{ Sys: frontend.System{
Name: "drone", Name: "drone",
@ -390,7 +416,7 @@ type builder struct {
} }
type buildItem struct { type buildItem struct {
Job *model.Job Proc *model.Proc
Platform string Platform string
Labels map[string]string Labels map[string]string
Config *backend.Config Config *backend.Config
@ -408,15 +434,15 @@ func (b *builder) Build() ([]*buildItem, error) {
var items []*buildItem var items []*buildItem
for i, axis := range axes { for i, axis := range axes {
job := &model.Job{ proc := &model.Proc{
BuildID: b.Curr.ID, BuildID: b.Curr.ID,
Number: i + 1, PID: i + 1,
Status: model.StatusPending, PGID: i + 1,
Environment: axis, State: model.StatusPending,
Enqueued: b.Curr.Created, Environ: axis,
} }
metadata := metadataFromStruct(b.Repo, b.Curr, b.Last, job, b.Link) metadata := metadataFromStruct(b.Repo, b.Curr, b.Last, proc, b.Link)
environ := metadata.Environ() environ := metadata.Environ()
for k, v := range metadata.EnvironDrone() { for k, v := range metadata.EnvironDrone() {
environ[k] = v environ[k] = v
@ -481,15 +507,16 @@ func (b *builder) Build() ([]*buildItem, error) {
compiler.WithPrefix( compiler.WithPrefix(
fmt.Sprintf( fmt.Sprintf(
"%d_%d", "%d_%d",
job.ID, proc.ID,
time.Now().Unix(), time.Now().Unix(),
), ),
), ),
compiler.WithEnviron(job.Environment), compiler.WithEnviron(proc.Environ),
compiler.WithProxy(), compiler.WithProxy(),
// TODO ability to set global volumes for things like certs // TODO ability to set global volumes for things like certs
compiler.WithVolumes(), compiler.WithVolumes(),
compiler.WithWorkspaceFromURL("/drone", b.Curr.Link), compiler.WithWorkspaceFromURL("/drone", b.Curr.Link),
compiler.WithMetadata(metadata),
).Compile(parsed) ).Compile(parsed)
for _, sec := range b.Secs { for _, sec := range b.Secs {
@ -506,7 +533,7 @@ func (b *builder) Build() ([]*buildItem, error) {
} }
item := &buildItem{ item := &buildItem{
Job: job, Proc: proc,
Config: ir, Config: ir,
Labels: parsed.Labels, Labels: parsed.Labels,
Platform: metadata.Sys.Arch, Platform: metadata.Sys.Arch,

296
server/rpc.go
View File

@ -4,13 +4,10 @@ import (
"bytes" "bytes"
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"io"
"log" "log"
"os" "os"
"strconv" "strconv"
"github.com/Sirupsen/logrus"
"github.com/cncd/logging" "github.com/cncd/logging"
"github.com/cncd/pipeline/pipeline/rpc" "github.com/cncd/pipeline/pipeline/rpc"
"github.com/cncd/pubsub" "github.com/cncd/pubsub"
@ -109,20 +106,26 @@ func (s *RPC) Extend(c context.Context, id string) error {
// Update implements the rpc.Update function // Update implements the rpc.Update function
func (s *RPC) Update(c context.Context, id string, state rpc.State) error { func (s *RPC) Update(c context.Context, id string, state rpc.State) error {
jobID, err := strconv.ParseInt(id, 10, 64) procID, err := strconv.ParseInt(id, 10, 64)
if err != nil { if err != nil {
return err return err
} }
job, err := s.store.GetJob(jobID) pproc, err := s.store.ProcLoad(procID)
if err != nil { if err != nil {
log.Printf("error: cannot find job with id %d: %s", jobID, err) log.Printf("error: rpc.update: cannot find pproc with id %d: %s", procID, err)
return err return err
} }
build, err := s.store.GetBuild(job.BuildID) build, err := s.store.GetBuild(pproc.BuildID)
if err != nil { if err != nil {
log.Printf("error: cannot find build with id %d: %s", job.BuildID, err) log.Printf("error: cannot find build with id %d: %s", pproc.BuildID, err)
return err
}
proc, err := s.store.ProcChild(build, pproc.PID, state.Proc)
if err != nil {
log.Printf("error: cannot find proc with name %s: %s", state.Proc, err)
return err return err
} }
@ -132,92 +135,231 @@ func (s *RPC) Update(c context.Context, id string, state rpc.State) error {
return err return err
} }
if build.Status != model.StatusRunning {
}
job.Started = state.Started
job.Finished = state.Finished
job.ExitCode = state.ExitCode
job.Status = model.StatusRunning
job.Error = state.Error
if build.Status == model.StatusPending {
build.Started = job.Started
build.Status = model.StatusRunning
s.store.UpdateBuild(build)
}
log.Printf("pipeline: update %s: exited=%v, exit_code=%d", id, state.Exited, state.ExitCode)
if state.Exited { if state.Exited {
proc.Stopped = state.Finished
job.Status = model.StatusSuccess proc.ExitCode = state.ExitCode
if job.ExitCode != 0 || job.Error != "" { proc.Error = state.Error
job.Status = model.StatusFailure proc.State = model.StatusSuccess
if state.ExitCode != 0 || state.Error != "" {
proc.State = model.StatusFailure
} }
} else {
// save the logs proc.Started = state.Started
var buf bytes.Buffer proc.State = model.StatusRunning
if serr := s.logger.Snapshot(context.Background(), id, &buf); serr != nil {
log.Printf("error: snapshotting logs: %s", serr)
}
if werr := s.store.WriteLog(job, &buf); werr != nil {
log.Printf("error: persisting logs: %s", werr)
}
// close the logger
s.logger.Close(c, id)
s.queue.Done(c, id)
} }
// hackity hack if err := s.store.ProcUpdate(proc); err != nil {
cc := context.WithValue(c, "store", s.store) log.Printf("error: rpc.update: cannot update proc: %s", err)
ok, uerr := store.UpdateBuildJob(cc, build, job)
if uerr != nil {
log.Printf("error: updating job: %s", uerr)
}
if ok {
// get the user because we transfer the user form the server to agent
// and back we lose the token which does not get serialized to json.
user, uerr := s.store.GetUser(repo.UserID)
if uerr != nil {
logrus.Errorf("Unable to find user. %s", err)
} else {
s.remote.Status(user, repo, build,
fmt.Sprintf("%s/%s/%d", s.host, repo.FullName, build.Number))
}
} }
message := pubsub.Message{} build.Procs, _ = s.store.ProcList(build)
build.Procs = model.Tree(build.Procs)
message := pubsub.Message{
Labels: map[string]string{
"repo": repo.FullName,
"private": strconv.FormatBool(repo.IsPrivate),
},
}
message.Data, _ = json.Marshal(model.Event{ message.Data, _ = json.Marshal(model.Event{
Type: func() model.EventType {
// HACK we don't even really care about the event type.
// so we should just simplify how events are triggered.
// WTF was this being used for?????????????????????????
if job.Status == model.StatusRunning {
return model.Started
}
return model.Finished
}(),
Repo: *repo, Repo: *repo,
Build: *build, Build: *build,
Job: *job,
}) })
message.Labels = map[string]string{
"repo": repo.FullName,
"private": strconv.FormatBool(repo.IsPrivate),
}
s.pubsub.Publish(c, "topic/events", message) s.pubsub.Publish(c, "topic/events", message)
log.Println("finish rpc.update")
return nil return nil
} }
// Upload implements the rpc.Upload function // Upload implements the rpc.Upload function
func (s *RPC) Upload(c context.Context, id, mime string, file io.Reader) error { return nil } func (s *RPC) Upload(c context.Context, id string, file *rpc.File) error {
procID, err := strconv.ParseInt(id, 10, 64)
if err != nil {
return err
}
pproc, err := s.store.ProcLoad(procID)
if err != nil {
log.Printf("error: cannot find parent proc with id %d: %s", procID, err)
return err
}
build, err := s.store.GetBuild(pproc.BuildID)
if err != nil {
log.Printf("error: cannot find build with id %d: %s", pproc.BuildID, err)
return err
}
proc, err := s.store.ProcChild(build, pproc.PID, file.Proc)
if err != nil {
log.Printf("error: cannot find child proc with name %s: %s", file.Proc, err)
return err
}
if file.Mime == "application/json+logs" {
return s.store.LogSave(
proc,
bytes.NewBuffer(file.Data),
)
}
return s.store.FileCreate(&model.File{
BuildID: proc.BuildID,
ProcID: proc.ID,
Mime: file.Mime,
Name: file.Name,
Size: file.Size,
Time: file.Time,
},
bytes.NewBuffer(file.Data),
)
}
// Init implements the rpc.Init function
func (s *RPC) Init(c context.Context, id string, state rpc.State) error {
procID, err := strconv.ParseInt(id, 10, 64)
if err != nil {
return err
}
proc, err := s.store.ProcLoad(procID)
if err != nil {
log.Printf("error: cannot find proc with id %d: %s", procID, err)
return err
}
build, err := s.store.GetBuild(proc.BuildID)
if err != nil {
log.Printf("error: cannot find build with id %d: %s", proc.BuildID, err)
return err
}
repo, err := s.store.GetRepo(build.RepoID)
if err != nil {
log.Printf("error: cannot find repo with id %d: %s", build.RepoID, err)
return err
}
if build.Status == model.StatusPending {
build.Status = model.StatusRunning
build.Started = state.Started
if err := s.store.UpdateBuild(build); err != nil {
log.Printf("error: init: cannot update build_id %d state: %s", build.ID, err)
}
}
defer func() {
build.Procs, _ = s.store.ProcList(build)
message := pubsub.Message{
Labels: map[string]string{
"repo": repo.FullName,
"private": strconv.FormatBool(repo.IsPrivate),
},
}
message.Data, _ = json.Marshal(model.Event{
Repo: *repo,
Build: *build,
})
s.pubsub.Publish(c, "topic/events", message)
}()
proc.Started = state.Started
proc.State = model.StatusRunning
return s.store.ProcUpdate(proc)
}
// Done implements the rpc.Done function // Done implements the rpc.Done function
func (s *RPC) Done(c context.Context, id string) error { return nil } func (s *RPC) Done(c context.Context, id string, state rpc.State) error {
procID, err := strconv.ParseInt(id, 10, 64)
if err != nil {
return err
}
proc, err := s.store.ProcLoad(procID)
if err != nil {
log.Printf("error: cannot find proc with id %d: %s", procID, err)
return err
}
build, err := s.store.GetBuild(proc.BuildID)
if err != nil {
log.Printf("error: cannot find build with id %d: %s", proc.BuildID, err)
return err
}
repo, err := s.store.GetRepo(build.RepoID)
if err != nil {
log.Printf("error: cannot find repo with id %d: %s", build.RepoID, err)
return err
}
proc.Stopped = state.Finished
proc.Error = state.Error
proc.ExitCode = state.ExitCode
proc.State = model.StatusSuccess
if proc.ExitCode != 0 || proc.Error != "" {
proc.State = model.StatusFailure
}
if err := s.store.ProcUpdate(proc); err != nil {
log.Printf("error: done: cannot update proc_id %d state: %s", procID, err)
}
if err := s.queue.Done(c, id); err != nil {
log.Printf("error: done: cannot ack proc_id %d: %s", procID, err)
}
// TODO handle this error
procs, _ := s.store.ProcList(build)
for _, p := range procs {
if p.Running() && p.PPID == proc.PID {
p.State = model.StatusSkipped
if p.Started != 0 {
p.State = model.StatusSuccess // for deamons that are killed
p.Stopped = proc.Stopped
}
if err := s.store.ProcUpdate(p); err != nil {
log.Printf("error: done: cannot update proc_id %d child state: %s", p.ID, err)
}
}
}
running := false
status := model.StatusSuccess
for _, p := range procs {
if p.PPID == 0 {
if p.Running() {
running = true
}
if p.Failing() {
status = p.State
}
}
}
if !running {
build.Status = status
build.Finished = proc.Stopped
if err := s.store.UpdateBuild(build); err != nil {
log.Printf("error: done: cannot update build_id %d final state: %s", build.ID, err)
}
}
if err := s.logger.Close(c, id); err != nil {
log.Printf("error: done: cannot close build_id %d logger: %s", proc.ID, err)
}
build.Procs = model.Tree(procs)
message := pubsub.Message{
Labels: map[string]string{
"repo": repo.FullName,
"private": strconv.FormatBool(repo.IsPrivate),
},
}
message.Data, _ = json.Marshal(model.Event{
Repo: *repo,
Build: *build,
})
s.pubsub.Publish(c, "topic/events", message)
return nil
}
// Log implements the rpc.Log function // Log implements the rpc.Log function
func (s *RPC) Log(c context.Context, id string, line *rpc.Line) error { func (s *RPC) Log(c context.Context, id string, line *rpc.Line) error {

8
server/stream.go
View File

@ -66,13 +66,13 @@ func LogStream(c *gin.Context) {
c.AbortWithError(404, err) c.AbortWithError(404, err)
return return
} }
job, err := store.GetJobNumber(c, build, jobn) proc, err := store.FromContext(c).ProcFind(build, jobn)
if err != nil { if err != nil {
logrus.Debugln("stream cannot get job number.", err) logrus.Debugln("stream cannot get proc number.", err)
c.AbortWithError(404, err) c.AbortWithError(404, err)
return return
} }
if job.Status != model.StatusRunning { if proc.State != model.StatusRunning {
logrus.Debugln("stream not found.") logrus.Debugln("stream not found.")
c.AbortWithStatus(404) c.AbortWithStatus(404)
return return
@ -102,7 +102,7 @@ func LogStream(c *gin.Context) {
go func() { go func() {
// TODO remove global variable // TODO remove global variable
config.logger.Tail(ctx, fmt.Sprint(job.ID), func(entries ...*logging.Entry) { config.logger.Tail(ctx, fmt.Sprint(proc.ID), func(entries ...*logging.Entry) {
for _, entry := range entries { for _, entry := range entries {
select { select {
case <-ctx.Done(): case <-ctx.Done():

10
store/datastore/builds.go
View File

@ -55,7 +55,7 @@ func (db *datastore) GetBuildQueue() ([]*model.Feed, error) {
return feed, err return feed, err
} }
func (db *datastore) CreateBuild(build *model.Build, jobs ...*model.Job) error { func (db *datastore) CreateBuild(build *model.Build, procs ...*model.Proc) error {
var number int var number int
db.QueryRow(rebind(buildNumberLast), build.RepoID).Scan(&number) db.QueryRow(rebind(buildNumberLast), build.RepoID).Scan(&number)
build.Number = number + 1 build.Number = number + 1
@ -65,11 +65,9 @@ func (db *datastore) CreateBuild(build *model.Build, jobs ...*model.Job) error {
if err != nil { if err != nil {
return err return err
} }
for i, job := range jobs { for _, proc := range procs {
job.BuildID = build.ID proc.BuildID = build.ID
job.Number = i + 1 err = meddler.Insert(db, "procs", proc)
job.Enqueued = build.Created
err = meddler.Insert(db, jobTable, job)
if err != nil { if err != nil {
return err return err
} }

36
store/datastore/builds_test.go
View File

@ -28,7 +28,7 @@ func TestBuilds(t *testing.T) {
Status: model.StatusSuccess, Status: model.StatusSuccess,
Commit: "85f8c029b902ed9400bc600bac301a0aadb144ac", Commit: "85f8c029b902ed9400bc600bac301a0aadb144ac",
} }
err := s.CreateBuild(&build, []*model.Job{}...) err := s.CreateBuild(&build, []*model.Proc{}...)
g.Assert(err == nil).IsTrue() g.Assert(err == nil).IsTrue()
g.Assert(build.ID != 0).IsTrue() g.Assert(build.ID != 0).IsTrue()
g.Assert(build.Number).Equal(1) g.Assert(build.Number).Equal(1)
@ -42,7 +42,7 @@ func TestBuilds(t *testing.T) {
Status: model.StatusSuccess, Status: model.StatusSuccess,
Commit: "85f8c029b902ed9400bc600bac301a0aadb144ac", Commit: "85f8c029b902ed9400bc600bac301a0aadb144ac",
} }
s.CreateBuild(&build, []*model.Job{}...) s.CreateBuild(&build, []*model.Proc{}...)
build.Status = model.StatusRunning build.Status = model.StatusRunning
err1 := s.UpdateBuild(&build) err1 := s.UpdateBuild(&build)
getbuild, err2 := s.GetBuild(build.ID) getbuild, err2 := s.GetBuild(build.ID)
@ -59,7 +59,7 @@ func TestBuilds(t *testing.T) {
RepoID: 1, RepoID: 1,
Status: model.StatusSuccess, Status: model.StatusSuccess,
} }
s.CreateBuild(&build, []*model.Job{}...) s.CreateBuild(&build, []*model.Proc{}...)
getbuild, err := s.GetBuild(build.ID) getbuild, err := s.GetBuild(build.ID)
g.Assert(err == nil).IsTrue() g.Assert(err == nil).IsTrue()
g.Assert(build.ID).Equal(getbuild.ID) g.Assert(build.ID).Equal(getbuild.ID)
@ -76,8 +76,8 @@ func TestBuilds(t *testing.T) {
RepoID: 1, RepoID: 1,
Status: model.StatusPending, Status: model.StatusPending,
} }
err1 := s.CreateBuild(build1, []*model.Job{}...) err1 := s.CreateBuild(build1, []*model.Proc{}...)
err2 := s.CreateBuild(build2, []*model.Job{}...) err2 := s.CreateBuild(build2, []*model.Proc{}...)
getbuild, err3 := s.GetBuildNumber(&model.Repo{ID: 1}, build2.Number) getbuild, err3 := s.GetBuildNumber(&model.Repo{ID: 1}, build2.Number)
g.Assert(err1 == nil).IsTrue() g.Assert(err1 == nil).IsTrue()
g.Assert(err2 == nil).IsTrue() g.Assert(err2 == nil).IsTrue()
@ -98,8 +98,8 @@ func TestBuilds(t *testing.T) {
Status: model.StatusPending, Status: model.StatusPending,
Ref: "refs/pull/6", Ref: "refs/pull/6",
} }
err1 := s.CreateBuild(build1, []*model.Job{}...) err1 := s.CreateBuild(build1, []*model.Proc{}...)
err2 := s.CreateBuild(build2, []*model.Job{}...) err2 := s.CreateBuild(build2, []*model.Proc{}...)
getbuild, err3 := s.GetBuildRef(&model.Repo{ID: 1}, "refs/pull/6") getbuild, err3 := s.GetBuildRef(&model.Repo{ID: 1}, "refs/pull/6")
g.Assert(err1 == nil).IsTrue() g.Assert(err1 == nil).IsTrue()
g.Assert(err2 == nil).IsTrue() g.Assert(err2 == nil).IsTrue()
@ -121,8 +121,8 @@ func TestBuilds(t *testing.T) {
Status: model.StatusPending, Status: model.StatusPending,
Ref: "refs/pull/6", Ref: "refs/pull/6",
} }
err1 := s.CreateBuild(build1, []*model.Job{}...) err1 := s.CreateBuild(build1, []*model.Proc{}...)
err2 := s.CreateBuild(build2, []*model.Job{}...) err2 := s.CreateBuild(build2, []*model.Proc{}...)
getbuild, err3 := s.GetBuildRef(&model.Repo{ID: 1}, "refs/pull/6") getbuild, err3 := s.GetBuildRef(&model.Repo{ID: 1}, "refs/pull/6")
g.Assert(err1 == nil).IsTrue() g.Assert(err1 == nil).IsTrue()
g.Assert(err2 == nil).IsTrue() g.Assert(err2 == nil).IsTrue()
@ -146,8 +146,8 @@ func TestBuilds(t *testing.T) {
Branch: "dev", Branch: "dev",
Commit: "85f8c029b902ed9400bc600bac301a0aadb144aa", Commit: "85f8c029b902ed9400bc600bac301a0aadb144aa",
} }
err1 := s.CreateBuild(build1, []*model.Job{}...) err1 := s.CreateBuild(build1, []*model.Proc{}...)
err2 := s.CreateBuild(build2, []*model.Job{}...) err2 := s.CreateBuild(build2, []*model.Proc{}...)
getbuild, err3 := s.GetBuildCommit(&model.Repo{ID: 1}, build2.Commit, build2.Branch) getbuild, err3 := s.GetBuildCommit(&model.Repo{ID: 1}, build2.Commit, build2.Branch)
g.Assert(err1 == nil).IsTrue() g.Assert(err1 == nil).IsTrue()
g.Assert(err2 == nil).IsTrue() g.Assert(err2 == nil).IsTrue()
@ -174,8 +174,8 @@ func TestBuilds(t *testing.T) {
Commit: "85f8c029b902ed9400bc600bac301a0aadb144aa", Commit: "85f8c029b902ed9400bc600bac301a0aadb144aa",
Event: model.EventPush, Event: model.EventPush,
} }
err1 := s.CreateBuild(build1, []*model.Job{}...) err1 := s.CreateBuild(build1, []*model.Proc{}...)
err2 := s.CreateBuild(build2, []*model.Job{}...) err2 := s.CreateBuild(build2, []*model.Proc{}...)
getbuild, err3 := s.GetBuildLast(&model.Repo{ID: 1}, build2.Branch) getbuild, err3 := s.GetBuildLast(&model.Repo{ID: 1}, build2.Branch)
g.Assert(err1 == nil).IsTrue() g.Assert(err1 == nil).IsTrue()
g.Assert(err2 == nil).IsTrue() g.Assert(err2 == nil).IsTrue()
@ -207,9 +207,9 @@ func TestBuilds(t *testing.T) {
Branch: "master", Branch: "master",
Commit: "85f8c029b902ed9400bc600bac301a0aadb144aa", Commit: "85f8c029b902ed9400bc600bac301a0aadb144aa",
} }
err1 := s.CreateBuild(build1, []*model.Job{}...) err1 := s.CreateBuild(build1, []*model.Proc{}...)
err2 := s.CreateBuild(build2, []*model.Job{}...) err2 := s.CreateBuild(build2, []*model.Proc{}...)
err3 := s.CreateBuild(build3, []*model.Job{}...) err3 := s.CreateBuild(build3, []*model.Proc{}...)
getbuild, err4 := s.GetBuildLastBefore(&model.Repo{ID: 1}, build3.Branch, build3.ID) getbuild, err4 := s.GetBuildLastBefore(&model.Repo{ID: 1}, build3.Branch, build3.ID)
g.Assert(err1 == nil).IsTrue() g.Assert(err1 == nil).IsTrue()
g.Assert(err2 == nil).IsTrue() g.Assert(err2 == nil).IsTrue()
@ -232,8 +232,8 @@ func TestBuilds(t *testing.T) {
RepoID: 1, RepoID: 1,
Status: model.StatusSuccess, Status: model.StatusSuccess,
} }
s.CreateBuild(build1, []*model.Job{}...) s.CreateBuild(build1, []*model.Proc{}...)
s.CreateBuild(build2, []*model.Job{}...) s.CreateBuild(build2, []*model.Proc{}...)
builds, err := s.GetBuildList(&model.Repo{ID: 1}) builds, err := s.GetBuildList(&model.Repo{ID: 1})
g.Assert(err == nil).IsTrue() g.Assert(err == nil).IsTrue()
g.Assert(len(builds)).Equal(2) g.Assert(len(builds)).Equal(2)

45
store/datastore/ddl/mysql/13.sql Normal file
View File

@ -0,0 +1,45 @@
-- +migrate Up
CREATE TABLE procs (
proc_id INTEGER PRIMARY KEY AUTO_INCREMENT
,proc_build_id INTEGER
,proc_pid INTEGER
,proc_ppid INTEGER
,proc_pgid INTEGER
,proc_name VARCHAR(250)
,proc_state VARCHAR(250)
,proc_error VARCHAR(500)
,proc_exit_code INTEGER
,proc_started INTEGER
,proc_stopped INTEGER
,proc_machine VARCHAR(250)
,proc_platform VARCHAR(250)
,proc_environ VARCHAR(2000)
,UNIQUE(proc_build_id, proc_pid)
);
CREATE INDEX proc_build_ix ON procs (proc_build_id);
CREATE TABLE files (
file_id INTEGER PRIMARY KEY AUTO_INCREMENT
,file_build_id INTEGER
,file_proc_id INTEGER
,file_name VARCHAR(250)
,file_mime VARCHAR(250)
,file_size INTEGER
,file_time INTEGER
,file_data MEDIUMBLOB
,UNIQUE(file_proc_id,file_name)
);
CREATE INDEX file_build_ix ON files (file_build_id);
CREATE INDEX file_proc_ix ON files (file_proc_id);
-- +migrate Down
DROP INDEX file_build_ix;
DROP INDEX file_proc_ix;
DROP TABLE files;
DROP INDEX proc_build_ix;
DROP TABLE procs;

47
store/datastore/ddl/postgres/13.sql Normal file
View File

@ -0,0 +1,47 @@
-- +migrate Up
CREATE TABLE procs (
proc_id SERIAL PRIMARY KEY
,proc_build_id INTEGER
,proc_pid INTEGER
,proc_ppid INTEGER
,proc_pgid INTEGER
,proc_name VARCHAR(250)
,proc_state VARCHAR(250)
,proc_error VARCHAR(500)
,proc_exit_code INTEGER
,proc_started INTEGER
,proc_stopped INTEGER
,proc_machine VARCHAR(250)
,proc_platform VARCHAR(250)
,proc_environ VARCHAR(2000)
,UNIQUE(proc_build_id, proc_pid)
);
CREATE INDEX proc_build_ix ON procs (proc_build_id);
CREATE TABLE files (
file_id SERIAL PRIMARY KEY
,file_build_id INTEGER
,file_proc_id INTEGER
,file_name VARCHAR(250)
,file_mime VARCHAR(250)
,file_size INTEGER
,file_time INTEGER
,file_data BYTEA
,UNIQUE(file_proc_id,file_name)
);
CREATE INDEX file_build_ix ON files (file_build_id);
CREATE INDEX file_proc_ix ON files (file_proc_id);
-- +migrate Down
DROP INDEX file_build_ix;
DROP INDEX file_proc_ix;
DROP TABLE files;
DROP INDEX proc_build_ix;
DROP TABLE procs;

46
store/datastore/ddl/sqlite3/13.sql Normal file
View File

@ -0,0 +1,46 @@
-- +migrate Up
CREATE TABLE procs (
proc_id INTEGER PRIMARY KEY AUTOINCREMENT
,proc_build_id INTEGER
,proc_pid INTEGER
,proc_ppid INTEGER
,proc_pgid INTEGER
,proc_name TEXT
,proc_state TEXT
,proc_error TEXT
,proc_exit_code INTEGER
,proc_started INTEGER
,proc_stopped INTEGER
,proc_machine TEXT
,proc_platform TEXT
,proc_environ TEXT
,UNIQUE(proc_build_id, proc_pid)
);
CREATE INDEX proc_build_ix ON procs (proc_build_id);
CREATE TABLE files (
file_id INTEGER PRIMARY KEY AUTOINCREMENT
,file_build_id INTEGER
,file_proc_id INTEGER
,file_name TEXT
,file_mime TEXT
,file_size INTEGER
,file_time INTEGER
,file_data BLOB
,UNIQUE(file_proc_id,file_name)
,FOREIGN KEY(file_proc_id) REFERENCES procs (proc_id) ON DELETE CASCADE
);
CREATE INDEX file_build_ix ON files (file_build_id);
CREATE INDEX file_proc_ix ON files (file_proc_id);
-- +migrate Down
DROP INDEX file_build_ix;
DROP INDEX file_proc_ix;
DROP TABLE files;
DROP INDEX proc_build_ix;
DROP TABLE procs;

63
store/datastore/files.go Normal file
View File

@ -0,0 +1,63 @@
package datastore
import (
"bytes"
"io"
"io/ioutil"
"github.com/drone/drone/model"
"github.com/drone/drone/store/datastore/sql"
"github.com/russross/meddler"
)
func (db *datastore) FileList(build *model.Build) ([]*model.File, error) {
stmt := sql.Lookup(db.driver, "files-find-build")
list := []*model.File{}
err := meddler.QueryAll(db, &list, stmt, build.ID)
return list, err
}
func (db *datastore) FileFind(proc *model.Proc, name string) (*model.File, error) {
stmt := sql.Lookup(db.driver, "files-find-proc-name")
file := new(model.File)
err := meddler.QueryRow(db, file, stmt, proc.ID, name)
return file, err
}
func (db *datastore) FileRead(proc *model.Proc, name string) (io.ReadCloser, error) {
stmt := sql.Lookup(db.driver, "files-find-proc-name-data")
file := new(fileData)
err := meddler.QueryRow(db, file, stmt, proc.ID, name)
buf := bytes.NewBuffer(file.Data)
return ioutil.NopCloser(buf), err
}
func (db *datastore) FileCreate(file *model.File, r io.Reader) error {
d, err := ioutil.ReadAll(r)
if err != nil {
return err
}
f := fileData{
ID: file.ID,
BuildID: file.BuildID,
ProcID: file.ProcID,
Name: file.Name,
Size: file.Size,
Mime: file.Mime,
Time: file.Time,
Data: d,
}
return meddler.Insert(db, "files", &f)
}
type fileData struct {
ID int64 `meddler:"file_id,pk"`
BuildID int64 `meddler:"file_build_id"`
ProcID int64 `meddler:"file_proc_id"`
Name string `meddler:"file_name"`
Size int `meddler:"file_size"`
Mime string `meddler:"file_mime"`
Time int64 `meddler:"file_time"`
Data []byte `meddler:"file_data"`
}

184
store/datastore/files_test.go Normal file
View File

@ -0,0 +1,184 @@
package datastore
import (
"bytes"
"io/ioutil"
"testing"
"github.com/drone/drone/model"
)
func TestFileFind(t *testing.T) {
s := newTest()
defer func() {
s.Exec("delete from files")
s.Close()
}()
if err := s.FileCreate(
&model.File{
BuildID: 2,
ProcID: 1,
Name: "hello.txt",
Mime: "text/plain",
Size: 11,
},
bytes.NewBufferString("hello world"),
); err != nil {
t.Errorf("Unexpected error: insert file: %s", err)
return
}
file, err := s.FileFind(&model.Proc{ID: 1}, "hello.txt")
if err != nil {
t.Error(err)
return
}
if got, want := file.ID, int64(1); got != want {
t.Errorf("Want file id %d, got %d", want, got)
}
if got, want := file.BuildID, int64(2); got != want {
t.Errorf("Want file build id %d, got %d", want, got)
}
if got, want := file.ProcID, int64(1); got != want {
t.Errorf("Want file proc id %d, got %d", want, got)
}
if got, want := file.Name, "hello.txt"; got != want {
t.Errorf("Want file name %s, got %s", want, got)
}
if got, want := file.Mime, "text/plain"; got != want {
t.Errorf("Want file mime %s, got %s", want, got)
}
if got, want := file.Size, 11; got != want {
t.Errorf("Want file size %d, got %d", want, got)
}
rc, err := s.FileRead(&model.Proc{ID: 1}, "hello.txt")
if err != nil {
t.Error(err)
return
}
out, _ := ioutil.ReadAll(rc)
if got, want := string(out), "hello world"; got != want {
t.Errorf("Want file data %s, got %s", want, got)
}
}
func TestFileList(t *testing.T) {
s := newTest()
defer func() {
s.Exec("delete from files")
s.Close()
}()
s.FileCreate(
&model.File{
BuildID: 1,
ProcID: 1,
Name: "hello.txt",
Mime: "text/plain",
Size: 11,
},
bytes.NewBufferString("hello world"),
)
s.FileCreate(
&model.File{
BuildID: 1,
ProcID: 1,
Name: "hola.txt",
Mime: "text/plain",
Size: 11,
},
bytes.NewBufferString("hola mundo"),
)
files, err := s.FileList(&model.Build{ID: 1})
if err != nil {
t.Errorf("Unexpected error: select files: %s", err)
return
}
if got, want := len(files), 2; got != want {
t.Errorf("Wanted %d files, got %d", want, got)
}
}
func TestFileIndexes(t *testing.T) {
s := newTest()
defer func() {
s.Exec("delete from files")
s.Close()
}()
if err := s.FileCreate(
&model.File{
BuildID: 1,
ProcID: 1,
Name: "hello.txt",
Size: 11,
Mime: "text/plain",
},
bytes.NewBufferString("hello world"),
); err != nil {
t.Errorf("Unexpected error: insert file: %s", err)
return
}
// fail due to duplicate file name
if err := s.FileCreate(
&model.File{
BuildID: 1,
ProcID: 1,
Name: "hello.txt",
Mime: "text/plain",
Size: 11,
},
bytes.NewBufferString("hello world"),
); err == nil {
t.Errorf("Unexpected error: dupliate pid")
}
}
// func TestFileCascade(t *testing.T) {
// s := newTest()
// defer s.Close()
//
//
// err1 := s.ProcCreate([]*model.Proc{
// {
// BuildID: 1,
// PID: 1,
// PGID: 1,
// Name: "build",
// State: "success",
// },
// })
// err2 := s.FileCreate(
// &model.File{
// BuildID: 1,
// ProcID: 1,
// Name: "hello.txt",
// Mime: "text/plain",
// Size: 11,
// },
// bytes.NewBufferString("hello world"),
// )
//
// if err1 != nil {
// t.Errorf("Unexpected error: cannot insert proc: %s", err1)
// } else if err2 != nil {
// t.Errorf("Unexpected error: cannot insert file: %s", err2)
// }
//
// if _, err3 := s.ProcFind(&model.Build{ID: 1}, 1); err3 != nil {
// t.Errorf("Unexpected error: cannot get inserted proc: %s", err3)
// }
//
// db.Exec("delete from procs where proc_id = 1")
//
// file, err4 := s.FileFind(&model.Proc{ID: 1}, "hello.txt")
// if err4 == nil {
// t.Errorf("Expected no rows in result set error")
// t.Log(file)
// }
// }

95
store/datastore/jobs.go
View File

@ -1,49 +1,50 @@
package datastore package datastore
import ( //
"github.com/drone/drone/model" // import (
"github.com/russross/meddler" // "github.com/drone/drone/model"
) // "github.com/russross/meddler"
// )
func (db *datastore) GetJob(id int64) (*model.Job, error) { //
var job = new(model.Job) // func (db *datastore) GetJob(id int64) (*model.Job, error) {
var err = meddler.Load(db, jobTable, job, id) // var job = new(model.Job)
return job, err // var err = meddler.Load(db, jobTable, job, id)
} // return job, err
// }
func (db *datastore) GetJobNumber(build *model.Build, num int) (*model.Job, error) { //
var job = new(model.Job) // func (db *datastore) GetJobNumber(build *model.Build, num int) (*model.Job, error) {
var err = meddler.QueryRow(db, job, rebind(jobNumberQuery), build.ID, num) // var job = new(model.Job)
return job, err // var err = meddler.QueryRow(db, job, rebind(jobNumberQuery), build.ID, num)
} // return job, err
// }
func (db *datastore) GetJobList(build *model.Build) ([]*model.Job, error) { //
var jobs = []*model.Job{} // func (db *datastore) GetJobList(build *model.Build) ([]*model.Job, error) {
var err = meddler.QueryAll(db, &jobs, rebind(jobListQuery), build.ID) // var jobs = []*model.Job{}
return jobs, err // var err = meddler.QueryAll(db, &jobs, rebind(jobListQuery), build.ID)
} // return jobs, err
// }
func (db *datastore) CreateJob(job *model.Job) error { //
return meddler.Insert(db, jobTable, job) // func (db *datastore) CreateJob(job *model.Job) error {
} // return meddler.Insert(db, jobTable, job)
// }
func (db *datastore) UpdateJob(job *model.Job) error { //
return meddler.Update(db, jobTable, job) // func (db *datastore) UpdateJob(job *model.Job) error {
} // return meddler.Update(db, jobTable, job)
// }
const jobTable = "jobs" //
// const jobTable = "jobs"
const jobListQuery = ` //
SELECT * // const jobListQuery = `
FROM jobs // SELECT *
WHERE job_build_id = ? // FROM jobs
ORDER BY job_number ASC // WHERE job_build_id = ?
` // ORDER BY job_number ASC
// `
const jobNumberQuery = ` //
SELECT * // const jobNumberQuery = `
FROM jobs // SELECT *
WHERE job_build_id = ? // FROM jobs
AND job_number = ? // WHERE job_build_id = ?
LIMIT 1 // AND job_number = ?
` // LIMIT 1
// `

233
store/datastore/jobs_test.go
View File

@ -1,118 +1,119 @@
package datastore package datastore
import ( //
"testing" // import (
// "testing"
"github.com/drone/drone/model" //
"github.com/franela/goblin" // "github.com/drone/drone/model"
) // "github.com/franela/goblin"
// )
func TestJobs(t *testing.T) { //
db := openTest() // func TestJobs(t *testing.T) {
defer db.Close() // db := openTest()
// defer db.Close()
s := From(db) //
g := goblin.Goblin(t) // s := From(db)
g.Describe("Job", func() { // g := goblin.Goblin(t)
// g.Describe("Job", func() {
// before each test we purge the package table data from the database. //
g.BeforeEach(func() { // // before each test we purge the package table data from the database.
db.Exec("DELETE FROM jobs") // g.BeforeEach(func() {
db.Exec("DELETE FROM builds") // db.Exec("DELETE FROM jobs")
}) // db.Exec("DELETE FROM builds")
// })
g.It("Should Set a job", func() { //
job := &model.Job{ // g.It("Should Set a job", func() {
BuildID: 1, // job := &model.Job{
Status: "pending", // BuildID: 1,
ExitCode: 0, // Status: "pending",
Number: 1, // ExitCode: 0,
} // Number: 1,
err1 := s.CreateJob(job) // }
g.Assert(err1 == nil).IsTrue() // err1 := s.CreateJob(job)
g.Assert(job.ID != 0).IsTrue() // g.Assert(err1 == nil).IsTrue()
// g.Assert(job.ID != 0).IsTrue()
job.Status = "started" //
err2 := s.UpdateJob(job) // job.Status = "started"
g.Assert(err2 == nil).IsTrue() // err2 := s.UpdateJob(job)
// g.Assert(err2 == nil).IsTrue()
getjob, err3 := s.GetJob(job.ID) //
g.Assert(err3 == nil).IsTrue() // getjob, err3 := s.GetJob(job.ID)
g.Assert(getjob.Status).Equal(job.Status) // g.Assert(err3 == nil).IsTrue()
}) // g.Assert(getjob.Status).Equal(job.Status)
// })
g.It("Should Get a Job by ID", func() { //
job := &model.Job{ // g.It("Should Get a Job by ID", func() {
BuildID: 1, // job := &model.Job{
Status: "pending", // BuildID: 1,
ExitCode: 1, // Status: "pending",
Number: 1, // ExitCode: 1,
Environment: map[string]string{"foo": "bar"}, // Number: 1,
} // Environment: map[string]string{"foo": "bar"},
err1 := s.CreateJob(job) // }
g.Assert(err1 == nil).IsTrue() // err1 := s.CreateJob(job)
g.Assert(job.ID != 0).IsTrue() // g.Assert(err1 == nil).IsTrue()
// g.Assert(job.ID != 0).IsTrue()
getjob, err2 := s.GetJob(job.ID) //
g.Assert(err2 == nil).IsTrue() // getjob, err2 := s.GetJob(job.ID)
g.Assert(getjob.ID).Equal(job.ID) // g.Assert(err2 == nil).IsTrue()
g.Assert(getjob.Status).Equal(job.Status) // g.Assert(getjob.ID).Equal(job.ID)
g.Assert(getjob.ExitCode).Equal(job.ExitCode) // g.Assert(getjob.Status).Equal(job.Status)
g.Assert(getjob.Environment).Equal(job.Environment) // g.Assert(getjob.ExitCode).Equal(job.ExitCode)
g.Assert(getjob.Environment["foo"]).Equal("bar") // g.Assert(getjob.Environment).Equal(job.Environment)
}) // g.Assert(getjob.Environment["foo"]).Equal("bar")
// })
g.It("Should Get a Job by Number", func() { //
job := &model.Job{ // g.It("Should Get a Job by Number", func() {
BuildID: 1, // job := &model.Job{
Status: "pending", // BuildID: 1,
ExitCode: 1, // Status: "pending",
Number: 1, // ExitCode: 1,
} // Number: 1,
err1 := s.CreateJob(job) // }
g.Assert(err1 == nil).IsTrue() // err1 := s.CreateJob(job)
g.Assert(job.ID != 0).IsTrue() // g.Assert(err1 == nil).IsTrue()
// g.Assert(job.ID != 0).IsTrue()
getjob, err2 := s.GetJobNumber(&model.Build{ID: 1}, 1) //
g.Assert(err2 == nil).IsTrue() // getjob, err2 := s.GetJobNumber(&model.Build{ID: 1}, 1)
g.Assert(getjob.ID).Equal(job.ID) // g.Assert(err2 == nil).IsTrue()
g.Assert(getjob.Status).Equal(job.Status) // g.Assert(getjob.ID).Equal(job.ID)
}) // g.Assert(getjob.Status).Equal(job.Status)
// })
g.It("Should Get a List of Jobs by Commit", func() { //
// g.It("Should Get a List of Jobs by Commit", func() {
build := model.Build{ //
RepoID: 1, // build := model.Build{
Status: model.StatusSuccess, // RepoID: 1,
} // Status: model.StatusSuccess,
jobs := []*model.Job{ // }
{ // jobs := []*model.Job{
BuildID: 1, // {
Status: "success", // BuildID: 1,
ExitCode: 0, // Status: "success",
Number: 1, // ExitCode: 0,
}, // Number: 1,
{ // },
BuildID: 3, // {
Status: "error", // BuildID: 3,
ExitCode: 1, // Status: "error",
Number: 2, // ExitCode: 1,
}, // Number: 2,
{ // },
BuildID: 5, // {
Status: "pending", // BuildID: 5,
ExitCode: 0, // Status: "pending",
Number: 3, // ExitCode: 0,
}, // Number: 3,
} // },
// }
err1 := s.CreateBuild(&build, jobs...) //
g.Assert(err1 == nil).IsTrue() // err1 := s.CreateBuild(&build, jobs...)
getjobs, err2 := s.GetJobList(&build) // g.Assert(err1 == nil).IsTrue()
g.Assert(err2 == nil).IsTrue() // getjobs, err2 := s.GetJobList(&build)
g.Assert(len(getjobs)).Equal(3) // g.Assert(err2 == nil).IsTrue()
g.Assert(getjobs[0].Number).Equal(1) // g.Assert(len(getjobs)).Equal(3)
g.Assert(getjobs[0].Status).Equal(model.StatusSuccess) // g.Assert(getjobs[0].Number).Equal(1)
}) // g.Assert(getjobs[0].Status).Equal(model.StatusSuccess)
}) // })
} // })
// }

20
store/datastore/logs.go
View File

@ -9,23 +9,29 @@ import (
"github.com/russross/meddler" "github.com/russross/meddler"
) )
func (db *datastore) ReadLog(job *model.Job) (io.ReadCloser, error) { func (db *datastore) LogFind(proc *model.Proc) (io.ReadCloser, error) {
var log = new(model.Log) var log = new(logData)
var err = meddler.QueryRow(db, log, rebind(logQuery), job.ID) var err = meddler.QueryRow(db, log, rebind(logQuery), proc.ID)
var buf = bytes.NewBuffer(log.Data) var buf = bytes.NewBuffer(log.Data)
return ioutil.NopCloser(buf), err return ioutil.NopCloser(buf), err
} }
func (db *datastore) WriteLog(job *model.Job, r io.Reader) error { func (db *datastore) LogSave(proc *model.Proc, r io.Reader) error {
var log = new(model.Log) var log = new(logData)
var err = meddler.QueryRow(db, log, rebind(logQuery), job.ID) var err = meddler.QueryRow(db, log, rebind(logQuery), proc.ID)
if err != nil { if err != nil {
log = &model.Log{JobID: job.ID} log = &logData{ProcID: proc.ID}
} }
log.Data, _ = ioutil.ReadAll(r) log.Data, _ = ioutil.ReadAll(r)
return meddler.Save(db, logTable, log) return meddler.Save(db, logTable, log)
} }
type logData struct {
ID int64 `meddler:"log_id,pk"`
ProcID int64 `meddler:"log_job_id"`
Data []byte `meddler:"log_data"`
}
const logTable = "logs" const logTable = "logs"
const logQuery = ` const logQuery = `

14
store/datastore/logs_test.go
View File

@ -24,14 +24,14 @@ func TestLogs(t *testing.T) {
}) })
g.It("Should create a log", func() { g.It("Should create a log", func() {
job := model.Job{ proc := model.Proc{
ID: 1, ID: 1,
} }
buf := bytes.NewBufferString("echo hi") buf := bytes.NewBufferString("echo hi")
err := s.WriteLog(&job, buf) err := s.LogSave(&proc, buf)
g.Assert(err == nil).IsTrue() g.Assert(err == nil).IsTrue()
rc, err := s.ReadLog(&job) rc, err := s.LogFind(&proc)
g.Assert(err == nil).IsTrue() g.Assert(err == nil).IsTrue()
defer rc.Close() defer rc.Close()
out, _ := ioutil.ReadAll(rc) out, _ := ioutil.ReadAll(rc)
@ -39,17 +39,17 @@ func TestLogs(t *testing.T) {
}) })
g.It("Should update a log", func() { g.It("Should update a log", func() {
job := model.Job{ proc := model.Proc{
ID: 1, ID: 1,
} }
buf1 := bytes.NewBufferString("echo hi") buf1 := bytes.NewBufferString("echo hi")
buf2 := bytes.NewBufferString("echo allo?") buf2 := bytes.NewBufferString("echo allo?")
err1 := s.WriteLog(&job, buf1) err1 := s.LogSave(&proc, buf1)
err2 := s.WriteLog(&job, buf2) err2 := s.LogSave(&proc, buf2)
g.Assert(err1 == nil).IsTrue() g.Assert(err1 == nil).IsTrue()
g.Assert(err2 == nil).IsTrue() g.Assert(err2 == nil).IsTrue()
rc, err := s.ReadLog(&job) rc, err := s.LogFind(&proc)
g.Assert(err == nil).IsTrue() g.Assert(err == nil).IsTrue()
defer rc.Close() defer rc.Close()
out, _ := ioutil.ReadAll(rc) out, _ := ioutil.ReadAll(rc)

59
store/datastore/procs.go Normal file
View File

@ -0,0 +1,59 @@
package datastore
import (
"github.com/drone/drone/model"
"github.com/drone/drone/store/datastore/sql"
"github.com/russross/meddler"
)
func (db *datastore) ProcLoad(id int64) (*model.Proc, error) {
stmt := sql.Lookup(db.driver, "procs-find-id")
proc := new(model.Proc)
err := meddler.QueryRow(db, proc, stmt, id)
return proc, err
}
func (db *datastore) ProcFind(build *model.Build, pid int) (*model.Proc, error) {
stmt := sql.Lookup(db.driver, "procs-find-build-pid")
proc := new(model.Proc)
err := meddler.QueryRow(db, proc, stmt, build.ID, pid)
return proc, err
}
func (db *datastore) ProcChild(build *model.Build, pid int, child string) (*model.Proc, error) {
stmt := sql.Lookup(db.driver, "procs-find-build-ppid")
proc := new(model.Proc)
err := meddler.QueryRow(db, proc, stmt, build.ID, pid, child)
return proc, err
}
func (db *datastore) ProcList(build *model.Build) ([]*model.Proc, error) {
stmt := sql.Lookup(db.driver, "procs-find-build")
list := []*model.Proc{}
err := meddler.QueryAll(db, &list, stmt, build.ID)
return list, err
}
func (db *datastore) ProcCreate(procs []*model.Proc) error {
for _, proc := range procs {
if err := meddler.Insert(db, "procs", proc); err != nil {
return err
}
}
return nil
}
func (db *datastore) ProcUpdate(proc *model.Proc) error {
return meddler.Update(db, "procs", proc)
}
func (db *datastore) ProcClear(build *model.Build) (err error) {
stmt1 := sql.Lookup(db.driver, "files-delete-build")
stmt2 := sql.Lookup(db.driver, "procs-delete-build")
_, err = db.Exec(stmt1, build.ID)
if err != nil {
return
}
_, err = db.Exec(stmt2, build.ID)
return
}

239
store/datastore/procs_test.go Normal file
View File

@ -0,0 +1,239 @@
package datastore
import (
"testing"
"github.com/drone/drone/model"
)
func TestProcFind(t *testing.T) {
s := newTest()
defer func() {
s.Exec("delete from procs")
s.Close()
}()
err := s.ProcCreate([]*model.Proc{
{
BuildID: 1000,
PID: 1,
PPID: 2,
PGID: 3,
Name: "build",
State: model.StatusSuccess,
Error: "pc load letter",
ExitCode: 255,
Machine: "localhost",
Platform: "linux/amd64",
Environ: map[string]string{"GOLANG": "tip"},
},
})
if err != nil {
t.Errorf("Unexpected error: insert procs: %s", err)
return
}
proc, err := s.ProcFind(&model.Build{ID: 1000}, 1)
if err != nil {
t.Error(err)
return
}
if got, want := proc.BuildID, int64(1000); got != want {
t.Errorf("Want proc fk %d, got %d", want, got)
}
if got, want := proc.ID, int64(1); got != want {
t.Errorf("Want proc pk %d, got %d", want, got)
}
if got, want := proc.PID, 1; got != want {
t.Errorf("Want proc ppid %d, got %d", want, got)
}
if got, want := proc.PPID, 2; got != want {
t.Errorf("Want proc ppid %d, got %d", want, got)
}
if got, want := proc.PGID, 3; got != want {
t.Errorf("Want proc pgid %d, got %d", want, got)
}
if got, want := proc.Name, "build"; got != want {
t.Errorf("Want proc name %s, got %s", want, got)
}
}
func TestProcChild(t *testing.T) {
s := newTest()
defer func() {
s.Exec("delete from procs")
s.Close()
}()
err := s.ProcCreate([]*model.Proc{
{
BuildID: 1,
PID: 1,
PPID: 1,
PGID: 1,
State: "success",
},
{
BuildID: 1,
PID: 2,
PGID: 2,
PPID: 1,
Name: "build",
State: "success",
},
})
if err != nil {
t.Errorf("Unexpected error: insert procs: %s", err)
return
}
proc, err := s.ProcChild(&model.Build{ID: 1}, 1, "build")
if err != nil {
t.Error(err)
return
}
if got, want := proc.PID, 2; got != want {
t.Errorf("Want proc pid %d, got %d", want, got)
}
if got, want := proc.Name, "build"; got != want {
t.Errorf("Want proc name %s, got %s", want, got)
}
}
func TestProcList(t *testing.T) {
s := newTest()
defer func() {
s.Exec("delete from procs")
s.Close()
}()
err := s.ProcCreate([]*model.Proc{
{
BuildID: 2,
PID: 1,
PPID: 1,
PGID: 1,
State: "success",
},
{
BuildID: 1,
PID: 1,
PPID: 1,
PGID: 1,
State: "success",
},
{
BuildID: 1,
PID: 2,
PGID: 2,
PPID: 1,
Name: "build",
State: "success",
},
})
if err != nil {
t.Errorf("Unexpected error: insert procs: %s", err)
return
}
procs, err := s.ProcList(&model.Build{ID: 1})
if err != nil {
t.Error(err)
return
}
if got, want := len(procs), 2; got != want {
t.Errorf("Want %d procs, got %d", want, got)
}
}
func TestProcUpdate(t *testing.T) {
s := newTest()
defer func() {
s.Exec("delete from procs")
s.Close()
}()
proc := &model.Proc{
BuildID: 1,
PID: 1,
PPID: 2,
PGID: 3,
Name: "build",
State: "pending",
Error: "pc load letter",
ExitCode: 255,
Machine: "localhost",
Platform: "linux/amd64",
Environ: map[string]string{"GOLANG": "tip"},
}
if err := s.ProcCreate([]*model.Proc{proc}); err != nil {
t.Errorf("Unexpected error: insert proc: %s", err)
return
}
proc.State = "running"
if err := s.ProcUpdate(proc); err != nil {
t.Errorf("Unexpected error: update proc: %s", err)
return
}
updated, err := s.ProcFind(&model.Build{ID: 1}, 1)
if err != nil {
t.Error(err)
return
}
if got, want := updated.State, "running"; got != want {
t.Errorf("Want proc name %s, got %s", want, got)
}
}
func TestProcIndexes(t *testing.T) {
s := newTest()
defer func() {
s.Exec("delete from procs")
s.Close()
}()
if err := s.ProcCreate([]*model.Proc{
{
BuildID: 1,
PID: 1,
PPID: 1,
PGID: 1,
State: "running",
Name: "build",
},
}); err != nil {
t.Errorf("Unexpected error: insert procs: %s", err)
return
}
// fail due to duplicate pid
if err := s.ProcCreate([]*model.Proc{
{
BuildID: 1,
PID: 1,
PPID: 1,
PGID: 1,
State: "success",
Name: "clone",
},
}); err == nil {
t.Errorf("Unexpected error: dupliate pid")
}
// // fail due to duplicate process name
// if err := s.ProcCreate([]*model.Proc{
// {
// BuildID: 1,
// PID: 2,
// PPID: 1,
// PGID: 1,
// State: "success",
// Name: "build",
// },
// }); err == nil {
// t.Errorf("Unexpected error: dupliate name")
// }
}
// func TestProcCascade(t *testing.T) {
//
// }

24
store/datastore/sql/lookup.go Normal file
View File

@ -0,0 +1,24 @@
package sql
import (
"github.com/drone/drone/store/datastore/sql/postgres"
"github.com/drone/drone/store/datastore/sql/sqlite"
)
// Supported database drivers
const (
DriverSqlite = "sqlite3"
DriverMysql = "mysql"
DriverPostgres = "postgres"
)
// Lookup returns the named sql statement compatible with
// the specified database driver.
func Lookup(driver string, name string) string {
switch driver {
case DriverPostgres:
return postgres.Lookup(name)
default:
return sqlite.Lookup(name)
}
}

45
store/datastore/sql/postgres/files/files.sql Normal file
View File

@ -0,0 +1,45 @@
-- name: files-find-build
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
FROM files
WHERE file_build_id = $1
-- name: files-find-proc-name
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
FROM files
WHERE file_proc_id = $1
AND file_name = $2
-- name: files-find-proc-name-data
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
,file_data
FROM files
WHERE file_proc_id = $1
AND file_name = $2
-- name: files-delete-build
DELETE FROM files WHERE file_build_id = $1

86
store/datastore/sql/postgres/files/procs.sql Normal file
View File

@ -0,0 +1,86 @@
-- name: procs-find-id
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_id = $1
-- name: procs-find-build
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = $1
-- name: procs-find-build-pid
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = $1
AND proc_pid = $2
-- name: procs-find-build-ppid
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = $1
AND proc_ppid = $2
AND proc_name = $3
-- name: procs-delete-build
DELETE FROM procs WHERE proc_build_id = $1

3
store/datastore/sql/postgres/sql.go Normal file
View File

@ -0,0 +1,3 @@
package postgres
//go:generate sqlbin sql --package=postgres

151
store/datastore/sql/postgres/sql_gen.go Normal file
View File

@ -0,0 +1,151 @@
package postgres
// Lookup returns the named statement.
func Lookup(name string) string {
return index[name]
}
var index = map[string]string{
"files-find-build": filesFindBuild,
"files-find-proc-name": filesFindProcName,
"files-find-proc-name-data": filesFindProcNameData,
"files-delete-build": filesDeleteBuild,
"procs-find-id": procsFindId,
"procs-find-build": procsFindBuild,
"procs-find-build-pid": procsFindBuildPid,
"procs-find-build-ppid": procsFindBuildPpid,
"procs-delete-build": procsDeleteBuild,
}
var filesFindBuild = `
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
FROM files
WHERE file_build_id = $1
`
var filesFindProcName = `
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
FROM files
WHERE file_proc_id = $1
AND file_name = $2
`
var filesFindProcNameData = `
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
,file_data
FROM files
WHERE file_proc_id = $1
AND file_name = $2
`
var filesDeleteBuild = `
DELETE FROM files WHERE file_build_id = $1
`
var procsFindId = `
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_id = $1
`
var procsFindBuild = `
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = $1
`
var procsFindBuildPid = `
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = $1
AND proc_pid = $2
`
var procsFindBuildPpid = `
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = $1
AND proc_ppid = $2
AND proc_name = $3
`
var procsDeleteBuild = `
DELETE FROM procs WHERE proc_build_id = $1
`

45
store/datastore/sql/sqlite/files/files.sql Normal file
View File

@ -0,0 +1,45 @@
-- name: files-find-build
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
FROM files
WHERE file_build_id = ?
-- name: files-find-proc-name
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
FROM files
WHERE file_proc_id = ?
AND file_name = ?
-- name: files-find-proc-name-data
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
,file_data
FROM files
WHERE file_proc_id = ?
AND file_name = ?
-- name: files-delete-build
DELETE FROM files WHERE file_build_id = ?

86
store/datastore/sql/sqlite/files/procs.sql Normal file
View File

@ -0,0 +1,86 @@
-- name: procs-find-id
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_id = ?
-- name: procs-find-build
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = ?
-- name: procs-find-build-pid
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = ?
AND proc_pid = ?
-- name: procs-find-build-ppid
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = ?
AND proc_ppid = ?
AND proc_name = ?
-- name: procs-delete-build
DELETE FROM procs WHERE proc_build_id = ?

3
store/datastore/sql/sqlite/sql.go Normal file
View File

@ -0,0 +1,3 @@
package sqlite
//go:generate sqlbin sql --package=sqlite

151
store/datastore/sql/sqlite/sql_gen.go Normal file
View File

@ -0,0 +1,151 @@
package sqlite
// Lookup returns the named statement.
func Lookup(name string) string {
return index[name]
}
var index = map[string]string{
"files-find-build": filesFindBuild,
"files-find-proc-name": filesFindProcName,
"files-find-proc-name-data": filesFindProcNameData,
"files-delete-build": filesDeleteBuild,
"procs-find-id": procsFindId,
"procs-find-build": procsFindBuild,
"procs-find-build-pid": procsFindBuildPid,
"procs-find-build-ppid": procsFindBuildPpid,
"procs-delete-build": procsDeleteBuild,
}
var filesFindBuild = `
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
FROM files
WHERE file_build_id = ?
`
var filesFindProcName = `
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
FROM files
WHERE file_proc_id = ?
AND file_name = ?
`
var filesFindProcNameData = `
SELECT
file_id
,file_build_id
,file_proc_id
,file_name
,file_mime
,file_size
,file_time
,file_data
FROM files
WHERE file_proc_id = ?
AND file_name = ?
`
var filesDeleteBuild = `
DELETE FROM files WHERE file_build_id = ?
`
var procsFindId = `
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_id = ?
`
var procsFindBuild = `
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = ?
`
var procsFindBuildPid = `
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = ?
AND proc_pid = ?
`
var procsFindBuildPpid = `
SELECT
proc_id
,proc_build_id
,proc_pid
,proc_ppid
,proc_pgid
,proc_name
,proc_state
,proc_error
,proc_exit_code
,proc_started
,proc_stopped
,proc_machine
,proc_platform
,proc_environ
FROM procs
WHERE proc_build_id = ?
AND proc_ppid = ?
AND proc_name = ?
`
var procsDeleteBuild = `
DELETE FROM procs WHERE proc_build_id = ?
`

34
store/datastore/store.go
View File

@ -17,19 +17,24 @@ import (
// of the sql/database driver with a relational database backend. // of the sql/database driver with a relational database backend.
type datastore struct { type datastore struct {
*sql.DB *sql.DB
driver string
config string
} }
// New creates a database connection for the given driver and datasource // New creates a database connection for the given driver and datasource
// and returns a new Store. // and returns a new Store.
func New(driver, config string) store.Store { func New(driver, config string) store.Store {
return From( return &datastore{
open(driver, config), DB: open(driver, config),
) driver: driver,
config: config,
}
} }
// From returns a Store using an existing database connection. // From returns a Store using an existing database connection.
func From(db *sql.DB) store.Store { func From(db *sql.DB) store.Store {
return &datastore{db} return &datastore{DB: db}
} }
// open opens a new database connection with the specified // open opens a new database connection with the specified
@ -60,7 +65,7 @@ func open(driver, config string) *sql.DB {
return db return db
} }
// OpenTest opens a new database connection for testing purposes. // openTest opens a new database connection for testing purposes.
// The database driver and connection string are provided by // The database driver and connection string are provided by
// environment variables, with fallback to in-memory sqlite. // environment variables, with fallback to in-memory sqlite.
func openTest() *sql.DB { func openTest() *sql.DB {
@ -75,6 +80,25 @@ func openTest() *sql.DB {
return open(driver, config) return open(driver, config)
} }
// newTest creates a new database connection for testing purposes.
// The database driver and connection string are provided by
// environment variables, with fallback to in-memory sqlite.
func newTest() *datastore {
var (
driver = "sqlite3"
config = ":memory:"
)
if os.Getenv("DATABASE_DRIVER") != "" {
driver = os.Getenv("DATABASE_DRIVER")
config = os.Getenv("DATABASE_CONFIG")
}
return &datastore{
DB: open(driver, config),
driver: driver,
config: config,
}
}
// helper function to ping the database with backoff to ensure // helper function to ping the database with backoff to ensure
// a connection can be established before we proceed with the // a connection can be established before we proceed with the
// database setup and migration. // database setup and migration.

157
store/store.go
View File

@ -107,31 +107,31 @@ type Store interface {
GetBuildQueue() ([]*model.Feed, error) GetBuildQueue() ([]*model.Feed, error)
// CreateBuild creates a new build and jobs. // CreateBuild creates a new build and jobs.
CreateBuild(*model.Build, ...*model.Job) error CreateBuild(*model.Build, ...*model.Proc) error
// UpdateBuild updates a build. // UpdateBuild updates a build.
UpdateBuild(*model.Build) error UpdateBuild(*model.Build) error
// GetJob gets a job by unique ID. // // GetJob gets a job by unique ID.
GetJob(int64) (*model.Job, error) // GetJob(int64) (*model.Job, error)
//
// GetJobNumber gets a job by number. // // GetJobNumber gets a job by number.
GetJobNumber(*model.Build, int) (*model.Job, error) // GetJobNumber(*model.Build, int) (*model.Job, error)
//
// GetJobList gets a list of all users in the system. // // GetJobList gets a list of all users in the system.
GetJobList(*model.Build) ([]*model.Job, error) // GetJobList(*model.Build) ([]*model.Job, error)
//
// CreateJob creates a job. // // CreateJob creates a job.
CreateJob(*model.Job) error // CreateJob(*model.Job) error
//
// UpdateJob updates a job. // // UpdateJob updates a job.
UpdateJob(*model.Job) error // UpdateJob(*model.Job) error
//
// ReadLog reads the Job logs from the datastore. // // ReadLog reads the Job logs from the datastore.
ReadLog(*model.Job) (io.ReadCloser, error) // ReadLog(*model.Job) (io.ReadCloser, error)
//
// WriteLog writes the job logs to the datastore. // // WriteLog writes the job logs to the datastore.
WriteLog(*model.Job, io.Reader) error // WriteLog(*model.Job, io.Reader) error
GetAgent(int64) (*model.Agent, error) GetAgent(int64) (*model.Agent, error)
@ -144,6 +144,22 @@ type Store interface {
UpdateAgent(*model.Agent) error UpdateAgent(*model.Agent) error
DeleteAgent(*model.Agent) error DeleteAgent(*model.Agent) error
ProcLoad(int64) (*model.Proc, error)
ProcFind(*model.Build, int) (*model.Proc, error)
ProcChild(*model.Build, int, string) (*model.Proc, error)
ProcList(*model.Build) ([]*model.Proc, error)
ProcCreate([]*model.Proc) error
ProcUpdate(*model.Proc) error
ProcClear(*model.Build) error
LogFind(*model.Proc) (io.ReadCloser, error)
LogSave(*model.Proc, io.Reader) error
FileList(*model.Build) ([]*model.File, error)
FileFind(*model.Proc, string) (*model.File, error)
FileRead(*model.Proc, string) (io.ReadCloser, error)
FileCreate(*model.File, io.Reader) error
} }
const globalTeamName = "__global__" const globalTeamName = "__global__"
@ -336,82 +352,41 @@ func GetBuildQueue(c context.Context) ([]*model.Feed, error) {
return FromContext(c).GetBuildQueue() return FromContext(c).GetBuildQueue()
} }
func CreateBuild(c context.Context, build *model.Build, jobs ...*model.Job) error { func CreateBuild(c context.Context, build *model.Build, procs ...*model.Proc) error {
return FromContext(c).CreateBuild(build, jobs...) return FromContext(c).CreateBuild(build, procs...)
} }
func UpdateBuild(c context.Context, build *model.Build) error { func UpdateBuild(c context.Context, build *model.Build) error {
return FromContext(c).UpdateBuild(build) return FromContext(c).UpdateBuild(build)
} }
func UpdateBuildJob(c context.Context, build *model.Build, job *model.Job) (bool, error) { // func GetJob(c context.Context, id int64) (*model.Job, error) {
if err := UpdateJob(c, job); err != nil { // return FromContext(c).GetJob(id)
return false, err // }
} //
// func GetJobNumber(c context.Context, build *model.Build, num int) (*model.Job, error) {
// if the job is running or started we don't need to update the build // return FromContext(c).GetJobNumber(build, num)
// status since. // }
if job.Status == model.StatusRunning || job.Status == model.StatusPending { //
return false, nil // func GetJobList(c context.Context, build *model.Build) ([]*model.Job, error) {
} // return FromContext(c).GetJobList(build)
// }
jobs, err := GetJobList(c, build) //
if err != nil { // func CreateJob(c context.Context, job *model.Job) error {
return false, err // return FromContext(c).CreateJob(job)
} // }
// check to see if all jobs are finished for this build. If yes, we need to //
// calcualte the overall build status and finish time. // func UpdateJob(c context.Context, job *model.Job) error {
status := model.StatusSuccess // return FromContext(c).UpdateJob(job)
finish := job.Finished // }
for _, job := range jobs { //
if job.Finished > finish { // func ReadLog(c context.Context, job *model.Job) (io.ReadCloser, error) {
finish = job.Finished // return FromContext(c).ReadLog(job)
} // }
switch job.Status { //
case model.StatusSuccess: // func WriteLog(c context.Context, job *model.Job, r io.Reader) error {
// no-op // return FromContext(c).WriteLog(job, r)
case model.StatusRunning, model.StatusPending: // }
return false, nil
default:
status = job.Status
}
}
build.Status = status
build.Finished = finish
if err := FromContext(c).UpdateBuild(build); err != nil {
return false, err
}
return true, nil
}
func GetJob(c context.Context, id int64) (*model.Job, error) {
return FromContext(c).GetJob(id)
}
func GetJobNumber(c context.Context, build *model.Build, num int) (*model.Job, error) {
return FromContext(c).GetJobNumber(build, num)
}
func GetJobList(c context.Context, build *model.Build) ([]*model.Job, error) {
return FromContext(c).GetJobList(build)
}
func CreateJob(c context.Context, job *model.Job) error {
return FromContext(c).CreateJob(job)
}
func UpdateJob(c context.Context, job *model.Job) error {
return FromContext(c).UpdateJob(job)
}
func ReadLog(c context.Context, job *model.Job) (io.ReadCloser, error) {
return FromContext(c).ReadLog(job)
}
func WriteLog(c context.Context, job *model.Job, r io.Reader) error {
return FromContext(c).WriteLog(job, r)
}
func GetAgent(c context.Context, id int64) (*model.Agent, error) { func GetAgent(c context.Context, id int64) (*model.Agent, error) {
return FromContext(c).GetAgent(id) return FromContext(c).GetAgent(id)

24
vendor/github.com/cncd/pipeline/pipeline/rpc/client.go generated vendored
View File

@ -3,7 +3,6 @@ package rpc
import ( import (
"context" "context"
"io" "io"
"io/ioutil"
"log" "log"
"math" "math"
"net/http" "net/http"
@ -18,6 +17,7 @@ import (
const ( const (
methodNext = "next" methodNext = "next"
methodWait = "wait" methodWait = "wait"
methodInit = "init"
methodDone = "done" methodDone = "done"
methodExtend = "extend" methodExtend = "extend"
methodUpdate = "update" methodUpdate = "update"
@ -28,8 +28,7 @@ const (
type ( type (
uploadReq struct { uploadReq struct {
ID string `json:"id"` ID string `json:"id"`
Mime string `json:"mime"` File *File `json:"file"`
Data []byte `json:"data"`
} }
updateReq struct { updateReq struct {
@ -90,9 +89,16 @@ func (t *Client) Wait(c context.Context, id string) error {
return t.call(c, methodWait, id, nil) return t.call(c, methodWait, id, nil)
} }
// Init signals the pipeline is initialized.
func (t *Client) Init(c context.Context, id string, state State) error {
params := updateReq{id, state}
return t.call(c, methodInit, &params, nil)
}
// Done signals the pipeline is complete. // Done signals the pipeline is complete.
func (t *Client) Done(c context.Context, id string) error { func (t *Client) Done(c context.Context, id string, state State) error {
return t.call(c, methodDone, id, nil) params := updateReq{id, state}
return t.call(c, methodDone, &params, nil)
} }
// Extend extends the pipeline deadline. // Extend extends the pipeline deadline.
@ -113,12 +119,8 @@ func (t *Client) Log(c context.Context, id string, line *Line) error {
} }
// Upload uploads the pipeline artifact. // Upload uploads the pipeline artifact.
func (t *Client) Upload(c context.Context, id, mime string, file io.Reader) error { func (t *Client) Upload(c context.Context, id string, file *File) error {
data, err := ioutil.ReadAll(file) params := uploadReq{id, file}
if err != nil {
return err
}
params := uploadReq{id, mime, data}
return t.call(c, methodUpload, params, nil) return t.call(c, methodUpload, params, nil)
} }

24
vendor/github.com/cncd/pipeline/pipeline/rpc/line.go generated vendored
View File

@ -36,12 +36,13 @@ func (l *Line) String() string {
// LineWriter sends logs to the client. // LineWriter sends logs to the client.
type LineWriter struct { type LineWriter struct {
peer Peer peer Peer
id string id string
name string name string
num int num int
now time.Time now time.Time
rep *strings.Replacer rep *strings.Replacer
lines []*Line
} }
// NewLineWriter returns a new line reader. // NewLineWriter returns a new line reader.
@ -91,5 +92,16 @@ func (w *LineWriter) Write(p []byte) (n int, err error) {
// w.peer.Log(context.Background(), w.id, line) // w.peer.Log(context.Background(), w.id, line)
// w.num++ // w.num++
// } // }
w.lines = append(w.lines, line)
return len(p), nil return len(p), nil
} }
// Lines returns the line history
func (w *LineWriter) Lines() []*Line {
return w.lines
}
// Clear clears the line history
func (w *LineWriter) Clear() {
w.lines = w.lines[:0]
}

20
vendor/github.com/cncd/pipeline/pipeline/rpc/peer.go generated vendored
View File

@ -2,7 +2,6 @@ package rpc
import ( import (
"context" "context"
"io"
"github.com/cncd/pipeline/pipeline/backend" "github.com/cncd/pipeline/pipeline/backend"
) )
@ -33,6 +32,16 @@ type (
Config *backend.Config `json:"config"` Config *backend.Config `json:"config"`
Timeout int64 `json:"timeout"` Timeout int64 `json:"timeout"`
} }
// File defines a pipeline artifact.
File struct {
Name string `json:"name"`
Proc string `json:"proc"`
Mime string `json:"mime"`
Time int64 `json:"time"`
Size int `json:"size"`
Data []byte `json:"data"`
}
) )
// NoFilter is an empty filter. // NoFilter is an empty filter.
@ -43,11 +52,14 @@ type Peer interface {
// Next returns the next pipeline in the queue. // Next returns the next pipeline in the queue.
Next(c context.Context, f Filter) (*Pipeline, error) Next(c context.Context, f Filter) (*Pipeline, error)
// Wait blocks untilthe pipeline is complete. // Wait blocks until the pipeline is complete.
Wait(c context.Context, id string) error Wait(c context.Context, id string) error
// Init signals the pipeline is initialized.
Init(c context.Context, id string, state State) error
// Done signals the pipeline is complete. // Done signals the pipeline is complete.
Done(c context.Context, id string) error Done(c context.Context, id string, state State) error
// Extend extends the pipeline deadline // Extend extends the pipeline deadline
Extend(c context.Context, id string) error Extend(c context.Context, id string) error
@ -56,7 +68,7 @@ type Peer interface {
Update(c context.Context, id string, state State) error Update(c context.Context, id string, state State) error
// Upload uploads the pipeline artifact. // Upload uploads the pipeline artifact.
Upload(c context.Context, id, mime string, file io.Reader) error Upload(c context.Context, id string, file *File) error
// Log writes the pipeline log entry. // Log writes the pipeline log entry.
Log(c context.Context, id string, line *Line) error Log(c context.Context, id string, line *Line) error

22
vendor/github.com/cncd/pipeline/pipeline/rpc/server.go generated vendored
View File

@ -1,7 +1,6 @@
package rpc package rpc
import ( import (
"bytes"
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
@ -54,6 +53,8 @@ func (s *Server) router(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.
return s.next(ctx, req) return s.next(ctx, req)
case methodWait: case methodWait:
return s.wait(ctx, req) return s.wait(ctx, req)
case methodInit:
return s.init(ctx, req)
case methodDone: case methodDone:
return s.done(ctx, req) return s.done(ctx, req)
case methodExtend: case methodExtend:
@ -90,15 +91,24 @@ func (s *Server) wait(ctx context.Context, req *jsonrpc2.Request) (interface{},
return nil, s.peer.Wait(ctx, id) return nil, s.peer.Wait(ctx, id)
} }
// init unmarshals the rpc request parameters and invokes the peer.Init
// procedure. The results are retuned and written to the rpc response.
func (s *Server) init(ctx context.Context, req *jsonrpc2.Request) (interface{}, error) {
in := new(updateReq)
if err := json.Unmarshal([]byte(*req.Params), in); err != nil {
return nil, err
}
return nil, s.peer.Init(ctx, in.ID, in.State)
}
// done unmarshals the rpc request parameters and invokes the peer.Done // done unmarshals the rpc request parameters and invokes the peer.Done
// procedure. The results are retuned and written to the rpc response. // procedure. The results are retuned and written to the rpc response.
func (s *Server) done(ctx context.Context, req *jsonrpc2.Request) (interface{}, error) { func (s *Server) done(ctx context.Context, req *jsonrpc2.Request) (interface{}, error) {
var id string in := new(updateReq)
err := json.Unmarshal([]byte(*req.Params), &id) if err := json.Unmarshal([]byte(*req.Params), in); err != nil {
if err != nil {
return nil, err return nil, err
} }
return nil, s.peer.Done(ctx, id) return nil, s.peer.Done(ctx, in.ID, in.State)
} }
// extend unmarshals the rpc request parameters and invokes the peer.Extend // extend unmarshals the rpc request parameters and invokes the peer.Extend
@ -137,5 +147,5 @@ func (s *Server) upload(req *jsonrpc2.Request) (interface{}, error) {
if err := json.Unmarshal([]byte(*req.Params), in); err != nil { if err := json.Unmarshal([]byte(*req.Params), in); err != nil {
return nil, err return nil, err
} }
return nil, s.peer.Upload(noContext, in.ID, in.Mime, bytes.NewBuffer(in.Data)) return nil, s.peer.Upload(noContext, in.ID, in.File)
} }

46
vendor/vendor.json vendored
View File

@ -28,68 +28,68 @@
{ {
"checksumSHA1": "W3AuK8ocqHwlUajGmQLFvnRhTZE=", "checksumSHA1": "W3AuK8ocqHwlUajGmQLFvnRhTZE=",
"path": "github.com/cncd/pipeline/pipeline", "path": "github.com/cncd/pipeline/pipeline",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "Qu2FreqaMr8Yx2bW9O0cxAGgjr0=", "checksumSHA1": "Qu2FreqaMr8Yx2bW9O0cxAGgjr0=",
"path": "github.com/cncd/pipeline/pipeline/backend", "path": "github.com/cncd/pipeline/pipeline/backend",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "0CGXRaYwZhJxGIrGhn8WGpkFqPo=", "checksumSHA1": "0CGXRaYwZhJxGIrGhn8WGpkFqPo=",
"path": "github.com/cncd/pipeline/pipeline/backend/docker", "path": "github.com/cncd/pipeline/pipeline/backend/docker",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "/8wE+cVb7T4PQZgpLNu0DHzKGuE=", "checksumSHA1": "/8wE+cVb7T4PQZgpLNu0DHzKGuE=",
"path": "github.com/cncd/pipeline/pipeline/frontend", "path": "github.com/cncd/pipeline/pipeline/frontend",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "O0sulBQAHJeNLg3lO38Cq5uf/eg=", "checksumSHA1": "O0sulBQAHJeNLg3lO38Cq5uf/eg=",
"path": "github.com/cncd/pipeline/pipeline/frontend/yaml", "path": "github.com/cncd/pipeline/pipeline/frontend/yaml",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "ftyr9EJQl9D5OvzOcqGBS6stt0g=", "checksumSHA1": "ftyr9EJQl9D5OvzOcqGBS6stt0g=",
"path": "github.com/cncd/pipeline/pipeline/frontend/yaml/compiler", "path": "github.com/cncd/pipeline/pipeline/frontend/yaml/compiler",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "Q0GkNUFamVYIA1Fd8r0A5M6Gx54=", "checksumSHA1": "Q0GkNUFamVYIA1Fd8r0A5M6Gx54=",
"path": "github.com/cncd/pipeline/pipeline/frontend/yaml/linter", "path": "github.com/cncd/pipeline/pipeline/frontend/yaml/linter",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "kx2sPUIMozPC/g6E4w48h3FfH3k=", "checksumSHA1": "kx2sPUIMozPC/g6E4w48h3FfH3k=",
"path": "github.com/cncd/pipeline/pipeline/frontend/yaml/matrix", "path": "github.com/cncd/pipeline/pipeline/frontend/yaml/matrix",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "2/3f3oNmxXy5kcrRLCFa24Oc9O4=", "checksumSHA1": "2/3f3oNmxXy5kcrRLCFa24Oc9O4=",
"path": "github.com/cncd/pipeline/pipeline/interrupt", "path": "github.com/cncd/pipeline/pipeline/interrupt",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "uOjTfke7Qxosrivgz/nVTHeIP5g=", "checksumSHA1": "uOjTfke7Qxosrivgz/nVTHeIP5g=",
"path": "github.com/cncd/pipeline/pipeline/multipart", "path": "github.com/cncd/pipeline/pipeline/multipart",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "MratmNKJ78/IhWvDsZphN01CtmE=", "checksumSHA1": "TP5lK1T8cOKv5QjZ2nqdlYczSTo=",
"path": "github.com/cncd/pipeline/pipeline/rpc", "path": "github.com/cncd/pipeline/pipeline/rpc",
"revision": "addc99dad68008570994f8de318101adfe4161a6", "revision": "4b348532eddd31220de9a179c197d31a78b200f5",
"revisionTime": "2017-03-19T09:04:25Z" "revisionTime": "2017-03-29T08:36:18Z"
}, },
{ {
"checksumSHA1": "7Qj1DK0ceAXkYztW0l3+L6sn+V8=", "checksumSHA1": "7Qj1DK0ceAXkYztW0l3+L6sn+V8=",