| 1 | // The analytics switch pass (spec 2026-09-07 §3.3): resolve |
| 2 | // telemetry.json app → account → default for every (account, app) this |
| 3 | // box serves and hand the edge a fresh switch map. Rides the serial- |
| 4 | // triggered full pass and the five-minute instances loop, so a console |
| 5 | // write lands within seconds and the hourly floor stands underneath. |
| 6 | package host |
| 7 | |
| 8 | import ( |
| 9 | "context" |
| 10 | "log" |
| 11 | "sync" |
| 12 | "sync/atomic" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/registry" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | "github.com/carlosframework/platform/internal/telemetry" |
| 18 | ) |
| 19 | |
| 20 | var analyticsSink atomic.Pointer[func(map[string]bool)] |
| 21 | |
| 22 | // setAnalyticsSink registers where a resolved switch map goes (the edge's |
| 23 | // SetAnalyticsOn), the way setUpstreamSource publishes the prober table. |
| 24 | func setAnalyticsSink(fn func(map[string]bool)) { analyticsSink.Store(&fn) } |
| 25 | |
| 26 | // telemetryReads is what the switch pass remembers between passes while the |
| 27 | // change index is armed (S3 change-index design §5): a gate over the app and |
| 28 | // account scopes, and the last good settings object per key, so a read the |
| 29 | // gate skips, or one that fails, reuses the last good value. A failed read |
| 30 | // marks its scope failed, so the next pass reads it again. |
| 31 | // |
| 32 | // Disarmed, each pass gets a fresh one over a nil watch, and the pass is |
| 33 | // main's (telemetrySwitchesResolve): every app resolves both levels on its |
| 34 | // own, nothing is remembered within or across passes, and a failed read |
| 35 | // leaves only that app off, as before the index. |
| 36 | type telemetryReads struct { |
| 37 | // mu is held across a whole pass, reads included: the fast converge loop |
| 38 | // and the instances loop both run the pass and take turns here. Nothing |
| 39 | // else is locked under it. |
| 40 | mu sync.Mutex |
| 41 | w *release.ChangeWatch |
| 42 | gate *release.ReadGate |
| 43 | last map[string]telemetry.Settings // telemetry.AccountKey / AppKey -> last good object |
| 44 | } |
| 45 | |
| 46 | func newTelemetryReads(w *release.ChangeWatch) *telemetryReads { |
| 47 | return &telemetryReads{w: w, gate: newChangeGateOver(w), last: map[string]telemetry.Settings{}} |
| 48 | } |
| 49 | |
| 50 | // telemetryGate is the process's telemetryReads for the armed watch. It is |
| 51 | // rebuilt whenever the armed watch differs from the one it was built over, |
| 52 | // so a pass that ran before arming never pins a nil watch. |
| 53 | var telemetryGate struct { |
| 54 | mu sync.Mutex |
| 55 | r *telemetryReads |
| 56 | } |
| 57 | |
| 58 | // currentTelemetryReads is the state a pass uses: telemetryGate's while a |
| 59 | // watch is armed, a fresh one otherwise. |
| 60 | func currentTelemetryReads() *telemetryReads { |
| 61 | w := armedChangeWatch.Load() |
| 62 | if w == nil { |
| 63 | return newTelemetryReads(nil) |
| 64 | } |
| 65 | telemetryGate.mu.Lock() |
| 66 | defer telemetryGate.mu.Unlock() |
| 67 | if telemetryGate.r == nil || telemetryGate.r.w != w { |
| 68 | telemetryGate.r = newTelemetryReads(w) |
| 69 | } |
| 70 | return telemetryGate.r |
| 71 | } |
| 72 | |
| 73 | // telemetrySwitches resolves the count switch for each distinct |
| 74 | // (account, app) among the box's routes. Status routes carry no app and |
| 75 | // are skipped. |
| 76 | func telemetrySwitches(ctx context.Context, s release.Store, reg *registry.Registry) (map[string]bool, error) { |
| 77 | return telemetrySwitchesWith(ctx, s, reg, currentTelemetryReads(), time.Now()) |
| 78 | } |
| 79 | |
| 80 | // telemetrySwitchesWith is telemetrySwitches over r at now. Over a watch, |
| 81 | // each account's telemetry.json is read at most once per pass, however many |
| 82 | // of its apps the box serves, and each app's at most once; a read is skipped |
| 83 | // when r's gate says its scope has not moved and r holds a last good value. |
| 84 | // A pair with no good value at either level is left OFF and logged: off is |
| 85 | // the safe side of this switch. Without one (disarmed), it is main's pass. |
| 86 | func telemetrySwitchesWith(ctx context.Context, s release.Store, reg *registry.Registry, r *telemetryReads, now time.Time) (map[string]bool, error) { |
| 87 | if r == nil || r.w == nil { |
| 88 | return telemetrySwitchesResolve(ctx, s, reg) |
| 89 | } |
| 90 | routes, err := reg.List() |
| 91 | if err != nil { |
| 92 | return nil, err |
| 93 | } |
| 94 | r.mu.Lock() |
| 95 | defer r.mu.Unlock() |
| 96 | pass := map[string]error{} // keys this pass has settled, and the error when there is no good value |
| 97 | on := map[string]bool{} |
| 98 | seen := map[string]bool{} |
| 99 | for _, rt := range routes { |
| 100 | if rt.Kind == "status" || rt.App == "" { |
| 101 | continue |
| 102 | } |
| 103 | sqid := release.RouteAccount(rt) |
| 104 | key := sqid + "/" + rt.App |
| 105 | if seen[key] { |
| 106 | continue |
| 107 | } |
| 108 | seen[key] = true |
| 109 | acct, err := r.read(ctx, s, telemetry.AccountKey(sqid), now, pass) |
| 110 | if err != nil { |
| 111 | log.Printf("agent: telemetry resolve %s: %v (analytics stays off)", key, err) |
| 112 | continue |
| 113 | } |
| 114 | app, err := r.read(ctx, s, telemetry.AppKey(sqid, rt.App), now, pass) |
| 115 | if err != nil { |
| 116 | log.Printf("agent: telemetry resolve %s: %v (analytics stays off)", key, err) |
| 117 | continue |
| 118 | } |
| 119 | if telemetry.Fold(acct, app).Count == telemetry.CountEdge { |
| 120 | on[key] = true |
| 121 | } |
| 122 | } |
| 123 | // Objects no route needs any more are forgotten, so a returning app is |
| 124 | // read afresh. |
| 125 | for k := range r.last { |
| 126 | if _, used := pass[k]; !used { |
| 127 | delete(r.last, k) |
| 128 | } |
| 129 | } |
| 130 | return on, nil |
| 131 | } |
| 132 | |
| 133 | // telemetrySwitchesResolve is the pass with no change index armed, exactly |
| 134 | // as before the index: one Resolve (two Gets) per (account, app); a pair |
| 135 | // that fails to resolve is left OFF and logged, and no other pair is. |
| 136 | func telemetrySwitchesResolve(ctx context.Context, s release.Store, reg *registry.Registry) (map[string]bool, error) { |
| 137 | routes, err := reg.List() |
| 138 | if err != nil { |
| 139 | return nil, err |
| 140 | } |
| 141 | on := map[string]bool{} |
| 142 | seen := map[string]bool{} |
| 143 | for _, rt := range routes { |
| 144 | if rt.Kind == "status" || rt.App == "" { |
| 145 | continue |
| 146 | } |
| 147 | sqid := release.RouteAccount(rt) |
| 148 | key := sqid + "/" + rt.App |
| 149 | if seen[key] { |
| 150 | continue |
| 151 | } |
| 152 | seen[key] = true |
| 153 | res, err := telemetry.Resolve(ctx, s, sqid, rt.App) |
| 154 | if err != nil { |
| 155 | log.Printf("agent: telemetry resolve %s: %v (analytics stays off)", key, err) |
| 156 | continue |
| 157 | } |
| 158 | if res.Count == telemetry.CountEdge { |
| 159 | on[key] = true |
| 160 | } |
| 161 | } |
| 162 | return on, nil |
| 163 | } |
| 164 | |
| 165 | // read is key's settings for this pass: settled once per pass in pass. The |
| 166 | // last good value stands in when the gate says the key's scope has not moved; |
| 167 | // otherwise the object is read. A failed read keeps the last good value, logs |
| 168 | // it and marks the scope failed; with no last good value it returns the |
| 169 | // error. |
| 170 | func (r *telemetryReads) read(ctx context.Context, s release.Store, key string, now time.Time, pass map[string]error) (telemetry.Settings, error) { |
| 171 | if err, done := pass[key]; done { |
| 172 | return r.last[key], err |
| 173 | } |
| 174 | scope, gated := release.ChangeScope(key) |
| 175 | gen, due := r.gate.Due(scope, now) |
| 176 | if prev, ok := r.last[key]; ok && gated && !due { |
| 177 | pass[key] = nil |
| 178 | return prev, nil |
| 179 | } |
| 180 | st, _, err := telemetry.Read(ctx, s, key) |
| 181 | if gated { |
| 182 | r.gate.Mark(scope, gen, now, err == nil) |
| 183 | } |
| 184 | if err == nil { |
| 185 | r.last[key] = st |
| 186 | pass[key] = nil |
| 187 | return st, nil |
| 188 | } |
| 189 | if prev, ok := r.last[key]; ok { |
| 190 | log.Printf("agent: telemetry read %s: %v (the last good value stands; retried next pass)", key, err) |
| 191 | pass[key] = nil |
| 192 | return prev, nil |
| 193 | } |
| 194 | pass[key] = err |
| 195 | return telemetry.Settings{}, err |
| 196 | } |
| 197 | |
| 198 | // telemetryPass resolves and pushes; errors are one log line each. |
| 199 | func telemetryPass(ctx context.Context, s release.Store, reg *registry.Registry) { |
| 200 | sink := analyticsSink.Load() |
| 201 | if sink == nil { |
| 202 | return |
| 203 | } |
| 204 | on, err := telemetrySwitches(ctx, s, reg) |
| 205 | if err != nil { |
| 206 | log.Printf("agent: telemetry pass: %v", err) |
| 207 | return |
| 208 | } |
| 209 | (*sink)(on) |
| 210 | } |
| 211 | |
amadan