package kube import ( "context" "fmt" "net/http" "net/url" "strconv" "strings" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/cli-runtime/pkg/genericclioptions" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" "k8s.io/client-go/tools/clientcmd" "k8s.io/client-go/tools/portforward" "k8s.io/client-go/transport/spdy" "github.com/balabanovds/pgcli/internal/config" "github.com/balabanovds/pgcli/internal/core" ) type Client struct { config *rest.Config appConfig *config.KubeConfig client *kubernetes.Clientset } func NewClient(kubeconfig string, appConfig *config.KubeConfig) (*Client, error) { config, err := clientcmd.BuildConfigFromFlags("", kubeconfig) if err != nil { config, err = defaultConfig() if err != nil { return nil, fmt.Errorf("failed to read Kubernetes config: %w", err) } } client, err := kubernetes.NewForConfig(config) if err != nil { return nil, fmt.Errorf("failed to create Kubernetes client: %w", err) } return &Client{ config: config, appConfig: appConfig, client: client, }, nil } type NamespacedName struct { Namespace string Name string } func (c *Client) PodNames(ctx context.Context, namespace string) ([]NamespacedName, error) { podsList, err := c.client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{}) if err != nil { return nil, fmt.Errorf("failed to get Pods for namespace %s: %w", namespace, err) } pods := podsList.Items if len(pods) == 0 { allpods, err := c.client.CoreV1().Pods("").List(ctx, metav1.ListOptions{}) if err != nil { return nil, fmt.Errorf("failed to get Pods for namespace %s: %w", namespace, err) } for _, p := range allpods.Items { if strings.Contains(p.Namespace, namespace) { pods = append(pods, p) } } } res := make([]NamespacedName, 0, 1) for _, p := range pods { if strings.Contains(p.Name, c.appConfig.PodDB.Substring) { res = append(res, NamespacedName{ Namespace: p.Namespace, Name: p.Name, }) } } return res, nil } type Secret struct { User string Password string DBName string Port int } func (c *Client) GetSecret(ctx context.Context, namespace string) (*Secret, error) { secrets, err := c.client.CoreV1().Secrets(namespace).List(ctx, metav1.ListOptions{}) if err != nil { return nil, fmt.Errorf("failed to get Pods for namespace %s: %w", namespace, err) } var scr *v1.Secret for _, s := range secrets.Items { if strings.Contains(s.Name, c.appConfig.Secret.Substring) { scr = &s break } } if scr == nil { return nil, fmt.Errorf("secret for namespace %q containing substring %q not found", namespace, c.appConfig.Secret.Substring) } port, err := strconv.Atoi(string(scr.Data[c.appConfig.Secret.DataFields.Port])) if err != nil { port = 5432 } return &Secret{ User: string(scr.Data[c.appConfig.Secret.DataFields.Username]), Password: string(scr.Data[c.appConfig.Secret.DataFields.Password]), DBName: string(scr.Data[c.appConfig.Secret.DataFields.DBName]), Port: port, }, nil } type ForwardPodRequest struct { // Pod is the selected pod for this port forwarding Pod NamespacedName // LocalPort is the local port that will be selected to expose the PodPort LocalPort int // PodPort is the target port for the pod PodPort int // Steam configures where to write or read input from Stream core.Stream // StopCh is the channel used to manage the port forward lifecycle StopCh <-chan struct{} // ReadyCh communicates when the tunnel is ready to receive traffic ReadyCh chan struct{} } func (c *Client) ForwardPort(req *ForwardPodRequest) error { // stream is used to tell the port forwarder where to place its output or // where to expect input if needed. For the port forwarding we just need // the output eventually iostream := genericclioptions.IOStreams{ In: req.Stream.In, Out: req.Stream.Out, ErrOut: req.Stream.Err, } path := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/portforward", req.Pod.Namespace, req.Pod.Name) hostIP := strings.TrimLeft(c.config.Host, "htps:/") transport, upgrader, err := spdy.RoundTripperFor(c.config) if err != nil { return err } dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, http.MethodPost, &url.URL{Scheme: "https", Path: path, Host: hostIP}) fw, err := portforward.New(dialer, []string{fmt.Sprintf("%d:%d", req.LocalPort, req.PodPort)}, req.StopCh, req.ReadyCh, iostream.Out, iostream.ErrOut) if err != nil { return err } return fw.ForwardPorts() } func defaultConfig() (*rest.Config, error) { pathOptions := clientcmd.NewDefaultPathOptions() config, err := clientcmd.BuildConfigFromKubeconfigGetter("", pathOptions.GetStartingConfig) if err != nil { return nil, fmt.Errorf("error loading REST configuration using default path options: %w", err) } return config, nil }