feat(traffic): add udp passthrough entrypoints

commit: 02269b2ef604272f6bc7c6f82f3ec0ac822b1a9d
author: brice · 3mo ago · 2026-06-29 21:54 UTC
parent: 771cd59
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +692 -12

M internal/adapters/in/traffic/manager.go +129 -12


@@ -25,9 +25,10 @@ const (
25	25		type Manager struct {
26	26		log zerowrap.Logger
27	27
28		-	mu sync.Mutex
29		-	snapshot atomic.Pointer[domain.TrafficGraph]
30		-	listeners map[string]*entryPointRuntime
	28	+	mu sync.Mutex
	29	+	snapshot atomic.Pointer[domain.TrafficGraph]
	30	+	listeners map[string]*entryPointRuntime
	31	+	udpListeners map[string]*udpEntryPointRuntime
31	32
32	33		lastReloadStatus string
33	34		lastReloadError string

@@ -39,6 +40,7 @@ func NewManager(log zerowrap.Logger) *Manager {
39	40		manager := &Manager{
40	41		log: log,
41	42		listeners: map[string]*entryPointRuntime{},
	43	+	udpListeners: map[string]*udpEntryPointRuntime{},
42	44		lastReloadStatus: reloadStatusOK,
43	45		}
44	46		manager.tlsFallbacks.Store(tlsFallbacks{})

@@ -64,24 +66,48 @@ func (m Manager) Apply(ctx context.Context, graph domain.TrafficGraph) error {
64	66		m.lastReloadError = err.Error()
65	67		return err
66	68		}
	69	+	newUDPListeners, createdUDPListeners, err := m.prepareUDPListeners(ctx, &nextGraph)
	70	+	if err != nil {
	71	+	for _, runtime := range createdListeners {
	72	+	runtime.stop(ctx, effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout)
	73	+	}
	74	+	m.lastReloadStatus = reloadStatusError
	75	+	m.lastReloadError = err.Error()
	76	+	return err
	77	+	}
67	78
68	79		oldListeners := m.listeners
	80	+	oldUDPListeners := m.udpListeners
69	81		m.listeners = newListeners
	82	+	m.udpListeners = newUDPListeners
70	83		m.snapshot.Store(&nextGraph)
71	84		for _, runtime := range createdListeners {
72	85		runtime.start()
73	86		}
	87	+	for _, runtime := range createdUDPListeners {
	88	+	runtime.start()
	89	+	}
74	90		m.lastReloadStatus = reloadStatusOK
75	91		m.lastReloadError = ""
76	92
77		-	drainTimeout := effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout
	93	+	tcpDrainTimeout := effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout
78	94		for name, runtime := range oldListeners {
79	95		if newListeners[name] == runtime {
80	96		continue
81	97		}
82	98		if runtime.shouldStopWith(newListeners[name]) {
83		-	runtime.stop(ctx, drainTimeout)
	99	+	runtime.stop(ctx, tcpDrainTimeout)
	100	+	}
	101	+	}
	102	+	udpDrainTimeout := effectiveUDPOptions(snapshotUDPOptions(&nextGraph)).DrainTimeout
	103	+	for name, runtime := range oldUDPListeners {
	104	+	if newUDPListeners[name] == runtime {
	105	+	if _, ok := runtime.resolveUDPBackend(); !ok {
	106	+	runtime.drainSessionsAfter(udpDrainTimeout)
	107	+	}
	108	+	continue
84	109		}
	110	+	runtime.stop(ctx, udpDrainTimeout)
85	111		}
86	112		return nil
87	113		}

@@ -92,16 +118,23 @@ func (m *Manager) Shutdown(ctx context.Context) error {
92	118		defer m.mu.Unlock()
93	119
94	120		listeners := m.listeners
	121	+	udpListeners := m.udpListeners
95	122		m.listeners = map[string]*entryPointRuntime{}
	123	+	m.udpListeners = map[string]*udpEntryPointRuntime{}
96	124		graph := m.snapshot.Load()
97	125		m.snapshot.Store(nil)
98	126
99		-	drainTimeout := defaultTCPOptions().DrainTimeout
	127	+	tcpDrainTimeout := defaultTCPOptions().DrainTimeout
	128	+	udpDrainTimeout := defaultUDPOptions().DrainTimeout
100	129		if graph != nil {
101		-	drainTimeout = effectiveTCPOptions(graph.Options.TCP).DrainTimeout
	130	+	tcpDrainTimeout = effectiveTCPOptions(graph.Options.TCP).DrainTimeout
	131	+	udpDrainTimeout = effectiveUDPOptions(graph.Options.UDP).DrainTimeout
102	132		}
103	133		for _, runtime := range listeners {
104		-	runtime.stop(ctx, drainTimeout)
	134	+	runtime.stop(ctx, tcpDrainTimeout)
	135	+	}
	136	+	for _, runtime := range udpListeners {
	137	+	runtime.stop(ctx, udpDrainTimeout)
105	138		}
106	139		return nil
107	140		}

@@ -113,6 +146,10 @@ func (m *Manager) Status() domain.TrafficStatus {
113	146		for name, runtime := range m.listeners {
114	147		listeners[name] = runtime
115	148		}
	149	+	udpListeners := make(map[string]*udpEntryPointRuntime, len(m.udpListeners))
	150	+	for name, runtime := range m.udpListeners {
	151	+	udpListeners[name] = runtime
	152	+	}
116	153		status := domain.TrafficStatus{LastReloadStatus: m.lastReloadStatus, LastReloadError: m.lastReloadError}
117	154		graph := m.snapshot.Load()
118	155		m.mu.Unlock()

@@ -121,8 +158,8 @@ func (m *Manager) Status() domain.TrafficStatus {
121	158		return status
122	159		}
123	160
124		-	status.EntryPoints = entryPointStatuses(graph.EntryPoints, listeners)
125		-	status.Routers = routerStatuses(graph.Routers, listeners)
	161	+	status.EntryPoints = entryPointStatuses(graph.EntryPoints, listeners, udpListeners)
	162	+	status.Routers = routerStatuses(graph.Routers, listeners, udpListeners)
126	163		status.Services = serviceStatuses(graph.Services)
127	164		status.Counters = aggregateCounters(status.EntryPoints)
128	165		return status

@@ -178,7 +215,44 @@ func (m *Manager) bindTCPEntryPoint(ctx context.Context, entryPoint domain.Entry
178	215		return runtime, nil
179	216		}
180	217
181		-	func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]*entryPointRuntime) []domain.EntryPointStatus {
	218	+	func (m Manager) prepareUDPListeners(ctx context.Context, graph domain.TrafficGraph) (map[string]udpEntryPointRuntime, []udpEntryPointRuntime, error) {
	219	+	current := make(map[string]*udpEntryPointRuntime, len(m.udpListeners))
	220	+	for name, runtime := range m.udpListeners {
	221	+	current[name] = runtime
	222	+	}
	223	+
	224	+	next := make(map[string]*udpEntryPointRuntime, len(current))
	225	+	created := []*udpEntryPointRuntime{}
	226	+	for _, entryPoint := range graph.EntryPoints {
	227	+	if entryPoint.Protocol != domain.EntryPointProtocolUDP {
	228	+	continue
	229	+	}
	230	+	if runtime := current[entryPoint.Name]; runtime != nil && runtime.matches(entryPoint) {
	231	+	next[entryPoint.Name] = runtime
	232	+	continue
	233	+	}
	234	+	runtime, err := m.bindUDPEntryPoint(ctx, entryPoint)
	235	+	if err != nil {
	236	+	for _, createdRuntime := range created {
	237	+	createdRuntime.stop(ctx, effectiveUDPOptions(graph.Options.UDP).DrainTimeout)
	238	+	}
	239	+	return nil, nil, err
	240	+	}
	241	+	next[entryPoint.Name] = runtime
	242	+	created = append(created, runtime)
	243	+	}
	244	+	return next, created, nil
	245	+	}
	246	+
	247	+	func (m Manager) bindUDPEntryPoint(ctx context.Context, entryPoint domain.EntryPoint) (udpEntryPointRuntime, error) {
	248	+	packetConn, err := (&net.ListenConfig{}).ListenPacket(ctx, "udp", entryPoint.Address)
	249	+	if err != nil {
	250	+	return nil, fmt.Errorf("bind udp entrypoint %q on %s: %w", entryPoint.Name, entryPoint.Address, err)
	251	+	}
	252	+	return newUDPEntryPointRuntime(m, entryPoint, packetConn), nil
	253	+	}
	254	+
	255	+	func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]entryPointRuntime, udpListeners map[string]udpEntryPointRuntime) []domain.EntryPointStatus {
182	256		statuses := make([]domain.EntryPointStatus, 0, len(entries))
183	257		for _, entry := range entries {
184	258		status := domain.EntryPointStatus{Name: entry.Name, Address: entry.Address, Protocol: entry.Protocol}

@@ -192,13 +266,23 @@ func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]*entry
192	266		status.BytesIn = counters.BytesIn
193	267		status.BytesOut = counters.BytesOut
194	268		}
	269	+	if runtime := udpListeners[entry.Name]; runtime != nil {
	270	+	counters := runtime.counters.snapshot()
	271	+	status.Active = !runtime.isClosed()
	272	+	status.ActiveUDPSessions = counters.ActiveUDPSessions
	273	+	status.TotalAccepted = counters.TotalAccepted
	274	+	status.TotalRefused = counters.TotalRefused
	275	+	status.TotalErrors = counters.TotalErrors
	276	+	status.BytesIn = counters.BytesIn
	277	+	status.BytesOut = counters.BytesOut
	278	+	}
195	279		statuses = append(statuses, status)
196	280		}
197	281		sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
198	282		return statuses
199	283		}
200	284
201		-	func routerStatuses(routers []domain.TrafficRouter, listeners map[string]*entryPointRuntime) []domain.TrafficRouterStatus {
	285	+	func routerStatuses(routers []domain.TrafficRouter, listeners map[string]entryPointRuntime, udpListeners map[string]udpEntryPointRuntime) []domain.TrafficRouterStatus {
202	286		statuses := make([]domain.TrafficRouterStatus, 0, len(routers))
203	287		for _, router := range routers {
204	288		status := domain.TrafficRouterStatus{Name: router.Name, EntryPoint: router.EntryPoint, Protocol: router.Protocol, Active: true}

@@ -211,6 +295,15 @@ func routerStatuses(routers []domain.TrafficRouter, listeners map[string]*entryP
211	295		status.BytesIn = counters.BytesIn
212	296		status.BytesOut = counters.BytesOut
213	297		}
	298	+	if runtime := udpListeners[router.EntryPoint]; runtime != nil {
	299	+	counters := runtime.counters.snapshot()
	300	+	status.ActiveUDPSessions = counters.ActiveUDPSessions
	301	+	status.TotalAccepted = counters.TotalAccepted
	302	+	status.TotalRefused = counters.TotalRefused
	303	+	status.TotalErrors = counters.TotalErrors
	304	+	status.BytesIn = counters.BytesIn
	305	+	status.BytesOut = counters.BytesOut
	306	+	}
214	307		statuses = append(statuses, status)
215	308		}
216	309		sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })

@@ -262,8 +355,16 @@ func snapshotTCPOptions(graph *domain.TrafficGraph) domain.TCPOptions {
262	355		return graph.Options.TCP
263	356		}
264	357
	358	+	func snapshotUDPOptions(graph *domain.TrafficGraph) domain.UDPOptions {
	359	+	if graph == nil {
	360	+	return domain.UDPOptions{}
	361	+	}
	362	+	return graph.Options.UDP
	363	+	}
	364	+
265	365		type trafficCounters struct {
266	366		activeTCPConnections atomic.Int64
	367	+	activeUDPSessions atomic.Int64
267	368		totalAccepted atomic.Int64
268	369		totalRefused atomic.Int64
269	370		totalErrors atomic.Int64

@@ -274,6 +375,7 @@ type trafficCounters struct {
274	375		func (c *trafficCounters) snapshot() domain.TrafficCounters {
275	376		return domain.TrafficCounters{
276	377		ActiveTCPConnections: c.activeTCPConnections.Load(),
	378	+	ActiveUDPSessions: c.activeUDPSessions.Load(),
277	379		TotalAccepted: c.totalAccepted.Load(),
278	380		TotalRefused: c.totalRefused.Load(),
279	381		TotalErrors: c.totalErrors.Load(),

@@ -303,3 +405,18 @@ func effectiveTCPOptions(options domain.TCPOptions) domain.TCPOptions {
303	405		}
304	406		return options
305	407		}
	408	+
	409	+	func defaultUDPOptions() domain.UDPOptions {
	410	+	return domain.UDPOptions{IdleTimeout: 30 * time.Second, DrainTimeout: 30 * time.Second}
	411	+	}
	412	+
	413	+	func effectiveUDPOptions(options domain.UDPOptions) domain.UDPOptions {
	414	+	defaults := defaultUDPOptions()
	415	+	if options.IdleTimeout == 0 {
	416	+	options.IdleTimeout = defaults.IdleTimeout
	417	+	}
	418	+	if options.DrainTimeout == 0 {
	419	+	options.DrainTimeout = defaults.DrainTimeout
	420	+	}
	421	+	return options
	422	+	}


@@ -25,9 +25,10 @@ const (
25		type Manager struct {	25		type Manager struct {
26		log zerowrap.Logger	26		log zerowrap.Logger
27			27
28	-	mu sync.Mutex	28	+	mu sync.Mutex
29	-	snapshot atomic.Pointer[domain.TrafficGraph]	29	+	snapshot atomic.Pointer[domain.TrafficGraph]
30	-	listeners map[string]*entryPointRuntime	30	+	listeners map[string]*entryPointRuntime
			31	+	udpListeners map[string]*udpEntryPointRuntime
31			32
32		lastReloadStatus string	33		lastReloadStatus string
33		lastReloadError string	34		lastReloadError string

@@ -39,6 +40,7 @@ func NewManager(log zerowrap.Logger) *Manager {
39		manager := &Manager{	40		manager := &Manager{
40		log: log,	41		log: log,
41		listeners: map[string]*entryPointRuntime{},	42		listeners: map[string]*entryPointRuntime{},
			43	+	udpListeners: map[string]*udpEntryPointRuntime{},
42		lastReloadStatus: reloadStatusOK,	44		lastReloadStatus: reloadStatusOK,
43		}	45		}
44		manager.tlsFallbacks.Store(tlsFallbacks{})	46		manager.tlsFallbacks.Store(tlsFallbacks{})

@@ -64,24 +66,48 @@ func (m Manager) Apply(ctx context.Context, graph domain.TrafficGraph) error {
64		m.lastReloadError = err.Error()	66		m.lastReloadError = err.Error()
65		return err	67		return err
66		}	68		}
			69	+	newUDPListeners, createdUDPListeners, err := m.prepareUDPListeners(ctx, &nextGraph)
			70	+	if err != nil {
			71	+	for _, runtime := range createdListeners {
			72	+	runtime.stop(ctx, effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout)
			73	+	}
			74	+	m.lastReloadStatus = reloadStatusError
			75	+	m.lastReloadError = err.Error()
			76	+	return err
			77	+	}
67			78
68		oldListeners := m.listeners	79		oldListeners := m.listeners
			80	+	oldUDPListeners := m.udpListeners
69		m.listeners = newListeners	81		m.listeners = newListeners
			82	+	m.udpListeners = newUDPListeners
70		m.snapshot.Store(&nextGraph)	83		m.snapshot.Store(&nextGraph)
71		for _, runtime := range createdListeners {	84		for _, runtime := range createdListeners {
72		runtime.start()	85		runtime.start()
73		}	86		}
			87	+	for _, runtime := range createdUDPListeners {
			88	+	runtime.start()
			89	+	}
74		m.lastReloadStatus = reloadStatusOK	90		m.lastReloadStatus = reloadStatusOK
75		m.lastReloadError = ""	91		m.lastReloadError = ""
76			92
77	-	drainTimeout := effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout	93	+	tcpDrainTimeout := effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout
78		for name, runtime := range oldListeners {	94		for name, runtime := range oldListeners {
79		if newListeners[name] == runtime {	95		if newListeners[name] == runtime {
80		continue	96		continue
81		}	97		}
82		if runtime.shouldStopWith(newListeners[name]) {	98		if runtime.shouldStopWith(newListeners[name]) {
83	-	runtime.stop(ctx, drainTimeout)	99	+	runtime.stop(ctx, tcpDrainTimeout)
			100	+	}
			101	+	}
			102	+	udpDrainTimeout := effectiveUDPOptions(snapshotUDPOptions(&nextGraph)).DrainTimeout
			103	+	for name, runtime := range oldUDPListeners {
			104	+	if newUDPListeners[name] == runtime {
			105	+	if _, ok := runtime.resolveUDPBackend(); !ok {
			106	+	runtime.drainSessionsAfter(udpDrainTimeout)
			107	+	}
			108	+	continue
84		}	109		}
			110	+	runtime.stop(ctx, udpDrainTimeout)
85		}	111		}
86		return nil	112		return nil
87		}	113		}

@@ -92,16 +118,23 @@ func (m *Manager) Shutdown(ctx context.Context) error {
92		defer m.mu.Unlock()	118		defer m.mu.Unlock()
93			119
94		listeners := m.listeners	120		listeners := m.listeners
			121	+	udpListeners := m.udpListeners
95		m.listeners = map[string]*entryPointRuntime{}	122		m.listeners = map[string]*entryPointRuntime{}
			123	+	m.udpListeners = map[string]*udpEntryPointRuntime{}
96		graph := m.snapshot.Load()	124		graph := m.snapshot.Load()
97		m.snapshot.Store(nil)	125		m.snapshot.Store(nil)
98			126
99	-	drainTimeout := defaultTCPOptions().DrainTimeout	127	+	tcpDrainTimeout := defaultTCPOptions().DrainTimeout
			128	+	udpDrainTimeout := defaultUDPOptions().DrainTimeout
100		if graph != nil {	129		if graph != nil {
101	-	drainTimeout = effectiveTCPOptions(graph.Options.TCP).DrainTimeout	130	+	tcpDrainTimeout = effectiveTCPOptions(graph.Options.TCP).DrainTimeout
			131	+	udpDrainTimeout = effectiveUDPOptions(graph.Options.UDP).DrainTimeout
102		}	132		}
103		for _, runtime := range listeners {	133		for _, runtime := range listeners {
104	-	runtime.stop(ctx, drainTimeout)	134	+	runtime.stop(ctx, tcpDrainTimeout)
			135	+	}
			136	+	for _, runtime := range udpListeners {
			137	+	runtime.stop(ctx, udpDrainTimeout)
105		}	138		}
106		return nil	139		return nil
107		}	140		}

@@ -113,6 +146,10 @@ func (m *Manager) Status() domain.TrafficStatus {
113		for name, runtime := range m.listeners {	146		for name, runtime := range m.listeners {
114		listeners[name] = runtime	147		listeners[name] = runtime
115		}	148		}
			149	+	udpListeners := make(map[string]*udpEntryPointRuntime, len(m.udpListeners))
			150	+	for name, runtime := range m.udpListeners {
			151	+	udpListeners[name] = runtime
			152	+	}
116		status := domain.TrafficStatus{LastReloadStatus: m.lastReloadStatus, LastReloadError: m.lastReloadError}	153		status := domain.TrafficStatus{LastReloadStatus: m.lastReloadStatus, LastReloadError: m.lastReloadError}
117		graph := m.snapshot.Load()	154		graph := m.snapshot.Load()
118		m.mu.Unlock()	155		m.mu.Unlock()

@@ -121,8 +158,8 @@ func (m *Manager) Status() domain.TrafficStatus {
121		return status	158		return status
122		}	159		}
123			160
124	-	status.EntryPoints = entryPointStatuses(graph.EntryPoints, listeners)	161	+	status.EntryPoints = entryPointStatuses(graph.EntryPoints, listeners, udpListeners)
125	-	status.Routers = routerStatuses(graph.Routers, listeners)	162	+	status.Routers = routerStatuses(graph.Routers, listeners, udpListeners)
126		status.Services = serviceStatuses(graph.Services)	163		status.Services = serviceStatuses(graph.Services)
127		status.Counters = aggregateCounters(status.EntryPoints)	164		status.Counters = aggregateCounters(status.EntryPoints)
128		return status	165		return status

@@ -178,7 +215,44 @@ func (m *Manager) bindTCPEntryPoint(ctx context.Context, entryPoint domain.Entry
178		return runtime, nil	215		return runtime, nil
179		}	216		}
180			217
181	-	func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]*entryPointRuntime) []domain.EntryPointStatus {	218	+	func (m Manager) prepareUDPListeners(ctx context.Context, graph domain.TrafficGraph) (map[string]udpEntryPointRuntime, []udpEntryPointRuntime, error) {
			219	+	current := make(map[string]*udpEntryPointRuntime, len(m.udpListeners))
			220	+	for name, runtime := range m.udpListeners {
			221	+	current[name] = runtime
			222	+	}
			223	+
			224	+	next := make(map[string]*udpEntryPointRuntime, len(current))
			225	+	created := []*udpEntryPointRuntime{}
			226	+	for _, entryPoint := range graph.EntryPoints {
			227	+	if entryPoint.Protocol != domain.EntryPointProtocolUDP {
			228	+	continue
			229	+	}
			230	+	if runtime := current[entryPoint.Name]; runtime != nil && runtime.matches(entryPoint) {
			231	+	next[entryPoint.Name] = runtime
			232	+	continue
			233	+	}
			234	+	runtime, err := m.bindUDPEntryPoint(ctx, entryPoint)
			235	+	if err != nil {
			236	+	for _, createdRuntime := range created {
			237	+	createdRuntime.stop(ctx, effectiveUDPOptions(graph.Options.UDP).DrainTimeout)
			238	+	}
			239	+	return nil, nil, err
			240	+	}
			241	+	next[entryPoint.Name] = runtime
			242	+	created = append(created, runtime)
			243	+	}
			244	+	return next, created, nil
			245	+	}
			246	+
			247	+	func (m Manager) bindUDPEntryPoint(ctx context.Context, entryPoint domain.EntryPoint) (udpEntryPointRuntime, error) {
			248	+	packetConn, err := (&net.ListenConfig{}).ListenPacket(ctx, "udp", entryPoint.Address)
			249	+	if err != nil {
			250	+	return nil, fmt.Errorf("bind udp entrypoint %q on %s: %w", entryPoint.Name, entryPoint.Address, err)
			251	+	}
			252	+	return newUDPEntryPointRuntime(m, entryPoint, packetConn), nil
			253	+	}
			254	+
			255	+	func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]entryPointRuntime, udpListeners map[string]udpEntryPointRuntime) []domain.EntryPointStatus {
182		statuses := make([]domain.EntryPointStatus, 0, len(entries))	256		statuses := make([]domain.EntryPointStatus, 0, len(entries))
183		for _, entry := range entries {	257		for _, entry := range entries {
184		status := domain.EntryPointStatus{Name: entry.Name, Address: entry.Address, Protocol: entry.Protocol}	258		status := domain.EntryPointStatus{Name: entry.Name, Address: entry.Address, Protocol: entry.Protocol}

@@ -192,13 +266,23 @@ func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]*entry
192		status.BytesIn = counters.BytesIn	266		status.BytesIn = counters.BytesIn
193		status.BytesOut = counters.BytesOut	267		status.BytesOut = counters.BytesOut
194		}	268		}
			269	+	if runtime := udpListeners[entry.Name]; runtime != nil {
			270	+	counters := runtime.counters.snapshot()
			271	+	status.Active = !runtime.isClosed()
			272	+	status.ActiveUDPSessions = counters.ActiveUDPSessions
			273	+	status.TotalAccepted = counters.TotalAccepted
			274	+	status.TotalRefused = counters.TotalRefused
			275	+	status.TotalErrors = counters.TotalErrors
			276	+	status.BytesIn = counters.BytesIn
			277	+	status.BytesOut = counters.BytesOut
			278	+	}
195		statuses = append(statuses, status)	279		statuses = append(statuses, status)
196		}	280		}
197		sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })	281		sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
198		return statuses	282		return statuses
199		}	283		}
200			284
201	-	func routerStatuses(routers []domain.TrafficRouter, listeners map[string]*entryPointRuntime) []domain.TrafficRouterStatus {	285	+	func routerStatuses(routers []domain.TrafficRouter, listeners map[string]entryPointRuntime, udpListeners map[string]udpEntryPointRuntime) []domain.TrafficRouterStatus {
202		statuses := make([]domain.TrafficRouterStatus, 0, len(routers))	286		statuses := make([]domain.TrafficRouterStatus, 0, len(routers))
203		for _, router := range routers {	287		for _, router := range routers {
204		status := domain.TrafficRouterStatus{Name: router.Name, EntryPoint: router.EntryPoint, Protocol: router.Protocol, Active: true}	288		status := domain.TrafficRouterStatus{Name: router.Name, EntryPoint: router.EntryPoint, Protocol: router.Protocol, Active: true}

@@ -211,6 +295,15 @@ func routerStatuses(routers []domain.TrafficRouter, listeners map[string]*entryP
211		status.BytesIn = counters.BytesIn	295		status.BytesIn = counters.BytesIn
212		status.BytesOut = counters.BytesOut	296		status.BytesOut = counters.BytesOut
213		}	297		}
			298	+	if runtime := udpListeners[router.EntryPoint]; runtime != nil {
			299	+	counters := runtime.counters.snapshot()
			300	+	status.ActiveUDPSessions = counters.ActiveUDPSessions
			301	+	status.TotalAccepted = counters.TotalAccepted
			302	+	status.TotalRefused = counters.TotalRefused
			303	+	status.TotalErrors = counters.TotalErrors
			304	+	status.BytesIn = counters.BytesIn
			305	+	status.BytesOut = counters.BytesOut
			306	+	}
214		statuses = append(statuses, status)	307		statuses = append(statuses, status)
215		}	308		}
216		sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })	309		sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })

@@ -262,8 +355,16 @@ func snapshotTCPOptions(graph *domain.TrafficGraph) domain.TCPOptions {
262		return graph.Options.TCP	355		return graph.Options.TCP
263		}	356		}
264			357
			358	+	func snapshotUDPOptions(graph *domain.TrafficGraph) domain.UDPOptions {
			359	+	if graph == nil {
			360	+	return domain.UDPOptions{}
			361	+	}
			362	+	return graph.Options.UDP
			363	+	}
			364	+
265		type trafficCounters struct {	365		type trafficCounters struct {
266		activeTCPConnections atomic.Int64	366		activeTCPConnections atomic.Int64
			367	+	activeUDPSessions atomic.Int64
267		totalAccepted atomic.Int64	368		totalAccepted atomic.Int64
268		totalRefused atomic.Int64	369		totalRefused atomic.Int64
269		totalErrors atomic.Int64	370		totalErrors atomic.Int64

@@ -274,6 +375,7 @@ type trafficCounters struct {
274		func (c *trafficCounters) snapshot() domain.TrafficCounters {	375		func (c *trafficCounters) snapshot() domain.TrafficCounters {
275		return domain.TrafficCounters{	376		return domain.TrafficCounters{
276		ActiveTCPConnections: c.activeTCPConnections.Load(),	377		ActiveTCPConnections: c.activeTCPConnections.Load(),
			378	+	ActiveUDPSessions: c.activeUDPSessions.Load(),
277		TotalAccepted: c.totalAccepted.Load(),	379		TotalAccepted: c.totalAccepted.Load(),
278		TotalRefused: c.totalRefused.Load(),	380		TotalRefused: c.totalRefused.Load(),
279		TotalErrors: c.totalErrors.Load(),	381		TotalErrors: c.totalErrors.Load(),

@@ -303,3 +405,18 @@ func effectiveTCPOptions(options domain.TCPOptions) domain.TCPOptions {
303		}	405		}
304		return options	406		return options
305		}	407		}
			408	+
			409	+	func defaultUDPOptions() domain.UDPOptions {
			410	+	return domain.UDPOptions{IdleTimeout: 30 * time.Second, DrainTimeout: 30 * time.Second}
			411	+	}
			412	+
			413	+	func effectiveUDPOptions(options domain.UDPOptions) domain.UDPOptions {
			414	+	defaults := defaultUDPOptions()
			415	+	if options.IdleTimeout == 0 {
			416	+	options.IdleTimeout = defaults.IdleTimeout
			417	+	}
			418	+	if options.DrainTimeout == 0 {
			419	+	options.DrainTimeout = defaults.DrainTimeout
			420	+	}
			421	+	return options
			422	+	}

A internal/adapters/in/traffic/udp.go +331 -0

@@ -0,0 +1,331 @@
	1	+	package traffic
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"net"
	7	+	"strconv"
	8	+	"sync"
	9	+	"sync/atomic"
	10	+	"time"
	11	+
	12	+	"github.com/bnema/gordon/internal/domain"
	13	+	)
	14	+
	15	+	const udpBufferSize = 64 << 10
	16	+
	17	+	type udpEntryPointRuntime struct {
	18	+	manager *Manager
	19	+	entryPoint domain.EntryPoint
	20	+	packetConn net.PacketConn
	21	+	counters trafficCounters
	22	+
	23	+	started atomic.Bool
	24	+	closed atomic.Bool
	25	+
	26	+	ctx context.Context
	27	+	cancel context.CancelFunc
	28	+	done chan struct{}
	29	+	doneOnce sync.Once
	30	+
	31	+	mu sync.Mutex
	32	+	sessions map[string]*udpSession
	33	+	}
	34	+
	35	+	type udpSession struct {
	36	+	clientAddr net.Addr
	37	+	backend net.Conn
	38	+	lastSeen atomic.Int64
	39	+	done chan struct{}
	40	+	once sync.Once
	41	+	}
	42	+
	43	+	func newUDPEntryPointRuntime(manager Manager, entryPoint domain.EntryPoint, packetConn net.PacketConn) udpEntryPointRuntime {
	44	+	ctx, cancel := context.WithCancel(context.Background())
	45	+	return &udpEntryPointRuntime{
	46	+	manager: manager,
	47	+	entryPoint: entryPoint,
	48	+	packetConn: packetConn,
	49	+	ctx: ctx,
	50	+	cancel: cancel,
	51	+	done: make(chan struct{}),
	52	+	sessions: map[string]*udpSession{},
	53	+	}
	54	+	}
	55	+
	56	+	func (r *udpEntryPointRuntime) start() {
	57	+	if !r.started.CompareAndSwap(false, true) {
	58	+	return
	59	+	}
	60	+	go r.readLoop()
	61	+	go r.expireLoop()
	62	+	}
	63	+
	64	+	func (r *udpEntryPointRuntime) readLoop() {
	65	+	defer r.closeDone()
	66	+	buf := make([]byte, udpBufferSize)
	67	+	for {
	68	+	n, clientAddr, err := r.packetConn.ReadFrom(buf)
	69	+	if err != nil {
	70	+	if r.isClosed() \|\| errors.Is(err, net.ErrClosed) {
	71	+	return
	72	+	}
	73	+	r.counters.totalErrors.Add(1)
	74	+	continue
	75	+	}
	76	+	packet := append([]byte(nil), buf[:n]...)
	77	+	r.handleDatagram(clientAddr, packet)
	78	+	}
	79	+	}
	80	+
	81	+	func (r *udpEntryPointRuntime) handleDatagram(clientAddr net.Addr, packet []byte) {
	82	+	options := effectiveUDPOptions(snapshotUDPOptions(r.manager.snapshot.Load()))
	83	+	session, ok := r.session(clientAddr.String(), clientAddr, options)
	84	+	if !ok {
	85	+	return
	86	+	}
	87	+	session.touch()
	88	+	n, err := session.backend.Write(packet)
	89	+	r.counters.bytesIn.Add(int64(n))
	90	+	if err != nil {
	91	+	r.counters.totalErrors.Add(1)
	92	+	r.removeSession(clientAddr.String())
	93	+	}
	94	+	}
	95	+
	96	+	func (r udpEntryPointRuntime) session(key string, clientAddr net.Addr, options domain.UDPOptions) (udpSession, bool) {
	97	+	r.mu.Lock()
	98	+	if session := r.sessions[key]; session != nil {
	99	+	r.mu.Unlock()
	100	+	return session, true
	101	+	}
	102	+	if options.MaxSessions > 0 && len(r.sessions) >= options.MaxSessions {
	103	+	r.mu.Unlock()
	104	+	r.counters.totalRefused.Add(1)
	105	+	return nil, false
	106	+	}
	107	+	r.mu.Unlock()
	108	+
	109	+	backend, ok := r.resolveUDPBackend()
	110	+	if !ok {
	111	+	r.counters.totalRefused.Add(1)
	112	+	return nil, false
	113	+	}
	114	+	backendConn, err := net.Dial("udp", net.JoinHostPort(backend.Host, strconv.Itoa(backend.Port)))
	115	+	if err != nil {
	116	+	r.counters.totalErrors.Add(1)
	117	+	return nil, false
	118	+	}
	119	+	session := &udpSession{clientAddr: clientAddr, backend: backendConn, done: make(chan struct{})}
	120	+	session.touch()
	121	+
	122	+	r.mu.Lock()
	123	+	if existing := r.sessions[key]; existing != nil {
	124	+	r.mu.Unlock()
	125	+	_ = backendConn.Close()
	126	+	return existing, true
	127	+	}
	128	+	if options.MaxSessions > 0 && len(r.sessions) >= options.MaxSessions {
	129	+	r.mu.Unlock()
	130	+	_ = backendConn.Close()
	131	+	r.counters.totalRefused.Add(1)
	132	+	return nil, false
	133	+	}
	134	+	r.sessions[key] = session
	135	+	r.mu.Unlock()
	136	+
	137	+	r.counters.activeUDPSessions.Add(1)
	138	+	r.counters.totalAccepted.Add(1)
	139	+	go r.backendLoop(key, session)
	140	+	return session, true
	141	+	}
	142	+
	143	+	func (r udpEntryPointRuntime) backendLoop(key string, session udpSession) {
	144	+	defer r.removeSession(key)
	145	+	buf := make([]byte, udpBufferSize)
	146	+	for {
	147	+	n, err := session.backend.Read(buf)
	148	+	if err != nil {
	149	+	if !errors.Is(err, net.ErrClosed) && !r.isClosed() {
	150	+	r.counters.totalErrors.Add(1)
	151	+	}
	152	+	return
	153	+	}
	154	+	written, err := r.packetConn.WriteTo(buf[:n], session.clientAddr)
	155	+	r.counters.bytesOut.Add(int64(written))
	156	+	if err != nil {
	157	+	if !r.isClosed() {
	158	+	r.counters.totalErrors.Add(1)
	159	+	}
	160	+	return
	161	+	}
	162	+	}
	163	+	}
	164	+
	165	+	func (r *udpEntryPointRuntime) expireLoop() {
	166	+	ticker := time.NewTicker(25 * time.Millisecond)
	167	+	defer ticker.Stop()
	168	+	for {
	169	+	select {
	170	+	case <-r.ctx.Done():
	171	+	return
	172	+	case <-ticker.C:
	173	+	r.expireIdleSessions(effectiveUDPOptions(snapshotUDPOptions(r.manager.snapshot.Load())).IdleTimeout)
	174	+	}
	175	+	}
	176	+	}
	177	+
	178	+	func (r *udpEntryPointRuntime) expireIdleSessions(idleTimeout time.Duration) {
	179	+	if idleTimeout <= 0 {
	180	+	idleTimeout = defaultUDPOptions().IdleTimeout
	181	+	}
	182	+	cutoff := time.Now().Add(-idleTimeout).UnixNano()
	183	+	r.mu.Lock()
	184	+	keys := make([]string, 0)
	185	+	for key, session := range r.sessions {
	186	+	if session.lastSeen.Load() <= cutoff {
	187	+	keys = append(keys, key)
	188	+	}
	189	+	}
	190	+	r.mu.Unlock()
	191	+	for _, key := range keys {
	192	+	r.removeSession(key)
	193	+	}
	194	+	}
	195	+
	196	+	func (r *udpEntryPointRuntime) removeSession(key string) {
	197	+	r.mu.Lock()
	198	+	session := r.sessions[key]
	199	+	if session != nil {
	200	+	delete(r.sessions, key)
	201	+	}
	202	+	r.mu.Unlock()
	203	+	if session == nil {
	204	+	return
	205	+	}
	206	+	session.close()
	207	+	r.counters.activeUDPSessions.Add(-1)
	208	+	}
	209	+
	210	+	func (r *udpEntryPointRuntime) resolveUDPBackend() (domain.TrafficBackend, bool) {
	211	+	graph := r.manager.snapshot.Load()
	212	+	if graph == nil {
	213	+	return domain.TrafficBackend{}, false
	214	+	}
	215	+	var router domain.TrafficRouter
	216	+	for _, candidate := range graph.Routers {
	217	+	if candidate.EntryPoint == r.entryPoint.Name && candidate.Protocol == domain.RouterProtocolUDP {
	218	+	if router.Name != "" {
	219	+	return domain.TrafficBackend{}, false
	220	+	}
	221	+	router = candidate
	222	+	}
	223	+	}
	224	+	if router.Name == "" {
	225	+	return domain.TrafficBackend{}, false
	226	+	}
	227	+	for _, service := range graph.Services {
	228	+	if service.Name == router.Service && len(service.Backends) == 1 && service.Backends[0].Protocol == domain.NetworkProtocolUDP {
	229	+	return service.Backends[0], true
	230	+	}
	231	+	}
	232	+	return domain.TrafficBackend{}, false
	233	+	}
	234	+
	235	+	func (r *udpEntryPointRuntime) stop(ctx context.Context, drainTimeout time.Duration) {
	236	+	if r.closed.CompareAndSwap(false, true) {
	237	+	r.cancel()
	238	+	_ = r.packetConn.Close()
	239	+	if !r.started.Load() {
	240	+	r.closeDone()
	241	+	}
	242	+	}
	243	+	select {
	244	+	case <-r.done:
	245	+	case <-ctx.Done():
	246	+	return
	247	+	}
	248	+	if r.waitSessions(ctx, drainTimeout) {
	249	+	return
	250	+	}
	251	+	r.closeSessions()
	252	+	select {
	253	+	case <-r.sessionsDone():
	254	+	case <-ctx.Done():
	255	+	}
	256	+	}
	257	+
	258	+	func (r *udpEntryPointRuntime) drainSessionsAfter(drainTimeout time.Duration) {
	259	+	go func() {
	260	+	if drainTimeout <= 0 {
	261	+	drainTimeout = defaultUDPOptions().DrainTimeout
	262	+	}
	263	+	select {
	264	+	case <-time.After(drainTimeout):
	265	+	r.closeSessions()
	266	+	case <-r.ctx.Done():
	267	+	}
	268	+	}()
	269	+	}
	270	+
	271	+	func (r *udpEntryPointRuntime) waitSessions(ctx context.Context, drainTimeout time.Duration) bool {
	272	+	if drainTimeout <= 0 {
	273	+	drainTimeout = defaultUDPOptions().DrainTimeout
	274	+	}
	275	+	drainCtx, cancel := context.WithTimeout(ctx, drainTimeout)
	276	+	defer cancel()
	277	+	select {
	278	+	case <-r.sessionsDone():
	279	+	return true
	280	+	case <-drainCtx.Done():
	281	+	return false
	282	+	}
	283	+	}
	284	+
	285	+	func (r *udpEntryPointRuntime) closeDone() {
	286	+	r.doneOnce.Do(func() { close(r.done) })
	287	+	}
	288	+
	289	+	func (r *udpEntryPointRuntime) sessionsDone() <-chan struct{} {
	290	+	done := make(chan struct{})
	291	+	go func() {
	292	+	for {
	293	+	r.mu.Lock()
	294	+	count := len(r.sessions)
	295	+	r.mu.Unlock()
	296	+	if count == 0 {
	297	+	close(done)
	298	+	return
	299	+	}
	300	+	time.Sleep(10 * time.Millisecond)
	301	+	}
	302	+	}()
	303	+	return done
	304	+	}
	305	+
	306	+	func (r *udpEntryPointRuntime) closeSessions() {
	307	+	r.mu.Lock()
	308	+	keys := make([]string, 0, len(r.sessions))
	309	+	for key := range r.sessions {
	310	+	keys = append(keys, key)
	311	+	}
	312	+	r.mu.Unlock()
	313	+	for _, key := range keys {
	314	+	r.removeSession(key)
	315	+	}
	316	+	}
	317	+
	318	+	func (r *udpEntryPointRuntime) matches(entryPoint domain.EntryPoint) bool {
	319	+	return r.entryPoint.Name == entryPoint.Name && r.entryPoint.Address == entryPoint.Address && r.entryPoint.Protocol == entryPoint.Protocol
	320	+	}
	321	+
	322	+	func (r *udpEntryPointRuntime) isClosed() bool { return r.closed.Load() }
	323	+
	324	+	func (s *udpSession) touch() { s.lastSeen.Store(time.Now().UnixNano()) }
	325	+
	326	+	func (s *udpSession) close() {
	327	+	s.once.Do(func() {
	328	+	_ = s.backend.Close()
	329	+	close(s.done)
	330	+	})
	331	+	}

@@ -0,0 +1,331 @@
	1	+	package traffic
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"net"
	7	+	"strconv"
	8	+	"sync"
	9	+	"sync/atomic"
	10	+	"time"
	11	+
	12	+	"github.com/bnema/gordon/internal/domain"
	13	+	)
	14	+
	15	+	const udpBufferSize = 64 << 10
	16	+
	17	+	type udpEntryPointRuntime struct {
	18	+	manager *Manager
	19	+	entryPoint domain.EntryPoint
	20	+	packetConn net.PacketConn
	21	+	counters trafficCounters
	22	+
	23	+	started atomic.Bool
	24	+	closed atomic.Bool
	25	+
	26	+	ctx context.Context
	27	+	cancel context.CancelFunc
	28	+	done chan struct{}
	29	+	doneOnce sync.Once
	30	+
	31	+	mu sync.Mutex
	32	+	sessions map[string]*udpSession
	33	+	}
	34	+
	35	+	type udpSession struct {
	36	+	clientAddr net.Addr
	37	+	backend net.Conn
	38	+	lastSeen atomic.Int64
	39	+	done chan struct{}
	40	+	once sync.Once
	41	+	}
	42	+
	43	+	func newUDPEntryPointRuntime(manager Manager, entryPoint domain.EntryPoint, packetConn net.PacketConn) udpEntryPointRuntime {
	44	+	ctx, cancel := context.WithCancel(context.Background())
	45	+	return &udpEntryPointRuntime{
	46	+	manager: manager,
	47	+	entryPoint: entryPoint,
	48	+	packetConn: packetConn,
	49	+	ctx: ctx,
	50	+	cancel: cancel,
	51	+	done: make(chan struct{}),
	52	+	sessions: map[string]*udpSession{},
	53	+	}
	54	+	}
	55	+
	56	+	func (r *udpEntryPointRuntime) start() {
	57	+	if !r.started.CompareAndSwap(false, true) {
	58	+	return
	59	+	}
	60	+	go r.readLoop()
	61	+	go r.expireLoop()
	62	+	}
	63	+
	64	+	func (r *udpEntryPointRuntime) readLoop() {
	65	+	defer r.closeDone()
	66	+	buf := make([]byte, udpBufferSize)
	67	+	for {
	68	+	n, clientAddr, err := r.packetConn.ReadFrom(buf)
	69	+	if err != nil {
	70	+	if r.isClosed() \|\| errors.Is(err, net.ErrClosed) {
	71	+	return
	72	+	}
	73	+	r.counters.totalErrors.Add(1)
	74	+	continue
	75	+	}
	76	+	packet := append([]byte(nil), buf[:n]...)
	77	+	r.handleDatagram(clientAddr, packet)
	78	+	}
	79	+	}
	80	+
	81	+	func (r *udpEntryPointRuntime) handleDatagram(clientAddr net.Addr, packet []byte) {
	82	+	options := effectiveUDPOptions(snapshotUDPOptions(r.manager.snapshot.Load()))
	83	+	session, ok := r.session(clientAddr.String(), clientAddr, options)
	84	+	if !ok {
	85	+	return
	86	+	}
	87	+	session.touch()
	88	+	n, err := session.backend.Write(packet)
	89	+	r.counters.bytesIn.Add(int64(n))
	90	+	if err != nil {
	91	+	r.counters.totalErrors.Add(1)
	92	+	r.removeSession(clientAddr.String())
	93	+	}
	94	+	}
	95	+
	96	+	func (r udpEntryPointRuntime) session(key string, clientAddr net.Addr, options domain.UDPOptions) (udpSession, bool) {
	97	+	r.mu.Lock()
	98	+	if session := r.sessions[key]; session != nil {
	99	+	r.mu.Unlock()
	100	+	return session, true
	101	+	}
	102	+	if options.MaxSessions > 0 && len(r.sessions) >= options.MaxSessions {
	103	+	r.mu.Unlock()
	104	+	r.counters.totalRefused.Add(1)
	105	+	return nil, false
	106	+	}
	107	+	r.mu.Unlock()
	108	+
	109	+	backend, ok := r.resolveUDPBackend()
	110	+	if !ok {
	111	+	r.counters.totalRefused.Add(1)
	112	+	return nil, false
	113	+	}
	114	+	backendConn, err := net.Dial("udp", net.JoinHostPort(backend.Host, strconv.Itoa(backend.Port)))
	115	+	if err != nil {
	116	+	r.counters.totalErrors.Add(1)
	117	+	return nil, false
	118	+	}
	119	+	session := &udpSession{clientAddr: clientAddr, backend: backendConn, done: make(chan struct{})}
	120	+	session.touch()
	121	+
	122	+	r.mu.Lock()
	123	+	if existing := r.sessions[key]; existing != nil {
	124	+	r.mu.Unlock()
	125	+	_ = backendConn.Close()
	126	+	return existing, true
	127	+	}
	128	+	if options.MaxSessions > 0 && len(r.sessions) >= options.MaxSessions {
	129	+	r.mu.Unlock()
	130	+	_ = backendConn.Close()
	131	+	r.counters.totalRefused.Add(1)
	132	+	return nil, false
	133	+	}
	134	+	r.sessions[key] = session
	135	+	r.mu.Unlock()
	136	+
	137	+	r.counters.activeUDPSessions.Add(1)
	138	+	r.counters.totalAccepted.Add(1)
	139	+	go r.backendLoop(key, session)
	140	+	return session, true
	141	+	}
	142	+
	143	+	func (r udpEntryPointRuntime) backendLoop(key string, session udpSession) {
	144	+	defer r.removeSession(key)
	145	+	buf := make([]byte, udpBufferSize)
	146	+	for {
	147	+	n, err := session.backend.Read(buf)
	148	+	if err != nil {
	149	+	if !errors.Is(err, net.ErrClosed) && !r.isClosed() {
	150	+	r.counters.totalErrors.Add(1)
	151	+	}
	152	+	return
	153	+	}
	154	+	written, err := r.packetConn.WriteTo(buf[:n], session.clientAddr)
	155	+	r.counters.bytesOut.Add(int64(written))
	156	+	if err != nil {
	157	+	if !r.isClosed() {
	158	+	r.counters.totalErrors.Add(1)
	159	+	}
	160	+	return
	161	+	}
	162	+	}
	163	+	}
	164	+
	165	+	func (r *udpEntryPointRuntime) expireLoop() {
	166	+	ticker := time.NewTicker(25 * time.Millisecond)
	167	+	defer ticker.Stop()
	168	+	for {
	169	+	select {
	170	+	case <-r.ctx.Done():
	171	+	return
	172	+	case <-ticker.C:
	173	+	r.expireIdleSessions(effectiveUDPOptions(snapshotUDPOptions(r.manager.snapshot.Load())).IdleTimeout)
	174	+	}
	175	+	}
	176	+	}
	177	+
	178	+	func (r *udpEntryPointRuntime) expireIdleSessions(idleTimeout time.Duration) {
	179	+	if idleTimeout <= 0 {
	180	+	idleTimeout = defaultUDPOptions().IdleTimeout
	181	+	}
	182	+	cutoff := time.Now().Add(-idleTimeout).UnixNano()
	183	+	r.mu.Lock()
	184	+	keys := make([]string, 0)
	185	+	for key, session := range r.sessions {
	186	+	if session.lastSeen.Load() <= cutoff {
	187	+	keys = append(keys, key)
	188	+	}
	189	+	}
	190	+	r.mu.Unlock()
	191	+	for _, key := range keys {
	192	+	r.removeSession(key)
	193	+	}
	194	+	}
	195	+
	196	+	func (r *udpEntryPointRuntime) removeSession(key string) {
	197	+	r.mu.Lock()
	198	+	session := r.sessions[key]
	199	+	if session != nil {
	200	+	delete(r.sessions, key)
	201	+	}
	202	+	r.mu.Unlock()
	203	+	if session == nil {
	204	+	return
	205	+	}
	206	+	session.close()
	207	+	r.counters.activeUDPSessions.Add(-1)
	208	+	}
	209	+
	210	+	func (r *udpEntryPointRuntime) resolveUDPBackend() (domain.TrafficBackend, bool) {
	211	+	graph := r.manager.snapshot.Load()
	212	+	if graph == nil {
	213	+	return domain.TrafficBackend{}, false
	214	+	}
	215	+	var router domain.TrafficRouter
	216	+	for _, candidate := range graph.Routers {
	217	+	if candidate.EntryPoint == r.entryPoint.Name && candidate.Protocol == domain.RouterProtocolUDP {
	218	+	if router.Name != "" {
	219	+	return domain.TrafficBackend{}, false
	220	+	}
	221	+	router = candidate
	222	+	}
	223	+	}
	224	+	if router.Name == "" {
	225	+	return domain.TrafficBackend{}, false
	226	+	}
	227	+	for _, service := range graph.Services {
	228	+	if service.Name == router.Service && len(service.Backends) == 1 && service.Backends[0].Protocol == domain.NetworkProtocolUDP {
	229	+	return service.Backends[0], true
	230	+	}
	231	+	}
	232	+	return domain.TrafficBackend{}, false
	233	+	}
	234	+
	235	+	func (r *udpEntryPointRuntime) stop(ctx context.Context, drainTimeout time.Duration) {
	236	+	if r.closed.CompareAndSwap(false, true) {
	237	+	r.cancel()
	238	+	_ = r.packetConn.Close()
	239	+	if !r.started.Load() {
	240	+	r.closeDone()
	241	+	}
	242	+	}
	243	+	select {
	244	+	case <-r.done:
	245	+	case <-ctx.Done():
	246	+	return
	247	+	}
	248	+	if r.waitSessions(ctx, drainTimeout) {
	249	+	return
	250	+	}
	251	+	r.closeSessions()
	252	+	select {
	253	+	case <-r.sessionsDone():
	254	+	case <-ctx.Done():
	255	+	}
	256	+	}
	257	+
	258	+	func (r *udpEntryPointRuntime) drainSessionsAfter(drainTimeout time.Duration) {
	259	+	go func() {
	260	+	if drainTimeout <= 0 {
	261	+	drainTimeout = defaultUDPOptions().DrainTimeout
	262	+	}
	263	+	select {
	264	+	case <-time.After(drainTimeout):
	265	+	r.closeSessions()
	266	+	case <-r.ctx.Done():
	267	+	}
	268	+	}()
	269	+	}
	270	+
	271	+	func (r *udpEntryPointRuntime) waitSessions(ctx context.Context, drainTimeout time.Duration) bool {
	272	+	if drainTimeout <= 0 {
	273	+	drainTimeout = defaultUDPOptions().DrainTimeout
	274	+	}
	275	+	drainCtx, cancel := context.WithTimeout(ctx, drainTimeout)
	276	+	defer cancel()
	277	+	select {
	278	+	case <-r.sessionsDone():
	279	+	return true
	280	+	case <-drainCtx.Done():
	281	+	return false
	282	+	}
	283	+	}
	284	+
	285	+	func (r *udpEntryPointRuntime) closeDone() {
	286	+	r.doneOnce.Do(func() { close(r.done) })
	287	+	}
	288	+
	289	+	func (r *udpEntryPointRuntime) sessionsDone() <-chan struct{} {
	290	+	done := make(chan struct{})
	291	+	go func() {
	292	+	for {
	293	+	r.mu.Lock()
	294	+	count := len(r.sessions)
	295	+	r.mu.Unlock()
	296	+	if count == 0 {
	297	+	close(done)
	298	+	return
	299	+	}
	300	+	time.Sleep(10 * time.Millisecond)
	301	+	}
	302	+	}()
	303	+	return done
	304	+	}
	305	+
	306	+	func (r *udpEntryPointRuntime) closeSessions() {
	307	+	r.mu.Lock()
	308	+	keys := make([]string, 0, len(r.sessions))
	309	+	for key := range r.sessions {
	310	+	keys = append(keys, key)
	311	+	}
	312	+	r.mu.Unlock()
	313	+	for _, key := range keys {
	314	+	r.removeSession(key)
	315	+	}
	316	+	}
	317	+
	318	+	func (r *udpEntryPointRuntime) matches(entryPoint domain.EntryPoint) bool {
	319	+	return r.entryPoint.Name == entryPoint.Name && r.entryPoint.Address == entryPoint.Address && r.entryPoint.Protocol == entryPoint.Protocol
	320	+	}
	321	+
	322	+	func (r *udpEntryPointRuntime) isClosed() bool { return r.closed.Load() }
	323	+
	324	+	func (s *udpSession) touch() { s.lastSeen.Store(time.Now().UnixNano()) }
	325	+
	326	+	func (s *udpSession) close() {
	327	+	s.once.Do(func() {
	328	+	_ = s.backend.Close()
	329	+	close(s.done)
	330	+	})
	331	+	}

A internal/adapters/in/traffic/udp_test.go +232 -0

@@ -0,0 +1,232 @@
	1	+	package traffic
	2	+
	3	+	import (
	4	+	"context"
	5	+	"net"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/zerowrap"
	10	+	"github.com/stretchr/testify/assert"
	11	+	"github.com/stretchr/testify/require"
	12	+
	13	+	"github.com/bnema/gordon/internal/domain"
	14	+	)
	15	+
	16	+	func TestUDPPassthroughEcho(t *testing.T) {
	17	+	backend := startUDPEchoServer(t)
	18	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	19	+	manager := NewManager(zerowrap.Default())
	20	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	21	+	defer shutdownManager(t, manager)
	22	+
	23	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	24	+	defer conn.Close()
	25	+	assertUDPRoundTrip(t, conn, "hello udp")
	26	+	}
	27	+
	28	+	func TestUDPTwoClientsGetIsolatedSessions(t *testing.T) {
	29	+	backend := startUDPEchoServer(t)
	30	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	31	+	manager := NewManager(zerowrap.Default())
	32	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	33	+	defer shutdownManager(t, manager)
	34	+
	35	+	first := dialUDP(t, graph.EntryPoints[0].Address)
	36	+	defer first.Close()
	37	+	second := dialUDP(t, graph.EntryPoints[0].Address)
	38	+	defer second.Close()
	39	+	assertUDPRoundTrip(t, first, "first")
	40	+	assertUDPRoundTrip(t, second, "second")
	41	+	assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 2 }, time.Second, 10*time.Millisecond)
	42	+	}
	43	+
	44	+	func TestUDPIdleSessionExpires(t *testing.T) {
	45	+	backend := startUDPEchoServer(t)
	46	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	47	+	graph.Options.UDP.IdleTimeout = 40 * time.Millisecond
	48	+	manager := NewManager(zerowrap.Default())
	49	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	50	+	defer shutdownManager(t, manager)
	51	+
	52	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	53	+	defer conn.Close()
	54	+	assertUDPRoundTrip(t, conn, "expire")
	55	+	assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
	56	+	}
	57	+
	58	+	func TestUDPMaxSessionsRejectsOverflow(t *testing.T) {
	59	+	backend := startUDPEchoServer(t)
	60	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	61	+	graph.Options.UDP.MaxSessions = 1
	62	+	manager := NewManager(zerowrap.Default())
	63	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	64	+	defer shutdownManager(t, manager)
	65	+
	66	+	first := dialUDP(t, graph.EntryPoints[0].Address)
	67	+	defer first.Close()
	68	+	assertUDPRoundTrip(t, first, "first")
	69	+
	70	+	second := dialUDP(t, graph.EntryPoints[0].Address)
	71	+	defer second.Close()
	72	+	_, err := second.Write([]byte("second"))
	73	+	require.NoError(t, err)
	74	+	_ = second.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
	75	+	buf := make([]byte, 32)
	76	+	_, err = second.Read(buf)
	77	+	require.Error(t, err)
	78	+	assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
	79	+	}
	80	+
	81	+	func TestUDPRemovedRouterWithRetainedEntryPointDrainsSession(t *testing.T) {
	82	+	backend := startUDPEchoServer(t)
	83	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	84	+	graph.Options.UDP.DrainTimeout = 50 * time.Millisecond
	85	+	manager := NewManager(zerowrap.Default())
	86	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	87	+	defer shutdownManager(t, manager)
	88	+
	89	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	90	+	defer conn.Close()
	91	+	assertUDPRoundTrip(t, conn, "held")
	92	+
	93	+	next := graph
	94	+	next.Routers = nil
	95	+	next.Services = nil
	96	+	require.NoError(t, manager.Apply(context.Background(), &next))
	97	+	assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
	98	+
	99	+	_, err := conn.Write([]byte("after"))
	100	+	require.NoError(t, err)
	101	+	_ = conn.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
	102	+	buf := make([]byte, 32)
	103	+	_, err = conn.Read(buf)
	104	+	require.Error(t, err)
	105	+	}
	106	+
	107	+	func TestUDPRemovedRouterDrainsThenClosesSessions(t *testing.T) {
	108	+	backend := startUDPEchoServer(t)
	109	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	110	+	graph.Options.UDP.DrainTimeout = 50 * time.Millisecond
	111	+	manager := NewManager(zerowrap.Default())
	112	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	113	+	defer shutdownManager(t, manager)
	114	+
	115	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	116	+	defer conn.Close()
	117	+	assertUDPRoundTrip(t, conn, "held")
	118	+	require.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 1 }, time.Second, 10*time.Millisecond)
	119	+
	120	+	empty := domain.TrafficGraph{Options: graph.Options}
	121	+	started := time.Now()
	122	+	require.NoError(t, manager.Apply(context.Background(), &empty))
	123	+	assert.GreaterOrEqual(t, time.Since(started), 40*time.Millisecond)
	124	+	assert.Equal(t, int64(0), manager.Status().Counters.ActiveUDPSessions)
	125	+	}
	126	+
	127	+	func TestUDPShutdownClosesSocket(t *testing.T) {
	128	+	backend := startUDPEchoServer(t)
	129	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	130	+	manager := NewManager(zerowrap.Default())
	131	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	132	+	require.NoError(t, manager.Shutdown(context.Background()))
	133	+
	134	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	135	+	defer conn.Close()
	136	+	_, err := conn.Write([]byte("closed"))
	137	+	require.NoError(t, err)
	138	+	_ = conn.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
	139	+	buf := make([]byte, 32)
	140	+	_, err = conn.Read(buf)
	141	+	require.Error(t, err)
	142	+	}
	143	+
	144	+	func TestUDPStatusCountersTrackDatagrams(t *testing.T) {
	145	+	backend := startUDPEchoServer(t)
	146	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	147	+	manager := NewManager(zerowrap.Default())
	148	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	149	+	defer shutdownManager(t, manager)
	150	+
	151	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	152	+	defer conn.Close()
	153	+	assertUDPRoundTrip(t, conn, "bytes")
	154	+
	155	+	assert.Eventually(t, func() bool {
	156	+	status := manager.Status()
	157	+	return status.Counters.TotalAccepted == 1 &&
	158	+	status.Counters.ActiveUDPSessions == 1 &&
	159	+	status.Counters.BytesIn >= int64(len("bytes")) &&
	160	+	status.Counters.BytesOut >= int64(len("bytes"))
	161	+	}, time.Second, 10*time.Millisecond)
	162	+	}
	163	+
	164	+	func udpGraph(t *testing.T, listenAddress string, backendAddress string) domain.TrafficGraph {
	165	+	t.Helper()
	166	+	backend, err := backendFromAddress("echo:udp", backendAddress)
	167	+	require.NoError(t, err)
	168	+	backend.Protocol = domain.NetworkProtocolUDP
	169	+	ref := serviceRef("echo", "udp")
	170	+	graph := domain.TrafficGraph{
	171	+	Options: domain.TrafficOptions{UDP: domain.UDPOptions{IdleTimeout: time.Minute, DrainTimeout: 50 * time.Millisecond}},
	172	+	EntryPoints: []domain.EntryPoint{{Name: "udp", Address: listenAddress, Protocol: domain.EntryPointProtocolUDP}},
	173	+	Routers: []domain.TrafficRouter{{Name: "echo", EntryPoint: "udp", Protocol: domain.RouterProtocolUDP, Service: ref}},
	174	+	Services: []domain.TrafficService{{Name: ref, Backends: []domain.TrafficBackend{backend}}},
	175	+	}
	176	+	require.NoError(t, graph.Validate())
	177	+	return graph
	178	+	}
	179	+
	180	+	func freeUDPAddress(t *testing.T) string {
	181	+	t.Helper()
	182	+	conn, err := net.ListenPacket("udp", "127.0.0.1:0")
	183	+	require.NoError(t, err)
	184	+	address := conn.LocalAddr().String()
	185	+	require.NoError(t, conn.Close())
	186	+	return address
	187	+	}
	188	+
	189	+	func dialUDP(t testing.T, address string) net.UDPConn {
	190	+	t.Helper()
	191	+	udpAddr, err := net.ResolveUDPAddr("udp", address)
	192	+	require.NoError(t, err)
	193	+	conn, err := net.DialUDP("udp", nil, udpAddr)
	194	+	require.NoError(t, err)
	195	+	return conn
	196	+	}
	197	+
	198	+	func assertUDPRoundTrip(t testing.T, conn net.UDPConn, message string) {
	199	+	t.Helper()
	200	+	_, err := conn.Write([]byte(message))
	201	+	require.NoError(t, err)
	202	+	_ = conn.SetReadDeadline(time.Now().Add(time.Second))
	203	+	buf := make([]byte, 128)
	204	+	n, err := conn.Read(buf)
	205	+	require.NoError(t, err)
	206	+	assert.Equal(t, message, string(buf[:n]))
	207	+	}
	208	+
	209	+	type udpEchoServer struct{ address string }
	210	+
	211	+	func startUDPEchoServer(t *testing.T) udpEchoServer {
	212	+	t.Helper()
	213	+	conn, err := net.ListenPacket("udp", "127.0.0.1:0")
	214	+	require.NoError(t, err)
	215	+	done := make(chan struct{})
	216	+	go func() {
	217	+	defer close(done)
	218	+	buf := make([]byte, udpBufferSize)
	219	+	for {
	220	+	n, addr, err := conn.ReadFrom(buf)
	221	+	if err != nil {
	222	+	return
	223	+	}
	224	+	_, _ = conn.WriteTo(buf[:n], addr)
	225	+	}
	226	+	}()
	227	+	t.Cleanup(func() {
	228	+	_ = conn.Close()
	229	+	<-done
	230	+	})
	231	+	return udpEchoServer{address: conn.LocalAddr().String()}
	232	+	}

@@ -0,0 +1,232 @@
	1	+	package traffic
	2	+
	3	+	import (
	4	+	"context"
	5	+	"net"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/zerowrap"
	10	+	"github.com/stretchr/testify/assert"
	11	+	"github.com/stretchr/testify/require"
	12	+
	13	+	"github.com/bnema/gordon/internal/domain"
	14	+	)
	15	+
	16	+	func TestUDPPassthroughEcho(t *testing.T) {
	17	+	backend := startUDPEchoServer(t)
	18	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	19	+	manager := NewManager(zerowrap.Default())
	20	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	21	+	defer shutdownManager(t, manager)
	22	+
	23	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	24	+	defer conn.Close()
	25	+	assertUDPRoundTrip(t, conn, "hello udp")
	26	+	}
	27	+
	28	+	func TestUDPTwoClientsGetIsolatedSessions(t *testing.T) {
	29	+	backend := startUDPEchoServer(t)
	30	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	31	+	manager := NewManager(zerowrap.Default())
	32	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	33	+	defer shutdownManager(t, manager)
	34	+
	35	+	first := dialUDP(t, graph.EntryPoints[0].Address)
	36	+	defer first.Close()
	37	+	second := dialUDP(t, graph.EntryPoints[0].Address)
	38	+	defer second.Close()
	39	+	assertUDPRoundTrip(t, first, "first")
	40	+	assertUDPRoundTrip(t, second, "second")
	41	+	assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 2 }, time.Second, 10*time.Millisecond)
	42	+	}
	43	+
	44	+	func TestUDPIdleSessionExpires(t *testing.T) {
	45	+	backend := startUDPEchoServer(t)
	46	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	47	+	graph.Options.UDP.IdleTimeout = 40 * time.Millisecond
	48	+	manager := NewManager(zerowrap.Default())
	49	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	50	+	defer shutdownManager(t, manager)
	51	+
	52	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	53	+	defer conn.Close()
	54	+	assertUDPRoundTrip(t, conn, "expire")
	55	+	assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
	56	+	}
	57	+
	58	+	func TestUDPMaxSessionsRejectsOverflow(t *testing.T) {
	59	+	backend := startUDPEchoServer(t)
	60	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	61	+	graph.Options.UDP.MaxSessions = 1
	62	+	manager := NewManager(zerowrap.Default())
	63	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	64	+	defer shutdownManager(t, manager)
	65	+
	66	+	first := dialUDP(t, graph.EntryPoints[0].Address)
	67	+	defer first.Close()
	68	+	assertUDPRoundTrip(t, first, "first")
	69	+
	70	+	second := dialUDP(t, graph.EntryPoints[0].Address)
	71	+	defer second.Close()
	72	+	_, err := second.Write([]byte("second"))
	73	+	require.NoError(t, err)
	74	+	_ = second.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
	75	+	buf := make([]byte, 32)
	76	+	_, err = second.Read(buf)
	77	+	require.Error(t, err)
	78	+	assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
	79	+	}
	80	+
	81	+	func TestUDPRemovedRouterWithRetainedEntryPointDrainsSession(t *testing.T) {
	82	+	backend := startUDPEchoServer(t)
	83	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	84	+	graph.Options.UDP.DrainTimeout = 50 * time.Millisecond
	85	+	manager := NewManager(zerowrap.Default())
	86	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	87	+	defer shutdownManager(t, manager)
	88	+
	89	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	90	+	defer conn.Close()
	91	+	assertUDPRoundTrip(t, conn, "held")
	92	+
	93	+	next := graph
	94	+	next.Routers = nil
	95	+	next.Services = nil
	96	+	require.NoError(t, manager.Apply(context.Background(), &next))
	97	+	assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
	98	+
	99	+	_, err := conn.Write([]byte("after"))
	100	+	require.NoError(t, err)
	101	+	_ = conn.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
	102	+	buf := make([]byte, 32)
	103	+	_, err = conn.Read(buf)
	104	+	require.Error(t, err)
	105	+	}
	106	+
	107	+	func TestUDPRemovedRouterDrainsThenClosesSessions(t *testing.T) {
	108	+	backend := startUDPEchoServer(t)
	109	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	110	+	graph.Options.UDP.DrainTimeout = 50 * time.Millisecond
	111	+	manager := NewManager(zerowrap.Default())
	112	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	113	+	defer shutdownManager(t, manager)
	114	+
	115	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	116	+	defer conn.Close()
	117	+	assertUDPRoundTrip(t, conn, "held")
	118	+	require.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 1 }, time.Second, 10*time.Millisecond)
	119	+
	120	+	empty := domain.TrafficGraph{Options: graph.Options}
	121	+	started := time.Now()
	122	+	require.NoError(t, manager.Apply(context.Background(), &empty))
	123	+	assert.GreaterOrEqual(t, time.Since(started), 40*time.Millisecond)
	124	+	assert.Equal(t, int64(0), manager.Status().Counters.ActiveUDPSessions)
	125	+	}
	126	+
	127	+	func TestUDPShutdownClosesSocket(t *testing.T) {
	128	+	backend := startUDPEchoServer(t)
	129	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	130	+	manager := NewManager(zerowrap.Default())
	131	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	132	+	require.NoError(t, manager.Shutdown(context.Background()))
	133	+
	134	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	135	+	defer conn.Close()
	136	+	_, err := conn.Write([]byte("closed"))
	137	+	require.NoError(t, err)
	138	+	_ = conn.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
	139	+	buf := make([]byte, 32)
	140	+	_, err = conn.Read(buf)
	141	+	require.Error(t, err)
	142	+	}
	143	+
	144	+	func TestUDPStatusCountersTrackDatagrams(t *testing.T) {
	145	+	backend := startUDPEchoServer(t)
	146	+	graph := udpGraph(t, freeUDPAddress(t), backend.address)
	147	+	manager := NewManager(zerowrap.Default())
	148	+	require.NoError(t, manager.Apply(context.Background(), &graph))
	149	+	defer shutdownManager(t, manager)
	150	+
	151	+	conn := dialUDP(t, graph.EntryPoints[0].Address)
	152	+	defer conn.Close()
	153	+	assertUDPRoundTrip(t, conn, "bytes")
	154	+
	155	+	assert.Eventually(t, func() bool {
	156	+	status := manager.Status()
	157	+	return status.Counters.TotalAccepted == 1 &&
	158	+	status.Counters.ActiveUDPSessions == 1 &&
	159	+	status.Counters.BytesIn >= int64(len("bytes")) &&
	160	+	status.Counters.BytesOut >= int64(len("bytes"))
	161	+	}, time.Second, 10*time.Millisecond)
	162	+	}
	163	+
	164	+	func udpGraph(t *testing.T, listenAddress string, backendAddress string) domain.TrafficGraph {
	165	+	t.Helper()
	166	+	backend, err := backendFromAddress("echo:udp", backendAddress)
	167	+	require.NoError(t, err)
	168	+	backend.Protocol = domain.NetworkProtocolUDP
	169	+	ref := serviceRef("echo", "udp")
	170	+	graph := domain.TrafficGraph{
	171	+	Options: domain.TrafficOptions{UDP: domain.UDPOptions{IdleTimeout: time.Minute, DrainTimeout: 50 * time.Millisecond}},
	172	+	EntryPoints: []domain.EntryPoint{{Name: "udp", Address: listenAddress, Protocol: domain.EntryPointProtocolUDP}},
	173	+	Routers: []domain.TrafficRouter{{Name: "echo", EntryPoint: "udp", Protocol: domain.RouterProtocolUDP, Service: ref}},
	174	+	Services: []domain.TrafficService{{Name: ref, Backends: []domain.TrafficBackend{backend}}},
	175	+	}
	176	+	require.NoError(t, graph.Validate())
	177	+	return graph
	178	+	}
	179	+
	180	+	func freeUDPAddress(t *testing.T) string {
	181	+	t.Helper()
	182	+	conn, err := net.ListenPacket("udp", "127.0.0.1:0")
	183	+	require.NoError(t, err)
	184	+	address := conn.LocalAddr().String()
	185	+	require.NoError(t, conn.Close())
	186	+	return address
	187	+	}
	188	+
	189	+	func dialUDP(t testing.T, address string) net.UDPConn {
	190	+	t.Helper()
	191	+	udpAddr, err := net.ResolveUDPAddr("udp", address)
	192	+	require.NoError(t, err)
	193	+	conn, err := net.DialUDP("udp", nil, udpAddr)
	194	+	require.NoError(t, err)
	195	+	return conn
	196	+	}
	197	+
	198	+	func assertUDPRoundTrip(t testing.T, conn net.UDPConn, message string) {
	199	+	t.Helper()
	200	+	_, err := conn.Write([]byte(message))
	201	+	require.NoError(t, err)
	202	+	_ = conn.SetReadDeadline(time.Now().Add(time.Second))
	203	+	buf := make([]byte, 128)
	204	+	n, err := conn.Read(buf)
	205	+	require.NoError(t, err)
	206	+	assert.Equal(t, message, string(buf[:n]))
	207	+	}
	208	+
	209	+	type udpEchoServer struct{ address string }
	210	+
	211	+	func startUDPEchoServer(t *testing.T) udpEchoServer {
	212	+	t.Helper()
	213	+	conn, err := net.ListenPacket("udp", "127.0.0.1:0")
	214	+	require.NoError(t, err)
	215	+	done := make(chan struct{})
	216	+	go func() {
	217	+	defer close(done)
	218	+	buf := make([]byte, udpBufferSize)
	219	+	for {
	220	+	n, addr, err := conn.ReadFrom(buf)
	221	+	if err != nil {
	222	+	return
	223	+	}
	224	+	_, _ = conn.WriteTo(buf[:n], addr)
	225	+	}
	226	+	}()
	227	+	t.Cleanup(func() {
	228	+	_ = conn.Close()
	229	+	<-done
	230	+	})
	231	+	return udpEchoServer{address: conn.LocalAddr().String()}
	232	+	}