Add the Kubernetes pod provider (replaces the removed mock)

Create/Get/Delete/ListByTag against real corev1.Pod objects in the same
cluster the operator runs in, running an ubuntu/squid container -- picked
by actually checking Docker Hub metadata (Canonical-published, rebuilt
the same day this was decided, 50M+ pulls) rather than guessing an image
reference. It's a public image, so kind nodes pull it directly with no
build/load step.

providerID is "<namespace>/<podName>", parsed via
cache.SplitMetaNamespaceKey -- the same self-contained-providerID
reasoning the plan already calls for on the GCP provider's zone-qualified
IDs. Pod state maps to InstanceState with Succeeded/Failed/Unknown all
collapsing to Terminated, since the reconciler already treats Stopped and
Terminated identically; Running-without-PodIP maps to Provisioning so an
empty IP is never published.

The client is built internally via ctrl.GetConfig() (in-cluster or local
kubeconfig, whichever applies), not threaded through the registry
Constructor signature -- this is what lets `make run` against a local
kind cluster and running in-cluster share the exact same code path with
no provider-specific wiring in cmd/main.go. New() is deliberately
untested (0% coverage): it's the one function that must never run under
`go test`, since it would happily connect to whatever cluster the
developer's kubeconfig points at. Tests construct Provider via an
unexported newWithClient(client, cfg) instead.

ListByTag lists Pods across every namespace (orphan GC needs to find
every tagged Pod regardless of where it landed), which means this
provider's RBAC has to be a ClusterRole rather than namespace-scoped --
flagged now, wired in Step 10.

provider.Config gains KubernetesConfig (replacing MockConfig) and drops
the FailWith*/fault-injection surface entirely, since that need is now
served by a small in-test stub Provider for reconciler tests (Step 4),
not a config-driven mechanism on a real provider package.

Tests use sigs.k8s.io/controller-runtime/pkg/client/fake -- real Pod
objects, the real client.Client interface -- at 77.6% coverage.
make test green across the whole repo.

Co-Authored-By: Claude <noreply@anthropic.com>

.claude/settings.json

@@ -51,13 +51,18 @@
       "Bash(kind get *)",
       "Bash(curl -s --max-time 5 \"https://hub.docker.com/v2/repositories/ubuntu/squid/tags?page_size=10\")",
       "Bash(python3 -c ' *)",
-      "Bash(curl -s --max-time 5 \"https://hub.docker.com/v2/repositories/ubuntu/squid/tags?page_size=100\")"
+      "Bash(curl -s --max-time 5 \"https://hub.docker.com/v2/repositories/ubuntu/squid/tags?page_size=100\")",
+      "Bash(go doc *)",
+      "Bash(go list *)",
+      "Bash(gofmt -w internal/provider/config.go)",
+      "Bash(gofmt -l .)"
     ],
     "additionalDirectories": [
       "/Users/jan.novak/srv/go/egress-proxies-operator/.claude",
       "/Users/jan.novak/srv/go/egress-proxies-operator/docs/plans",
       "/Users/jan.novak/srv/go/egress-proxies-operator/docs",
-      "/Users/jan.novak/srv/go/egress-proxies-operator/docs/prompts"
+      "/Users/jan.novak/srv/go/egress-proxies-operator/docs/prompts",
+      "/tmp"
     ]
   }
 }

docs/plans-executions/2026-08-07-1747-proxy-operator.md

@@ -303,3 +303,110 @@ end-to-end healthcheck, not a simulated one.
 
 `internal/provider/mock` at 91.1% coverage. `make test` green across the
 whole repo.
+
+## Step 3 (revised) — Kubernetes pod provider replaces the mock
+
+After the mock provider above was built and working, the user pushed back:
+it felt too far from the real system to build confidence in, and they'd
+rather have simpler, more "real" code than a fast-but-simulated test
+double — a `kind`-based verification pass "once in a while" is an
+acceptable trade. They proposed replacing it outright with a provider that
+creates real Pods in the operator's own cluster, rather than keeping mock
+around as a fallback.
+
+**Talked through the trade-off before agreeing to it, since it wasn't free
+of downsides.** Pods sharing a cluster's egress IPs don't solve what this
+operator actually exists for (routing around IP-based rate limiting needs
+genuinely distinct egress paths — that's still only `gcp`), and `envtest`
+has no kubelet, so a Pod-based provider can never be exercised by the fast
+test suite either. Net effect: this is a replacement for the mock's role
+in local dev/CI confidence-building, not a new alternative to GCP, and the
+reconciler's own state-machine tests (Step 4) still need a minimal
+in-test stub `Provider` — a handful of lines in the test file, not a
+package with its own config format or fault-injection surface, which is
+the specific kind of complexity the user was pushing back on.
+
+**Picked the proxy software by actually checking what's out there,** not
+by guessing a Docker Hub path:
+
+```bash
+curl -s "https://hub.docker.com/v2/repositories/vimagick/tinyproxy/" | head -c 400
+# last_updated 2021-07-22 — stale
+curl -s "https://hub.docker.com/v2/repositories/ubuntu/squid/" | head -c 400
+# last_updated 2026-08-07T04:36:14Z — updated the same day, Canonical-published, 50M+ pulls
+```
+
+`ubuntu/squid` won clearly: actively maintained (rebuilt the same day this
+check ran), official publisher, and — since it's a public image — `kind`
+nodes pull it directly, no build/load step needed for the quickstart.
+Pinned to `6.6-24.04_edge` (Ubuntu 24.04 LTS base) rather than floating
+`latest`, for reproducibility.
+
+**Design, in `internal/provider/kubernetes/`:**
+
+- **`pod.go`** — `buildPod` is a pure function (mirrors the GCP provider's
+  planned `buildInsertRequest`): builds a `corev1.Pod` with one Squid
+  container. Config is generated in Go and passed via a `SQUID_CONF` env
+  var that the container's command writes to `/etc/squid/squid.conf`
+  before `exec squid` — deliberately not a separate `ConfigMap`, so
+  there's still exactly one Kubernetes object per proxy instance to
+  create, track, and clean up. The config itself is intentionally
+  permissive (`http_access allow all`, `via off`, `forwarded_for off`) —
+  documented in-code as a prototype-for-a-private-cluster choice, the same
+  posture the spec already takes toward cloud-init on the GCP provider
+  (installing/configuring proxy software is explicitly out of scope
+  there; this provider doesn't try to do more).
+- **`kubernetes.go`** — `Create`/`Get`/`Delete`/`ListByTag` against a
+  `client.Client`. **`providerID` is `<namespace>/<podName>`**, parsed
+  with `k8s.io/client-go/tools/cache.SplitMetaNamespaceKey` — the exact
+  same reasoning as the GCP provider's planned zone-qualified providerID
+  (Step 8): `Get`/`Delete` need to be self-contained without re-deriving
+  where the resource lives. State mapping collapses `Succeeded`/`Failed`/
+  `Unknown` all into `Terminated`, since the reconciler already treats
+  Stopped and Terminated identically (delete + recreate) — no finer
+  distinction would change any behavior. `Running` with no `PodIP` yet
+  maps to `Provisioning`, not `Running`, so an empty IP is never
+  published — the same rule the plan already called out for GCP's
+  `RUNNING`-without-`NatIP` case.
+- **Client construction is the one genuinely new pattern this provider
+  needed** that GCP/mock didn't: it builds its own `client.Client` 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 zero provider-specific wiring in
+  `cmd/main.go`. The corresponding risk: `New()` must never run under
+  `go test` — it would happily connect to whatever real cluster the
+  developer's kubeconfig points at. Solved the same way `mock.Provider`
+  solved clock injection: an unexported `newWithClient(c, cfg)` constructor
+  that tests call directly, bypassing `ctrl.GetConfig()` entirely.
+  `New()` sits at 0% test coverage, deliberately — it's the one function
+  that must stay untested by design.
+- **RBAC implication worth flagging now** (implemented in Step 10):
+  `ListByTag` lists Pods across every namespace, not just the namespace(s)
+  Proxies live in, because orphan GC needs to find every Pod this operator
+  tagged regardless of where it landed. That means the operator's Pod
+  permissions have to be a `ClusterRole`, not scoped to a single
+  namespace — a real trade-off against the "fleet lives in one namespace,
+  keeps RBAC simple" principle the spec states for the CRD itself.
+  Documented rather than special-cased away.
+- **One known simplification, documented in a code comment rather than
+  solved:** Kubernetes surfaces both RBAC-denied and quota-exceeded as the
+  same 403 Forbidden, and `apierrors` has no helper to tell them apart.
+  Both classify as `ErrPermanent` — the safer of the two defaults (stop
+  retrying rather than hammering an API server that will never allow the
+  request), but a real ResourceQuota failure that would clear once other
+  proxies are deleted won't get the 5-minute-backoff-and-retry treatment
+  `ErrQuotaExceeded` gives on the GCP path.
+
+**Testing:** unlike the mock, this package's tests use
+`sigs.k8s.io/controller-runtime/pkg/client/fake` — real `corev1.Pod`
+objects, the real `client.Client` interface, not a hand-rolled in-memory
+map. That's a strictly more realistic test double than what mock.Provider
+was, while still being fast (no cluster, no kubelet) — it just can't prove
+a container actually starts and serves traffic, which is exactly the gap
+the `kind`-based verification pass is for. 77.6% coverage; the only
+meaningfully uncovered function is `New()` itself, deliberately.
+
+`make test` green across the whole repo (`go build`/`go vet` clean,
+`internal/provider` 96.0%, `internal/provider/kubernetes` 77.6%,
+`internal/provider/registry` 100%, unchanged).

go.mod

@@ -5,6 +5,7 @@ go 1.26.0
 require (
 	github.com/onsi/ginkgo/v2 v2.27.4
 	github.com/onsi/gomega v1.39.0
+	k8s.io/api v0.36.0
 	k8s.io/apimachinery v0.36.0
 	k8s.io/client-go v0.36.0
 	sigs.k8s.io/controller-runtime v0.24.1
@@ -85,7 +86,6 @@ require (
 	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/api v0.36.0 // indirect
 	k8s.io/apiextensions-apiserver v0.36.0 // indirect
 	k8s.io/apiserver v0.36.0 // indirect
 	k8s.io/component-base v0.36.0 // indirect

internal/provider/config.go

@@ -7,23 +7,6 @@ import (
 	"sigs.k8s.io/yaml"
 )
 
-// Fail-with classes accepted by MockConfig.FailWith, shared with the mock
-// provider's fault injection so the two sides never drift on the string
-// values.
-const (
-	FailWithNotFound  = "notfound"
-	FailWithQuota     = "quota"
-	FailWithTransient = "transient"
-	FailWithPermanent = "permanent"
-)
-
-var validFailClasses = map[string]bool{
-	FailWithNotFound:  true,
-	FailWithQuota:     true,
-	FailWithTransient: true,
-	FailWithPermanent: true,
-}
-
 // Config is the top-level shape of the --providers-config file.
 type Config struct {
 	Providers []ProviderConfig `json:"providers"`
@@ -34,27 +17,17 @@ type Config struct {
 // blocks. Exactly one of the type-specific blocks below should be set,
 // matching Type.
 type ProviderConfig struct {
-	Name string      `json:"name"`
+	Name       string            `json:"name"`
-	Type string      `json:"type"`
+	Type       string            `json:"type"`
-	Mock *MockConfig `json:"mock,omitempty"`
+	Kubernetes *KubernetesConfig `json:"kubernetes,omitempty"`
-	GCP  *GCPConfig  `json:"gcp,omitempty"`
+	GCP        *GCPConfig        `json:"gcp,omitempty"`
 }
 
-// MockConfig configures the in-memory mock provider.
+// KubernetesConfig configures the kubernetes-pod provider, which creates
-type MockConfig struct {
+// proxy Pods in the same cluster the operator itself runs in.
-	// ProvisionDelaySeconds is how long a created instance reports
+type KubernetesConfig struct {
-	// Provisioning before Running. Default 5.
+	// Image is the proxy container image. Default "ubuntu/squid:6.6-24.04_edge".
-	ProvisionDelaySeconds int32 `json:"provisionDelaySeconds,omitempty"`
+	Image string `json:"image,omitempty"`
-	// DeleteDelaySeconds is how long a deleted instance reports Terminated
-	// before Get starts returning ErrNotFound. Default 1.
-	DeleteDelaySeconds int32 `json:"deleteDelaySeconds,omitempty"`
-	// FailNextCreates makes the next N Create calls fail with FailWith,
-	// for exercising the reconciler's error handling in demos.
-	FailNextCreates int `json:"failNextCreates,omitempty"`
-	// FailWith selects the error class injected failures return: one of
-	// FailWithNotFound/FailWithQuota/FailWithTransient/FailWithPermanent.
-	// Default FailWithTransient.
-	FailWith string `json:"failWith,omitempty"`
 }
 
 // GCPConfig configures a named GCP provider instance.
@@ -109,16 +82,13 @@ func (c *Config) validate() error {
 		seen[p.Name] = true
 
 		switch p.Type {
-		case "mock":
+		case "kubernetes":
 			if p.GCP != nil {
-				return fmt.Errorf("providers[%d] %q: type is mock but a gcp block is set", i, p.Name)
+				return fmt.Errorf("providers[%d] %q: type is kubernetes but a gcp block is set", i, p.Name)
-			}
-			if p.Mock != nil && p.Mock.FailWith != "" && !validFailClasses[p.Mock.FailWith] {
-				return fmt.Errorf("providers[%d] %q: unknown mock.failWith %q", i, p.Name, p.Mock.FailWith)
 			}
 		case "gcp":
-			if p.Mock != nil {
+			if p.Kubernetes != nil {
-				return fmt.Errorf("providers[%d] %q: type is gcp but a mock block is set", i, p.Name)
+				return fmt.Errorf("providers[%d] %q: type is gcp but a kubernetes block is set", i, p.Name)
 			}
 			if p.GCP == nil || p.GCP.Project == "" {
 				return fmt.Errorf("providers[%d] %q: gcp.project is required", i, p.Name)

internal/provider/config_test.go

@@ -9,10 +9,10 @@ func TestLoadConfig_valid(t *testing.T) {
 	t.Parallel()
 	data := []byte(`
 providers:
-  - name: mock
+  - name: kubernetes
-    type: mock
+    type: kubernetes
-    mock:
+    kubernetes:
-      provisionDelaySeconds: 2
+      image: ubuntu/squid:6.6-24.04_edge
   - name: gcp-eu
     type: gcp
     gcp:
@@ -26,8 +26,8 @@ providers:
 	if len(cfg.Providers) != 2 {
 		t.Fatalf("len(cfg.Providers) = %d, want 2", len(cfg.Providers))
 	}
-	if cfg.Providers[0].Mock == nil || cfg.Providers[0].Mock.ProvisionDelaySeconds != 2 {
+	if cfg.Providers[0].Kubernetes == nil || cfg.Providers[0].Kubernetes.Image != "ubuntu/squid:6.6-24.04_edge" {
-		t.Errorf("providers[0].mock = %+v, want ProvisionDelaySeconds=2", cfg.Providers[0].Mock)
+		t.Errorf("providers[0].kubernetes = %+v, want Image=ubuntu/squid:6.6-24.04_edge", cfg.Providers[0].Kubernetes)
 	}
 	if cfg.Providers[1].GCP == nil || cfg.Providers[1].GCP.Project != "my-project" {
 		t.Errorf("providers[1].gcp = %+v, want Project=my-project", cfg.Providers[1].GCP)
@@ -50,17 +50,17 @@ func TestLoadConfig_invalid(t *testing.T) {
 			name: "missing name",
 			yaml: `
 providers:
-  - type: mock`,
+  - type: kubernetes`,
 			wantErrSub: "name is required",
 		},
 		{
 			name: "duplicate name",
 			yaml: `
 providers:
-  - name: mock
+  - name: kubernetes
-    type: mock
+    type: kubernetes
-  - name: mock
+  - name: kubernetes
-    type: mock`,
+    type: kubernetes`,
 			wantErrSub: "duplicate provider name",
 		},
 		{
@@ -97,43 +97,33 @@ providers:
 			wantErrSub: "gcp.project is required",
 		},
 		{
-			name: "mock type with gcp block",
+			name: "kubernetes type with gcp block",
 			yaml: `
 providers:
   - name: p1
-    type: mock
+    type: kubernetes
     gcp:
       project: my-project`,
-			wantErrSub: "type is mock but a gcp block is set",
+			wantErrSub: "type is kubernetes but a gcp block is set",
 		},
 		{
-			name: "gcp type with mock block",
+			name: "gcp type with kubernetes block",
 			yaml: `
 providers:
   - name: p1
     type: gcp
     gcp:
       project: my-project
-    mock:
+    kubernetes:
-      failNextCreates: 1`,
+      image: custom-image`,
-			wantErrSub: "type is gcp but a mock block is set",
+			wantErrSub: "type is gcp but a kubernetes block is set",
-		},
-		{
-			name: "unknown mock.failWith",
-			yaml: `
-providers:
-  - name: p1
-    type: mock
-    mock:
-      failWith: oops`,
-			wantErrSub: `unknown mock.failWith "oops"`,
 		},
 		{
 			name: "strict mode rejects unknown top-level key",
 			yaml: `
 providers:
   - name: p1
-    type: mock
+    type: kubernetes
 extraneous: true`,
 			wantErrSub: "parsing providers config",
 		},
@@ -142,7 +132,7 @@ extraneous: true`,
 			yaml: `
 providers:
   - name: p1
-    type: mock
+    type: kubernetes
     bogus: true`,
 			wantErrSub: "parsing providers config",
 		},

internal/provider/kubernetes/kubernetes.go

@@ -0,0 +1,202 @@
+// 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)
+	}
+}

internal/provider/kubernetes/kubernetes_test.go

@@ -0,0 +1,245 @@
+package kubernetes
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	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/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
+)
+
+func newTestProvider(objs ...runtime.Object) *Provider {
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		panic(err)
+	}
+	c := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(objs...).Build()
+	return newWithClient(c, provider.ProviderConfig{Name: "kubernetes"})
+}
+
+func testPod(namespace, name, uid string, phase corev1.PodPhase, ip string) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+			Labels: map[string]string{
+				provider.LabelManaged: provider.LabelManagedYes,
+				provider.LabelUID:     uid,
+			},
+		},
+		Status: corev1.PodStatus{Phase: phase, PodIP: ip},
+	}
+}
+
+func TestProvider_Create_buildsPodAndReturnsProviderID(t *testing.T) {
+	t.Parallel()
+	p := newTestProvider()
+	id, err := p.Create(context.Background(), provider.CreateRequest{
+		Name: "proxy-abc", UID: "uid-1", Namespace: "crawl", Port: 3128,
+	})
+	if err != nil {
+		t.Fatalf("Create() error = %v", err)
+	}
+	if id != "crawl/proxy-abc" {
+		t.Errorf("Create() id = %q, want %q", id, "crawl/proxy-abc")
+	}
+
+	var pod corev1.Pod
+	key := types.NamespacedName{Namespace: "crawl", Name: "proxy-abc"}
+	if err := p.client.Get(context.Background(), key, &pod); err != nil {
+		t.Fatalf("expected the Pod to exist: %v", err)
+	}
+	if pod.Labels[provider.LabelUID] != "uid-1" {
+		t.Errorf("pod UID label = %q, want %q", pod.Labels[provider.LabelUID], "uid-1")
+	}
+}
+
+func TestProvider_Create_idempotent(t *testing.T) {
+	t.Parallel()
+	p := newTestProvider()
+	ctx := context.Background()
+	req := provider.CreateRequest{Name: "proxy-dup", UID: "uid-2", Namespace: "crawl", Port: 3128}
+
+	id1, err := p.Create(ctx, req)
+	if err != nil {
+		t.Fatalf("Create() #1 error = %v", err)
+	}
+	id2, err := p.Create(ctx, req)
+	if err != nil {
+		t.Fatalf("Create() #2 error = %v, want nil (AlreadyExists must be absorbed)", err)
+	}
+	if id1 != id2 {
+		t.Errorf("Create() not idempotent: %q != %q", id1, id2)
+	}
+}
+
+func TestProvider_Get_stateMapping(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name      string
+		phase     corev1.PodPhase
+		ip        string
+		wantState provider.InstanceState
+		wantIP    string
+	}{
+		{"pending", corev1.PodPending, "", provider.StateProvisioning, ""},
+		{"running with IP", corev1.PodRunning, "10.0.0.5", provider.StateRunning, "10.0.0.5"},
+		{"running without IP yet", corev1.PodRunning, "", provider.StateProvisioning, ""},
+		{"succeeded", corev1.PodSucceeded, "10.0.0.5", provider.StateTerminated, ""},
+		{"failed", corev1.PodFailed, "10.0.0.5", provider.StateTerminated, ""},
+		{"unknown", corev1.PodUnknown, "10.0.0.5", provider.StateTerminated, ""},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			pod := testPod("crawl", "proxy-"+tc.name, "uid", tc.phase, tc.ip)
+			p := newTestProvider(pod)
+			inst, err := p.Get(context.Background(), "crawl/"+pod.Name)
+			if err != nil {
+				t.Fatalf("Get() error = %v", err)
+			}
+			if inst.State != tc.wantState {
+				t.Errorf("State = %v, want %v", inst.State, tc.wantState)
+			}
+			if inst.IP != tc.wantIP {
+				t.Errorf("IP = %q, want %q", inst.IP, tc.wantIP)
+			}
+			if inst.UID != "uid" {
+				t.Errorf("UID = %q, want %q", inst.UID, "uid")
+			}
+		})
+	}
+}
+
+func TestProvider_Get_notFound(t *testing.T) {
+	t.Parallel()
+	p := newTestProvider()
+	_, err := p.Get(context.Background(), "crawl/does-not-exist")
+	if !errors.Is(err, provider.ErrNotFound) {
+		t.Errorf("Get() error = %v, want ErrNotFound", err)
+	}
+}
+
+func TestProvider_Get_malformedProviderID(t *testing.T) {
+	t.Parallel()
+	p := newTestProvider()
+	_, err := p.Get(context.Background(), "no-namespace-here")
+	if err == nil {
+		t.Fatal("Get() error = nil, want error for a providerID with no namespace")
+	}
+}
+
+func TestProvider_Delete_idempotent(t *testing.T) {
+	t.Parallel()
+	pod := testPod("crawl", "proxy-del", "uid", corev1.PodRunning, "10.0.0.5")
+	p := newTestProvider(pod)
+	ctx := context.Background()
+
+	if err := p.Delete(ctx, "crawl/proxy-del"); err != nil {
+		t.Fatalf("Delete() error = %v", err)
+	}
+	if err := p.Delete(ctx, "crawl/proxy-del"); err != nil {
+		t.Errorf("Delete() (repeat) error = %v, want nil", err)
+	}
+	if err := p.Delete(ctx, "crawl/never-existed"); err != nil {
+		t.Errorf("Delete() on unknown ID error = %v, want nil", err)
+	}
+
+	if _, err := p.Get(ctx, "crawl/proxy-del"); !errors.Is(err, provider.ErrNotFound) {
+		t.Errorf("Get() after Delete() error = %v, want ErrNotFound", err)
+	}
+}
+
+func TestProvider_Delete_malformedProviderID(t *testing.T) {
+	t.Parallel()
+	p := newTestProvider()
+	if err := p.Delete(context.Background(), "no-namespace-here"); err == nil {
+		t.Fatal("Delete() error = nil, want error for a providerID with no namespace")
+	}
+}
+
+func TestProvider_ListByTag_filtersByLabelAcrossNamespaces(t *testing.T) {
+	t.Parallel()
+	managed1 := testPod("crawl", "proxy-a", "uid-a", corev1.PodRunning, "10.0.0.1")
+	managed2 := testPod("other-ns", "proxy-b", "uid-b", corev1.PodPending, "")
+	unmanaged := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "not-ours", Namespace: "crawl"}}
+
+	p := newTestProvider(managed1, managed2, unmanaged)
+	instances, err := p.ListByTag(context.Background())
+	if err != nil {
+		t.Fatalf("ListByTag() error = %v", err)
+	}
+	if len(instances) != 2 {
+		t.Fatalf("len(instances) = %d, want 2 (unmanaged Pod must be excluded)", len(instances))
+	}
+
+	byID := make(map[string]provider.Instance, len(instances))
+	for _, inst := range instances {
+		byID[inst.ID] = inst
+	}
+	a, ok := byID["crawl/proxy-a"]
+	if !ok {
+		t.Fatal(`instances missing "crawl/proxy-a"`)
+	}
+	if a.State != provider.StateRunning || a.IP != "10.0.0.1" {
+		t.Errorf("crawl/proxy-a = %+v, want Running/10.0.0.1", a)
+	}
+	if _, ok := byID["other-ns/proxy-b"]; !ok {
+		t.Fatal(`instances missing "other-ns/proxy-b" (ListByTag must not be namespace-scoped)`)
+	}
+}
+
+func TestNewWithClient_image(t *testing.T) {
+	t.Parallel()
+	scheme := runtime.NewScheme()
+	_ = corev1.AddToScheme(scheme)
+	c := fake.NewClientBuilder().WithScheme(scheme).Build()
+
+	def := newWithClient(c, provider.ProviderConfig{Name: "k8s"})
+	if def.image != defaultImage {
+		t.Errorf("default image = %q, want %q", def.image, defaultImage)
+	}
+
+	custom := newWithClient(c, provider.ProviderConfig{
+		Name:       "k8s",
+		Kubernetes: &provider.KubernetesConfig{Image: "myregistry/squid:custom"},
+	})
+	if custom.image != "myregistry/squid:custom" {
+		t.Errorf("custom image = %q, want %q", custom.image, "myregistry/squid:custom")
+	}
+}
+
+func TestClassify(t *testing.T) {
+	t.Parallel()
+	gr := schema.GroupResource{Group: "", Resource: "pods"}
+	tests := []struct {
+		name string
+		err  error
+		want error
+	}{
+		{"not found", apierrors.NewNotFound(gr, "x"), provider.ErrNotFound},
+		{"too many requests", apierrors.NewTooManyRequests("slow down", 5), provider.ErrTransient},
+		{"server timeout", apierrors.NewServerTimeout(gr, "create", 5), provider.ErrTransient},
+		{"forbidden", apierrors.NewForbidden(gr, "x", errors.New("denied")), provider.ErrPermanent},
+		{"bad request", apierrors.NewBadRequest("bad"), provider.ErrPermanent},
+		{"unauthorized", apierrors.NewUnauthorized("no creds"), provider.ErrPermanent},
+		{"unclassified", errors.New("boom"), provider.ErrTransient},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			got := classify("op", "id", tc.err)
+			if !errors.Is(got, tc.want) {
+				t.Errorf("classify(%v) = %v, want class %v", tc.err, got, tc.want)
+			}
+		})
+	}
+}

internal/provider/kubernetes/pod.go

@@ -0,0 +1,67 @@
+package kubernetes
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
+)
+
+const proxyContainerName = "squid"
+
+// writeConfigAndExec is the container's entrypoint: write the config the
+// SQUID_CONF env var carries to disk, then exec squid against it. Avoids a
+// separate ConfigMap object per proxy instance — there's still only one
+// Kubernetes object (the Pod) to create, track, and clean up per proxy.
+const writeConfigAndExec = `printf '%s' "$SQUID_CONF" > /etc/squid/squid.conf && exec squid -N -f /etc/squid/squid.conf`
+
+// buildPod constructs the Pod for a proxy instance. Pure and side-effect
+// free, so it's unit-tested directly without a cluster — the same pattern
+// the GCP provider's buildInsertRequest uses.
+func buildPod(image string, req provider.CreateRequest) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      req.Name,
+			Namespace: req.Namespace,
+			Labels: map[string]string{
+				provider.LabelManaged: provider.LabelManagedYes,
+				provider.LabelUID:     req.UID,
+			},
+		},
+		Spec: corev1.PodSpec{
+			RestartPolicy: corev1.RestartPolicyAlways,
+			Containers: []corev1.Container{{
+				Name:    proxyContainerName,
+				Image:   image,
+				Command: []string{"/bin/sh", "-c"},
+				Args:    []string{writeConfigAndExec},
+				Env: []corev1.EnvVar{{
+					Name:  "SQUID_CONF",
+					Value: squidConf(req.Port),
+				}},
+				Ports: []corev1.ContainerPort{{
+					ContainerPort: req.Port,
+					Protocol:      corev1.ProtocolTCP,
+				}},
+			}},
+		},
+	}
+}
+
+// squidConf generates a minimal Squid config listening on port, permissive
+// enough to forward CONNECT and plain HTTP to any destination. Open by
+// design: this is a proxy for a private cluster, not internet-facing, and
+// installing/configuring proxy software is explicitly out of scope for
+// this operator's real (GCP) provider too — cloud-init is passed through
+// there, never interpreted. via/forwarded_for are turned off so the proxy
+// doesn't leak the Pod's identity to the origin.
+func squidConf(port int32) string {
+	return fmt.Sprintf(`http_port %d
+acl all src 0.0.0.0/0
+http_access allow all
+via off
+forwarded_for off
+`, port)
+}

internal/provider/kubernetes/pod_test.go

@@ -0,0 +1,91 @@
+package kubernetes
+
+import (
+	"strings"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
+)
+
+func TestBuildPod(t *testing.T) {
+	t.Parallel()
+	req := provider.CreateRequest{
+		Name:      "proxy-abc123",
+		UID:       "uid-1",
+		Namespace: "crawl",
+		ProxyName: "proxy-eu-1",
+		Port:      3128,
+	}
+	pod := buildPod("ubuntu/squid:6.6-24.04_edge", req)
+
+	if pod.Name != req.Name {
+		t.Errorf("pod.Name = %q, want %q", pod.Name, req.Name)
+	}
+	if pod.Namespace != req.Namespace {
+		t.Errorf("pod.Namespace = %q, want %q", pod.Namespace, req.Namespace)
+	}
+	if pod.Labels[provider.LabelManaged] != provider.LabelManagedYes {
+		t.Errorf("labels[%s] = %q, want %q", provider.LabelManaged, pod.Labels[provider.LabelManaged], provider.LabelManagedYes)
+	}
+	if pod.Labels[provider.LabelUID] != req.UID {
+		t.Errorf("labels[%s] = %q, want %q", provider.LabelUID, pod.Labels[provider.LabelUID], req.UID)
+	}
+	if pod.Spec.RestartPolicy != corev1.RestartPolicyAlways {
+		t.Errorf("RestartPolicy = %v, want Always", pod.Spec.RestartPolicy)
+	}
+	if len(pod.Spec.Containers) != 1 {
+		t.Fatalf("len(Containers) = %d, want 1", len(pod.Spec.Containers))
+	}
+	c := pod.Spec.Containers[0]
+	if c.Image != "ubuntu/squid:6.6-24.04_edge" {
+		t.Errorf("Image = %q, want ubuntu/squid:6.6-24.04_edge", c.Image)
+	}
+	if len(c.Ports) != 1 || c.Ports[0].ContainerPort != req.Port {
+		t.Errorf("Ports = %+v, want a single entry on port %d", c.Ports, req.Port)
+	}
+	var confEnv string
+	for _, e := range c.Env {
+		if e.Name == "SQUID_CONF" {
+			confEnv = e.Value
+		}
+	}
+	if !strings.Contains(confEnv, "http_port 3128") {
+		t.Errorf("SQUID_CONF env = %q, want it to contain %q", confEnv, "http_port 3128")
+	}
+}
+
+func TestBuildPod_usesRequestPort(t *testing.T) {
+	t.Parallel()
+	req := provider.CreateRequest{Name: "proxy-x", UID: "uid-2", Namespace: "ns", Port: 8080}
+	pod := buildPod("img", req)
+	conf := envValue(t, pod, "SQUID_CONF")
+	if !strings.Contains(conf, "http_port 8080") {
+		t.Errorf("SQUID_CONF = %q, want it to contain %q", conf, "http_port 8080")
+	}
+	if pod.Spec.Containers[0].Ports[0].ContainerPort != 8080 {
+		t.Errorf("ContainerPort = %d, want 8080", pod.Spec.Containers[0].Ports[0].ContainerPort)
+	}
+}
+
+func TestSquidConf_permissive(t *testing.T) {
+	t.Parallel()
+	conf := squidConf(3128)
+	for _, want := range []string{"http_port 3128", "http_access allow all", "via off", "forwarded_for off"} {
+		if !strings.Contains(conf, want) {
+			t.Errorf("squidConf() = %q, want it to contain %q", conf, want)
+		}
+	}
+}
+
+func envValue(t *testing.T, pod *corev1.Pod, name string) string {
+	t.Helper()
+	for _, e := range pod.Spec.Containers[0].Env {
+		if e.Name == name {
+			return e.Value
+		}
+	}
+	t.Fatalf("env var %q not found on container", name)
+	return ""
+}