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