ability to stream builds

drone.go

@@ -141,6 +141,7 @@ func main() {
 	{
 		stream.Use(server.SetRepo())
 		stream.Use(server.SetPerm())
+		stream.Use(server.SetRunner(&runner_))
 		stream.GET("/:owner/:name", server.GetRepoEvents)
 		stream.GET("/:owner/:name/:build/:number", server.GetStream)
 

runner/builtin/copy.go

@@ -0,0 +1,124 @@
+package builtin
+
+import (
+	"encoding/binary"
+	"errors"
+	"io"
+)
+
+const (
+	StdWriterPrefixLen = 8
+	StdWriterFdIndex   = 0
+	StdWriterSizeIndex = 4
+)
+
+type StdType [StdWriterPrefixLen]byte
+
+var (
+	Stdin  StdType = StdType{0: 0}
+	Stdout StdType = StdType{0: 1}
+	Stderr StdType = StdType{0: 2}
+)
+
+type StdWriter struct {
+	io.Writer
+	prefix  StdType
+	sizeBuf []byte
+}
+
+var ErrInvalidStdHeader = errors.New("Unrecognized input header")
+
+// StdCopy is a modified version of io.Copy.
+//
+// StdCopy will demultiplex `src`, assuming that it contains two streams,
+// previously multiplexed together using a StdWriter instance.
+// As it reads from `src`, StdCopy will write to `dstout` and `dsterr`.
+//
+// StdCopy will read until it hits EOF on `src`. It will then return a nil error.
+// In other words: if `err` is non nil, it indicates a real underlying error.
+//
+// `written` will hold the total number of bytes written to `dstout` and `dsterr`.
+func StdCopy(dstout, dsterr io.Writer, src io.Reader) (written int64, err error) {
+	var (
+		buf       = make([]byte, 32*1024+StdWriterPrefixLen+1)
+		bufLen    = len(buf)
+		nr, nw    int
+		er, ew    error
+		out       io.Writer
+		frameSize int
+	)
+
+	for {
+		// Make sure we have at least a full header
+		for nr < StdWriterPrefixLen {
+			var nr2 int
+			nr2, er = src.Read(buf[nr:])
+			nr += nr2
+			if er == io.EOF {
+				if nr < StdWriterPrefixLen {
+					return written, nil
+				}
+				break
+			}
+			if er != nil {
+				return 0, er
+			}
+		}
+
+		// Check the first byte to know where to write
+		switch buf[StdWriterFdIndex] {
+		case 0:
+			fallthrough
+		case 1:
+			// Write on stdout
+			out = dstout
+		case 2:
+			// Write on stderr
+			out = dsterr
+		default:
+			return 0, ErrInvalidStdHeader
+		}
+
+		// Retrieve the size of the frame
+		frameSize = int(binary.BigEndian.Uint32(buf[StdWriterSizeIndex : StdWriterSizeIndex+4]))
+
+		// Check if the buffer is big enough to read the frame.
+		// Extend it if necessary.
+		if frameSize+StdWriterPrefixLen > bufLen {
+			buf = append(buf, make([]byte, frameSize+StdWriterPrefixLen-bufLen+1)...)
+			bufLen = len(buf)
+		}
+
+		// While the amount of bytes read is less than the size of the frame + header, we keep reading
+		for nr < frameSize+StdWriterPrefixLen {
+			var nr2 int
+			nr2, er = src.Read(buf[nr:])
+			nr += nr2
+			if er == io.EOF {
+				if nr < frameSize+StdWriterPrefixLen {
+					return written, nil
+				}
+				break
+			}
+			if er != nil {
+				return 0, er
+			}
+		}
+
+		// Write the retrieved frame (without header)
+		nw, ew = out.Write(buf[StdWriterPrefixLen : frameSize+StdWriterPrefixLen])
+		if ew != nil {
+			return 0, ew
+		}
+		// If the frame has not been fully written: error
+		if nw != frameSize {
+			return 0, io.ErrShortWrite
+		}
+		written += int64(nw)
+
+		// Move the rest of the buffer to the beginning
+		copy(buf, buf[frameSize+StdWriterPrefixLen:])
+		// Move the index
+		nr -= frameSize + StdWriterPrefixLen
+	}
+}

runner/builtin/runner.go

@@ -140,15 +140,18 @@ func (r *Runner) Run(w *queue.Work) error {
 		}
 
 		// send the logs to the datastore
+		var buf bytes.Buffer
 		rc, err := worker.Logs()
 		if err != nil && builderr != nil {
-			var buf bytes.Buffer
 			buf.WriteString(builderr.Error())
-			rc = ioutil.NopCloser(&buf)
 		} else if err != nil {
+			buf.WriteString(err.Error())
 			return err
+		} else {
+			defer rc.Close()
+			StdCopy(&buf, &buf, rc)
 		}
-		err = r.SetLogs(w.Repo, w.Build, task, rc)
+		err = r.SetLogs(w.Repo, w.Build, task, ioutil.NopCloser(&buf))
 		if err != nil {
 			return err
 		}
@@ -210,7 +213,39 @@ func (r *Runner) Logs(repo string, build, task int) (io.ReadCloser, error) {
 	if err != nil {
 		return nil, err
 	}
-	return client.ContainerLogs(cname(repo, build, task), logOptsTail)
+	// make sure this container actually exists
+	info, err := client.InspectContainer(cname(repo, build, task))
+	if err != nil {
+		return nil, err
+	}
+	// verify the container is running. if not we'll
+	// do an exponential backoff and attempt to wait
+	if !info.State.Running {
+		for i := 0; ; i++ {
+			time.Sleep(1 * time.Second)
+			info, err = client.InspectContainer(info.Id)
+			if err != nil {
+				return nil, err
+			}
+			if info.State.Running {
+				break
+			}
+			if i == 5 {
+				return nil, dockerclient.ErrNotFound
+			}
+		}
+	}
+
+	rc, err := client.ContainerLogs(info.Id, logOptsTail)
+	if err != nil {
+		return nil, err
+	}
+	pr, pw := io.Pipe()
+	go func() {
+		defer rc.Close()
+		StdCopy(pw, pw, rc)
+	}()
+	return pr, nil
 }
 
 func cname(repo string, number, task int) string {

runner/builtin/updater.go

@@ -33,6 +33,17 @@ func (u *updater) SetBuild(r *common.Repo, b *common.Build) error {
 		return err
 	}
 
+	// if the build is complete we may need to update
+	if b.State != common.StatePending && b.State != common.StateRunning {
+		repo, err := u.store.Repo(r.FullName)
+		if err == nil {
+			if repo.Last == nil || b.Number >= repo.Last.Number {
+				repo.Last = b
+				u.store.SetRepo(repo)
+			}
+		}
+	}
+
 	msg, err := json.Marshal(b)
 	if err != nil {
 		return err

runner/builtin/worker.go

@@ -33,10 +33,10 @@ var (
 
 var (
 	// name of the build agent container.
-	DefaultAgent = "drone/drone-build"
+	DefaultAgent = "drone/drone-build:latest"
 
 	// default name of the build agent executable
-	DefaultEntrypoint = []string{"/bin/drone-build"}
+	DefaultEntrypoint = []string{"/go/bin/drone-build"}
 
 	// default argument to invoke build steps
 	DefaultBuildArgs = []string{"--build", "--clone", "--publish", "--deploy"}
@@ -98,7 +98,7 @@ func (w *worker) Build(name string, stdin []byte) (_ int, err error) {
 
 	w.build, err = run(w.client, conf, name, w.timeout)
 	if err != nil {
-		return
+		return 1, err
 	}
 	return w.build.State.ExitCode, err
 }

server/server.go

@@ -11,6 +11,7 @@ import (
 	"github.com/drone/drone/eventbus"
 	"github.com/drone/drone/queue"
 	"github.com/drone/drone/remote"
+	"github.com/drone/drone/runner"
 	"github.com/drone/drone/server/session"
 	"github.com/drone/drone/settings"
 )
@@ -60,6 +61,21 @@ func SetRemote(r remote.Remote) gin.HandlerFunc {
 	}
 }
 
+func ToRunner(c *gin.Context) runner.Runner {
+	v, err := c.Get("runner")
+	if err != nil {
+		return nil
+	}
+	return v.(runner.Runner)
+}
+
+func SetRunner(r runner.Runner) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		c.Set("runner", r)
+		c.Next()
+	}
+}
+
 func ToSettings(c *gin.Context) *settings.Settings {
 	v, err := c.Get("settings")
 	if err != nil {

server/static/scripts/controllers/builds.js

@@ -87,7 +87,7 @@
 
 			// subscribes to the build otuput.
 			logs.subscribe(fullName, number, step, function(data){
-				term.innerHTML += convert.toHtml(data)+"\n";
+				term.innerHTML += convert.toHtml(data);
 				if (tail) {
 					// scrolls to the bottom of the page if enabled
 					$window.scrollTo(0, $window.document.body.scrollHeight);

server/ws.go

@@ -1,9 +1,8 @@
 package server
 
 import (
-	"fmt"
+	"bufio"
 	"net/http"
-	"net/url"
 	"strconv"
 	"time"
 
@@ -12,7 +11,8 @@ import (
 	log "github.com/Sirupsen/logrus"
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
-	"github.com/koding/websocketproxy"
+
+	// "github.com/koding/websocketproxy"
 )
 
 const (
@@ -88,27 +88,91 @@ func GetRepoEvents(c *gin.Context) {
 }
 
 func GetStream(c *gin.Context) {
-	store := ToDatastore(c)
+	// store := ToDatastore(c)
 	repo := ToRepo(c)
+	runner := ToRunner(c)
 	build, _ := strconv.Atoi(c.Params.ByName("build"))
 	task, _ := strconv.Atoi(c.Params.ByName("number"))
 
-	agent, err := store.BuildAgent(repo.FullName, build)
+	// agent, err := store.BuildAgent(repo.FullName, build)
+	// if err != nil {
+	// 	c.Fail(404, err)
+	// 	return
+	// }
+
+	rc, err := runner.Logs(repo.FullName, build, task)
 	if err != nil {
 		c.Fail(404, err)
 		return
 	}
 
-	url_, err := url.Parse("ws://" + agent.Addr)
+	// upgrade the websocket
+	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
 	if err != nil {
-		c.Fail(500, err)
+		c.Fail(400, err)
 		return
 	}
-	url_.Path = fmt.Sprintf("/stream/%s/%v/%v", repo.FullName, build, task)
-	proxy := websocketproxy.NewProxy(url_)
-	proxy.ServeHTTP(c.Writer, c.Request)
 
-	log.Debugf("closed websocket")
+	var ticker = time.NewTicker(pingPeriod)
+	var out = make(chan []byte)
+	defer func() {
+		log.Infof("closed stdout websocket")
+		ticker.Stop()
+		rc.Close()
+		ws.Close()
+	}()
+
+	go func() {
+		for {
+			select {
+			case <-c.Writer.CloseNotify():
+				rc.Close()
+				ws.Close()
+				return
+			case line := <-out:
+				ws.WriteMessage(websocket.TextMessage, line)
+			case <-ticker.C:
+				ws.SetWriteDeadline(time.Now().Add(writeWait))
+				err := ws.WriteMessage(websocket.PingMessage, []byte{})
+				if err != nil {
+					rc.Close()
+					ws.Close()
+					return
+				}
+			}
+		}
+	}()
+
+	go func() {
+		rd := bufio.NewReader(rc)
+		for {
+			str, err := rd.ReadBytes('\n')
+
+			if err != nil {
+				break
+			}
+			if len(str) == 0 {
+				break
+			}
+
+			out <- str
+		}
+		rc.Close()
+		ws.Close()
+	}()
+
+	readWebsocket(ws)
+
+	// url_, err := url.Parse("ws://" + agent.Addr)
+	// if err != nil {
+	// 	c.Fail(500, err)
+	// 	return
+	// }
+	// url_.Path = fmt.Sprintf("/stream/%s/%v/%v", repo.FullName, build, task)
+	// proxy := websocketproxy.NewProxy(url_)
+	// proxy.ServeHTTP(c.Writer, c.Request)
+
+	// log.Debugf("closed websocket")
 }
 
 // readWebsocket will block while reading the websocket data