egress-proxies-operator/internal/provider/kubernetes/kubernetes.go

// Package kubernetes is a provider.Provider that creates real Pods running
// a Squid container in the same cluster the operator itself runs in —
// unlike a cloud provider, "creating compute" here means talking back to
// the very Kubernetes API the operator is already watching.
package kubernetes

import (
	"context"
	"fmt"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/cache"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)

// defaultImage is Canonical's actively maintained Squid image for Ubuntu
// 24.04 LTS, verified before picking it: a public image (no build/load
// step needed for a kind demo), 50M+ pulls, updated the same day this
// decision was made.
const defaultImage = "ubuntu/squid:6.6-24.04_edge"

// Provider creates proxy Pods. It builds its own client rather than
// depending on the manager's, via ctrl.GetConfig() — which auto-detects
// in-cluster config when running as a Pod and falls back to the local
// kubeconfig otherwise. That's what makes `make run` against a local kind
// cluster and running in-cluster use the exact same code path with no
// provider-specific wiring in cmd/main.go.
type Provider struct {
	client client.Client
	image  string
}

// New builds a kubernetes Provider from its config block. Satisfies
// registry.Constructor.
func New(_ context.Context, cfg provider.ProviderConfig) (provider.Provider, error) {
	restCfg, err := ctrl.GetConfig()
	if err != nil {
		return nil, fmt.Errorf("kubernetes provider %q: loading kubeconfig: %w", cfg.Name, err)
	}
	scheme := runtime.NewScheme()
	if err := corev1.AddToScheme(scheme); err != nil {
		return nil, fmt.Errorf("kubernetes provider %q: %w", cfg.Name, err)
	}
	c, err := client.New(restCfg, client.Options{Scheme: scheme})
	if err != nil {
		return nil, fmt.Errorf("kubernetes provider %q: building client: %w", cfg.Name, err)
	}
	return newWithClient(c, cfg), nil
}

// newWithClient builds a Provider around an already-constructed client,
// bypassing ctrl.GetConfig(). Tests use this exclusively — New() must
// never run under `go test`, since ctrl.GetConfig() would happily connect
// to whatever real cluster the developer's kubeconfig points at.
func newWithClient(c client.Client, cfg provider.ProviderConfig) *Provider {
	image := defaultImage
	if cfg.Kubernetes != nil && cfg.Kubernetes.Image != "" {
		image = cfg.Kubernetes.Image
	}
	return &Provider{client: c, image: image}
}

// Create creates a Pod running the proxy container. Idempotent by
// req.Name: if a Pod with that name already exists in req.Namespace, its
// providerID is returned rather than erroring, so a repeat call after a
// crash finds the existing Pod instead of creating a duplicate.
func (p *Provider) Create(ctx context.Context, req provider.CreateRequest) (string, error) {
	pod := buildPod(p.image, req)
	if err := p.client.Create(ctx, pod); err != nil {
		if apierrors.IsAlreadyExists(err) {
			return providerID(req.Namespace, req.Name), nil
		}
		return "", classify("create", req.Name, err)
	}
	return providerID(req.Namespace, req.Name), nil
}

// Get returns the current state of a previously created Pod.
func (p *Provider) Get(ctx context.Context, id string) (*provider.Instance, error) {
	ns, name, err := parseProviderID(id)
	if err != nil {
		return nil, provider.Wrap(provider.ErrPermanent, "get", "kubernetes", id, err)
	}
	var pod corev1.Pod
	if err := p.client.Get(ctx, types.NamespacedName{Namespace: ns, Name: name}, &pod); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, provider.Wrap(provider.ErrNotFound, "get", "kubernetes", id, nil)
		}
		return nil, classify("get", id, err)
	}
	return instanceFromPod(&pod), nil
}

// Delete is idempotent: deleting an unknown Pod is not an error.
func (p *Provider) Delete(ctx context.Context, id string) error {
	ns, name, err := parseProviderID(id)
	if err != nil {
		return provider.Wrap(provider.ErrPermanent, "delete", "kubernetes", id, err)
	}
	pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: name}}
	if err := p.client.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) {
		return classify("delete", id, err)
	}
	return nil
}

// ListByTag returns every Pod carrying LabelManaged, across every
// namespace — orphan GC needs to find proxy Pods regardless of which
// namespace Proxy CRs happen to live in, which is why this provider's RBAC
// is cluster-scoped rather than namespaced.
func (p *Provider) ListByTag(ctx context.Context) ([]provider.Instance, error) {
	var pods corev1.PodList
	if err := p.client.List(ctx, &pods, client.MatchingLabels{
		provider.LabelManaged: provider.LabelManagedYes,
	}); err != nil {
		return nil, classify("list", "", err)
	}
	out := make([]provider.Instance, 0, len(pods.Items))
	for i := range pods.Items {
		out = append(out, *instanceFromPod(&pods.Items[i]))
	}
	return out, nil
}

// providerID encodes namespace and name into the opaque providerID string
// the Provider interface exposes, so Get/Delete are self-contained without
// needing to separately track or re-derive which namespace a Pod lives in
// — the same reasoning as the GCP provider's zone-qualified providerID.
func providerID(namespace, name string) string {
	return namespace + "/" + name
}

func parseProviderID(id string) (namespace, name string, err error) {
	ns, n, err := cache.SplitMetaNamespaceKey(id)
	if err != nil {
		return "", "", err
	}
	if ns == "" {
		return "", "", fmt.Errorf("providerID %q missing a namespace", id)
	}
	return ns, n, nil
}

func instanceFromPod(pod *corev1.Pod) *provider.Instance {
	inst := &provider.Instance{
		ID:        providerID(pod.Namespace, pod.Name),
		UID:       pod.Labels[provider.LabelUID],
		CreatedAt: pod.CreationTimestamp.Time,
		State:     stateFromPod(pod),
	}
	if inst.State == provider.StateRunning {
		inst.IP = pod.Status.PodIP
	}
	return inst
}

// stateFromPod maps a Pod's phase to InstanceState. Succeeded/Failed/
// Unknown all collapse to Terminated: the reconciler treats Stopped and
// Terminated identically (delete and recreate — cattle, not pets), so a
// finer-grained distinction between "the container exited" and "the node
// went unreachable" wouldn't change any behavior.
func stateFromPod(pod *corev1.Pod) provider.InstanceState {
	switch pod.Status.Phase {
	case corev1.PodPending:
		return provider.StateProvisioning
	case corev1.PodRunning:
		if pod.Status.PodIP == "" {
			// Never publish an empty IP while the kubelet is still
			// finishing setup.
			return provider.StateProvisioning
		}
		return provider.StateRunning
	default: // Succeeded, Failed, Unknown
		return provider.StateTerminated
	}
}

// classify maps a Kubernetes API error to the provider error taxonomy.
// Quota-exceeded and RBAC-denied both surface as 403 Forbidden from the
// API server with nothing in apierrors to tell them apart programmatically
// — a known simplification for this prototype; both classify as
// ErrPermanent, which is the safer default of the two (stop retrying
// rather than hammering an API server that will never allow the request).
func classify(op, id string, err error) error {
	switch {
	case apierrors.IsNotFound(err):
		return provider.Wrap(provider.ErrNotFound, op, "kubernetes", id, err)
	case apierrors.IsTooManyRequests(err), apierrors.IsServerTimeout(err), apierrors.IsTimeout(err):
		return provider.Wrap(provider.ErrTransient, op, "kubernetes", id, err)
	case apierrors.IsForbidden(err), apierrors.IsInvalid(err), apierrors.IsBadRequest(err), apierrors.IsUnauthorized(err):
		return provider.Wrap(provider.ErrPermanent, op, "kubernetes", id, err)
	default:
		return provider.Wrap(provider.ErrTransient, op, "kubernetes", id, err)
	}
}