feat: log structured reconnect diagnostics for kiosk display layers (#2)
Build Ground Station / build (push) Successful in 1m2s
Build Ground Station / build (push) Successful in 1m2s
This commit was merged in pull request #13.
This commit is contained in:
@@ -44,6 +44,28 @@ func buildTarget(name string) ([]display.Layer, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// wireDiagnostics gives every layer the shared structured reconnect hook so
|
||||||
|
// each layer reports its own reconnect boundaries, identified by target and
|
||||||
|
// layer name.
|
||||||
|
func wireDiagnostics(target string, layers []display.Layer) {
|
||||||
|
for i := range layers {
|
||||||
|
layers[i].Options = display.Options{
|
||||||
|
Target: target,
|
||||||
|
Layer: layers[i].Name,
|
||||||
|
Diagnostic: func(d display.Diagnostic) {
|
||||||
|
log.Printf("display reconnect: %s", formatDiagnostic(d))
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// formatDiagnostic renders one reconnect diagnostic as key=value fields so
|
||||||
|
// reconnects stay greppable and machine-parseable in the journal.
|
||||||
|
func formatDiagnostic(d display.Diagnostic) string {
|
||||||
|
return fmt.Sprintf("target=%s layer=%s event=%s error=%q retry_delay=%s",
|
||||||
|
d.Target, d.Layer, d.Event, d.Err, d.RetryDelay)
|
||||||
|
}
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
target := flag.String("target", "x11", "display target: x11 (Viture) today; mipi planned")
|
target := flag.String("target", "x11", "display target: x11 (Viture) today; mipi planned")
|
||||||
flag.Parse()
|
flag.Parse()
|
||||||
@@ -52,11 +74,13 @@ func main() {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatal(err)
|
log.Fatal(err)
|
||||||
}
|
}
|
||||||
|
wireDiagnostics(*target, layers)
|
||||||
|
|
||||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||||
defer stop()
|
defer stop()
|
||||||
|
|
||||||
// Temporary X11 unavailability never exits this process; every layer
|
// Temporary X11 unavailability never exits this process; every layer
|
||||||
// keeps retrying internally until the context is cancelled.
|
// keeps retrying internally until the context is cancelled, reporting
|
||||||
|
// each reconnect boundary through the structured diagnostic hook.
|
||||||
display.RunLayers(ctx, layers)
|
display.RunLayers(ctx, layers)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.piksar.eu/piotr/groundstation/internal/display"
|
"git.piksar.eu/piotr/groundstation/internal/display"
|
||||||
)
|
)
|
||||||
@@ -83,3 +85,41 @@ func TestBuildTargetRejectsUnknownTarget(t *testing.T) {
|
|||||||
t.Fatal("unknown target must fail")
|
t.Fatal("unknown target must fail")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestFormatDiagnosticRendersStructuredFields(t *testing.T) {
|
||||||
|
got := formatDiagnostic(display.Diagnostic{
|
||||||
|
Target: "x11",
|
||||||
|
Layer: "fallback-canvas",
|
||||||
|
Event: display.EventOpenFailed,
|
||||||
|
Err: errors.New("connect X11: refused"),
|
||||||
|
RetryDelay: 500 * time.Millisecond,
|
||||||
|
})
|
||||||
|
want := `target=x11 layer=fallback-canvas event=open_failed error="connect X11: refused" retry_delay=500ms`
|
||||||
|
if got != want {
|
||||||
|
t.Fatalf("formatDiagnostic=%q want %q", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestWireDiagnosticsTagsEveryLayerWithItsName(t *testing.T) {
|
||||||
|
layers, err := buildTarget("x11")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
wireDiagnostics("x11", layers)
|
||||||
|
|
||||||
|
for _, layer := range layers {
|
||||||
|
if layer.Options.Target != "x11" {
|
||||||
|
t.Fatalf("layer %q target=%q want x11", layer.Name, layer.Options.Target)
|
||||||
|
}
|
||||||
|
if layer.Options.Layer != layer.Name {
|
||||||
|
t.Fatalf("layer %q diagnostic layer=%q", layer.Name, layer.Options.Layer)
|
||||||
|
}
|
||||||
|
if layer.Options.Diagnostic == nil {
|
||||||
|
t.Fatalf("layer %q must carry the reconnect diagnostic hook", layer.Name)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The shared hook must be safe to call from any layer goroutine.
|
||||||
|
layers[0].Options.Diagnostic(display.Diagnostic{Target: "x11", Layer: layers[0].Name})
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,182 @@
|
|||||||
|
package display
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// diagnosticCapture collects structured reconnect records from Run.
|
||||||
|
type diagnosticCapture struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
records []Diagnostic
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *diagnosticCapture) record(d Diagnostic) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
c.records = append(c.records, d)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *diagnosticCapture) all() []Diagnostic {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
return append([]Diagnostic(nil), c.records...)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunDiagnosesEachFailedOpenWithStructuredFields(t *testing.T) {
|
||||||
|
session := newFakeSession()
|
||||||
|
var opens int
|
||||||
|
opener := fakeOpener{open: func(int) (Session, error) {
|
||||||
|
opens++
|
||||||
|
if opens <= 2 {
|
||||||
|
return nil, errors.New("x11 down")
|
||||||
|
}
|
||||||
|
return session, nil
|
||||||
|
}}
|
||||||
|
|
||||||
|
capture := &diagnosticCapture{}
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
done <- Run(ctx, opener, NewStatusTemplate("hello"), Options{
|
||||||
|
Target: "x11",
|
||||||
|
Layer: "fallback-canvas",
|
||||||
|
Diagnostic: capture.record,
|
||||||
|
RetryDelay: time.Millisecond,
|
||||||
|
MaxDelay: 2 * time.Millisecond,
|
||||||
|
})
|
||||||
|
}()
|
||||||
|
|
||||||
|
waitFor(t, func() bool { return session.canvas.lineCount() > 0 }, "connect after retries")
|
||||||
|
cancel()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("run: %v", err)
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("run did not stop after cancel")
|
||||||
|
}
|
||||||
|
|
||||||
|
records := capture.all()
|
||||||
|
if len(records) != 2 {
|
||||||
|
t.Fatalf("records=%d want exactly one per failed open (2)", len(records))
|
||||||
|
}
|
||||||
|
first, second := records[0], records[1]
|
||||||
|
if first.Target != "x11" {
|
||||||
|
t.Fatalf("target=%q want x11", first.Target)
|
||||||
|
}
|
||||||
|
if first.Layer != "fallback-canvas" {
|
||||||
|
t.Fatalf("layer=%q want fallback-canvas", first.Layer)
|
||||||
|
}
|
||||||
|
if first.Event != EventOpenFailed {
|
||||||
|
t.Fatalf("event=%q want %q", first.Event, EventOpenFailed)
|
||||||
|
}
|
||||||
|
if first.Err == nil || first.Err.Error() != "x11 down" {
|
||||||
|
t.Fatalf("underlying error %v must be carried in the record", first.Err)
|
||||||
|
}
|
||||||
|
if first.RetryDelay != time.Millisecond {
|
||||||
|
t.Fatalf("first retry delay=%v want %v", first.RetryDelay, time.Millisecond)
|
||||||
|
}
|
||||||
|
if second.RetryDelay != 2*time.Millisecond {
|
||||||
|
t.Fatalf("backed-off retry delay=%v want %v", second.RetryDelay, 2*time.Millisecond)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunDiagnosesBrokenSessionOnceAndRecovers(t *testing.T) {
|
||||||
|
first := newFakeSession()
|
||||||
|
second := newFakeSession()
|
||||||
|
var opens int
|
||||||
|
opener := fakeOpener{open: func(int) (Session, error) {
|
||||||
|
opens++
|
||||||
|
if opens == 1 {
|
||||||
|
return first, nil
|
||||||
|
}
|
||||||
|
return second, nil
|
||||||
|
}}
|
||||||
|
|
||||||
|
capture := &diagnosticCapture{}
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
done <- Run(ctx, opener, NewStatusTemplate("hello"), Options{
|
||||||
|
Target: "x11",
|
||||||
|
Diagnostic: capture.record,
|
||||||
|
RetryDelay: time.Millisecond,
|
||||||
|
MaxDelay: 2 * time.Millisecond,
|
||||||
|
})
|
||||||
|
}()
|
||||||
|
|
||||||
|
waitFor(t, func() bool { return first.canvas.lineCount() == 1 }, "draw on first session")
|
||||||
|
first.events <- eventResult{err: errors.New("connection lost")}
|
||||||
|
waitFor(t, func() bool { return second.canvas.lineCount() == 1 }, "recover on reconnected session")
|
||||||
|
cancel()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("run: %v", err)
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("run did not stop after cancel")
|
||||||
|
}
|
||||||
|
|
||||||
|
records := capture.all()
|
||||||
|
if len(records) != 1 {
|
||||||
|
t.Fatalf("records=%d want exactly one per broken session (1)", len(records))
|
||||||
|
}
|
||||||
|
record := records[0]
|
||||||
|
if record.Target != "x11" {
|
||||||
|
t.Fatalf("target=%q want x11", record.Target)
|
||||||
|
}
|
||||||
|
if record.Event != EventSessionEnded {
|
||||||
|
t.Fatalf("event=%q want %q", record.Event, EventSessionEnded)
|
||||||
|
}
|
||||||
|
if record.Err == nil || record.Err.Error() != "connection lost" {
|
||||||
|
t.Fatalf("underlying error %v must be carried in the record", record.Err)
|
||||||
|
}
|
||||||
|
if record.RetryDelay != time.Millisecond {
|
||||||
|
t.Fatalf("retry delay=%v want the reset %v", record.RetryDelay, time.Millisecond)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// blockingOpener fails only after the context is done, so cancellation is
|
||||||
|
// the sole reason Run stops: no diagnostic may fire for shutdown.
|
||||||
|
type blockingOpener struct{}
|
||||||
|
|
||||||
|
func (blockingOpener) Open(ctx context.Context) (Session, error) {
|
||||||
|
<-ctx.Done()
|
||||||
|
return nil, ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestRunStaysSilentWhenCancellationEndsTheLoop(t *testing.T) {
|
||||||
|
capture := &diagnosticCapture{}
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
done <- Run(ctx, blockingOpener{}, NewStatusTemplate("hello"), Options{
|
||||||
|
Target: "x11",
|
||||||
|
Diagnostic: capture.record,
|
||||||
|
RetryDelay: time.Millisecond,
|
||||||
|
MaxDelay: 2 * time.Millisecond,
|
||||||
|
})
|
||||||
|
}()
|
||||||
|
|
||||||
|
cancel()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("run: %v", err)
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("run did not stop after cancel")
|
||||||
|
}
|
||||||
|
if records := capture.all(); len(records) != 0 {
|
||||||
|
t.Fatalf("records=%v must stay empty: shutdown is not a reconnect boundary", records)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -49,12 +49,47 @@ type Opener interface {
|
|||||||
Open(ctx context.Context) (Session, error)
|
Open(ctx context.Context) (Session, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Reconnect-boundary diagnostic events.
|
||||||
|
const (
|
||||||
|
// EventOpenFailed marks a failed attempt to open the display target.
|
||||||
|
EventOpenFailed = "open_failed"
|
||||||
|
// EventSessionEnded marks a session broken by a target failure.
|
||||||
|
EventSessionEnded = "session_ended"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Diagnostic is one structured record from a reconnect boundary of the
|
||||||
|
// kiosk loop. Exactly one record is emitted per failed open or broken
|
||||||
|
// session; cancellation and shutdown stay silent.
|
||||||
|
type Diagnostic struct {
|
||||||
|
// Target is the display target name, for example "x11".
|
||||||
|
Target string
|
||||||
|
// Layer is the display layer name within the target, for example
|
||||||
|
// "fallback-canvas"; empty for single-layer runs.
|
||||||
|
Layer string
|
||||||
|
// Event names the boundary: EventOpenFailed or EventSessionEnded.
|
||||||
|
Event string
|
||||||
|
// Err is the underlying error from Opener.Open or Session.WaitEvent.
|
||||||
|
Err error
|
||||||
|
// RetryDelay is the wait scheduled after this boundary.
|
||||||
|
RetryDelay time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
// Options tune the kiosk lifecycle loop. Zero values select the defaults.
|
// Options tune the kiosk lifecycle loop. Zero values select the defaults.
|
||||||
type Options struct {
|
type Options struct {
|
||||||
// RetryDelay is the first wait after a failed connect or lost session.
|
// RetryDelay is the first wait after a failed connect or lost session.
|
||||||
RetryDelay time.Duration
|
RetryDelay time.Duration
|
||||||
// MaxDelay bounds the exponential reconnect backoff.
|
// MaxDelay bounds the exponential reconnect backoff.
|
||||||
MaxDelay time.Duration
|
MaxDelay time.Duration
|
||||||
|
// Target names the display target in diagnostics. Empty selects
|
||||||
|
// "unknown".
|
||||||
|
Target string
|
||||||
|
// Layer names the kiosk layer in diagnostics; empty for single-layer
|
||||||
|
// runs.
|
||||||
|
Layer string
|
||||||
|
// Diagnostic receives one structured record per reconnect boundary.
|
||||||
|
// Nil disables diagnostics. It runs on the loop goroutine and must not
|
||||||
|
// block.
|
||||||
|
Diagnostic func(Diagnostic)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (o Options) withDefaults() Options {
|
func (o Options) withDefaults() Options {
|
||||||
@@ -67,10 +102,35 @@ func (o Options) withDefaults() Options {
|
|||||||
return o
|
return o
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (o Options) targetName() string {
|
||||||
|
if o.Target == "" {
|
||||||
|
return "unknown"
|
||||||
|
}
|
||||||
|
return o.Target
|
||||||
|
}
|
||||||
|
|
||||||
|
// diagnose emits one reconnect-boundary record. Cancellation is shutdown,
|
||||||
|
// not a failure, so it stays silent; so does a nil hook.
|
||||||
|
func (o Options) diagnose(ctx context.Context, event string, err error, retryDelay time.Duration) {
|
||||||
|
if o.Diagnostic == nil || err == nil || ctx.Err() != nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
o.Diagnostic(Diagnostic{
|
||||||
|
Target: o.targetName(),
|
||||||
|
Layer: o.Layer,
|
||||||
|
Event: event,
|
||||||
|
Err: err,
|
||||||
|
RetryDelay: retryDelay,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
// Run owns the kiosk display lifecycle until ctx is cancelled. Temporary
|
// Run owns the kiosk display lifecycle until ctx is cancelled. Temporary
|
||||||
// unavailability of the target, for example X11 not being up yet, never
|
// unavailability of the target, for example X11 not being up yet, never
|
||||||
// returns an error: Run retries with bounded backoff and only stops when the
|
// returns an error: Run retries with bounded backoff and only stops when the
|
||||||
// context is cancelled, so the Ground Station supervisor stays up.
|
// context is cancelled, so the Ground Station supervisor stays up. Each
|
||||||
|
// reconnect boundary (failed open, broken session) emits one structured
|
||||||
|
// Diagnostic record through Options.Diagnostic, never through an error
|
||||||
|
// return; cancellation stays silent.
|
||||||
func Run(ctx context.Context, opener Opener, template Template, options Options) error {
|
func Run(ctx context.Context, opener Opener, template Template, options Options) error {
|
||||||
opts := options.withDefaults()
|
opts := options.withDefaults()
|
||||||
delay := opts.RetryDelay
|
delay := opts.RetryDelay
|
||||||
@@ -81,6 +141,7 @@ func Run(ctx context.Context, opener Opener, template Template, options Options)
|
|||||||
|
|
||||||
session, err := opener.Open(ctx)
|
session, err := opener.Open(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
opts.diagnose(ctx, EventOpenFailed, err, delay)
|
||||||
if !sleep(ctx, delay) {
|
if !sleep(ctx, delay) {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -91,9 +152,11 @@ func Run(ctx context.Context, opener Opener, template Template, options Options)
|
|||||||
delay = opts.RetryDelay
|
delay = opts.RetryDelay
|
||||||
canvas := session.Canvas()
|
canvas := session.Canvas()
|
||||||
template.Draw(canvas)
|
template.Draw(canvas)
|
||||||
|
var sessionErr error
|
||||||
for ctx.Err() == nil {
|
for ctx.Err() == nil {
|
||||||
redraw, err := session.WaitEvent(ctx)
|
var redraw bool
|
||||||
if err != nil {
|
redraw, sessionErr = session.WaitEvent(ctx)
|
||||||
|
if sessionErr != nil {
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
if redraw {
|
if redraw {
|
||||||
@@ -105,6 +168,7 @@ func Run(ctx context.Context, opener Opener, template Template, options Options)
|
|||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
opts.diagnose(ctx, EventSessionEnded, sessionErr, delay)
|
||||||
if !sleep(ctx, delay) {
|
if !sleep(ctx, delay) {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user