pgcli/internal/kube/client.go

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
}