Connection Sweeper CLI Command #2530
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,179 @@ | ||
| package kube | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/skupperproject/skupper/internal/cmd/skupper/common" | ||
| "github.com/skupperproject/skupper/internal/cmd/skupper/debug/sweeper" | ||
| "github.com/skupperproject/skupper/internal/kube/client" | ||
| "github.com/spf13/cobra" | ||
| corev1 "k8s.io/api/core/v1" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| "k8s.io/client-go/kubernetes" | ||
| "k8s.io/client-go/kubernetes/scheme" | ||
| restclient "k8s.io/client-go/rest" | ||
| "k8s.io/client-go/tools/clientcmd" | ||
| ) | ||
|
|
||
| const ( | ||
| routerPodSelector = "app.kubernetes.io/name=skupper-router" | ||
| routerContainer = "router" | ||
| podExecTimeout = 30 * time.Second | ||
| ) | ||
|
|
||
| // CmdConnSweeper is the kubernetes entry point for `skupper debug sweep`. It | ||
| // finds every ready router pod and runs the sweeper against each one, with | ||
| // all commands exec'd inside the router container so they see the pod's own | ||
| // network namespace (no port-forward needed). | ||
| type CmdConnSweeper struct { | ||
| CobraCmd *cobra.Command | ||
| Flags *common.CommandConnSweeperFlags | ||
| KubeClient kubernetes.Interface | ||
| Rest *restclient.Config | ||
| Namespace string | ||
| } | ||
|
|
||
| func NewCmdConnSweeper() *CmdConnSweeper { | ||
| return &CmdConnSweeper{} | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) NewClient(cobraCommand *cobra.Command, args []string) { | ||
| cli, err := client.NewClient(cobraCommand.Flag("namespace").Value.String(), cobraCommand.Flag("context").Value.String(), cobraCommand.Flag("kubeconfig").Value.String()) | ||
| if err != nil { | ||
| return | ||
| } | ||
| cmd.KubeClient = cli.GetKubeClient() | ||
| cmd.Namespace = cli.Namespace | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() | ||
| if kubeConfigPath := cobraCommand.Flag("kubeconfig").Value.String(); kubeConfigPath != "" { | ||
| loadingRules = &clientcmd.ClientConfigLoadingRules{ExplicitPath: kubeConfigPath} | ||
| } | ||
| kubeconfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig( | ||
| loadingRules, | ||
| &clientcmd.ConfigOverrides{CurrentContext: cobraCommand.Flag("context").Value.String()}, | ||
| ) | ||
| restconfig, err := kubeconfig.ClientConfig() | ||
| if err != nil { | ||
| return | ||
| } | ||
| restconfig.APIPath = "/api" | ||
| restconfig.GroupVersion = &corev1.SchemeGroupVersion | ||
| restconfig.NegotiatedSerializer = scheme.Codecs.WithoutConversion() | ||
|
coderabbitai[bot] marked this conversation as resolved.
Outdated
|
||
| cmd.Rest = restconfig | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) ValidateInput(args []string) error { | ||
| if cmd.Flags.IdleThreshold <= 0 { | ||
|
evanwang9x marked this conversation as resolved.
Outdated
|
||
| return fmt.Errorf("--idle-threshold must be a positive number of seconds") | ||
| } | ||
| if cmd.KubeClient == nil || cmd.Rest == nil { | ||
| return fmt.Errorf("could not initialize kubernetes client") | ||
| } | ||
| return nil | ||
|
evanwang9x marked this conversation as resolved.
Outdated
evanwang9x marked this conversation as resolved.
Outdated
|
||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) InputToOptions() {} | ||
|
|
||
| func (cmd *CmdConnSweeper) Run() error { | ||
| podNames, err := cmd.findRouterPods() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // Each ready replica has its own connections, so sweep every pod. A | ||
| // failure on one pod (e.g. exec cut off mid-sweep) must not leave the | ||
| // remaining replicas unswept. | ||
| var total sweeper.Result | ||
| var failedPods []string | ||
| for _, podName := range podNames { | ||
| fmt.Printf("=== router pod %s (namespace %s) ===\n", podName, cmd.Namespace) | ||
| res, err := sweeper.Run(sweeper.Config{ | ||
| URL: cmd.Flags.URL, | ||
| Skmanage: cmd.Flags.Skmanage, | ||
| IdleThresholdSecs: cmd.Flags.IdleThreshold, | ||
| DryRun: cmd.Flags.DryRun, | ||
| Exec: cmd.podExecer(podName), | ||
| }) | ||
| if err != nil { | ||
| fmt.Printf("sweep of pod %s failed: %v\n", podName, err) | ||
| failedPods = append(failedPods, podName) | ||
| continue | ||
| } | ||
| total.Total += res.Total | ||
| total.Killed += res.Killed | ||
| total.Skipped += res.Skipped | ||
| total.Failed += res.Failed | ||
| } | ||
|
|
||
| if len(podNames) > 1 { | ||
| fmt.Printf("=== all pods: total:%d killed:%d skipped:%d failed:%d ===\n", | ||
| total.Total, total.Killed, total.Skipped, total.Failed) | ||
| } | ||
| if len(failedPods) > 0 { | ||
| return fmt.Errorf("sweep failed on %d of %d router pod(s): %v", len(failedPods), len(podNames), failedPods) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) WaitUntil() error { return nil } | ||
|
|
||
| // findRouterPods returns every router pod whose router container is ready, | ||
| // with HA there are multiple replicas and each holds its own connections. | ||
|
evanwang9x marked this conversation as resolved.
Outdated
|
||
| func (cmd *CmdConnSweeper) findRouterPods() ([]string, error) { | ||
| pods, err := cmd.KubeClient.CoreV1().Pods(cmd.Namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: routerPodSelector}) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("could not list router pods: %w", err) | ||
| } | ||
| var ready []string | ||
| for _, pod := range pods.Items { | ||
| // Phase stays "Running" even while a container crash-loops, so | ||
| // require the router container itself to be ready. | ||
| for _, cs := range pod.Status.ContainerStatuses { | ||
| if cs.Name == routerContainer && cs.Ready { | ||
| ready = append(ready, pod.Name) | ||
| } | ||
| } | ||
| } | ||
| if len(ready) == 0 { | ||
| return nil, fmt.Errorf("no ready skupper-router pod found in namespace %q", cmd.Namespace) | ||
| } | ||
| return ready, nil | ||
| } | ||
|
|
||
| // podExecer returns a sweeper.Execer that runs argv inside the router | ||
| // container, where skmanage, python3 and the router's own network namespace | ||
| // are all available. | ||
| func (cmd *CmdConnSweeper) podExecer(podName string) sweeper.Execer { | ||
| return func(argv []string) ([]byte, error) { | ||
| // ExecCommandInContainer uses the deprecated context-less Stream, so | ||
| // it can't be cancelled directly; run it in a goroutine and give up | ||
| // on the sweep's behalf after podExecTimeout. The buffered channel | ||
| // lets the goroutine finish its send if Stream ever returns. | ||
|
evanwang9x marked this conversation as resolved.
Outdated
|
||
| 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) | ||
| } | ||
| } | ||
| } | ||
|
Comment on lines
+167
to
+193
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.
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,128 @@ | ||
| package nonkube | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "os/exec" | ||
| "time" | ||
|
|
||
| "github.com/skupperproject/skupper/internal/cmd/skupper/common" | ||
| "github.com/skupperproject/skupper/internal/cmd/skupper/debug/sweeper" | ||
| "github.com/skupperproject/skupper/internal/nonkube/client/runtime" | ||
| nonkubecommon "github.com/skupperproject/skupper/internal/nonkube/common" | ||
| "github.com/spf13/cobra" | ||
| ) | ||
|
|
||
| // Cert paths inside the router container (see compat.SiteStateRenderer, which | ||
| // mounts the runtime certs at /etc/skupper-router/runtime/certs). | ||
| const containerCertsPath = "/etc/skupper-router/runtime/certs/skupper-local-client" | ||
|
|
||
| type CmdConnSweeper struct { | ||
| CobraCmd *cobra.Command | ||
| Flags *common.CommandConnSweeperFlags | ||
| namespace string | ||
| platform string | ||
| url string | ||
| sslArgs []string | ||
| exec sweeper.Execer | ||
| } | ||
|
|
||
| func NewCmdConnSweeper() *CmdConnSweeper { | ||
| return &CmdConnSweeper{} | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) NewClient(cobraCommand *cobra.Command, args []string) { | ||
| cmd.namespace = cobraCommand.Flag(common.FlagNameNamespace).Value.String() | ||
| } | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| func (cmd *CmdConnSweeper) ValidateInput(args []string) error { | ||
| if cmd.Flags.IdleThreshold <= 0 { | ||
| return fmt.Errorf("--idle-threshold must be a positive number of seconds") | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) InputToOptions() { | ||
| if cmd.namespace == "" { | ||
| cmd.namespace = "default" | ||
| } | ||
| cmd.url = cmd.Flags.URL | ||
|
|
||
| // Detect the platform from the namespace's site config. | ||
| platformLoader := &nonkubecommon.NamespacePlatformLoader{} | ||
| platform, err := platformLoader.Load(cmd.namespace) | ||
| if err != nil { | ||
| return | ||
| } | ||
| cmd.platform = platform | ||
|
|
||
| switch platform { | ||
|
|
||
| case "podman", "docker": | ||
| // exec inside it so skmanage and the socket | ||
| // query see the router's network namespace. | ||
| containerName := cmd.namespace + "-skupper-router" | ||
| cmd.exec = containerExecer(platform, containerName) | ||
| if !cmd.urlOverridden() { | ||
| // The site's local management listener is amqps with client-cert | ||
| // auth; certs are mounted inside the container. | ||
| if url, err := runtime.GetLocalRouterAddress(cmd.namespace); err == nil { | ||
| cmd.url = url | ||
| cmd.sslArgs = sslArgs(containerCertsPath+"/tls.crt", containerCertsPath+"/tls.key", containerCertsPath+"/ca.crt") | ||
| } | ||
| } | ||
| default: // linux | ||
| if !cmd.urlOverridden() { | ||
| if url, err := runtime.GetLocalRouterAddress(cmd.namespace); err == nil { | ||
| certs := runtime.GetRuntimeTlsCert(cmd.namespace, "skupper-local-client") | ||
| cmd.url = url | ||
| cmd.sslArgs = sslArgs(certs.CertPath, certs.KeyPath, certs.CaPath) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) Run() error { | ||
| if cmd.exec != nil { | ||
| fmt.Printf("running against %s container %s-skupper-router\n", cmd.platform, cmd.namespace) | ||
| } | ||
| _, err := sweeper.Run(sweeper.Config{ | ||
| URL: cmd.url, | ||
| Skmanage: cmd.Flags.Skmanage, | ||
| IdleThresholdSecs: cmd.Flags.IdleThreshold, | ||
| DryRun: cmd.Flags.DryRun, | ||
| Exec: cmd.exec, | ||
| SkmanageExtraArgs: cmd.sslArgs, | ||
| }) | ||
| return err | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) WaitUntil() error { return nil } | ||
|
|
||
| // urlOverridden reports whether the user set --url. | ||
| func (cmd *CmdConnSweeper) urlOverridden() bool { | ||
| return cmd.CobraCmd != nil && cmd.CobraCmd.Flags().Changed("url") | ||
| } | ||
|
|
||
| func sslArgs(cert, key, ca string) []string { | ||
| return []string{"--ssl-certificate", cert, "--ssl-key", key, "--ssl-trustfile", ca} | ||
| } | ||
|
|
||
| // containerExecer returns a sweeper.Execer that runs argv in the router | ||
| // container via `<engine> exec` so that skmanage and python3 are | ||
| // available. | ||
| func containerExecer(engine, containerName string) sweeper.Execer { | ||
| return func(argv []string) ([]byte, error) { | ||
| ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) | ||
| defer cancel() | ||
| full := append([]string{"exec", containerName}, argv...) | ||
| out, err := exec.CommandContext(ctx, engine, full...).Output() | ||
| if err != nil { | ||
| if ee, ok := err.(*exec.ExitError); ok && len(ee.Stderr) > 0 { | ||
| return nil, fmt.Errorf("%s exec %q failed: %w (stderr: %s)", engine, argv[0], err, ee.Stderr) | ||
| } | ||
| return nil, fmt.Errorf("%s exec %q failed: %w", engine, argv[0], err) | ||
| } | ||
| return out, nil | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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
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