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,193 @@ | ||
| package kube | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "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/skupperproject/skupper/internal/utils/validator" | ||
| "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 | ||
| clientErr error | ||
| } | ||
|
|
||
| func NewCmdConnSweeper() *CmdConnSweeper { | ||
| return &CmdConnSweeper{} | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) NewClient(cobraCommand *cobra.Command, args []string) { | ||
| cmd.CobraCmd = cobraCommand | ||
| namespaceFlag, _ := cobraCommand.Flags().GetString("namespace") | ||
| contextFlag, _ := cobraCommand.Flags().GetString("context") | ||
| kubeconfigFlag, _ := cobraCommand.Flags().GetString("kubeconfig") | ||
|
|
||
| cli, err := client.NewClient(namespaceFlag, contextFlag, kubeconfigFlag) | ||
| if err != nil { | ||
| cmd.clientErr = fmt.Errorf("failed to initialize kubernetes client: %w", err) | ||
| return | ||
| } | ||
| cmd.KubeClient = cli.GetKubeClient() | ||
| cmd.Namespace = cli.Namespace | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() | ||
| if kubeconfigFlag != "" { | ||
| loadingRules = &clientcmd.ClientConfigLoadingRules{ExplicitPath: kubeconfigFlag} | ||
| } | ||
| kubeconfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig( | ||
| loadingRules, | ||
| &clientcmd.ConfigOverrides{CurrentContext: contextFlag}, | ||
| ) | ||
| restconfig, err := kubeconfig.ClientConfig() | ||
| if err != nil { | ||
| cmd.clientErr = fmt.Errorf("failed to build kubernetes client config: %w", err) | ||
| return | ||
| } | ||
|
|
||
| execConfig := *restconfig | ||
| execConfig.APIPath = "/api" | ||
| execConfig.GroupVersion = &corev1.SchemeGroupVersion | ||
| execConfig.NegotiatedSerializer = scheme.Codecs.WithoutConversion() | ||
| cmd.Rest = &execConfig | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) ValidateInput(args []string) error { | ||
| var validationErrors []error | ||
| numberValidator := validator.NewNumberValidator() | ||
| numberValidator.IncludeZero = false | ||
| if ok, err := numberValidator.Evaluate(cmd.Flags.IdleThreshold); !ok { | ||
| validationErrors = append(validationErrors, fmt.Errorf("idle-threshold is not valid: %s", err)) | ||
| } | ||
| if int64(cmd.Flags.IdleThreshold) > sweeper.MaxIdleThreshold { | ||
| validationErrors = append(validationErrors, fmt.Errorf("idle-threshold is too large (max %d seconds)", sweeper.MaxIdleThreshold)) | ||
| } | ||
| return errors.Join(validationErrors...) | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) InputToOptions() {} | ||
|
|
||
| func (cmd *CmdConnSweeper) Run() error { | ||
| if cmd.clientErr != nil { | ||
| return cmd.clientErr | ||
| } | ||
| if cmd.KubeClient == nil || cmd.Rest == nil { | ||
| return fmt.Errorf("could not initialize kubernetes client") | ||
| } | ||
|
|
||
| podNames, err := cmd.findRouterPods() | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // Each ready replica has its own connections, so sweep every pod. | ||
| 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: sweeper.DefaultURL, | ||
| Skmanage: sweeper.DefaultSkmanage, | ||
| IdleThresholdSecs: cmd.Flags.IdleThreshold, | ||
| Execute: cmd.Flags.Execute, | ||
| 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) | ||
| } | ||
| if total.Failed > 0 { | ||
| return fmt.Errorf("%d idle connection(s) failed to close (%d closed)", total.Failed, total.Killed) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) WaitUntil() error { return nil } | ||
|
|
||
| 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 { | ||
| 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. | ||
| 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) | ||
| } | ||
| } | ||
| } | ||
|
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,140 @@ | ||
| package nonkube | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "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/skupperproject/skupper/internal/utils/validator" | ||
| "github.com/spf13/cobra" | ||
| ) | ||
|
|
||
| const containerCertsPath = "/etc/skupper-router/runtime/certs/skupper-local-client" | ||
| const ( | ||
| containerSkmanage = "/bin/skmanage" | ||
| systemdSkmanage = "/usr/bin/skmanage" | ||
| ) | ||
|
|
||
| type CmdConnSweeper struct { | ||
| CobraCmd *cobra.Command | ||
| Flags *common.CommandConnSweeperFlags | ||
| namespace string | ||
| platform string | ||
| url string | ||
| skmanage string | ||
| sslArgs []string | ||
| exec sweeper.Execer | ||
| setupErr error | ||
| } | ||
|
|
||
| func NewCmdConnSweeper() *CmdConnSweeper { | ||
| return &CmdConnSweeper{} | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) NewClient(cobraCommand *cobra.Command, args []string) { | ||
| cmd.CobraCmd = cobraCommand | ||
| cmd.namespace, _ = cobraCommand.Flags().GetString(common.FlagNameNamespace) | ||
| } | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| func (cmd *CmdConnSweeper) ValidateInput(args []string) error { | ||
| var validationErrors []error | ||
| numberValidator := validator.NewNumberValidator() | ||
| numberValidator.IncludeZero = false | ||
| if ok, err := numberValidator.Evaluate(cmd.Flags.IdleThreshold); !ok { | ||
| validationErrors = append(validationErrors, fmt.Errorf("idle-threshold is not valid: %s", err)) | ||
| } | ||
| if int64(cmd.Flags.IdleThreshold) > sweeper.MaxIdleThreshold { | ||
| validationErrors = append(validationErrors, fmt.Errorf("idle-threshold is too large (max %d seconds)", sweeper.MaxIdleThreshold)) | ||
| } | ||
| return errors.Join(validationErrors...) | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) InputToOptions() { | ||
| if cmd.namespace == "" { | ||
| cmd.namespace = "default" | ||
| } | ||
|
|
||
| platformLoader := &nonkubecommon.NamespacePlatformLoader{} | ||
| platform, err := platformLoader.Load(cmd.namespace) | ||
| if err != nil { | ||
| cmd.setupErr = fmt.Errorf("could not load platform for namespace %q: %w", cmd.namespace, err) | ||
| return | ||
| } | ||
| cmd.platform = platform | ||
|
|
||
| url, err := runtime.GetLocalRouterAddress(cmd.namespace) | ||
| if err != nil { | ||
| cmd.setupErr = fmt.Errorf("could not determine router address for namespace %q: %w", cmd.namespace, err) | ||
| return | ||
| } | ||
| cmd.url = url | ||
|
|
||
| switch platform { | ||
| case "podman", "docker": | ||
| cmd.skmanage = containerSkmanage | ||
| cmd.exec = containerExecer(platform, cmd.namespace+"-skupper-router") | ||
| cmd.sslArgs = sslArgs(containerCertsPath+"/tls.crt", containerCertsPath+"/tls.key", containerCertsPath+"/ca.crt") | ||
| default: | ||
| cmd.skmanage = systemdSkmanage | ||
| certs := runtime.GetRuntimeTlsCert(cmd.namespace, "skupper-local-client") | ||
| cmd.sslArgs = sslArgs(certs.CertPath, certs.KeyPath, certs.CaPath) | ||
| } | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) Run() error { | ||
| if cmd.setupErr != nil { | ||
| return cmd.setupErr | ||
| } | ||
| if cmd.url == "" || cmd.skmanage == "" { | ||
| return fmt.Errorf("could not determine router management address for namespace %q", cmd.namespace) | ||
| } | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| if cmd.exec != nil { | ||
| fmt.Printf("running against %s container %s-skupper-router\n", cmd.platform, cmd.namespace) | ||
| } | ||
| res, err := sweeper.Run(sweeper.Config{ | ||
| URL: cmd.url, | ||
| Skmanage: cmd.skmanage, | ||
| IdleThresholdSecs: cmd.Flags.IdleThreshold, | ||
| Execute: cmd.Flags.Execute, | ||
| Exec: cmd.exec, | ||
| SkmanageExtraArgs: cmd.sslArgs, | ||
| }) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if res.Failed > 0 { | ||
| return fmt.Errorf("%d idle connection(s) failed to close (%d closed)", res.Failed, res.Killed) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (cmd *CmdConnSweeper) WaitUntil() error { return nil } | ||
|
|
||
| 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