diff --git a/CLAUDE.md b/CLAUDE.md index 184a269..59e620c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -42,8 +42,8 @@ maintains it. Memory holds live readings, and only while the roster's grant serves a copy; SQLite holds only the home id, the box key, the signed roster, owners' suspensions, tombstones, APNs registrations, the hashes of - extension tokens when each away box's alert is due, browser notice sent markers and - anonymous daily fleet and energy totals. Browser endpoints and box-encrypted offline + extension tokens when each away box's alert is due, browser notice sent markers, + anonymous daily fleet and energy totals, and the service's own hourly counts. Browser endpoints and box-encrypted offline notices live only in the signed roster. Nothing logs addresses. - **A phone can only narrow, or talk to its own box.** On `/v1/app` it can @@ -70,6 +70,10 @@ maintains it. one, because other parties' clients follow no redirect. - **Counts name nobody.** `/metrics` listens on loopback only, and it and the minute's log line sum over all homes: no home id, device or address. + The registry keeps the same sums an hour at a time for 90 days, and + `/v1/ops` serves them, with the closed fleet and energy day totals, only to + the stats Worker's key. A label is a word fixed in the code + (`TestCountsNameNobody`); never add one that could hold a home. - **A deploy drains.** On SIGTERM every socket, handshakes included, hears 4503 with a spread before the process exits, and no drained box starts `box.unreachable`. A part that holds sockets is a `drainer` in diff --git a/README.md b/README.md index 94db4c5..bda3b0b 100644 --- a/README.md +++ b/README.md @@ -26,8 +26,8 @@ to it yet. - What the service holds follows the newest roster's grant, never `hello`: the readable copy can turn on or off while the uplink stays up. - The registry (SQLite) holds the home id, the box key, the signed roster, - owners' suspensions, APNs registrations, the hashes of extension tokens - and when an away box's alert is due. A box whose owner turns the readable + owners' suspensions, APNs registrations, the hashes of extension tokens, + when an away box's alert is due, and the service's own hourly counts. A box whose owner turns the readable copy off gets that copy deleted and a signed receipt (`purged {what: "copy"}`); a box whose link ends gets its home deleted and a receipt. - **Room to grow.** Each home lives on one shard; an instance that does not @@ -36,7 +36,8 @@ to it yet. one instance holds every home. A deploy closes every socket with 4503 and a spread, so boxes and phones come back over seconds rather than at once. `/metrics` on loopback and one log line a minute count what the service - does without naming a home. `deploy/README.md` has the load numbers and + does without naming a home; the registry keeps the same counts an hour + at a time for 90 days ([Internal operations](#internal-operations)). `deploy/README.md` has the load numbers and when to add a shard. - **A restart forgets no one.** At start the service reads every signed roster back from the registry and checks it with the home's key, so a @@ -585,3 +586,48 @@ OHTTP should be outside GDPR, but reading data from the owner's equipment to make them may still need consent under the ePrivacy rule (Swedish LEK, chapter 9, section 28). If it does, neither the opt-out for boxes without Cloud nor the requirement with Cloud holds. + +## Internal operations + +Sourceful runs the service and the fleet from one internal view: the private +page of stats.ftw.energy, behind Cloudflare Access. It needs no system of +its own. The service keeps its counts; the stats Worker pulls them once an +hour into its D1 database. + +**What the service keeps.** Every hour `internal/metrics` hands the hour to +the registry (`ops_hours`, one row an hour): how much each counter rose, +each gauge's low, mean and high over the minutes sampled, each latency's +buckets, how much of the hour a process ran, and how many processes started +in it. A deploy hands over its part of the hour after the drain, and the +next process adds the rest to the same row. The series are the ones +`/metrics` serves: sockets by kind and by transport (QUIC or WebSocket), +homes held and connected, bytes written to boxes and phones, tun frames and +bytes, drops by reason, push queues and the answers of Apple and of browsers' +push services, MCP requests and tool calls by outcome and their time, +handshake and forward latency, the drain's sockets and the Go runtime. +Hours go after 90 days, at each write and at upkeep. + +**What it serves.** `GET /v1/ops` answers only the key the Worker holds, +`Authorization: Bearer `. The service holds the key's SHA-256 +(`-ops-key-sha256`), which is no secret; without it the route answers 404 +to everyone, and the hours are kept all the same. An answer carries at most +a week of settled hours (`?since=`, five minutes after an hour ends), and the +exact totals of the anonymous fleet and energy reports for closed days: a +fleet day once it has ended, an energy date once no report may name it, as +`/v1/energy/public` has it. Below the public group size of 5 too, but never +an open day, so two answers never differ by one report. `/v1/ops` is +`ftw.cloud-ops/1`; the Worker refuses any field it does not name. + +**Why none of it is about a household.** Each series sums over the whole +process. A label is a word fixed in the code, never a home, a device, a key +or an address (`TestCountsNameNobody`), and no series is made for a home +(`TestAnHourOfCountsNamesNobody`). The fleet and energy days are totals of +reports that carry no id and keep nothing per report. The limits, plainly: +while the fleet is a handful of homes, an hour's phone count or a closed +day's single report says something about those few, and staff who know +them could guess. The feed gives nobody more than the totals, and nobody +outside Access sees them. [privacy/internal-operations.md](privacy/internal-operations.md) +has the argument and the retention in full. + +Per-home data stays where it is: in memory for the home's own phones, and in +support access only when an owner opts in for a ticket. diff --git a/cmd/ftwcloudd/main.go b/cmd/ftwcloudd/main.go index 6a0e1ca..239e65e 100644 --- a/cmd/ftwcloudd/main.go +++ b/cmd/ftwcloudd/main.go @@ -44,6 +44,7 @@ import ( "github.com/srcfl/ftw-cloud/internal/mcp" "github.com/srcfl/ftw-cloud/internal/metrics" "github.com/srcfl/ftw-cloud/internal/oauth" + "github.com/srcfl/ftw-cloud/internal/ops" "github.com/srcfl/ftw-cloud/internal/push" "github.com/srcfl/ftw-cloud/internal/quicdoor" "github.com/srcfl/ftw-cloud/internal/registry" @@ -104,6 +105,7 @@ func serve(args []string) error { shardsPath := fs.String("shards", "", "shard directory, JSON; without it this instance holds every home") statsPolicy := fs.String("statistics-policy", "", "approved statistics policy; empty keeps hourly collection off pending C8 sign-off") metricsAddr := fs.String("metrics-listen", defaultMetricsAddr, "loopback address for /metrics; empty turns it off") + opsKey := fs.String("ops-key-sha256", "", "hex SHA-256 of the key the project stats Worker reads /v1/ops with; empty serves no one") restartRate := fs.Int("restart-rate", defaultRestartRate, "sockets a second the next instance can take after a restart; sets the spread of the redial") restartMin := fs.Duration("restart-spread-min", 15*time.Second, "the shortest spread: longer than the port stays closed while the next container starts") h3 := fs.Bool("h3", true, "also serve HTTP/3 on the same address over UDP, and say so in Alt-Svc") @@ -179,9 +181,17 @@ func serve(args []string) error { if err != nil { return err } + // The service's own hours, kept 90 days for Sourceful's internal view. + opsSvc, err := ops.New(reg, shards.Self()) + if err != nil { + return err + } + if err := opsSvc.UseKey(*opsKey); err != nil { + log.Error("ops-key-sha256 is not 64 hex digits; /v1/ops stays off, and the hours are kept", "err", err) + } tokens := token.New(st, reg, log) tokens.Shards = shards - srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Fleet: fleetSvc, Gateway: gatewaySvc, Statistics: usageSvc, Log: log} + srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Fleet: fleetSvc, Gateway: gatewaySvc, Statistics: usageSvc, Ops: opsSvc, Log: log} // Assistants: the authorization server, with agent keys in ftwkeysd. agentKeys := keys.Dial(*keysSocket) @@ -348,6 +358,10 @@ func serve(args []string) error { } } go metrics.Default.Summarise(ctx, os.Stdout, shards.Self(), time.Minute) + // The same series, an hour at a time, into the registry. A deploy hands + // over the hour so far after the drain, so the drain is in it. + keeper := metrics.Default.NewKeeper(opsSvc, func(err error) { log.Warn("an hour of metrics was not kept", "err", err) }) + go keeper.Run(ctx) // A deploy: stop accepting, tell every box and phone to come back within // a spread the next instance can take, let the queued alerts go, then @@ -382,9 +396,13 @@ func serve(args []string) error { left += d.WaitClosed(time.Until(deadline)) } log.Info("drained for a restart", "sockets", sockets, "still_open", left, "spread", spread) + drains.Inc() + drainedSockets.Add(int64(sockets)) + drainLeftOpen.Add(int64(left)) if left := pushSvc.Flush(3 * time.Second); left > 0 { log.Warn("alerts queued or waiting to go again at exit; they are lost", "count", left) } + keeper.Close() stopPush() }() if door != nil { @@ -425,6 +443,13 @@ type drainer interface { // drainWait bounds the wait for the clients' answers to the drain's close. const drainWait = 2 * time.Second +// What a deploy's drain did, kept in the hour it ends. +var ( + drains = metrics.Default.Counter("ftwcloud_drains_total", "Drains for a restart.") + drainedSockets = metrics.Default.Counter("ftwcloud_drained_sockets_total", "Sockets a drain closed with 4503, handshakes included.") + drainLeftOpen = metrics.Default.Counter("ftwcloud_drain_left_open_total", "Sockets whose clients had not answered the drain's close when it stopped waiting.") +) + // defaultRestartRate is how many sockets a second one instance takes on // the host, with room to spare: deploy/README.md, "Load", has the numbers. const defaultRestartRate = 500 diff --git a/cmd/ftwcloudd/main_test.go b/cmd/ftwcloudd/main_test.go index 2694c31..79736e9 100644 --- a/cmd/ftwcloudd/main_test.go +++ b/cmd/ftwcloudd/main_test.go @@ -1,10 +1,43 @@ package main import ( + "regexp" + "slices" + "strings" "testing" "time" + + "github.com/srcfl/ftw-cloud/internal/metrics" ) +// Every series the service keeps, which /v1/ops keeps for 90 days, carries +// only labels from this list, each a word fixed in the code. A label that +// could hold a home, a device or an address is a change to argue for here. +func TestCountsNameNobody(t *testing.T) { + var b strings.Builder + if _, err := metrics.Default.WriteTo(&b); err != nil { + t.Fatal(err) + } + pair := regexp.MustCompile(`([a-z_]+)="([^"]*)"`) + word := regexp.MustCompile(`^[a-z0-9_]{1,40}$|^[0-9.e+-]+$|^\+Inf$`) + allowed := []string{"kind", "reason", "path", "dir", "result", "transport", "outcome", "le"} + n := 0 + for _, line := range strings.Split(b.String(), "\n") { + if strings.HasPrefix(line, "#") || line == "" { + continue + } + n++ + for _, m := range pair.FindAllStringSubmatch(line, -1) { + if !slices.Contains(allowed, m[1]) || !word.MatchString(m[2]) { + t.Fatalf("label %s=%q in %s", m[1], m[2], line) + } + } + } + if n < 50 { + t.Fatalf("only %d series: is every package's set registered?", n) + } +} + func TestTheRestartSpreadFollowsTheSockets(t *testing.T) { for _, c := range []struct { sockets, rate int diff --git a/deploy/README.md b/deploy/README.md index b36ad15..b6c6194 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -215,6 +215,23 @@ closed port, so the spread's floor can drop to a few seconds. Logs (the `awslogs` log driver, and `logs:CreateLogStream` and `logs:PutLogEvents` for the instance role, which the owner adds), each field is a metric in the `FTWCloud` namespace, by shard. +- **The hours, for 90 days.** The registry keeps the same counts an hour + at a time, and the private page of stats.ftw.energy reads them from + `/v1/ops` once an hour ([README](../README.md#internal-operations)). To + turn the feed on, make a key and give it to both ends: + + ```bash + key=$(openssl rand -base64 32 | tr '+/' '-_' | tr -d '=') + printf %s "$key" | npx wrangler secret put CLOUD_OPS_KEY --config stats/wrangler.jsonc # in srcfl/ftw-webapp + printf %s "$key" | shasum -a 256 | cut -d' ' -f1 # FTW_CLOUD_OPS_KEY_SHA256 + ``` + + Put the hash in `/etc/ftw-cloud/compose.env` as + `FTW_CLOUD_OPS_KEY_SHA256=`; the next deploy serves the feed. The hash is + no secret, the key is: it lives only in the Worker. Until the service has + a hash the Worker logs `awaiting key on Cloud`; while the two differ, a + 401. A hash that is not 64 hex digits is logged at start and leaves the + feed off; the service runs. To rotate, do all three again. - What to watch: `forward_seconds_*_p99` (time inside the service), `drops_*` (any rise), `push_queued_alert`, `homes_connected`, `go_memory_bytes`. diff --git a/deploy/compose.yml b/deploy/compose.yml index cf23143..cef7cc0 100644 --- a/deploy/compose.yml +++ b/deploy/compose.yml @@ -30,6 +30,9 @@ services: - -noise-key=/run/secrets/noise/noise.key - -ohttp-key=/run/secrets/ohttp/ohttp.key - -keys-socket=/run/ftw-cloud-keys/keys.sock + # The SHA-256 of the key the stats Worker reads /v1/ops with: no + # secret. Empty turns the feed off; the hours are kept all the same. + - -ops-key-sha256=${FTW_CLOUD_OPS_KEY_SHA256:-} environment: # /source publishes it, so anyone can compare it with what CI built. FTW_CLOUD_IMAGE: ${FTW_CLOUD_IMAGE} diff --git a/internal/appsession/appsession.go b/internal/appsession/appsession.go index 75f122e..22678ba 100644 --- a/internal/appsession/appsession.go +++ b/internal/appsession/appsession.go @@ -94,6 +94,8 @@ type Exchanger interface { // What the app sessions count, summed over all homes. var ( appSockets = metrics.Sockets("app") + appBytesOut = metrics.BytesOut("app") + appOver = map[string]*metrics.Gauge{cloudwire.TransportQUIC: metrics.TransportSockets("app", cloudwire.TransportQUIC), cloudwire.TransportWebSocket: metrics.TransportSockets("app", cloudwire.TransportWebSocket)} appHandshake = metrics.Handshake("app") onceHandshake = metrics.Handshake("once") envelopeOut = metrics.Forward("envelope") @@ -221,6 +223,10 @@ func (s *Server) ServeConn(conn wire.Conn, home string) { defer conn.Close() appSockets.Add(1) defer appSockets.Add(-1) + if over := appOver[wire.Transport(conn)]; over != nil { + over.Add(1) + defer over.Add(-1) + } if !cloudwire.ValidHome(home) { _ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(cloudwire.CloseProtocol, ""), time.Now().Add(time.Second)) return @@ -538,7 +544,11 @@ func (c *phoneConn) writeAs(kind int, msg []byte) error { if err := c.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { return err } - return c.conn.WriteMessage(wire.Kind(c.conn, kind), msg) + if err := c.conn.WriteMessage(wire.Kind(c.conn, kind), msg); err != nil { + return err + } + appBytesOut.Add(int64(len(msg))) + return nil } // forward sends one box envelope as the box sent it. An envelope too large diff --git a/internal/mcp/mcp.go b/internal/mcp/mcp.go index 71121e4..b2763c1 100644 --- a/internal/mcp/mcp.go +++ b/internal/mcp/mcp.go @@ -32,9 +32,22 @@ import ( "sync" "time" + "github.com/srcfl/ftw-cloud/internal/metrics" "github.com/srcfl/ftw-cloud/internal/oauth" ) +// What the service counts of assistants, summed over every home and agent. +var ( + mcpOutcomes = map[string]*metrics.Counter{} + mcpCall = metrics.Default.Histogram("ftwcloud_mcp_call_seconds", "From a tool's start to its result, the box's answer included.") +) + +func init() { + for _, o := range []string{"refused", "rate_limited", "ok", "tool_error", "failed", "unknown_tool", "unsupported", "step_up", "busy"} { + mcpOutcomes[o] = metrics.Default.Counter("ftwcloud_mcp_total", "MCP requests refused before a call, and tool calls, by outcome.", "outcome", o) + } +} + // Versions. const ( // Modern is the revision this server is written to. @@ -196,10 +209,12 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { _ = http.NewResponseController(w).SetReadDeadline(time.Now().Add(10 * time.Second)) grant, aerr := s.auth.Authenticate(r) if aerr != nil { + mcpOutcomes["refused"].Inc() oauth.Challenge(w, profileAuthError(r, aerr)) return } if !s.limit.allow(grant.Subject.Home+"/"+grant.Subject.Agent, s.now()) { + mcpOutcomes["rate_limited"].Inc() w.Header().Set("Retry-After", "1") writeRPCError(w, http.StatusTooManyRequests, nil, codeInternal, "too many requests; slow down", nil) return @@ -375,6 +390,7 @@ type callParams struct { func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage, p callParams, g *oauth.Grant, modern bool) { if unsupportedDiagnostic(p.Name) { + mcpOutcomes["unsupported"].Inc() result := toolError("This connector does not provide this operation (E_UNSUPPORTED). Use measurement_diagnostics, health_get or logs_tail.") result["structuredContent"] = map[string]any{"source": "service", "data": map[string]any{"available": false, "code": "E_UNSUPPORTED", "operation": p.Name, "alternatives": []string{"measurement_diagnostics", "health_get", "logs_tail"}}} writeResult(w, id, modern, result) @@ -385,6 +401,7 @@ func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage t = nil } if t == nil { + mcpOutcomes["unknown_tool"].Inc() status := http.StatusOK if modern { status = http.StatusBadRequest @@ -395,6 +412,7 @@ func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage required := t.requiredScopes() if slices.ContainsFunc(required, func(scope string) bool { return !g.Has(scope) }) { // Step up: the client asks the owner for the right, then retries. + mcpOutcomes["step_up"].Inc() scopes := append(slices.Clone(g.Scopes), required...) slices.Sort(scopes) scopes = slices.Compact(scopes) @@ -410,6 +428,7 @@ func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage case s.inflight <- struct{}{}: defer func() { <-s.inflight }() default: + mcpOutcomes["busy"].Inc() writeResult(w, id, modern, toolError("FTW Cloud is busy. Try again in a moment.")) return } @@ -418,22 +437,29 @@ func (s *Server) call(w http.ResponseWriter, r *http.Request, id json.RawMessage args = json.RawMessage("{}") } ctx := r.Context() + start := time.Now() data, err := t.run(ctx, &callCtx{s: s, grant: g, rpcID: string(id)}, args) + mcpCall.Since(start) if err != nil { var ue userError + outcome := "tool_error" if !errors.As(err, &ue) { s.log.Warn("a tool call failed", "tool", t.name, "err", err) ue = userError("The call failed inside FTW Cloud.") + outcome = "failed" } + mcpOutcomes[outcome].Inc() writeResult(w, id, modern, toolError(string(ue))) return } if failure, ok := data.(failedToolData); ok { + mcpOutcomes["tool_error"].Inc() result := toolResult(g.Subject.Home, failure.data) result["isError"] = true writeResult(w, id, modern, result) return } + mcpOutcomes["ok"].Inc() writeResult(w, id, modern, toolResult(g.Subject.Home, data)) } diff --git a/internal/mcp/mcp_test.go b/internal/mcp/mcp_test.go index e9a550a..a656c8f 100644 --- a/internal/mcp/mcp_test.go +++ b/internal/mcp/mcp_test.go @@ -712,3 +712,18 @@ func TestARestartDrainsAgentSessions(t *testing.T) { t.Fatalf("a drained instance opened a session: %d handshakes", hs) } } + +// Calls are counted by outcome and timed, summed over every home and agent. +func TestCallsAreCountedByOutcome(t *testing.T) { + w := newWorld(t) + b := w.connectBox() + a := w.connect(b, "") + ok, unknown, took := mcpOutcomes["ok"].Load(), mcpOutcomes["unknown_tool"].Load(), mcpCall.Count() + if _, isErr, text := a.call("status", nil); isErr { + t.Fatal(text) + } + _ = a.rpc("tools/call", map[string]any{"name": "no_such_tool"}, false) + if mcpOutcomes["ok"].Load() != ok+1 || mcpOutcomes["unknown_tool"].Load() != unknown+1 || mcpCall.Count() != took+1 { + t.Fatalf("ok %d, unknown %d, timed %d", mcpOutcomes["ok"].Load()-ok, mcpOutcomes["unknown_tool"].Load()-unknown, mcpCall.Count()-took) + } +} diff --git a/internal/metrics/hours.go b/internal/metrics/hours.go new file mode 100644 index 0000000..aacfcf0 --- /dev/null +++ b/internal/metrics/hours.go @@ -0,0 +1,207 @@ +package metrics + +import ( + "context" + "slices" + "sync" + "time" +) + +// Hour is one clock hour of the service, for the operator to keep: how much +// each counter rose, each gauge's low, high and mean over the minutes +// sampled, and each histogram's buckets. Like every series it sums over the +// whole process: no home, device or address. Keys are the names of the +// minute's log line (Summarise). +type Hour struct { + // Start is the hour, in Unix milliseconds, UTC. + Start int64 `json:"start_ms"` + // Ms is how much of the hour a process sampled. Less than an hour + // means the hour held a start, a stop or a crash. + Ms int64 `json:"ms"` + // Starts is how many processes started in the hour: deploys and crashes. + Starts int64 `json:"starts"` + Counters map[string]int64 `json:"counters"` + Gauges map[string]GaugeHour `json:"gauges"` + Histograms map[string][]int64 `json:"histograms"` +} + +// GaugeHour is a gauge over an hour: its low, its high, and the sum and +// number of its samples. +type GaugeHour struct { + Min int64 `json:"min"` + Max int64 `json:"max"` + Sum int64 `json:"sum"` + N int64 `json:"n"` +} + +func newHour(t time.Time) Hour { + return Hour{Start: t.UTC().Truncate(time.Hour).UnixMilli(), Counters: map[string]int64{}, + Gauges: map[string]GaugeHour{}, Histograms: map[string][]int64{}} +} + +// Merge adds o, a later part of the same hour, as when a process restarts +// inside it. +func (h *Hour) Merge(o Hour) { + h.Ms += o.Ms + h.Starts += o.Starts + for k, v := range o.Counters { + h.Counters[k] += v + } + for k, g := range o.Gauges { + h.Gauges[k] = h.Gauges[k].merge(g) + } + for k, b := range o.Histograms { + old := h.Histograms[k] + if len(old) != len(b) { + // The buckets changed between versions: the later ones win. + h.Histograms[k] = slices.Clone(b) + continue + } + for i := range b { + old[i] += b[i] + } + } +} + +func (g GaugeHour) merge(o GaugeHour) GaugeHour { + if g.N == 0 { + return o + } + if o.N == 0 { + return g + } + return GaugeHour{Min: min(g.Min, o.Min), Max: max(g.Max, o.Max), Sum: g.Sum + o.Sum, N: g.N + o.N} +} + +func (g GaugeHour) add(v int64) GaugeHour { return g.merge(GaugeHour{Min: v, Max: v, Sum: v, N: 1}) } + +// Quantile is the upper bound, in seconds, of the LatencyBounds bucket that +// holds the q-th quantile of counts, +Inf past the last bound, or 0 without +// any. +func Quantile(counts []int64, q float64) float64 { return quantile(LatencyBounds, counts, q) } + +// HourStore keeps hours. PutHour merges h into the hour already kept for +// h.Start, if there is one. +type HourStore interface { + PutHour(h Hour) error +} + +// Keeper rolls the series up into hours and hands each to a store when it +// ends. Run samples it on each whole minute; Close hands over the hour it is +// in, which a deploy calls after the drain so the drain's counts are in it. +type Keeper struct { + set *Set + store HourStore + failed func(error) + now func() time.Time + + mu sync.Mutex + last state + at time.Time + hour Hour + closed bool +} + +// NewKeeper starts the hour now, counting this process's start in it. +// failed hears each hour the store would not take, which is then lost. +func (s *Set) NewKeeper(store HourStore, failed func(error)) *Keeper { + k := &Keeper{set: s, store: store, failed: failed, now: time.Now} + k.begin() + return k +} + +func (k *Keeper) begin() { + k.at = k.now() + k.last = k.set.state() + k.hour = newHour(k.at) + k.hour.Starts = 1 + k.gauges(k.last) +} + +// Run samples on each whole minute until ctx ends. The minute stays the +// unit after a slow tick, so no interval crosses an hour's end. +func (k *Keeper) Run(ctx context.Context) { + for { + now := k.now() + t := time.NewTimer(now.Truncate(time.Minute).Add(time.Minute).Sub(now)) + select { + case <-ctx.Done(): + t.Stop() + return + case <-t.C: + k.sample() + } + } +} + +// sample adds the time since the last sample to the hour it began in, and +// hands that hour over once the clock has left it. +func (k *Keeper) sample() { + k.mu.Lock() + defer k.mu.Unlock() + if k.closed { + return + } + now, cur := k.now(), k.set.state() + k.rise(cur, now) + if start := now.UTC().Truncate(time.Hour).UnixMilli(); start != k.hour.Start { + k.put(k.hour) + k.hour = newHour(now) + } + k.gauges(cur) +} + +// Close hands over the hour so far and stops the keeper. +func (k *Keeper) Close() { + k.mu.Lock() + defer k.mu.Unlock() + if k.closed { + return + } + k.closed = true + k.rise(k.set.state(), k.now()) + k.put(k.hour) +} + +// rise adds each counter's and histogram's rise since the last sample. +func (k *Keeper) rise(cur state, now time.Time) { + k.hour.Ms += max(0, now.Sub(k.at).Milliseconds()) + for x, v := range cur { + before := k.last[x] + switch x.kind { + case counter: + var prev int64 + if len(before) == 1 { + prev = before[0] + } + k.hour.Counters[field(x)] += v[0] - prev + case histogram: + b := k.hour.Histograms[field(x)] + if len(b) != len(v) { + b = make([]int64, len(v)) + } + for i := range v { + b[i] += v[i] + if i < len(before) { + b[i] -= before[i] + } + } + k.hour.Histograms[field(x)] = b + } + } + k.last, k.at = cur, now +} + +func (k *Keeper) gauges(cur state) { + for x, v := range cur { + if x.kind == gauge { + k.hour.Gauges[field(x)] = k.hour.Gauges[field(x)].add(v[0]) + } + } +} + +func (k *Keeper) put(h Hour) { + if err := k.store.PutHour(h); err != nil && k.failed != nil { + k.failed(err) + } +} diff --git a/internal/metrics/hours_test.go b/internal/metrics/hours_test.go new file mode 100644 index 0000000..a357881 --- /dev/null +++ b/internal/metrics/hours_test.go @@ -0,0 +1,129 @@ +package metrics + +import ( + "errors" + "testing" + "time" +) + +type hours struct{ got []Hour } + +func (h *hours) PutHour(x Hour) error { + h.got = append(h.got, x) + return nil +} + +func at(hh, mm int) time.Time { return time.Date(2026, 10, 24, hh, mm, 0, 0, time.UTC) } + +// An hour holds each counter's rise, each gauge's low, high and mean over the +// minutes sampled, and each histogram's buckets; it goes to the store once +// the clock has left it, and Close hands over the one it is in. +func TestTheKeeperRollsTheSeriesUpIntoHours(t *testing.T) { + s := &Set{} + c := s.Counter("ftwcloud_drops_total", "", "reason", "x") + g := s.Gauge("ftwcloud_sockets", "", "kind", "box") + h := s.Histogram("ftwcloud_forward_seconds", "", "path", "tun_up") + c.Add(7) // before this process's keeper: not in its hours + g.Add(2) + + now := at(5, 10) + store := &hours{} + k := &Keeper{set: s, store: store, now: func() time.Time { return now }} + k.begin() + + c.Add(3) + now = at(5, 11) + k.sample() + g.Add(3) + for range 10 { + h.Observe(2 * time.Millisecond) + } + now = at(5, 12) + k.sample() + now = at(6, 0) + k.sample() + if len(store.got) != 1 { + t.Fatalf("%d hours handed over", len(store.got)) + } + five := store.got[0] + if five.Start != at(5, 0).UnixMilli() || five.Ms != (50*time.Minute).Milliseconds() || five.Starts != 1 { + t.Fatalf("hour %+v", five) + } + if five.Counters["drops_x"] != 3 { + t.Fatalf("counters %v", five.Counters) + } + if g := five.Gauges["sockets_box"]; g != (GaugeHour{Min: 2, Max: 5, Sum: 9, N: 3}) { + t.Fatalf("gauge %+v", g) + } + if b := five.Histograms["forward_seconds_tun_up"]; Quantile(b, 0.99) != 0.0025 || sum(b) != 10 { + t.Fatalf("buckets %v", b) + } + + c.Inc() + now = at(6, 30) + k.Close() + k.sample() + k.Close() + if len(store.got) != 2 { + t.Fatalf("%d hours handed over", len(store.got)) + } + six := store.got[1] + if six.Start != at(6, 0).UnixMilli() || six.Ms != (30*time.Minute).Milliseconds() || six.Starts != 0 || + six.Counters["drops_x"] != 1 || six.Gauges["sockets_box"] != (GaugeHour{Min: 5, Max: 5, Sum: 5, N: 1}) { + t.Fatalf("hour %+v", six) + } +} + +// A restart inside an hour adds the new process's part to the old one's. +func TestARestartAddsToTheHourItSplit(t *testing.T) { + a := newHour(at(5, 0)) + a.Ms, a.Starts = 40*60_000, 0 + a.Counters["tun_frames_up"] = 100 + a.Gauges["sockets_box"] = GaugeHour{Min: 90, Max: 100, Sum: 950, N: 10} + a.Histograms["handshake_seconds_box"] = []int64{1, 2, 0} + b := newHour(at(5, 0)) + b.Ms, b.Starts = 19*60_000, 1 + b.Counters["tun_frames_up"] = 40 + b.Counters["drains"] = 1 + b.Gauges["sockets_box"] = GaugeHour{Min: 0, Max: 99, Sum: 500, N: 10} + b.Gauges["homes"] = GaugeHour{Min: 3, Max: 3, Sum: 3, N: 1} + b.Histograms["handshake_seconds_box"] = []int64{0, 5, 1} + a.Merge(b) + if a.Ms != 59*60_000 || a.Starts != 1 || a.Counters["tun_frames_up"] != 140 || a.Counters["drains"] != 1 { + t.Fatalf("%+v", a) + } + if a.Gauges["sockets_box"] != (GaugeHour{Min: 0, Max: 100, Sum: 1450, N: 20}) || a.Gauges["homes"].N != 1 { + t.Fatalf("gauges %+v", a.Gauges) + } + if b := a.Histograms["handshake_seconds_box"]; b[0] != 1 || b[1] != 7 || b[2] != 1 { + t.Fatalf("buckets %v", b) + } +} + +// An hour the store refuses is reported, and the keeper goes on. +func TestAnHourTheStoreRefusesIsReported(t *testing.T) { + s := &Set{} + var heard error + now := at(5, 59) + k := &Keeper{set: s, store: refusing{}, failed: func(err error) { heard = err }, now: func() time.Time { return now }} + k.begin() + now = at(6, 0) + k.sample() + if heard == nil { + t.Fatal("the refusal went unheard") + } + if k.hour.Start != at(6, 0).UnixMilli() { + t.Fatal("the keeper stopped at the refusal") + } +} + +type refusing struct{} + +func (refusing) PutHour(Hour) error { return errors.New("disk full") } + +func sum(b []int64) (n int64) { + for _, c := range b { + n += c + } + return n +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index db3bc05..0ceee55 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -72,6 +72,9 @@ func (h *Histogram) Observe(d time.Duration) { // Since counts the time since t. func (h *Histogram) Since(t time.Time) { h.Observe(time.Since(t)) } +// Count is how many durations it has counted. +func (h *Histogram) Count() int64 { return h.n.Load() } + // snapshot is the bucket counts now. func (h *Histogram) snapshot() []int64 { out := make([]int64, len(h.counts)) @@ -83,7 +86,9 @@ func (h *Histogram) snapshot() []int64 { // quantile is the upper bound of the bucket that holds the q-th quantile of // counts, a delta between two snapshots, or 0 without any. -func (h *Histogram) quantile(counts []int64, q float64) float64 { +func (h *Histogram) quantile(counts []int64, q float64) float64 { return quantile(h.bounds, counts, q) } + +func quantile(bounds []float64, counts []int64, q float64) float64 { var total int64 for _, c := range counts { total += c @@ -96,10 +101,10 @@ func (h *Histogram) quantile(counts []int64, q float64) float64 { for i, c := range counts { seen += c if seen >= rank { - if i == len(h.bounds) { + if i == len(bounds) { return math.Inf(1) } - return h.bounds[i] + return bounds[i] } } return math.Inf(1) @@ -270,6 +275,18 @@ func Sockets(kind string) *Gauge { return Default.Gauge("ftwcloud_sockets", "Open sockets, by kind.", "kind", kind) } +// TransportSockets is the open sockets of one kind over one transport, +// quic or websocket. +func TransportSockets(kind, transport string) *Gauge { + return Default.Gauge("ftwcloud_transport_sockets", "Open sockets, by kind and transport.", "kind", kind, "transport", transport) +} + +// BytesOut counts the bytes of the messages the service writes to one kind +// of socket. TLS, WebSocket and QUIC framing come on top. +func BytesOut(kind string) *Counter { + return Default.Counter("ftwcloud_bytes_out_total", "Bytes of messages written to sockets, by kind; TLS, WebSocket and QUIC framing not counted.", "kind", kind) +} + // Handshake times one kind of handshake, from the upgrade or the request to // the answer that ends it. func Handshake(kind string) *Histogram { diff --git a/internal/ops/ops.go b/internal/ops/ops.go new file mode 100644 index 0000000..8c51d13 --- /dev/null +++ b/internal/ops/ops.go @@ -0,0 +1,289 @@ +// Package ops is Sourceful's internal view of the service, for running it and +// for understanding the fleet. It keeps the service's own hours (package +// metrics) for 90 days, and serves them with the exact daily totals of the +// anonymous fleet and energy reports on GET /v1/ops to one key: the one the +// project stats Worker holds. +// +// Nothing in it names a home, a device or an address. Every hour sums over +// the whole process, as every series does. The days are the totals that +// /v1/fleet/public and /v1/energy/public hold to a group size, which keep no +// report; here they come exact, but only once closed, so that two answers +// never differ by one report. +package ops + +import ( + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "encoding/json" + "errors" + "math" + "net/http" + "strings" + "time" + + "github.com/srcfl/ftw-cloud/internal/metrics" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol/energywire" + "github.com/srcfl/ftw/go/protocol/fleetwire" +) + +// Path is where the feed answers. +const Path = "/v1/ops" + +const ( + Schema = "ftw.cloud-ops/1" + Meaning = "sums over the whole service, and closed daily totals over all reports; no home, device or address" + // MaxHours is the most hours one answer carries: a week. + MaxHours = 168 + // settle is how long after its end an hour may still grow: the next + // process adds the rest of an hour a restart split at its first sample. + settle = 5 * time.Minute + // defaultBack is how far back an answer starts without since. + defaultBack = 48 * time.Hour +) + +// Service keeps the hours and serves the feed. +type Service struct { + reg *registry.Registry + key []byte // SHA-256 of the key; nil serves no one + shard string + now func() time.Time +} + +// New keeps hours in reg. Until UseKey takes a key, the feed serves no one, +// and the hours are kept all the same. +func New(reg *registry.Registry, shard string) (*Service, error) { + if err := reg.OpsInit(); err != nil { + return nil, err + } + if err := reg.FleetInit(); err != nil { + return nil, err + } + if err := reg.EnergyInit(); err != nil { + return nil, err + } + if err := reg.OpsPrune(time.Now()); err != nil { + return nil, err + } + return &Service{reg: reg, shard: shard, now: time.Now}, nil +} + +// UseKey makes the feed answer the key whose SHA-256 is keySHA256, in hex, +// which is no secret: it does not help anyone make the key. Spaces, line +// ends and quotes around it, as an env file may leave them, are dropped. +// Empty turns the feed off. A hash that is not 64 hex digits turns it off +// too and is an error to log: a typo in an internal setting must not stop +// the service. +func (s *Service) UseKey(keySHA256 string) error { + s.key = nil + h := strings.Trim(keySHA256, " \t\r\n\"'") + if h == "" { + return nil + } + key, err := hex.DecodeString(h) + if err != nil || len(key) != sha256.Size { + return errors.New("ops: the key's hash is not 64 hex digits") + } + s.key = key + return nil +} + +// PutHour adds h to the hour kept for its start. +func (s *Service) PutHour(h metrics.Hour) error { + return s.reg.OpsPut(h.Start, s.now(), func(old []byte) ([]byte, error) { + if old == nil { + return json.Marshal(h) + } + var kept metrics.Hour + if err := json.Unmarshal(old, &kept); err != nil { + return nil, err + } + kept.Merge(h) + return json.Marshal(kept) + }) +} + +// Prune drops hours past registry.OpsKept. +func (s *Service) Prune(now time.Time) error { return s.reg.OpsPrune(now) } + +// allowed reports whether r carries the key. +func (s *Service) allowed(r *http.Request) bool { + scheme, token, ok := strings.Cut(r.Header.Get("Authorization"), " ") + if !ok || !strings.EqualFold(scheme, "Bearer") || len(token) < 32 { + return false + } + sum := sha256.Sum256([]byte(token)) + return subtle.ConstantTimeCompare(sum[:], s.key) == 1 +} + +// Feed is the whole answer. +type Feed struct { + Schema string `json:"schema"` + Meaning string `json:"meaning"` + Shard string `json:"shard"` + ObservedAt string `json:"observed_at"` + Hours []HourView `json:"hours"` + FleetDays []fleetwire.Day `json:"fleet_days"` + // EnergyDays are the closed dates' energy totals, newest first: the + // date, reports and each of energywire.Fields. + EnergyDays []map[string]any `json:"energy_days"` +} + +// HourView is one settled hour as the feed carries it. +type HourView struct { + Hour string `json:"hour"` + Seconds int64 `json:"seconds"` + Starts int64 `json:"starts"` + Counters map[string]int64 `json:"counters"` + Gauges map[string]GaugeView `json:"gauges"` + LatencyMs map[string]LatencyView `json:"latency_ms"` +} + +// GaugeView is a gauge's low, mean and high over the hour's samples. +type GaugeView struct { + Min int64 `json:"min"` + Mean float64 `json:"mean"` + Max int64 `json:"max"` +} + +// LatencyView is how many durations an hour counted and the upper bounds of +// the buckets that hold its quantiles, in milliseconds. A quantile is null +// when there is none or it lies past the last bucket. +type LatencyView struct { + N int64 `json:"n"` + P50 *float64 `json:"p50"` + P95 *float64 `json:"p95"` + P99 *float64 `json:"p99"` +} + +func view(h metrics.Hour) HourView { + v := HourView{Hour: time.UnixMilli(h.Start).UTC().Format(time.RFC3339), Seconds: int64(math.Round(float64(h.Ms) / 1000)), + Starts: h.Starts, Counters: h.Counters, Gauges: map[string]GaugeView{}, LatencyMs: map[string]LatencyView{}} + if v.Counters == nil { + v.Counters = map[string]int64{} + } + for k, g := range h.Gauges { + if g.N > 0 { + v.Gauges[k] = GaugeView{Min: g.Min, Mean: math.Round(float64(g.Sum)/float64(g.N)*1000) / 1000, Max: g.Max} + } + } + for k, b := range h.Histograms { + var n int64 + for _, c := range b { + n += c + } + l := LatencyView{N: n} + if n > 0 { + l.P50, l.P95, l.P99 = ms(b, 0.5), ms(b, 0.95), ms(b, 0.99) + } + v.LatencyMs[k] = l + } + return v +} + +func ms(counts []int64, q float64) *float64 { + s := metrics.Quantile(counts, q) + if math.IsInf(s, 0) { + return nil + } + v := math.Round(s*1e6) / 1e3 + return &v +} + +// ServeHTTP answers GET /v1/ops?since= to the key, with the settled +// hours from since's hour on, at most MaxHours, and the fleet's days. +func (s *Service) ServeHTTP(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-store") + if r.Method != http.MethodGet { + w.Header().Set("Allow", http.MethodGet) + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + if s.key == nil { + http.NotFound(w, r) + return + } + if !s.allowed(r) { + w.Header().Set("WWW-Authenticate", `Bearer realm="ftw-cloud-ops"`) + w.WriteHeader(http.StatusUnauthorized) + return + } + now := s.now() + from := now.Add(-defaultBack) + if q := r.URL.Query().Get("since"); q != "" { + t, err := time.Parse(time.RFC3339, q) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + from = t + } + from = maxTime(from, now.Add(-registry.OpsKept)).UTC().Truncate(time.Hour) + // An hour is served once it can no longer grow. + until := now.Add(-settle).UTC().Truncate(time.Hour).Add(-time.Hour) + rows, err := s.reg.OpsHours(from.UnixMilli(), until.Add(time.Hour).UnixMilli(), MaxHours) + if err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + fleet, err := s.reg.FleetView(now) + if err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + energy, err := s.energyDays(now) + if err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + feed := Feed{Schema: Schema, Meaning: Meaning, Shard: s.shard, ObservedAt: now.UTC().Format(time.RFC3339), Hours: []HourView{}, FleetDays: []fleetwire.Day{}, EnergyDays: energy} + // A fleet day closes at its end: until then each report adds to it. + today := now.UTC().Format(time.DateOnly) + for _, d := range fleet.Days { + if d.Date < today { + feed.FleetDays = append(feed.FleetDays, d) + } + } + for _, raw := range rows { + var h metrics.Hour + if err := json.Unmarshal(raw, &h); err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + feed.Hours = append(feed.Hours, view(h)) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(feed) +} + +// energyDays are the closed dates of the energy reports, as +// /v1/energy/public names them, without its group size. +func (s *Service) energyDays(now time.Time) ([]map[string]any, error) { + today := now.UTC().Truncate(24 * time.Hour) + first, last := today.AddDate(0, 0, -registry.EnergyDays), today.AddDate(0, 0, -energywire.AcceptDays-1) + totals, err := s.reg.EnergyTotals(first.Format(time.DateOnly), last.Format(time.DateOnly)) + if err != nil { + return nil, err + } + days := []map[string]any{} + for d := last; !d.Before(first); d = d.AddDate(0, 0, -1) { + day, ok := totals[d.Format(time.DateOnly)] + if !ok || day.Reports == 0 { + continue + } + row := map[string]any{"date": day.Date, "reports": day.Reports} + for i, name := range energywire.Fields { + row[name] = day.Sums[i] + } + days = append(days, row) + } + return days, nil +} + +func maxTime(a, b time.Time) time.Time { + if a.After(b) { + return a + } + return b +} diff --git a/internal/ops/ops_test.go b/internal/ops/ops_test.go new file mode 100644 index 0000000..eabfbc9 --- /dev/null +++ b/internal/ops/ops_test.go @@ -0,0 +1,348 @@ +package ops + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "maps" + "net/http" + "net/http/httptest" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/srcfl/ftw-cloud/internal/fleet" + "github.com/srcfl/ftw-cloud/internal/metrics" + "github.com/srcfl/ftw-cloud/internal/registry" + "github.com/srcfl/ftw/go/protocol/energywire" + "github.com/srcfl/ftw/go/protocol/fleetwire" +) + +const key = "a-test-key-of-forty-three-characters-000000" + +func hash(k string) string { + s := sha256.Sum256([]byte(k)) + return hex.EncodeToString(s[:]) +} + +var now = time.Date(2026, 10, 24, 6, 23, 0, 0, time.UTC) + +func newService(t *testing.T, keyHash string) (*Service, *registry.Registry) { + t.Helper() + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reg.Close() }) + s, err := New(reg, "s1") + if err != nil { + t.Fatal(err) + } + if err := s.UseKey(keyHash); err != nil { + t.Fatal(err) + } + s.now = func() time.Time { return now } + return s, reg +} + +func get(s *Service, path, auth string) *httptest.ResponseRecorder { + r := httptest.NewRequest(http.MethodGet, path, nil) + if auth != "" { + r.Header.Set("Authorization", auth) + } + w := httptest.NewRecorder() + s.ServeHTTP(w, r) + return w +} + +func feed(t *testing.T, s *Service, path string) Feed { + t.Helper() + w := get(s, path, "Bearer "+key) + if w.Code != http.StatusOK { + t.Fatalf("%s: %d", path, w.Code) + } + var f Feed + d := json.NewDecoder(w.Body) + d.DisallowUnknownFields() + if err := d.Decode(&f); err != nil { + t.Fatal(err) + } + return f +} + +func hourAt(t time.Time) metrics.Hour { + return metrics.Hour{Start: t.UnixMilli(), Ms: 3_600_000, Counters: map[string]int64{}, Gauges: map[string]metrics.GaugeHour{}, Histograms: map[string][]int64{}} +} + +// The feed answers only the key; without one configured it answers no one. +func TestTheFeedAnswersOnlyTheKey(t *testing.T) { + s, _ := newService(t, hash(key)) + for _, auth := range []string{"", "Bearer", "Bearer " + key[:31], "Bearer " + key + "x", "Basic " + key, key, "Bearer " + key} { + w := get(s, Path, auth) + if w.Code != http.StatusUnauthorized || w.Body.Len() != 0 || w.Header().Get("WWW-Authenticate") == "" { + t.Fatalf("%q: %d %q", auth, w.Code, w.Body.String()) + } + } + if w := get(s, Path, "bearer "+key); w.Code != http.StatusOK { + t.Fatalf("the key: %d", w.Code) + } + r := httptest.NewRequest(http.MethodPost, Path, nil) + r.Header.Set("Authorization", "Bearer "+key) + w := httptest.NewRecorder() + s.ServeHTTP(w, r) + if w.Code != http.StatusMethodNotAllowed { + t.Fatalf("POST: %d", w.Code) + } + + off, _ := newService(t, "") + if w := get(off, Path, "Bearer "+key); w.Code != http.StatusNotFound { + t.Fatalf("no key configured: %d", w.Code) + } + // A hash that is not one is an error to log, and the feed stays off; the + // service still starts. Quotes and a line end an env file leaves are + // not part of the hash. + for _, bad := range []string{"00", strings.Repeat("g", 64), hash(key) + "00", "'" + hash(key)[1:] + "'"} { + if err := s.UseKey(bad); err == nil { + t.Fatalf("a hash of %q was taken", bad) + } + if w := get(s, Path, "Bearer "+key); w.Code != http.StatusNotFound { + t.Fatalf("after %q the feed answers %d", bad, w.Code) + } + } + if err := s.UseKey(`"` + hash(key) + "\"\r\n"); err != nil { + t.Fatal(err) + } + if w := get(s, Path, "Bearer "+key); w.Code != http.StatusOK { + t.Fatalf("a quoted hash with a line end: %d", w.Code) + } +} + +// Two parts of an hour, a restart between them, are served as one hour: +// counters added, gauges' low, mean and high over both, latency quantiles +// from both parts' buckets. +func TestARestartInsideAnHourIsServedAsOneHour(t *testing.T) { + s, _ := newService(t, hash(key)) + five := time.Date(2026, 10, 24, 5, 0, 0, 0, time.UTC) + a := hourAt(five) + a.Ms = 2_400_000 + a.Counters["tun_bytes_down"] = 1000 + a.Gauges["sockets_box"] = metrics.GaugeHour{Min: 9, Max: 10, Sum: 95, N: 10} + b := hourAt(five) + b.Ms, b.Starts = 1_199_600, 1 + b.Counters["tun_bytes_down"] = 500 + b.Counters["drains"] = 1 + b.Gauges["sockets_box"] = metrics.GaugeHour{Min: 0, Max: 10, Sum: 25, N: 5} + bounds := len(metrics.LatencyBounds) + 1 + a.Histograms["handshake_seconds_box"] = make([]int64, bounds) + b.Histograms["handshake_seconds_box"] = make([]int64, bounds) + a.Histograms["handshake_seconds_box"][3] = 90 // 1 ms + b.Histograms["handshake_seconds_box"][9] = 10 // 100 ms + b.Histograms["forward_seconds_tun_up"] = make([]int64, bounds) + b.Histograms["forward_seconds_tun_up"][bounds-1] = 1 // past 10 s + if err := s.PutHour(a); err != nil { + t.Fatal(err) + } + if err := s.PutHour(b); err != nil { + t.Fatal(err) + } + f := feed(t, s, Path) + if f.Schema != Schema || f.Shard != "s1" || f.ObservedAt != "2026-10-24T06:23:00Z" || len(f.Hours) != 1 { + t.Fatalf("%+v", f) + } + h := f.Hours[0] + if h.Hour != "2026-10-24T05:00:00Z" || h.Seconds != 3600 || h.Starts != 1 || h.Counters["tun_bytes_down"] != 1500 || h.Counters["drains"] != 1 { + t.Fatalf("%+v", h) + } + if g := h.Gauges["sockets_box"]; g.Min != 0 || g.Max != 10 || g.Mean != 8 { + t.Fatalf("gauge %+v", g) + } + l := h.LatencyMs["handshake_seconds_box"] + if l.N != 100 || *l.P50 != 1 || *l.P95 != 100 || *l.P99 != 100 { + t.Fatalf("latency %+v", l) + } + if l := h.LatencyMs["forward_seconds_tun_up"]; l.N != 1 || l.P50 != nil || l.P99 != nil { + t.Fatalf("past the last bucket: %+v", l) + } +} + +// An hour is served once it can no longer grow, from since's hour on, at +// most a week of hours an answer, and never one older than 90 days. +func TestTheFeedServesSettledHoursInWeeks(t *testing.T) { + s, reg := newService(t, hash(key)) + first := now.Add(-registry.OpsKept - 2*time.Hour).Truncate(time.Hour) + for h := first; !h.After(now); h = h.Add(time.Hour) { + if err := reg.OpsPut(h.UnixMilli(), first, func([]byte) ([]byte, error) { return json.Marshal(hourAt(h)) }); err != nil { + t.Fatal(err) + } + } + f := feed(t, s, Path) + if len(f.Hours) != 48 || f.Hours[0].Hour != "2026-10-22T06:00:00Z" || f.Hours[47].Hour != "2026-10-24T05:00:00Z" { + t.Fatalf("default: %d hours, %s to %s", len(f.Hours), f.Hours[0].Hour, f.Hours[len(f.Hours)-1].Hour) + } + // At 06:23 the hour that ended at 06:00 is settled; at 06:03 it is not. + now = time.Date(2026, 10, 24, 7, 23, 0, 0, time.UTC) + if f := feed(t, s, Path); f.Hours[len(f.Hours)-1].Hour != "2026-10-24T06:00:00Z" { + t.Fatalf("at 07:23 the last hour is %s", f.Hours[len(f.Hours)-1].Hour) + } + now = time.Date(2026, 10, 24, 7, 3, 0, 0, time.UTC) + if f := feed(t, s, Path); f.Hours[len(f.Hours)-1].Hour != "2026-10-24T05:00:00Z" { + t.Fatalf("at 07:03 the last hour is %s", f.Hours[len(f.Hours)-1].Hour) + } + now = time.Date(2026, 10, 24, 6, 23, 0, 0, time.UTC) + f = feed(t, s, Path+"?since=2020-01-01T00:00:00Z") + if len(f.Hours) != MaxHours || f.Hours[0].Hour != now.Add(-registry.OpsKept).Truncate(time.Hour).Format(time.RFC3339) { + t.Fatalf("since long ago: %d hours from %s", len(f.Hours), f.Hours[0].Hour) + } + if f := feed(t, s, Path+"?since=2026-10-24T03:30:00Z"); len(f.Hours) != 3 || f.Hours[0].Hour != "2026-10-24T03:00:00Z" { + t.Fatalf("since 03:30: %+v", f.Hours) + } + if w := get(s, Path+"?since=yesterday", "Bearer "+key); w.Code != http.StatusBadRequest { + t.Fatalf("a bad since: %d", w.Code) + } +} + +// Hours older than 90 days go, at upkeep and at each write. +func TestHoursGoAfterNinetyDays(t *testing.T) { + s, reg := newService(t, hash(key)) + old := now.Add(-registry.OpsKept - time.Hour).Truncate(time.Hour) + keep := now.Add(-registry.OpsKept + time.Hour).Truncate(time.Hour) + for _, h := range []time.Time{old, keep} { + if err := reg.OpsPut(h.UnixMilli(), old, func([]byte) ([]byte, error) { return json.Marshal(hourAt(h)) }); err != nil { + t.Fatal(err) + } + } + count := func() int { + rows, err := reg.OpsHours(0, now.UnixMilli(), 1000) + if err != nil { + t.Fatal(err) + } + return len(rows) + } + if count() != 2 { + t.Fatal("both hours should be there before upkeep") + } + if err := s.Prune(now); err != nil || count() != 1 { + t.Fatalf("after upkeep: %d %v", count(), err) + } + if err := reg.OpsPut(old.UnixMilli(), old, func([]byte) ([]byte, error) { return json.Marshal(hourAt(old)) }); err != nil { + t.Fatal(err) + } + if err := s.PutHour(hourAt(now.Truncate(time.Hour))); err != nil || count() != 2 { + t.Fatalf("a write should drop the old hour: %d %v", count(), err) + } +} + +// The fleet's and the energy reports' days come exactly as the cloud holds +// them, below the public group size of five too, once closed: a fleet day at +// its end, an energy date once no report may name it. +func TestTheDaysAreExactOnceClosed(t *testing.T) { + s, reg := newService(t, hash(key)) + s.now = time.Now + fl, err := fleet.New(reg) + if err != nil { + t.Fatal(err) + } + raw := `{"schema":"ftw.fleet/1","ftw_version":"v0.155.1-beta.1","channel":"beta","drivers":["sungrow"],"battery_kwh":"5-15","price_zone":"SE3","install_age":"1-6m"}` + report, err := fleetwire.Decode([]byte(raw)) + if err != nil { + t.Fatal(err) + } + yesterday := time.Now().UTC().AddDate(0, 0, -1) + for range 2 { + if ok, err := reg.FleetPut(report, yesterday); !ok || err != nil { + t.Fatal(ok, err) + } + } + w := httptest.NewRecorder() + fl.ServeHTTP(w, httptest.NewRequest(http.MethodPost, "/v1/fleet", strings.NewReader(raw))) + if w.Code != http.StatusNoContent { + t.Fatalf("report: %d", w.Code) + } + closed := time.Now().UTC().AddDate(0, 0, -energywire.AcceptDays-1).Format(time.DateOnly) + open := time.Now().UTC().AddDate(0, 0, -1).Format(time.DateOnly) + for _, r := range []energywire.Report{ + {Schema: energywire.Schema, Date: closed, PVGeneration: 12, GridImport: 30, GridExport: 4, BatteryCharge: 9, BatteryDischarge: 8}, + {Schema: energywire.Schema, Date: closed, PVGeneration: 3, GridImport: 10, VehicleCharge: 7}, + {Schema: energywire.Schema, Date: open, PVGeneration: 1}, + } { + if ok, err := reg.EnergyAdd(r); !ok || err != nil { + t.Fatal(ok, err) + } + } + + f := feed(t, s, Path) + if len(f.FleetDays) != 1 || f.FleetDays[0].Date != yesterday.Format(time.DateOnly) || f.FleetDays[0].Reports != 2 || + f.FleetDays[0].Drivers["sungrow"] != 2 || f.FleetDays[0].Zones["SE3"] != 2 { + t.Fatalf("fleet days %+v", f.FleetDays) + } + if len(f.EnergyDays) != 1 { + t.Fatalf("energy days %+v", f.EnergyDays) + } + want := map[string]any{"date": closed, "reports": 2.0, "pv_generation_kwh": 15.0, "grid_import_kwh": 40.0, "grid_export_kwh": 4.0, + "battery_charge_kwh": 9.0, "battery_discharge_kwh": 8.0, "vehicle_charge_kwh": 7.0} + if !maps.Equal(f.EnergyDays[0], want) { + t.Fatalf("energy day %v", f.EnergyDays[0]) + } + var public struct { + Days []map[string]any `json:"days"` + } + w = httptest.NewRecorder() + fl.Public(w, httptest.NewRequest(http.MethodGet, "/v1/fleet/public", nil)) + _ = json.Unmarshal(w.Body.Bytes(), &public) + for _, d := range public.Days { + if d["reports"] != nil { + t.Fatalf("the public view should hide days below five: %s", w.Body.String()) + } + } +} + +// The feed's top level and each hour hold exactly these fields; a new one +// is a change to review, here and in the Worker's parser. +func TestTheFeedHoldsOnlyItsFields(t *testing.T) { + s, _ := newService(t, hash(key)) + h := hourAt(time.Date(2026, 10, 24, 5, 0, 0, 0, time.UTC)) + h.Counters["drops_x"] = 1 + if err := s.PutHour(h); err != nil { + t.Fatal(err) + } + var top map[string]json.RawMessage + _ = json.Unmarshal(get(s, Path, "Bearer "+key).Body.Bytes(), &top) + if got := keys(top); !slices.Equal(got, []string{"energy_days", "fleet_days", "hours", "meaning", "observed_at", "schema", "shard"}) { + t.Fatalf("top level %v", got) + } + var hours []map[string]json.RawMessage + _ = json.Unmarshal(top["hours"], &hours) + if got := keys(hours[0]); !slices.Equal(got, []string{"counters", "gauges", "hour", "latency_ms", "seconds", "starts"}) { + t.Fatalf("hour %v", got) + } + if err := s.reg.EnergyInit(); err != nil { + t.Fatal(err) + } + if _, err := s.reg.EnergyAdd(energywire.Report{Schema: energywire.Schema, Date: "2026-10-20", GridImport: 1}); err != nil { + t.Fatal(err) + } + var energy struct { + Days []map[string]json.RawMessage `json:"energy_days"` + } + _ = json.Unmarshal(get(s, Path, "Bearer "+key).Body.Bytes(), &energy) + if len(energy.Days) != 1 || !slices.Equal(keys(energy.Days[0]), slices.Sorted(slices.Values(append([]string{"date", "reports"}, energywire.Fields...)))) { + t.Fatalf("energy day %v", energy.Days) + } + var day map[string]json.RawMessage + b, _ := json.Marshal(fleetwire.NewDay("2026-10-24")) + _ = json.Unmarshal(b, &day) + if got := keys(day); !slices.Equal(got, []string{"battery_kwh", "channels", "date", "drivers", "ftw_versions", "install_age", "price_zones", "reports"}) { + t.Fatalf("fleet day %v", got) + } +} + +func keys(m map[string]json.RawMessage) []string { + var out []string + for k := range m { + out = append(out, k) + } + slices.Sort(out) + return out +} diff --git a/internal/registry/ops.go b/internal/registry/ops.go new file mode 100644 index 0000000..5eb425a --- /dev/null +++ b/internal/registry/ops.go @@ -0,0 +1,68 @@ +package registry + +import ( + "database/sql" + "errors" + "time" +) + +// OpsKept is how long the service keeps its own hours (package metrics). +const OpsKept = 90 * 24 * time.Hour + +// OpsInit creates the service's own hours: one row an hour, each a sum over +// the whole service. There is no home column, and none may be added. +func (r *Registry) OpsInit() error { + _, err := r.db.Exec(`CREATE TABLE IF NOT EXISTS ops_hours (hour_ms INTEGER PRIMARY KEY, summary BLOB NOT NULL)`) + return err +} + +// OpsPut stores the hour that starts at hourMS as merge makes it from the +// one stored, nil if none, in one transaction, and drops hours past OpsKept. +func (r *Registry) OpsPut(hourMS int64, now time.Time, merge func(old []byte) ([]byte, error)) error { + tx, err := r.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + var old []byte + err = tx.QueryRow(`SELECT summary FROM ops_hours WHERE hour_ms = ?`, hourMS).Scan(&old) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + data, err := merge(old) + if err != nil { + return err + } + if _, err := tx.Exec(`INSERT INTO ops_hours(hour_ms, summary) VALUES (?, ?) ON CONFLICT(hour_ms) DO UPDATE SET summary=excluded.summary`, hourMS, data); err != nil { + return err + } + if _, err := tx.Exec(`DELETE FROM ops_hours WHERE hour_ms < ?`, now.Add(-OpsKept).UnixMilli()); err != nil { + return err + } + return tx.Commit() +} + +// OpsHours returns at most limit hours that start in [fromMS, toMS), oldest +// first. +func (r *Registry) OpsHours(fromMS, toMS int64, limit int) ([][]byte, error) { + rows, err := r.rd.Query(`SELECT summary FROM ops_hours WHERE hour_ms >= ? AND hour_ms < ? ORDER BY hour_ms LIMIT ?`, fromMS, toMS, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out [][]byte + for rows.Next() { + var data []byte + if err := rows.Scan(&data); err != nil { + return nil, err + } + out = append(out, data) + } + return out, rows.Err() +} + +// OpsPrune drops hours past OpsKept, including when none is written. +func (r *Registry) OpsPrune(now time.Time) error { + _, err := r.db.Exec(`DELETE FROM ops_hours WHERE hour_ms < ?`, now.Add(-OpsKept).UnixMilli()) + return err +} diff --git a/internal/server/ops_e2e_test.go b/internal/server/ops_e2e_test.go new file mode 100644 index 0000000..d0a8848 --- /dev/null +++ b/internal/server/ops_e2e_test.go @@ -0,0 +1,95 @@ +package server + +import ( + "bytes" + "encoding/base64" + "encoding/json" + "maps" + "math" + "slices" + "testing" + + "github.com/srcfl/ftw-cloud/internal/metrics" + "github.com/srcfl/ftw-cloud/internal/ops" + "github.com/srcfl/ftw/go/protocol" + "github.com/srcfl/ftw/go/protocol/appwire" + "github.com/srcfl/ftw/go/protocol/cloudwire" +) + +// An hour of the service's counts, over a box on QUIC and a phone on a +// WebSocket that send a command through the tunnel, holds the traffic and +// names no home, device, key or address. Its fields are the series the +// service had before the traffic: none is made for a home. +func TestAnHourOfCountsNamesNobody(t *testing.T) { + s := newStack(t) + svc, err := ops.New(s.reg, "s1") + if err != nil { + t.Fatal(err) + } + before := &held{} + metrics.Default.NewKeeper(before, nil).Close() + keeper := metrics.Default.NewKeeper(svc, func(err error) { t.Error(err) }) + + box := s.connectTunBoxQUIC() + key, _ := appwire.GenerateKeyPair() + box.roster(2, grantOff, nil, key.Public()) + var phone *tunPhone + eventually(t, "the phone is admitted", func() bool { + phone = s.dialPhone(box.home, key, cloudwire.Now{Hz: cloudwire.HzLooking}, "") + return phone.conn != nil && phone.tr != nil + }) + phone.open(box.static.Public(), 7, nil) + if ack, res := phone.cmd(7, "0192f2a0-7c1e-7000-8000-0000000000a0"); ack != protocol.MsgCmdAck || res != protocol.CmdApplied { + t.Fatalf("ack %q, result %q", ack, res) + } + keeper.Close() + + rows, err := s.reg.OpsHours(0, math.MaxInt64, 10) + if err != nil || len(rows) == 0 { + t.Fatalf("%d hours, %v", len(rows), err) + } + var h metrics.Hour + if err := json.Unmarshal(rows[len(rows)-1], &h); err != nil { + t.Fatal(err) + } + if h.Counters["tun_frames_up"] == 0 || h.Counters["tun_frames_down"] == 0 || h.Histograms["handshake_seconds_box"] == nil { + t.Fatalf("the hour missed the traffic: %v", h.Counters) + } + if g := h.Gauges["transport_sockets_box_quic"]; g.N == 0 { + t.Fatalf("no box over QUIC in the hour: %v", h.Gauges) + } + boxKey, _ := cloudwire.PublicKeyBytesOf(&box.key.PublicKey) + for _, row := range rows { + for _, needle := range []string{box.home, cloudwire.DeviceID(key.Public()), base64.RawURLEncoding.EncodeToString(boxKey), + base64.RawURLEncoding.EncodeToString(key.Public()), base64.RawURLEncoding.EncodeToString(box.static.Public()), "127.0.0.1", "::1"} { + if bytes.Contains(row, []byte(needle)) { + t.Fatalf("an hour holds %q: %s", needle, row) + } + } + } + for name, fields := range map[string][]string{ + "counters": slices.Sorted(maps.Keys(h.Counters)), + "gauges": slices.Sorted(maps.Keys(h.Gauges)), + "histograms": slices.Sorted(maps.Keys(h.Histograms)), + } { + var want []string + switch name { + case "counters": + want = slices.Sorted(maps.Keys(before.got.Counters)) + case "gauges": + want = slices.Sorted(maps.Keys(before.got.Gauges)) + default: + want = slices.Sorted(maps.Keys(before.got.Histograms)) + } + if !slices.Equal(fields, want) { + t.Fatalf("%s changed with the traffic: %v, before %v", name, fields, want) + } + } +} + +type held struct{ got metrics.Hour } + +func (h *held) PutHour(x metrics.Hour) error { + h.got = x + return nil +} diff --git a/internal/server/server.go b/internal/server/server.go index aa50ba8..33ddaa5 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -12,6 +12,8 @@ // /healthz liveness, and nothing about any home // /oauth/*, /.well-known/oauth-* the authorization server for assistants (package oauth) // /mcp the MCP server, for assistants with a token (package mcp) +// /v1/ops the service's own hours and the fleet's exact days, to the +// project stats Worker's key only (package ops) // // The same handler serves HTTP/3 (package quicdoor), where the sockets do not // open: they are WebSockets over TCP. @@ -42,7 +44,7 @@ const SourceArchive = "/usr/share/ftw-cloud/source.tar.gz" // Routes are every path the service answers. var Routes = append([]string{"/v1/box", "/v1/app", "/v1/app/once", "/v1/now", "/v1/fleet", "/v1/fleet/public", - "/v1/ohttp", "/.well-known/ohttp-gateway", "/v1/energy/public", "/v1/stats", "/source", "/healthz", "/mcp"}, oauth.Routes...) + "/v1/ohttp", "/.well-known/ohttp-gateway", "/v1/energy/public", "/v1/stats", "/v1/ops", "/source", "/healthz", "/mcp"}, oauth.Routes...) // devRoutes adds a local inspection route in binaries built with // -tags ftwclouddev. It is nil in every other build. @@ -67,6 +69,8 @@ type Server struct { Statistics http.Handler MCP http.Handler Log *slog.Logger + // Ops is Sourceful's internal feed, to one key; nil serves none. + Ops http.Handler // AltSvc, when set, is the Alt-Svc header that points TCP clients at // HTTP/3 (package quicdoor). It goes on plain responses only: a client // that learns HTTP/3 from a WebSocket upgrade might try the socket over @@ -118,6 +122,11 @@ func (s *Server) Handler() http.Handler { } else { mux.HandleFunc("/v1/stats", notYet) } + if s.Ops != nil { + mux.Handle("/v1/ops", s.Ops) + } else { + mux.HandleFunc("/v1/ops", notYet) + } mux.HandleFunc("/source", source) mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Cache-Control", "no-store") @@ -220,6 +229,11 @@ func (s *Server) Upkeep(ctx context.Context, every time.Duration) { s.Log.Warn("statistics upkeep failed") } } + if f, ok := s.Ops.(interface{ Prune(time.Time) error }); ok { + if err := f.Prune(time.Now()); err != nil { + s.Log.Warn("ops upkeep failed") + } + } if err := s.Registry.TokenExpire(time.Now()); err != nil { s.Log.Warn("token upkeep failed", "err", err) } diff --git a/internal/server/server_test.go b/internal/server/server_test.go index b3a5a43..90c4343 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -16,6 +16,7 @@ import ( "github.com/srcfl/ftw-cloud/internal/appsession" "github.com/srcfl/ftw-cloud/internal/fleet" "github.com/srcfl/ftw-cloud/internal/gateway" + "github.com/srcfl/ftw-cloud/internal/ops" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw-cloud/internal/statistics" "github.com/srcfl/ftw-cloud/internal/store" @@ -51,6 +52,15 @@ func newServer(t *testing.T) (*Server, string) { if s.Gateway, err = gateway.New(reg, testGateway(t)); err != nil { t.Fatal(err) } + // The internal feed with a key set: a request without it reads nothing. + opsSvc, err := ops.New(reg, "s1") + if err != nil { + t.Fatal(err) + } + if err := opsSvc.UseKey("9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08"); err != nil { + t.Fatal(err) + } + s.Ops = opsSvc up.Statistics = true s.Statistics.(*statistics.Service).Enabled = true // A home with readings in it, held the way a box's uplink holds them. diff --git a/internal/uplink/uplink.go b/internal/uplink/uplink.go index 0856385..eb2760a 100644 --- a/internal/uplink/uplink.go +++ b/internal/uplink/uplink.go @@ -82,6 +82,7 @@ const ( // What the uplink counts, summed over all homes. var ( boxSockets = metrics.Sockets("box") + boxOver = map[string]*metrics.Gauge{cloudwire.TransportQUIC: metrics.TransportSockets("box", cloudwire.TransportQUIC), cloudwire.TransportWebSocket: metrics.TransportSockets("box", cloudwire.TransportWebSocket)} boxHandshake = metrics.Handshake("box") tunDown = metrics.Forward("tun_down") wrongShard = metrics.WrongShard("box") @@ -97,6 +98,7 @@ var ( dropConnect = metrics.Drop("box_connect_rate") dropXBusy = metrics.Drop("x_in_flight") dropXUnknown = metrics.Drop("x_unknown") + boxBytesOut = metrics.BytesOut("box") boxesFollow = metrics.Default.Gauge("ftwcloud_boxes_following_shards", "Connected boxes that list shard in hello.caps.") ) @@ -308,6 +310,10 @@ func (u *Uplink) ServeConn(conn wire.Conn, ekm []byte) { conn.SetReadLimit(maxBoxFrame) boxSockets.Add(1) defer boxSockets.Add(-1) + if over := boxOver[wire.Transport(conn)]; over != nil { + over.Add(1) + defer over.Add(-1) + } sock := u.sockets.Add(conn) defer u.sockets.Remove(sock) c := &boxConn{u: u, conn: conn, sock: sock, ekm: ekm, started: time.Now()} @@ -607,8 +613,10 @@ func (c *boxConn) writeFrame(frame []byte) error { } if err != nil { _ = c.conn.Close() + return err } - return err + boxBytesOut.Add(int64(len(frame))) + return nil } // closeAndWait ends the uplink with a close code and reason, and waits for diff --git a/internal/webpush/webpush.go b/internal/webpush/webpush.go index dae0deb..ed3b2e0 100644 --- a/internal/webpush/webpush.go +++ b/internal/webpush/webpush.go @@ -17,11 +17,21 @@ import ( "strings" "time" + "github.com/srcfl/ftw-cloud/internal/metrics" "github.com/srcfl/ftw-cloud/internal/registry" "github.com/srcfl/ftw/go/protocol" "github.com/srcfl/ftw/go/protocol/cloudwire" ) +// Push services' answers per browser, summed over every home. +var results = map[string]*metrics.Counter{} + +func init() { + for _, r := range []string{"ok", "refused", "retry"} { + results[r] = metrics.Default.Counter("ftwcloud_webpush_total", "Push services' answers per browser, by result.", "result", r) + } +} + type Service struct { reg *registry.Registry client *http.Client @@ -172,17 +182,22 @@ func (s *Service) Send(ctx context.Context, home string) (took int, retry bool) req.Header.Set("Authorization", row.Auth) resp, e := s.client.Do(req) if e != nil { + results["retry"].Inc() retry = true continue } _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1024)) _ = resp.Body.Close() if resp.StatusCode == 429 || resp.StatusCode >= 500 { + results["retry"].Inc() retry = true continue } if resp.StatusCode >= 200 && resp.StatusCode < 300 { + results["ok"].Inc() took++ + } else { + results["refused"].Inc() } // Terminal refusals (including expired subscriptions) do not repeat. if s.reg.WebPushMark(home, due, mark) != nil { diff --git a/internal/wire/wire.go b/internal/wire/wire.go index 6e04271..0b5950a 100644 --- a/internal/wire/wire.go +++ b/internal/wire/wire.go @@ -12,6 +12,7 @@ import ( "time" "github.com/gorilla/websocket" + "github.com/srcfl/ftw/go/protocol/cloudwire" ) // Conn is a message connection: gorilla/websocket's method set. @@ -63,3 +64,12 @@ func Kind(conn Conn, want int) int { func Data(messageType int) bool { return messageType == websocket.BinaryMessage || messageType == Bulk || messageType == Datagram } + +// Transport is the registry's name for conn's transport: what a Carrier +// reports, and websocket for a plain WebSocket. +func Transport(conn Conn) string { + if c, ok := conn.(Carrier); ok { + return c.Transport() + } + return cloudwire.TransportWebSocket +} diff --git a/privacy/internal-operations.md b/privacy/internal-operations.md new file mode 100644 index 0000000..8250458 --- /dev/null +++ b/privacy/internal-operations.md @@ -0,0 +1,83 @@ +# Intern driftstatistik för FTW Cloud + +24 oktober 2026. Beskriver vad koden gör. Inte juridiskt granskad. + +Sourceful driver FTW Cloud och behöver förstå driften, felen och flottan för +att kunna förbättra tjänsten. Fredrik bad om det 24 oktober 2026: statistik +för internt bruk, samlad på ett ställe, utan ett nytt system. Den visas i den +privata vyn på stats.ftw.energy. + +## Vad som sparas och hur länge + +| Uppgift | Var | Hur länge | Vem når den | +|---|---|---|---| +| Tjänstens timsummor | Cloud-registret, `ops_hours` | 90 dagar | Workern, med sin nyckel | +| Samma timmar | Workerns D1, `cloud_ops_hours` | 90 dagar | Privata vyn, bakom Access | +| Exakta summor för stängda dagar med flottrapporter | Cloud: `fleet_days`, 90 dagar. D1: senaste vyn | Högst 90 dagar | Privata vyn | +| Exakta summor för stängda datum med energirapporter | Cloud: `energy_days`, 30 dagar. D1: senaste vyn | 30 dagar i Cloud, senaste vyn i D1 | Privata vyn | +| `/metrics` och minutraden, oförändrade | Minnet och Docker-loggen | Till nästa driftsättning | Den som har tillgång till värden | + +En timme håller hur mycket varje räknare steg, varje mätares lägsta, högsta +och medelvärde, varje fördröjnings fördelning, hur stor del av timmen en +process körde och hur många processer som startade. Serierna är de som +`/metrics` redan har: uttag per slag och transport, hem i minnet och +anslutna, byte till boxar och telefoner, tunnelns ramar och byte, tappade +meddelanden per skäl, Apples och webbläsarnas svar på notiser, MCP-anrop per +utfall och deras tid, handskakningar, dräneringar och Go-körtiden. + +En flottdag visas när den är slut, ett energidatum när ingen rapport längre +får nämna det. Därför skiljer sig två hämtningar aldrig med en enda rapport. + +## Vem ser den + +Bara den som Cloudflare Access släpper in på `stats.ftw.energy/admin*` och +`/api/admin*`. Policyn tar med GitHub-organisationen `srcfl`. Workern prövar +dessutom tokenets signatur, utfärdare och publik, så ett fel i Access-regeln +stänger i stället för att öppna. + +`GET /v1/ops` svarar bara den nyckel Workern har som hemlighet +(`CLOUD_OPS_KEY`). Cloud har bara nyckelns SHA-256, som inte är hemlig. Utan +den svarar vägen 404 till alla. Det finns ingen offentlig väg till +timmarna och ingen väg från dem till ett hem. + +## Varför det inte är personuppgifter + +- Varje serie summerar över hela processen. En etikett är ett fast ord i + koden, aldrig ett hem, en enhet, en nyckel eller en adress. + `TestCountsNameNobody` håller listan över tillåtna etiketter, och + `TestAnHourOfCountsNamesNobody` visar att trafik från en box och en + telefon inte skapar någon ny serie och inte lämnar deras id, nycklar eller + adress i timmen. +- Den finaste tiden är en timme. Ingen rad gäller ett hem. +- Flott- och energidagarna är summor av rapporter utan id. Cloud sparar + ingen enskild rapport. Flottrapporten räknas per dimension var för sig, + aldrig som kombination. Energirapporten går genom ett oberoende OHTTP-relä, + så Cloud ser aldrig avsändarens adress. +- Workern tar bara de fält flödet `ftw.cloud-ops/1` namnger och stoppar allt + annat innan något sparas (`stats/src/ops.ts` i srcfl/ftw-webapp). + +## Det som återstår + +Så länge flottan är liten kan en summa ändå säga något om ett fåtal kända +hushåll. En timme med en telefon ansluten klockan tre på natten, eller ett +stängt datum med en enda rapport, beskriver då någon som personalen kanske +känner. Fredrik valde 24 oktober 2026 att visa dagarna exakt internt, utan +den offentliga gruppstorleken fem. Det här är det valets pris. + +Det som håller risken nere: bara Access-skyddad personal ser vyn, inget fält +pekar ut ett hem, tiden är grov, dagar visas först när de är stängda, och +personal ska inte försöka koppla en summa till ett hushåll. Om risken ska +minska mer kan små dagar visas först när de är äldre, eller slås ihop till +veckor. Det är ett beslut för Fredrik. + +## Per-hem-data + +Den stannar där den är: i minnet för hemmets egna telefoner, och i +supportåtkomst bara när ägaren slår på den för ett ärende. Driftstatistiken +läser den inte och ger ingen väg dit. + +## Radering och rättigheter + +Det finns ingen rad per person att lämna ut eller radera. Timmar raderas +efter 90 dagar: i Cloud vid varje skrivning och vid underhåll, i D1 vid +Workerns timkörning. Dagssummorna följer sina egna gränser ovan.