2016-05-02 21:21:25 +02:00
|
|
|
package server
|
2016-04-20 03:37:53 +02:00
|
|
|
|
|
|
|
import (
|
2016-09-26 07:39:28 +02:00
|
|
|
"bytes"
|
2016-04-20 03:37:53 +02:00
|
|
|
"fmt"
|
2016-05-23 23:35:58 +02:00
|
|
|
"net/http"
|
2016-09-26 10:29:05 +02:00
|
|
|
"strconv"
|
2016-09-28 02:33:13 +02:00
|
|
|
"time"
|
2016-09-26 07:39:28 +02:00
|
|
|
|
|
|
|
"golang.org/x/net/context"
|
2016-04-20 03:37:53 +02:00
|
|
|
|
|
|
|
"github.com/Sirupsen/logrus"
|
|
|
|
"github.com/drone/drone/model"
|
|
|
|
"github.com/drone/drone/remote"
|
|
|
|
"github.com/drone/drone/store"
|
2016-09-26 07:39:28 +02:00
|
|
|
"github.com/drone/mq/stomp"
|
2016-05-23 23:35:58 +02:00
|
|
|
"github.com/gorilla/websocket"
|
2016-04-20 03:37:53 +02:00
|
|
|
)
|
|
|
|
|
2016-09-26 07:39:28 +02:00
|
|
|
// newline defines a newline constant to separate lines in the build output
|
|
|
|
var newline = []byte{'\n'}
|
2016-04-20 03:37:53 +02:00
|
|
|
|
2016-09-26 07:39:28 +02:00
|
|
|
// upgrader defines the default behavior for upgrading the websocket.
|
|
|
|
var upgrader = websocket.Upgrader{
|
|
|
|
ReadBufferSize: 1024,
|
|
|
|
WriteBufferSize: 1024,
|
|
|
|
CheckOrigin: func(r *http.Request) bool {
|
|
|
|
return true
|
|
|
|
},
|
2016-04-20 03:37:53 +02:00
|
|
|
}
|
|
|
|
|
2016-09-26 07:39:28 +02:00
|
|
|
// HandleUpdate handles build updates from the agent and persists to the database.
|
|
|
|
func HandleUpdate(c context.Context, message *stomp.Message) {
|
|
|
|
defer func() {
|
|
|
|
message.Release()
|
|
|
|
if r := recover(); r != nil {
|
|
|
|
err := r.(error)
|
|
|
|
logrus.Errorf("Panic recover: broker update handler: %s", err)
|
2016-04-20 03:37:53 +02:00
|
|
|
}
|
2016-09-26 07:39:28 +02:00
|
|
|
}()
|
2016-04-20 03:37:53 +02:00
|
|
|
|
2016-09-28 03:30:28 +02:00
|
|
|
work := new(model.Work)
|
2016-09-26 07:39:28 +02:00
|
|
|
if err := message.Unmarshal(work); err != nil {
|
2016-04-20 03:37:53 +02:00
|
|
|
logrus.Errorf("Invalid input. %s", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
// TODO(bradrydzewski) it is really annoying that we have to do this lookup
|
|
|
|
// and I'd prefer not to. The reason we do this is because the Build and Job
|
|
|
|
// have fields that aren't serialized to json and would be reset to their
|
|
|
|
// empty values if we just saved what was coming in the http.Request body.
|
|
|
|
build, err := store.GetBuild(c, work.Build.ID)
|
|
|
|
if err != nil {
|
2016-09-26 07:39:28 +02:00
|
|
|
logrus.Errorf("Unable to find build. %s", err)
|
2016-04-20 03:37:53 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
job, err := store.GetJob(c, work.Job.ID)
|
|
|
|
if err != nil {
|
2016-09-26 07:39:28 +02:00
|
|
|
logrus.Errorf("Unable to find job. %s", err)
|
2016-04-20 03:37:53 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
build.Started = work.Build.Started
|
|
|
|
build.Finished = work.Build.Finished
|
|
|
|
build.Status = work.Build.Status
|
|
|
|
job.Started = work.Job.Started
|
|
|
|
job.Finished = work.Job.Finished
|
|
|
|
job.Status = work.Job.Status
|
|
|
|
job.ExitCode = work.Job.ExitCode
|
2016-05-11 02:03:24 +02:00
|
|
|
job.Error = work.Job.Error
|
2016-04-20 03:37:53 +02:00
|
|
|
|
2016-04-21 09:25:30 +02:00
|
|
|
if build.Status == model.StatusPending {
|
2016-06-14 22:07:05 +02:00
|
|
|
build.Started = work.Job.Started
|
2016-04-21 09:25:30 +02:00
|
|
|
build.Status = model.StatusRunning
|
|
|
|
store.UpdateBuild(c, build)
|
|
|
|
}
|
|
|
|
|
2016-05-23 23:35:58 +02:00
|
|
|
// if job.Status == model.StatusRunning {
|
|
|
|
// err := stream.Create(c, stream.ToKey(job.ID))
|
|
|
|
// if err != nil {
|
|
|
|
// logrus.Errorf("Unable to create stream. %s", err)
|
|
|
|
// }
|
|
|
|
// }
|
2016-04-23 02:35:32 +02:00
|
|
|
|
2016-04-20 03:37:53 +02:00
|
|
|
ok, err := store.UpdateBuildJob(c, build, job)
|
|
|
|
if err != nil {
|
2016-09-26 07:39:28 +02:00
|
|
|
logrus.Errorf("Unable to update job. %s", err)
|
2016-04-20 03:37:53 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-09-26 10:29:05 +02:00
|
|
|
if ok {
|
2016-04-20 03:37:53 +02:00
|
|
|
// 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.
|
2016-09-26 07:39:28 +02:00
|
|
|
user, uerr := store.GetUser(c, work.User.ID)
|
|
|
|
if uerr != nil {
|
|
|
|
logrus.Errorf("Unable to find user. %s", err)
|
2016-04-20 03:37:53 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
remote.Status(c, user, work.Repo, build,
|
|
|
|
fmt.Sprintf("%s/%s/%d", work.System.Link, work.Repo.FullName, work.Build.Number))
|
|
|
|
}
|
2016-04-21 09:25:30 +02:00
|
|
|
|
2016-09-26 10:29:05 +02:00
|
|
|
client := stomp.MustFromContext(c)
|
2016-09-28 03:30:28 +02:00
|
|
|
err = client.SendJSON("/topic/events", model.Event{
|
2016-09-29 23:23:26 +02:00
|
|
|
Type: func() model.EventType {
|
|
|
|
// HACK we don't even really care about the event type.
|
|
|
|
// so we should just simplify how events are triggered.
|
|
|
|
if job.Status == model.StatusRunning {
|
|
|
|
return model.Started
|
|
|
|
}
|
|
|
|
return model.Finished
|
|
|
|
}(),
|
2016-09-26 10:29:05 +02:00
|
|
|
Repo: *work.Repo,
|
|
|
|
Build: *build,
|
|
|
|
Job: *job,
|
|
|
|
},
|
|
|
|
stomp.WithHeader("repo", work.Repo.FullName),
|
|
|
|
stomp.WithHeader("private", strconv.FormatBool(work.Repo.IsPrivate)),
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
logrus.Errorf("Unable to publish to /topic/events. %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
if job.Status == model.StatusRunning {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-09-26 07:39:28 +02:00
|
|
|
var buf bytes.Buffer
|
|
|
|
var sub []byte
|
2016-05-23 23:35:58 +02:00
|
|
|
|
2016-09-26 07:39:28 +02:00
|
|
|
done := make(chan bool)
|
2016-09-26 10:29:05 +02:00
|
|
|
dest := fmt.Sprintf("/topic/logs.%d", job.ID)
|
2016-09-26 07:39:28 +02:00
|
|
|
sub, err = client.Subscribe(dest, stomp.HandlerFunc(func(m *stomp.Message) {
|
2016-09-26 10:29:05 +02:00
|
|
|
defer m.Release()
|
|
|
|
if m.Header.GetBool("eof") {
|
2016-09-26 07:39:28 +02:00
|
|
|
done <- true
|
2016-09-26 10:29:05 +02:00
|
|
|
return
|
2016-09-26 07:39:28 +02:00
|
|
|
}
|
|
|
|
buf.Write(m.Body)
|
|
|
|
buf.WriteByte('\n')
|
|
|
|
}))
|
2016-09-26 10:29:05 +02:00
|
|
|
|
2016-05-23 23:35:58 +02:00
|
|
|
if err != nil {
|
2016-09-26 07:39:28 +02:00
|
|
|
logrus.Errorf("Unable to read logs from broker. %s", err)
|
2016-05-23 23:35:58 +02:00
|
|
|
return
|
|
|
|
}
|
2016-09-28 02:33:13 +02:00
|
|
|
|
|
|
|
defer func() {
|
|
|
|
client.Unsubscribe(sub)
|
|
|
|
client.Send(dest, []byte{}, stomp.WithRetain("remove"))
|
|
|
|
}()
|
|
|
|
|
|
|
|
select {
|
|
|
|
case <-done:
|
|
|
|
case <-time.After(30 * time.Second):
|
|
|
|
logrus.Errorf("Unable to read logs from broker. Timeout. %s", err)
|
|
|
|
return
|
|
|
|
}
|
2016-05-23 23:35:58 +02:00
|
|
|
|
2016-09-26 07:39:28 +02:00
|
|
|
if err := store.WriteLog(c, job, &buf); err != nil {
|
|
|
|
logrus.Errorf("Unable to write logs to store. %s", err)
|
2016-05-23 23:35:58 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|