1// Package traffic owns Gordon's runtime traffic entrypoints.
2package traffic
3
4import (
5 "context"
6 "errors"
7 "fmt"
8 "maps"
9 "net"
10 "sort"
11 "strings"
12 "sync"
13 "sync/atomic"
14 "syscall"
15 "time"
16
17 "github.com/bnema/zerowrap"
18 "github.com/rs/zerolog"
19
20 "github.com/bnema/gordon/internal/domain"
21)
22
23const (
24 reloadStatusOK = "ok"
25 reloadStatusError = "error"
26)
27
28func trafficLog(ctx context.Context) zerowrap.Logger {
29 return zerowrap.FromCtxWithFields(ctx, map[string]any{
30 zerowrap.FieldLayer: "adapter",
31 zerowrap.FieldAdapter: "traffic",
32 })
33}
34
35func trafficInfo(ctx context.Context) *zerolog.Event {
36 log := trafficLog(ctx)
37 return log.Info()
38}
39
40func trafficDebug(ctx context.Context) *zerolog.Event {
41 log := trafficLog(ctx)
42 return log.Debug()
43}
44
45func trafficWarn(ctx context.Context) *zerolog.Event {
46 log := trafficLog(ctx)
47 return log.Warn()
48}
49
50// Manager owns runtime entrypoints for the traffic plane.
51type Manager struct {
52 mu sync.Mutex
53 snapshot atomic.Pointer[domain.TrafficGraph]
54 listeners map[string]*entryPointRuntime
55 udpListeners map[string]*udpEntryPointRuntime
56
57 lastReloadStatus string
58 lastReloadError string
59 tlsHTTPServers atomic.Value
60 smartHTTPServers atomic.Value
61 smartTLSServers atomic.Value
62}
63
64// NewManager creates a traffic manager.
65func NewManager() *Manager {
66 manager := &Manager{
67 listeners: map[string]*entryPointRuntime{},
68 udpListeners: map[string]*udpEntryPointRuntime{},
69 lastReloadStatus: reloadStatusOK,
70 }
71 manager.tlsHTTPServers.Store(tlsHTTPServers{})
72 manager.smartHTTPServers.Store(smartHTTPServers{})
73 manager.smartTLSServers.Store(smartTLSServers{})
74 return manager
75}
76
77// Apply validates and applies a new traffic graph snapshot.
78func (m *Manager) Apply(ctx context.Context, graph *domain.TrafficGraph) error {
79 if graph == nil {
80 return m.recordReloadError(ctx, fmt.Errorf("%w: traffic graph is required", domain.ErrTrafficGraphRequired))
81 }
82 if err := graph.Validate(); err != nil {
83 return m.recordReloadError(ctx, fmt.Errorf("validate traffic graph: %w", err))
84 }
85 trafficDebug(ctx).Int("entrypoints", len(graph.EntryPoints)).Int("routers", len(graph.Routers)).Int("services", len(graph.Services)).Msg("applying traffic graph")
86
87 nextGraph := cloneTrafficGraph(*graph)
88 m.mu.Lock()
89 newListeners, createdListeners, tcpUpdates, err := m.prepareTCPListeners(ctx, &nextGraph)
90 if err != nil {
91 m.lastReloadStatus = reloadStatusError
92 m.lastReloadError = err.Error()
93 m.mu.Unlock()
94 trafficWarn(ctx).Err(err).Msg("failed to prepare tcp traffic listeners")
95 return err
96 }
97 newUDPListeners, createdUDPListeners, udpUpdates, err := m.prepareUDPListeners(ctx, &nextGraph)
98 if err != nil {
99 m.lastReloadStatus = reloadStatusError
100 m.lastReloadError = err.Error()
101 stopTCPRuntimes(ctx, createdListeners, effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout)
102 m.mu.Unlock()
103 trafficWarn(ctx).Err(err).Msg("failed to prepare udp traffic listeners")
104 return err
105 }
106
107 for _, update := range tcpUpdates {
108 update.runtime.updateEntryPoint(update.entryPoint, update.trusted, update.rawTrusted)
109 }
110 for _, update := range udpUpdates {
111 update.runtime.updateEntryPoint(update.entryPoint, update.trusted)
112 }
113
114 oldListeners := m.listeners
115 oldUDPListeners := m.udpListeners
116 m.listeners = newListeners
117 m.udpListeners = newUDPListeners
118 m.snapshot.Store(&nextGraph)
119 for _, runtime := range createdListeners {
120 runtime.start()
121 }
122 for _, runtime := range createdUDPListeners {
123 runtime.start()
124 }
125 m.lastReloadStatus = reloadStatusOK
126 m.lastReloadError = ""
127 m.mu.Unlock()
128
129 tcpDrainTimeout := effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout
130 stoppedTCP := 0
131 for _, runtime := range oldListeners {
132 if tcpRuntimeRetained(newListeners, runtime) {
133 continue
134 }
135 stoppedTCP++
136 runtime.stop(ctx, tcpDrainTimeout)
137 }
138 udpDrainTimeout := effectiveUDPOptions(snapshotUDPOptions(&nextGraph)).DrainTimeout
139 stoppedUDP := 0
140 for _, runtime := range oldUDPListeners {
141 if udpRuntimeRetained(newUDPListeners, runtime) {
142 if backend, ok := runtime.resolveUDPBackend(); ok {
143 runtime.drainSessionsNotMatchingAfter(backend, udpDrainTimeout)
144 } else {
145 runtime.drainSessionsAfter(udpDrainTimeout)
146 }
147 continue
148 }
149 stoppedUDP++
150 runtime.stop(ctx, udpDrainTimeout)
151 }
152 logAppliedTrafficGraph(ctx, newListeners, newUDPListeners, createdListeners, createdUDPListeners, stoppedTCP, stoppedUDP)
153 return nil
154}
155
156func tcpRuntimeRetained(listeners map[string]*entryPointRuntime, runtime *entryPointRuntime) bool {
157 for _, candidate := range listeners {
158 if candidate == runtime {
159 return true
160 }
161 }
162 return false
163}
164
165func udpRuntimeRetained(listeners map[string]*udpEntryPointRuntime, runtime *udpEntryPointRuntime) bool {
166 for _, candidate := range listeners {
167 if candidate == runtime {
168 return true
169 }
170 }
171 return false
172}
173
174func logAppliedTrafficGraph(
175 ctx context.Context,
176 newListeners map[string]*entryPointRuntime,
177 newUDPListeners map[string]*udpEntryPointRuntime,
178 createdListeners []*entryPointRuntime,
179 createdUDPListeners []*udpEntryPointRuntime,
180 stoppedTCP int,
181 stoppedUDP int,
182) {
183 logEvent := trafficInfo(ctx)
184 if len(newListeners) == 0 && len(newUDPListeners) == 0 && stoppedTCP == 0 && stoppedUDP == 0 {
185 logEvent = trafficDebug(ctx)
186 }
187 logEvent.
188 Int("tcp_listeners", len(newListeners)).
189 Int("udp_listeners", len(newUDPListeners)).
190 Int("created_tcp_listeners", len(createdListeners)).
191 Int("created_udp_listeners", len(createdUDPListeners)).
192 Int("stopped_tcp_listeners", stoppedTCP).
193 Int("stopped_udp_listeners", stoppedUDP).
194 Msg("traffic graph applied")
195}
196
197// Shutdown stops all listeners and active TCP streams.
198func (m *Manager) Shutdown(ctx context.Context) error {
199 trafficInfo(ctx).Msg("shutting down traffic manager")
200 m.mu.Lock()
201 defer m.mu.Unlock()
202
203 listeners := m.listeners
204 udpListeners := m.udpListeners
205 m.listeners = map[string]*entryPointRuntime{}
206 m.udpListeners = map[string]*udpEntryPointRuntime{}
207 graph := m.snapshot.Load()
208 m.snapshot.Store(nil)
209
210 tcpDrainTimeout := defaultTCPOptions().DrainTimeout
211 udpDrainTimeout := defaultUDPOptions().DrainTimeout
212 if graph != nil {
213 tcpDrainTimeout = effectiveTCPOptions(graph.Options.TCP).DrainTimeout
214 udpDrainTimeout = effectiveUDPOptions(graph.Options.UDP).DrainTimeout
215 }
216 for _, runtime := range listeners {
217 runtime.stop(ctx, tcpDrainTimeout)
218 }
219 for _, runtime := range udpListeners {
220 runtime.stop(ctx, udpDrainTimeout)
221 }
222 trafficInfo(ctx).Int("tcp_listeners", len(listeners)).Int("udp_listeners", len(udpListeners)).Msg("traffic manager shut down")
223 return nil
224}
225
226// Status returns a consistent snapshot of manager state and counters.
227func (m *Manager) Status() domain.TrafficStatus {
228 m.mu.Lock()
229 listeners := make(map[string]*entryPointRuntime, len(m.listeners))
230 maps.Copy(listeners, m.listeners)
231 udpListeners := make(map[string]*udpEntryPointRuntime, len(m.udpListeners))
232 maps.Copy(udpListeners, m.udpListeners)
233 status := domain.TrafficStatus{LastReloadStatus: m.lastReloadStatus, LastReloadError: m.lastReloadError}
234 graph := m.snapshot.Load()
235 m.mu.Unlock()
236
237 if graph == nil {
238 return status
239 }
240
241 status.EntryPoints = entryPointStatuses(graph.EntryPoints, listeners, udpListeners)
242 status.Routers = routerStatuses(graph.Routers, listeners, udpListeners)
243 status.Services = serviceStatuses(graph.Services)
244 status.Counters = aggregateCounters(status.EntryPoints)
245 return status
246}
247
248func (m *Manager) recordReloadError(ctx context.Context, err error) error {
249 m.mu.Lock()
250 m.lastReloadStatus = reloadStatusError
251 m.lastReloadError = err.Error()
252 m.mu.Unlock()
253 trafficWarn(ctx).Err(err).Msg("traffic graph rejected")
254 return err
255}
256
257type tcpRuntimeUpdate struct {
258 runtime *entryPointRuntime
259 entryPoint domain.EntryPoint
260 trusted []*net.IPNet
261 rawTrusted []*net.IPNet
262}
263
264type udpRuntimeUpdate struct {
265 runtime *udpEntryPointRuntime
266 entryPoint domain.EntryPoint
267 trusted []*net.IPNet
268}
269
270func (m *Manager) prepareTCPListeners(ctx context.Context, graph *domain.TrafficGraph) (map[string]*entryPointRuntime, []*entryPointRuntime, []tcpRuntimeUpdate, error) {
271 current := make(map[string]*entryPointRuntime, len(m.listeners))
272 maps.Copy(current, m.listeners)
273
274 next := make(map[string]*entryPointRuntime, len(current))
275 created := []*entryPointRuntime{}
276 updates := []tcpRuntimeUpdate{}
277 for _, entryPoint := range graph.EntryPoints {
278 if !isTCPListenerProtocol(entryPoint.Protocol) {
279 continue
280 }
281 if runtime := current[entryPoint.Name]; runtime != nil && runtime.matches(entryPoint) {
282 next[entryPoint.Name] = runtime
283 delete(current, entryPoint.Name)
284 continue
285 }
286 if runtime := conflictingTCPRuntime(current, entryPoint); runtime != nil {
287 trusted, err := parseTrustedCIDRs(entryPoint.TrustedCIDRs)
288 if err != nil {
289 stopTCPRuntimes(ctx, created, effectiveTCPOptions(graph.Options.TCP).DrainTimeout)
290 return nil, nil, nil, fmt.Errorf("parse trusted cidrs for tcp entrypoint %q: %w", entryPoint.Name, err)
291 }
292 rawTrusted, err := parseTrustedCIDRs(entryPoint.RawFallbackTrustedCIDRs)
293 if err != nil {
294 stopTCPRuntimes(ctx, created, effectiveTCPOptions(graph.Options.TCP).DrainTimeout)
295 return nil, nil, nil, fmt.Errorf("parse raw fallback trusted cidrs for tcp entrypoint %q: %w", entryPoint.Name, err)
296 }
297 trafficDebug(ctx).Str("entrypoint", entryPoint.Name).Str("address", entryPoint.Address).Msg("reusing tcp traffic listener for same-address entrypoint update")
298 updates = append(updates, tcpRuntimeUpdate{runtime: runtime, entryPoint: entryPoint, trusted: trusted, rawTrusted: rawTrusted})
299 next[entryPoint.Name] = runtime
300 delete(current, runtime.entryPointSnapshot().Name)
301 continue
302 }
303 runtime, err := m.bindTCPEntryPoint(ctx, entryPoint)
304 if err != nil {
305 stopTCPRuntimes(ctx, created, effectiveTCPOptions(graph.Options.TCP).DrainTimeout)
306 return nil, nil, nil, err
307 }
308 next[entryPoint.Name] = runtime
309 created = append(created, runtime)
310 }
311 return next, created, updates, nil
312}
313
314func stopTCPRuntimes(ctx context.Context, runtimes []*entryPointRuntime, drainTimeout time.Duration) {
315 for _, runtime := range runtimes {
316 runtime.stop(ctx, drainTimeout)
317 }
318}
319
320func stopUDPRuntimes(ctx context.Context, runtimes []*udpEntryPointRuntime, drainTimeout time.Duration) {
321 for _, runtime := range runtimes {
322 runtime.stop(ctx, drainTimeout)
323 }
324}
325
326func isTCPListenerProtocol(protocol domain.EntryPointProtocol) bool {
327 return protocol == domain.EntryPointProtocolTCP || protocol == domain.EntryPointProtocolTLSMux || protocol == domain.EntryPointProtocolSmartTCP
328}
329
330func conflictingTCPRuntime(current map[string]*entryPointRuntime, entryPoint domain.EntryPoint) *entryPointRuntime {
331 for _, runtime := range current {
332 if runtime.sameAddress(entryPoint) {
333 return runtime
334 }
335 }
336 return nil
337}
338
339func (m *Manager) bindTCPEntryPoint(ctx context.Context, entryPoint domain.EntryPoint) (*entryPointRuntime, error) {
340 listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", entryPoint.Address)
341 if err != nil {
342 if errors.Is(err, syscall.EADDRINUSE) {
343 return nil, fmt.Errorf("bind tcp entrypoint %q on %s: inspect host-level TCP listeners, including other users or Gordon instances: %w", entryPoint.Name, entryPoint.Address, err)
344 }
345 return nil, fmt.Errorf("bind tcp entrypoint %q on %s: %w", entryPoint.Name, entryPoint.Address, err)
346 }
347 trusted, err := parseTrustedCIDRs(entryPoint.TrustedCIDRs)
348 if err != nil {
349 _ = listener.Close()
350 return nil, fmt.Errorf("parse trusted cidrs for tcp entrypoint %q: %w", entryPoint.Name, err)
351 }
352 rawTrusted, err := parseTrustedCIDRs(entryPoint.RawFallbackTrustedCIDRs)
353 if err != nil {
354 _ = listener.Close()
355 return nil, fmt.Errorf("parse raw fallback trusted cidrs for tcp entrypoint %q: %w", entryPoint.Name, err)
356 }
357 trafficInfo(ctx).Str("entrypoint", entryPoint.Name).Str("address", entryPoint.Address).Str("protocol", string(entryPoint.Protocol)).Msg("bound tcp traffic entrypoint")
358 runtime := newEntryPointRuntime(ctx, m, entryPoint, listener, trusted, rawTrusted)
359 return runtime, nil
360}
361
362func (m *Manager) prepareUDPListeners(ctx context.Context, graph *domain.TrafficGraph) (map[string]*udpEntryPointRuntime, []*udpEntryPointRuntime, []udpRuntimeUpdate, error) {
363 current := make(map[string]*udpEntryPointRuntime, len(m.udpListeners))
364 maps.Copy(current, m.udpListeners)
365
366 next := make(map[string]*udpEntryPointRuntime, len(current))
367 created := []*udpEntryPointRuntime{}
368 updates := []udpRuntimeUpdate{}
369 for _, entryPoint := range graph.EntryPoints {
370 if entryPoint.Protocol != domain.EntryPointProtocolUDP {
371 continue
372 }
373 if runtime := current[entryPoint.Name]; runtime != nil && runtime.matches(entryPoint) {
374 next[entryPoint.Name] = runtime
375 delete(current, entryPoint.Name)
376 continue
377 }
378 if runtime := conflictingUDPRuntime(current, entryPoint); runtime != nil {
379 trusted, err := parseTrustedCIDRs(entryPoint.TrustedCIDRs)
380 if err != nil {
381 stopUDPRuntimes(ctx, created, effectiveUDPOptions(graph.Options.UDP).DrainTimeout)
382 return nil, nil, nil, fmt.Errorf("parse trusted cidrs for udp entrypoint %q: %w", entryPoint.Name, err)
383 }
384 trafficDebug(ctx).Str("entrypoint", entryPoint.Name).Str("address", entryPoint.Address).Msg("reusing udp traffic listener for same-address entrypoint update")
385 updates = append(updates, udpRuntimeUpdate{runtime: runtime, entryPoint: entryPoint, trusted: trusted})
386 next[entryPoint.Name] = runtime
387 delete(current, runtime.entryPointSnapshot().Name)
388 continue
389 }
390 runtime, err := m.bindUDPEntryPoint(ctx, entryPoint)
391 if err != nil {
392 stopUDPRuntimes(ctx, created, effectiveUDPOptions(graph.Options.UDP).DrainTimeout)
393 return nil, nil, nil, err
394 }
395 next[entryPoint.Name] = runtime
396 created = append(created, runtime)
397 }
398 return next, created, updates, nil
399}
400
401func conflictingUDPRuntime(current map[string]*udpEntryPointRuntime, entryPoint domain.EntryPoint) *udpEntryPointRuntime {
402 for _, runtime := range current {
403 if runtime.sameAddress(entryPoint) {
404 return runtime
405 }
406 }
407 return nil
408}
409
410func (m *Manager) bindUDPEntryPoint(ctx context.Context, entryPoint domain.EntryPoint) (*udpEntryPointRuntime, error) {
411 packetConn, err := (&net.ListenConfig{}).ListenPacket(ctx, "udp", entryPoint.Address)
412 if err != nil {
413 if errors.Is(err, syscall.EADDRINUSE) {
414 return nil, fmt.Errorf("bind udp entrypoint %q on %s: inspect host-level UDP listeners, including other users or Gordon instances: %w", entryPoint.Name, entryPoint.Address, err)
415 }
416 return nil, fmt.Errorf("bind udp entrypoint %q on %s: %w", entryPoint.Name, entryPoint.Address, err)
417 }
418 trusted, err := parseTrustedCIDRs(entryPoint.TrustedCIDRs)
419 if err != nil {
420 _ = packetConn.Close()
421 return nil, fmt.Errorf("parse trusted cidrs for udp entrypoint %q: %w", entryPoint.Name, err)
422 }
423 trafficInfo(ctx).Str("entrypoint", entryPoint.Name).Str("address", entryPoint.Address).Str("protocol", string(entryPoint.Protocol)).Msg("bound udp traffic entrypoint")
424 return newUDPEntryPointRuntime(ctx, m, entryPoint, packetConn, trusted), nil
425}
426
427func parseTrustedCIDRs(values []string) ([]*net.IPNet, error) {
428 if len(values) == 0 {
429 return nil, nil
430 }
431 trusted := make([]*net.IPNet, 0, len(values))
432 for _, value := range values {
433 _, network, err := net.ParseCIDR(strings.TrimSpace(value))
434 if err != nil {
435 return nil, err
436 }
437 trusted = append(trusted, network)
438 }
439 return trusted, nil
440}
441
442func trustedCIDRsEqual(left []string, right []string) bool {
443 left = normalizedCIDRs(left)
444 right = normalizedCIDRs(right)
445 if len(left) != len(right) {
446 return false
447 }
448 for i := range left {
449 if left[i] != right[i] {
450 return false
451 }
452 }
453 return true
454}
455
456func normalizedCIDRs(values []string) []string {
457 out := make([]string, 0, len(values))
458 for _, value := range values {
459 out = append(out, strings.TrimSpace(value))
460 }
461 sort.Strings(out)
462 return out
463}
464
465func trustedRemoteAddr(trusted []*net.IPNet, addr net.Addr) bool {
466 if len(trusted) == 0 {
467 return true
468 }
469 host, _, err := net.SplitHostPort(addr.String())
470 if err != nil {
471 return false
472 }
473 ip := net.ParseIP(host)
474 if ip == nil {
475 return false
476 }
477 for _, network := range trusted {
478 if network.Contains(ip) {
479 return true
480 }
481 }
482 return false
483}
484
485func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]*entryPointRuntime, udpListeners map[string]*udpEntryPointRuntime) []domain.EntryPointStatus {
486 statuses := make([]domain.EntryPointStatus, 0, len(entries))
487 for _, entry := range entries {
488 status := domain.EntryPointStatus{Name: entry.Name, Address: entry.Address, Protocol: entry.Protocol}
489 if runtime := listeners[entry.Name]; runtime != nil {
490 counters := runtime.counters.snapshot()
491 status.Active = !runtime.isClosed()
492 status.ActiveTCPConnections = counters.ActiveTCPConnections
493 status.TotalAccepted = counters.TotalAccepted
494 status.TotalRefused = counters.TotalRefused
495 status.TotalErrors = counters.TotalErrors
496 status.BytesIn = counters.BytesIn
497 status.BytesOut = counters.BytesOut
498 status.SmartTCP = counters.SmartTCP
499 }
500 if runtime := udpListeners[entry.Name]; runtime != nil {
501 counters := runtime.counters.snapshot()
502 status.Active = !runtime.isClosed()
503 status.ActiveUDPSessions = counters.ActiveUDPSessions
504 status.TotalAccepted = counters.TotalAccepted
505 status.TotalRefused = counters.TotalRefused
506 status.TotalErrors = counters.TotalErrors
507 status.BytesIn = counters.BytesIn
508 status.BytesOut = counters.BytesOut
509 }
510 statuses = append(statuses, status)
511 }
512 sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
513 return statuses
514}
515
516func routerStatuses(routers []domain.TrafficRouter, listeners map[string]*entryPointRuntime, udpListeners map[string]*udpEntryPointRuntime) []domain.TrafficRouterStatus {
517 statuses := make([]domain.TrafficRouterStatus, 0, len(routers))
518 for _, router := range routers {
519 tcpActive := false
520 if runtime := listeners[router.EntryPoint]; runtime != nil {
521 tcpActive = !runtime.isClosed()
522 }
523 udpActive := false
524 if runtime := udpListeners[router.EntryPoint]; runtime != nil {
525 udpActive = !runtime.isClosed()
526 }
527 statuses = append(statuses, domain.TrafficRouterStatus{
528 Name: router.Name, EntryPoint: router.EntryPoint, Protocol: router.Protocol,
529 Rule: router.Rule, Service: router.Service, Active: tcpActive || udpActive,
530 })
531 }
532 sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
533 return statuses
534}
535
536func serviceStatuses(services []domain.TrafficService) []domain.TrafficServiceStatus {
537 statuses := make([]domain.TrafficServiceStatus, 0, len(services))
538 for _, service := range services {
539 status := domain.TrafficServiceStatus{Name: service.Name, Active: true, Backends: make([]domain.TrafficBackendStatus, 0, len(service.Backends))}
540 for _, backend := range service.Backends {
541 status.Backends = append(status.Backends, domain.TrafficBackendStatus{Name: backend.Name, Host: backend.Host, Port: backend.Port, Protocol: backend.Protocol, Active: true})
542 }
543 statuses = append(statuses, status)
544 }
545 sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
546 return statuses
547}
548
549func aggregateCounters(entries []domain.EntryPointStatus) domain.TrafficCounters {
550 var counters domain.TrafficCounters
551 for _, entry := range entries {
552 counters.ActiveTCPConnections += entry.ActiveTCPConnections
553 counters.ActiveUDPSessions += entry.ActiveUDPSessions
554 counters.TotalAccepted += entry.TotalAccepted
555 counters.TotalRefused += entry.TotalRefused
556 counters.TotalErrors += entry.TotalErrors
557 counters.BytesIn += entry.BytesIn
558 counters.BytesOut += entry.BytesOut
559 counters.SmartTCP.HTTPAccepted += entry.SmartTCP.HTTPAccepted
560 counters.SmartTCP.H2CAccepted += entry.SmartTCP.H2CAccepted
561 counters.SmartTCP.HTTPSFallbackAccepted += entry.SmartTCP.HTTPSFallbackAccepted
562 counters.SmartTCP.TLSPassthroughAccepted += entry.SmartTCP.TLSPassthroughAccepted
563 counters.SmartTCP.RawFallbackAccepted += entry.SmartTCP.RawFallbackAccepted
564 counters.SmartTCP.EntrypointCIDRRefused += entry.SmartTCP.EntrypointCIDRRefused
565 counters.SmartTCP.RawFallbackCIDRRefused += entry.SmartTCP.RawFallbackCIDRRefused
566 counters.SmartTCP.PROXYRefused += entry.SmartTCP.PROXYRefused
567 counters.SmartTCP.UnknownNoFallbackRefused += entry.SmartTCP.UnknownNoFallbackRefused
568 counters.SmartTCP.MalformedRejected += entry.SmartTCP.MalformedRejected
569 counters.SmartTCP.SniffTimeout += entry.SmartTCP.SniffTimeout
570 counters.SmartTCP.ClientHelloTooLarge += entry.SmartTCP.ClientHelloTooLarge
571 }
572 return counters
573}
574
575func cloneTrafficGraph(graph domain.TrafficGraph) domain.TrafficGraph {
576 clone := graph
577 clone.EntryPoints = append([]domain.EntryPoint{}, graph.EntryPoints...)
578 for i := range clone.EntryPoints {
579 clone.EntryPoints[i].TrustedCIDRs = append([]string(nil), graph.EntryPoints[i].TrustedCIDRs...)
580 clone.EntryPoints[i].RawFallbackTrustedCIDRs = append([]string(nil), graph.EntryPoints[i].RawFallbackTrustedCIDRs...)
581 }
582 clone.Routers = append([]domain.TrafficRouter{}, graph.Routers...)
583 clone.Services = append([]domain.TrafficService{}, graph.Services...)
584 for i := range clone.Services {
585 clone.Services[i].Backends = append([]domain.TrafficBackend{}, graph.Services[i].Backends...)
586 }
587 return clone
588}
589
590func snapshotTCPOptions(graph *domain.TrafficGraph) domain.TCPOptions {
591 if graph == nil {
592 return domain.TCPOptions{}
593 }
594 return graph.Options.TCP
595}
596
597func snapshotUDPOptions(graph *domain.TrafficGraph) domain.UDPOptions {
598 if graph == nil {
599 return domain.UDPOptions{}
600 }
601 return graph.Options.UDP
602}
603
604type trafficCounters struct {
605 activeTCPConnections atomic.Int64
606 activeUDPSessions atomic.Int64
607 totalAccepted atomic.Int64
608 totalRefused atomic.Int64
609 totalErrors atomic.Int64
610 bytesIn atomic.Int64
611 bytesOut atomic.Int64
612 smartTCP smartTCPCounterSet
613}
614
615type smartTCPCounterSet struct {
616 httpAccepted atomic.Int64
617 h2cAccepted atomic.Int64
618 httpsFallbackAccepted atomic.Int64
619 tlsPassthroughAccepted atomic.Int64
620 rawFallbackAccepted atomic.Int64
621 entrypointCIDRRefused atomic.Int64
622 rawFallbackCIDRRefused atomic.Int64
623 proxyRefused atomic.Int64
624 unknownNoFallbackRefused atomic.Int64
625 malformedRejected atomic.Int64
626 sniffTimeout atomic.Int64
627 clientHelloTooLarge atomic.Int64
628}
629
630func (c *trafficCounters) snapshot() domain.TrafficCounters {
631 return domain.TrafficCounters{
632 ActiveTCPConnections: c.activeTCPConnections.Load(),
633 ActiveUDPSessions: c.activeUDPSessions.Load(),
634 TotalAccepted: c.totalAccepted.Load(),
635 TotalRefused: c.totalRefused.Load(),
636 TotalErrors: c.totalErrors.Load(),
637 BytesIn: c.bytesIn.Load(),
638 BytesOut: c.bytesOut.Load(),
639 SmartTCP: domain.SmartTCPCounters{
640 HTTPAccepted: c.smartTCP.httpAccepted.Load(),
641 H2CAccepted: c.smartTCP.h2cAccepted.Load(),
642 HTTPSFallbackAccepted: c.smartTCP.httpsFallbackAccepted.Load(),
643 TLSPassthroughAccepted: c.smartTCP.tlsPassthroughAccepted.Load(),
644 RawFallbackAccepted: c.smartTCP.rawFallbackAccepted.Load(),
645 EntrypointCIDRRefused: c.smartTCP.entrypointCIDRRefused.Load(),
646 RawFallbackCIDRRefused: c.smartTCP.rawFallbackCIDRRefused.Load(),
647 PROXYRefused: c.smartTCP.proxyRefused.Load(),
648 UnknownNoFallbackRefused: c.smartTCP.unknownNoFallbackRefused.Load(),
649 MalformedRejected: c.smartTCP.malformedRejected.Load(),
650 SniffTimeout: c.smartTCP.sniffTimeout.Load(),
651 ClientHelloTooLarge: c.smartTCP.clientHelloTooLarge.Load(),
652 },
653 }
654}
655
656func defaultTCPOptions() domain.TCPOptions {
657 return domain.TCPOptions{
658 DialTimeout: 10 * time.Second,
659 IdleTimeout: 5 * time.Minute,
660 DrainTimeout: 30 * time.Second,
661 MaxConnections: 1024,
662 }
663}
664
665func effectiveTCPOptions(options domain.TCPOptions) domain.TCPOptions {
666 defaults := defaultTCPOptions()
667 if options.DialTimeout == 0 {
668 options.DialTimeout = defaults.DialTimeout
669 }
670 if options.IdleTimeout == 0 {
671 options.IdleTimeout = defaults.IdleTimeout
672 }
673 if options.DrainTimeout == 0 {
674 options.DrainTimeout = defaults.DrainTimeout
675 }
676 if options.MaxConnections <= 0 {
677 options.MaxConnections = defaults.MaxConnections
678 }
679 return options
680}
681
682func defaultUDPOptions() domain.UDPOptions {
683 return domain.UDPOptions{IdleTimeout: 30 * time.Second, DrainTimeout: 30 * time.Second, MaxSessions: 4096}
684}
685
686func effectiveUDPOptions(options domain.UDPOptions) domain.UDPOptions {
687 defaults := defaultUDPOptions()
688 if options.IdleTimeout == 0 {
689 options.IdleTimeout = defaults.IdleTimeout
690 }
691 if options.DrainTimeout == 0 {
692 options.DrainTimeout = defaults.DrainTimeout
693 }
694 if options.MaxSessions <= 0 {
695 options.MaxSessions = defaults.MaxSessions
696 }
697 return options
698}