albatros/pkg/handler.go
2023-05-03 10:52:34 +02:00

189 lines
5.8 KiB
Go

package pkg
import (
b64 "encoding/base64"
"encoding/json"
"fmt"
"golang.org/x/exp/slices"
"io"
"log"
"net/http"
"strings"
nomad "github.com/hashicorp/nomad/api"
)
type HookHandler struct {
Config *Config
Cluster *Cluster
}
type BuildHandler struct {
Config *Config
Cluster *Cluster
}
func (h HookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Hook only support POST requests", http.StatusBadRequest)
return
}
q := r.URL.Query()
maybeToken, ok := q["token"]
if !ok || len(maybeToken) < 1 {
http.Error(w, "Missing query parameter 'token'. Try adding '?token=xxx'", http.StatusBadRequest)
return
}
token := maybeToken[0]
flavor := "default"
var notification GiteaNotification
dec := json.NewDecoder(r.Body)
if err := dec.Decode(&notification); err != nil {
http.Error(w, "Can't parse your request JSON", http.StatusBadRequest)
return
}
notifInfo := notifSummary(&notification)
log.Printf("Received gitea notification for %s\n", notifInfo)
// Fetch our repo descriptor
kv := h.Cluster.Consul.KV()
encodedRepoUrl := b64.StdEncoding.EncodeToString([]byte(notification.Repository.CloneUrl))
key := "albatros/" + encodedRepoUrl
log.Printf("Fetching key %s for %s\n", key, notifInfo)
pair, _, err := kv.Get(key, nil)
if err != nil {
http.Error(w, "Can't fetch the repo descriptor in Consul", http.StatusInternalServerError)
log.Printf("Job error for %s, can't fetch repo descriptor in consul: %+v\n", notifInfo, err)
return
}
if pair == nil || pair.Value == nil {
http.Error(w, "You must declare %s in Consul in order to build it", http.StatusForbidden)
return
}
// Parse our repo descriptor
var repoDesc ConsulSecret
if err = json.Unmarshal(pair.Value, &repoDesc); err != nil {
http.Error(w, "Can't decode your Consul configuration for this repo", http.StatusInternalServerError)
log.Printf("Job error for %s, can't unmarshal the Consul descriptor: %+v\n", notifInfo, err)
return
}
// Check token
if repoDesc.Hook.Token != token {
http.Error(w, "Your albatros token does not match the one declared in Consul", http.StatusForbidden)
return
}
// Build job parameters for Nomad
meta := map[string]string{
"REPO_URL": notification.Repository.CloneUrl,
"COMMIT": notification.After,
"FLAVOR": flavor,
// @FIXME: this code is not correct, this is a hack
"BRANCH": strings.ReplaceAll(notification.Ref, "refs/heads/", ""),
}
// Check sender
payload := []byte{}
if slices.Contains(repoDesc.Trusted.Senders, notification.Sender.Username) {
log.Printf("Trusted build of %s as %s in the list of allowed senders, inject secrets\n", notifInfo, notification.Sender.Username)
// Write payload
payload = []byte(repoDesc.Inject)
}
jobs := h.Cluster.Nomad.Jobs()
dres, _, err := jobs.Dispatch("builder", meta, payload, "albatros", &nomad.WriteOptions{})
if err != nil {
http.Error(w, "Can't submit your job to Nomad", http.StatusInternalServerError)
}
log.Printf("Created job %s for %s\n", dres.DispatchedJobID, notifInfo)
// Start a lifecycle observer to update gitea status
j := Job{
Cluster: h.Cluster,
Config: h.Config,
Notification: &notification,
Dispatch: dres,
Creds: &repoDesc.Gitea,
Flavor: flavor,
}
go j.follow()
io.WriteString(w, dres.DispatchedJobID)
}
func (b BuildHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
jobID, ok := q["job"]
if !ok || len(jobID) < 1 {
http.Error(w, "Missing query parameter 'job'. Try adding '?job=xxx'", http.StatusBadRequest)
return
}
logType, ok := q["log"]
log.Printf("Follow logs for %s\n", jobID)
if !ok || len(logType) < 1 || !(logType[0] == "stdout" || logType[0] == "stderr") {
http.Error(w, "Missing or wrong query parameter 'log'.\nTry adding '?log=stdout' or '?log=stderr'", http.StatusBadRequest)
return
}
logFilter := logType[0]
jobs := b.Cluster.Nomad.Jobs()
allocList, _, err := jobs.Allocations(jobID[0], true, &nomad.QueryOptions{})
if err != nil {
http.Error(w, "Unable to fetch your job on Nomad. Might be garbage collected.", http.StatusInternalServerError)
return
}
if len(allocList) < 1 {
http.Error(w, "Job does not contain at least 1 allocation, can't display logs. Job might be garbage collected. ", http.StatusNotFound)
return
}
myAllocStub := allocList[0]
allocations := b.Cluster.Nomad.Allocations()
myAlloc, _, err := allocations.Info(myAllocStub.ID, &nomad.QueryOptions{})
if err != nil {
http.Error(w, "Allocation does not exist anymore. Allocation might be garbage collected", http.StatusNotFound)
return
}
log.Printf("Selected alloc %s for job %s\n", myAlloc.ID, jobID)
allocFS := b.Cluster.Nomad.AllocFS()
scancel := make(chan struct{})
sframe, serr := allocFS.Logs(myAlloc, true, "executor", logFilter, "start", 0, scancel, &nomad.QueryOptions{})
// stream logs to client's browser
io.WriteString(w, fmt.Sprintf("--- Streaming logs for %s, job can take some times to start, be patient ---\n", jobID))
build_loop:
for {
select {
case <-r.Context().Done():
// client disconnect, cleaning
break build_loop
case nomadErr := <-serr:
// an error occured in nomad, inform user and clean
_, _ = io.WriteString(w, fmt.Sprintf("\n--- Broken stream: %+v ---\n", nomadErr))
break build_loop
case chunk := <-sframe:
// we get some data from nomad, send it to the client
for i := 0; i < len(chunk.Data); {
written, err := w.Write(chunk.Data[i:])
w.(http.Flusher).Flush() // flush the buffer to send it right now
i += written
if err == io.EOF {
_, _ = io.WriteString(w, "\n--- End of file :-) ---\n")
break build_loop
} else if err != nil {
_, _ = io.WriteString(w, fmt.Sprintf("\n--- Broken stream: %+v ---\n", err))
break build_loop
}
}
}
}
log.Printf("Connection closed, cleaning listeners for %s\n", jobID)
scancel <- struct{}{}
}