newHEADmain

1 files changed, 230 insertions, 0 deletions

diff --git a/main.go b/main.go
new file mode 100644
index 0000000..2b7be34
--- /dev/null
+++ b/main.go
@@ -0,0 +1,230 @@
+// gps-clock-dash: a lightweight live dashboard for a GPS + PPS chrony time server.
+package main
+
+import (
+	"context"
+	"embed"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io/fs"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+)
+
+//go:embed web
+var webFS embed.FS
+
+var version = "dev"
+
+// HistPoint is one 1 Hz sample kept for the history charts.
+type HistPoint struct {
+	T    float64  `json:"t"`
+	Sys  *float64 `json:"sys"`  // chrony system time offset
+	PPS  *float64 `json:"pps"`  // last sample offset of the PPS refclock
+	Freq *float64 `json:"freq"` // chrony frequency, ppm
+	Used *int     `json:"used"`
+	Seen *int     `json:"seen"`
+}
+
+type Snapshot struct {
+	Now      float64    `json:"now"`
+	Host     string     `json:"host"`
+	Version  string     `json:"version"`
+	PPSRefID string     `json:"ppsRefId"`
+	Chrony   ChronySnap `json:"chrony"`
+	GPS      GPSSnap    `json:"gps"`
+	Point    HistPoint  `json:"point"`
+}
+
+type hub struct {
+	mu      sync.Mutex
+	clients map[chan []byte]struct{}
+	latest  []byte
+	history []HistPoint
+	maxHist int
+}
+
+func (h *hub) publish(s *Snapshot) {
+	b, err := json.Marshal(s)
+	if err != nil {
+		log.Printf("marshal: %v", err)
+		return
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.latest = b
+	h.history = append(h.history, s.Point)
+	if over := len(h.history) - h.maxHist; over > 0 {
+		h.history = append(h.history[:0:0], h.history[over:]...)
+	}
+	for c := range h.clients {
+		select {
+		case c <- b:
+		default: // slow client; it will get the next one
+		}
+	}
+}
+
+func (h *hub) subscribe() (chan []byte, []byte) {
+	c := make(chan []byte, 2)
+	h.mu.Lock()
+	h.clients[c] = struct{}{}
+	latest := h.latest
+	h.mu.Unlock()
+	return c, latest
+}
+
+func (h *hub) unsubscribe(c chan []byte) {
+	h.mu.Lock()
+	delete(h.clients, c)
+	h.mu.Unlock()
+}
+
+func unixNow() float64 { return float64(time.Now().UnixNano()) / 1e9 }
+
+func main() {
+	listen := flag.String("listen", ":8473", "HTTP listen address")
+	gpsdAddr := flag.String("gpsd", "127.0.0.1:2947", "gpsd address")
+	chronyc := flag.String("chronyc", "chronyc", "path to chronyc")
+	ppsRef := flag.String("pps-refid", "GPS", "chrony refid of the PPS refclock (for the history chart)")
+	histLen := flag.Int("history", 3600, "number of 1-second history samples to keep")
+	showVer := flag.Bool("version", false, "print version and exit")
+	flag.Parse()
+	if *showVer {
+		fmt.Println(version)
+		return
+	}
+
+	host, _ := os.Hostname()
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	gps := newGPSD(*gpsdAddr)
+	go gps.Run(ctx)
+	ch := &chronyClient{path: *chronyc}
+	h := &hub{clients: map[chan []byte]struct{}{}, maxHist: *histLen}
+
+	// Sample once per second, shortly after the top of the second so the
+	// GPS fix for that second has arrived.
+	go func() {
+		for {
+			now := time.Now()
+			next := now.Truncate(time.Second).Add(time.Second + 150*time.Millisecond)
+			if next.Sub(now) > time.Second {
+				next = next.Add(-time.Second)
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(time.Until(next)):
+			}
+			s := &Snapshot{Host: host, Version: version, PPSRefID: *ppsRef}
+			s.Chrony = ch.Poll(ctx)
+			s.GPS = gps.Snapshot()
+			s.Now = unixNow()
+			p := HistPoint{T: float64(time.Now().Unix())}
+			if t := s.Chrony.Tracking; t != nil {
+				p.Sys, p.Freq = &t.SystemOffset, &t.FreqPPM
+			}
+			for i := range s.Chrony.Sources {
+				if src := &s.Chrony.Sources[i]; src.Name == *ppsRef && src.Reach != 0 {
+					p.PPS = &src.Offset
+				}
+			}
+			if s.GPS.Connected {
+				p.Used, p.Seen = s.GPS.USat, s.GPS.NSat
+			}
+			s.Point = p
+			h.publish(s)
+		}
+	}()
+
+	mux := http.NewServeMux()
+	static, _ := fs.Sub(webFS, "web")
+	mux.Handle("GET /", http.FileServer(http.FS(static)))
+
+	mux.HandleFunc("GET /api/time", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Cache-Control", "no-store")
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprintf(w, `{"now":%.6f}`, unixNow())
+	})
+
+	mux.HandleFunc("GET /api/status", func(w http.ResponseWriter, r *http.Request) {
+		h.mu.Lock()
+		b := h.latest
+		h.mu.Unlock()
+		w.Header().Set("Cache-Control", "no-store")
+		w.Header().Set("Content-Type", "application/json")
+		if b == nil {
+			http.Error(w, `{"error":"no data yet"}`, http.StatusServiceUnavailable)
+			return
+		}
+		w.Write(b)
+	})
+
+	mux.HandleFunc("GET /api/history", func(w http.ResponseWriter, r *http.Request) {
+		h.mu.Lock()
+		b, err := json.Marshal(h.history)
+		h.mu.Unlock()
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Cache-Control", "no-store")
+		w.Header().Set("Content-Type", "application/json")
+		w.Write(b)
+	})
+
+	mux.HandleFunc("GET /events", func(w http.ResponseWriter, r *http.Request) {
+		fl, ok := w.(http.Flusher)
+		if !ok {
+			http.Error(w, "streaming unsupported", http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.Header().Set("Cache-Control", "no-cache")
+		w.Header().Set("X-Accel-Buffering", "no") // tell nginx not to buffer
+		c, latest := h.subscribe()
+		defer h.unsubscribe(c)
+		fmt.Fprint(w, "retry: 3000\n\n")
+		if latest != nil {
+			fmt.Fprintf(w, "data: %s\n\n", latest)
+		}
+		fl.Flush()
+		for {
+			select {
+			case <-r.Context().Done():
+				return
+			case b := <-c:
+				if _, err := fmt.Fprintf(w, "data: %s\n\n", b); err != nil {
+					return
+				}
+				fl.Flush()
+			}
+		}
+	})
+
+	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok\n")) })
+
+	srv := &http.Server{
+		Addr: *listen, Handler: mux, ReadHeaderTimeout: 10 * time.Second,
+		BaseContext: func(net.Listener) context.Context { return ctx }, // ends SSE streams on shutdown
+	}
+	go func() {
+		<-ctx.Done()
+		sctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		defer cancel()
+		srv.Shutdown(sctx)
+	}()
+	log.Printf("gps-clock-dash %s listening on %s (gpsd %s)", version, *listen, *gpsdAddr)
+	if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+		log.Fatal(err)
+	}
+}