cmd, docker: Updates for infrastructure

These are some changes to the relay pool server, upgrade server, and crash receiver to run under Kubernetes. It's been in production for a while.
2023-01-31 11:17:52 +01:00
parent a6c2a5a0ce
commit 65cfefaa3c
12 changed files with 653 additions and 161 deletions
@@ -0,0 +1,181 @@
+package main
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"io"
+	"log"
+	"os"
+	"path/filepath"
+	"sort"
+	"time"
+)
+
+type diskStore struct {
+	dir      string
+	inbox    chan diskEntry
+	maxBytes int64
+	maxFiles int
+
+	currentFiles []currentFile
+	currentSize  int64
+}
+
+type diskEntry struct {
+	path string
+	data []byte
+}
+
+type currentFile struct {
+	path  string
+	size  int64
+	mtime int64
+}
+
+func (d *diskStore) Serve(ctx context.Context) {
+	if err := os.MkdirAll(d.dir, 0750); err != nil {
+		log.Println("Creating directory:", err)
+		return
+	}
+
+	if err := d.inventory(); err != nil {
+		log.Println("Failed to inventory disk store:", err)
+	}
+	d.clean()
+
+	cleanTimer := time.NewTicker(time.Minute)
+	inventoryTimer := time.NewTicker(24 * time.Hour)
+
+	buf := new(bytes.Buffer)
+	gw := gzip.NewWriter(buf)
+	for {
+		select {
+		case entry := <-d.inbox:
+			path := d.fullPath(entry.path)
+
+			if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
+				log.Println("Creating directory:", err)
+				continue
+			}
+
+			buf.Reset()
+			gw.Reset(buf)
+			if _, err := gw.Write(entry.data); err != nil {
+				log.Println("Failed to compress crash report:", err)
+				continue
+			}
+			if err := gw.Close(); err != nil {
+				log.Println("Failed to compress crash report:", err)
+				continue
+			}
+			if err := os.WriteFile(path, buf.Bytes(), 0644); err != nil {
+				log.Printf("Failed to write %s: %v", entry.path, err)
+				_ = os.Remove(path)
+				continue
+			}
+
+			d.currentSize += int64(buf.Len())
+			d.currentFiles = append(d.currentFiles, currentFile{
+				size: int64(len(entry.data)),
+				path: path,
+			})
+
+		case <-cleanTimer.C:
+			d.clean()
+
+		case <-inventoryTimer.C:
+			if err := d.inventory(); err != nil {
+				log.Println("Failed to inventory disk store:", err)
+			}
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (d *diskStore) Put(path string, data []byte) bool {
+	select {
+	case d.inbox <- diskEntry{
+		path: path,
+		data: data,
+	}:
+		return true
+	default:
+		return false
+	}
+}
+
+func (d *diskStore) Get(path string) ([]byte, error) {
+	path = d.fullPath(path)
+	bs, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	gr, err := gzip.NewReader(bytes.NewReader(bs))
+	if err != nil {
+		return nil, err
+	}
+	defer gr.Close()
+	return io.ReadAll(gr)
+}
+
+func (d *diskStore) Exists(path string) bool {
+	path = d.fullPath(path)
+	_, err := os.Lstat(path)
+	return err == nil
+}
+
+func (d *diskStore) clean() {
+	for len(d.currentFiles) > 0 && (len(d.currentFiles) > d.maxFiles || d.currentSize > d.maxBytes) {
+		f := d.currentFiles[0]
+		log.Println("Removing", f.path)
+		if err := os.Remove(f.path); err != nil {
+			log.Println("Failed to remove file:", err)
+		}
+		d.currentFiles = d.currentFiles[1:]
+		d.currentSize -= f.size
+	}
+	var oldest time.Duration
+	if len(d.currentFiles) > 0 {
+		oldest = time.Since(time.Unix(d.currentFiles[0].mtime, 0)).Truncate(time.Minute)
+	}
+	log.Printf("Clean complete: %d files, %d MB, oldest is %v ago", len(d.currentFiles), d.currentSize>>20, oldest)
+}
+
+func (d *diskStore) inventory() error {
+	d.currentFiles = nil
+	d.currentSize = 0
+	err := filepath.Walk(d.dir, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if info.IsDir() {
+			return nil
+		}
+		if filepath.Ext(path) != ".gz" {
+			return nil
+		}
+		d.currentSize += info.Size()
+		d.currentFiles = append(d.currentFiles, currentFile{
+			path:  path,
+			size:  info.Size(),
+			mtime: info.ModTime().Unix(),
+		})
+		return nil
+	})
+	sort.Slice(d.currentFiles, func(i, j int) bool {
+		return d.currentFiles[i].mtime < d.currentFiles[j].mtime
+	})
+	var oldest time.Duration
+	if len(d.currentFiles) > 0 {
+		oldest = time.Since(time.Unix(d.currentFiles[0].mtime, 0)).Truncate(time.Minute)
+	}
+	log.Printf("Inventory complete: %d files, %d MB, oldest is %v ago", len(d.currentFiles), d.currentSize>>20, oldest)
+	return err
+}
+
+func (d *diskStore) fullPath(path string) string {
+	return filepath.Join(d.dir, path[0:2], path[2:]) + ".gz"
+}
@@ -13,15 +13,17 @@
 package main

 import (
+	"context"
 	"encoding/json"
-	"flag"
 	"fmt"
 	"io"
 	"log"
 	"net/http"
 	"os"
 	"path/filepath"
+	"time"

+	"github.com/alecthomas/kong"
 	"github.com/syncthing/syncthing/lib/sha256"
 	"github.com/syncthing/syncthing/lib/ur"

@@ -30,26 +32,50 @@ import (

 const maxRequestSize = 1 << 20 // 1 MiB

+type cli struct {
+	Dir           string        `help:"Parent directory to store crash and failure reports in" env:"REPORTS_DIR" default:"."`
+	DSN           string        `help:"Sentry DSN" env:"SENTRY_DSN"`
+	Listen        string        `help:"HTTP listen address" default:":8080" env:"LISTEN_ADDRESS"`
+	MaxDiskFiles  int           `help:"Maximum number of reports on disk" default:"100000" env:"MAX_DISK_FILES"`
+	MaxDiskSizeMB int64         `help:"Maximum disk space to use for reports" default:"1024" env:"MAX_DISK_SIZE_MB"`
+	CleanInterval time.Duration `help:"Interval between cleaning up old reports" default:"12h" env:"CLEAN_INTERVAL"`
+	SentryQueue   int           `help:"Maximum number of reports to queue for sending to Sentry" default:"64" env:"SENTRY_QUEUE"`
+	DiskQueue     int           `help:"Maximum number of reports to queue for writing to disk" default:"64" env:"DISK_QUEUE"`
+}
+
 func main() {
-	dir := flag.String("dir", ".", "Parent directory to store crash and failure reports in")
-	dsn := flag.String("dsn", "", "Sentry DSN")
-	listen := flag.String("listen", ":22039", "HTTP listen address")
-	flag.Parse()
+	var params cli
+	kong.Parse(&params)

 	mux := http.NewServeMux()

-	cr := &crashReceiver{
-		dir: filepath.Join(*dir, "crash_reports"),
-		dsn: *dsn,
+	ds := &diskStore{
+		dir:      filepath.Join(params.Dir, "crash_reports"),
+		inbox:    make(chan diskEntry, params.DiskQueue),
+		maxFiles: params.MaxDiskFiles,
+		maxBytes: params.MaxDiskSizeMB << 20,
 	}
+	go ds.Serve(context.Background())
+
+	ss := &sentryService{
+		dsn:   params.DSN,
+		inbox: make(chan sentryRequest, params.SentryQueue),
+	}
+	go ss.Serve(context.Background())
+
+	cr := &crashReceiver{
+		store:  ds,
+		sentry: ss,
+	}
+
 	mux.Handle("/", cr)

-	if *dsn != "" {
-		mux.HandleFunc("/newcrash/failure", handleFailureFn(*dsn, filepath.Join(*dir, "failure_reports")))
+	if params.DSN != "" {
+		mux.HandleFunc("/newcrash/failure", handleFailureFn(params.DSN, filepath.Join(params.Dir, "failure_reports")))
 	}

 	log.SetOutput(os.Stdout)
-	if err := http.ListenAndServe(*listen, mux); err != nil {
+	if err := http.ListenAndServe(params.Listen, mux); err != nil {
 		log.Fatalln("HTTP serve:", err)
 	}
 }
@@ -8,8 +8,10 @@ package main

 import (
 	"bytes"
+	"context"
 	"errors"
 	"io"
+	"log"
 	"regexp"
 	"strings"
 	"sync"
@@ -31,6 +33,44 @@ var (
 	clientsMut sync.Mutex
 )

+type sentryService struct {
+	dsn   string
+	inbox chan sentryRequest
+}
+
+type sentryRequest struct {
+	reportID string
+	data     []byte
+}
+
+func (s *sentryService) Serve(ctx context.Context) {
+	for {
+		select {
+		case req := <-s.inbox:
+			pkt, err := parseCrashReport(req.reportID, req.data)
+			if err != nil {
+				log.Println("Failed to parse crash report:", err)
+				continue
+			}
+			if err := sendReport(s.dsn, pkt, req.reportID); err != nil {
+				log.Println("Failed to send crash report:", err)
+			}
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (s *sentryService) Send(reportID string, data []byte) bool {
+	select {
+	case s.inbox <- sentryRequest{reportID, data}:
+		return true
+	default:
+		return false
+	}
+}
+
 func sendReport(dsn string, pkt *raven.Packet, userID string) error {
 	pkt.Interfaces = append(pkt.Interfaces, &raven.User{ID: userID})

@@ -7,20 +7,16 @@
 package main

 import (
-	"bytes"
-	"compress/gzip"
 	"io"
 	"log"
 	"net/http"
-	"os"
 	"path"
-	"path/filepath"
 	"strings"
 )

 type crashReceiver struct {
-	dir string
-	dsn string
+	store  *diskStore
+	sentry *sentryService
 }

 func (r *crashReceiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {
@@ -43,55 +39,38 @@ func (r *crashReceiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		return
 	}

-	// The location of the report on disk, compressed
-	fullPath := filepath.Join(r.dir, r.dirFor(reportID), reportID) + ".gz"
-
 	switch req.Method {
 	case http.MethodGet:
-		r.serveGet(fullPath, w, req)
+		r.serveGet(reportID, w, req)
 	case http.MethodHead:
-		r.serveHead(fullPath, w, req)
+		r.serveHead(reportID, w, req)
 	case http.MethodPut:
-		r.servePut(reportID, fullPath, w, req)
+		r.servePut(reportID, w, req)
 	default:
 		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
 	}
 }

 // serveGet responds to GET requests by serving the uncompressed report.
-func (*crashReceiver) serveGet(fullPath string, w http.ResponseWriter, _ *http.Request) {
-	fd, err := os.Open(fullPath)
+func (r *crashReceiver) serveGet(reportID string, w http.ResponseWriter, _ *http.Request) {
+	bs, err := r.store.Get(reportID)
 	if err != nil {
 		http.Error(w, "Not found", http.StatusNotFound)
 		return
 	}
-
-	defer fd.Close()
-	gr, err := gzip.NewReader(fd)
-	if err != nil {
-		http.Error(w, "Internal server error", http.StatusInternalServerError)
-		return
-	}
-	_, _ = io.Copy(w, gr) // best effort
+	w.Write(bs)
 }

 // serveHead responds to HEAD requests by checking if the named report
 // already exists in the system.
-func (*crashReceiver) serveHead(fullPath string, w http.ResponseWriter, _ *http.Request) {
-	if _, err := os.Lstat(fullPath); err != nil {
+func (r *crashReceiver) serveHead(reportID string, w http.ResponseWriter, _ *http.Request) {
+	if !r.store.Exists(reportID) {
 		http.Error(w, "Not found", http.StatusNotFound)
 	}
 }

 // servePut accepts and stores the given report.
-func (r *crashReceiver) servePut(reportID, fullPath string, w http.ResponseWriter, req *http.Request) {
-	// Ensure the destination directory exists
-	if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil {
-		log.Println("Creating directory:", err)
-		http.Error(w, "Internal server error", http.StatusInternalServerError)
-		return
-	}
-
+func (r *crashReceiver) servePut(reportID string, w http.ResponseWriter, req *http.Request) {
 	// Read at most maxRequestSize of report data.
 	log.Println("Receiving report", reportID)
 	lr := io.LimitReader(req.Body, maxRequestSize)
@@ -102,40 +81,13 @@ func (r *crashReceiver) servePut(reportID, fullPath string, w http.ResponseWrite
 		return
 	}

-	// Compress the report for storage
-	buf := new(bytes.Buffer)
-	gw := gzip.NewWriter(buf)
-	_, _ = gw.Write(bs) // can't fail
-	gw.Close()
-
-	// Create an output file with the compressed report
-	err = os.WriteFile(fullPath, buf.Bytes(), 0644)
-	if err != nil {
-		log.Println("Saving report:", err)
-		http.Error(w, "Internal server error", http.StatusInternalServerError)
-		return
+	// Store the report
+	if !r.store.Put(reportID, bs) {
+		log.Println("Failed to store report (queue full):", reportID)
 	}

 	// Send the report to Sentry
-	if r.dsn != "" {
-		// Remote ID
-		user := userIDFor(req)
-
-		go func() {
-			// There's no need for the client to have to wait for this part.
-			pkt, err := parseCrashReport(reportID, bs)
-			if err != nil {
-				log.Println("Failed to parse crash report:", err)
-				return
-			}
-			if err := sendReport(r.dsn, pkt, user); err != nil {
-				log.Println("Failed to send crash report:", err)
-			}
-		}()
+	if !r.sentry.Send(reportID, bs) {
+		log.Println("Failed to send report to sentry (queue full):", reportID)
 	}
 }
-
-// 01234567890abcdef... => 01/23
-func (*crashReceiver) dirFor(base string) string {
-	return filepath.Join(base[0:2], base[2:4])
-}