path: root/main.go

// 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)
	}
}