// 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" "k8s.io/client-go/transport" 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(ctx context.Context, cfg provider.ProviderConfig) (provider.Provider, error) { return NewWithTransportWrapper(ctx, cfg, nil) } // NewWithTransportWrapper is New with an optional transport wrapper applied // to the provider's own rest.Config (this client is built independently of // the manager's, so the composition root must wrap it separately for // tracing). A nil wrapper means a plain client. func NewWithTransportWrapper(_ context.Context, cfg provider.ProviderConfig, wrap transport.WrapperFunc) (provider.Provider, error) { restCfg, err := ctrl.GetConfig() if err != nil { return nil, fmt.Errorf("kubernetes provider %q: loading kubeconfig: %w", cfg.Name, err) } if wrap != nil { restCfg.Wrap(wrap) } 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) } }