From fc01782554033612f4d1571d7aa078d3ed91d2c0 Mon Sep 17 00:00:00 2001 From: Alex Suraci Date: Mon, 24 Feb 2014 15:41:14 -0800 Subject: [PATCH 1/5] add constructor for Builder this makes it easier to track required dependencies as they change (todo: actually, like, use it for required dependencies) --- cmd/drone/drone.go | 4 ++-- pkg/build/build.go | 4 ++++ pkg/queue/queue.go | 6 +++--- 3 files changed, 9 insertions(+), 5 deletions(-) diff --git a/cmd/drone/drone.go b/cmd/drone/drone.go index 9fdc6b3c7..c12e3ef20 100644 --- a/cmd/drone/drone.go +++ b/cmd/drone/drone.go @@ -174,7 +174,7 @@ func run(path string) { // loop through and create builders for _, b := range builds { //script.Builds { - builder := build.Builder{} + builder := build.New() builder.Build = b builder.Repo = &code builder.Key = key @@ -186,7 +186,7 @@ func run(path string) { builder.Stdout = &buf } - builders = append(builders, &builder) + builders = append(builders, builder) } switch *parallel { diff --git a/pkg/build/build.go b/pkg/build/build.go index b086d4f47..ba8d39691 100644 --- a/pkg/build/build.go +++ b/pkg/build/build.go @@ -35,6 +35,10 @@ type BuildState struct { // Max RAM, Max Swap, Disk space, and more. } +func New() *Builder { + return &Builder{} +} + // Builder represents a build process being prepared // to run. type Builder struct { diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go index e931c46f0..cadaee932 100644 --- a/pkg/queue/queue.go +++ b/pkg/queue/queue.go @@ -3,7 +3,7 @@ package queue import ( "bytes" "fmt" - bldr "github.com/drone/drone/pkg/build" + "github.com/drone/drone/pkg/build" "github.com/drone/drone/pkg/build/git" r "github.com/drone/drone/pkg/build/repo" "github.com/drone/drone/pkg/build/script" @@ -134,7 +134,7 @@ func (b *BuildTask) execute() error { } // execute the build - builder := bldr.Builder{} + builder := build.New() builder.Build = b.Script builder.Repo = &r.Repo{Path: b.Repo.URL, Branch: b.Commit.Branch, Commit: b.Commit.Hash, PR: b.Commit.PullRequest, Dir: filepath.Join("/var/cache/drone/src", b.Repo.Slug), Depth: git.GitDepth(b.Script.Git)} builder.Key = []byte(b.Repo.PrivateKey) @@ -225,7 +225,7 @@ func updateGitHubStatus(repo *Repo, commit *Commit) error { } client := github.New(user.GithubToken) - client.ApiUrl = settings.GitHubApiUrl; + client.ApiUrl = settings.GitHubApiUrl var url string url = settings.URL().String() + "/" + repo.Slug + "/commit/" + commit.Hash From 12989b187c0fd9adccae1e0b6027ccc41598ba5d Mon Sep 17 00:00:00 2001 From: Alex Suraci Date: Mon, 24 Feb 2014 16:53:28 -0800 Subject: [PATCH 2/5] introduce Queue object this is an intermediate step towards pushing configuration up. Signed-off-by: Abhijit Hiremagalur --- pkg/queue/init.go | 22 ---- pkg/queue/queue.go | 246 +++++--------------------------------------- pkg/queue/worker.go | 234 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 261 insertions(+), 241 deletions(-) delete mode 100644 pkg/queue/init.go create mode 100644 pkg/queue/worker.go diff --git a/pkg/queue/init.go b/pkg/queue/init.go deleted file mode 100644 index c0b4b6972..000000000 --- a/pkg/queue/init.go +++ /dev/null @@ -1,22 +0,0 @@ -package queue - -import ( - "runtime" -) - -func init() { - // get the number of CPUs. Since builds - // tend to be CPU-intensive we should only - // execute 1 build per CPU. - ncpu := runtime.NumCPU() - - // must be at least 1 - if ncpu < 1 { - ncpu = 1 - } - - // spawn a worker for each CPU - for i := 0; i < ncpu; i++ { - go work() - } -} diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go index cadaee932..2b69f33c9 100644 --- a/pkg/queue/queue.go +++ b/pkg/queue/queue.go @@ -1,46 +1,14 @@ package queue import ( - "bytes" - "fmt" - "github.com/drone/drone/pkg/build" - "github.com/drone/drone/pkg/build/git" - r "github.com/drone/drone/pkg/build/repo" "github.com/drone/drone/pkg/build/script" - "github.com/drone/drone/pkg/channel" - "github.com/drone/drone/pkg/database" . "github.com/drone/drone/pkg/model" - "github.com/drone/drone/pkg/plugin/notify" - "github.com/drone/go-github/github" - "log" - "path/filepath" - "time" + "runtime" ) -// queue that will store all build tasks until -// they are processed by a worker. -var queue = make(chan *BuildTask) - -// work is a function that will infinitely -// run in the background waiting for tasks that -// it can pull off the queue and execute. -func work() { - var task *BuildTask - for { - // get work item (pointer) from the queue - task = <-queue - if task == nil { - continue - } - - // execute the task - task.execute() - } -} - -// Add adds the task to the build queue. -func Add(task *BuildTask) { - queue <- task +// A Queue dispatches tasks to workers. +type Queue struct { + tasks chan<- *BuildTask } // BuildTasks represents a build that is pending @@ -55,193 +23,33 @@ type BuildTask struct { Script *script.Build } -// execute will execute the build task and persist -// the results to the datastore. -func (b *BuildTask) execute() error { - // we need to be sure that we can recover - // from any sort panic that could occur - // to avoid brining down the entire application - defer func() { - if e := recover(); e != nil { - b.Build.Finished = time.Now().UTC() - b.Commit.Finished = time.Now().UTC() - b.Build.Duration = b.Build.Finished.Unix() - b.Build.Started.Unix() - b.Commit.Duration = b.Build.Finished.Unix() - b.Build.Started.Unix() - b.Commit.Status = "Error" - b.Build.Status = "Error" - database.SaveBuild(b.Build) - database.SaveCommit(b.Commit) - } - }() +var defaultQueue = Start(runtime.NumCPU()) // TEMPORARY; INJECT PLEASE - // update commit and build status - b.Commit.Status = "Started" - b.Build.Status = "Started" - b.Build.Started = time.Now().UTC() - b.Commit.Started = time.Now().UTC() +var Add = defaultQueue.Add // TEMPORARY; INJECT PLEASE - // persist the commit to the database - if err := database.SaveCommit(b.Commit); err != nil { - return err +func Start(workers int) *Queue { + // get the number of CPUs. Since builds + // tend to be CPU-intensive we should only + // execute 1 build per CPU. + // must be at least 1 + // if ncpu < 1 { + // ncpu = 1 + // } + + tasks := make(chan *BuildTask) + + queue := &Queue{tasks: tasks} + + // spawn a worker for each CPU + for i := 0; i < workers; i++ { + worker := worker{} + go worker.work(tasks) } - // persist the build to the database - if err := database.SaveBuild(b.Build); err != nil { - return err - } - - // get settings - settings, _ := database.GetSettings() - - // notification context - context := ¬ify.Context{ - Repo: b.Repo, - Commit: b.Commit, - Host: settings.URL().String(), - } - - // send all "started" notifications - if b.Script.Notifications != nil { - b.Script.Notifications.Send(context) - } - - // Send "started" notification to Github - if err := updateGitHubStatus(b.Repo, b.Commit); err != nil { - log.Printf("error updating github status: %s\n", err.Error()) - } - - // make sure a channel exists for the repository, - // the commit, and the commit output (TODO) - reposlug := fmt.Sprintf("%s/%s/%s", b.Repo.Host, b.Repo.Owner, b.Repo.Name) - commitslug := fmt.Sprintf("%s/%s/%s/commit/%s", b.Repo.Host, b.Repo.Owner, b.Repo.Name, b.Commit.Hash) - consoleslug := fmt.Sprintf("%s/%s/%s/commit/%s/builds/%s", b.Repo.Host, b.Repo.Owner, b.Repo.Name, b.Commit.Hash, b.Build.Slug) - channel.Create(reposlug) - channel.Create(commitslug) - channel.CreateStream(consoleslug) - - // notify the channels that the commit and build started - channel.SendJSON(reposlug, b.Commit) - channel.SendJSON(commitslug, b.Build) - - var buf = &bufferWrapper{channel: consoleslug} - - // append private parameters to the environment - // variable section of the .drone.yml file - if b.Repo.Params != nil { - for k, v := range b.Repo.Params { - b.Script.Env = append(b.Script.Env, k+"="+v) - } - } - - // execute the build - builder := build.New() - builder.Build = b.Script - builder.Repo = &r.Repo{Path: b.Repo.URL, Branch: b.Commit.Branch, Commit: b.Commit.Hash, PR: b.Commit.PullRequest, Dir: filepath.Join("/var/cache/drone/src", b.Repo.Slug), Depth: git.GitDepth(b.Script.Git)} - builder.Key = []byte(b.Repo.PrivateKey) - builder.Stdout = buf - builder.Timeout = 300 * time.Minute - - defer func() { - // update the status of the commit using the - // GitHub status API. - if err := updateGitHubStatus(b.Repo, b.Commit); err != nil { - log.Printf("error updating github status: %s\n", err.Error()) - } - }() - - buildErr := builder.Run() - - b.Build.Finished = time.Now().UTC() - b.Commit.Finished = time.Now().UTC() - b.Build.Duration = b.Build.Finished.UnixNano() - b.Build.Started.UnixNano() - b.Commit.Duration = b.Build.Finished.UnixNano() - b.Build.Started.UnixNano() - b.Commit.Status = "Success" - b.Build.Status = "Success" - b.Build.Stdout = buf.buf.String() - - // if exit code != 0 set to failure - if builder.BuildState == nil || builder.BuildState.ExitCode != 0 { - b.Commit.Status = "Failure" - b.Build.Status = "Failure" - if buildErr != nil && b.Build.Stdout == "" { - // TODO: If you wanted to have very friendly error messages, you could do that here - b.Build.Stdout = buildErr.Error() + "\n" - } - } - - // persist the build to the database - if err := database.SaveBuild(b.Build); err != nil { - return err - } - - // persist the commit to the database - if err := database.SaveCommit(b.Commit); err != nil { - return err - } - - // notify the channels that the commit and build finished - channel.SendJSON(reposlug, b.Commit) - channel.SendJSON(commitslug, b.Build) - channel.Close(consoleslug) - - // send all "finished" notifications - if b.Script.Notifications != nil { - b.Script.Notifications.Send(context) - } - - return nil + return queue } -// updateGitHubStatus is a helper function that will send -// the build status to GitHub using the Status API. -// see https://github.com/blog/1227-commit-status-api -func updateGitHubStatus(repo *Repo, commit *Commit) error { - - // convert from drone status to github status - var message, status string - switch commit.Status { - case "Success": - status = "success" - message = "The build succeeded on drone.io" - case "Failure": - status = "failure" - message = "The build failed on drone.io" - case "Started": - status = "pending" - message = "The build is pending on drone.io" - default: - status = "error" - message = "The build errored on drone.io" - } - - // get the system settings - settings, _ := database.GetSettings() - - // get the user from the database - // since we need his / her GitHub token - user, err := database.GetUser(repo.UserID) - if err != nil { - return err - } - - client := github.New(user.GithubToken) - client.ApiUrl = settings.GitHubApiUrl - - var url string - url = settings.URL().String() + "/" + repo.Slug + "/commit/" + commit.Hash - - return client.Repos.CreateStatus(repo.Owner, repo.Name, status, url, message, commit.Hash) -} - -type bufferWrapper struct { - buf bytes.Buffer - - // name of the channel - channel string -} - -func (b *bufferWrapper) Write(p []byte) (n int, err error) { - n, err = b.buf.Write(p) - channel.SendBytes(b.channel, p) - return +// Add adds the task to the build queue. +func (q *Queue) Add(task *BuildTask) { + q.tasks <- task } diff --git a/pkg/queue/worker.go b/pkg/queue/worker.go new file mode 100644 index 000000000..e2cf8f6d8 --- /dev/null +++ b/pkg/queue/worker.go @@ -0,0 +1,234 @@ +package queue + +import ( + "bytes" + "fmt" + "github.com/drone/drone/pkg/build" + "github.com/drone/drone/pkg/build/git" + r "github.com/drone/drone/pkg/build/repo" + "github.com/drone/drone/pkg/channel" + "github.com/drone/drone/pkg/database" + . "github.com/drone/drone/pkg/model" + "github.com/drone/drone/pkg/plugin/notify" + "github.com/drone/go-github/github" + "io" + "log" + "path/filepath" + "time" +) + +type worker struct{} + +// work is a function that will infinitely +// run in the background waiting for tasks that +// it can pull off the queue and execute. +func (w *worker) work(queue <-chan *BuildTask) { + var task *BuildTask + for { + // get work item (pointer) from the queue + task = <-queue + if task == nil { + continue + } + + // execute the task + w.execute(task) + } +} + +// execute will execute the build task and persist +// the results to the datastore. +func (w *worker) execute(task *BuildTask) error { + // we need to be sure that we can recover + // from any sort panic that could occur + // to avoid brining down the entire application + defer func() { + if e := recover(); e != nil { + task.Build.Finished = time.Now().UTC() + task.Commit.Finished = time.Now().UTC() + task.Build.Duration = task.Build.Finished.Unix() - task.Build.Started.Unix() + task.Commit.Duration = task.Build.Finished.Unix() - task.Build.Started.Unix() + task.Commit.Status = "Error" + task.Build.Status = "Error" + database.SaveBuild(task.Build) + database.SaveCommit(task.Commit) + } + }() + + // update commit and build status + task.Commit.Status = "Started" + task.Build.Status = "Started" + task.Build.Started = time.Now().UTC() + task.Commit.Started = time.Now().UTC() + + // persist the commit to the database + if err := database.SaveCommit(task.Commit); err != nil { + return err + } + + // persist the build to the database + if err := database.SaveBuild(task.Build); err != nil { + return err + } + + // get settings + settings, _ := database.GetSettings() + + // notification context + context := ¬ify.Context{ + Repo: task.Repo, + Commit: task.Commit, + Host: settings.URL().String(), + } + + // send all "started" notifications + if task.Script.Notifications != nil { + task.Script.Notifications.Send(context) + } + + // Send "started" notification to Github + if err := updateGitHubStatus(task.Repo, task.Commit); err != nil { + log.Printf("error updating github status: %s\n", err.Error()) + } + + // make sure a channel exists for the repository, + // the commit, and the commit output (TODO) + reposlug := fmt.Sprintf("%s/%s/%s", task.Repo.Host, task.Repo.Owner, task.Repo.Name) + commitslug := fmt.Sprintf("%s/%s/%s/commit/%s", task.Repo.Host, task.Repo.Owner, task.Repo.Name, task.Commit.Hash) + consoleslug := fmt.Sprintf("%s/%s/%s/commit/%s/builds/%s", task.Repo.Host, task.Repo.Owner, task.Repo.Name, task.Commit.Hash, task.Build.Slug) + channel.Create(reposlug) + channel.Create(commitslug) + channel.CreateStream(consoleslug) + + // notify the channels that the commit and build started + channel.SendJSON(reposlug, task.Commit) + channel.SendJSON(commitslug, task.Build) + + var buf = &bufferWrapper{channel: consoleslug} + + // append private parameters to the environment + // variable section of the .drone.yml file + if task.Repo.Params != nil { + for k, v := range task.Repo.Params { + task.Script.Env = append(task.Script.Env, k+"="+v) + } + } + + defer func() { + // update the status of the commit using the + // GitHub status API. + if err := updateGitHubStatus(task.Repo, task.Commit); err != nil { + log.Printf("error updating github status: %s\n", err.Error()) + } + }() + + // execute the build + passed, buildErr := runBuild(task, buf) //w.builder.Build(script, repo, task.Repo.PrivateKey, buf) + + task.Build.Finished = time.Now().UTC() + task.Commit.Finished = time.Now().UTC() + task.Build.Duration = task.Build.Finished.UnixNano() - task.Build.Started.UnixNano() + task.Commit.Duration = task.Build.Finished.UnixNano() - task.Build.Started.UnixNano() + task.Commit.Status = "Success" + task.Build.Status = "Success" + task.Build.Stdout = buf.buf.String() + + // if exit code != 0 set to failure + if passed { + task.Commit.Status = "Failure" + task.Build.Status = "Failure" + if buildErr != nil && task.Build.Stdout == "" { + // TODO: If you wanted to have very friendly error messages, you could do that here + task.Build.Stdout = buildErr.Error() + "\n" + } + } + + // persist the build to the database + if err := database.SaveBuild(task.Build); err != nil { + return err + } + + // persist the commit to the database + if err := database.SaveCommit(task.Commit); err != nil { + return err + } + + // notify the channels that the commit and build finished + channel.SendJSON(reposlug, task.Commit) + channel.SendJSON(commitslug, task.Build) + channel.Close(consoleslug) + + // send all "finished" notifications + if task.Script.Notifications != nil { + task.Script.Notifications.Send(context) + } + + return nil +} + +func runBuild(b *BuildTask, buf io.Writer) (bool, error) { + builder := build.New() + builder.Build = b.Script + builder.Repo = &r.Repo{Path: b.Repo.URL, Branch: b.Commit.Branch, Commit: b.Commit.Hash, PR: b.Commit.PullRequest, Dir: filepath.Join("/var/cache/drone/src", b.Repo.Slug), Depth: git.GitDepth(b.Script.Git)} + builder.Key = []byte(b.Repo.PrivateKey) + builder.Stdout = buf + builder.Timeout = 300 * time.Minute + + buildErr := builder.Run() + + return builder.BuildState == nil || builder.BuildState.ExitCode != 0, buildErr +} + +// updateGitHubStatus is a helper function that will send +// the build status to GitHub using the Status API. +// see https://github.com/blog/1227-commit-status-api +func updateGitHubStatus(repo *Repo, commit *Commit) error { + + // convert from drone status to github status + var message, status string + switch commit.Status { + case "Success": + status = "success" + message = "The build succeeded on drone.io" + case "Failure": + status = "failure" + message = "The build failed on drone.io" + case "Started": + status = "pending" + message = "The build is pending on drone.io" + default: + status = "error" + message = "The build errored on drone.io" + } + + // get the system settings + settings, _ := database.GetSettings() + + // get the user from the database + // since we need his / her GitHub token + user, err := database.GetUser(repo.UserID) + if err != nil { + return err + } + + client := github.New(user.GithubToken) + client.ApiUrl = settings.GitHubApiUrl + + var url string + url = settings.URL().String() + "/" + repo.Slug + "/commit/" + commit.Hash + + return client.Repos.CreateStatus(repo.Owner, repo.Name, status, url, message, commit.Hash) +} + +type bufferWrapper struct { + buf bytes.Buffer + + // name of the channel + channel string +} + +func (b *bufferWrapper) Write(p []byte) (n int, err error) { + n, err = b.buf.Write(p) + channel.SendBytes(b.channel, p) + return +} From acc51e83fdb732d69dced3d3a2104d2008f74682 Mon Sep 17 00:00:00 2001 From: Alex Suraci Date: Mon, 24 Feb 2014 17:02:57 -0800 Subject: [PATCH 3/5] inject docker client into Builder Signed-off-by: Abhijit Hiremagalur --- cmd/drone/drone.go | 3 ++- pkg/build/build.go | 47 +++++++++++++++++++------------------- pkg/build/docker/client.go | 2 ++ pkg/queue/worker.go | 3 ++- 4 files changed, 30 insertions(+), 25 deletions(-) diff --git a/cmd/drone/drone.go b/cmd/drone/drone.go index c12e3ef20..812db1f88 100644 --- a/cmd/drone/drone.go +++ b/cmd/drone/drone.go @@ -11,6 +11,7 @@ import ( "time" "github.com/drone/drone/pkg/build" + "github.com/drone/drone/pkg/build/docker" "github.com/drone/drone/pkg/build/log" "github.com/drone/drone/pkg/build/repo" "github.com/drone/drone/pkg/build/script" @@ -174,7 +175,7 @@ func run(path string) { // loop through and create builders for _, b := range builds { //script.Builds { - builder := build.New() + builder := build.New(docker.DefaultClient) builder.Build = b builder.Repo = &code builder.Key = key diff --git a/pkg/build/build.go b/pkg/build/build.go index ba8d39691..f5e73ecf8 100644 --- a/pkg/build/build.go +++ b/pkg/build/build.go @@ -19,9 +19,6 @@ import ( "github.com/drone/drone/pkg/build/script" ) -// instance of the Docker client -var client = docker.New() - // BuildState stores information about a build // process including the Exit status and various // Runtime statistics (coming soon). @@ -35,8 +32,10 @@ type BuildState struct { // Max RAM, Max Swap, Disk space, and more. } -func New() *Builder { - return &Builder{} +func New(dockerClient *docker.Client) *Builder { + return &Builder{ + dockerClient: dockerClient, + } } // Builder represents a build process being prepared @@ -86,6 +85,8 @@ type Builder struct { // specified services and linked to // this build. services []*docker.Container + + dockerClient *docker.Client } func (b *Builder) Run() error { @@ -182,19 +183,19 @@ func (b *Builder) setup() error { log.Infof("starting service container %s", image.Tag) // Run the contianer - run, err := client.Containers.RunDaemonPorts(image.Tag, image.Ports...) + run, err := b.dockerClient.Containers.RunDaemonPorts(image.Tag, image.Ports...) if err != nil { return err } // Get the container info - info, err := client.Containers.Inspect(run.ID) + info, err := b.dockerClient.Containers.Inspect(run.ID) if err != nil { // on error kill the container since it hasn't yet been // added to the array and would therefore not get // removed in the defer statement. - client.Containers.Stop(run.ID, 10) - client.Containers.Remove(run.ID) + b.dockerClient.Containers.Stop(run.ID, 10) + b.dockerClient.Containers.Remove(run.ID) return err } @@ -224,16 +225,16 @@ func (b *Builder) setup() error { // check for build container (ie bradrydzewski/go:1.2) // and download if it doesn't already exist - if _, err := client.Images.Inspect(b.Build.Image); err == docker.ErrNotFound { + if _, err := b.dockerClient.Images.Inspect(b.Build.Image); err == docker.ErrNotFound { // download the image if it doesn't exist - if err := client.Images.Pull(b.Build.Image); err != nil { + if err := b.dockerClient.Images.Pull(b.Build.Image); err != nil { return err } } // create the Docker image id := createUID() - if err := client.Images.Build(id, dir); err != nil { + if err := b.dockerClient.Images.Build(id, dir); err != nil { return err } @@ -241,11 +242,11 @@ func (b *Builder) setup() error { log.Infof("copying repository to %s", b.Repo.Dir) // get the image details - b.image, err = client.Images.Inspect(id) + b.image, err = b.dockerClient.Images.Inspect(id) if err != nil { // if we have problems with the image make sure // we remove it before we exit - client.Images.Remove(id) + b.dockerClient.Images.Remove(id) return err } @@ -264,10 +265,10 @@ func (b *Builder) teardown() error { log.Info("removing build container") // stop the container, ignore error message - client.Containers.Stop(b.container.ID, 15) + b.dockerClient.Containers.Stop(b.container.ID, 15) // remove the container, ignore error message - if err := client.Containers.Remove(b.container.ID); err != nil { + if err := b.dockerClient.Containers.Remove(b.container.ID); err != nil { log.Errf("failed to delete build container %s", b.container.ID) } } @@ -278,10 +279,10 @@ func (b *Builder) teardown() error { log.Infof("removing service container %s", b.Build.Services[i]) // stop the service container, ignore the error - client.Containers.Stop(container.ID, 15) + b.dockerClient.Containers.Stop(container.ID, 15) // remove the service container, ignore the error - if err := client.Containers.Remove(container.ID); err != nil { + if err := b.dockerClient.Containers.Remove(container.ID); err != nil { log.Errf("failed to delete service container %s", container.ID) } } @@ -291,7 +292,7 @@ func (b *Builder) teardown() error { // debugging log.Info("removing build image") - if _, err := client.Images.Remove(b.image.ID); err != nil { + if _, err := b.dockerClient.Images.Remove(b.image.ID); err != nil { log.Errf("failed to completely delete build image %s. %s", b.image.ID, err.Error()) } } @@ -326,7 +327,7 @@ func (b *Builder) run() error { } // create the container from the image - run, err := client.Containers.Create(&conf) + run, err := b.dockerClient.Containers.Create(&conf) if err != nil { return err } @@ -336,18 +337,18 @@ func (b *Builder) run() error { // attach to the container go func() { - client.Containers.Attach(run.ID, &writer{b.Stdout}) + b.dockerClient.Containers.Attach(run.ID, &writer{b.Stdout}) }() // start the container - if err := client.Containers.Start(run.ID, &host); err != nil { + if err := b.dockerClient.Containers.Start(run.ID, &host); err != nil { b.BuildState.ExitCode = 1 b.BuildState.Finished = time.Now().UTC().Unix() return err } // wait for the container to stop - wait, err := client.Containers.Wait(run.ID) + wait, err := b.dockerClient.Containers.Wait(run.ID) if err != nil { b.BuildState.ExitCode = 1 b.BuildState.Finished = time.Now().UTC().Unix() diff --git a/pkg/build/docker/client.go b/pkg/build/docker/client.go index bb33a28fb..123e52468 100644 --- a/pkg/build/docker/client.go +++ b/pkg/build/docker/client.go @@ -29,6 +29,8 @@ const ( // Enables verbose logging to the Terminal window var Logging = true +var DefaultClient = New() // TEMPORARY; PLEASE CONSTRUCT/INJECT YOURSELF + // New creates an instance of the Docker Client func New() *Client { c := &Client{} diff --git a/pkg/queue/worker.go b/pkg/queue/worker.go index e2cf8f6d8..cb27d6a8b 100644 --- a/pkg/queue/worker.go +++ b/pkg/queue/worker.go @@ -4,6 +4,7 @@ import ( "bytes" "fmt" "github.com/drone/drone/pkg/build" + "github.com/drone/drone/pkg/build/docker" "github.com/drone/drone/pkg/build/git" r "github.com/drone/drone/pkg/build/repo" "github.com/drone/drone/pkg/channel" @@ -167,7 +168,7 @@ func (w *worker) execute(task *BuildTask) error { } func runBuild(b *BuildTask, buf io.Writer) (bool, error) { - builder := build.New() + builder := build.New(docker.DefaultClient) builder.Build = b.Script builder.Repo = &r.Repo{Path: b.Repo.URL, Branch: b.Commit.Branch, Commit: b.Commit.Hash, PR: b.Commit.PullRequest, Dir: filepath.Join("/var/cache/drone/src", b.Repo.Slug), Depth: git.GitDepth(b.Script.Git)} builder.Key = []byte(b.Repo.PrivateKey) From ddc8e7a56f54d793bb01957194726859c033592d Mon Sep 17 00:00:00 2001 From: Alex Suraci Date: Mon, 24 Feb 2014 17:32:40 -0800 Subject: [PATCH 4/5] split build construction out of worker (todo: inject runner into worker) Signed-off-by: Abhijit Hiremagalur --- pkg/queue/worker.go | 49 +++++++++++++++++++++++++++++++++++++-------- 1 file changed, 41 insertions(+), 8 deletions(-) diff --git a/pkg/queue/worker.go b/pkg/queue/worker.go index cb27d6a8b..5103de056 100644 --- a/pkg/queue/worker.go +++ b/pkg/queue/worker.go @@ -7,6 +7,7 @@ import ( "github.com/drone/drone/pkg/build/docker" "github.com/drone/drone/pkg/build/git" r "github.com/drone/drone/pkg/build/repo" + "github.com/drone/drone/pkg/build/script" "github.com/drone/drone/pkg/channel" "github.com/drone/drone/pkg/database" . "github.com/drone/drone/pkg/model" @@ -167,17 +168,49 @@ func (w *worker) execute(task *BuildTask) error { return nil } +type runner struct { + dockerClient *docker.Client + timeout time.Duration +} + +func (r *runner) Build(buildScript *script.Build, repo *r.Repo, key []byte, buildOutput io.Writer) (bool, error) { + builder := build.New(r.dockerClient) + builder.Build = buildScript + builder.Repo = repo + builder.Key = key + builder.Stdout = buildOutput + builder.Timeout = r.timeout + + err := builder.Run() + + return builder.BuildState == nil || builder.BuildState.ExitCode != 0, err +} + +func newRunner(dockerClient *docker.Client, timeout time.Duration) *runner { + return &runner{ + dockerClient: dockerClient, + timeout: timeout, + } +} + func runBuild(b *BuildTask, buf io.Writer) (bool, error) { - builder := build.New(docker.DefaultClient) - builder.Build = b.Script - builder.Repo = &r.Repo{Path: b.Repo.URL, Branch: b.Commit.Branch, Commit: b.Commit.Hash, PR: b.Commit.PullRequest, Dir: filepath.Join("/var/cache/drone/src", b.Repo.Slug), Depth: git.GitDepth(b.Script.Git)} - builder.Key = []byte(b.Repo.PrivateKey) - builder.Stdout = buf - builder.Timeout = 300 * time.Minute + runner := newRunner(docker.DefaultClient, 300*time.Minute) - buildErr := builder.Run() + repo := &r.Repo{ + Path: b.Repo.URL, + Branch: b.Commit.Branch, + Commit: b.Commit.Hash, + PR: b.Commit.PullRequest, + Dir: filepath.Join("/var/cache/drone/src", b.Repo.Slug), + Depth: git.GitDepth(b.Script.Git), + } - return builder.BuildState == nil || builder.BuildState.ExitCode != 0, buildErr + return runner.Build( + b.Script, + repo, + []byte(b.Repo.PrivateKey), + buf, + ) } // updateGitHubStatus is a helper function that will send From 4b52fcad1a1cac0774b0cfe6ea56a6b343daabba Mon Sep 17 00:00:00 2001 From: Alex Suraci Date: Mon, 24 Feb 2014 17:51:25 -0800 Subject: [PATCH 5/5] provide runner to workers, use it for all builds Signed-off-by: Abhijit Hiremagalur --- pkg/queue/queue.go | 11 ++++++--- pkg/queue/runner.go | 40 ++++++++++++++++++++++++++++++++ pkg/queue/worker.go | 56 ++++++++++++--------------------------------- 3 files changed, 62 insertions(+), 45 deletions(-) create mode 100644 pkg/queue/runner.go diff --git a/pkg/queue/queue.go b/pkg/queue/queue.go index 2b69f33c9..057a2c5aa 100644 --- a/pkg/queue/queue.go +++ b/pkg/queue/queue.go @@ -1,9 +1,11 @@ package queue import ( + "github.com/drone/drone/pkg/build/docker" "github.com/drone/drone/pkg/build/script" . "github.com/drone/drone/pkg/model" "runtime" + "time" ) // A Queue dispatches tasks to workers. @@ -23,11 +25,11 @@ type BuildTask struct { Script *script.Build } -var defaultQueue = Start(runtime.NumCPU()) // TEMPORARY; INJECT PLEASE +var defaultQueue = Start(runtime.NumCPU(), newRunner(docker.DefaultClient, 300*time.Second)) // TEMPORARY; INJECT PLEASE var Add = defaultQueue.Add // TEMPORARY; INJECT PLEASE -func Start(workers int) *Queue { +func Start(workers int, runner Runner) *Queue { // get the number of CPUs. Since builds // tend to be CPU-intensive we should only // execute 1 build per CPU. @@ -42,7 +44,10 @@ func Start(workers int) *Queue { // spawn a worker for each CPU for i := 0; i < workers; i++ { - worker := worker{} + worker := worker{ + runner: runner, + } + go worker.work(tasks) } diff --git a/pkg/queue/runner.go b/pkg/queue/runner.go new file mode 100644 index 000000000..8bb0a7f0b --- /dev/null +++ b/pkg/queue/runner.go @@ -0,0 +1,40 @@ +package queue + +import ( + "io" + "time" + + "github.com/drone/drone/pkg/build" + "github.com/drone/drone/pkg/build/docker" + "github.com/drone/drone/pkg/build/repo" + "github.com/drone/drone/pkg/build/script" +) + +type Runner interface { + Run(buildScript *script.Build, repo *repo.Repo, key []byte, buildOutput io.Writer) (success bool, err error) +} + +type runner struct { + dockerClient *docker.Client + timeout time.Duration +} + +func newRunner(dockerClient *docker.Client, timeout time.Duration) *runner { + return &runner{ + dockerClient: dockerClient, + timeout: timeout, + } +} + +func (r *runner) Run(buildScript *script.Build, repo *repo.Repo, key []byte, buildOutput io.Writer) (bool, error) { + builder := build.New(r.dockerClient) + builder.Build = buildScript + builder.Repo = repo + builder.Key = key + builder.Stdout = buildOutput + builder.Timeout = r.timeout + + err := builder.Run() + + return builder.BuildState == nil || builder.BuildState.ExitCode != 0, err +} diff --git a/pkg/queue/worker.go b/pkg/queue/worker.go index 5103de056..a43071091 100644 --- a/pkg/queue/worker.go +++ b/pkg/queue/worker.go @@ -3,11 +3,8 @@ package queue import ( "bytes" "fmt" - "github.com/drone/drone/pkg/build" - "github.com/drone/drone/pkg/build/docker" "github.com/drone/drone/pkg/build/git" r "github.com/drone/drone/pkg/build/repo" - "github.com/drone/drone/pkg/build/script" "github.com/drone/drone/pkg/channel" "github.com/drone/drone/pkg/database" . "github.com/drone/drone/pkg/model" @@ -19,7 +16,9 @@ import ( "time" ) -type worker struct{} +type worker struct { + runner Runner +} // work is a function that will infinitely // run in the background waiting for tasks that @@ -125,7 +124,7 @@ func (w *worker) execute(task *BuildTask) error { }() // execute the build - passed, buildErr := runBuild(task, buf) //w.builder.Build(script, repo, task.Repo.PrivateKey, buf) + passed, buildErr := w.runBuild(task, buf) task.Build.Finished = time.Now().UTC() task.Commit.Finished = time.Now().UTC() @@ -168,47 +167,20 @@ func (w *worker) execute(task *BuildTask) error { return nil } -type runner struct { - dockerClient *docker.Client - timeout time.Duration -} - -func (r *runner) Build(buildScript *script.Build, repo *r.Repo, key []byte, buildOutput io.Writer) (bool, error) { - builder := build.New(r.dockerClient) - builder.Build = buildScript - builder.Repo = repo - builder.Key = key - builder.Stdout = buildOutput - builder.Timeout = r.timeout - - err := builder.Run() - - return builder.BuildState == nil || builder.BuildState.ExitCode != 0, err -} - -func newRunner(dockerClient *docker.Client, timeout time.Duration) *runner { - return &runner{ - dockerClient: dockerClient, - timeout: timeout, - } -} - -func runBuild(b *BuildTask, buf io.Writer) (bool, error) { - runner := newRunner(docker.DefaultClient, 300*time.Minute) - +func (w *worker) runBuild(task *BuildTask, buf io.Writer) (bool, error) { repo := &r.Repo{ - Path: b.Repo.URL, - Branch: b.Commit.Branch, - Commit: b.Commit.Hash, - PR: b.Commit.PullRequest, - Dir: filepath.Join("/var/cache/drone/src", b.Repo.Slug), - Depth: git.GitDepth(b.Script.Git), + Path: task.Repo.URL, + Branch: task.Commit.Branch, + Commit: task.Commit.Hash, + PR: task.Commit.PullRequest, + Dir: filepath.Join("/var/cache/drone/src", task.Repo.Slug), + Depth: git.GitDepth(task.Script.Git), } - return runner.Build( - b.Script, + return w.runner.Run( + task.Script, repo, - []byte(b.Repo.PrivateKey), + []byte(task.Repo.PrivateKey), buf, ) }