// Copyright 2019 Drone IO, Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package events import ( "context" "encoding/json" "io" "net/http" "strconv" "time" "github.com/drone/drone/core" "github.com/drone/drone/handler/api/render" "github.com/go-chi/chi" ) // HandleLogStream creates an http.HandlerFunc that streams builds logs // to the http.Response in an event stream format. func HandleLogStream( repos core.RepositoryStore, builds core.BuildStore, stages core.StageStore, steps core.StepStore, stream core.LogStream, ) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var ( namespace = chi.URLParam(r, "owner") name = chi.URLParam(r, "name") ) number, err := strconv.ParseInt(chi.URLParam(r, "number"), 10, 64) if err != nil { render.BadRequest(w, err) return } stageNumber, err := strconv.Atoi(chi.URLParam(r, "stage")) if err != nil { render.BadRequest(w, err) return } stepNumber, err := strconv.Atoi(chi.URLParam(r, "step")) if err != nil { render.BadRequest(w, err) return } repo, err := repos.FindName(r.Context(), namespace, name) if err != nil { render.NotFound(w, err) return } build, err := builds.FindNumber(r.Context(), repo.ID, number) if err != nil { render.NotFound(w, err) return } stage, err := stages.FindNumber(r.Context(), build.ID, stageNumber) if err != nil { render.NotFound(w, err) return } step, err := steps.FindNumber(r.Context(), stage.ID, stepNumber) if err != nil { render.NotFound(w, err) return } h := w.Header() h.Set("Content-Type", "text/event-stream") h.Set("Cache-Control", "no-cache") h.Set("Connection", "keep-alive") h.Set("X-Accel-Buffering", "no") f, ok := w.(http.Flusher) if !ok { return } io.WriteString(w, ": ping\n\n") f.Flush() ctx, cancel := context.WithCancel(r.Context()) defer cancel() enc := json.NewEncoder(w) linec, errc := stream.Tail(ctx, step.ID) if errc == nil { io.WriteString(w, "event: error\ndata: eof\n\n") return } L: for { select { case <-ctx.Done(): break L case <-errc: break L case <-time.After(time.Hour): break L case <-time.After(pingInterval): io.WriteString(w, ": ping\n\n") case line := <-linec: io.WriteString(w, "data: ") enc.Encode(line) io.WriteString(w, "\n\n") f.Flush() } } io.WriteString(w, "event: error\ndata: eof\n\n") f.Flush() } }