detached 2eed61d

feat(traffic): add tcp passthrough entrypoints

commit
2eed61d58a507ce7d2b2e12e1c89157af6fab2d0
author
brice · · 2026-06-29 21:27 UTC
parent
2443a00
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +910 -0
Unified
UnifiedSplit
A internal/adapters/in/traffic/manager.go +298 -0
@@ -0,0 +1,298 @@
1+// Package traffic owns Gordon's runtime traffic entrypoints.
2+package traffic
3+
4+import (
5+ "context"
6+ "errors"
7+ "fmt"
8+ "net"
9+ "sort"
10+ "sync"
11+ "sync/atomic"
12+ "time"
13+
14+ "github.com/bnema/zerowrap"
15+
16+ "github.com/bnema/gordon/internal/domain"
17+)
18+
19+const (
20+ reloadStatusOK = "ok"
21+ reloadStatusError = "error"
22+)
23+
24+// Manager owns runtime entrypoints for the traffic plane.
25+type Manager struct {
26+ log zerowrap.Logger
27+
28+ mu sync.Mutex
29+ snapshot atomic.Pointer[domain.TrafficGraph]
30+ listeners map[string]*entryPointRuntime
31+
32+ lastReloadStatus string
33+ lastReloadError string
34+}
35+
36+// NewManager creates a traffic manager.
37+func NewManager(log zerowrap.Logger) *Manager {
38+ return &Manager{
39+ log: log,
40+ listeners: map[string]*entryPointRuntime{},
41+ lastReloadStatus: reloadStatusOK,
42+ }
43+}
44+
45+// Apply validates and applies a new traffic graph snapshot.
46+func (m *Manager) Apply(ctx context.Context, graph *domain.TrafficGraph) error {
47+ if graph == nil {
48+ return m.recordReloadError(errors.New("traffic graph is required"))
49+ }
50+ if err := graph.Validate(); err != nil {
51+ return m.recordReloadError(err)
52+ }
53+
54+ nextGraph := cloneTrafficGraph(*graph)
55+ m.mu.Lock()
56+ defer m.mu.Unlock()
57+
58+ newListeners, createdListeners, err := m.prepareTCPListeners(ctx, &nextGraph)
59+ if err != nil {
60+ m.lastReloadStatus = reloadStatusError
61+ m.lastReloadError = err.Error()
62+ return err
63+ }
64+
65+ oldListeners := m.listeners
66+ m.listeners = newListeners
67+ m.snapshot.Store(&nextGraph)
68+ for _, runtime := range createdListeners {
69+ runtime.start()
70+ }
71+ m.lastReloadStatus = reloadStatusOK
72+ m.lastReloadError = ""
73+
74+ drainTimeout := effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout
75+ for name, runtime := range oldListeners {
76+ if newListeners[name] == runtime {
77+ continue
78+ }
79+ if runtime.shouldStopWith(newListeners[name]) {
80+ runtime.stop(ctx, drainTimeout)
81+ }
82+ }
83+ return nil
84+}
85+
86+// Shutdown stops all listeners and active TCP streams.
87+func (m *Manager) Shutdown(ctx context.Context) error {
88+ m.mu.Lock()
89+ defer m.mu.Unlock()
90+
91+ listeners := m.listeners
92+ m.listeners = map[string]*entryPointRuntime{}
93+ graph := m.snapshot.Load()
94+ m.snapshot.Store(nil)
95+
96+ drainTimeout := defaultTCPOptions().DrainTimeout
97+ if graph != nil {
98+ drainTimeout = effectiveTCPOptions(graph.Options.TCP).DrainTimeout
99+ }
100+ for _, runtime := range listeners {
101+ runtime.stop(ctx, drainTimeout)
102+ }
103+ return nil
104+}
105+
106+// Status returns a consistent snapshot of manager state and counters.
107+func (m *Manager) Status() domain.TrafficStatus {
108+ m.mu.Lock()
109+ listeners := make(map[string]*entryPointRuntime, len(m.listeners))
110+ for name, runtime := range m.listeners {
111+ listeners[name] = runtime
112+ }
113+ status := domain.TrafficStatus{LastReloadStatus: m.lastReloadStatus, LastReloadError: m.lastReloadError}
114+ graph := m.snapshot.Load()
115+ m.mu.Unlock()
116+
117+ if graph == nil {
118+ return status
119+ }
120+
121+ status.EntryPoints = entryPointStatuses(graph.EntryPoints, listeners)
122+ status.Routers = routerStatuses(graph.Routers, listeners)
123+ status.Services = serviceStatuses(graph.Services)
124+ status.Counters = aggregateCounters(status.EntryPoints)
125+ return status
126+}
127+
128+func (m *Manager) recordReloadError(err error) error {
129+ m.mu.Lock()
130+ m.lastReloadStatus = reloadStatusError
131+ m.lastReloadError = err.Error()
132+ m.mu.Unlock()
133+ return err
134+}
135+
136+func (m *Manager) prepareTCPListeners(ctx context.Context, graph *domain.TrafficGraph) (map[string]*entryPointRuntime, []*entryPointRuntime, error) {
137+ current := make(map[string]*entryPointRuntime, len(m.listeners))
138+ for name, runtime := range m.listeners {
139+ current[name] = runtime
140+ }
141+
142+ next := make(map[string]*entryPointRuntime, len(current))
143+ created := []*entryPointRuntime{}
144+ for _, entryPoint := range graph.EntryPoints {
145+ if entryPoint.Protocol != domain.EntryPointProtocolTCP {
146+ continue
147+ }
148+ if runtime := current[entryPoint.Name]; runtime != nil && runtime.matches(entryPoint) {
149+ next[entryPoint.Name] = runtime
150+ continue
151+ }
152+ runtime, err := m.bindTCPEntryPoint(ctx, entryPoint)
153+ if err != nil {
154+ for _, createdRuntime := range created {
155+ createdRuntime.stop(ctx, effectiveTCPOptions(graph.Options.TCP).DrainTimeout)
156+ }
157+ return nil, nil, err
158+ }
159+ next[entryPoint.Name] = runtime
160+ created = append(created, runtime)
161+ }
162+ return next, created, nil
163+}
164+
165+func (m *Manager) bindTCPEntryPoint(ctx context.Context, entryPoint domain.EntryPoint) (*entryPointRuntime, error) {
166+ listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", entryPoint.Address)
167+ if err != nil {
168+ return nil, fmt.Errorf("bind tcp entrypoint %q on %s: %w", entryPoint.Name, entryPoint.Address, err)
169+ }
170+ runtime := newEntryPointRuntime(m, entryPoint, listener)
171+ return runtime, nil
172+}
173+
174+func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]*entryPointRuntime) []domain.EntryPointStatus {
175+ statuses := make([]domain.EntryPointStatus, 0, len(entries))
176+ for _, entry := range entries {
177+ status := domain.EntryPointStatus{Name: entry.Name, Address: entry.Address, Protocol: entry.Protocol}
178+ if runtime := listeners[entry.Name]; runtime != nil {
179+ counters := runtime.counters.snapshot()
180+ status.Active = !runtime.isClosed()
181+ status.ActiveTCPConnections = counters.ActiveTCPConnections
182+ status.TotalAccepted = counters.TotalAccepted
183+ status.TotalRefused = counters.TotalRefused
184+ status.TotalErrors = counters.TotalErrors
185+ status.BytesIn = counters.BytesIn
186+ status.BytesOut = counters.BytesOut
187+ }
188+ statuses = append(statuses, status)
189+ }
190+ sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
191+ return statuses
192+}
193+
194+func routerStatuses(routers []domain.TrafficRouter, listeners map[string]*entryPointRuntime) []domain.TrafficRouterStatus {
195+ statuses := make([]domain.TrafficRouterStatus, 0, len(routers))
196+ for _, router := range routers {
197+ status := domain.TrafficRouterStatus{Name: router.Name, EntryPoint: router.EntryPoint, Protocol: router.Protocol, Active: true}
198+ if runtime := listeners[router.EntryPoint]; runtime != nil {
199+ counters := runtime.counters.snapshot()
200+ status.ActiveTCPConnections = counters.ActiveTCPConnections
201+ status.TotalAccepted = counters.TotalAccepted
202+ status.TotalRefused = counters.TotalRefused
203+ status.TotalErrors = counters.TotalErrors
204+ status.BytesIn = counters.BytesIn
205+ status.BytesOut = counters.BytesOut
206+ }
207+ statuses = append(statuses, status)
208+ }
209+ sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
210+ return statuses
211+}
212+
213+func serviceStatuses(services []domain.TrafficService) []domain.TrafficServiceStatus {
214+ statuses := make([]domain.TrafficServiceStatus, 0, len(services))
215+ for _, service := range services {
216+ status := domain.TrafficServiceStatus{Name: service.Name, Active: true, Backends: make([]domain.TrafficBackendStatus, 0, len(service.Backends))}
217+ for _, backend := range service.Backends {
218+ status.Backends = append(status.Backends, domain.TrafficBackendStatus{Name: backend.Name, Active: true})
219+ }
220+ statuses = append(statuses, status)
221+ }
222+ sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
223+ return statuses
224+}
225+
226+func aggregateCounters(entries []domain.EntryPointStatus) domain.TrafficCounters {
227+ var counters domain.TrafficCounters
228+ for _, entry := range entries {
229+ counters.ActiveTCPConnections += entry.ActiveTCPConnections
230+ counters.ActiveUDPSessions += entry.ActiveUDPSessions
231+ counters.TotalAccepted += entry.TotalAccepted
232+ counters.TotalRefused += entry.TotalRefused
233+ counters.TotalErrors += entry.TotalErrors
234+ counters.BytesIn += entry.BytesIn
235+ counters.BytesOut += entry.BytesOut
236+ }
237+ return counters
238+}
239+
240+func cloneTrafficGraph(graph domain.TrafficGraph) domain.TrafficGraph {
241+ clone := graph
242+ clone.EntryPoints = append([]domain.EntryPoint{}, graph.EntryPoints...)
243+ clone.Routers = append([]domain.TrafficRouter{}, graph.Routers...)
244+ clone.Services = append([]domain.TrafficService{}, graph.Services...)
245+ for i := range clone.Services {
246+ clone.Services[i].Backends = append([]domain.TrafficBackend{}, graph.Services[i].Backends...)
247+ }
248+ return clone
249+}
250+
251+func snapshotTCPOptions(graph *domain.TrafficGraph) domain.TCPOptions {
252+ if graph == nil {
253+ return domain.TCPOptions{}
254+ }
255+ return graph.Options.TCP
256+}
257+
258+type trafficCounters struct {
259+ activeTCPConnections atomic.Int64
260+ totalAccepted atomic.Int64
261+ totalRefused atomic.Int64
262+ totalErrors atomic.Int64
263+ bytesIn atomic.Int64
264+ bytesOut atomic.Int64
265+}
266+
267+func (c *trafficCounters) snapshot() domain.TrafficCounters {
268+ return domain.TrafficCounters{
269+ ActiveTCPConnections: c.activeTCPConnections.Load(),
270+ TotalAccepted: c.totalAccepted.Load(),
271+ TotalRefused: c.totalRefused.Load(),
272+ TotalErrors: c.totalErrors.Load(),
273+ BytesIn: c.bytesIn.Load(),
274+ BytesOut: c.bytesOut.Load(),
275+ }
276+}
277+
278+func defaultTCPOptions() domain.TCPOptions {
279+ return domain.TCPOptions{
280+ DialTimeout: 10 * time.Second,
281+ IdleTimeout: 5 * time.Minute,
282+ DrainTimeout: 30 * time.Second,
283+ }
284+}
285+
286+func effectiveTCPOptions(options domain.TCPOptions) domain.TCPOptions {
287+ defaults := defaultTCPOptions()
288+ if options.DialTimeout == 0 {
289+ options.DialTimeout = defaults.DialTimeout
290+ }
291+ if options.IdleTimeout == 0 {
292+ options.IdleTimeout = defaults.IdleTimeout
293+ }
294+ if options.DrainTimeout == 0 {
295+ options.DrainTimeout = defaults.DrainTimeout
296+ }
297+ return options
298+}
@@ -0,0 +1,298 @@
1+// Package traffic owns Gordon's runtime traffic entrypoints.
2+package traffic
3+
4+import (
5+ "context"
6+ "errors"
7+ "fmt"
8+ "net"
9+ "sort"
10+ "sync"
11+ "sync/atomic"
12+ "time"
13+
14+ "github.com/bnema/zerowrap"
15+
16+ "github.com/bnema/gordon/internal/domain"
17+)
18+
19+const (
20+ reloadStatusOK = "ok"
21+ reloadStatusError = "error"
22+)
23+
24+// Manager owns runtime entrypoints for the traffic plane.
25+type Manager struct {
26+ log zerowrap.Logger
27+
28+ mu sync.Mutex
29+ snapshot atomic.Pointer[domain.TrafficGraph]
30+ listeners map[string]*entryPointRuntime
31+
32+ lastReloadStatus string
33+ lastReloadError string
34+}
35+
36+// NewManager creates a traffic manager.
37+func NewManager(log zerowrap.Logger) *Manager {
38+ return &Manager{
39+ log: log,
40+ listeners: map[string]*entryPointRuntime{},
41+ lastReloadStatus: reloadStatusOK,
42+ }
43+}
44+
45+// Apply validates and applies a new traffic graph snapshot.
46+func (m *Manager) Apply(ctx context.Context, graph *domain.TrafficGraph) error {
47+ if graph == nil {
48+ return m.recordReloadError(errors.New("traffic graph is required"))
49+ }
50+ if err := graph.Validate(); err != nil {
51+ return m.recordReloadError(err)
52+ }
53+
54+ nextGraph := cloneTrafficGraph(*graph)
55+ m.mu.Lock()
56+ defer m.mu.Unlock()
57+
58+ newListeners, createdListeners, err := m.prepareTCPListeners(ctx, &nextGraph)
59+ if err != nil {
60+ m.lastReloadStatus = reloadStatusError
61+ m.lastReloadError = err.Error()
62+ return err
63+ }
64+
65+ oldListeners := m.listeners
66+ m.listeners = newListeners
67+ m.snapshot.Store(&nextGraph)
68+ for _, runtime := range createdListeners {
69+ runtime.start()
70+ }
71+ m.lastReloadStatus = reloadStatusOK
72+ m.lastReloadError = ""
73+
74+ drainTimeout := effectiveTCPOptions(snapshotTCPOptions(&nextGraph)).DrainTimeout
75+ for name, runtime := range oldListeners {
76+ if newListeners[name] == runtime {
77+ continue
78+ }
79+ if runtime.shouldStopWith(newListeners[name]) {
80+ runtime.stop(ctx, drainTimeout)
81+ }
82+ }
83+ return nil
84+}
85+
86+// Shutdown stops all listeners and active TCP streams.
87+func (m *Manager) Shutdown(ctx context.Context) error {
88+ m.mu.Lock()
89+ defer m.mu.Unlock()
90+
91+ listeners := m.listeners
92+ m.listeners = map[string]*entryPointRuntime{}
93+ graph := m.snapshot.Load()
94+ m.snapshot.Store(nil)
95+
96+ drainTimeout := defaultTCPOptions().DrainTimeout
97+ if graph != nil {
98+ drainTimeout = effectiveTCPOptions(graph.Options.TCP).DrainTimeout
99+ }
100+ for _, runtime := range listeners {
101+ runtime.stop(ctx, drainTimeout)
102+ }
103+ return nil
104+}
105+
106+// Status returns a consistent snapshot of manager state and counters.
107+func (m *Manager) Status() domain.TrafficStatus {
108+ m.mu.Lock()
109+ listeners := make(map[string]*entryPointRuntime, len(m.listeners))
110+ for name, runtime := range m.listeners {
111+ listeners[name] = runtime
112+ }
113+ status := domain.TrafficStatus{LastReloadStatus: m.lastReloadStatus, LastReloadError: m.lastReloadError}
114+ graph := m.snapshot.Load()
115+ m.mu.Unlock()
116+
117+ if graph == nil {
118+ return status
119+ }
120+
121+ status.EntryPoints = entryPointStatuses(graph.EntryPoints, listeners)
122+ status.Routers = routerStatuses(graph.Routers, listeners)
123+ status.Services = serviceStatuses(graph.Services)
124+ status.Counters = aggregateCounters(status.EntryPoints)
125+ return status
126+}
127+
128+func (m *Manager) recordReloadError(err error) error {
129+ m.mu.Lock()
130+ m.lastReloadStatus = reloadStatusError
131+ m.lastReloadError = err.Error()
132+ m.mu.Unlock()
133+ return err
134+}
135+
136+func (m *Manager) prepareTCPListeners(ctx context.Context, graph *domain.TrafficGraph) (map[string]*entryPointRuntime, []*entryPointRuntime, error) {
137+ current := make(map[string]*entryPointRuntime, len(m.listeners))
138+ for name, runtime := range m.listeners {
139+ current[name] = runtime
140+ }
141+
142+ next := make(map[string]*entryPointRuntime, len(current))
143+ created := []*entryPointRuntime{}
144+ for _, entryPoint := range graph.EntryPoints {
145+ if entryPoint.Protocol != domain.EntryPointProtocolTCP {
146+ continue
147+ }
148+ if runtime := current[entryPoint.Name]; runtime != nil && runtime.matches(entryPoint) {
149+ next[entryPoint.Name] = runtime
150+ continue
151+ }
152+ runtime, err := m.bindTCPEntryPoint(ctx, entryPoint)
153+ if err != nil {
154+ for _, createdRuntime := range created {
155+ createdRuntime.stop(ctx, effectiveTCPOptions(graph.Options.TCP).DrainTimeout)
156+ }
157+ return nil, nil, err
158+ }
159+ next[entryPoint.Name] = runtime
160+ created = append(created, runtime)
161+ }
162+ return next, created, nil
163+}
164+
165+func (m *Manager) bindTCPEntryPoint(ctx context.Context, entryPoint domain.EntryPoint) (*entryPointRuntime, error) {
166+ listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", entryPoint.Address)
167+ if err != nil {
168+ return nil, fmt.Errorf("bind tcp entrypoint %q on %s: %w", entryPoint.Name, entryPoint.Address, err)
169+ }
170+ runtime := newEntryPointRuntime(m, entryPoint, listener)
171+ return runtime, nil
172+}
173+
174+func entryPointStatuses(entries []domain.EntryPoint, listeners map[string]*entryPointRuntime) []domain.EntryPointStatus {
175+ statuses := make([]domain.EntryPointStatus, 0, len(entries))
176+ for _, entry := range entries {
177+ status := domain.EntryPointStatus{Name: entry.Name, Address: entry.Address, Protocol: entry.Protocol}
178+ if runtime := listeners[entry.Name]; runtime != nil {
179+ counters := runtime.counters.snapshot()
180+ status.Active = !runtime.isClosed()
181+ status.ActiveTCPConnections = counters.ActiveTCPConnections
182+ status.TotalAccepted = counters.TotalAccepted
183+ status.TotalRefused = counters.TotalRefused
184+ status.TotalErrors = counters.TotalErrors
185+ status.BytesIn = counters.BytesIn
186+ status.BytesOut = counters.BytesOut
187+ }
188+ statuses = append(statuses, status)
189+ }
190+ sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
191+ return statuses
192+}
193+
194+func routerStatuses(routers []domain.TrafficRouter, listeners map[string]*entryPointRuntime) []domain.TrafficRouterStatus {
195+ statuses := make([]domain.TrafficRouterStatus, 0, len(routers))
196+ for _, router := range routers {
197+ status := domain.TrafficRouterStatus{Name: router.Name, EntryPoint: router.EntryPoint, Protocol: router.Protocol, Active: true}
198+ if runtime := listeners[router.EntryPoint]; runtime != nil {
199+ counters := runtime.counters.snapshot()
200+ status.ActiveTCPConnections = counters.ActiveTCPConnections
201+ status.TotalAccepted = counters.TotalAccepted
202+ status.TotalRefused = counters.TotalRefused
203+ status.TotalErrors = counters.TotalErrors
204+ status.BytesIn = counters.BytesIn
205+ status.BytesOut = counters.BytesOut
206+ }
207+ statuses = append(statuses, status)
208+ }
209+ sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
210+ return statuses
211+}
212+
213+func serviceStatuses(services []domain.TrafficService) []domain.TrafficServiceStatus {
214+ statuses := make([]domain.TrafficServiceStatus, 0, len(services))
215+ for _, service := range services {
216+ status := domain.TrafficServiceStatus{Name: service.Name, Active: true, Backends: make([]domain.TrafficBackendStatus, 0, len(service.Backends))}
217+ for _, backend := range service.Backends {
218+ status.Backends = append(status.Backends, domain.TrafficBackendStatus{Name: backend.Name, Active: true})
219+ }
220+ statuses = append(statuses, status)
221+ }
222+ sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
223+ return statuses
224+}
225+
226+func aggregateCounters(entries []domain.EntryPointStatus) domain.TrafficCounters {
227+ var counters domain.TrafficCounters
228+ for _, entry := range entries {
229+ counters.ActiveTCPConnections += entry.ActiveTCPConnections
230+ counters.ActiveUDPSessions += entry.ActiveUDPSessions
231+ counters.TotalAccepted += entry.TotalAccepted
232+ counters.TotalRefused += entry.TotalRefused
233+ counters.TotalErrors += entry.TotalErrors
234+ counters.BytesIn += entry.BytesIn
235+ counters.BytesOut += entry.BytesOut
236+ }
237+ return counters
238+}
239+
240+func cloneTrafficGraph(graph domain.TrafficGraph) domain.TrafficGraph {
241+ clone := graph
242+ clone.EntryPoints = append([]domain.EntryPoint{}, graph.EntryPoints...)
243+ clone.Routers = append([]domain.TrafficRouter{}, graph.Routers...)
244+ clone.Services = append([]domain.TrafficService{}, graph.Services...)
245+ for i := range clone.Services {
246+ clone.Services[i].Backends = append([]domain.TrafficBackend{}, graph.Services[i].Backends...)
247+ }
248+ return clone
249+}
250+
251+func snapshotTCPOptions(graph *domain.TrafficGraph) domain.TCPOptions {
252+ if graph == nil {
253+ return domain.TCPOptions{}
254+ }
255+ return graph.Options.TCP
256+}
257+
258+type trafficCounters struct {
259+ activeTCPConnections atomic.Int64
260+ totalAccepted atomic.Int64
261+ totalRefused atomic.Int64
262+ totalErrors atomic.Int64
263+ bytesIn atomic.Int64
264+ bytesOut atomic.Int64
265+}
266+
267+func (c *trafficCounters) snapshot() domain.TrafficCounters {
268+ return domain.TrafficCounters{
269+ ActiveTCPConnections: c.activeTCPConnections.Load(),
270+ TotalAccepted: c.totalAccepted.Load(),
271+ TotalRefused: c.totalRefused.Load(),
272+ TotalErrors: c.totalErrors.Load(),
273+ BytesIn: c.bytesIn.Load(),
274+ BytesOut: c.bytesOut.Load(),
275+ }
276+}
277+
278+func defaultTCPOptions() domain.TCPOptions {
279+ return domain.TCPOptions{
280+ DialTimeout: 10 * time.Second,
281+ IdleTimeout: 5 * time.Minute,
282+ DrainTimeout: 30 * time.Second,
283+ }
284+}
285+
286+func effectiveTCPOptions(options domain.TCPOptions) domain.TCPOptions {
287+ defaults := defaultTCPOptions()
288+ if options.DialTimeout == 0 {
289+ options.DialTimeout = defaults.DialTimeout
290+ }
291+ if options.IdleTimeout == 0 {
292+ options.IdleTimeout = defaults.IdleTimeout
293+ }
294+ if options.DrainTimeout == 0 {
295+ options.DrainTimeout = defaults.DrainTimeout
296+ }
297+ return options
298+}
A internal/adapters/in/traffic/tcp.go +319 -0
@@ -0,0 +1,319 @@
1+package traffic
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "io"
8+ "net"
9+ "strconv"
10+ "sync"
11+ "sync/atomic"
12+ "time"
13+
14+ "github.com/bnema/gordon/internal/domain"
15+)
16+
17+type entryPointRuntime struct {
18+ manager *Manager
19+ entryPoint domain.EntryPoint
20+ listener net.Listener
21+ counters trafficCounters
22+
23+ started atomic.Bool
24+ closed atomic.Bool
25+
26+ ctx context.Context
27+ cancel context.CancelFunc
28+ acceptDone chan struct{}
29+ acceptDoneOnce sync.Once
30+ activeWG sync.WaitGroup
31+
32+ mu sync.Mutex
33+ activeConns map[*trackedTCPConn]struct{}
34+}
35+
36+type trackedTCPConn struct {
37+ client net.Conn
38+ backend net.Conn
39+}
40+
41+func newEntryPointRuntime(manager *Manager, entryPoint domain.EntryPoint, listener net.Listener) *entryPointRuntime {
42+ ctx, cancel := context.WithCancel(context.Background())
43+ return &entryPointRuntime{
44+ manager: manager,
45+ entryPoint: entryPoint,
46+ listener: listener,
47+ ctx: ctx,
48+ cancel: cancel,
49+ acceptDone: make(chan struct{}),
50+ activeConns: map[*trackedTCPConn]struct{}{},
51+ }
52+}
53+
54+func (r *entryPointRuntime) start() {
55+ if !r.started.CompareAndSwap(false, true) {
56+ return
57+ }
58+ go r.acceptLoop()
59+}
60+
61+func (r *entryPointRuntime) acceptLoop() {
62+ defer r.closeAcceptDone()
63+ for {
64+ conn, err := r.listener.Accept()
65+ if err != nil {
66+ if r.isClosed() || errors.Is(err, net.ErrClosed) {
67+ return
68+ }
69+ r.counters.totalErrors.Add(1)
70+ if temporary, ok := err.(interface{ Temporary() bool }); ok && temporary.Temporary() {
71+ continue
72+ }
73+ return
74+ }
75+ r.activeWG.Add(1)
76+ go r.handleTCPConn(conn)
77+ }
78+}
79+
80+func (r *entryPointRuntime) handleTCPConn(client net.Conn) {
81+ defer r.activeWG.Done()
82+
83+ options := effectiveTCPOptions(snapshotTCPOptions(r.manager.snapshot.Load()))
84+ if !r.reserveConnection(options.MaxConnections) {
85+ r.counters.totalRefused.Add(1)
86+ _ = client.Close()
87+ return
88+ }
89+ defer r.releaseConnection()
90+
91+ router, backend, ok := r.resolveTCPBackend()
92+ if !ok {
93+ r.counters.totalRefused.Add(1)
94+ _ = client.Close()
95+ return
96+ }
97+
98+ dialCtx, cancel := context.WithTimeout(r.ctx, options.DialTimeout)
99+ defer cancel()
100+ backendConn, err := (&net.Dialer{}).DialContext(dialCtx, "tcp", net.JoinHostPort(backend.Host, strconv.Itoa(backend.Port)))
101+ if err != nil {
102+ r.counters.totalErrors.Add(1)
103+ _ = client.Close()
104+ return
105+ }
106+
107+ tracked := &trackedTCPConn{client: client, backend: backendConn}
108+ r.track(tracked)
109+ defer r.untrack(tracked)
110+
111+ r.counters.totalAccepted.Add(1)
112+ _ = router
113+ r.proxyTCP(client, backendConn, options.IdleTimeout)
114+}
115+
116+func (r *entryPointRuntime) resolveTCPBackend() (domain.TrafficRouter, domain.TrafficBackend, bool) {
117+ graph := r.manager.snapshot.Load()
118+ if graph == nil {
119+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
120+ }
121+ var router domain.TrafficRouter
122+ for _, candidate := range graph.Routers {
123+ if candidate.EntryPoint == r.entryPoint.Name && candidate.Protocol == domain.RouterProtocolTCP {
124+ if router.Name != "" {
125+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
126+ }
127+ router = candidate
128+ }
129+ }
130+ if router.Name == "" {
131+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
132+ }
133+ for _, service := range graph.Services {
134+ if service.Name != router.Service || len(service.Backends) != 1 {
135+ continue
136+ }
137+ backend := service.Backends[0]
138+ if backend.Protocol != domain.NetworkProtocolTCP {
139+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
140+ }
141+ return router, backend, true
142+ }
143+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
144+}
145+
146+func (r *entryPointRuntime) proxyTCP(client net.Conn, backend net.Conn, idleTimeout time.Duration) {
147+ defer client.Close()
148+ defer backend.Close()
149+
150+ var wg sync.WaitGroup
151+ wg.Add(2)
152+ go func() {
153+ defer wg.Done()
154+ n, err := copyWithIdleTimeout(backend, client, idleTimeout)
155+ r.counters.bytesIn.Add(n)
156+ if err != nil && !isExpectedCopyError(err) {
157+ r.counters.totalErrors.Add(1)
158+ }
159+ closeWrite(backend)
160+ }()
161+ go func() {
162+ defer wg.Done()
163+ n, err := copyWithIdleTimeout(client, backend, idleTimeout)
164+ r.counters.bytesOut.Add(n)
165+ if err != nil && !isExpectedCopyError(err) {
166+ r.counters.totalErrors.Add(1)
167+ }
168+ closeWrite(client)
169+ }()
170+ wg.Wait()
171+}
172+
173+func copyWithIdleTimeout(dst net.Conn, src net.Conn, idleTimeout time.Duration) (int64, error) {
174+ if idleTimeout <= 0 {
175+ return io.Copy(dst, src)
176+ }
177+ return io.Copy(dst, idleConn{Conn: src, idleTimeout: idleTimeout})
178+}
179+
180+type idleConn struct {
181+ net.Conn
182+ idleTimeout time.Duration
183+}
184+
185+func (c idleConn) Read(p []byte) (int, error) {
186+ if err := c.SetReadDeadline(time.Now().Add(c.idleTimeout)); err != nil {
187+ return 0, err
188+ }
189+ return c.Conn.Read(p)
190+}
191+
192+func closeWrite(conn net.Conn) {
193+ if writeCloser, ok := conn.(interface{ CloseWrite() error }); ok {
194+ _ = writeCloser.CloseWrite()
195+ }
196+}
197+
198+func isExpectedCopyError(err error) bool {
199+ if err == nil || errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
200+ return true
201+ }
202+ var netErr net.Error
203+ if errors.As(err, &netErr) && netErr.Timeout() {
204+ return true
205+ }
206+ return false
207+}
208+
209+func (r *entryPointRuntime) reserveConnection(maxConnections int) bool {
210+ active := r.counters.activeTCPConnections.Add(1)
211+ if maxConnections > 0 && active > int64(maxConnections) {
212+ r.counters.activeTCPConnections.Add(-1)
213+ return false
214+ }
215+ return true
216+}
217+
218+func (r *entryPointRuntime) releaseConnection() {
219+ r.counters.activeTCPConnections.Add(-1)
220+}
221+
222+func (r *entryPointRuntime) track(conn *trackedTCPConn) {
223+ r.mu.Lock()
224+ r.activeConns[conn] = struct{}{}
225+ r.mu.Unlock()
226+}
227+
228+func (r *entryPointRuntime) untrack(conn *trackedTCPConn) {
229+ r.mu.Lock()
230+ delete(r.activeConns, conn)
231+ r.mu.Unlock()
232+}
233+
234+func (r *entryPointRuntime) stop(ctx context.Context, drainTimeout time.Duration) {
235+ if r.closed.CompareAndSwap(false, true) {
236+ r.cancel()
237+ _ = r.listener.Close()
238+ if !r.started.Load() {
239+ r.closeAcceptDone()
240+ }
241+ }
242+ select {
243+ case <-r.acceptDone:
244+ case <-ctx.Done():
245+ return
246+ }
247+ if r.waitActive(ctx, drainTimeout) {
248+ return
249+ }
250+ r.closeActiveConns()
251+ select {
252+ case <-r.activeDone():
253+ case <-ctx.Done():
254+ }
255+}
256+
257+func (r *entryPointRuntime) waitActive(ctx context.Context, drainTimeout time.Duration) bool {
258+ if drainTimeout <= 0 {
259+ drainTimeout = defaultTCPOptions().DrainTimeout
260+ }
261+ drainCtx, cancel := context.WithTimeout(ctx, drainTimeout)
262+ defer cancel()
263+ select {
264+ case <-r.activeDone():
265+ return true
266+ case <-drainCtx.Done():
267+ return false
268+ }
269+}
270+
271+func (r *entryPointRuntime) closeAcceptDone() {
272+ r.acceptDoneOnce.Do(func() { close(r.acceptDone) })
273+}
274+
275+func (r *entryPointRuntime) activeDone() <-chan struct{} {
276+ done := make(chan struct{})
277+ go func() {
278+ r.activeWG.Wait()
279+ close(done)
280+ }()
281+ return done
282+}
283+
284+func (r *entryPointRuntime) closeActiveConns() {
285+ r.mu.Lock()
286+ defer r.mu.Unlock()
287+ for conn := range r.activeConns {
288+ _ = conn.client.Close()
289+ _ = conn.backend.Close()
290+ }
291+}
292+
293+func (r *entryPointRuntime) matches(entryPoint domain.EntryPoint) bool {
294+ return r.entryPoint.Name == entryPoint.Name && r.entryPoint.Address == entryPoint.Address && r.entryPoint.Protocol == entryPoint.Protocol
295+}
296+
297+func (r *entryPointRuntime) shouldStopWith(next *entryPointRuntime) bool {
298+ return next == nil || next != r
299+}
300+
301+func (r *entryPointRuntime) isClosed() bool {
302+ return r.closed.Load()
303+}
304+
305+func backendFromAddress(name string, address string) (domain.TrafficBackend, error) {
306+ host, portValue, err := net.SplitHostPort(address)
307+ if err != nil {
308+ return domain.TrafficBackend{}, err
309+ }
310+ port, err := strconv.Atoi(portValue)
311+ if err != nil {
312+ return domain.TrafficBackend{}, err
313+ }
314+ return domain.TrafficBackend{Name: name, Host: host, Port: port, Protocol: domain.NetworkProtocolTCP}, nil
315+}
316+
317+func serviceRef(name string, portName string) string {
318+ return fmt.Sprintf("network_service:%s:%s", name, portName)
319+}
@@ -0,0 +1,319 @@
1+package traffic
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "io"
8+ "net"
9+ "strconv"
10+ "sync"
11+ "sync/atomic"
12+ "time"
13+
14+ "github.com/bnema/gordon/internal/domain"
15+)
16+
17+type entryPointRuntime struct {
18+ manager *Manager
19+ entryPoint domain.EntryPoint
20+ listener net.Listener
21+ counters trafficCounters
22+
23+ started atomic.Bool
24+ closed atomic.Bool
25+
26+ ctx context.Context
27+ cancel context.CancelFunc
28+ acceptDone chan struct{}
29+ acceptDoneOnce sync.Once
30+ activeWG sync.WaitGroup
31+
32+ mu sync.Mutex
33+ activeConns map[*trackedTCPConn]struct{}
34+}
35+
36+type trackedTCPConn struct {
37+ client net.Conn
38+ backend net.Conn
39+}
40+
41+func newEntryPointRuntime(manager *Manager, entryPoint domain.EntryPoint, listener net.Listener) *entryPointRuntime {
42+ ctx, cancel := context.WithCancel(context.Background())
43+ return &entryPointRuntime{
44+ manager: manager,
45+ entryPoint: entryPoint,
46+ listener: listener,
47+ ctx: ctx,
48+ cancel: cancel,
49+ acceptDone: make(chan struct{}),
50+ activeConns: map[*trackedTCPConn]struct{}{},
51+ }
52+}
53+
54+func (r *entryPointRuntime) start() {
55+ if !r.started.CompareAndSwap(false, true) {
56+ return
57+ }
58+ go r.acceptLoop()
59+}
60+
61+func (r *entryPointRuntime) acceptLoop() {
62+ defer r.closeAcceptDone()
63+ for {
64+ conn, err := r.listener.Accept()
65+ if err != nil {
66+ if r.isClosed() || errors.Is(err, net.ErrClosed) {
67+ return
68+ }
69+ r.counters.totalErrors.Add(1)
70+ if temporary, ok := err.(interface{ Temporary() bool }); ok && temporary.Temporary() {
71+ continue
72+ }
73+ return
74+ }
75+ r.activeWG.Add(1)
76+ go r.handleTCPConn(conn)
77+ }
78+}
79+
80+func (r *entryPointRuntime) handleTCPConn(client net.Conn) {
81+ defer r.activeWG.Done()
82+
83+ options := effectiveTCPOptions(snapshotTCPOptions(r.manager.snapshot.Load()))
84+ if !r.reserveConnection(options.MaxConnections) {
85+ r.counters.totalRefused.Add(1)
86+ _ = client.Close()
87+ return
88+ }
89+ defer r.releaseConnection()
90+
91+ router, backend, ok := r.resolveTCPBackend()
92+ if !ok {
93+ r.counters.totalRefused.Add(1)
94+ _ = client.Close()
95+ return
96+ }
97+
98+ dialCtx, cancel := context.WithTimeout(r.ctx, options.DialTimeout)
99+ defer cancel()
100+ backendConn, err := (&net.Dialer{}).DialContext(dialCtx, "tcp", net.JoinHostPort(backend.Host, strconv.Itoa(backend.Port)))
101+ if err != nil {
102+ r.counters.totalErrors.Add(1)
103+ _ = client.Close()
104+ return
105+ }
106+
107+ tracked := &trackedTCPConn{client: client, backend: backendConn}
108+ r.track(tracked)
109+ defer r.untrack(tracked)
110+
111+ r.counters.totalAccepted.Add(1)
112+ _ = router
113+ r.proxyTCP(client, backendConn, options.IdleTimeout)
114+}
115+
116+func (r *entryPointRuntime) resolveTCPBackend() (domain.TrafficRouter, domain.TrafficBackend, bool) {
117+ graph := r.manager.snapshot.Load()
118+ if graph == nil {
119+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
120+ }
121+ var router domain.TrafficRouter
122+ for _, candidate := range graph.Routers {
123+ if candidate.EntryPoint == r.entryPoint.Name && candidate.Protocol == domain.RouterProtocolTCP {
124+ if router.Name != "" {
125+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
126+ }
127+ router = candidate
128+ }
129+ }
130+ if router.Name == "" {
131+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
132+ }
133+ for _, service := range graph.Services {
134+ if service.Name != router.Service || len(service.Backends) != 1 {
135+ continue
136+ }
137+ backend := service.Backends[0]
138+ if backend.Protocol != domain.NetworkProtocolTCP {
139+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
140+ }
141+ return router, backend, true
142+ }
143+ return domain.TrafficRouter{}, domain.TrafficBackend{}, false
144+}
145+
146+func (r *entryPointRuntime) proxyTCP(client net.Conn, backend net.Conn, idleTimeout time.Duration) {
147+ defer client.Close()
148+ defer backend.Close()
149+
150+ var wg sync.WaitGroup
151+ wg.Add(2)
152+ go func() {
153+ defer wg.Done()
154+ n, err := copyWithIdleTimeout(backend, client, idleTimeout)
155+ r.counters.bytesIn.Add(n)
156+ if err != nil && !isExpectedCopyError(err) {
157+ r.counters.totalErrors.Add(1)
158+ }
159+ closeWrite(backend)
160+ }()
161+ go func() {
162+ defer wg.Done()
163+ n, err := copyWithIdleTimeout(client, backend, idleTimeout)
164+ r.counters.bytesOut.Add(n)
165+ if err != nil && !isExpectedCopyError(err) {
166+ r.counters.totalErrors.Add(1)
167+ }
168+ closeWrite(client)
169+ }()
170+ wg.Wait()
171+}
172+
173+func copyWithIdleTimeout(dst net.Conn, src net.Conn, idleTimeout time.Duration) (int64, error) {
174+ if idleTimeout <= 0 {
175+ return io.Copy(dst, src)
176+ }
177+ return io.Copy(dst, idleConn{Conn: src, idleTimeout: idleTimeout})
178+}
179+
180+type idleConn struct {
181+ net.Conn
182+ idleTimeout time.Duration
183+}
184+
185+func (c idleConn) Read(p []byte) (int, error) {
186+ if err := c.SetReadDeadline(time.Now().Add(c.idleTimeout)); err != nil {
187+ return 0, err
188+ }
189+ return c.Conn.Read(p)
190+}
191+
192+func closeWrite(conn net.Conn) {
193+ if writeCloser, ok := conn.(interface{ CloseWrite() error }); ok {
194+ _ = writeCloser.CloseWrite()
195+ }
196+}
197+
198+func isExpectedCopyError(err error) bool {
199+ if err == nil || errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
200+ return true
201+ }
202+ var netErr net.Error
203+ if errors.As(err, &netErr) && netErr.Timeout() {
204+ return true
205+ }
206+ return false
207+}
208+
209+func (r *entryPointRuntime) reserveConnection(maxConnections int) bool {
210+ active := r.counters.activeTCPConnections.Add(1)
211+ if maxConnections > 0 && active > int64(maxConnections) {
212+ r.counters.activeTCPConnections.Add(-1)
213+ return false
214+ }
215+ return true
216+}
217+
218+func (r *entryPointRuntime) releaseConnection() {
219+ r.counters.activeTCPConnections.Add(-1)
220+}
221+
222+func (r *entryPointRuntime) track(conn *trackedTCPConn) {
223+ r.mu.Lock()
224+ r.activeConns[conn] = struct{}{}
225+ r.mu.Unlock()
226+}
227+
228+func (r *entryPointRuntime) untrack(conn *trackedTCPConn) {
229+ r.mu.Lock()
230+ delete(r.activeConns, conn)
231+ r.mu.Unlock()
232+}
233+
234+func (r *entryPointRuntime) stop(ctx context.Context, drainTimeout time.Duration) {
235+ if r.closed.CompareAndSwap(false, true) {
236+ r.cancel()
237+ _ = r.listener.Close()
238+ if !r.started.Load() {
239+ r.closeAcceptDone()
240+ }
241+ }
242+ select {
243+ case <-r.acceptDone:
244+ case <-ctx.Done():
245+ return
246+ }
247+ if r.waitActive(ctx, drainTimeout) {
248+ return
249+ }
250+ r.closeActiveConns()
251+ select {
252+ case <-r.activeDone():
253+ case <-ctx.Done():
254+ }
255+}
256+
257+func (r *entryPointRuntime) waitActive(ctx context.Context, drainTimeout time.Duration) bool {
258+ if drainTimeout <= 0 {
259+ drainTimeout = defaultTCPOptions().DrainTimeout
260+ }
261+ drainCtx, cancel := context.WithTimeout(ctx, drainTimeout)
262+ defer cancel()
263+ select {
264+ case <-r.activeDone():
265+ return true
266+ case <-drainCtx.Done():
267+ return false
268+ }
269+}
270+
271+func (r *entryPointRuntime) closeAcceptDone() {
272+ r.acceptDoneOnce.Do(func() { close(r.acceptDone) })
273+}
274+
275+func (r *entryPointRuntime) activeDone() <-chan struct{} {
276+ done := make(chan struct{})
277+ go func() {
278+ r.activeWG.Wait()
279+ close(done)
280+ }()
281+ return done
282+}
283+
284+func (r *entryPointRuntime) closeActiveConns() {
285+ r.mu.Lock()
286+ defer r.mu.Unlock()
287+ for conn := range r.activeConns {
288+ _ = conn.client.Close()
289+ _ = conn.backend.Close()
290+ }
291+}
292+
293+func (r *entryPointRuntime) matches(entryPoint domain.EntryPoint) bool {
294+ return r.entryPoint.Name == entryPoint.Name && r.entryPoint.Address == entryPoint.Address && r.entryPoint.Protocol == entryPoint.Protocol
295+}
296+
297+func (r *entryPointRuntime) shouldStopWith(next *entryPointRuntime) bool {
298+ return next == nil || next != r
299+}
300+
301+func (r *entryPointRuntime) isClosed() bool {
302+ return r.closed.Load()
303+}
304+
305+func backendFromAddress(name string, address string) (domain.TrafficBackend, error) {
306+ host, portValue, err := net.SplitHostPort(address)
307+ if err != nil {
308+ return domain.TrafficBackend{}, err
309+ }
310+ port, err := strconv.Atoi(portValue)
311+ if err != nil {
312+ return domain.TrafficBackend{}, err
313+ }
314+ return domain.TrafficBackend{Name: name, Host: host, Port: port, Protocol: domain.NetworkProtocolTCP}, nil
315+}
316+
317+func serviceRef(name string, portName string) string {
318+ return fmt.Sprintf("network_service:%s:%s", name, portName)
319+}
A internal/adapters/in/traffic/tcp_test.go +293 -0
@@ -0,0 +1,293 @@
1+package traffic
2+
3+import (
4+ "bufio"
5+ "context"
6+ "io"
7+ "net"
8+ "testing"
9+ "time"
10+
11+ "github.com/bnema/zerowrap"
12+ "github.com/stretchr/testify/assert"
13+ "github.com/stretchr/testify/require"
14+
15+ "github.com/bnema/gordon/internal/domain"
16+)
17+
18+func TestTCPManagerLifecycle(t *testing.T) {
19+ backend := startTCPEchoServer(t, 0)
20+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
21+ manager := NewManager(zerowrap.Default())
22+
23+ require.NoError(t, manager.Apply(context.Background(), &graph))
24+ defer shutdownManager(t, manager)
25+
26+ conn := dialTCP(t, graph.EntryPoints[0].Address)
27+ require.NoError(t, conn.Close())
28+
29+ status := manager.Status()
30+ require.Equal(t, reloadStatusOK, status.LastReloadStatus)
31+ require.Len(t, status.EntryPoints, 1)
32+ assert.True(t, status.EntryPoints[0].Active)
33+
34+ require.NoError(t, manager.Shutdown(context.Background()))
35+ _, err := net.DialTimeout("tcp", graph.EntryPoints[0].Address, 100*time.Millisecond)
36+ require.Error(t, err)
37+}
38+
39+func TestTCPManagerRejectsInvalidGraphWithoutReplacingSnapshot(t *testing.T) {
40+ backend := startTCPEchoServer(t, 0)
41+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
42+ manager := NewManager(zerowrap.Default())
43+ require.NoError(t, manager.Apply(context.Background(), &graph))
44+ defer shutdownManager(t, manager)
45+
46+ invalid := graph
47+ invalid.EntryPoints = append(invalid.EntryPoints, graph.EntryPoints[0])
48+ err := manager.Apply(context.Background(), &invalid)
49+ require.Error(t, err)
50+
51+ status := manager.Status()
52+ assert.Equal(t, reloadStatusError, status.LastReloadStatus)
53+ assert.Contains(t, status.LastReloadError, "duplicate entrypoint")
54+
55+ conn := dialTCP(t, graph.EntryPoints[0].Address)
56+ defer conn.Close()
57+ assertRoundTrip(t, conn, "still works")
58+}
59+
60+func TestTCPPassthroughEcho(t *testing.T) {
61+ backend := startTCPEchoServer(t, 0)
62+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
63+ manager := NewManager(zerowrap.Default())
64+ require.NoError(t, manager.Apply(context.Background(), &graph))
65+ defer shutdownManager(t, manager)
66+
67+ conn := dialTCP(t, graph.EntryPoints[0].Address)
68+ defer conn.Close()
69+ assertRoundTrip(t, conn, "hello tcp")
70+}
71+
72+func TestTCPPassthroughUnknownRouterCloses(t *testing.T) {
73+ graph := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
74+ graph.Routers = nil
75+ graph.Services = nil
76+ manager := NewManager(zerowrap.Default())
77+ require.NoError(t, manager.Apply(context.Background(), &graph))
78+ defer shutdownManager(t, manager)
79+
80+ conn := dialTCP(t, graph.EntryPoints[0].Address)
81+ defer conn.Close()
82+ _, err := conn.Write([]byte("hello"))
83+ require.NoError(t, err)
84+ _, err = bufio.NewReader(conn).ReadByte()
85+ require.Error(t, err)
86+ assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
87+}
88+
89+func TestTCPPassthroughBackendDialFailure(t *testing.T) {
90+ graph := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
91+ graph.Options.TCP.DialTimeout = 50 * time.Millisecond
92+ manager := NewManager(zerowrap.Default())
93+ require.NoError(t, manager.Apply(context.Background(), &graph))
94+ defer shutdownManager(t, manager)
95+
96+ conn := dialTCP(t, graph.EntryPoints[0].Address)
97+ defer conn.Close()
98+ _, err := conn.Write([]byte("hello"))
99+ require.NoError(t, err)
100+ _, err = bufio.NewReader(conn).ReadByte()
101+ require.Error(t, err)
102+ assert.Eventually(t, func() bool { return manager.Status().Counters.TotalErrors == 1 }, time.Second, 10*time.Millisecond)
103+}
104+
105+func TestTCPPassthroughShutdownStopsAccepting(t *testing.T) {
106+ backend := startTCPEchoServer(t, 0)
107+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
108+ manager := NewManager(zerowrap.Default())
109+ require.NoError(t, manager.Apply(context.Background(), &graph))
110+
111+ require.NoError(t, manager.Shutdown(context.Background()))
112+ _, err := net.DialTimeout("tcp", graph.EntryPoints[0].Address, 100*time.Millisecond)
113+ require.Error(t, err)
114+}
115+
116+func TestTCPPassthroughIdleTimeoutClosesConnection(t *testing.T) {
117+ backend := startTCPEchoServer(t, 0)
118+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
119+ graph.Options.TCP.IdleTimeout = 50 * time.Millisecond
120+ manager := NewManager(zerowrap.Default())
121+ require.NoError(t, manager.Apply(context.Background(), &graph))
122+ defer shutdownManager(t, manager)
123+
124+ conn := dialTCP(t, graph.EntryPoints[0].Address)
125+ defer conn.Close()
126+ require.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second)))
127+ _, err := bufio.NewReader(conn).ReadByte()
128+ require.Error(t, err)
129+}
130+
131+func TestTCPPassthroughMaxConnectionsRejectsOverflow(t *testing.T) {
132+ backend := startTCPEchoServer(t, 0)
133+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
134+ graph.Options.TCP.MaxConnections = 1
135+ manager := NewManager(zerowrap.Default())
136+ require.NoError(t, manager.Apply(context.Background(), &graph))
137+ defer shutdownManager(t, manager)
138+
139+ first := dialTCP(t, graph.EntryPoints[0].Address)
140+ defer first.Close()
141+ assertRoundTrip(t, first, "first")
142+ require.Eventually(t, func() bool { return manager.Status().Counters.ActiveTCPConnections == 1 }, time.Second, 10*time.Millisecond)
143+
144+ second := dialTCP(t, graph.EntryPoints[0].Address)
145+ defer second.Close()
146+ _, err := second.Write([]byte("second"))
147+ require.NoError(t, err)
148+ _, err = bufio.NewReader(second).ReadByte()
149+ require.Error(t, err)
150+ assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
151+}
152+
153+func TestTCPPassthroughDrainWaitsForActiveConnectionThenTimesOut(t *testing.T) {
154+ backend := startTCPEchoServer(t, 0)
155+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
156+ graph.Options.TCP.DrainTimeout = 50 * time.Millisecond
157+ manager := NewManager(zerowrap.Default())
158+ require.NoError(t, manager.Apply(context.Background(), &graph))
159+ defer shutdownManager(t, manager)
160+
161+ conn := dialTCP(t, graph.EntryPoints[0].Address)
162+ defer conn.Close()
163+ assertRoundTrip(t, conn, "held")
164+ require.Eventually(t, func() bool { return manager.Status().Counters.ActiveTCPConnections == 1 }, time.Second, 10*time.Millisecond)
165+
166+ empty := domain.TrafficGraph{Options: graph.Options}
167+ started := time.Now()
168+ require.NoError(t, manager.Apply(context.Background(), &empty))
169+ assert.GreaterOrEqual(t, time.Since(started), 40*time.Millisecond)
170+
171+ _, err := conn.Write([]byte("after drain"))
172+ if err == nil {
173+ _, err = bufio.NewReader(conn).ReadByte()
174+ }
175+ require.Error(t, err)
176+}
177+
178+func TestTCPPassthroughStatusCountersTrackAcceptedRefusedErrorsAndBytes(t *testing.T) {
179+ backend := startTCPEchoServer(t, 0)
180+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
181+ graph.Options.TCP.MaxConnections = 1
182+ manager := NewManager(zerowrap.Default())
183+ require.NoError(t, manager.Apply(context.Background(), &graph))
184+ defer shutdownManager(t, manager)
185+
186+ conn := dialTCP(t, graph.EntryPoints[0].Address)
187+ assertRoundTrip(t, conn, "bytes")
188+ second := dialTCP(t, graph.EntryPoints[0].Address)
189+ _, err := second.Write([]byte("refused"))
190+ require.NoError(t, err)
191+ _, _ = bufio.NewReader(second).ReadByte()
192+ _ = second.Close()
193+ _ = conn.Close()
194+
195+ assert.Eventually(t, func() bool {
196+ status := manager.Status()
197+ return status.Counters.TotalAccepted == 1 &&
198+ status.Counters.TotalRefused == 1 &&
199+ status.Counters.BytesIn >= int64(len("bytes\n")) &&
200+ status.Counters.BytesOut >= int64(len("bytes\n"))
201+ }, time.Second, 10*time.Millisecond)
202+
203+ bad := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
204+ bad.Options.TCP.DialTimeout = 50 * time.Millisecond
205+ require.NoError(t, manager.Apply(context.Background(), &bad))
206+ badConn := dialTCP(t, bad.EntryPoints[0].Address)
207+ _, err = badConn.Write([]byte("boom"))
208+ require.NoError(t, err)
209+ _, _ = bufio.NewReader(badConn).ReadByte()
210+ _ = badConn.Close()
211+ assert.Eventually(t, func() bool { return manager.Status().Counters.TotalErrors == 1 }, time.Second, 10*time.Millisecond)
212+}
213+
214+func tcpGraph(t *testing.T, listenAddress string, backendAddress string) domain.TrafficGraph {
215+ t.Helper()
216+ backend, err := backendFromAddress("echo:tcp", backendAddress)
217+ require.NoError(t, err)
218+ ref := serviceRef("echo", "tcp")
219+ graph := domain.TrafficGraph{
220+ Options: domain.TrafficOptions{TCP: domain.TCPOptions{DialTimeout: time.Second, IdleTimeout: time.Minute, DrainTimeout: time.Second}},
221+ EntryPoints: []domain.EntryPoint{{Name: "tcp", Address: listenAddress, Protocol: domain.EntryPointProtocolTCP}},
222+ Routers: []domain.TrafficRouter{{Name: "echo", EntryPoint: "tcp", Protocol: domain.RouterProtocolTCP, Service: ref}},
223+ Services: []domain.TrafficService{{Name: ref, Backends: []domain.TrafficBackend{backend}}},
224+ }
225+ require.NoError(t, graph.Validate())
226+ return graph
227+}
228+
229+func freeTCPAddress(t *testing.T) string {
230+ t.Helper()
231+ listener, err := net.Listen("tcp", "127.0.0.1:0")
232+ require.NoError(t, err)
233+ address := listener.Addr().String()
234+ require.NoError(t, listener.Close())
235+ return address
236+}
237+
238+func dialTCP(t *testing.T, address string) net.Conn {
239+ t.Helper()
240+ conn, err := net.DialTimeout("tcp", address, time.Second)
241+ require.NoError(t, err)
242+ return conn
243+}
244+
245+func assertRoundTrip(t *testing.T, conn net.Conn, message string) {
246+ t.Helper()
247+ _, err := conn.Write([]byte(message + "\n"))
248+ require.NoError(t, err)
249+ line, err := bufio.NewReader(conn).ReadString('\n')
250+ require.NoError(t, err)
251+ assert.Equal(t, message+"\n", line)
252+}
253+
254+func shutdownManager(t *testing.T, manager *Manager) {
255+ t.Helper()
256+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
257+ defer cancel()
258+ require.NoError(t, manager.Shutdown(ctx))
259+}
260+
261+type tcpEchoServer struct {
262+ address string
263+ close func()
264+}
265+
266+func startTCPEchoServer(t *testing.T, readDelay time.Duration) tcpEchoServer {
267+ t.Helper()
268+ listener, err := net.Listen("tcp", "127.0.0.1:0")
269+ require.NoError(t, err)
270+ done := make(chan struct{})
271+ go func() {
272+ defer close(done)
273+ for {
274+ conn, err := listener.Accept()
275+ if err != nil {
276+ return
277+ }
278+ go func() {
279+ defer conn.Close()
280+ if readDelay > 0 {
281+ time.Sleep(readDelay)
282+ }
283+ _, _ = io.Copy(conn, conn)
284+ }()
285+ }
286+ }()
287+ server := tcpEchoServer{address: listener.Addr().String(), close: func() {
288+ _ = listener.Close()
289+ <-done
290+ }}
291+ t.Cleanup(server.close)
292+ return server
293+}
@@ -0,0 +1,293 @@
1+package traffic
2+
3+import (
4+ "bufio"
5+ "context"
6+ "io"
7+ "net"
8+ "testing"
9+ "time"
10+
11+ "github.com/bnema/zerowrap"
12+ "github.com/stretchr/testify/assert"
13+ "github.com/stretchr/testify/require"
14+
15+ "github.com/bnema/gordon/internal/domain"
16+)
17+
18+func TestTCPManagerLifecycle(t *testing.T) {
19+ backend := startTCPEchoServer(t, 0)
20+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
21+ manager := NewManager(zerowrap.Default())
22+
23+ require.NoError(t, manager.Apply(context.Background(), &graph))
24+ defer shutdownManager(t, manager)
25+
26+ conn := dialTCP(t, graph.EntryPoints[0].Address)
27+ require.NoError(t, conn.Close())
28+
29+ status := manager.Status()
30+ require.Equal(t, reloadStatusOK, status.LastReloadStatus)
31+ require.Len(t, status.EntryPoints, 1)
32+ assert.True(t, status.EntryPoints[0].Active)
33+
34+ require.NoError(t, manager.Shutdown(context.Background()))
35+ _, err := net.DialTimeout("tcp", graph.EntryPoints[0].Address, 100*time.Millisecond)
36+ require.Error(t, err)
37+}
38+
39+func TestTCPManagerRejectsInvalidGraphWithoutReplacingSnapshot(t *testing.T) {
40+ backend := startTCPEchoServer(t, 0)
41+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
42+ manager := NewManager(zerowrap.Default())
43+ require.NoError(t, manager.Apply(context.Background(), &graph))
44+ defer shutdownManager(t, manager)
45+
46+ invalid := graph
47+ invalid.EntryPoints = append(invalid.EntryPoints, graph.EntryPoints[0])
48+ err := manager.Apply(context.Background(), &invalid)
49+ require.Error(t, err)
50+
51+ status := manager.Status()
52+ assert.Equal(t, reloadStatusError, status.LastReloadStatus)
53+ assert.Contains(t, status.LastReloadError, "duplicate entrypoint")
54+
55+ conn := dialTCP(t, graph.EntryPoints[0].Address)
56+ defer conn.Close()
57+ assertRoundTrip(t, conn, "still works")
58+}
59+
60+func TestTCPPassthroughEcho(t *testing.T) {
61+ backend := startTCPEchoServer(t, 0)
62+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
63+ manager := NewManager(zerowrap.Default())
64+ require.NoError(t, manager.Apply(context.Background(), &graph))
65+ defer shutdownManager(t, manager)
66+
67+ conn := dialTCP(t, graph.EntryPoints[0].Address)
68+ defer conn.Close()
69+ assertRoundTrip(t, conn, "hello tcp")
70+}
71+
72+func TestTCPPassthroughUnknownRouterCloses(t *testing.T) {
73+ graph := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
74+ graph.Routers = nil
75+ graph.Services = nil
76+ manager := NewManager(zerowrap.Default())
77+ require.NoError(t, manager.Apply(context.Background(), &graph))
78+ defer shutdownManager(t, manager)
79+
80+ conn := dialTCP(t, graph.EntryPoints[0].Address)
81+ defer conn.Close()
82+ _, err := conn.Write([]byte("hello"))
83+ require.NoError(t, err)
84+ _, err = bufio.NewReader(conn).ReadByte()
85+ require.Error(t, err)
86+ assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
87+}
88+
89+func TestTCPPassthroughBackendDialFailure(t *testing.T) {
90+ graph := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
91+ graph.Options.TCP.DialTimeout = 50 * time.Millisecond
92+ manager := NewManager(zerowrap.Default())
93+ require.NoError(t, manager.Apply(context.Background(), &graph))
94+ defer shutdownManager(t, manager)
95+
96+ conn := dialTCP(t, graph.EntryPoints[0].Address)
97+ defer conn.Close()
98+ _, err := conn.Write([]byte("hello"))
99+ require.NoError(t, err)
100+ _, err = bufio.NewReader(conn).ReadByte()
101+ require.Error(t, err)
102+ assert.Eventually(t, func() bool { return manager.Status().Counters.TotalErrors == 1 }, time.Second, 10*time.Millisecond)
103+}
104+
105+func TestTCPPassthroughShutdownStopsAccepting(t *testing.T) {
106+ backend := startTCPEchoServer(t, 0)
107+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
108+ manager := NewManager(zerowrap.Default())
109+ require.NoError(t, manager.Apply(context.Background(), &graph))
110+
111+ require.NoError(t, manager.Shutdown(context.Background()))
112+ _, err := net.DialTimeout("tcp", graph.EntryPoints[0].Address, 100*time.Millisecond)
113+ require.Error(t, err)
114+}
115+
116+func TestTCPPassthroughIdleTimeoutClosesConnection(t *testing.T) {
117+ backend := startTCPEchoServer(t, 0)
118+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
119+ graph.Options.TCP.IdleTimeout = 50 * time.Millisecond
120+ manager := NewManager(zerowrap.Default())
121+ require.NoError(t, manager.Apply(context.Background(), &graph))
122+ defer shutdownManager(t, manager)
123+
124+ conn := dialTCP(t, graph.EntryPoints[0].Address)
125+ defer conn.Close()
126+ require.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second)))
127+ _, err := bufio.NewReader(conn).ReadByte()
128+ require.Error(t, err)
129+}
130+
131+func TestTCPPassthroughMaxConnectionsRejectsOverflow(t *testing.T) {
132+ backend := startTCPEchoServer(t, 0)
133+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
134+ graph.Options.TCP.MaxConnections = 1
135+ manager := NewManager(zerowrap.Default())
136+ require.NoError(t, manager.Apply(context.Background(), &graph))
137+ defer shutdownManager(t, manager)
138+
139+ first := dialTCP(t, graph.EntryPoints[0].Address)
140+ defer first.Close()
141+ assertRoundTrip(t, first, "first")
142+ require.Eventually(t, func() bool { return manager.Status().Counters.ActiveTCPConnections == 1 }, time.Second, 10*time.Millisecond)
143+
144+ second := dialTCP(t, graph.EntryPoints[0].Address)
145+ defer second.Close()
146+ _, err := second.Write([]byte("second"))
147+ require.NoError(t, err)
148+ _, err = bufio.NewReader(second).ReadByte()
149+ require.Error(t, err)
150+ assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
151+}
152+
153+func TestTCPPassthroughDrainWaitsForActiveConnectionThenTimesOut(t *testing.T) {
154+ backend := startTCPEchoServer(t, 0)
155+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
156+ graph.Options.TCP.DrainTimeout = 50 * time.Millisecond
157+ manager := NewManager(zerowrap.Default())
158+ require.NoError(t, manager.Apply(context.Background(), &graph))
159+ defer shutdownManager(t, manager)
160+
161+ conn := dialTCP(t, graph.EntryPoints[0].Address)
162+ defer conn.Close()
163+ assertRoundTrip(t, conn, "held")
164+ require.Eventually(t, func() bool { return manager.Status().Counters.ActiveTCPConnections == 1 }, time.Second, 10*time.Millisecond)
165+
166+ empty := domain.TrafficGraph{Options: graph.Options}
167+ started := time.Now()
168+ require.NoError(t, manager.Apply(context.Background(), &empty))
169+ assert.GreaterOrEqual(t, time.Since(started), 40*time.Millisecond)
170+
171+ _, err := conn.Write([]byte("after drain"))
172+ if err == nil {
173+ _, err = bufio.NewReader(conn).ReadByte()
174+ }
175+ require.Error(t, err)
176+}
177+
178+func TestTCPPassthroughStatusCountersTrackAcceptedRefusedErrorsAndBytes(t *testing.T) {
179+ backend := startTCPEchoServer(t, 0)
180+ graph := tcpGraph(t, freeTCPAddress(t), backend.address)
181+ graph.Options.TCP.MaxConnections = 1
182+ manager := NewManager(zerowrap.Default())
183+ require.NoError(t, manager.Apply(context.Background(), &graph))
184+ defer shutdownManager(t, manager)
185+
186+ conn := dialTCP(t, graph.EntryPoints[0].Address)
187+ assertRoundTrip(t, conn, "bytes")
188+ second := dialTCP(t, graph.EntryPoints[0].Address)
189+ _, err := second.Write([]byte("refused"))
190+ require.NoError(t, err)
191+ _, _ = bufio.NewReader(second).ReadByte()
192+ _ = second.Close()
193+ _ = conn.Close()
194+
195+ assert.Eventually(t, func() bool {
196+ status := manager.Status()
197+ return status.Counters.TotalAccepted == 1 &&
198+ status.Counters.TotalRefused == 1 &&
199+ status.Counters.BytesIn >= int64(len("bytes\n")) &&
200+ status.Counters.BytesOut >= int64(len("bytes\n"))
201+ }, time.Second, 10*time.Millisecond)
202+
203+ bad := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
204+ bad.Options.TCP.DialTimeout = 50 * time.Millisecond
205+ require.NoError(t, manager.Apply(context.Background(), &bad))
206+ badConn := dialTCP(t, bad.EntryPoints[0].Address)
207+ _, err = badConn.Write([]byte("boom"))
208+ require.NoError(t, err)
209+ _, _ = bufio.NewReader(badConn).ReadByte()
210+ _ = badConn.Close()
211+ assert.Eventually(t, func() bool { return manager.Status().Counters.TotalErrors == 1 }, time.Second, 10*time.Millisecond)
212+}
213+
214+func tcpGraph(t *testing.T, listenAddress string, backendAddress string) domain.TrafficGraph {
215+ t.Helper()
216+ backend, err := backendFromAddress("echo:tcp", backendAddress)
217+ require.NoError(t, err)
218+ ref := serviceRef("echo", "tcp")
219+ graph := domain.TrafficGraph{
220+ Options: domain.TrafficOptions{TCP: domain.TCPOptions{DialTimeout: time.Second, IdleTimeout: time.Minute, DrainTimeout: time.Second}},
221+ EntryPoints: []domain.EntryPoint{{Name: "tcp", Address: listenAddress, Protocol: domain.EntryPointProtocolTCP}},
222+ Routers: []domain.TrafficRouter{{Name: "echo", EntryPoint: "tcp", Protocol: domain.RouterProtocolTCP, Service: ref}},
223+ Services: []domain.TrafficService{{Name: ref, Backends: []domain.TrafficBackend{backend}}},
224+ }
225+ require.NoError(t, graph.Validate())
226+ return graph
227+}
228+
229+func freeTCPAddress(t *testing.T) string {
230+ t.Helper()
231+ listener, err := net.Listen("tcp", "127.0.0.1:0")
232+ require.NoError(t, err)
233+ address := listener.Addr().String()
234+ require.NoError(t, listener.Close())
235+ return address
236+}
237+
238+func dialTCP(t *testing.T, address string) net.Conn {
239+ t.Helper()
240+ conn, err := net.DialTimeout("tcp", address, time.Second)
241+ require.NoError(t, err)
242+ return conn
243+}
244+
245+func assertRoundTrip(t *testing.T, conn net.Conn, message string) {
246+ t.Helper()
247+ _, err := conn.Write([]byte(message + "\n"))
248+ require.NoError(t, err)
249+ line, err := bufio.NewReader(conn).ReadString('\n')
250+ require.NoError(t, err)
251+ assert.Equal(t, message+"\n", line)
252+}
253+
254+func shutdownManager(t *testing.T, manager *Manager) {
255+ t.Helper()
256+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
257+ defer cancel()
258+ require.NoError(t, manager.Shutdown(ctx))
259+}
260+
261+type tcpEchoServer struct {
262+ address string
263+ close func()
264+}
265+
266+func startTCPEchoServer(t *testing.T, readDelay time.Duration) tcpEchoServer {
267+ t.Helper()
268+ listener, err := net.Listen("tcp", "127.0.0.1:0")
269+ require.NoError(t, err)
270+ done := make(chan struct{})
271+ go func() {
272+ defer close(done)
273+ for {
274+ conn, err := listener.Accept()
275+ if err != nil {
276+ return
277+ }
278+ go func() {
279+ defer conn.Close()
280+ if readDelay > 0 {
281+ time.Sleep(readDelay)
282+ }
283+ _, _ = io.Copy(conn, conn)
284+ }()
285+ }
286+ }()
287+ server := tcpEchoServer{address: listener.Addr().String(), close: func() {
288+ _ = listener.Close()
289+ <-done
290+ }}
291+ t.Cleanup(server.close)
292+ return server
293+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 7

Tags 106

↑↓ moveenter openesc close