442 lines
13 KiB
Go
442 lines
13 KiB
Go
// Copyright (C) 2023-2024 Umorpha Systems
|
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"embed"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"html/template"
|
|
"io"
|
|
"io/fs"
|
|
"log"
|
|
"mime"
|
|
"mime/multipart"
|
|
"net/http"
|
|
"net/url"
|
|
"os"
|
|
"path"
|
|
"path/filepath"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.lukeshu.com/go/containers/typedsync"
|
|
"github.com/datawire/ocibuild/pkg/cliutil"
|
|
"github.com/spf13/cobra"
|
|
|
|
source "git.mothstuff.lol/lukeshu/eclipse"
|
|
"git.mothstuff.lol/lukeshu/eclipse/lib/eclipse"
|
|
)
|
|
|
|
func main() {
|
|
argparser := &cobra.Command{
|
|
Use: os.Args[0] + " [flags]",
|
|
Short: "HTTP server",
|
|
Args: cliutil.WrapPositionalArgs(cobra.NoArgs),
|
|
|
|
SilenceErrors: true, // we'll handle this ourselves after .ExecuteContext()
|
|
SilenceUsage: true, // FlagErrorFunc will handle this
|
|
|
|
CompletionOptions: cobra.CompletionOptions{
|
|
DisableDefaultCmd: true,
|
|
},
|
|
}
|
|
argparser.SetFlagErrorFunc(cliutil.FlagErrorFunc)
|
|
argparser.SetHelpTemplate(cliutil.HelpTemplate)
|
|
|
|
var cfgFile string
|
|
argparser.Flags().StringVar(&cfgFile, "config", source.DefaultConfigFile,
|
|
"Config file to use")
|
|
argparser.MarkFlagFilename("config", "yml", "yaml")
|
|
|
|
var socket string
|
|
argparser.Flags().StringVar(&socket, "socket", "tcp::8080",
|
|
"the socket to listen on; a \"family:address\" pair, where the\n"+
|
|
"\"family\" is one of \"tcp\", \"tcp4\", \"tcp6\", \"unix\", \"unixpacket\"\n"+
|
|
"(the usual Go net.Listen() network names), or \"fd\" (to listen\n"+
|
|
"on an already-open file descriptor). The \"address\" part for\n"+
|
|
"the \"fd\" family is one of\n"+
|
|
" 1. a file descriptor number (or the special names \"stdin\",\n"+
|
|
" \"stdout\", or \"stderr\" to refer to 0, 1, or 2\n"+
|
|
" respectively),\n"+
|
|
" 2. \"systemd:N\" to refer to the Nth file descriptor passed in\n"+
|
|
" via systemd (or \"systemd\" as shorthand for \"systemd:0\")\n"+
|
|
" (see sd_listend_fds(3)),\n"+
|
|
" 3. \"systemd:NAME\" to refer to a named file descriptor passed\n"+
|
|
" in via systemd (see sd_listend_fds_with_names(3)).\n")
|
|
|
|
argparser.RunE = func(cmd *cobra.Command, args []string) error {
|
|
stype, saddr, ok := strings.Cut(socket, ":")
|
|
if !ok {
|
|
return cliutil.FlagErrorFunc(cmd, fmt.Errorf("invalid address: %q", socket))
|
|
}
|
|
return Serve(cfgFile, stype, saddr)
|
|
}
|
|
|
|
ctx := context.Background()
|
|
if err := argparser.ExecuteContext(ctx); err != nil {
|
|
fmt.Fprintf(os.Stderr, "%v: error: %v\n", argparser.CommandPath(), err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
func Serve(cfgFile, stype, saddr string) (err error) {
|
|
maybeSetErr := func(_err error) {
|
|
if err == nil && _err != nil {
|
|
err = _err
|
|
}
|
|
}
|
|
|
|
cfg, err := eclipse.LoadConfig(cfgFile)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
db, err := cfg.DB()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { maybeSetErr(db.Close()) }()
|
|
|
|
sock, err := netListen(stype, saddr)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := db.GCRunningJobs(); err != nil {
|
|
return err
|
|
}
|
|
|
|
state := &ServerState{
|
|
db: db,
|
|
}
|
|
|
|
router := http.NewServeMux()
|
|
router.HandleFunc("/", state.serveIndex)
|
|
router.HandleFunc("/style.css", state.serveStatic)
|
|
router.HandleFunc("/favicon.ico", state.serveStatic)
|
|
router.HandleFunc("/eclipse.tar", state.serveStatic)
|
|
router.HandleFunc("/jobs/", state.serveJobs)
|
|
|
|
log.Printf("Serving on %v...", sock.Addr())
|
|
return http.Serve(sock, router)
|
|
}
|
|
|
|
type ServerState struct {
|
|
db *eclipse.DB
|
|
running typedsync.Map[eclipse.JobID, chan struct{}]
|
|
}
|
|
|
|
//go:embed style.css
|
|
var fileStyle []byte
|
|
|
|
//go:embed favicon.ico
|
|
var fileFavicon []byte
|
|
|
|
func (*ServerState) serveStatic(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
http.Error(w, "HTTP 405: only GET is supported", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
switch r.URL.Path {
|
|
case "/style.css":
|
|
w.Header().Set("Content-Type", "text/css")
|
|
_, _ = w.Write(fileStyle)
|
|
case "/favicon.ico":
|
|
w.Header().Set("Content-Type", "image/vnd.microsoft.icon")
|
|
_, _ = w.Write(fileFavicon)
|
|
case "/eclipse.tar":
|
|
w.Header().Set("Content-Type", "application/x-tar")
|
|
_, _ = w.Write(source.Tarball)
|
|
default:
|
|
http.NotFound(w, r)
|
|
}
|
|
}
|
|
|
|
//go:embed html
|
|
var tmplFS embed.FS
|
|
|
|
var (
|
|
tmplIndex = template.Must(template.ParseFS(tmplFS, "html/pages/index.html.tmpl", "html/lib/*.html.tmpl"))
|
|
tmplJobs = template.Must(template.ParseFS(tmplFS, "html/pages/jobs.html.tmpl", "html/lib/*.html.tmpl"))
|
|
tmplJob = template.Must(template.ParseFS(tmplFS, "html/pages/job.html.tmpl", "html/lib/*.html.tmpl"))
|
|
)
|
|
|
|
func httpAddSlash(w http.ResponseWriter, r *http.Request) {
|
|
u := &url.URL{Path: r.URL.Path + "/", RawQuery: r.URL.RawQuery}
|
|
http.Redirect(w, r, u.String(), http.StatusMovedPermanently)
|
|
}
|
|
|
|
func (*ServerState) serveIndex(w http.ResponseWriter, r *http.Request) {
|
|
if r.URL.Path != "/" {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
if r.Method != http.MethodGet {
|
|
http.Error(w, "HTTP 405: only GET is supported", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
|
if err := tmplIndex.Execute(w, nil); err != nil {
|
|
log.Printf("error: %q: %v", r.URL, err)
|
|
}
|
|
}
|
|
|
|
func (o *ServerState) serveJobs(w http.ResponseWriter, r *http.Request) {
|
|
rest := strings.TrimPrefix(r.URL.Path, "/jobs/")
|
|
if rest == "" {
|
|
switch r.Method {
|
|
case http.MethodGet:
|
|
jobs, err := o.db.ListJobs()
|
|
if err != nil {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
|
if err := tmplJobs.Execute(w, map[string]any{"Jobs": jobs}); err != nil {
|
|
log.Printf("error: %q: %v", r.URL, err)
|
|
}
|
|
// TODO: content-negotiate /jobs/ to have JSON
|
|
case http.MethodPost:
|
|
// TODO: allow POSTing to /jobs/ to create a job
|
|
default:
|
|
http.Error(w, "HTTP 405: only GET and POST are supported", http.StatusMethodNotAllowed)
|
|
}
|
|
return
|
|
}
|
|
jobIDStr, file, haveSlash := strings.Cut(rest, "/")
|
|
jobID, err := strconv.Atoi(jobIDStr)
|
|
if err != nil {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
job, err := o.db.GetJob(eclipse.JobID(jobID))
|
|
if err != nil {
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
if !haveSlash {
|
|
httpAddSlash(w, r)
|
|
return
|
|
}
|
|
switch file {
|
|
case "":
|
|
switch r.Method {
|
|
case http.MethodGet:
|
|
dirfs := job.DirFS()
|
|
artifactNames := []string{}
|
|
if err := fs.WalkDir(dirfs, "artifacts", func(path string, d fs.DirEntry, err error) error {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if d.Type() == 0 {
|
|
artifactNames = append(artifactNames, strings.TrimPrefix(path, "artifacts/"))
|
|
}
|
|
return nil
|
|
}); err != nil && !os.IsNotExist(err) {
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
|
if err := tmplJob.Execute(w, map[string]any{
|
|
"Job": job,
|
|
"ArtifactNames": artifactNames,
|
|
}); err != nil {
|
|
log.Printf("error: %q: %v", r.URL, err)
|
|
}
|
|
case http.MethodPost:
|
|
// TODO: Validate API key
|
|
if err := o.runJob(w, r, job); err != nil {
|
|
http.Error(w, err.Error(), err.Code)
|
|
log.Printf("error: %q: %v", r.URL, err)
|
|
}
|
|
default:
|
|
http.Error(w, "HTTP 405: only GET and POST are supported", http.StatusMethodNotAllowed)
|
|
}
|
|
default:
|
|
if r.Method != http.MethodGet {
|
|
http.Error(w, "HTTP 405: only GET is supported", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
if file == "log.txt" && job.Status == eclipse.StatusRunning {
|
|
fh, _ := os.Open(filepath.Join(job.Dir, "log.txt"))
|
|
ch, _ := o.running.Load(job.ID)
|
|
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
|
|
|
|
if fh != nil {
|
|
if _, err := io.Copy(w, fh); err != nil {
|
|
log.Printf("error: %q: %v", r.URL, err)
|
|
return
|
|
}
|
|
}
|
|
|
|
ticker := time.NewTicker(1 * time.Second)
|
|
defer ticker.Stop()
|
|
for keepGoing := true; keepGoing; {
|
|
select {
|
|
case <-ch:
|
|
keepGoing = false
|
|
case <-ticker.C:
|
|
}
|
|
if fh == nil {
|
|
fh, _ = os.Open(filepath.Join(job.Dir, "log.txt"))
|
|
}
|
|
if fh != nil {
|
|
if _, err := io.Copy(w, fh); err != nil {
|
|
log.Printf("error: %q: %v", r.URL, err)
|
|
keepGoing = false
|
|
}
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
r.URL.Path = "/"+file
|
|
http.FileServer(http.Dir(job.Dir)).ServeHTTP(w, r)
|
|
}
|
|
}
|
|
|
|
type httpError struct {
|
|
Err error
|
|
Code int
|
|
}
|
|
|
|
func (e *httpError) Error() string {
|
|
return fmt.Sprintf("HTTP %d: %v", e.Code, e.Err)
|
|
}
|
|
|
|
func (o *ServerState) runJob(w http.ResponseWriter, r *http.Request, job eclipse.Job) (_reterr *httpError) {
|
|
log.Printf("running job: %q", r.URL)
|
|
|
|
ctl := http.NewResponseController(w)
|
|
if err := ctl.EnableFullDuplex(); err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: err}
|
|
}
|
|
|
|
ch := make(chan struct{})
|
|
if _, conflict := o.running.LoadOrStore(job.ID, ch); conflict {
|
|
return &httpError{Code: http.StatusConflict, Err: errors.New("job has already started (ch)")}
|
|
}
|
|
defer func() {
|
|
close(ch)
|
|
o.running.Delete(job.ID)
|
|
}()
|
|
swapped, err := o.db.SwapJobStatus(job.ID, eclipse.StatusNew, eclipse.StatusRunning)
|
|
if err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: err}
|
|
}
|
|
if !swapped {
|
|
return &httpError{Code: http.StatusConflict, Err: errors.New("job has already started (db)")}
|
|
}
|
|
defer func() {
|
|
if _, err := o.db.SwapJobStatus(job.ID, eclipse.StatusRunning, eclipse.StatusError); err != nil {
|
|
_reterr = &httpError{Code: http.StatusInternalServerError, Err: err}
|
|
}
|
|
}()
|
|
|
|
bs, err := json.Marshal(job)
|
|
if err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: fmt.Errorf("encoding job description: %w", err)}
|
|
}
|
|
mediaType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
|
|
if err != nil {
|
|
return &httpError{Code: http.StatusBadRequest, Err: fmt.Errorf("invalid Content-Type: %q: %w", r.Header.Get("Content-Type"), err)}
|
|
}
|
|
if mediaType != "multipart/mixed" {
|
|
return &httpError{Code: http.StatusBadRequest, Err: fmt.Errorf("invalid Content-Type: %q", r.Header.Get("Content-Type"))}
|
|
}
|
|
reader := multipart.NewReader(r.Body, params["boundary"])
|
|
|
|
// Last chance to
|
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
if _, err := w.Write(bs); err != nil {
|
|
return &httpError{Code: http.StatusRequestTimeout, Err: fmt.Errorf("write job description: %w", err)}
|
|
}
|
|
if err := ctl.Flush(); err != nil {
|
|
return &httpError{Code: http.StatusRequestTimeout, Err: fmt.Errorf("write job description: %w", err)}
|
|
}
|
|
|
|
// At this point we can no longer return a status code /////////////////
|
|
|
|
var status string
|
|
var buffer [32 * 1024]byte
|
|
for {
|
|
part, err := reader.NextPart()
|
|
if err != nil {
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
return &httpError{Code: http.StatusRequestTimeout, Err: fmt.Errorf("read next part: %w", err)}
|
|
}
|
|
log.Printf("part: %q", part.Header.Get("Content-Disposition"))
|
|
disposition, params, err := mime.ParseMediaType(part.Header.Get("Content-Disposition"))
|
|
if err != nil {
|
|
return &httpError{Code: http.StatusBadRequest, Err: fmt.Errorf("invalid Content-Disposition: %q: %w", part.Header.Get("Content-Disposition"), err)}
|
|
}
|
|
filename := path.Clean(params["filename"])
|
|
switch {
|
|
case disposition == "attachment" && filename == "log.txt" || strings.HasPrefix(filename, "artifacts/"):
|
|
fsFilename := filepath.Join(job.Dir, filepath.FromSlash(filename))
|
|
if err := os.MkdirAll(filepath.Dir(fsFilename), 0755); err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: fmt.Errorf("saving file: %q: %w", filename, err)}
|
|
}
|
|
fh, err := os.Create(fsFilename)
|
|
if err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: fmt.Errorf("saving file: %q: %w", filename, err)}
|
|
}
|
|
for {
|
|
nr, er := part.Read(buffer[:])
|
|
if nr > 0 {
|
|
nw, ew := fh.Write(buffer[:nr])
|
|
if ew == nil && nw < nr {
|
|
ew = io.ErrShortWrite
|
|
}
|
|
if ew != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: fmt.Errorf("saving file: %q: %w", filename, err)}
|
|
}
|
|
}
|
|
if er != nil {
|
|
if er == io.EOF {
|
|
break
|
|
}
|
|
return &httpError{Code: http.StatusRequestTimeout, Err: fmt.Errorf("saving file: %q: %w", filename, err)}
|
|
}
|
|
}
|
|
if err := fh.Close(); err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: fmt.Errorf("saving file: %q: %w", filename, err)}
|
|
}
|
|
case disposition == "form-data" && params["name"] == "status":
|
|
statusBytes, err := io.ReadAll(part)
|
|
if err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: fmt.Errorf("reading status: %q: %w", filename, err)}
|
|
}
|
|
status = strings.TrimSpace(string(statusBytes))
|
|
default:
|
|
return &httpError{Code: http.StatusBadRequest, Err: fmt.Errorf("invalid Content-Disposition: %q", part.Header.Get("Content-Disposition"))}
|
|
}
|
|
if err := part.Close(); err != nil {
|
|
return &httpError{Code: http.StatusRequestTimeout, Err: fmt.Errorf("closing part: %w", err)}
|
|
}
|
|
}
|
|
|
|
switch status {
|
|
case "0":
|
|
if _, err := o.db.SwapJobStatus(job.ID, eclipse.StatusRunning, eclipse.StatusSucceeded); err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: err}
|
|
}
|
|
case "":
|
|
return &httpError{Code: http.StatusRequestTimeout, Err: fmt.Errorf("no status: %w", io.ErrUnexpectedEOF)}
|
|
default:
|
|
if _, err := o.db.SwapJobStatus(job.ID, eclipse.StatusRunning, eclipse.StatusFailed); err != nil {
|
|
return &httpError{Code: http.StatusInternalServerError, Err: err}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|