Connection Sweeper CLI Command - #2530
Conversation
Cross-references the router's TCP adaptor connections (skmanage QUERY) with kernel socket state (ss -tin) to get an idle-time signal more accurate than the router's own lastDlvSeconds, which only reflects delivery-object creation and not ongoing byte traffic on a long-lived stream. Includes gatherdump, a standalone tool for inspecting a Snapshot against a live router.
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (4)
🚧 Files skipped from review as they are similar to previous changes (4)
📝 WalkthroughWalkthroughAdds a hidden ChangesConnection sweeper pipeline
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Operator
participant SweepCommand
participant Sweeper
participant Router
participant Kernel
Operator->>SweepCommand: Run hidden sweep command
SweepCommand->>Sweeper: Pass platform-specific configuration
Sweeper->>Router: Query TCP adaptor connections
Router-->>Sweeper: Connection JSON
Sweeper->>Kernel: Gather socket activity
Kernel-->>Sweeper: Socket state
Sweeper->>Sweeper: Evaluate idle connections
Sweeper->>Router: Delete selected connections
Router-->>Sweeper: Deletion status
Sweeper-->>Operator: Return sweep result
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (2)
internal/cmd/skupper/debug/sweeper/gather.go (1)
96-104: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider surfacing the
sserror for debuggability.
gatherSocketsdiscards the error fromexecFnentirely. While the comment documents the intended "noss→ empty maps" behavior, a debug tool's value is diagnosis — silently swallowing the error makes it impossible to distinguish "ssnot installed" from "ssran but found no matching sockets." Consider capturing the error in theSnapshot(e.g., anSocketsError stringfield) or returning it fromgatherSocketssoGathercan log or include it.♻️ Optional: add error field to Snapshot
type Snapshot struct { Now time.Time TCPConns []connInfo // Sockets is keyed by peer "host:port", matching an 'in' connection's // Host (each client has a unique peer address). Sockets map[string]socketInfo // SocketsByLocal is keyed by the socket's own local "host:port", // matching an 'out' connection's LocalSocket ('out' peers all share the // backend's address, so only the local side is unique). SocketsByLocal map[string]socketInfo + // SocketsError is non-empty when the kernel socket query failed (e.g., + // `ss` is not installed), so the caller can distinguish a missing tool + // from an empty result. + SocketsError string }func gatherSockets(execFn Execer) (byPeer, byLocal map[string]socketInfo, errMsg string) { out, err := execFn([]string{"ss", "-tin"}) if err != nil { - return map[string]socketInfo{}, map[string]socketInfo{} + return map[string]socketInfo{}, map[string]socketInfo{}, err.Error() } - return socketsFromSS(out) + byPeer, byLocal = socketsFromSS(out) + return byPeer, byLocal, "" }internal/cmd/skupper/debug/sweeper/gatherdump/main.go (1)
30-38: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider sorting map keys for deterministic output.
Go map iteration order is randomized, so the "sockets by peer" and "sockets by local" sections will print in a different order on each run. For a debug tool where you may diff outputs across runs, sorting by key would make comparison easier.
♻️ Proposed refactor: sort keys before printing
import ( "flag" "fmt" "os" + "sort" "github.com/skupperproject/skupper/internal/cmd/skupper/debug/sweeper" )fmt.Println("\nsockets by peer (host:port -> lastrcv/lastsnd s):") - for host, s := range snap.Sockets { + peers := make([]string, 0, len(snap.Sockets)) + for host := range snap.Sockets { + peers = append(peers, host) + } + sort.Strings(peers) + for _, host := range peers { + s := snap.Sockets[host] fmt.Printf(" %-25s lastrcv=%.1fs lastsnd=%.1fs\n", host, float64(s.LastRcvMs)/1000, float64(s.LastSndMs)/1000) } fmt.Println("\nsockets by local (host:port -> lastrcv/lastsnd s):") - for host, s := range snap.SocketsByLocal { + locals := make([]string, 0, len(snap.SocketsByLocal)) + for host := range snap.SocketsByLocal { + locals = append(locals, host) + } + sort.Strings(locals) + for _, host := range locals { + s := snap.SocketsByLocal[host] fmt.Printf(" %-25s lastrcv=%.1fs lastsnd=%.1fs\n", host, float64(s.LastRcvMs)/1000, float64(s.LastSndMs)/1000) }
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 4fdf3a3b-5472-4b31-877e-d125fb23a88b
📒 Files selected for processing (3)
internal/cmd/skupper/debug/sweeper/README.mdinternal/cmd/skupper/debug/sweeper/gather.gointernal/cmd/skupper/debug/sweeper/gatherdump/main.go
evanwang9x
changed the title
59b631a to
2c79b1a
Compare
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (4)
internal/cmd/skupper/debug/sweeper/criteria.go (1)
14-14: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueFix typographical error.
The word "criterias" is grammatically incorrect; the plural of "criterion" is "criteria".
♻️ Proposed fix
-// Evaluate applies all kill criterias (currently only based off of time) to all +// Evaluate applies all kill criteria (currently only based off of time) to allSource: Pipeline failures
internal/cmd/skupper/debug/sweeper/sweeper.go (1)
71-74: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAvoid concatenating format strings.
Concatenating raw variables into a format string can enable log injection and violates standard
go vetrules forPrintfformatting. Pass the timestamp as a separate format argument instead.♻️ Proposed fix
func logf(format string, args ...any) { ts := time.Now().Format("15:04:05") - fmt.Printf("["+ts+"] "+format+"\n", args...) + fmt.Printf("[%s] %s\n", ts, fmt.Sprintf(format, args...)) }Source: Linters/SAST tools
internal/cmd/skupper/debug/kube/conn_sweeper.go (2)
125-129: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valuePropagate command context instead of using
context.TODO().If the cobra command provides a context (e.g.
cmd.CobraCmd.Context()), it is safer to pass that to the Kubernetes client instead ofcontext.TODO(). This ensures that if the user cancels the CLI command (e.g. via Ctrl+C), the pending network request is correctly aborted.♻️ Proposed fix
- pods, err := cmd.KubeClient.CoreV1().Pods(cmd.Namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: routerPodSelector}) + ctx := context.Background() + if cmd.CobraCmd != nil && cmd.CobraCmd.Context() != nil { + ctx = cmd.CobraCmd.Context() + } + pods, err := cmd.KubeClient.CoreV1().Pods(cmd.Namespace).List(ctx, metav1.ListOptions{LabelSelector: routerPodSelector})
50-66: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
cli.Restinstead of rebuilding kubeconfig.client.NewClientalready keeps a usable*rest.ConfigonKubeClient.Rest, and this method only needs that config forExecCommandInContainer; if you need the/api/serializer tweaks isolated, copycli.Restfirst and mutate the copy.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 1333e949-f1bd-4bb6-bd88-be3419c97fe2
📒 Files selected for processing (9)
internal/cmd/skupper/common/flags.gointernal/cmd/skupper/debug/debug.gointernal/cmd/skupper/debug/kube/conn_sweeper.gointernal/cmd/skupper/debug/nonkube/conn_sweeper.gointernal/cmd/skupper/debug/sweeper/criteria.gointernal/cmd/skupper/debug/sweeper/gather.gointernal/cmd/skupper/debug/sweeper/inetdiag.gointernal/cmd/skupper/debug/sweeper/kill.gointernal/cmd/skupper/debug/sweeper/sweeper.go
There was a problem hiding this comment.
🧹 Nitpick comments (1)
internal/cmd/skupper/debug/sweeper/sweeper.go (1)
71-74: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAvoid concatenating the format string.
While the current usages of
logfare safe because they only pass string literals, dynamically concatenating theformatargument with the timestamp defeats thego vettool's ability to statically verify format string arguments and triggers static analysis warnings.Consider formatting the message first to keep the layout string completely static.
♻️ Proposed refactor
func logf(format string, args ...any) { ts := time.Now().Format("15:04:05") - fmt.Printf("["+ts+"] "+format+"\n", args...) + fmt.Printf("[%s] %s\n", ts, fmt.Sprintf(format, args...)) }Source: Linters/SAST tools
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 185dd6c6-3a1e-4aa4-a1a5-f7a63c04a210
📒 Files selected for processing (9)
internal/cmd/skupper/common/flags.gointernal/cmd/skupper/debug/debug.gointernal/cmd/skupper/debug/kube/conn_sweeper.gointernal/cmd/skupper/debug/nonkube/conn_sweeper.gointernal/cmd/skupper/debug/sweeper/criteria.gointernal/cmd/skupper/debug/sweeper/gather.gointernal/cmd/skupper/debug/sweeper/inetdiag.gointernal/cmd/skupper/debug/sweeper/kill.gointernal/cmd/skupper/debug/sweeper/sweeper.go
🚧 Files skipped from review as they are similar to previous changes (7)
- internal/cmd/skupper/debug/debug.go
- internal/cmd/skupper/debug/sweeper/criteria.go
- internal/cmd/skupper/debug/kube/conn_sweeper.go
- internal/cmd/skupper/debug/nonkube/conn_sweeper.go
- internal/cmd/skupper/debug/sweeper/kill.go
- internal/cmd/skupper/debug/sweeper/gather.go
- internal/cmd/skupper/debug/sweeper/inetdiag.go
f19fd76 to
404c992
Compare
404c992 to
f529c33
Compare
|
Thank you for all your work! I left some comments regarding the kube implementation of the command. I suggest to fix also the same issues in the non-kube implementation. I also suggest to review all the generated comments: if they explain things in the current code that are not easy to understand, they should be kept; if not, I would suggest to delete them. Thanks! |
| type CommandDebugFlags struct { | ||
| } | ||
|
|
||
| type CommandConnSweeperFlags struct { |
There was a problem hiding this comment.
The URL and skmanage should not be flags for the sub-command
Suggest following the pattern that the dump subcommand uses for getting skstat as it accounts for the three scenarios
- k8s
- container (e.g.)
- systemd
We can assume skmanage is present in the router container and for systemd it can be assumed user already had to install the skupper-router binary
| // back to the python netlink script when ss isn't available (e.g. inside the | ||
| // router container, which ships python3 but not iproute). | ||
| func gatherSockets(execFn Execer) (byPeer, byLocal map[string]socketInfo) { | ||
| if out, err := execFn([]string{"ss", "-tin"}); err == nil { |
There was a problem hiding this comment.
iproute is lightweight and has some other useful utilities...we should look to add to the router container
There was a problem hiding this comment.
@ajssmith do i remember us somewhat recently stripping out non essential system dependencies to minimize churn from security scanners? Could be worth raising with the group to decide where our priorities are there.
EDIT: I do see that we've got a python script embedded to work around that - not ideal. I think I'd prefer anything that exercises router system dependencies (python or iproute included) live with the router codebase, get tested there, and bear some obvious sk-part-of-our-public-api name so that we don't accidentally break this debug option down the line.
| URL string | ||
| Skmanage string | ||
| IdleThresholdSecs int | ||
| DryRun bool |
There was a problem hiding this comment.
Should we use dryRun, which is false by default, causing the sweeper command to kill connections on execution.
Or change it with something like: execute bool, to prevent accidentally killling idle connections?
There was a problem hiding this comment.
I think execute bool is more intuitive than dryrun so I'll change it to be that.
There was a problem hiding this comment.
🧹 Nitpick comments (2)
internal/cmd/skupper/debug/nonkube/conn_sweeper.go (2)
44-52: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valueReject positional arguments to prevent user confusion.
Since
skupper debug sweepdoes not require any positional arguments, consider explicitly rejecting them inValidateInput. Otherwise, any extra arguments passed by mistake are silently ignored.💡 Proposed fix
func (cmd *CmdConnSweeper) ValidateInput(args []string) error { var validationErrors []error + if len(args) > 0 { + validationErrors = append(validationErrors, fmt.Errorf("this command does not accept positional arguments")) + } numberValidator := validator.NewNumberValidator() numberValidator.IncludeZero = false
59-71: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCapture and return initialization errors instead of swallowing them.
When
platformLoader.Loadorruntime.GetLocalRouterAddressfails, the error is swallowed and the function returns early. This causes the command to fail later inRun()with a generic "could not determine router management address" error, masking the root cause (e.g., missing configuration, permission denied).Consider adding an
initError errorfield to theCmdConnSweeperstruct to capture these errors inInputToOptionsand bubble them up inRun().♻️ Proposed refactoring
- Add the field to the struct:
type CmdConnSweeper struct { CobraCmd *cobra.Command // ... exec sweeper.Execer + initError error }
- Capture the errors in
InputToOptions:platformLoader := &nonkubecommon.NamespacePlatformLoader{} platform, err := platformLoader.Load(cmd.namespace) if err != nil { + cmd.initError = fmt.Errorf("failed to load platform: %w", err) return } cmd.platform = platform url, err := runtime.GetLocalRouterAddress(cmd.namespace) if err != nil { + cmd.initError = fmt.Errorf("failed to get local router address: %w", err) return }
- Return the error in
Run():func (cmd *CmdConnSweeper) Run() error { + if cmd.initError != nil { + return cmd.initError + } if cmd.url == "" || cmd.skmanage == "" {
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 857447a8-d6df-47d8-92ea-bf3ed20ffb6d
📒 Files selected for processing (8)
internal/cmd/skupper/common/flags.gointernal/cmd/skupper/debug/debug.gointernal/cmd/skupper/debug/kube/conn_sweeper.gointernal/cmd/skupper/debug/nonkube/conn_sweeper.gointernal/cmd/skupper/debug/sweeper/gather.gointernal/cmd/skupper/debug/sweeper/inetdiag.gointernal/cmd/skupper/debug/sweeper/kill.gointernal/cmd/skupper/debug/sweeper/sweeper.go
💤 Files with no reviewable changes (1)
- internal/cmd/skupper/common/flags.go
🚧 Files skipped from review as they are similar to previous changes (5)
- internal/cmd/skupper/debug/sweeper/kill.go
- internal/cmd/skupper/debug/sweeper/sweeper.go
- internal/cmd/skupper/debug/sweeper/inetdiag.go
- internal/cmd/skupper/debug/kube/conn_sweeper.go
- internal/cmd/skupper/debug/sweeper/gather.go
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
internal/cmd/skupper/debug/sweeper/sweeper.go (2)
47-56: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winValidate the idle threshold before enabling cleanup.
IdleThresholdSecsaccepts unrestricted integer input, so negative values—or values that overflowtime.Durationduring the seconds conversion—can make the evaluator classify unintended connections as idle. Reject invalid thresholds before the--executepath can delete connections.
64-67: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPropagate connection-deletion failures to the CLI.
killAllreports failures throughResult.Failed, but the Kubernetes and non-Kubernetes callers only propagateerror. As a result, a sweep can fail to close connections while still exiting successfully. Return an error when deletions fail, or update both callers to fail based onResult.Failed.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 2a0974cc-c7a5-4c3b-8a33-eb5be94d5a8a
📒 Files selected for processing (5)
internal/cmd/skupper/common/flags.gointernal/cmd/skupper/debug/debug.gointernal/cmd/skupper/debug/kube/conn_sweeper.gointernal/cmd/skupper/debug/nonkube/conn_sweeper.gointernal/cmd/skupper/debug/sweeper/sweeper.go
🚧 Files skipped from review as they are similar to previous changes (3)
- internal/cmd/skupper/debug/debug.go
- internal/cmd/skupper/debug/kube/conn_sweeper.go
- internal/cmd/skupper/debug/nonkube/conn_sweeper.go
|
@coderabbitai full review |
✅ Action performedFull review finished. |
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (2)
internal/cmd/skupper/debug/kube/conn_sweeper.go (2)
1-1: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate
ValidateInputlogic across kube and nonkube sweeper commands. Both implementations validateIdleThresholdidentically (non-zero number check + max-threshold check +errors.Join), which risks drift if the rule changes in only one place.
internal/cmd/skupper/debug/kube/conn_sweeper.go#L80-91: extract this validation into a shared helper (e.g.sweeper.ValidateIdleThreshold(int) error) and call it here.internal/cmd/skupper/debug/nonkube/conn_sweeper.go#L44-55: call the same shared helper instead of duplicating the validator/threshold logic.
45-78: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winReuse the existing REST config for exec.
client.NewClient(...)resolves the kubeconfig once and stores a copy inkubeClient.Rest, but it does not expose that config.NewClientduplicates the sameclientcmdloading rules/exec config patching, which can drift if the paths diverge. Add an exported accessor for the already-resolved*rest.Configand copy/patch it forcmd.Restinstead of reloading it.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: d211da5a-2394-4591-9b0f-0bb168eaff63
📒 Files selected for processing (9)
internal/cmd/skupper/common/flags.gointernal/cmd/skupper/debug/debug.gointernal/cmd/skupper/debug/kube/conn_sweeper.gointernal/cmd/skupper/debug/nonkube/conn_sweeper.gointernal/cmd/skupper/debug/sweeper/criteria.gointernal/cmd/skupper/debug/sweeper/gather.gointernal/cmd/skupper/debug/sweeper/inetdiag.gointernal/cmd/skupper/debug/sweeper/kill.gointernal/cmd/skupper/debug/sweeper/sweeper.go
| func (cmd *CmdConnSweeper) podExecer(podName string) sweeper.Execer { | ||
| return func(argv []string) ([]byte, error) { | ||
| type execResult struct { | ||
| out []byte | ||
| err error | ||
| } | ||
| done := make(chan execResult, 1) | ||
| go func() { | ||
| out, err := client.ExecCommandInContainer(argv, podName, routerContainer, cmd.Namespace, cmd.KubeClient, cmd.Rest) | ||
| if err != nil { | ||
| done <- execResult{nil, err} | ||
| return | ||
| } | ||
| done <- execResult{out.Bytes(), nil} | ||
| }() | ||
|
|
||
| select { | ||
| case r := <-done: | ||
| if r.err != nil { | ||
| return nil, fmt.Errorf("exec %q in pod %s failed: %w", argv[0], podName, r.err) | ||
| } | ||
| return r.out, nil | ||
| case <-time.After(podExecTimeout): | ||
| return nil, fmt.Errorf("exec %q in pod %s timed out after %s", argv[0], podName, podExecTimeout) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether ExecCommandInContainer accepts a context for cancellation
rg -n -A 10 'func ExecCommandInContainer' internal/kube/clientRepository: skupperproject/skupper
Length of output: 1056
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- conn_sweeper relevant section ---\n'
sed -n '130,210p' internal/cmd/skupper/debug/kube/conn_sweeper.go
printf '\n--- kube client exec util ---\n'
sed -n '1,120p' internal/kube/client/exec_util.go
printf '\n--- all ExecCommandInContainer usages ---\n'
rg -n 'ExecCommandInContainer\(' internal/cmd internal/kubeRepository: skupperproject/skupper
Length of output: 251
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- conn_sweeper relevant section ---'
sed -n '130,210p' internal/cmd/skupper/debug/kube/conn_sweeper.go
printf '%s\n' '--- kube client exec util ---'
sed -n '1,140p' internal/kube/client/exec_util.go
printf '%s\n' '--- all ExecCommandInContainer usages ---'
rg -n 'ExecCommandInContainer\(' internal/cmd internal/kubeRepository: skupperproject/skupper
Length of output: 4365
Cancel pending pod execs on timeout.
ExecCommandInContainer does not accept a cancellation context, so when time.After returns the goroutine may still be blocked on exec.Stream(...) and only later write to the buffered channel. Change ExecCommandInContainer to accept a context and use it on the Pod(...).Get(...) / RESTClientFor(...) path(s), then cancel it from the timeout branch before returning.
| // | ||
| // <local ip:port> <peer ip:port> <lastrcv ms> <lastsnd ms> | ||
| // | ||
| // TODO: IPv4 only; extend for IPv6. |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Support IPv6 in the fallback.
Line 16 limits the only router-container fallback to AF_INET; IPv6 TCP connections are silently unmatched whenever ss is unavailable, so they can never be reported or swept. Query AF_INET6 too and normalize its address format to the router connection keys.
5 participants
Added a CLI command skupper debug sweep to scan through TCP connections and see how long they've been idle and if it exceeds a user set time kill it
Summary by CodeRabbit
skupper debug sweepcommand to identify idle/orphaned router TCP-adaptor connections and optionally delete them.--idle-threshold(default 4 hours) and--execute(default off) to switch between reporting and deletion.