ALL

Solo para leer y verificar. No puedes copiar, usar, modificar ni redistribuir este código. Aviso completo en la página principal.

tunnel/metrics.go

SHA-256: ee0660cd3b88aa19237eae313441b5f8c771eaa9c8f11c4aa1741b9176375273
294 líneas · 13895 bytes · Ver en GitHub

1package main
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "fmt"
8 "net"
9 "net/http"
10 "runtime"
11 "sync/atomic"
12 "time"
13)
14
15// /metrics and /healthz (the design plan, section 5): loopback only, aggregate counters only. There is NO series per device, per label, per
16// address or per name, by construction: every counter is a member of a family whose words are a closed list written below, and a word that is not in
17// the list is counted as "other". The number of series therefore does not depend on how many devices or clients the node serves, and nothing a client
18// or a device sends can make a new one. The operator's tools read it from the machine (the monitor of WP22 reads the node certificate's not_after and the
19// age of the control plane's keys from here, and the bytes per user come from the devices' own heartbeats, not from the node).
20
21// family is a counter with one label whose values are a closed list of words.
22type family struct {
23 name, help, label string
24 words []string
25 vals []atomic.Int64
26 idx map[string]int
27}
28
29func newFamily(name, help, label string, words []string) *family {
30 f := &family{name: name, help: help, label: label, words: append([]string(nil), words...), idx: map[string]int{}}
31 f.vals = make([]atomic.Int64, len(f.words))
32 for i, w := range f.words {
33 f.idx[w] = i
34 }
35 if _, ok := f.idx["other"]; !ok {
36 f.idx["other"] = len(f.words)
37 f.words = append(f.words, "other")
38 f.vals = make([]atomic.Int64, len(f.words))
39 }
40 return f
41}
42
43func (f *family) add(word string, n int64) {
44 i, ok := f.idx[word]
45 if !ok {
46 i = f.idx["other"]
47 }
48 f.vals[i].Add(n)
49}
50
51func (f *family) inc(word string) { f.add(word, 1) }
52
53// get is the count of a word (0 for one that is not in the list).
54func (f *family) get(word string) int64 {
55 if i, ok := f.idx[word]; ok {
56 return f.vals[i].Load()
57 }
58 return 0
59}
60
61func (f *family) write(b *bytes.Buffer) {
62 fmt.Fprintf(b, "# HELP %s %s\n# TYPE %s counter\n", f.name, f.help, f.name)
63 for i, w := range f.words {
64 fmt.Fprintf(b, "%s{%s=%q} %d\n", f.name, f.label, w, f.vals[i].Load())
65 }
66}
67
68// The closed lists of words.
69var (
70 // why a connection or a stream was refused before it carried anything
71 rejectWords = []string{
72 refuseBanned, refuseRate, refuseConcurrent, "unauth_limit", "unauth_anon_limit", "not_tls", "bad_hello", "hello_too_large", "hello_timeout", "closed_before_hello",
73 "read_error", "no_sni", "bad_sni", "unknown_host", "device_offline", "host_limit", "label_rate", "global_limit", "anon_global_limit", "draining",
74 "gated_streams", "gated_bytes", "attach_timeout", "control_write_failed", "shutting_down", "other",
75 }
76 // why a stream that was served ended
77 streamEndWords = []string{
78 "closed", "idle", "max_age", "byte_cap", "conn_error", "shutdown", "half_close_timeout", "gated_bytes",
79 closeRevoked, closeExpired, closeReleased, closeKilled, closePaused, closeDrain, "other",
80 }
81 authorizeWords = []string{"ok", "refused", "unavailable", "other"}
82 feedWords = []string{feedRevoke, feedPause, feedResume, feedLicenseEnd, feedRelease, feedKill, feedKillClear, "other"}
83 prefixWords = []string{prefixOK, prefixFetch, prefixEmpty, prefixCorrupt, prefixTooLarge, prefixTooMany, prefixBadPrefix, "other"}
84 directionWords = []string{"up", "down", "other"}
85)
86
87// metrics holds the node's counters. Every member is aggregate and every label value comes from a list above or from the observer's tables.
88type metrics struct {
89 rejected *family
90 streamEnd *family
91 bytes *family
92 authorize *family
93 feedEvents *family
94 prefixRefresh *family
95 streamsOpened atomic.Int64
96 gatedCut atomic.Int64 // streams cut because a gated device's byte budget ran out
97 hintsMarked atomic.Int64
98 bansTotal atomic.Int64
99
100 obsOffered *family // groups a ClientHello offered
101 obsVersion *family // what the ServerHello selected
102 obsCipher *family
103 obsGroup *family
104 obsHellos atomic.Int64 // ClientHellos observed
105 obsECH atomic.Int64 // of them, how many carried an ECH extension
106 obsRetry atomic.Int64 // HelloRetryRequests
107 obsUnparsed atomic.Int64 // streams whose first bytes from the device were not a ServerHello the observer follows
108}
109
110func newMetrics() *metrics {
111 return &metrics{
112 rejected: newFamily("all_tunnel_conn_rejected_total", "Connections and streams refused before they carried anything, by reason.", "reason", rejectWords),
113 streamEnd: newFamily("all_tunnel_stream_closed_total", "Client streams that were served and ended, by why.", "reason", streamEndWords),
114 bytes: newFamily("all_tunnel_bytes_total", "Bytes spliced, client to device (up) and device to client (down).", "direction", directionWords),
115 authorize: newFamily("all_tunnel_authorize_total", "Calls to the control plane's authorize, by outcome.", "result", authorizeWords),
116 feedEvents: newFamily("all_tunnel_feed_events_total", "Feed events applied, by type.", "type", feedWords),
117 prefixRefresh: newFamily("all_tunnel_openai_prefix_refresh_total", "Refreshes of the OpenAI prefix list, by outcome.", "result", prefixWords),
118 obsOffered: newFamily("all_tunnel_tls_client_offered_group_total", "ClientHellos that offered a named group (each counted once per hello).", "group", obsGroupWords),
119 obsVersion: newFamily("all_tunnel_tls_selected_version_total", "TLS version a device's ServerHello selected.", "version", obsVersionWords),
120 obsCipher: newFamily("all_tunnel_tls_selected_cipher_total", "Cipher suite a device's ServerHello selected.", "cipher", obsCipherWords),
121 obsGroup: newFamily("all_tunnel_tls_selected_group_total", "Key-exchange group a device's ServerHello selected (none: TLS 1.2).", "group", append(append([]string(nil), obsGroupWords...), obsNoGroup)),
122 }
123}
124
125func writeCounter(b *bytes.Buffer, name, help string, v int64) {
126 fmt.Fprintf(b, "# HELP %s %s\n# TYPE %s counter\n%s %d\n", name, help, name, name, v)
127}
128
129func writeGauge(b *bytes.Buffer, name, help string, v any) {
130 fmt.Fprintf(b, "# HELP %s %s\n# TYPE %s gauge\n%s %v\n", name, help, name, name, v)
131}
132
133// render is the text of /metrics (Prometheus exposition format).
134func (s *Server) renderMetrics() string {
135 m := s.met
136 var b bytes.Buffer
137 now := s.now()
138 fmt.Fprintf(&b, "# HELP all_tunnel_build_info The node's build.\n# TYPE all_tunnel_build_info gauge\nall_tunnel_build_info{version=%q} 1\n", version)
139 writeGauge(&b, "all_tunnel_uptime_seconds", "Seconds since the node started.", int64(now.Sub(s.startedAt).Seconds()))
140 writeGauge(&b, "all_tunnel_goroutines", "Goroutines.", runtime.NumGoroutine())
141
142 writeGauge(&b, "all_tunnel_streams_open", "Client streams being served now.", s.streams.Load())
143 writeCounter(&b, "all_tunnel_streams_opened_total", "Client streams served since start.", m.streamsOpened.Load())
144 writeGauge(&b, "all_tunnel_control_connections", "Control connections of devices, authenticated or not yet.", s.controlCount())
145 writeGauge(&b, "all_tunnel_devices_online", "Devices online (one number: no list of them exists in the metrics).", s.routes.onlineCount())
146 writeGauge(&b, "all_tunnel_unauthenticated_connections", "Connections that have not finished their first message.", s.unauth.Load())
147 writeGauge(&b, "all_tunnel_pump_big_buffers", "32 KiB buffers held by a stream's pump now (an idle stream holds none).", s.bigBuffersInUse.Load())
148 writeGauge(&b, "all_tunnel_unauthenticated_anonymous_connections", "Of those, the ones whose source is not ChatGPT's, not a device's and not marked good (they share what the reserve leaves).", s.anonUnauth.Load())
149 writeGauge(&b, "all_tunnel_pending_attach", "OPENs waiting for their ATTACH.", s.pendingCount())
150 writeGauge(&b, "all_tunnel_draining", "1 while the node drains (SIGTERM).", boolNum(s.draining.Load()))
151 writeGauge(&b, "all_tunnel_kill_switch", "1 while the control plane's kill switch is on.", boolNum(s.killed.Load()))
152
153 m.rejected.write(&b)
154 m.streamEnd.write(&b)
155 m.bytes.write(&b)
156 m.authorize.write(&b)
157 m.feedEvents.write(&b)
158
159 size, banned, bans, overflowed := s.ips.snapshot()
160 writeGauge(&b, "all_tunnel_ip_table_entries", "Sources remembered (a number: no address leaves the node).", size)
161 writeGauge(&b, "all_tunnel_ip_banned", "Sources dropped at accept now for trying too many names.", banned)
162 writeCounter(&b, "all_tunnel_ip_bans_total", "Sources banned since start.", int64(bans))
163 writeCounter(&b, "all_tunnel_ip_table_overflow_total", "Sources that did not fit in the table and shared the overflow entry.", int64(overflowed))
164 writeGauge(&b, "all_tunnel_hint_addresses", "Addresses devices marked good, all devices together.", s.hints.size())
165 writeCounter(&b, "all_tunnel_hints_total", "HINT messages accepted.", m.hintsMarked.Load())
166 writeGauge(&b, "all_tunnel_gated_records", "Device hosts with a gated-day record.", s.gates.size())
167 writeCounter(&b, "all_tunnel_gated_cut_total", "Streams cut because a gated device's byte budget ran out.", m.gatedCut.Load())
168
169 writeGauge(&b, "all_tunnel_openai_prefixes", "Prefixes in the OpenAI list in use.", s.openai.size())
170 writeGauge(&b, "all_tunnel_openai_prefix_age_seconds", "Seconds since the OpenAI list in use was installed (-1: none).", int64(s.openai.age().Seconds()))
171 m.prefixRefresh.write(&b)
172
173 if s.cfg.CertNotAfter != nil {
174 if na := s.cfg.CertNotAfter(); !na.IsZero() {
175 writeGauge(&b, "all_tunnel_node_cert_not_after_seconds", "The end of the node's own certificate, Unix seconds (the monitor alerts under 14 and 5 days).", na.Unix())
176 writeGauge(&b, "all_tunnel_node_cert_seconds_left", "Seconds the node's own certificate has left.", int64(na.Sub(now).Seconds()))
177 }
178 }
179 if kh, ok := s.cfg.Keys.(interface{ Age() time.Duration }); ok {
180 writeGauge(&b, "all_tunnel_jwks_age_seconds", "Seconds since the control plane's keys were last fetched (-1: never).", int64(kh.Age().Seconds()))
181 }
182 if kh, ok := s.cfg.Keys.(keyHealth); ok {
183 writeGauge(&b, "all_tunnel_jwks_healthy", "1 while the node can vouch for the control plane's keys.", boolNum(kh.Healthy()))
184 }
185 writeGauge(&b, "all_tunnel_control_plane_down", "1 while nothing has answered from the control plane for the policy's quiet period.", boolNum(s.health.down(now, s.cfg.DownQuiet)))
186
187 writeCounter(&b, "all_tunnel_tls_client_hellos_observed_total", "ClientHellos the handshake observer read.", m.obsHellos.Load())
188 m.obsOffered.write(&b)
189 writeCounter(&b, "all_tunnel_tls_client_ech_total", "ClientHellos that carried an ECH extension.", m.obsECH.Load())
190 m.obsVersion.write(&b)
191 m.obsCipher.write(&b)
192 m.obsGroup.write(&b)
193 writeCounter(&b, "all_tunnel_tls_hello_retry_total", "HelloRetryRequests seen from devices.", m.obsRetry.Load())
194 writeCounter(&b, "all_tunnel_tls_server_hello_unparsed_total", "Streams whose first bytes from the device were not a ServerHello the observer follows.", m.obsUnparsed.Load())
195 return b.String()
196}
197
198func boolNum(v bool) int {
199 if v {
200 return 1
201 }
202 return 0
203}
204
205// healthState is the answer of /healthz: whether the node serves, in one word. A node that drains, has its kill switch on, or cannot vouch for the control
206// plane's keys is not serving new devices, and says so with 503; a control plane that is down is reported but is 200, because what is connected goes on.
207func (s *Server) healthState() (code int, body map[string]any) {
208 state := "ok"
209 switch {
210 case s.draining.Load():
211 state = "draining"
212 case s.killed.Load():
213 state = "kill_switch"
214 }
215 if kh, ok := s.cfg.Keys.(keyHealth); ok && state == "ok" && !kh.Healthy() {
216 state = "keys_stale"
217 }
218 code = http.StatusOK
219 if state != "ok" {
220 code = http.StatusServiceUnavailable
221 }
222 return code, map[string]any{"ok": state == "ok", "state": state, "control_plane_down": s.health.down(s.now(), s.cfg.DownQuiet)}
223}
224
225// metricsServer is the loopback HTTP listener of /metrics and /healthz.
226type metricsServer struct {
227 ln net.Listener
228 srv *http.Server
229}
230
231func (s *Server) startMetrics() error {
232 if s.cfg.MetricsAddr == "" {
233 return nil
234 }
235 ln, err := net.Listen("tcp", s.cfg.MetricsAddr)
236 if err != nil {
237 return fmt.Errorf("metrics listener: %w", err)
238 }
239 mux := http.NewServeMux()
240 handle := func(path string, serve func(w http.ResponseWriter)) {
241 mux.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) {
242 if r.Method != http.MethodGet && r.Method != http.MethodHead {
243 w.Header().Set("allow", "GET, HEAD")
244 http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
245 return
246 }
247 if host, _, err := net.SplitHostPort(r.RemoteAddr); err != nil || !net.ParseIP(host).IsLoopback() {
248 http.NotFound(w, r) // the listener is loopback already; a connection that is not is not served
249 return
250 }
251 serve(w)
252 })
253 }
254 handle("/metrics", func(w http.ResponseWriter) {
255 w.Header().Set("content-type", "text/plain; version=0.0.4; charset=utf-8")
256 _, _ = w.Write([]byte(s.renderMetrics()))
257 })
258 handle("/healthz", func(w http.ResponseWriter) {
259 code, body := s.healthState()
260 raw, _ := json.Marshal(body)
261 w.Header().Set("content-type", "application/json")
262 w.WriteHeader(code)
263 _, _ = w.Write(append(raw, '\n'))
264 })
265 srv := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 10 * time.Second, MaxHeaderBytes: 4 << 10}
266 s.metricsSrv = &metricsServer{ln: ln, srv: srv}
267 go func() { _ = srv.Serve(ln) }()
268 return nil
269}
270
271func (s *Server) stopMetrics() {
272 if s.metricsSrv == nil {
273 return
274 }
275 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
276 defer cancel()
277 _ = s.metricsSrv.srv.Shutdown(ctx)
278 _ = s.metricsSrv.srv.Close()
279}
280
281// MetricsAddr is where /metrics listens ("" when it does not).
282func (s *Server) MetricsAddr() string {
283 if s.metricsSrv == nil {
284 return ""
285 }
286 return s.metricsSrv.ln.Addr().String()
287}
288
289// controlCount is the number of control connections.
290func (s *Server) controlCount() int {
291 s.mu.Lock()
292 defer s.mu.Unlock()
293 return len(s.conns)
294}