Merge pull request #1719 from bradrydzewski/master

finish streaming logs before killing images

build/pipeline.go

@@ -3,6 +3,7 @@ package build
 import (
 	"bufio"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/Sirupsen/logrus"
@@ -20,6 +21,7 @@ type Pipeline struct {
 	conf *yaml.Config
 	head *element
 	tail *element
+	wait sync.WaitGroup
 	pipe chan (*Line)
 	next chan (error)
 	done chan (error)
@@ -87,6 +89,11 @@ func (p *Pipeline) Tail() *yaml.Container {
 // Stop stops the pipeline.
 func (p *Pipeline) Stop() {
 	go func() {
+		defer func() {
+			if r := recover(); r != nil {
+				logrus.Errorln("recover stopping the pipeline", r)
+			}
+		}()
 		p.done <- ErrTerm
 	}()
 }
@@ -98,9 +105,11 @@ func (p *Pipeline) Setup() error {
 
 // Teardown removes the pipeline environment.
 func (p *Pipeline) Teardown() {
+
 	for _, id := range p.containers {
 		p.engine.ContainerRemove(id)
 	}
+
 	close(p.next)
 	close(p.done)
 
@@ -114,10 +123,32 @@ func (p *Pipeline) Teardown() {
 func (p *Pipeline) step() {
 	if p.head == p.tail {
 		go func() {
+			defer func() {
+				if r := recover(); r != nil {
+					logrus.Errorln("recover executing step function", r)
+				}
+			}()
+
+			// stop all containers
+			for _, id := range p.containers {
+				p.engine.ContainerStop(id)
+			}
+
+			// wait for all logs to terminate
+			// p.wait.Done() // this is for the ambassador
+			p.wait.Wait()
+
+			// signal completion
 			p.done <- nil
 		}()
 	} else {
 		go func() {
+			defer func() {
+				if r := recover(); r != nil {
+					logrus.Errorln("recover executing step to head function", r)
+				}
+			}()
+
 			p.head = p.head.next
 			p.next <- nil
 		}()
@@ -137,17 +168,23 @@ func (p *Pipeline) close(err error) {
 }
 
 func (p *Pipeline) exec(c *yaml.Container) error {
+
 	name, err := p.engine.ContainerStart(c)
 	if err != nil {
 		return err
 	}
 	p.containers = append(p.containers, name)
 
+	logrus.Debugf("wait.add(1) for %s logs", name)
+	p.wait.Add(1)
 	go func() {
 		defer func() {
 			if r := recover(); r != nil {
 				logrus.Errorln("recover writing build output", r)
 			}
+
+			logrus.Debugf("wait.done() for %s logs", name)
+			p.wait.Done()
 		}()
 
 		rc, rerr := p.engine.ContainerLogs(name)
@@ -179,17 +216,16 @@ func (p *Pipeline) exec(c *yaml.Container) error {
 	if err != nil {
 		return err
 	}
-	if state.OOMKilled {
-		return &OomError{c.Name}
-	} else if state.ExitCode != 0 {
-		return &ExitError{c.Name, state.ExitCode}
-	}
 
+	logrus.Debugf("wait.add(1) for %s exit code", name)
+	p.wait.Add(1)
 	go func() {
 		defer func() {
 			if r := recover(); r != nil {
 				logrus.Errorln("recover writing exit code to output", r)
 			}
+			p.wait.Done()
+			logrus.Debugf("wait.done() for %s exit code", name)
 		}()
 
 		p.pipe <- &Line{
@@ -198,5 +234,12 @@ func (p *Pipeline) exec(c *yaml.Container) error {
 			Out:  strconv.Itoa(state.ExitCode),
 		}
 	}()
+
+	if state.OOMKilled {
+		return &OomError{c.Name}
+	} else if state.ExitCode != 0 {
+		return &ExitError{c.Name, state.ExitCode}
+	}
+
 	return nil
 }

drone/build_start.go

@@ -43,9 +43,9 @@ func buildStart(c *cli.Context) (err error) {
 
 	var build *model.Build
 	if c.Bool("fork") {
-		build, err = client.BuildStart(owner, name, number)
-	} else {
 		build, err = client.BuildFork(owner, name, number)
+	} else {
+		build, err = client.BuildStart(owner, name, number)
 	}
 	if err != nil {
 		return err

drone/server.go

@@ -1,9 +1,7 @@
 package main
 
 import (
-	"fmt"
 	"net/http"
-	"os"
 	"time"
 
 	"github.com/drone/drone/router"
@@ -263,30 +261,11 @@ var serverCmd = cli.Command{
 			Name:   "stash-skip-verify",
 			Usage:  "stash skip ssl verification",
 		},
-		//
-		// remove these eventually
-		//
-
-		cli.BoolFlag{
-			Name:   "agreement.ack",
-			EnvVar: "I_UNDERSTAND_I_AM_USING_AN_UNSTABLE_VERSION",
-			Usage:  "agree to terms of use.",
-		},
-		cli.BoolFlag{
-			Name:   "agreement.fix",
-			EnvVar: "I_AGREE_TO_FIX_BUGS_AND_NOT_FILE_BUGS",
-			Usage:  "agree to terms of use.",
-		},
 	},
 }
 
 func server(c *cli.Context) error {
 
-	if c.Bool("agreement.ack") == false || c.Bool("agreement.fix") == false {
-		fmt.Println(agreement)
-		os.Exit(1)
-	}
-
 	// debug level if requested by user
 	if c.Bool("debug") {
 		logrus.SetLevel(logrus.DebugLevel)
@@ -324,28 +303,3 @@ func server(c *cli.Context) error {
 		handler,
 	)
 }
-
-var agreement = `
----
-
-
-You are attempting to use the unstable channel. This build is experimental and
-has known bugs and compatibility issues. It is not intended for general use.
-
-Please consider using the latest stable release instead:
-
-		drone/drone:0.4.2
-
-If you are attempting to build from source please use the latest stable tag:
-
-		v0.4.2
-
-If you are interested in testing this experimental build AND assisting with
-development you may proceed by setting the following environment:
-
-		I_UNDERSTAND_I_AM_USING_AN_UNSTABLE_VERSION=true
-		I_AGREE_TO_FIX_BUGS_AND_NOT_FILE_BUGS=true
-
-
----
-`

server/build.go

@@ -135,6 +135,19 @@ func DeleteBuild(c *gin.Context) {
 		return
 	}
 
+	if job.Status != model.StatusRunning {
+		c.String(400, "Cannot cancel a non-running build")
+		return
+	}
+
+	job.Status = model.StatusKilled
+	job.Finished = time.Now().Unix()
+	if job.Started == 0 {
+		job.Started = job.Finished
+	}
+	job.ExitCode = 137
+	store.UpdateBuildJob(c, build, job)
+
 	bus.Publish(c, bus.NewEvent(bus.Cancelled, repo, build, job))
 	c.String(204, "")
 }
@@ -242,6 +255,7 @@ func PostBuild(c *gin.Context) {
 	build.Finished = 0
 	build.Enqueued = time.Now().UTC().Unix()
 	for _, job := range jobs {
+		job.Error = ""
 		job.Status = model.StatusPending
 		job.Started = 0
 		job.Finished = 0