1 files changed, 138 insertions, 0 deletions
diff --git a/logs.go b/logs.go
new file mode 100644
index 0000000..9b01e94
--- /dev/null
+++ b/logs.go
@@ -0,0 +1,138 @@
+package main
+
+// Cloudflare does not forward container stdout to observability, so syncd
+// ships its own output and git-forge's to the worker. Drops rather than
+// blocks, and never logs through log itself.
+
+import (
+	"bytes"
+	"fmt"
+	"net/http"
+	"os"
+	"strings"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	shipQueue   = 256
+	shipEvery   = 500 * time.Millisecond
+	shipTimeout = 3 * time.Second
+	// The bound on getting run's attention; the flush itself is waited out,
+	// since those lines say how the container stopped.
+	shipWait = 5 * time.Second
+)
+
+type shipper struct {
+	url     string
+	lines   chan string
+	flush   chan chan struct{}
+	dropped atomic.Int64 // never queued, because the queue was full
+	unsent  atomic.Int64 // queued, but the send failed
+}
+
+// newShipper tees to stdout and to the worker, or to stdout alone with no
+// endpoint.
+func newShipper(base string) *shipper {
+	s := &shipper{
+		url:   base,
+		lines: make(chan string, shipQueue),
+		flush: make(chan chan struct{}),
+	}
+	if base != "" {
+		go s.run()
+	}
+	return s
+}
+
+// close sends what is queued and waits: the last lines say how the container
+// shut down.
+func (s *shipper) close() {
+	if s.url == "" {
+		return
+	}
+	ack := make(chan struct{})
+	select {
+	case s.flush <- ack:
+		<-ack
+	case <-time.After(shipWait):
+	}
+}
+
+// Write always reports success: the caller is a logger.
+func (s *shipper) Write(p []byte) (int, error) {
+	_, _ = os.Stdout.Write(p)
+	if s.url == "" {
+		return len(p), nil
+	}
+	for _, line := range strings.Split(string(bytes.TrimRight(p, "\n")), "\n") {
+		if line == "" {
+			continue
+		}
+		select {
+		case s.lines <- line:
+		default:
+			// Dropped rather than stalling a request; counted, so the gap is
+			// reported.
+			s.dropped.Add(1)
+		}
+	}
+	return len(p), nil
+}
+
+func (s *shipper) run() {
+	client := &http.Client{Timeout: shipTimeout}
+	t := time.NewTicker(shipEvery)
+	defer t.Stop()
+
+	var batch []string
+	send := func() {
+		lines := int64(len(batch))
+		dropped, unsent := s.dropped.Swap(0), s.unsent.Swap(0)
+		if dropped > 0 {
+			batch = append(batch, fmt.Sprintf("syncd: dropped %d log lines, the queue was full", dropped))
+		}
+		if unsent > 0 {
+			batch = append(batch, fmt.Sprintf("syncd: %d log lines did not reach the worker", unsent))
+		}
+		if len(batch) == 0 {
+			return
+		}
+		body := strings.Join(batch, "\n")
+		batch = batch[:0]
+		// The status too: a 500 loses the batch as a refused connection does.
+		landed := false
+		if resp, err := client.Post(s.url, "text/plain", strings.NewReader(body)); err == nil {
+			landed = resp.StatusCode/100 == 2
+			resp.Body.Close()
+		}
+		if !landed {
+			// Put back, so the next send that works reports one total.
+			s.dropped.Add(dropped)
+			s.unsent.Add(unsent + lines)
+		}
+	}
+
+	for {
+		select {
+		case line := <-s.lines:
+			batch = append(batch, line)
+			if len(batch) >= shipQueue/2 {
+				send()
+			}
+		case <-t.C:
+			send()
+		case ack := <-s.flush:
+			for drained := false; !drained; {
+				select {
+				case line := <-s.lines:
+					batch = append(batch, line)
+				default:
+					drained = true
+				}
+			}
+			send()
+			close(ack)
+		}
+	}
+}