kacerr/egress-proxies-operator
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

proxy-operator: Kubernetes operator for crawling-proxy fleets #1

Merged
kacerr merged 34 commits from feat/proxy-operator into main 2026-08-11 19:21:14 +02:00
Conversation 0 Commits 34 Files Changed 95 +110 -957
6 changed files with 110 additions and 957 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 4529594fb7 - Show all commits

8
.claude/settings.json
View File

@@ -45,7 +45,13 @@
"Bash(echo \"build: $?\")",
"Bash(echo \"vet: $?\")",
"Bash(perl -i -pe 's{^\\\\t\\\\t\\\\t\\\\t\\\\t// TODO\\\\\\(user\\\\\\): Specify other spec details if needed\\\\.\\\\n}{\\\\t\\\\t\\\\t\\\\t\\\\t// A minimal, schema-valid spec so this placeholder test survives the\\\\n\\\\t\\\\t\\\\t\\\\t\\\\t// CEL/CRD validation added in Step 1. Rewritten wholesale in Step 4\\\\n\\\\t\\\\t\\\\t\\\\t\\\\t// alongside the real reconciler and envtest suite.\\\\n\\\\t\\\\t\\\\t\\\\t\\\\tSpec: crawlv1alpha1.ProxySpec{\\\\n\\\\t\\\\t\\\\t\\\\t\\\\t\\\\tMode: crawlv1alpha1.ModeExternal,\\\\n\\\\t\\\\t\\\\t\\\\t\\\\t\\\\tEndpoint: &crawlv1alpha1.EndpointSpec{Host: \"10.0.0.1\"},\\\\n\\\\t\\\\t\\\\t\\\\t\\\\t},\\\\n}' internal/controller/proxy_controller_test.go)",
"Bash(go tool *)"
"Bash(go tool *)",
"Bash(docker version *)",
"Bash(curl -sI --max-time 5 https://hub.docker.com)",
"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\")"
],
"additionalDirectories": [
"/Users/jan.novak/srv/go/egress-proxies-operator/.claude",

150
docs/plans/2026-08-07-1747-proxy-operator.md
View File

@@ -22,7 +22,7 @@ spec change deletes and recreates the VM. No in-place update logic.
| API group | `crawl.example.com`, `v1alpha1`, kind `Proxy`, **namespaced** |
| Git flow | Scaffold commit, then work on `feat/proxy-operator`, MR via `tea`, no merge/delete from CLI |
| kubebuilder | `go install sigs.k8s.io/kubebuilder/v4/cmd/kubebuilder@v4.15.0` |
| Mock health | Mock provider runs a **real in-process HTTP CONNECT proxy** per instance, so healthchecks genuinely pass and the kind demo is truly end-to-end |
| Local/CI provider | **Kubernetes pod provider**, not an in-memory mock: creates real `ubuntu/squid` pods in-cluster. Revised mid-build — see Step 3 |
| Verification | vet + unit + envtest, then a throwaway kind cluster running the README quickstart, then delete it |
### Verified environment — no version substitutions needed
@@ -38,7 +38,7 @@ omitting the substitutions section.
### Milestone order — each ends green on `go build ./... && go vet ./...`
1. Scaffold + pins → 2. `api/v1alpha1` + CEL → 3. `internal/provider` + mock →
1. Scaffold + pins → 2. `api/v1alpha1` + CEL → 3. `internal/provider` + kubernetes-pod →
4. reconciler + envtest → 5. health engine → 6. lease + discovery → 7. GCP provider →
8. orphan GC + metrics → 9. `cmd/main.go` wiring + `config/` → 10. docs + kind run.
@@ -64,8 +64,9 @@ plan to `docs/plans/2026-08-07-1747-proxy-operator.md` per CLAUDE.md. **Commit t
untouched scaffold on its own** so every later diff is reviewable.
Post-scaffold hand-edits: `CONTROLLER_TOOLS_VERSION ?= v0.21.0` in the Makefile (CEL
emission at the 1.36 API level); add a `run-mock` target; delete the scaffolded
`.github/workflows/` (the remote is Gitea).
emission at the 1.36 API level); add a `run-dev` target wired to a sample
`--providers-config` (Step 10); delete the scaffolded `.github/workflows/` (the
remote is Gitea).
---
@@ -120,12 +121,15 @@ health, discovery, and the hash.
```
internal/provider/{provider,errors,name,config,metrics}.go
internal/provider/registry/registry.go # type→constructor — SEPARATE package
internal/provider/{mock,gcp}/
internal/provider/{kubernetes,gcp}/
```
**Import-cycle trap:** a registry inside `internal/provider` would have to import
`internal/provider/mock`, which imports `internal/provider`. `init()` self-registration
is banned by CLAUDE.md, so the registry goes in its own leaf-importing package.
`internal/provider/kubernetes` (or `.../gcp`), both of which import `internal/provider`
for the interface. `init()` self-registration is banned by CLAUDE.md, so the registry
goes in its own leaf-importing package, and its `Build` function takes the
type→constructor map as a parameter instead — see Step 2's execution log entry for why
this ended up better than a package-level map even beyond avoiding the cycle.
`Instance` needs **two fields the spec omits**, or orphan GC is unimplementable:
`UID string` (from the label, for the liveness match) and `CreatedAt time.Time` (for
@@ -157,23 +161,67 @@ same 80 bits. 80 bits → birthday collision at ~2^40 objects against a fleet of
---
## Step 3 — Mock provider (`internal/provider/mock/`)
## Step 3 — Kubernetes pod provider (`internal/provider/kubernetes/`)
**State is a pure function of an injectable clock — no background timers.** `Get`/
`ListByTag` derive state from `createdAt`/`deletedAt` vs `now()`:
`< provisionDelay` → Provisioning; else Running; `deletedAt` set and `< deleteDelay` →
Terminated; beyond that → purged, `ErrNotFound`. Deterministic under a fake clock,
correct under the real one, and no goroutine lifecycle to leak.
**Revised after Step 3 was first built as an in-memory mock provider** (state-machine
simulation + a hand-rolled CONNECT proxy on a shared, refcounted local listener). The
user found that too far from the real system to build confidence in, and didn't need
tests to be fast enough to justify the complexity it cost — a real `kind`-cluster
verification pass "once in a while" is an acceptable trade for tests that actually
look like the final product. Full narrative of the reversal is in
`docs/plans-executions/2026-08-07-1747-proxy-operator.md`; this section describes the
replacement, which is what actually gets built for local dev/CI going forward. No
in-memory provider remains in the tree — GCP is now the only other provider, per the
spec's original two-provider scope.
**Real proxy listener (the user's decision):** when a record first reports `Running`,
lazily start an `http.Server` on `127.0.0.1:0` implementing HTTP `CONNECT` tunnelling
(hijack + bidirectional `io.Copy`) plus plain-HTTP forwarding, and report `127.0.0.1`
plus the real listener port. `Delete` shuts it down. This is what makes the health
engine's probe a genuine CONNECT through a real proxy, so the kind quickstart actually
reaches Ready and leasable. One `http.Server` per instance, bounded by fleet size.
**What it does:** `Create` creates a `corev1.Pod` running a proxy container in the
same cluster (and same namespace as the owning Proxy CR — `req.Namespace`); `Get`
reads the Pod's phase/IP; `Delete` deletes it (tolerating NotFound); `ListByTag` lists
Pods by the standard `LabelManaged`/`LabelUID` labels, unscoped by namespace (the
operator's RBAC needs cluster-scoped Pod permissions — see RBAC note below).
Fault injection: config-driven `failNextCreates`/`failWith` for the demo, plus
`InjectCreateFailures(n int, class error)` for tests. `Create` is idempotent by name.
**Proxy software: `ubuntu/squid`** (Canonical's actively maintained LTS image on
Docker Hub, verified before picking it — 50M+ pulls, updated the same day this
decision was made), not a hand-rolled proxy. It's a public image, so `kind` nodes
pull it directly; no build/load step needed for the quickstart. Squid's config
(`http_port <req.Port>`, permissive ACL) is generated in Go and injected via an env
var the container's command writes to `/etc/squid/squid.conf` before exec'ing squid —
no separate ConfigMap object, so there's still only one Kubernetes object per proxy
instance to create, track, and clean up.
**providerID format: `<namespace>/<podName>`** (parseable with
`k8s.io/client-go/tools/cache.SplitMetaNamespaceKey`), so `Get`/`Delete` are
self-contained without needing to re-derive the namespace — the same reasoning as the
GCP provider's zone-qualified providerID in Step 8.
**Pod naming:** reuses `provider.NameFromUID` unchanged — the same deterministic name
satisfies Kubernetes Pod naming rules (`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$`, ≤253 chars)
with room to spare.
**State mapping:** Pod phase `Pending`, or `Running` with no PodIP yet → Provisioning
(never publish an empty IP); `Running` with a PodIP → Running; `Succeeded`/`Failed`/
`Unknown` → Terminated (the reconciler treats Stopped and Terminated identically —
delete and recreate, cattle not pets — so collapsing three failure-ish phases into one
is enough).
**Client:** built internally via `ctrl.GetConfig()` (auto-detects in-cluster config,
falls back to the local kubeconfig otherwise), not threaded through the registry
`Constructor` signature — this is 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`.
**Testing, given envtest can't schedule real Pods** (no kubelet — a Pod created
against `envtest`'s API server just sits `Pending` forever): `internal/provider/kubernetes`
itself is unit-tested against `sigs.k8s.io/controller-runtime/pkg/client/fake` — real
Pod objects, real client interface, fully exercises Create/Get/Delete/ListByTag
logic and the Pod-construction function in isolation, just without a kubelet actually
starting a container. Real end-to-end proof (does a probe actually tunnel through a
real Squid pod) only happens against a real `kind` cluster, in the Verification
section — which is exactly what the user asked for. Step 4's reconciler tests use a
small `Provider`-interface stub defined directly in the controller test file (a
handful of lines, not a package) for exercising the state-machine's branching logic —
categorically simpler than what mock.Provider was, since it has no config format, no
fault-injection surface, and exists only inside test code.
---
@@ -425,14 +473,17 @@ registry → manager → `mgr.Add` health engine, GC, lease expiry loop, discove
`SetupWithManager`. Contexts from `ctrl.SetupSignalHandler()` throughout.
RBAC markers: proxies CRUD + status + finalizers, secrets get/list/watch, events
create/patch. `config/`: providers ConfigMap mount, `DISCOVERY_TOKEN` from a Secret,
containerPort 8090 + Service. Samples: `proxy_mock.yaml`, `proxy_gcp.yaml`,
`proxy_external.yaml`, `providers-config.yaml`.
create/patch, and (for the kubernetes-pod provider) pods get/list/watch/create/delete
— cluster-scoped, since `ListByTag` enumerates across namespaces. `config/`: providers
ConfigMap mount, `DISCOVERY_TOKEN` from a Secret, containerPort 8090 + Service.
Samples: `proxy_kubernetes.yaml`, `proxy_gcp.yaml`, `proxy_external.yaml`,
`providers-config.yaml`.
`docs/architecture.md`: components table, ASCII data-flow diagram, and a **Decisions**
section covering the channel-vs-patch choice, replacement-polls-to-NotFound, base32
naming, discovery without leader election, single-mutex lease store, mock-runs-a-real-
proxy, leader-handover health seeding, the latency-suppression refinement, the
naming, discovery without leader election, single-mutex lease store, the mock→
kubernetes-pod-provider revision (why, and why `ubuntu/squid` over a hand-rolled
proxy), leader-handover health seeding, the latency-suppression refinement, the
`lastHealthCheckTime` semantics, and the logr-not-slog deviation (CLAUDE.md says
`slog`, but `log.FromContext(ctx)` returns logr inside controller paths — noted, not
silently ignored).
@@ -448,16 +499,17 @@ note confirming no substitutions were needed. Then a `CHANGELOG.md` entry with a
## Step 11 — Tests
**envtest** (`internal/controller/`), mock provider + a fake `HealthSnapshotter`,
intervals shrunk to 50–200 ms, whole suite behind `testing.Short()`:
Managed→Ready; spec change → old mock instance gone and providerID changed; delete →
finalizer runs and instance removed; External → Ready on first health pass, no
finalizer; injected quota → `Provisioned=False/QuotaExceeded` and phase *not* Failed;
permanent error → Failed and no further provider calls; adopt (strip annotation →
restored, providerID unchanged); and the **CEL cases only a real API server can test** —
mode/provider mutation rejected, Managed-without-provider, External-without-endpoint,
cloudInit both/neither, and `healthCheck` omitted → nested defaults materialized (the
`default={}` assertion).
**envtest** (`internal/controller/`), a small in-test stub `Provider` (not the real
kubernetes-pod provider — envtest has no kubelet, so a real Pod never leaves Pending)
plus a fake `HealthSnapshotter`, intervals shrunk to 50–200 ms, whole suite behind
`testing.Short()`: Managed→Ready; spec change → old stub instance gone and
providerID changed; delete → finalizer runs and instance removed; External → Ready
on first health pass, no finalizer; injected quota → `Provisioned=False/QuotaExceeded`
and phase *not* Failed; permanent error → Failed and no further provider calls; adopt
(strip annotation → restored, providerID unchanged); and the **CEL cases only a real
API server can test** — mode/provider mutation rejected, Managed-without-provider,
External-without-endpoint, cloudInit both/neither, and `healthCheck` omitted → nested
defaults materialized (the `default={}` assertion).
**Action-table unit tests** — the highest-value tests in the repo: `fake` client with
`WithStatusSubresource`, calling `Reconcile` directly, table-driven over every row of
@@ -466,12 +518,15 @@ client runs neither CEL nor defaulting — that's what the envtest CEL cases cov
**Units:** name derivation (idempotency, `^proxy-[a-z2-7]{16}$`, 10k-UID distinctness);
`Class()` mapping + `errors.Is`/`errors.As` through the multi-unwrap; config loading;
mock state at the delay boundary with a fake clock; `buildInsertRequest` field-by-field
+ GCP error classification + RUNNING-without-IP; `computePhase` truth table; `SpecHash`
stability *and* sensitivity; lease store (capacity, `MaxLeases=0`, least-loaded with
latency tie-break, cooldown with/without target, report on an expired-but-retained
lease, concurrent acquire under `-race` never exceeding `MaxLeases`); discovery
handlers over `httptest` + fake reader + real store; health thresholds against a real
kubernetes-pod provider Create/Get/Delete/ListByTag against
`sigs.k8s.io/controller-runtime/pkg/client/fake` (real Pod objects, real client
interface, no kubelet needed for this level) plus pure tests of the generated Squid
config and Pod spec; `buildInsertRequest` field-by-field + GCP error classification +
RUNNING-without-IP; `computePhase` truth table; `SpecHash` stability *and*
sensitivity; lease store (capacity, `MaxLeases=0`, least-loaded with latency
tie-break, cooldown with/without target, report on an expired-but-retained lease,
concurrent acquire under `-race` never exceeding `MaxLeases`); discovery handlers
over `httptest` + fake reader + real store; health thresholds against a real
CONNECT-capable `httptest` proxy stub.
Everything runs with `-race`.
@@ -485,19 +540,20 @@ go vet ./... && make test && make build # unit + envtest, -race
kind create cluster --name proxy-operator-demo
make install
make run-mock & # --providers-config hack/providers-mock.yaml
kubectl apply -f config/samples/proxy_mock.yaml
kubectl get px -w # expect Ready with an IP
make run-dev & # --providers-config hack/providers-dev.yaml
kubectl apply -f config/samples/proxy_kubernetes.yaml
kubectl get px -w # expect Ready with an IP (a real squid Pod)
curl -s 'localhost:8090/v1/proxies?healthy=true' | jq
curl -s -XPOST localhost:8090/v1/leases -d '{"selector":{"geo":"eu"},"ttlSeconds":300}' | jq
curl -s -XPOST localhost:8090/v1/leases/<id>/report -d '{"result":"rate_limited","target":"example.com"}'
curl -si -XDELETE localhost:8090/v1/leases/<id> # 204, and 204 again
kubectl delete -f config/samples/proxy_mock.yaml # finalizer runs, object goes
kubectl delete -f config/samples/proxy_kubernetes.yaml # finalizer runs, Pod is deleted
kind delete cluster --name proxy-operator-demo
```
Success bar: a competent SRE clones the repo, follows the README, and holds a lease on a
healthy mock proxy in under 10 minutes.
healthy proxy — a real Squid pod running in their own `kind` cluster — in under 10
minutes.
Then commit on `feat/proxy-operator`, push with `-u`, open the MR with
`tea pr create --base main --head feat/proxy-operator`, print the URL. No merging or

257
internal/provider/mock/mock.go
View File

@@ -1,257 +0,0 @@
// Package mock is an in-memory provider.Provider for local development and
// tests. State transitions are a pure function of an injectable clock, not
// background timers, so behavior is deterministic under a fake clock and
// there is no goroutine lifecycle for provisioning/deletion to leak.
//
// Once an instance's state resolves to Running, the provider starts (or
// joins) a real, minimal HTTP CONNECT proxy listener — see proxy.go — so
// that a through-the-proxy healthcheck (internal/health) genuinely
// succeeds against it. That's the difference between this being a true
// end-to-end local demo and one that quietly bypasses the operator's core
// mechanism.
package mock
import (
"context"
"fmt"
"sync"
"time"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
const (
defaultProvisionDelay = 5 * time.Second
defaultDeleteDelay = 1 * time.Second
)
// Provider is a thread-safe, in-memory provider.Provider.
type Provider struct {
mu sync.Mutex
name string
now func() time.Time
provisionDelay time.Duration
deleteDelay time.Duration
instances map[string]*record // keyed by instance name (== providerID)
failNext int
failClass error
}
type record struct {
uid string
port int32
createdAt time.Time
deletedAt time.Time // zero == not deleted
proxyAcquired bool
releaseProxy func()
}
// New builds a mock Provider from its config block. Satisfies
// registry.Constructor.
func New(_ context.Context, cfg provider.ProviderConfig) (provider.Provider, error) {
p := &Provider{
name: cfg.Name,
now: time.Now,
provisionDelay: defaultProvisionDelay,
deleteDelay: defaultDeleteDelay,
instances: make(map[string]*record),
}
if cfg.Mock == nil {
return p, nil
}
if cfg.Mock.ProvisionDelaySeconds > 0 {
p.provisionDelay = time.Duration(cfg.Mock.ProvisionDelaySeconds) * time.Second
}
if cfg.Mock.DeleteDelaySeconds > 0 {
p.deleteDelay = time.Duration(cfg.Mock.DeleteDelaySeconds) * time.Second
}
if cfg.Mock.FailNextCreates > 0 {
class, err := failClassFromString(cfg.Mock.FailWith)
if err != nil {
return nil, fmt.Errorf("mock provider %q: %w", cfg.Name, err)
}
p.failNext = cfg.Mock.FailNextCreates
p.failClass = class
}
return p, nil
}
func failClassFromString(s string) (error, error) {
switch s {
case "", provider.FailWithTransient:
return provider.ErrTransient, nil
case provider.FailWithNotFound:
return provider.ErrNotFound, nil
case provider.FailWithQuota:
return provider.ErrQuotaExceeded, nil
case provider.FailWithPermanent:
return provider.ErrPermanent, nil
default:
return nil, fmt.Errorf("unknown failWith %q", s)
}
}
// InjectCreateFailures makes the next n calls to Create fail, each
// returning an error classified as class. Exported for tests exercising
// the reconciler's error handling; config-driven FailNextCreates/FailWith
// (config.go) drives the same mechanism for demos.
func (p *Provider) InjectCreateFailures(n int, class error) {
p.mu.Lock()
defer p.mu.Unlock()
p.failNext = n
p.failClass = class
}
// Create is idempotent by req.Name: a repeat call for an existing,
// non-deleted instance returns its existing name rather than creating a
// duplicate. This is what lets the reconciler recover cleanly if it
// crashes between calling Create and persisting providerID — the next
// reconcile's Create call finds the same instance by its deterministic
// name.
func (p *Provider) Create(_ context.Context, req provider.CreateRequest) (string, error) {
p.mu.Lock()
defer p.mu.Unlock()
if existing, ok := p.instances[req.Name]; ok && existing.deletedAt.IsZero() {
return req.Name, nil
}
if p.failNext > 0 {
p.failNext--
return "", provider.Wrap(p.failClass, "create", p.name, req.Name, nil)
}
p.instances[req.Name] = &record{
uid: req.UID,
port: req.Port,
createdAt: p.now(),
}
return req.Name, nil
}
// Get returns the current state of a previously created instance,
// deriving it from the record's timestamps against the provider's clock.
// The first Get to observe a Running instance lazily acquires its real
// proxy listener.
func (p *Provider) Get(_ context.Context, providerID string) (*provider.Instance, error) {
p.mu.Lock()
defer p.mu.Unlock()
rec, ok := p.instances[providerID]
if !ok {
return nil, provider.Wrap(provider.ErrNotFound, "get", p.name, providerID, nil)
}
state, purge := p.stateLocked(rec)
if purge {
p.releaseLocked(rec)
delete(p.instances, providerID)
return nil, provider.Wrap(provider.ErrNotFound, "get", p.name, providerID, nil)
}
inst := &provider.Instance{
ID: providerID,
State: state,
UID: rec.uid,
CreatedAt: rec.createdAt,
}
if state == provider.StateRunning {
if !rec.proxyAcquired {
release, err := acquireProxy(rec.port)
if err != nil {
return nil, provider.Wrap(provider.ErrTransient, "get", p.name, providerID, err)
}
rec.releaseProxy = release
rec.proxyAcquired = true
}
inst.IP = mockProxyIP
}
return inst, nil
}
// Delete is idempotent: deleting an unknown or already-deleted instance is
// not an error. The real proxy listener (if any) is released immediately;
// the record itself lingers, reporting Terminated, until deleteDelay has
// passed and a later Get/ListByTag purges it — mirroring a real provider
// where the API stops accepting the ID before the resource fully vanishes.
func (p *Provider) Delete(_ context.Context, providerID string) error {
p.mu.Lock()
defer p.mu.Unlock()
rec, ok := p.instances[providerID]
if !ok {
return nil
}
if rec.deletedAt.IsZero() {
rec.deletedAt = p.now()
}
p.releaseLocked(rec)
return nil
}
// ListByTag returns every non-purged instance, mirroring what a real
// provider's tag/label-filtered list call would return. Also lazily purges
// (and releases) any instance whose deleteDelay has elapsed, since orphan
// GC — the only caller that runs regardless of whether anyone still calls
// Get on a given proxy — is what's responsible for eventually reclaiming
// deleted-and-expired instances in a real fleet.
func (p *Provider) ListByTag(_ context.Context) ([]provider.Instance, error) {
p.mu.Lock()
defer p.mu.Unlock()
var out []provider.Instance
for id, rec := range p.instances {
state, purge := p.stateLocked(rec)
if purge {
p.releaseLocked(rec)
delete(p.instances, id)
continue
}
inst := provider.Instance{
ID: id,
State: state,
UID: rec.uid,
CreatedAt: rec.createdAt,
}
if state == provider.StateRunning && rec.proxyAcquired {
inst.IP = mockProxyIP
}
out = append(out, inst)
}
return out, nil
}
// stateLocked derives state from rec's timestamps against p.now(). Must be
// called with p.mu held. purge is true once the instance has been deleted
// long enough that it should behave as fully gone (ErrNotFound to Get,
// absent from ListByTag).
func (p *Provider) stateLocked(rec *record) (state provider.InstanceState, purge bool) {
now := p.now()
if !rec.deletedAt.IsZero() {
if now.Sub(rec.deletedAt) < p.deleteDelay {
return provider.StateTerminated, false
}
return "", true
}
if now.Sub(rec.createdAt) < p.provisionDelay {
return provider.StateProvisioning, false
}
return provider.StateRunning, false
}
// releaseLocked releases rec's proxy listener reference, if it holds one.
// Must be called with p.mu held; acquireProxy/release use a separate lock
// (sharedProxies.mu), so this never deadlocks against it.
func (p *Provider) releaseLocked(rec *record) {
if rec.proxyAcquired {
rec.releaseProxy()
rec.proxyAcquired = false
rec.releaseProxy = nil
}
}

430
internal/provider/mock/mock_test.go
View File

@@ -1,430 +0,0 @@
package mock
import (
"context"
"errors"
"io"
"net"
"net/http"
"net/http/httptest"
"net/url"
"strconv"
"sync"
"sync/atomic"
"testing"
"time"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
// fakeClock lets tests drive Provider's internal state machine
// deterministically instead of racing real time.
type fakeClock struct {
mu sync.Mutex
now time.Time
}
func newFakeClock() *fakeClock {
return &fakeClock{now: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)}
}
func (c *fakeClock) Now() time.Time {
c.mu.Lock()
defer c.mu.Unlock()
return c.now
}
func (c *fakeClock) Advance(d time.Duration) {
c.mu.Lock()
defer c.mu.Unlock()
c.now = c.now.Add(d)
}
func newTestProvider(clock *fakeClock) *Provider {
return &Provider{
name: "test",
now: clock.Now,
provisionDelay: 5 * time.Second,
deleteDelay: 1 * time.Second,
instances: make(map[string]*record),
}
}
// testPortCounter hands out distinct ports across this package's test run.
// A "bind to :0, read back the port, close it" approach looks more
// realistic but is a TOCTOU race under t.Parallel(): two tests can be
// handed the same "free" port before either actually claims it, since
// nothing holds it open in between. What these tests actually need is a
// port no *other test in this run* will also try — a monotonic counter
// guarantees that outright, at the acceptable cost of a (much rarer) clash
// with an unrelated process already using something in this range.
var testPortCounter atomic.Int32
func freePort(t *testing.T) int32 {
t.Helper()
return 20000 + testPortCounter.Add(1)
}
func TestProvider_Create_idempotent(t *testing.T) {
t.Parallel()
p := newTestProvider(newFakeClock())
ctx := context.Background()
req := provider.CreateRequest{Name: "proxy-abc", UID: "uid-1", Port: freePort(t)}
id1, err := p.Create(ctx, req)
if err != nil {
t.Fatalf("Create() error = %v", err)
}
id2, err := p.Create(ctx, req)
if err != nil {
t.Fatalf("Create() (repeat) error = %v", err)
}
if id1 != id2 {
t.Errorf("Create() not idempotent: %q != %q", id1, id2)
}
instances, err := p.ListByTag(ctx)
if err != nil {
t.Fatalf("ListByTag() error = %v", err)
}
if len(instances) != 1 {
t.Errorf("len(instances) = %d, want 1 (repeat Create must not duplicate)", len(instances))
}
}
func TestProvider_Get_notFound(t *testing.T) {
t.Parallel()
p := newTestProvider(newFakeClock())
_, err := p.Get(context.Background(), "does-not-exist")
if !errors.Is(err, provider.ErrNotFound) {
t.Errorf("Get() error = %v, want ErrNotFound", err)
}
}
func TestProvider_stateTransitions(t *testing.T) {
t.Parallel()
clock := newFakeClock()
p := newTestProvider(clock)
ctx := context.Background()
port := freePort(t)
id, err := p.Create(ctx, provider.CreateRequest{Name: "proxy-xyz", UID: "uid-2", Port: port})
if err != nil {
t.Fatalf("Create() error = %v", err)
}
inst, err := p.Get(ctx, id)
if err != nil {
t.Fatalf("Get() error = %v", err)
}
if inst.State != provider.StateProvisioning {
t.Errorf("State = %v, want Provisioning immediately after Create", inst.State)
}
if inst.IP != "" {
t.Errorf("IP = %q, want empty while Provisioning", inst.IP)
}
clock.Advance(5*time.Second + time.Millisecond)
inst, err = p.Get(ctx, id)
if err != nil {
t.Fatalf("Get() error = %v", err)
}
if inst.State != provider.StateRunning {
t.Errorf("State = %v, want Running after provisionDelay", inst.State)
}
if inst.IP == "" {
t.Error("IP is empty, want a real address once Running")
}
if inst.UID != "uid-2" {
t.Errorf("UID = %q, want %q", inst.UID, "uid-2")
}
if err := p.Delete(ctx, id); err != nil {
t.Fatalf("Delete() error = %v", err)
}
inst, err = p.Get(ctx, id)
if err != nil {
t.Fatalf("Get() error = %v", err)
}
if inst.State != provider.StateTerminated {
t.Errorf("State = %v, want Terminated immediately after Delete", inst.State)
}
clock.Advance(time.Second + time.Millisecond)
_, err = p.Get(ctx, id)
if !errors.Is(err, provider.ErrNotFound) {
t.Errorf("Get() error = %v, want ErrNotFound after deleteDelay", err)
}
}
func TestProvider_Delete_idempotent(t *testing.T) {
t.Parallel()
p := newTestProvider(newFakeClock())
ctx := context.Background()
if err := p.Delete(ctx, "never-existed"); err != nil {
t.Errorf("Delete() on unknown ID error = %v, want nil", err)
}
id, _ := p.Create(ctx, provider.CreateRequest{Name: "proxy-del", UID: "uid-3", Port: freePort(t)})
if err := p.Delete(ctx, id); err != nil {
t.Fatalf("Delete() error = %v", err)
}
if err := p.Delete(ctx, id); err != nil {
t.Errorf("Delete() (repeat) error = %v, want nil", err)
}
}
func TestProvider_FaultInjection_configDriven(t *testing.T) {
t.Parallel()
ctx := context.Background()
prov, err := New(ctx, provider.ProviderConfig{
Name: "flaky",
Type: "mock",
Mock: &provider.MockConfig{FailNextCreates: 1, FailWith: provider.FailWithQuota},
})
if err != nil {
t.Fatalf("New() error = %v", err)
}
_, err = prov.Create(ctx, provider.CreateRequest{Name: "proxy-1", UID: "uid-4", Port: freePort(t)})
if !errors.Is(err, provider.ErrQuotaExceeded) {
t.Fatalf("first Create() error = %v, want ErrQuotaExceeded", err)
}
_, err = prov.Create(ctx, provider.CreateRequest{Name: "proxy-1", UID: "uid-4", Port: freePort(t)})
if err != nil {
t.Fatalf("second Create() error = %v, want nil (failure budget exhausted)", err)
}
}
func TestProvider_InjectCreateFailures(t *testing.T) {
t.Parallel()
p := newTestProvider(newFakeClock())
ctx := context.Background()
p.InjectCreateFailures(2, provider.ErrPermanent)
for i := range 2 {
_, err := p.Create(ctx, provider.CreateRequest{Name: "proxy-fail", UID: "uid-5", Port: freePort(t)})
if !errors.Is(err, provider.ErrPermanent) {
t.Fatalf("Create() #%d error = %v, want ErrPermanent", i, err)
}
}
_, err := p.Create(ctx, provider.CreateRequest{Name: "proxy-fail", UID: "uid-5", Port: freePort(t)})
if err != nil {
t.Fatalf("Create() after budget exhausted, error = %v, want nil", err)
}
}
func TestProvider_New_unknownFailWith(t *testing.T) {
t.Parallel()
_, err := New(context.Background(), provider.ProviderConfig{
Name: "bad",
Type: "mock",
Mock: &provider.MockConfig{FailNextCreates: 1, FailWith: "oops"},
})
if err == nil {
t.Fatal("New() error = nil, want error for unknown failWith")
}
}
func TestNew_appliesConfigOverrides(t *testing.T) {
t.Parallel()
prov, err := New(context.Background(), provider.ProviderConfig{
Name: "custom",
Type: "mock",
Mock: &provider.MockConfig{ProvisionDelaySeconds: 30, DeleteDelaySeconds: 10},
})
if err != nil {
t.Fatalf("New() error = %v", err)
}
p := prov.(*Provider)
if p.provisionDelay != 30*time.Second {
t.Errorf("provisionDelay = %v, want 30s", p.provisionDelay)
}
if p.deleteDelay != 10*time.Second {
t.Errorf("deleteDelay = %v, want 10s", p.deleteDelay)
}
}
func TestFailClassFromString(t *testing.T) {
t.Parallel()
tests := []struct {
in string
want error
wantErr bool
}{
{in: "", want: provider.ErrTransient},
{in: provider.FailWithTransient, want: provider.ErrTransient},
{in: provider.FailWithNotFound, want: provider.ErrNotFound},
{in: provider.FailWithQuota, want: provider.ErrQuotaExceeded},
{in: provider.FailWithPermanent, want: provider.ErrPermanent},
{in: "bogus", wantErr: true},
}
for _, tc := range tests {
t.Run(tc.in, func(t *testing.T) {
t.Parallel()
got, err := failClassFromString(tc.in)
if tc.wantErr {
if err == nil {
t.Fatal("failClassFromString() error = nil, want error")
}
return
}
if err != nil {
t.Fatalf("failClassFromString() error = %v, want nil", err)
}
if got != tc.want {
t.Errorf("failClassFromString(%q) = %v, want %v", tc.in, got, tc.want)
}
})
}
}
func TestProvider_ListByTag_includesRunningAndPurgesExpired(t *testing.T) {
t.Parallel()
clock := newFakeClock()
p := newTestProvider(clock)
ctx := context.Background()
running, err := p.Create(ctx, provider.CreateRequest{Name: "proxy-running", UID: "uid-running", Port: freePort(t)})
if err != nil {
t.Fatalf("Create() error = %v", err)
}
expiring, err := p.Create(ctx, provider.CreateRequest{Name: "proxy-expiring", UID: "uid-expiring", Port: freePort(t)})
if err != nil {
t.Fatalf("Create() error = %v", err)
}
clock.Advance(5*time.Second + time.Millisecond)
if _, err := p.Get(ctx, running); err != nil {
t.Fatalf("Get(running) error = %v", err)
}
if err := p.Delete(ctx, expiring); err != nil {
t.Fatalf("Delete(expiring) error = %v", err)
}
clock.Advance(time.Second + time.Millisecond) // past deleteDelay
instances, err := p.ListByTag(ctx)
if err != nil {
t.Fatalf("ListByTag() error = %v", err)
}
if len(instances) != 1 {
t.Fatalf("len(instances) = %d, want 1 (expired instance should be purged)", len(instances))
}
if instances[0].ID != running {
t.Errorf("instances[0].ID = %q, want %q", instances[0].ID, running)
}
if instances[0].State != provider.StateRunning {
t.Errorf("instances[0].State = %v, want Running", instances[0].State)
}
if instances[0].IP == "" {
t.Error("instances[0].IP is empty, want a real address for a Running instance")
}
if _, err := p.Get(ctx, expiring); !errors.Is(err, provider.ErrNotFound) {
t.Errorf("Get(expiring) error = %v, want ErrNotFound (ListByTag should have purged it)", err)
}
}
// TestProvider_realProxyForwardsPlainHTTP covers the non-CONNECT path: a
// probeURL override using plain http:// instead of the default https://
// should also work, going through handleForward rather than handleConnect.
func TestProvider_realProxyForwardsPlainHTTP(t *testing.T) {
t.Parallel()
origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
}))
defer origin.Close()
clock := newFakeClock()
p := newTestProvider(clock)
ctx := context.Background()
port := freePort(t)
id, err := p.Create(ctx, provider.CreateRequest{Name: "proxy-forward", UID: "uid-forward", Port: port})
if err != nil {
t.Fatalf("Create() error = %v", err)
}
clock.Advance(5*time.Second + time.Millisecond)
inst, err := p.Get(ctx, id)
if err != nil {
t.Fatalf("Get() error = %v", err)
}
proxyURL, err := url.Parse("http://" + net.JoinHostPort(inst.IP, strconv.Itoa(int(port))))
if err != nil {
t.Fatalf("parsing proxy URL: %v", err)
}
client := &http.Client{
Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)},
Timeout: 5 * time.Second,
}
resp, err := client.Get(origin.URL)
if err != nil {
t.Fatalf("GET through mock proxy failed: %v", err)
}
defer resp.Body.Close()
_, _ = io.Copy(io.Discard, resp.Body)
if resp.StatusCode != http.StatusOK {
t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusOK)
}
}
// TestProvider_realProxyTunnelsConnect is the load-bearing test for this
// package's whole reason to exist: once an instance is Running, a real
// http.Client using it as an HTTP proxy must genuinely tunnel a CONNECT
// request to a real origin — the exact mechanism internal/health depends
// on. This is not simulated; it opens real sockets.
func TestProvider_realProxyTunnelsConnect(t *testing.T) {
t.Parallel()
origin := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusNoContent)
}))
defer origin.Close()
clock := newFakeClock()
p := newTestProvider(clock)
ctx := context.Background()
port := freePort(t)
id, err := p.Create(ctx, provider.CreateRequest{Name: "proxy-e2e", UID: "uid-6", Port: port})
if err != nil {
t.Fatalf("Create() error = %v", err)
}
clock.Advance(5*time.Second + time.Millisecond)
inst, err := p.Get(ctx, id)
if err != nil {
t.Fatalf("Get() error = %v", err)
}
if inst.State != provider.StateRunning {
t.Fatalf("State = %v, want Running", inst.State)
}
proxyURL, err := url.Parse("http://" + net.JoinHostPort(inst.IP, strconv.Itoa(int(port))))
if err != nil {
t.Fatalf("parsing proxy URL: %v", err)
}
client := &http.Client{
Transport: &http.Transport{
Proxy: http.ProxyURL(proxyURL),
// origin is an httptest TLS server with a self-signed cert;
// this test is about the CONNECT tunnel, not certificate
// trust, so skip verification the same way httptest's own
// .Client() helper would.
TLSClientConfig: origin.Client().Transport.(*http.Transport).TLSClientConfig,
},
Timeout: 5 * time.Second,
}
resp, err := client.Get(origin.URL)
if err != nil {
t.Fatalf("GET through mock proxy failed: %v", err)
}
defer resp.Body.Close()
_, _ = io.Copy(io.Discard, resp.Body)
if resp.StatusCode != http.StatusNoContent {
t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusNoContent)
}
}

171
internal/provider/mock/proxy.go
View File

@@ -1,171 +0,0 @@
package mock
import (
"fmt"
"io"
"net"
"net/http"
"sync"
"time"
)
// mockProxyIP is the address every real listener binds to. Instances don't
// get distinct addresses (unlike a real cloud provider): only 127.0.0.1 is
// guaranteed bindable without elevated privileges across platforms — macOS
// does not, by default, route the rest of 127.0.0.0/8 the way Linux does.
// Instances sharing one port are told apart by which listener they share,
// not by IP.
const mockProxyIP = "127.0.0.1"
// connectProxy is a minimal HTTP proxy: it tunnels CONNECT requests
// (hijack + bidirectional copy) and forwards plain absolute-form HTTP
// requests. It exists so the health engine's through-the-proxy probe
// genuinely exercises a CONNECT tunnel against the mock provider, rather
// than the healthcheck being simulated or bypassed for local development.
type connectProxy struct {
ln net.Listener
sv *http.Server
}
func newConnectProxy(addr string) (*connectProxy, error) {
ln, err := net.Listen("tcp", addr)
if err != nil {
return nil, fmt.Errorf("mock proxy: listen %s: %w", addr, err)
}
sv := &http.Server{Handler: http.HandlerFunc(handleProxyRequest)}
go func() {
// Serve returns http.ErrServerClosed on a clean Close; there is no
// caller left to report anything else to by the time it returns.
_ = sv.Serve(ln)
}()
return &connectProxy{ln: ln, sv: sv}, nil
}
func (c *connectProxy) close() {
_ = c.sv.Close()
}
func handleProxyRequest(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodConnect {
handleConnect(w, r)
return
}
handleForward(w, r)
}
// handleConnect implements the CONNECT tunnel: dial the real destination,
// hijack the client connection, and splice the two together. This is the
// exact mechanism the health engine's default https:// probe URL depends
// on — a proxy that TCP-accepts but can't actually tunnel must fail here,
// not succeed.
func handleConnect(w http.ResponseWriter, r *http.Request) {
dst, err := net.DialTimeout("tcp", r.Host, 10*time.Second)
if err != nil {
http.Error(w, err.Error(), http.StatusBadGateway)
return
}
defer dst.Close()
hijacker, ok := w.(http.Hijacker)
if !ok {
http.Error(w, "hijack unsupported", http.StatusInternalServerError)
return
}
src, buf, err := hijacker.Hijack()
if err != nil {
return
}
defer src.Close()
if _, err := src.Write([]byte("HTTP/1.1 200 Connection Established\r\n\r\n")); err != nil {
return
}
// Any bytes the client already sent past the CONNECT request line
// before we hijacked are sitting in buf's reader; forward them before
// starting the raw splice loop.
if n := buf.Reader.Buffered(); n > 0 {
if _, err := io.CopyN(dst, buf.Reader, int64(n)); err != nil {
return
}
}
done := make(chan struct{}, 2)
go func() { io.Copy(dst, src); done <- struct{}{} }()
go func() { io.Copy(src, dst); done <- struct{}{} }()
<-done
}
// handleForward proxies a plain absolute-form HTTP request. CONNECT is the
// path the health engine's default probe exercises, but a probeURL
// override using plain http:// should work too.
func handleForward(w http.ResponseWriter, r *http.Request) {
outReq := r.Clone(r.Context())
outReq.RequestURI = ""
resp, err := http.DefaultTransport.RoundTrip(outReq)
if err != nil {
http.Error(w, err.Error(), http.StatusBadGateway)
return
}
defer resp.Body.Close()
for k, vv := range resp.Header {
for _, v := range vv {
w.Header().Add(k, v)
}
}
w.WriteHeader(resp.StatusCode)
_, _ = io.Copy(w, resp.Body)
}
// sharedProxies tracks one real listener per port, reference-counted
// across every mock.Provider in the process. This is package-level rather
// than a field on Provider because a bound TCP port is a process-global OS
// resource: two independently configured mock-typed provider entries (e.g.
// two named "mock" instances in providers-config.yaml) must not both try
// to bind 127.0.0.1:<port> — the second bind would simply fail. Sharing by
// port, refcounted, means any number of instances across any number of
// Provider values can use the same port safely, and the listener is torn
// down once nothing needs it anymore.
var sharedProxies = struct {
mu sync.Mutex
byPort map[int32]*sharedProxyEntry
}{byPort: make(map[int32]*sharedProxyEntry)}
type sharedProxyEntry struct {
proxy *connectProxy
refs int
}
// acquireProxy returns a release func for a real listener on mockProxyIP:
// port, starting one if this is the first acquire for that port. Safe to
// call concurrently; each returned release func must be called exactly
// once.
func acquireProxy(port int32) (release func(), err error) {
sharedProxies.mu.Lock()
defer sharedProxies.mu.Unlock()
entry, ok := sharedProxies.byPort[port]
if !ok {
p, err := newConnectProxy(fmt.Sprintf("%s:%d", mockProxyIP, port))
if err != nil {
return nil, err
}
entry = &sharedProxyEntry{proxy: p}
sharedProxies.byPort[port] = entry
}
entry.refs++
var once sync.Once
release = func() {
once.Do(func() {
sharedProxies.mu.Lock()
defer sharedProxies.mu.Unlock()
entry.refs--
if entry.refs <= 0 {
entry.proxy.close()
delete(sharedProxies.byPort, port)
}
})
}
return release, nil
}

51
internal/provider/mock/proxy_test.go
View File

@@ -1,51 +0,0 @@
package mock
import (
"net"
"strconv"
"testing"
)
func TestAcquireProxy_sharedAcrossAcquires(t *testing.T) {
t.Parallel()
port := freePort(t)
addr := net.JoinHostPort(mockProxyIP, strconv.Itoa(int(port)))
release1, err := acquireProxy(port)
if err != nil {
t.Fatalf("acquireProxy() #1 error = %v", err)
}
// A second acquire for the same port must join the existing listener
// rather than fail trying to bind it again — this is the whole point
// of sharing by port instead of by IP.
release2, err := acquireProxy(port)
if err != nil {
t.Fatalf("acquireProxy() #2 error = %v, want nil (should share the existing listener)", err)
}
release1()
// One reference remains; the port must still be in use.
if ln, err := net.Listen("tcp", addr); err == nil {
ln.Close()
t.Fatal("port became bindable after releasing only one of two references")
}
release2()
// Last reference released: the port must now be free.
ln, err := net.Listen("tcp", addr)
if err != nil {
t.Fatalf("port not released after last reference: %v", err)
}
ln.Close()
}
func TestAcquireProxy_releaseIsIdempotent(t *testing.T) {
t.Parallel()
port := freePort(t)
release, err := acquireProxy(port)
if err != nil {
t.Fatalf("acquireProxy() error = %v", err)
}
release()
release() // must not panic or double-decrement into a negative refcount
}