package engine import ( "bytes" "crypto/tls" "crypto/x509" "errors" "fmt" "io" "io/ioutil" "runtime" "time" log "github.com/Sirupsen/logrus" "github.com/docker/docker/pkg/stdcopy" "github.com/drone/drone/model" "github.com/drone/drone/shared/docker" "github.com/drone/drone/shared/envconfig" "github.com/drone/drone/store" "github.com/samalba/dockerclient" "golang.org/x/net/context" ) type Engine interface { Schedule(context.Context, *Task) Cancel(int64, int64, *model.Node) error Stream(int64, int64, *model.Node) (io.ReadCloser, error) Deallocate(*model.Node) Allocate(*model.Node) error Subscribe(chan *Event) Unsubscribe(chan *Event) } var ( // options to fetch the stdout and stderr logs logOpts = &dockerclient.LogOptions{ Stdout: true, Stderr: true, } // options to fetch the stdout and stderr logs // by tailing the output. logOptsTail = &dockerclient.LogOptions{ Follow: true, Stdout: true, Stderr: true, } // error when the system cannot find logs errLogging = errors.New("Logs not available") ) type engine struct { bus *eventbus updater *updater pool *pool envs []string } // Load creates a new build engine, loaded with registered nodes from the // database. The registered nodes are added to the pool of nodes to immediately // start accepting workloads. func Load(env envconfig.Env, s store.Store) Engine { engine := &engine{} engine.bus = newEventbus() engine.pool = newPool() engine.updater = &updater{engine.bus} // quick fix to propogate HTTP_PROXY variables // throughout the build environment. var proxyVars = []string{"HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy", "NO_PROXY", "no_proxy"} for _, proxyVar := range proxyVars { proxyVal := env.Get(proxyVar) if len(proxyVal) != 0 { engine.envs = append(engine.envs, proxyVar+"="+proxyVal) } } nodes, err := s.Nodes().GetList() if err != nil { log.Fatalf("failed to get nodes from database. %s", err) } for _, node := range nodes { engine.pool.allocate(node) log.Infof("registered docker daemon %s", node.Addr) } return engine } // Cancel cancels the job running on the specified Node. func (e *engine) Cancel(build, job int64, node *model.Node) error { client, err := newDockerClient(node.Addr, node.Cert, node.Key, node.CA) if err != nil { return err } id := fmt.Sprintf("drone_build_%d_job_%d", build, job) return client.StopContainer(id, 30) } // Stream streams the job output from the specified Node. func (e *engine) Stream(build, job int64, node *model.Node) (io.ReadCloser, error) { client, err := newDockerClient(node.Addr, node.Cert, node.Key, node.CA) if err != nil { log.Errorf("cannot create Docker client for node %s", node.Addr) return nil, err } id := fmt.Sprintf("drone_build_%d_job_%d", build, job) log.Debugf("streaming container logs %s", id) return client.ContainerLogs(id, logOptsTail) } // Subscribe subscribes the channel to all build events. func (e *engine) Subscribe(c chan *Event) { e.bus.subscribe(c) } // Unsubscribe unsubscribes the channel from all build events. func (e *engine) Unsubscribe(c chan *Event) { e.bus.unsubscribe(c) } func (e *engine) Allocate(node *model.Node) error { // run the full build! client, err := newDockerClient(node.Addr, node.Cert, node.Key, node.CA) if err != nil { log.Errorf("error creating docker client %s. %s.", node.Addr, err) return err } version, err := client.Version() if err != nil { log.Errorf("error connecting to docker daemon %s. %s.", node.Addr, err) return err } log.Infof("registered docker daemon %s running version %s", node.Addr, version.Version) e.pool.allocate(node) return nil } func (e *engine) Deallocate(n *model.Node) { nodes := e.pool.list() for _, node := range nodes { if node.ID == n.ID { log.Infof("un-registered docker daemon %s", node.Addr) e.pool.deallocate(node) break } } } func (e *engine) Schedule(c context.Context, req *Task) { node := <-e.pool.reserve() // since we are probably running in a go-routine // make sure we recover from any panics so that // a bug doesn't crash the whole system. defer func() { if err := recover(); err != nil { const size = 64 << 10 buf := make([]byte, size) buf = buf[:runtime.Stack(buf, false)] log.Errorf("panic running build: %v\n%s", err, string(buf)) } e.pool.release(node) }() // update the node that was allocated to each job func(id int64) { for _, job := range req.Jobs { job.NodeID = id store.UpdateJob(c, job) } }(node.ID) // run the full build! client, err := newDockerClient(node.Addr, node.Cert, node.Key, node.CA) if err != nil { log.Errorln("error creating docker client", err) } // update the build state if any of the sub-tasks // had a non-success status req.Build.Started = time.Now().UTC().Unix() req.Build.Status = model.StatusRunning e.updater.SetBuild(c, req) // run all bulid jobs for _, job := range req.Jobs { req.Job = job e.runJob(c, req, e.updater, client) } // update overall status based on each job req.Build.Status = model.StatusSuccess for _, job := range req.Jobs { if job.Status != model.StatusSuccess { req.Build.Status = job.Status break } } req.Build.Finished = time.Now().UTC().Unix() err = e.updater.SetBuild(c, req) if err != nil { log.Errorf("error updating build completion status. %s", err) } // run notifications err = e.runJobNotify(req, client) if err != nil { log.Errorf("error executing notification step. %s", err) } } func newDockerClient(addr, cert, key, ca string) (dockerclient.Client, error) { var tlc *tls.Config // create the Docket client TLS config if len(cert) != 0 { pem, err := tls.X509KeyPair([]byte(cert), []byte(key)) if err != nil { log.Errorf("error loading X509 key pair. %s.", err) return dockerclient.NewDockerClient(addr, nil) } // create the TLS configuration for secure // docker communications. tlc = &tls.Config{} tlc.Certificates = []tls.Certificate{pem} // use the certificate authority if provided. // else don't use a certificate authority and set // skip verify to true if len(ca) != 0 { log.Infof("creating docker client %s with CA", addr) pool := x509.NewCertPool() pool.AppendCertsFromPEM([]byte(ca)) tlc.RootCAs = pool } else { log.Infof("creating docker client %s WITHOUT CA", addr) tlc.InsecureSkipVerify = true } } // create the Docker client. In this version of Drone (alpha) // we do not spread builds across clients, but this can and // (probably) will change in the future. return dockerclient.NewDockerClient(addr, tlc) } func (e *engine) runJob(c context.Context, r *Task, updater *updater, client dockerclient.Client) error { name := fmt.Sprintf("drone_build_%d_job_%d", r.Build.ID, r.Job.ID) defer func() { if r.Job.Status == model.StatusRunning { r.Job.Status = model.StatusError r.Job.Finished = time.Now().UTC().Unix() r.Job.ExitCode = 255 } if r.Job.Status == model.StatusPending { r.Job.Status = model.StatusError r.Job.Started = time.Now().UTC().Unix() r.Job.Finished = time.Now().UTC().Unix() r.Job.ExitCode = 255 } updater.SetJob(c, r) client.KillContainer(name, "9") client.RemoveContainer(name, true, true) }() // marks the task as running r.Job.Status = model.StatusRunning r.Job.Started = time.Now().UTC().Unix() // encode the build payload to write to stdin // when launching the build container in, err := encodeToLegacyFormat(r) if err != nil { log.Errorf("failure to marshal work. %s", err) return err } // CREATE AND START BUILD args := DefaultBuildArgs if r.Build.Event == model.EventPull { args = DefaultPullRequestArgs } args = append(args, "--") args = append(args, string(in)) conf := &dockerclient.ContainerConfig{ Image: DefaultAgent, Entrypoint: DefaultEntrypoint, Cmd: args, Env: e.envs, HostConfig: dockerclient.HostConfig{ Binds: []string{"/var/run/docker.sock:/var/run/docker.sock"}, }, Volumes: map[string]struct{}{ "/var/run/docker.sock": struct{}{}, }, } log.Infof("preparing container %s", name) client.PullImage(conf.Image, nil) _, err = docker.RunDaemon(client, conf, name) if err != nil { log.Errorf("error starting build container. %s", err) return err } // UPDATE STATUS err = updater.SetJob(c, r) if err != nil { log.Errorf("error updating job status as running. %s", err) return err } // WAIT FOR OUTPUT info, builderr := docker.Wait(client, name) switch { case info.State.ExitCode == 128: r.Job.ExitCode = info.State.ExitCode r.Job.Status = model.StatusKilled case info.State.ExitCode == 130: r.Job.ExitCode = info.State.ExitCode r.Job.Status = model.StatusKilled case builderr != nil: r.Job.Status = model.StatusError case info.State.ExitCode != 0: r.Job.ExitCode = info.State.ExitCode r.Job.Status = model.StatusFailure default: r.Job.Status = model.StatusSuccess } // send the logs to the datastore var buf bytes.Buffer rc, err := client.ContainerLogs(name, docker.LogOpts) if err != nil && builderr != nil { buf.WriteString("Error launching build") buf.WriteString(builderr.Error()) } else if err != nil { buf.WriteString("Error launching build") buf.WriteString(err.Error()) log.Errorf("error opening connection to logs. %s", err) return err } else { defer rc.Close() stdcopy.StdCopy(&buf, &buf, io.LimitReader(rc, 5000000)) } // update the task in the datastore r.Job.Finished = time.Now().UTC().Unix() err = updater.SetJob(c, r) if err != nil { log.Errorf("error updating job after completion. %s", err) return err } err = updater.SetLogs(c, r, ioutil.NopCloser(&buf)) if err != nil { log.Errorf("error updating logs. %s", err) return err } log.Debugf("completed job %d with status %s.", r.Job.ID, r.Job.Status) return nil } func (e *engine) runJobNotify(r *Task, client dockerclient.Client) error { name := fmt.Sprintf("drone_build_%d_notify", r.Build.ID) defer func() { client.KillContainer(name, "9") client.RemoveContainer(name, true, true) }() // encode the build payload to write to stdin // when launching the build container in, err := encodeToLegacyFormat(r) if err != nil { log.Errorf("failure to marshal work. %s", err) return err } args := DefaultNotifyArgs args = append(args, "--") args = append(args, string(in)) conf := &dockerclient.ContainerConfig{ Image: DefaultAgent, Entrypoint: DefaultEntrypoint, Cmd: args, Env: e.envs, HostConfig: dockerclient.HostConfig{ Binds: []string{"/var/run/docker.sock:/var/run/docker.sock"}, }, Volumes: map[string]struct{}{ "/var/run/docker.sock": struct{}{}, }, } log.Infof("preparing container %s", name) info, err := docker.Run(client, conf, name) if err != nil { log.Errorf("Error starting notification container %s. %s", name, err) } // for debugging purposes we print a failed notification executions // output to the logs. Otherwise we have no way to troubleshoot failed // notifications. This is temporary code until I've come up with // a better solution. if info != nil && info.State.ExitCode != 0 && log.GetLevel() >= log.InfoLevel { var buf bytes.Buffer rc, err := client.ContainerLogs(name, docker.LogOpts) if err == nil { defer rc.Close() stdcopy.StdCopy(&buf, &buf, io.LimitReader(rc, 50000)) } log.Infof("Notification container %s exited with %d", name, info.State.ExitCode) log.Infoln(buf.String()) } return err }