Author SHA1 Message Date
kacerrandClaude Opus 5 0bb699bf09 Fill the provider blueprint's gaps after a full sweep
A pass over the API types, spec hash, health and discovery call sites,
manager manifests, samples, README, provider docs, plans and the PR review
found rules the blueprint did not state: the IP must be reachable on the
proxy port by the health engine and clients, providers must be safe for
concurrent use, the manager's read-only/128Mi runtime envelope, catalogue
edits bypass the spec hash, Delete is re-called while in progress, and
provider names must be DNS-1123. Adds matrix rows, a deployment checklist
item, and the still-open review findings and broken samples as gaps.

README's "Adding a provider" restated an outdated contract; it now points
to docs/providers.md, and the providers table gains the kvm row.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-16 09:18:40 +00:00
kacerrandClaude Opus 5 01c27b5f35 Cover tracing in the provider blueprint
The blueprint reduced tracing to "don't open per-operation spans", which
reads as "nothing to do". Add an observability section: the operation
spans the decorators give for free, the child spans per backend call a
provider should add (parent-gated transport or a span decorator over the
SDK seam), the ctx/logger rules that keep traceID on log lines, and the
status per provider: kubernetes and gcp have child spans, kvm has none.
Matrix rows, the checklist item, gaps and a kvm proposal follow suit.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-16 09:14:06 +00:00
kacerrandClaude Opus 5 e08f23bcb3 Correct the provider blueprint after verification
A re-check against the code found four mismatches: the kvm connection is
opened at startup by the leader's first health check, so that isn't doc
drift; cmd/main.go, not the monitor, type-asserts Checker; QuotaExceeded
and PermanentError are reasons on Provisioned, not conditions; and the
latch line range. The checklist also now matches how the existing
providers do interface assertions and RBAC markers.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-16 08:57:52 +00:00
kacerrandClaude Opus 5 d96e56f1a0 Write the provider blueprint
docs/providers.md collects the provider contract that was spread across
comments, callers and per-provider docs: MUST/SHOULD/MAY rules per method,
each tied to the reconciler or GC behaviour it protects, the error taxonomy,
a feature matrix for kubernetes/gcp/kvm with file:line notes, a checklist for
adding a provider, and the known gaps. Linked from architecture.md and
CLAUDE.md.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-16 08:54:40 +00:00
kacerr cacd9583a3 Merge pull request 'Provider health checks: ProviderHealth CRD, per-provider checks and gauges' (#6) from feat/provider-health into main
Reviewed-on: #6
2026-09-16 10:46:47 +02:00
kacerrandClaude Opus 5 e6ed46fd7c Record the provider health validation and add the changelog entry
The user ran the real-hypervisor checks against beelink and confirmed they
passed, which completes Step 7 of the provider health plan.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-16 08:45:58 +00:00
kacerrandClaude Opus 5 80b2e9c85e Plan the provider blueprint document
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-16 08:41:42 +00:00
kacerrandClaude Opus 5 4e23347760 Point the KVM integration tests at a dedicated known_hosts
Leaving KVM_TEST_KNOWN_HOSTS unset silently disables host-key
verification for the run, so document it as part of the normal setup
rather than as an optional extra, and suggest a kvm_known_hosts holding
only the hypervisors.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-12 05:59:22 +00:00
kacerrandClaude Opus 5 4e5b5d62df Probe hosts for health instead of risking the shared connection
The health check shares each host's libvirt connection with the reconciles,
so it must not be able to take that connection away from them.

- A timed-out check no longer drops the link. The check goes through the new
  hostPool.probe, which abandons its RPCs when the context ends and leaves
  the connection alone, so an image upload or a Create running on the same
  link is not cut off. A link that really died is closed by the keepalive.
  One abandoned probe per host may stay waiting; until it returns, the next
  check reports Timeout at once (errProbeBusy).
- Waiting for a host's connection slot now ends with the caller's context.
  hostEntry used a mutex, so a check could wait out a reconcile redialing a
  dead host — minutes — past its own timeout. The slot is a one-slot channel
  and acquire bounds only the wait, never the uncontended take.
- --provider-check-timeout defaults to 0 and derives min(15s, interval).
  The flat 15s default contradicted the validation that the timeout may not
  exceed the interval, so --provider-check-interval=12s alone failed to
  start.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-12 05:04:45 +00:00
kacerrandClaude Opus 5 0e456affdd Record the automated validation of the provider health checks
make test (envtest, -race) and the kind e2e suite pass; the e2e
ProviderHealth for the kubernetes provider turned Ready within a second
of the manager starting and its gauge appeared in the scrape. make lint
is blocked by the local go1.27 toolchain crashing staticcheck on the
standard library, unrelated to this branch.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-11 20:31:15 +00:00
kacerrandClaude Opus 5 19b4a82ae9 Document the provider health checks
Architecture gets a provider-health section, component rows and four
decisions; the KVM provider doc gets a health-check section with the
reason table and the rotated-Secret gap; the testing guide lists the new
suites. CLAUDE.md's Architecture TODO and stale "only go.mod exists"
gotcha now point at the real docs, plus the RBAC-marker gotcha found
while wiring the monitor.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-11 20:28:09 +00:00
kacerrandClaude Opus 5 9a3ee963a0 Wire the provider health monitor into the manager, with gauges
--provider-check-interval (1m; 0 = off, else at least 10s) and
--provider-check-timeout (15s) are validated before the providers config
loads. Checkers are collected from the unwrapped providers in config
order, so checks never count as provider requests. Three scrape-time
gauges per {provider,target} read the monitor's snapshot: up, check
duration and check timestamp. The e2e suite waits for the kubernetes
ProviderHealth to turn Ready and asserts its gauge in the scrape.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-11 20:23:55 +00:00
kacerrandClaude Opus 5 88a15ecdb5 Add the provider health Monitor
A leader-only runnable checks every configured provider concurrently
under one timeout per round (first round at start, then every interval),
keeps the latest results for the metrics collector, logs targets going
down, changing reason or coming back, and publishes one ProviderHealth
object per provider: created with a managed-by label when missing, then
a status merge patch. Per-target lastTransitionTime survives rounds in
which ready does not flip. Objects of providers that are no longer
configured are deleted once at start; with checks turned off, all are.
Reads go through the API reader, so no cluster-wide informer is needed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-11 20:20:52 +00:00
kacerrandClaude Opus 5 04dd2c0284 Add the KVM health check: reachability and storage pool per host
Each configured host is checked concurrently over the pooled SSH+libvirt
connection: libvirt version and hostname prove the daemon answers, then
the configured pool must exist and be running, with its free space
reported. Dial failures are now tagged with their stage (SSH material,
TCP, SSH handshake, libvirt connect) by a dialError that keeps the
message and unwraps to the cause, so classification is unchanged and the
check can say why a host is unreachable. A pool verdict is decided after
the call returns so it never drops a working connection.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-11 20:17:35 +00:00
kacerrandClaude Opus 5 1962365f44 Add the provider Checker contract and the kubernetes and GCP checks
Checker is an optional interface next to Provider: Check returns one
verdict per target and never fails as a whole. The kubernetes provider
lists at most one managed Pod cluster-wide (the verb and scope GC relies
on); the GCP provider fetches a single page of at most one managed
instance through a new Probe seam method, since the SDK treats
MaxResults as a page size and the existing AggregatedList drains every
page.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-11 20:08:21 +00:00
kacerrandClaude Opus 5 1c773c24ad Add the cluster-scoped ProviderHealth CRD
One operator-owned object per configured provider will carry the
periodic check result: a Ready condition, per-target verdicts with their
own transition times, and a ready/total count for kubectl get. There is
no spec; only a viewer role is shipped since the operator is the sole
writer. Written by hand (kubebuilder is not installed), registered in
the existing AddKnownTypes call rather than a second init().

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-11 18:30:31 +00:00
kacerrandClaude Opus 5 18a3378507 Add plan: provider health checks
Periodic, leader-only checks of every configured provider (KVM per host
over the pooled SSH+libvirt connection including the storage pool,
kubernetes apiserver/RBAC, GCP credentials/API), surfaced through a new
cluster-scoped ProviderHealth CRD, Prometheus gauges and transition logs.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-11 18:08:59 +00:00
kacerr eb8f075f7a Merge pull request 'KVM/libvirt provider: proxies as VMs on your own hypervisors' (#5) from feat/kvm-provider into main
Reviewed-on: #5
2026-09-11 17:02:33 +02:00
46 changed files with 4987 additions and 80 deletions
+21
View File
@@ -1,5 +1,26 @@
# Changelog # Changelog
## 2026-09-16 08:45 UTC — Provider health checks: ProviderHealth CRD, per-provider checks, gauges
- **Each configured provider is now checked on a schedule** and reported in a
cluster-scoped `ProviderHealth` object (`kubectl get ph`), one per provider, with a
`Ready` condition, a reason, and per-target verdicts.
- **Checks:** kubernetes lists one object from the API server; GCP fetches a single
instance page; KVM checks SSH/libvirt reachability and the storage pool on every
host, in parallel. Failures map to fixed reasons (`Unreachable`,
`SSHHandshakeFailed`, `PoolNotFound`, `Timeout`, …).
- **The KVM check can't hurt reconciles:** it uses `hostPool.probe`, which gives up on
a slow RPC without dropping the shared connection, and waiting for a host slot can
be interrupted.
- **Metrics:** `proxy_operator_provider_target_up`, `_check_duration_seconds` and
`_check_timestamp_seconds`. The flags are `--provider-check-interval` and
`--provider-check-timeout` (defaults to `min(15s, interval)`).
- Validated with `make test`, `make test-e2e` and the real-hypervisor KVM run against
beelink. `make lint` is blocked by a Go 1.27 toolchain/staticcheck panic that has
nothing to do with this change.
- Key files: `api/v1alpha1/providerhealth_types.go`, `internal/provider/check.go`,
`internal/provider/kvm/`, `internal/providerhealth/`, `docs/architecture.md`.
## 2026-09-10 22:58 UTC — Code-review fixes on the KVM branch: deletion, validation, lazy SSH, guest agent ## 2026-09-10 22:58 UTC — Code-review fixes on the KVM branch: deletion, validation, lazy SSH, guest agent
Five findings from a review of `feat/kvm-provider`, all fixed on the branch. Five findings from a review of `feat/kvm-provider`, all fixed on the branch.
+12 -8
View File
@@ -25,10 +25,12 @@ entrypoint yet, so there's nothing to run locally.
## Architecture ## Architecture
Not yet built. TODO: once the controller-runtime manager and reconcile loop exist, Documented in [docs/architecture.md](docs/architecture.md). It holds the component
document the data-flow diagram and package/role table here, following the standard table, the reconcile flow, and one section per runnable: health engine, discovery and
reconcile-loop shape: fetch → resolve → list related → compute desired state → merge lease API, orphan GC, metrics, tracing, and provider health checks. The provider
→ patch only on diff. contract (MUST/SHOULD rules, feature matrix, checklist for adding a provider, known
gaps) is in [docs/providers.md](docs/providers.md). Provider-specific detail is in
`docs/kvm-provider.md` and the GCP docs; the test layout is in `docs/testing.md`.
## Conventions ## Conventions
@@ -214,7 +216,9 @@ A lightweight `check` job (`go vet` / `go build` / `go test -short`) gates the b
## Gotchas ## Gotchas
- This repo currently contains only `go.mod` (module - The "What this is" and "Commands" sections above predate the code and are stale.
`gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator`) — no controller code, The Makefile targets (`make test`, `make manifests generate`, `make test-e2e`,
`main.go`, Dockerfile, k8s manifests, Makefile, or CI config exist yet. Nothing in `make test-integration-kvm`) and `docs/testing.md` are authoritative.
this file describes working software; it's the convention scaffold to build against. - `+kubebuilder:rbac` markers are package-level. Keep them in their own comment block,
never inside a type's or function's doc comment: controller-gen silently ignores
them there. After `make manifests`, check `git diff config/rbac/role.yaml`.
+7
View File
@@ -18,4 +18,11 @@ resources:
kind: Proxy kind: Proxy
path: gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1 path: gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1
version: v1alpha1 version: v1alpha1
- api:
crdVersion: v1
domain: example.com
group: crawl
kind: ProviderHealth
path: gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1
version: v1alpha1
version: "3" version: "3"
+9 -38
View File
@@ -104,7 +104,8 @@ make deploy IMG=<registry>/egress-proxies-operator:dev
- Provider config comes from the `providers-config` ConfigMap - Provider config comes from the `providers-config` ConfigMap
([config/manager/providers_config.yaml](config/manager/providers_config.yaml)); ([config/manager/providers_config.yaml](config/manager/providers_config.yaml));
the default ships only the kubernetes provider. the default ships the kubernetes provider plus a `kvm-home` entry that needs
the `kvm-ssh` Secret (without it the manager starts and warns).
- The discovery API is exposed by the - The discovery API is exposed by the
`controller-manager-discovery-service` Service on port 8090. `controller-manager-discovery-service` Service on port 8090.
- Auth: create the token Secret, or the API serves **unauthenticated** - Auth: create the token Secret, or the API serves **unauthenticated**
@@ -185,46 +186,16 @@ can be two differently-configured instances of the same `gcp` type (see
| --- | --- | --- | --- | | --- | --- | --- | --- |
| `kubernetes` | A real Squid pod (`ubuntu/squid:6.6-24.04_edge` by default) in the same cluster the operator runs in | `image` (optional) | Needs no cloud account — local dev, CI, and the kind quickstart. Pods share the cluster's egress IP, so it exercises the full lifecycle but not distinct egress paths. | | `kubernetes` | A real Squid pod (`ubuntu/squid:6.6-24.04_edge` by default) in the same cluster the operator runs in | `image` (optional) | Needs no cloud account — local dev, CI, and the kind quickstart. Pods share the cluster's egress IP, so it exercises the full lifecycle but not distinct egress paths. |
| `gcp` | A Compute Engine VM with an ephemeral external IP | `project` (required), `network`, `networkTag`, `diskSizeGb` | The real egress fleet. Deliberately uses only four API calls (Insert / Get / Delete / AggregatedList), all fire-and-forget: `Create` returns as soon as the operation is submitted and the reconciler discovers progress by polling `Get`. Auth is Application Default Credentials — workload identity in-cluster, `gcloud` ADC locally; no key-file plumbing. | | `gcp` | A Compute Engine VM with an ephemeral external IP | `project` (required), `network`, `networkTag`, `diskSizeGb` | The real egress fleet. Deliberately uses only four API calls (Insert / Get / Delete / AggregatedList), all fire-and-forget: `Create` returns as soon as the operation is submitted and the reconciler discovers progress by polling `Get`. Auth is Application Default Credentials — workload identity in-cluster, `gcloud` ADC locally; no key-file plumbing. |
| `kvm` | A libvirt VM on one of your own hypervisors, reached over SSH | `sshKeyFile`, `knownHostsFile` (unless `insecureIgnoreHostKey`), `machineTypes`, `hosts` (required); `images`, `diskSizeGb`, `vendorData`, `installGuestAgent` | Home-lab or on-prem egress. The base image is downloaded into each host's pool on first use; `Create` waits for the VM to start. See [docs/kvm-provider.md](docs/kvm-provider.md). |
### Adding a provider ### Adding a provider
A new backend (Hetzner, AWS, ...) is four pieces; the contract lives in The contract, the rules each caller depends on, a feature matrix of the
[internal/provider/provider.go](internal/provider/provider.go): existing providers and a step-by-step checklist are in
[docs/providers.md](docs/providers.md). Start there: several rules (a
1. **Implement the 4-method `Provider` interface** in a new self-contained `providerID`, `Get` returning `ErrNotFound` only when the
`internal/provider/<type>/` package: instance is really gone, `CreatedAt` read back for orphan GC) are enforced by
- `Create` submits and returns — it never blocks until the VM runs, and callers, not by the interface.
must be idempotent keyed on `req.Name` (a deterministic name derived
from the Proxy's UID), so a repeat call after a crash finds the
existing instance instead of duplicating it.
- `Get` returns `provider.ErrNotFound` as a *normal* outcome — the
reconciler branches on it for replacement and adoption, so don't
treat it as exceptional.
- `Delete` is idempotent: deleting an already-gone instance is not an
error.
- `ListByTag` returns every instance the operator ever tagged, for
orphan GC.
2. **Tag every created resource** with `LabelManaged=true` and
`LabelUID=<Proxy UID>`, and report `CreatedAt` — orphan GC relies on
all three to find owned resources and skip in-flight creates.
3. **Classify every returned error** with `provider.Wrap` into the
four-sentinel taxonomy in
[internal/provider/errors.go](internal/provider/errors.go)
(`ErrNotFound` / `ErrQuotaExceeded` / `ErrTransient` / `ErrPermanent`)
— the reconciler decides retry, slow backoff, or latching `Failed`
purely from that classification, never from provider-specific types.
4. **Wire it up**: add a type-specific config block in
[internal/provider/config.go](internal/provider/config.go), and
register the constructor in the builtins map in
[cmd/main.go](cmd/main.go) (`"<type>": <pkg>.New`). The registry
([internal/provider/registry](internal/provider/registry/registry.go))
handles named instances, and the metrics wrapper is applied
automatically.
Test against a fake API seam rather than the real cloud — see the
`instancesAPI` seam in
[internal/provider/gcp/gcp.go](internal/provider/gcp/gcp.go) for the
pattern.
## Tracing ## Tracing
+127
View File
@@ -0,0 +1,127 @@
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package v1alpha1
import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// ConditionReady summarizes a ProviderHealth's targets: True when every
// target passed its last check.
const ConditionReady = "Ready"
// ProviderTarget is the latest check verdict for one thing a provider
// depends on: a KVM hypervisor, the Kubernetes API server, a GCP project.
type ProviderTarget struct {
// name identifies the target within its provider: the KVM host name,
// "apiserver", or the GCP project.
// +kubebuilder:validation:MinLength=1
// +required
Name string `json:"name"`
// address is where the check connected to.
// +optional
Address string `json:"address,omitempty"`
// ready is true when the last check succeeded.
// +required
Ready bool `json:"ready"`
// reason is a CamelCase verdict such as Reachable, Unreachable,
// Forbidden, SSHHandshakeFailed or PoolNotActive.
// +optional
Reason string `json:"reason,omitempty"`
// message is human-readable detail, usually the error of a failed check.
// +kubebuilder:validation:MaxLength=1024
// +optional
Message string `json:"message,omitempty"`
// latencyMillis is how long the last check took.
// +optional
LatencyMillis int64 `json:"latencyMillis,omitempty"`
// lastTransitionTime is when ready last changed.
// +required
LastTransitionTime metav1.Time `json:"lastTransitionTime"`
// details are provider-specific facts from the last check, e.g. the
// libvirt version and free storage-pool space of a KVM host.
// +optional
Details map[string]string `json:"details,omitempty"`
}
// ProviderHealthStatus is the result of the latest check of one provider.
type ProviderHealthStatus struct {
// type is the provider's type from the providers config: kubernetes,
// gcp or kvm.
// +optional
Type string `json:"type,omitempty"`
// conditions hold Ready, which summarizes the targets.
// +listType=map
// +listMapKey=type
// +optional
Conditions []metav1.Condition `json:"conditions,omitempty"`
// targets hold one verdict per checked target.
// +listType=map
// +listMapKey=name
// +optional
Targets []ProviderTarget `json:"targets,omitempty"`
// targetsReady counts the ready targets as "ready/total".
// +optional
TargetsReady string `json:"targetsReady,omitempty"`
// lastCheckTime is when the latest check finished.
// +optional
LastCheckTime *metav1.Time `json:"lastCheckTime,omitempty"`
}
// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:resource:scope=Cluster,shortName=ph
// +kubebuilder:printcolumn:name="Type",type=string,JSONPath=`.status.type`
// +kubebuilder:printcolumn:name="Ready",type=string,JSONPath=`.status.conditions[?(@.type=="Ready")].status`
// +kubebuilder:printcolumn:name="Targets",type=string,JSONPath=`.status.targetsReady`
// +kubebuilder:printcolumn:name="Last Check",type=date,JSONPath=`.status.lastCheckTime`
// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
// ProviderHealth reports the periodic health check of one configured
// provider. It is named after the provider, and the operator creates,
// updates and deletes it itself; there is no spec to edit.
type ProviderHealth struct {
metav1.TypeMeta `json:",inline"`
// metadata is a standard object metadata
// +optional
metav1.ObjectMeta `json:"metadata,omitzero"`
// status is the result of the latest check
// +optional
Status ProviderHealthStatus `json:"status,omitzero"`
}
// +kubebuilder:object:root=true
// ProviderHealthList contains a list of ProviderHealth
type ProviderHealthList struct {
metav1.TypeMeta `json:",inline"`
metav1.ListMeta `json:"metadata,omitzero"`
Items []ProviderHealth `json:"items"`
}
+1 -1
View File
@@ -361,7 +361,7 @@ type ProxyList struct {
func init() { func init() {
SchemeBuilder.Register(func(s *runtime.Scheme) error { SchemeBuilder.Register(func(s *runtime.Scheme) error {
s.AddKnownTypes(SchemeGroupVersion, &Proxy{}, &ProxyList{}) s.AddKnownTypes(SchemeGroupVersion, &Proxy{}, &ProxyList{}, &ProviderHealth{}, &ProviderHealthList{})
return nil return nil
}) })
} }
+114
View File
@@ -95,6 +95,120 @@ func (in *PlacementSpec) DeepCopy() *PlacementSpec {
return out return out
} }
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ProviderHealth) DeepCopyInto(out *ProviderHealth) {
*out = *in
out.TypeMeta = in.TypeMeta
in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
in.Status.DeepCopyInto(&out.Status)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ProviderHealth.
func (in *ProviderHealth) DeepCopy() *ProviderHealth {
if in == nil {
return nil
}
out := new(ProviderHealth)
in.DeepCopyInto(out)
return out
}
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
func (in *ProviderHealth) DeepCopyObject() runtime.Object {
if c := in.DeepCopy(); c != nil {
return c
}
return nil
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ProviderHealthList) DeepCopyInto(out *ProviderHealthList) {
*out = *in
out.TypeMeta = in.TypeMeta
in.ListMeta.DeepCopyInto(&out.ListMeta)
if in.Items != nil {
in, out := &in.Items, &out.Items
*out = make([]ProviderHealth, len(*in))
for i := range *in {
(*in)[i].DeepCopyInto(&(*out)[i])
}
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ProviderHealthList.
func (in *ProviderHealthList) DeepCopy() *ProviderHealthList {
if in == nil {
return nil
}
out := new(ProviderHealthList)
in.DeepCopyInto(out)
return out
}
// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
func (in *ProviderHealthList) DeepCopyObject() runtime.Object {
if c := in.DeepCopy(); c != nil {
return c
}
return nil
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ProviderHealthStatus) DeepCopyInto(out *ProviderHealthStatus) {
*out = *in
if in.Conditions != nil {
in, out := &in.Conditions, &out.Conditions
*out = make([]v1.Condition, len(*in))
for i := range *in {
(*in)[i].DeepCopyInto(&(*out)[i])
}
}
if in.Targets != nil {
in, out := &in.Targets, &out.Targets
*out = make([]ProviderTarget, len(*in))
for i := range *in {
(*in)[i].DeepCopyInto(&(*out)[i])
}
}
if in.LastCheckTime != nil {
in, out := &in.LastCheckTime, &out.LastCheckTime
*out = (*in).DeepCopy()
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ProviderHealthStatus.
func (in *ProviderHealthStatus) DeepCopy() *ProviderHealthStatus {
if in == nil {
return nil
}
out := new(ProviderHealthStatus)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ProviderTarget) DeepCopyInto(out *ProviderTarget) {
*out = *in
in.LastTransitionTime.DeepCopyInto(&out.LastTransitionTime)
if in.Details != nil {
in, out := &in.Details, &out.Details
*out = make(map[string]string, len(*in))
for key, val := range *in {
(*out)[key] = val
}
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ProviderTarget.
func (in *ProviderTarget) DeepCopy() *ProviderTarget {
if in == nil {
return nil
}
out := new(ProviderTarget)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *Proxy) DeepCopyInto(out *Proxy) { func (in *Proxy) DeepCopyInto(out *Proxy) {
*out = *in *out = *in
+59 -2
View File
@@ -16,8 +16,8 @@ limitations under the License.
// Package main is the composition root: it loads the provider config (fail // Package main is the composition root: it loads the provider config (fail
// fast), assembles the provider registry, and wires the reconciler, health // fast), assembles the provider registry, and wires the reconciler, health
// engine, lease store, discovery API, orphan GC, and metrics onto one // engine, lease store, discovery API, orphan GC, provider health checks and
// controller-runtime manager. // metrics onto one controller-runtime manager.
package main package main
import ( import (
@@ -60,6 +60,7 @@ import (
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kubernetes" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kubernetes"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kvm" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kvm"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/registry" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/registry"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/providerhealth"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/version" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/version"
// +kubebuilder:scaffold:imports // +kubebuilder:scaffold:imports
@@ -97,6 +98,7 @@ func main() {
var showVersion bool var showVersion bool
var gcpWireFullPayloads bool var gcpWireFullPayloads bool
var traceHealthProbes bool var traceHealthProbes bool
var providerCheckInterval, providerCheckTimeout time.Duration
flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+ flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
"Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.") "Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.")
@@ -140,6 +142,13 @@ func main() {
flag.BoolVar(&traceHealthProbes, "trace-health-probes", false, flag.BoolVar(&traceHealthProbes, "trace-health-probes", false,
"Emit one trace span per health probe. Off by default: probes run about "+ "Emit one trace span per health probe. Off by default: probes run about "+
"once per second per proxy and would dominate trace volume.") "once per second per proxy and would dominate trace volume.")
flag.DurationVar(&providerCheckInterval, "provider-check-interval", time.Minute,
"Interval between provider health checks (can the cluster reach each hypervisor, API "+
"and account), published as ProviderHealth objects and metrics. 0 turns them off; "+
"otherwise at least 10s.")
flag.DurationVar(&providerCheckTimeout, "provider-check-timeout", 0,
"Time limit for one round of provider health checks: 15s, or the interval when that is "+
"shorter. If set, at most the interval.")
opts := zap.Options{ opts := zap.Options{
Development: true, Development: true,
@@ -174,6 +183,10 @@ func main() {
setupLog.Error(nil, "--providers-config is required") setupLog.Error(nil, "--providers-config is required")
os.Exit(1) os.Exit(1)
} }
if err := providerhealth.ValidateSchedule(providerCheckInterval, providerCheckTimeout); err != nil {
setupLog.Error(err, "Invalid provider check flags")
os.Exit(1)
}
cfg, err := provider.LoadConfigFile(providersConfig) cfg, err := provider.LoadConfigFile(providersConfig)
if err != nil { if err != nil {
setupLog.Error(err, "Failed to load providers config", "path", providersConfig) setupLog.Error(err, "Failed to load providers config", "path", providersConfig)
@@ -195,6 +208,14 @@ func main() {
os.Exit(1) os.Exit(1)
} }
m := metrics.New() m := metrics.New()
// Checkers come from the unwrapped providers, in config order: the
// metrics and tracing wrappers implement only Provider, and a check is
// not a provider request to be counted anyway.
checkTargets := make([]providerhealth.Target, 0, len(cfg.Providers))
for _, pc := range cfg.Providers {
checker, _ := providers[pc.Name].(provider.Checker)
checkTargets = append(checkTargets, providerhealth.Target{Name: pc.Name, Type: pc.Type, Checker: checker})
}
for name, p := range providers { for name, p := range providers {
// Tracing outermost: the span covers the metrics recording too. // Tracing outermost: the span covers the metrics recording too.
providers[name] = provider.WithTracing(name, provider.WithMetrics(name, p, m)) providers[name] = provider.WithTracing(name, provider.WithMetrics(name, p, m))
@@ -314,6 +335,18 @@ func main() {
os.Exit(1) os.Exit(1)
} }
monitor := &providerhealth.Monitor{
Client: mgr.GetClient(),
Reader: mgr.GetAPIReader(),
Targets: checkTargets,
Interval: providerCheckInterval,
Timeout: providerCheckTimeout,
}
if err := mgr.Add(monitor); err != nil {
setupLog.Error(err, "Failed to add provider health monitor")
os.Exit(1)
}
if err := m.Register(ctrlmetrics.Registry, if err := m.Register(ctrlmetrics.Registry,
proxyPhaseCounts(mgr.GetClient()), proxyPhaseCounts(mgr.GetClient()),
func() int { func() int {
@@ -327,6 +360,10 @@ func main() {
setupLog.Error(err, "Failed to register metrics") setupLog.Error(err, "Failed to register metrics")
os.Exit(1) os.Exit(1)
} }
if err := metrics.RegisterProviderTargets(ctrlmetrics.Registry, providerTargetSamples(monitor)); err != nil {
setupLog.Error(err, "Failed to register provider check metrics")
os.Exit(1)
}
if err := (&controller.ProxyReconciler{ if err := (&controller.ProxyReconciler{
Client: mgr.GetClient(), Client: mgr.GetClient(),
@@ -389,3 +426,23 @@ func proxyPhaseCounts(c client.Reader) func() map[string]int {
return counts return counts
} }
} }
// providerTargetSamples flattens the monitor's latest results for the
// scrape-time provider-check gauges.
func providerTargetSamples(mon *providerhealth.Monitor) func() []metrics.ProviderTarget {
return func() []metrics.ProviderTarget {
var out []metrics.ProviderTarget
for _, r := range mon.Snapshot() {
for _, ts := range r.Targets {
out = append(out, metrics.ProviderTarget{
Provider: r.Provider,
Target: ts.Name,
Up: ts.Ready,
Duration: ts.Latency,
CheckedAt: r.CheckedAt,
})
}
}
return out
}
}
@@ -0,0 +1,193 @@
---
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
annotations:
controller-gen.kubebuilder.io/version: v0.21.0
name: providerhealths.crawl.example.com
spec:
group: crawl.example.com
names:
kind: ProviderHealth
listKind: ProviderHealthList
plural: providerhealths
shortNames:
- ph
singular: providerhealth
scope: Cluster
versions:
- additionalPrinterColumns:
- jsonPath: .status.type
name: Type
type: string
- jsonPath: .status.conditions[?(@.type=="Ready")].status
name: Ready
type: string
- jsonPath: .status.targetsReady
name: Targets
type: string
- jsonPath: .status.lastCheckTime
name: Last Check
type: date
- jsonPath: .metadata.creationTimestamp
name: Age
type: date
name: v1alpha1
schema:
openAPIV3Schema:
description: |-
ProviderHealth reports the periodic health check of one configured
provider. It is named after the provider, and the operator creates,
updates and deletes it itself; there is no spec to edit.
properties:
apiVersion:
description: |-
APIVersion defines the versioned schema of this representation of an object.
Servers should convert recognized schemas to the latest internal value, and
may reject unrecognized values.
More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources
type: string
kind:
description: |-
Kind is a string value representing the REST resource this object represents.
Servers may infer this from the endpoint the client submits requests to.
Cannot be updated.
In CamelCase.
More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds
type: string
metadata:
type: object
status:
description: status is the result of the latest check
properties:
conditions:
description: conditions hold Ready, which summarizes the targets.
items:
description: Condition contains details for one aspect of the current
state of this API Resource.
properties:
lastTransitionTime:
description: |-
lastTransitionTime is the last time the condition transitioned from one status to another.
This should be when the underlying condition changed. If that is not known, then using the time when the API field changed is acceptable.
format: date-time
type: string
message:
description: |-
message is a human readable message indicating details about the transition.
This may be an empty string.
maxLength: 32768
type: string
observedGeneration:
description: |-
observedGeneration represents the .metadata.generation that the condition was set based upon.
For instance, if .metadata.generation is currently 12, but the .status.conditions[x].observedGeneration is 9, the condition is out of date
with respect to the current state of the instance.
format: int64
minimum: 0
type: integer
reason:
description: |-
reason contains a programmatic identifier indicating the reason for the condition's last transition.
Producers of specific condition types may define expected values and meanings for this field,
and whether the values are considered a guaranteed API.
The value should be a CamelCase string.
This field may not be empty.
maxLength: 1024
minLength: 1
pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$
type: string
status:
description: status of the condition, one of True, False, Unknown.
enum:
- "True"
- "False"
- Unknown
type: string
type:
description: type of condition in CamelCase or in foo.example.com/CamelCase.
maxLength: 316
pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$
type: string
required:
- lastTransitionTime
- message
- reason
- status
- type
type: object
type: array
x-kubernetes-list-map-keys:
- type
x-kubernetes-list-type: map
lastCheckTime:
description: lastCheckTime is when the latest check finished.
format: date-time
type: string
targets:
description: targets hold one verdict per checked target.
items:
description: |-
ProviderTarget is the latest check verdict for one thing a provider
depends on: a KVM hypervisor, the Kubernetes API server, a GCP project.
properties:
address:
description: address is where the check connected to.
type: string
details:
additionalProperties:
type: string
description: |-
details are provider-specific facts from the last check, e.g. the
libvirt version and free storage-pool space of a KVM host.
type: object
lastTransitionTime:
description: lastTransitionTime is when ready last changed.
format: date-time
type: string
latencyMillis:
description: latencyMillis is how long the last check took.
format: int64
type: integer
message:
description: message is human-readable detail, usually the error
of a failed check.
maxLength: 1024
type: string
name:
description: |-
name identifies the target within its provider: the KVM host name,
"apiserver", or the GCP project.
minLength: 1
type: string
ready:
description: ready is true when the last check succeeded.
type: boolean
reason:
description: |-
reason is a CamelCase verdict such as Reachable, Unreachable,
Forbidden, SSHHandshakeFailed or PoolNotActive.
type: string
required:
- lastTransitionTime
- name
- ready
type: object
type: array
x-kubernetes-list-map-keys:
- name
x-kubernetes-list-type: map
targetsReady:
description: targetsReady counts the ready targets as "ready/total".
type: string
type:
description: |-
type is the provider's type from the providers config: kubernetes,
gcp or kvm.
type: string
type: object
type: object
served: true
storage: true
subresources:
status: {}
+1
View File
@@ -3,4 +3,5 @@
# It should be run by config/default # It should be run by config/default
resources: resources:
- bases/crawl.example.com_proxies.yaml - bases/crawl.example.com_proxies.yaml
- bases/crawl.example.com_providerhealths.yaml
# +kubebuilder:scaffold:crdkustomizeresource # +kubebuilder:scaffold:crdkustomizeresource
+4
View File
@@ -66,6 +66,10 @@ spec:
- --providers-config=/etc/proxy-operator/providers.yaml - --providers-config=/etc/proxy-operator/providers.yaml
# Add --trace-health-probes to emit one span per health probe # Add --trace-health-probes to emit one span per health probe
# (high volume: ~1/s per proxy). # (high volume: ~1/s per proxy).
# Provider health checks (kubectl get providerhealth) run every
# minute; tune with --provider-check-interval (0 turns them off,
# otherwise at least 10s) and --provider-check-timeout (15s, or
# the interval if that is shorter).
env: env:
# Bearer token for the discovery API. Optional: without the # Bearer token for the discovery API. Optional: without the
# Secret the API serves unauthenticated (with a loud warning). # Secret the API serves unauthenticated (with a loud warning).
+3
View File
@@ -25,4 +25,7 @@ resources:
- proxy_admin_role.yaml - proxy_admin_role.yaml
- proxy_editor_role.yaml - proxy_editor_role.yaml
- proxy_viewer_role.yaml - proxy_viewer_role.yaml
# ProviderHealth objects are written only by the operator, so they get a
# viewer role and nothing else.
- providerhealth_viewer_role.yaml
@@ -0,0 +1,29 @@
# This rule is not used by the project egress-proxies-operator itself.
# It is provided to allow the cluster admin to help manage permissions for users.
#
# Grants read-only access to ProviderHealth objects, the results of the
# operator's periodic provider checks. The operator creates, updates and
# deletes them itself, so no admin or editor role is provided.
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
labels:
app.kubernetes.io/name: egress-proxies-operator
app.kubernetes.io/managed-by: kustomize
name: providerhealth-viewer-role
rules:
- apiGroups:
- crawl.example.com
resources:
- providerhealths
verbs:
- get
- list
- watch
- apiGroups:
- crawl.example.com
resources:
- providerhealths/status
verbs:
- get
+16
View File
@@ -22,6 +22,22 @@ rules:
- get - get
- list - list
- watch - watch
- apiGroups:
- crawl.example.com
resources:
- providerhealths
verbs:
- create
- delete
- get
- list
- apiGroups:
- crawl.example.com
resources:
- providerhealths/status
verbs:
- get
- patch
- apiGroups: - apiGroups:
- crawl.example.com - crawl.example.com
resources: resources:
+88 -3
View File
@@ -4,9 +4,9 @@
| Component | Package | Runs as | Leader-elected | Role | | Component | Package | Runs as | Leader-elected | Role |
|---|---|---|---|---| |---|---|---|---|---|
| Proxy CRD + helpers | `api/v1alpha1` | types | — | `Proxy` spec/status, CEL validation, defaulting, pure helpers | | Proxy + ProviderHealth CRDs | `api/v1alpha1` | types | — | `Proxy` spec/status, CEL validation, defaulting, pure helpers; the status-only, cluster-scoped `ProviderHealth` |
| Reconciler | `internal/controller` | controller | yes (with the manager) | the state machine: provision, replace, delete, represent health | | Reconciler | `internal/controller` | controller | yes (with the manager) | the state machine: provision, replace, delete, represent health |
| Provider contract | `internal/provider` | library | — | `Provider` interface, error taxonomy, deterministic naming, config, metrics decorator | | Provider contract | `internal/provider` | library | — | `Provider` interface, optional `Checker`, error taxonomy, deterministic naming, config, metrics decorator; rules and per-provider matrix in [providers.md](providers.md) |
| kubernetes provider | `internal/provider/kubernetes` | library | — | real Squid pods in this cluster (local dev/CI) | | kubernetes provider | `internal/provider/kubernetes` | library | — | real Squid pods in this cluster (local dev/CI) |
| gcp provider | `internal/provider/gcp` | library | — | Compute Engine VMs, four API calls, fire-and-forget ops | | gcp provider | `internal/provider/gcp` | library | — | Compute Engine VMs, four API calls, fire-and-forget ops |
| kvm provider | `internal/provider/kvm` | library | — | VMs on libvirt hosts over SSH-forwarded sockets; image cache, overlay disks, NoCloud seeds | | kvm provider | `internal/provider/kvm` | library | — | VMs on libvirt hosts over SSH-forwarded sockets; image cache, overlay disks, NoCloud seeds |
@@ -14,6 +14,7 @@
| Lease store | `internal/lease` | Runnable (expiry sweep) | no | in-memory leases + cooldowns, single mutex | | Lease store | `internal/lease` | Runnable (expiry sweep) | no | in-memory leases + cooldowns, single mutex |
| Discovery API | `internal/discovery` | Runnable | no | HTTP list/lease/release/report on `:8090` | | Discovery API | `internal/discovery` | Runnable | no | HTTP list/lease/release/report on `:8090` |
| Orphan GC | `internal/gc` | Runnable | yes | deletes tagged instances whose CR is gone | | Orphan GC | `internal/gc` | Runnable | yes | deletes tagged instances whose CR is gone |
| Provider health monitor | `internal/providerhealth` | Runnable | yes | periodic provider checks → one `ProviderHealth` per provider, target gauges, transition logs |
| Metrics | `internal/metrics` | library | — | explicit registration, scrape-time collectors | | Metrics | `internal/metrics` | library | — | explicit registration, scrape-time collectors |
| Composition root | `cmd/main.go` | binary | — | flags, provider registry, wires everything onto one manager | | Composition root | `cmd/main.go` | binary | — | flags, provider registry, wires everything onto one manager |
@@ -306,6 +307,8 @@ registries). Two kinds:
- **Scrape-time collectors** — `proxy_operator_proxies{phase}` and - **Scrape-time collectors** — `proxy_operator_proxies{phase}` and
`proxy_operator_leases_active` read the cache / lease store at every `proxy_operator_leases_active` read the cache / lease store at every
scrape; reconcile-incremented gauges inevitably drift and leak series. scrape; reconcile-incremented gauges inevitably drift and leak series.
The provider-check gauges (§11) read the health monitor's snapshot the
same way.
- **Fed vectors** — `healthcheck_duration_seconds{proxy}` and - **Fed vectors** — `healthcheck_duration_seconds{proxy}` and
`healthcheck_failures_total{proxy}` observe EVERY probe (status writes `healthcheck_failures_total{proxy}` observe EVERY probe (status writes
are transition-only; metrics carry the high-frequency signal), and the are transition-only; metrics carry the high-frequency signal), and the
@@ -348,7 +351,8 @@ incoming trace context):
- Discovery API: one server span per request (named from the route - Discovery API: one server span per request (named from the route
pattern), incoming W3C `traceparent` honored so clients' traces continue pattern), incoming W3C `traceparent` honored so clients' traces continue
into the operator; `/healthz` excluded. into the operator; `/healthz` excluded.
- `gc.sweep` per GC pass; `health.probe` per probe **only** with - `gc.sweep` per GC pass; `providerhealth.check` per round of provider
checks; `health.probe` per probe **only** with
`--trace-health-probes` (default off — probes run ~1/s per proxy). `--trace-health-probes` (default off — probes run ~1/s per proxy).
Log correlation: logr sinks never see a context, so trace IDs ride on the Log correlation: logr sinks never see a context, so trace IDs ride on the
@@ -359,6 +363,63 @@ Re-deriving from the base rather than layering keeps zap from emitting
duplicate keys on nested spans. GCP V(5) wire logs get the same keys from duplicate keys on nested spans. GCP V(5) wire logs get the same keys from
the slog handler's ctx. the slog handler's ctx.
### 11. Provider health checks (`internal/providerhealth/`)
Answers "can this cluster reach each provider's backend right now?"
without waiting for a Proxy to fail. A provider opts in by implementing
`provider.Checker`: `Check(ctx) []TargetStatus` returns one verdict per
target and never fails as a whole. All three providers do:
| Provider | Target | What one check does |
|---|---|---|
| kvm | each `kvm.hosts` entry | over the pooled SSH+libvirt connection: libvirt version and hostname, then the configured pool must exist and be running; reports pool capacity and free space ([details](kvm-provider.md#health-check)) |
| kubernetes | `apiserver` | lists at most one managed Pod cluster-wide: the verb and scope `ListByTag` and GC rely on |
| gcp | the project | one page of an aggregated instance list with `maxResults=1`: credentials, Compute API enabled, `compute.instances.list` |
```text
Monitor.Start(ctx) ── leader-only; --provider-check-interval (1m; 0 = off,
│ otherwise ≥ 10s), --provider-check-timeout (15s, or
│ the interval if that is shorter)
├─ once: delete managed-by-labelled ProviderHealth objects whose provider
│ is not configured (all of them when checks are off) — the providers
│ config is read once per process
└─ at start, then every Interval: checkOnce
│ every provider's Check concurrently, under one context.WithTimeout
│ snapshot → the scrape-time gauges
│ log transitions: down or reason changed → "WARNING: provider target
│ not ready"; back up → "Provider target ready"; steady → silent
└ per provider: APIReader.Get → Create if missing → status merge patch
```
- **One status write per provider per round**, since `lastCheckTime`
moves every round. At the 10s minimum interval that is still far below
leader-election renewals. Proxy status avoids writes like this only
because its probes run once a second per proxy.
- **Reads go through the API reader**, not the cache. The objects are
cluster-scoped; a cached Get would start a cluster-wide informer (and
need `watch`) even with `--proxy-namespace`.
- **Checks bypass the provider decorators.** `cmd/main.go` collects the
Checkers before wrapping the providers in `WithTracing(WithMetrics(...))`,
so a check never counts as a provider request. KVM RPCs still appear in
the V(1) `libvirt call` trace, and each round is one
`providerhealth.check` span.
- **Not wired to `/readyz`.** A dead hypervisor must not make the manager
unready while the other providers keep working.
- **Gauges** `proxy_operator_provider_target_up`,
`proxy_operator_provider_check_duration_seconds` and
`proxy_operator_provider_check_timestamp_seconds`, each labelled
`{provider,target}`. Only the leader exports them.
Illustrative output:
```sh
$ kubectl get providerhealth # short name: ph
NAME TYPE READY TARGETS LAST CHECK AGE
kubernetes kubernetes True 1/1 12s 3d
kvm-home kvm False 0/1 12s 3d
$ kubectl get ph kvm-home -o yaml # per target: reason, message, details
```
## Decisions ## Decisions
Judgment calls the spec left open, and deliberate deviations — recorded so Judgment calls the spec left open, and deliberate deviations — recorded so
@@ -528,3 +589,27 @@ they read as choices, not accidents. Chronological by build step.
external sites. The kube-apiserver ignores incoming `traceparent` by external sites. The kube-apiserver ignores incoming `traceparent` by
design (public endpoint), so k8s client spans are leaves — in-process design (public endpoint), so k8s client spans are leaves — in-process
traces, not cross-process ones. traces, not cross-process ones.
- **Provider health is a status-only, cluster-scoped CRD** (user
decision). `kubectl get providerhealth` works without Prometheus, which
the deployment does not ship (its ServiceMonitor is commented out). There
is no spec: the providers config file stays the one place providers are
configured, and the object only reports on it. The gauges carry the same
verdicts for alerting.
- **KVM checks reuse the pooled connection instead of dialing fresh.** A
fresh dial would not pick up a rotated Secret anyway, since the SSH
material is cached. It would also open an sshd session per host per
minute. And a working cached connection is exactly what reconciles use.
Sharing must not cost the reconciles anything, so the check calls
through `hostPool.probe`. A check that runs out of time abandons its
RPCs instead of closing the connection, which would kill an upload or a
Create running on it. At most one abandoned check per host is left
waiting, and until it returns, the next check reports `Timeout` at once.
Waiting for a host's connection slot also ends with the caller's
deadline, so a reconcile redialing a dead host cannot stall the round.
- **Provider transitions are logs, not Events**, in line with the
EventRecorder decision above. They log at Info with a `WARNING:` prefix,
and each target's `lastTransitionTime` in the object records when it
flipped.
- **A provider name that is not a valid object name** gets metrics but no
`ProviderHealth` object, with a warning at start. Tightening config
validation instead would crash-loop existing deployments on upgrade.
+57 -2
View File
@@ -265,12 +265,65 @@ The provider logs through controller-runtime's logger, named `kvm`, with
only `length` and `duration`. There is no raw wire tier like the GCP only `length` and `duration`. There is no raw wire tier like the GCP
provider's V(5); go-libvirt exposes no hook for it. provider's V(5); go-libvirt exposes no hook for it.
## Health check
The operator checks every configured host every `--provider-check-interval`
(default 1m). It publishes the result in two places: the `ProviderHealth`
object named after the provider (`kubectl get ph kvm-home -o yaml`), and the
`proxy_operator_provider_target_up{provider,target}` gauge.
All hosts are checked concurrently, each over the same pooled connection
that Create and Get use:
1. The SSH and libvirt handshakes. These only happen on the first check, or
after the link dropped.
2. `virConnectGetLibVersion` and `virConnectGetHostname`.
3. `virStoragePoolLookupByName` and `virStoragePoolGetInfo` on the host's
`pool`.
A passing target's `details` carry `hostname`, `libvirtVersion`, `pool`,
`poolState`, `poolCapacity` and `poolAvailable` (in GiB). The free space is
the closest thing this provider has to a capacity signal. Every target also
shows its SSH `user`. A failing target says where the check broke:
| Reason | Meaning | Look at |
|---|---|---|
| `SSHMaterialUnavailable` | the key or known_hosts file is missing or does not parse | the `kvm-ssh` Secret (see *Cluster-side setup*) |
| `Unreachable` | the TCP connect failed, or the link died mid-check | DNS, Tailscale or a firewall between the cluster and `address` |
| `SSHHandshakeFailed` | the host key is missing from known_hosts or has changed, or `user` rejected the key | the known_hosts entry; `user`'s `authorized_keys` |
| `LibvirtConnectFailed` | SSH worked, the libvirt socket did not | `user` in the `libvirt` group, `libvirtd` running, the `socket` path |
| `Timeout` | no answer within `--provider-check-timeout` (default 15s), or the previous check of the host is still waiting | a hung `libvirtd`, a saturated link, or a reconcile stuck redialing the host |
| `PoolNotFound` | there is no pool named `pool` | `virsh pool-list --all` |
| `PoolNotActive` | the pool exists but is not running | `virsh pool-start <pool>` and `virsh pool-autostart <pool>` |
| `LibvirtError` | any other libvirt error | the message carries libvirt's own text |
Sharing the connection has two consequences:
- The leader now keeps one SSH connection per host open from startup, with a
keepalive every 15s. Before, the first connection opened with the first KVM
reconcile.
- A check that runs out of time gives up on its RPCs but leaves the
connection alone, so an image upload or a Create running on it is not cut
off. At most one abandoned check per host keeps waiting; until it
returns, the next check reports `Timeout` straight away. A link that is
really dead is closed by the keepalive within about two keepalive
intervals.
- A check never waits longer than its timeout for the host's connection,
even while a reconcile holds it to redial a host that is down.
## Known gaps ## Known gaps
- **No quota or capacity detection.** libvirt reports an out-of-space - **No quota or capacity detection.** libvirt reports an out-of-space
pool or an over-committed host as an ordinary error; the reconciler pool or an over-committed host as an ordinary error; the reconciler
treats it as transient and retries with backoff rather than the slower treats it as transient and retries with backoff rather than the slower
quota backoff. quota backoff. The health check does report each pool's free space
(`poolAvailable`), so a pool that is filling up shows before Creates
start failing.
- **A rotated `kvm-ssh` Secret needs a manager restart.** The key and
known_hosts files are read on first use and cached for the life of the
process. Replacing them in the Secret therefore changes nothing for
later connections, health checks included. After a host key change, the
check keeps reporting `SSHHandshakeFailed` until the manager restarts.
- **Orphan volumes without a domain are invisible to GC.** A crash - **Orphan volumes without a domain are invisible to GC.** A crash
between creating the volumes and defining the domain leaves an overlay between creating the volumes and defining the domain leaves an overlay
and seed that `ListByTag` cannot see (it lists domains). They are and seed that `ListByTag` cannot see (it lists domains). They are
@@ -315,4 +368,6 @@ Unit tests run over an in-memory fake of the libvirt API. The optional
integration test against a real host is described in integration test against a real host is described in
[testing.md](testing.md#the-kvm-integration-test-real-hypervisor); its [testing.md](testing.md#the-kvm-integration-test-real-hypervisor); its
`-v` output includes the `libvirt call` trace above, at the verbosity `-v` output includes the `libvirt call` trace above, at the verbosity
`KVM_TEST_LOG_LEVEL` selects (default 1). `KVM_TEST_LOG_LEVEL` selects (default 1). `TestKVMCheck_realHost` runs the
health check against the same host and creates nothing:
`go test -tags=integration -count=1 -v -run TestKVMCheck ./internal/provider/kvm/`.
@@ -0,0 +1,378 @@
# Execution: provider health checks
Plan: [docs/plans/2026-09-11-1526-provider-health-checks.md](../plans/2026-09-11-1526-provider-health-checks.md)
- [x] Step 0 — Save the plan
- [x] Step 1 — ProviderHealth API and CRD
- [x] Step 2 — Contract plus the kubernetes and GCP checks
- [x] Step 3 — KVM check
- [x] Step 4 — Monitor
- [x] Step 5 — Metrics and wiring
- [x] Step 6 — Docs
- [x] Step 7 — Validate, record, push, open the PR
## Step 0 — Save the plan
Branched off `main` (at the PR #5 merge, `eb8f075`). The plan was copied out of
`~/.claude/plans/` and committed on its own, before any code, as the repo's plan rules
require:
```bash
git checkout -b feat/provider-health main
cp ~/.claude/plans/now-i-would-like-flickering-wall.md docs/plans/2026-09-11-1526-provider-health-checks.md
```
The uncommitted `.claude/settings.json` change was left unstaged and stays out of this
branch.
## Step 1 — ProviderHealth API and CRD
`kubebuilder` isn't installed here (only `bin/controller-gen`, `bin/kustomize` and
`bin/setup-envtest` are), so `api/v1alpha1/providerhealth_types.go` and the `PROJECT`
entry were written by hand instead of with `kubebuilder create api`. For a cluster-scoped
resource `PROJECT` simply omits `namespaced`, which is what kubebuilder itself writes.
The new types are registered in the existing `AddKnownTypes` call in `proxy_types.go`,
not in a second `init()`.
The deepcopy code, the CRD and the RBAC come from the usual generators:
```bash
make generate manifests
go build ./... && go vet ./api/... && go test -short ./api/...
```
To confirm the new CRD and the viewer role actually reach the rendered install, I built
the kustomization and grepped for them:
```bash
bin/kustomize build config/default | grep -n providerhealth
# providerhealths.crawl.example.com (scope: Cluster, shortName ph)
# egress-proxies-operator-providerhealth-viewer-role
```
Worth noting:
- The only role added is the viewer role. The operator is the object's only writer, so
the admin and editor roles kubebuilder scaffolds for Proxy were not added.
- The type file has no unit test, since it has no logic. The schema, including the
required `ready`/`lastTransitionTime` and the list-map keys, is proven by the envtest
round-trip in Step 4.
- `config/rbac/role.yaml` is unchanged so far. The manager's own permissions arrive in
Step 4 with the Monitor's RBAC markers.
## Step 2 — Contract plus the kubernetes and GCP checks
`provider.Checker` is an optional interface in `internal/provider/check.go`, kept out of
`Provider`. Each implementation has a compile-time assertion, because the monitor's
type assertion would otherwise fail silently.
- **kubernetes:** lists at most one managed Pod cluster-wide, using the same client, verb
and scope as `ListByTag`. The target is `apiserver`, at the `restCfg.Host` now stored
in the new `apiHost` field.
- **GCP:** calls a new seam method, `Probe`. `realInstances.Probe` lives in `check.go`,
next to its only caller. `ListByTag`'s `Sprintf` filter became the shared
`managedFilter` constant, so the check and the GC sweep can't drift apart.
Before relying on "MaxResults is only a page size", I checked the SDK's REST
`AggregatedList` in the module cache:
```bash
grep -n "func (c \*instancesRESTClient) AggregatedList" -A80 \
"$(go list -m -f '{{.Dir}}' cloud.google.com/go/compute)"/apiv1/instances_client.go
# InternalFetch: req.MaxResults = pageSize ... return elems, resp.GetNextPageToken(), nil
```
```bash
go build ./... && go vet ./internal/provider/... && go test -race -count=1 ./internal/provider/...
```
Worth noting:
- The planned `TestRealInstances_Probe_fetchesOnePage` became a three-case
`TestRealInstances_Probe` table.
- It keeps the first page's `nextPageToken`, and asserts exactly one request with
`maxResults=1` for each case.
- "empty project" covers `iterator.Done`.
- "api error keeps its status" proves the real client returns a `*googleapi.Error`, so
a disabled Compute API is reported as `Forbidden` rather than `Unreachable`.
- Once a context has ended, both checks report `Timeout` whatever the error says. The
monitor's timeout is the usual cause, and the underlying error only repeats it.
## Step 3 — KVM check
`conn.go` now wraps each of the four dial stages in a `dialError`: ssh-material, tcp,
ssh-handshake and libvirt-connect. `Error()` returns the inner message unchanged and
`Unwrap()` returns the cause, so `classify`/`isClassified` and every existing
message-matching test behave as before. `TestDialError_keepsClassification` pins this
down.
The three new RPCs went into `libvirtAPI`, `tracingAPI` and `fakeAPI`. The method-count
guard in `TestTracingAPI_logsEveryMethod` fails until a traced row exists for each new
method, which is what forced the three new rows.
`Check` runs one goroutine per host, over the shared connection, through
`hostPool.call`. The pool verdict is decided only after `call` returns, because
returning it as an error from inside the callback would make `call` drop a perfectly
good connection.
Build, vet (also with the integration tag, since that file changed) and race tests:
```bash
go build ./internal/provider/... && go vet ./internal/provider/kvm/ \
&& go vet -tags=integration ./internal/provider/kvm/ \
&& go test -race -count=1 ./internal/provider/kvm/
```
Worth noting:
- **Pool state wording:** `poolState` uses virsh's words (`running`, `inactive`, …)
rather than go-libvirt's constant names, so it reads like `virsh pool-list` output.
- **Details:** each host also reports its SSH `user`. That is the first thing to check
when the reason is `SSHHandshakeFailed`.
- **Tests renamed or merged:**
- The planned `TestSSHAuth_dial_stages` is `TestLazySSH_dial_tagsTheFailingStage`. It
dials through `lazySSH`, so the ssh-material stage is covered as well as the three
stages the in-process SSH server can trigger.
- The pool-inactive and pool-missing cases are merged into `TestCheck_libvirtVerdicts`,
which also covers a daemon error. Each case runs two checks and asserts a single
dial.
- **Tests added beyond the plan:**
- `TestCheck_timeout`: a daemon that stops answering gives `Timeout`.
- `TestDialError_keepsClassification`.
- **Not run yet:** `TestKVMCheck_realHost` in the integration file only compiles here.
It needs beelink, so it goes on the Step 7 checklist for the user to run.
## Step 4 — Monitor
`internal/providerhealth` has the `Monitor` runnable plus the pure `buildStatus`.
- **Scheduling:** the Monitor is leader-only. It cleans up once at start, checks
immediately, then checks on a ticker.
- **Each round:** providers are checked concurrently under one timeout, then logged and
published one after another. A failed write is logged, and the next round writes
again.
- **Reads:** objects are read through the reader it is given. `main.go` passes the
manager's API reader (Step 5).
The first `make manifests` left `config/rbac/role.yaml` unchanged, with no error.
I had put the `+kubebuilder:rbac` markers inside `Monitor`'s doc comment, and
controller-gen only reads RBAC markers from package-level comment groups. So they were
silently attached to the type and ignored. An empty `git diff config/rbac/role.yaml` was
the only sign. Moving them into their own comment block fixed it:
```bash
make manifests && git diff config/rbac/role.yaml
# + providerhealths: create, delete, get, list
# + providerhealths/status: get, patch
```
Vet, race tests and the envtest round-trip (using the `bin/k8s/1.36.2-linux-amd64`
binaries `make setup-envtest` had already fetched):
```bash
go vet ./internal/providerhealth/ && go test -race -count=1 -v ./internal/providerhealth/
# --- PASS: TestMonitor_envtestRoundTrip (6.45s) ... ok
```
Worth noting:
- **Fake client scope:** the fake client needs a RESTMapper that registers
ProviderHealth as `RESTScopeRoot`. Its default mapper treats every kind as namespaced.
- **Why the envtest calls `publish` directly:** `checkOnce` only logs write errors, so a
schema rejection would look like a missing object. Calling `publish` makes the API
server's answer the test failure.
- Its first round writes a 2000-character message; the stored message is truncated to
1024.
- Its second round proves that only the target that flipped gets a new
`lastTransitionTime`.
- **No guard against a hung checker:** the Monitor trusts `Checker`'s contract to return
once the context ends, and doesn't abandon a checker that ignores it. All three
checkers stop when the context ends: KVM through `awaitOrCancel`, and kubernetes and
GCP through their clients.
- **Test renamed:** the planned `TestMonitor_timeoutBoundsBlockingChecker` is
`TestMonitor_checkOnce_timeoutBoundsEachRound`. It also asserts that a fast provider
still publishes next to a hung one.
## Step 5 — Metrics and wiring
`metrics.RegisterProviderTargets` registers a scrape-time collector for three gauges,
each labelled `{provider,target}`: `up`, check duration and check timestamp.
`Register`'s signature is unchanged.
`cmd/main.go`:
- gains `--provider-check-interval` (1m) and `--provider-check-timeout` (15s);
- validates them before even loading the providers config;
- collects the `Checker`s from the unwrapped providers in config order, just before the
metrics and tracing wrap loop;
- adds the Monitor with the manager's API reader;
- registers the collector through a small `providerTargetSamples` closure next to
`proxyPhaseCounts`.
Build, vet and metrics tests:
```bash
go build ./... && go vet ./cmd/... ./internal/metrics/ && go test -race -count=1 ./internal/metrics/
```
To confirm bad flags fail fast, before anything touches the cluster, I ran the binary
with a 5s interval. zap's development mode prints a stack trace after the error, hence
the grep:
```bash
go run ./cmd --providers-config=/dev/null --provider-check-interval=5s 2>&1 | grep -m1 "Invalid provider check flags"
# ERROR setup Invalid provider check flags {"error": "--provider-check-interval must be 0 (off) or at least 10s, got 5s"}
go vet -tags=e2e ./test/e2e/
```
The e2e suite gets a new ordered spec, "should publish provider health". It waits for
`providerhealth kubernetes` to report `Ready=True`. It runs before the existing metrics
spec, which now also asserts
`proxy_operator_provider_target_up{provider="kubernetes",target="apiserver"} 1`. By then
the first round is known to be done, so the series is certain to exist.
Worth noting:
- **Main-only wiring:** `providerTargetSamples` is an 8-line loop in `package main`,
which has no test file. The e2e metrics assertion is its only coverage.
- **The KVM entry isn't asserted in e2e:** the default providers config also ships
`kvm-home` (beelink), which would report `SSHMaterialUnavailable` in kind. That entry
is the user's own host, and asserting it would break the suite as soon as the default
config changes.
- **Manifest:** `config/manager/manager.yaml` only gained a comment about the two flags.
The defaults need no args.
## Step 6 — Docs
- **`docs/architecture.md`:**
- component rows for the CRD, the `Checker` contract and the monitor;
- a new §11 covering the provider table, the loop diagram, the write, read and
decorator choices, the gauges, and sample `kubectl get ph` output;
- cross-references from the metrics (§9) and tracing (§10) sections;
- four Decisions entries.
- **`docs/kvm-provider.md`:**
- a "Health check" section with the reason → meaning → where-to-look table and the
two consequences of sharing the connection;
- under Known gaps, the capacity bullet now points at `poolAvailable`, and a new
bullet says a rotated `kvm-ssh` Secret needs a restart;
- the Testing section names `TestKVMCheck_realHost`.
- **`docs/testing.md`:**
- suite rows for the three `check_test.go` files, `internal/providerhealth` (unit and
envtest) and the metrics gauges;
- "17 libvirt calls" is now 20;
- the integration and e2e sections describe the new tests.
- **`CLAUDE.md`:**
- the Architecture TODO now points at `docs/architecture.md`;
- the stale "only `go.mod` exists" gotcha is replaced by a note that the "What this
is" and "Commands" sections predate the code, plus Step 4's RBAC-marker gotcha.
Worth noting:
- `docs/api.md` is untouched. It documents the discovery HTTP API, which this change
doesn't alter.
- `CLAUDE.md`'s "What this is" and "Commands" sections were left stale on purpose.
Rewriting them is outside this plan, so the new gotcha flags them instead of leaving
them silently wrong.
## Step 7 — Validate, record, push, open the PR
The full suite runs codegen, `go fmt` and `go vet`, then `go test -race` on every
package, with both envtest suites included:
```bash
make test
# every package ok; internal/providerhealth 92.5 %, internal/provider/kvm 92.0 % coverage
```
`go fmt` rewrote nothing, and `git status` showed only the Step 6 docs.
`make lint` could not run, for a reason unrelated to this branch. The Makefile rebuilt
the custom golangci-lint, which reports "built with go1.27.0", and its bundled
staticcheck (honnef.co/go/tools v0.7.0) panics on the standard library before it
reaches any repo package:
```bash
make lint
# buildir: package "poll" (isInitialPkg: false, needAnalyzeSource: true): unexpected expr: *ast.KeyValueExpr
# make: *** [Makefile:107: lint] Error 3
```
The local toolchain is go1.27.0 while `go.mod` says 1.26.0; the otel-tracing plan's
log shows `make lint` running normally before the toolchain moved. Retrying with
`GOTOOLCHAIN=go1.26.0` would have meant downloading a toolchain, and the user declined
that. `go vet` is clean in all three configurations: plain inside `make test`, plus
`-tags=integration` and `-tags=e2e`.
The kind e2e suite creates and deletes its own `egress-proxies-operator-test-e2e`
cluster. I checked `KIND_CLUSTER` first, because the two clusters already on the
machine (`kind`, `proxy-operator-demo`) must not be torn down by `cleanup-test-e2e`:
```bash
make test-e2e > <scratch>/e2e.log 2>&1
# Ran 3 of 5 Specs in 86.210 seconds
# SUCCESS! -- 3 Passed | 0 Failed | 0 Pending | 2 Skipped
```
In the deployed manager, the `kubernetes` ProviderHealth reported `Ready=True` about a
second after the pod was up (20:30:05.6 → 20:30:06.7). The metrics scrape that followed
contained `proxy_operator_provider_target_up{provider="kubernetes",target="apiserver"} 1`.
That exercises the generated RBAC, the API-reader path and the `main.go` gauge wiring
end to end. The two skipped specs are the Tempo-gated tracing specs; `TEMPO_URL` was not
set.
### Follow-up — the check must not cost the reconciles their connection
Reviewing the shared-connection decision after the automated validation turned up two
ways the health check could hurt the work it shares the link with, both of which the
plan's "a check against a hung host drops the shared connection" line accepted too
easily:
- A check that timed out went through `hostPool.call`, whose cancel path drops the
connection. A `libvirtd` slow to answer one `ConnectGetLibVersion` would therefore
cut off an image upload or a `Create` running on the same link.
- `hostEntry` was guarded by a `sync.Mutex`, so waiting for a host's slot could not be
interrupted. A reconcile redialing a host that is down holds the slot for as long as
the SSH dial takes — minutes on a blackholed address — and the whole check round
would wait behind it, past its own timeout.
So the check now calls `hostPool.probe` instead of `call`: on `ctx` ending it abandons
the RPC and leaves the connection alone (a link that is really dead is closed by the
keepalive anyway, which ends the abandoned `fn` too). One abandoned probe per host is
allowed to stay waiting — `hostEntry.probing` is an `atomic.Bool`, not a counter, so a
daemon that never answers cannot pile up goroutines; further probes get the new
`errProbeBusy`, which `checkReason` maps to `Timeout`. `hostEntry.mu` became a one-slot
channel with an `acquire(ctx)` that only bounds *waiting*: an uncontended caller takes a
free slot even with an expired context, so `call`'s behaviour is unchanged where nothing
is contended.
The second change came out of the same pass: `--provider-check-timeout` defaulted to a
flat 15s while the flag validation rejected a timeout longer than the interval, so
`--provider-check-interval=12s` alone failed to start. The flag now defaults to 0 and
`Monitor.setDefaults` derives `min(15s, Interval)`; `ValidateSchedule` accepts 0 and
rejects only negatives and timeouts past the interval.
Re-verified with the same commands as above:
```bash
go vet ./...
go test -race -count=1 ./internal/provider/kvm/ ./internal/providerhealth/
make test
# every package ok; internal/provider/kvm 92.2 %, internal/providerhealth 93.7 % coverage
```
Worth noting:
- `make lint` still cannot run, for the toolchain reason recorded above; unchanged by
this follow-up.
- The two new `kvm` tests are timing-shaped by necessity — they assert that a caller
with a 50ms deadline does not wait out a dial that is blocked on a channel. The
assertions are one-sided (`> 1s` fails) so a slow machine does not make them flaky.
- `docs/architecture.md`, `docs/kvm-provider.md` and `docs/testing.md` were corrected in
the same commit: the old text promised the drop-on-timeout behaviour that is now gone.
- The real-hypervisor commands from the plan's Verification section are still the user's
to run; nothing here replaces them.
### Real-hypervisor validation and PR
The user ran the plan's real-hypervisor checks against beelink (the `TestKVMCheck`
integration test plus the manual `ProviderHealth` and metrics checks) and confirmed
they passed. After that the CHANGELOG entry went in, and the branch was pushed with
the token-file credential helper. The PR against `main` was opened through the Gitea
API because `tea` isn't installed.
Worth noting:
- The branch also carries `docs/plans/2026-09-16-0841-provider-blueprint.md`, on
purpose: that plan reaches `main` with this merge, and the blueprint work starts on a
new branch afterwards.
@@ -0,0 +1,177 @@
# Execution: provider blueprint
Plan: [docs/plans/2026-09-16-0841-provider-blueprint.md](../plans/2026-09-16-0841-provider-blueprint.md)
- [x] Step 0 — Save the plan; wait for the merge
- [x] Step 1 — Collect evidence (read-only, in parallel)
- [x] Step 2 — Write `docs/providers.md`
- [x] Step 3 — Verify, then open the MR
## Step 0 — Save the plan; wait for the merge
The plan was committed on `feat/provider-health` (`80b2e9c`) and reached `main`
through PR #6. After the user confirmed the merge, the branch was cut from the
updated `main`:
```bash
git switch main && git pull --ff-only
git switch -c feat/provider-blueprint
```
Worth noting: `.claude/settings.json` has local, uncommitted permission edits
that carried over through the switch. They are deliberately left out of this
branch.
## Step 1 — Collect evidence (read-only, in parallel)
Three Explore agents answered the plan's 12-item questionnaire, one each for
kubernetes, gcp and kvm. Meanwhile I read the shared layer:
`internal/provider/*.go`, `registry.go`, the provider wiring in `cmd/main.go`,
`reconcileManaged`/`replaceInstance`/`reconcileDelete` in the controller,
`internal/gc/gc.go`, and how the monitor uses `Checker`.
Before relying on the agents' claims, I spot-checked them against the code. One
was wrong and did not go into the doc: the kvm agent said `Delete` skips volume
removal when `DomainDestroy` returns `ErrNoDomain`. In fact `removeDomain`
returns nil there, and `Delete` still calls `removeVolumes` afterwards
(`kvm.go:477-482`).
Rules that exist only in the callers, and not in any comment, were found by
reading the controller and GC. They went into the doc as MUST rules:
- A zero `CreatedAt` makes an instance immediately collectable (`gc.go:120`).
- `Get` returning `ErrPermanent` on the deletion path makes the reconciler call
`Delete` anyway.
- Replacement waits for `Get` to return NotFound before creating an instance
with the same name.
- `Stopped` and `Terminated` both lead to delete-and-recreate.
Worth noting, found outside this plan's scope and not changed here:
- `docs/gcp-vm-validation.md` filters on `labels.proxy-operator-managed=yes`,
but the value is `true`.
- According to the gcp agent, `docs/gcp-in-specific-project.md` holds a real
project ID, IAM output and an egress IP. The user was told.
## Step 2 — Write `docs/providers.md`
Wrote the eight sections the plan lists. Linked the new doc from the Provider
contract row in `docs/architecture.md` and from the CLAUDE.md "Architecture"
section.
Judgment calls:
- The file:line evidence sits in per-provider notes under the matrix, as
planned. The contract rules in §2 cite the reconciler and GC lines each rule
protects.
- The gap list includes documentation drift (kvm-provider.md's ARP order and
log keys, gcp-vm-validation.md's label value and its claim that async errors
reach status), as well as code gaps. Both would mislead the next
provider author in the same way.
- "Proposed features" stays a flat list with no design, as the user decided.
Worth noting: the line numbers in the doc are a snapshot. The doc says so and
asks readers to look for the named function when a line has moved.
## Step 3 — Verify, then open the MR
An Explore agent re-checked the doc against the code: about 150 claims,
including every matrix cell. It found four mismatches, and all four were fixed:
- **Startup dial:** I had listed kvm-provider.md's "connection kept open from
startup" as doc drift. It isn't: the pool dials lazily, but the provider
health monitor runs a check as soon as the leader starts
(`monitor.go:147`, default interval 1m), and that check opens the connection.
The bullet was removed and the kvm notes now say this.
- **Checker wording:** §5 said the monitor type-asserts `Checker`. It is
`cmd/main.go:216`; the monitor only sees nil or not.
- **Reasons, not conditions:** §3 called `QuotaExceeded` and `PermanentError`
conditions. They are reasons on `Provisioned`.
- **Latch lines:** the line range is `:133-139`, not `:130-135`.
It also noted that none of the providers assert `var _ provider.Provider`, and
that the kubernetes provider's RBAC markers live in the controller package. The
§7 checklist now says both.
Link check (every relative `](…)` target and anchor in the new and edited docs
resolves), then the sanity build:
```bash
go build ./... && go test -short ./...
# all packages ok
```
Worth noting: no CHANGELOG entry yet. Per CLAUDE.md it goes in after the user
confirms the doc.
### Follow-up — tracing was missing from the blueprint
The user pointed out, after the PR was opened, that the blueprint said almost
nothing about tracing. The plan's questionnaire item 10 had collected the
evidence, but the doc reduced it to a matrix row about logging and a checklist
line telling authors not to open spans. That line even suggests tracing needs
no work. A provider author needs the opposite message: the operation span comes
free, and child spans for backend calls are theirs to add.
Added a new §6, "Observability: tracing, logs and metrics", and renumbered the
matrix, checklist and gaps sections to §7–§9. It covers:
- **Three layers:** decorator operation spans; child spans per backend call,
with parent-gated HTTP transports or a span decorator over the SDK seam; and
logs.
- **The ctx and logger rules** that keep `traceID` on log lines.
- **A per-provider status table:**
- kubernetes: parent-gated `RestConfigWrapper` spans.
- gcp: the Google SDK's own `otelhttp`, never disabled by the provider.
- kvm: no spans. `trace.go` is a log decorator.
Also added two tracing rows to the §7 matrix, rewrote checklist item 8 as
"Tracing and logging" with an in-memory span-recorder test, and added gaps (kvm
has no child spans; only kubernetes is checked against Tempo, and only at
operation level). A proposal notes that the kvm `libvirtAPI` seam takes no
`ctx`, so per-RPC spans need that plumbed in.
The links were re-checked, including the new `architecture.md#10-tracing-internaltracing` anchor.
### Follow-up — a completeness sweep
The user asked for one more pass over everything provider-related. After the
tracing miss, this pass read beyond the provider packages:
- the Proxy API types and the spec hash;
- the health engine and discovery call sites that consume `Instance.IP`;
- the manager Deployment, the providers ConfigMap and the samples;
- README, testing.md, kvm-provider.md and the GCP docs;
- the initial prompt, the proxy-operator and kvm plans, and the 2026-08-10 PR
review;
- the CHANGELOG.
What the blueprint was missing, now added:
- **What the instance is for.** The IP and port must be reachable from the
operator's health probes and from discovery clients. Providers don't know
which proxy runs; the proxy comes from cloud-init, except for kubernetes.
- **Concurrency.** Three reconciles, GC and checks can call a provider at once,
so it MUST be safe for concurrent use.
- **The runtime envelope:** read-only root filesystem, a 64Mi `/tmp`, a 128Mi
memory limit, credentials as optional mounts.
- **Create:** `placement` can't be validated by the CRD, so error messages must
name the field. Catalogue edits in config aren't in the spec hash.
- **Delete** is re-called every `DeletionPoll` while a delete is in progress.
- **Provider names** must be DNS-1123 for `ProviderHealth`, and `spec.provider`
is immutable.
- **Metric names** in §6.
- **Three matrix rows:** proxy software, IP reachability, distinct egress.
- **A deployment checklist item:** credentials mount, config entries, sample
Proxy, Makefile helper.
- **Gaps:**
- still open from the PR review: GC has no installation identity; an
unconfigured provider latches Proxies; kubernetes Pod RBAC always installed
and no `OwnerReference` on Pods;
- the duplicated type list;
- samples that can't work: `proxy_gcp.yaml` uses a debian-cloud image without
cloud-init, and `proxy_kvm.yaml` places on `nas1` where configs define
`beelink`;
- the `"mock"` example in the `spec.provider` doc.
Outside the plan's "leave other docs alone" line, README.md changed:
- Its "Adding a provider" section restated the contract, with the old "Create
never blocks" rule that kvm contradicts. It now points to `docs/providers.md`
instead of keeping a second copy that disagrees.
- The "Implemented providers" table gained the missing kvm row.
- The deploy note no longer claims the default ConfigMap ships only the
kubernetes provider.
The links were re-checked for providers.md and README.md.
@@ -0,0 +1,354 @@
# Plan: Provider health checks — see whether the cluster can reach each provider
**Created:** 2026-09-11 15:26 UTC
## Context
Today the operator can't say whether a provider's backend is reachable. The `Provider`
interface ([provider.go:89-109](internal/provider/provider.go#L89-L109)) only has
Create/Get/Delete/ListByTag, and `/healthz` and `/readyz` are bare `healthz.Ping`. A
broken KVM path only shows up later: as a Proxy stuck in Provisioning, or as an orphan-GC
sweep logging an error. Several failures stay invisible until a Proxy needs that host:
- a missing `kvm-ssh` Secret;
- a rotated host key;
- a Tailscale route that is down;
- a storage pool that is inactive or full.
The goal is a periodic sanity check of every configured provider from inside the cluster,
with results you can see through `kubectl` and Prometheus.
Decisions made by the user during planning:
1. **Surface:** a new operator-owned, read-only, cluster-scoped **`ProviderHealth`** CRD,
with one object per configured provider, a `Ready` condition and per-target detail.
Also Prometheus gauges, and a log line whenever a target goes up or down. No
EventRecorder, matching the earlier decision.
2. **Scope:** real checks for all three providers.
- **kvm:** checked per host.
- **kubernetes:** confirms the API server and RBAC work.
- **gcp:** confirms credentials and the API work.
3. **KVM depth:** reachability plus the storage pool.
- Over the existing pooled SSH+libvirt connection, call the libvirt version and
hostname RPCs.
- Confirm the configured pool is running and report its free space.
- No bridge check.
Not affected: the Proxy reconcile behaviour, and the manager's `/healthz` and `/readyz`.
A dead hypervisor must never make the manager unready.
## Design
### Contract: `internal/provider/check.go`
```go
type TargetStatus struct {
Name, Address string
Ready bool
Reason, Message string
Latency time.Duration
Details map[string]string
}
// Check never errors; each target carries its own verdict. Must return promptly once ctx ends.
type Checker interface{ Check(ctx context.Context) []TargetStatus }
```
- **Shared reasons:** `Reachable`, `Timeout`, `Unreachable`, `Unauthorized`, `Forbidden`,
`APIError`.
- **Compile-time checks:** each provider's `check.go` has `var _ provider.Checker = (*Provider)(nil)`.
- **Bypass the wrappers.** `cmd/main.go:197-201` wraps every provider in
`WithTracing(WithMetrics(...))`, so the Checkers are collected from the **raw** providers
map, before that loop. As a result, checks don't inflate `provider_requests_total`.
### The checks
**kvm** (`internal/provider/kvm/check.go`)
- **Concurrency:** one goroutine per `cfg.Hosts` entry (`wg.Go`), writing to `out[i]` so
results stay in config order.
- **Inside `p.hosts.call`:** `ConnectGetLibVersion`, `ConnectGetHostname`,
`StoragePoolLookupByName(host.Pool)`, then `StoragePoolGetInfo`.
- **Rules for `fn`:**
- It returns only raw RPC errors. `call` drops the shared connection on any other error
([conn.go:160](internal/provider/kvm/conn.go#L160)), so "pool not running" is decided
after `call` returns, never returned from `fn`.
- It fills a captured result struct only after all four calls succeed. That struct is
read only when `call` returned nil, which avoids a race with an abandoned goroutine.
- **Details:** `hostname`, `libvirtVersion` (decoded as major.minor.patch), `pool`,
`poolState`, `poolCapacity` and `poolAvailable` (in GiB).
- **New RPCs:** add the three to `libvirtAPI` ([conn.go:29-47](internal/provider/kvm/conn.go#L29-L47)),
`tracingAPI` (`trace.go`) and `fakeAPI` (`fake_test.go`). `fakeAPI` gets a hostname, a
fixed version, and a per-pool state map that defaults to Running.
- **Telling failure stages apart without matching strings:** add
`type dialError struct{ stage dialStage; err error }` in `conn.go`, with `Error()`
returning the inner message unchanged and `Unwrap()`. It wraps at four places:
- `lazySSH.dial`, when loading the key material fails (`ssh-material`)
- `dialSSH`, at the dial step (`tcp`)
- `dialSSH`, at the handshake (`ssh-handshake`)
- `connectLibvirt` (`libvirt-connect`)
Because `Unwrap` is kept, `classify` and `isClassified` still see `ErrPermanent` and
`fs.ErrNotExist`, so reconcile behaviour doesn't change.
- **`checkReason(err)`, tested in order:**
| Condition | Reason |
|---|---|
| stage ssh-material | `SSHMaterialUnavailable` |
| stage tcp | `Unreachable` |
| stage ssh-handshake | `SSHHandshakeFailed` |
| stage libvirt-connect | `LibvirtConnectFailed` |
| ctx deadline or cancel | `Timeout` |
| `ErrNoStoragePool` | `PoolNotFound` |
| any other libvirt error | `LibvirtError` |
| anything else | `Unreachable` |
| success, pool not running | `PoolNotActive` |
| success | `Reachable` |
- **Accepted behaviour:**
- A check that hits its timeout drops the shared connection. That already happens for
any cancelled reconcile context, and it only fires when the host isn't answering.
- The leader now keeps one SSH connection per host open from startup, with keepalives
every 15s. Before, the connection only opened with the first KVM reconcile.
- No forced periodic redial. `lazySSH` caches the key material forever, so a redial
would not pick up a rotated Secret anyway; that gap goes into the docs under
"Known gaps".
**kubernetes** (`internal/provider/kubernetes/check.go`)
- **Call:** `List` pods with the managed-label selector and `client.Limit(1)`,
cluster-wide. It uses the same uncached client, verb and RBAC as `ListByTag` and GC.
There is no namespace setting in the kubernetes config.
- **Target:** one target named `apiserver`, with the address from a new `apiHost` field
set in `NewWithTransportWrapper`.
- **Reasons:**
- `Timeout`: the context ended, or the error is `IsTimeout`/`IsServerTimeout`.
- `Unauthorized`, `Forbidden`.
- `APIError`: any other API status.
- `Unreachable`: anything else.
**gcp** (`internal/provider/gcp/check.go`, plus a seam change in `gcp.go`)
- **Why a new seam method:** the existing `AggregatedList` seam reads every page
([gcp.go:63-78](internal/provider/gcp/gcp.go#L63-L78)), and MaxResults is only the page
size. So add `Probe(ctx, *computepb.AggregatedListInstancesRequest) error`, which makes
one `it.Next()` call and treats `iterator.Done` as nil.
- **Request:** `Project`, `MaxResults: 1`, `ReturnPartialSuccess: true`, and the same
managed-label filter `ListByTag` uses.
- **Target:** one per project, `Name: cfg.Project`, `Address: compute.googleapis.com`.
GCP config has no zones, and reading them from Proxy placements would tie the check to
the Proxy cache.
- **Reasons:**
- `Timeout`: the context ended.
- For `*googleapi.Error`: 401 is `Unauthorized`, 403 is `Forbidden` (this also covers a
disabled API), anything else is `APIError`.
- `Unreachable`: anything else, including a failure to get a token.
### The `ProviderHealth` CRD: `api/v1alpha1/providerhealth_types.go`
- **Written by hand**, since kubebuilder isn't installed; `PROJECT` is edited by hand too.
- **Scope and markers:** cluster-scoped, `shortName=ph`, status subresource, no spec.
- **Scheme registration:** add the type to the existing `AddKnownTypes` call in
[proxy_types.go:362](api/v1alpha1/proxy_types.go#L362), not a second `init()`.
- **Status:**
- `type`
- `conditions` (`Ready`; `ConditionReady` constant)
- `targets[]`, as a list-map keyed by `name`, each with: `name`, `address`, `ready`,
`reason`, `message` (at most 1024 characters), `latencyMillis`,
`lastTransitionTime`, `details`
- `targetsReady` (for example `"1/2"`, used by a printer column)
- `lastCheckTime`
- **Printer columns:** Type, Ready, Targets, Last Check, Age.
- **Viewer role:** `config/rbac/providerhealth_viewer_role.yaml`. No editor role, since the
operator owns the object.
### Monitor: `internal/providerhealth`
- **Fields:** `Monitor{Client, Reader, Targets []Target{Name, Type, Checker}, Interval, Timeout}`.
- **Scheduling:**
- `NeedLeaderElection() true`.
- Checks once immediately, then on a ticker. Copy the ticker loop from
[gc.go:54-80](internal/gc/gc.go#L54-L80).
- **Each cycle:**
- Runs every provider concurrently under `context.WithTimeout(Timeout)`, inside one
`tracing.Start` span.
- Stores the snapshot under a mutex; `Snapshot()` returns a deep copy.
- Logs changes. Going down logs `WARNING: provider target not ready` with provider,
target, address, reason and message. Coming back logs `Provider target ready`. A
target that is ready on first sight logs at V(1) only.
- **Writes:**
- Get through `mgr.GetAPIReader()`, not the cache. That avoids a cluster-wide informer
and the watch permission it would need, whatever `--proxy-namespace` is set to.
- Create the object if it is missing, labelled
`app.kubernetes.io/managed-by=egress-proxies-operator`.
- Then `Status().Patch(MergeFrom(base))` when the status changed, following
[status.go:114-121](internal/controller/status.go#L114-L121).
- This means one write per provider per interval, because `lastCheckTime` moves each
time. That's acceptable given the 10s minimum interval.
- **`buildStatus(prev, result, now)` is a pure function:**
- A target keeps its `lastTransitionTime` unless `ready` flipped.
- The `Ready` condition is set through `apimeta.SetStatusCondition`:
- `True/AllTargetsReady` when every target is ready
- `False/TargetsNotReady` otherwise, with the failing targets in the message
- `Unknown/ChecksNotImplemented` when the provider has no Checker
- `Unknown/NoTargets` when there are zero targets
- **Cleanup, once at `Start`:** delete labelled objects whose name isn't in the config.
The config is loaded once per process, and unlabelled objects are never touched.
- **Interval 0 (disabled):** run the cleanup with nothing kept, then return nil, so no
object is left showing a stale `Ready=True`.
- **Invalid provider names:** provider names aren't validated as DNS-1123 names. For an
invalid name, warn once and skip the object writes; metrics are still emitted.
- **`ValidateSchedule(interval, timeout)`** is called in `main.go` before the manager
starts. Interval must be 0 or at least 10s; timeout must be in (0, interval].
- **RBAC markers on `Monitor`:**
- `providerhealths`: get, list, create, delete
- `providerhealths/status`: get, patch
- `make manifests` regenerates `config/rbac/role.yaml`.
### Metrics and wiring
- **New function:** `metrics.RegisterProviderTargets(reg, read func() []ProviderTarget) error`.
`Register`'s signature is left alone.
- **Gauges:** a scrape-time collector, in the style of `constCollector`, reads the Monitor
snapshot and emits three gauges labelled `{provider,target}`:
- `proxy_operator_provider_target_up`
- `proxy_operator_provider_check_duration_seconds`
- `proxy_operator_provider_check_timestamp_seconds`
- **Conversion closure:** the one that turns the Monitor snapshot into gauges lives in
`main.go`, next to `proxyPhaseCounts`. So `providerhealth` doesn't import `metrics`.
- **No pool-bytes gauge:** that value is in status, and node-exporter is the right source
for disk alerts.
- **`cmd/main.go`:**
- New flags `--provider-check-interval` (1m) and `--provider-check-timeout` (15s).
- Build the `Targets` from `cfg.Providers` in config order, with
`c, _ := raw[pc.Name].(provider.Checker)`.
- `mgr.Add(&providerhealth.Monitor{Client: mgr.GetClient(), Reader: mgr.GetAPIReader(), ...})`,
then register the collector.
## Steps (one commit each, execution-log entry per step)
**Step 0 — Save the plan.**
- Create branch `feat/provider-health` off `main`.
- Copy this plan to `docs/plans/2026-09-11-1526-provider-health-checks.md` and commit it
alone.
- The uncommitted `.claude/settings.json` change stays out of every commit.
**Step 1 — ProviderHealth API and CRD.**
- `api/v1alpha1/providerhealth_types.go`
- `zz_generated.deepcopy.go`, from `make generate`
- `config/crd/bases/crawl.example.com_providerhealths.yaml`, from `make manifests`
- the CRD and RBAC `kustomization.yaml` files, plus the viewer role
- `PROJECT`
**Step 2 — Contract plus the kubernetes and GCP checks.**
- `internal/provider/check.go`
- `internal/provider/kubernetes/{kubernetes.go,check.go,check_test.go}`
- `internal/provider/gcp/{gcp.go,check.go,check_test.go,gcp_test.go}`
**Step 3 — KVM check.**
- `internal/provider/kvm/{conn.go,trace.go,check.go,check_test.go,fake_test.go,conn_test.go,trace_test.go,kvm_integration_test.go}`
**Step 4 — Monitor.**
- `internal/providerhealth/{monitor.go,status.go,monitor_test.go,status_test.go,envtest_test.go}`
- `config/rbac/role.yaml`, from `make manifests`
**Step 5 — Metrics and wiring.**
- `internal/metrics/{metrics.go,metrics_test.go}`
- `cmd/main.go`
- `config/manager/manager.yaml`: a comment for the new flags
- `test/e2e/e2e_test.go`
**Step 6 — Docs.**
- `docs/architecture.md`: a component row plus a provider-health section
- `docs/kvm-provider.md`: the reason table, the shared-connection behaviour, and the
Secret-rotation gap under "Known gaps"
- `docs/testing.md`: the new tests and the commands
- `CLAUDE.md`: its Architecture and Gotchas sections still say "empty scaffold"; point
them at `docs/architecture.md`
**Step 7 — Validate, record, push, open the PR.**
- Run `go vet ./...`, `go test -short ./...`, `make test` and `make test-e2e`.
- Hand the user the KVM commands below, and record their results in the execution log.
- Add a CHANGELOG entry once the user confirms.
- Push, using the token-file credential helper from memory.
- Open the PR against `main` through the Gitea API (`tea` isn't installed).
- Don't merge.
## Tests
**Provider checks** (table-driven, `Test<Func>_<scenario>`):
- kubernetes:
- reachable
- list options: `Limit==1`, the label selector, no namespace
- failure reasons: forbidden, unauthorized, timeout, cancelled, internal, and a
network `*url.Error`
- gcp:
- reachable, asserting the request fields
- failure reasons: 401, 403, 500, deadline, network
- `TestRealInstances_Probe_fetchesOnePage` against `httptest`: exactly one request with
`maxResults=1`, even though the page has a `nextPageToken`
- kvm (using `hostFakes`/`newTestProvider` from `kvm_test.go`):
- reachable, with details filled in
- per-host verdicts, order preserved
- pool inactive, pool missing
- hosts checked concurrently (a sequential check would deadlock)
- a pool verdict keeps the connection (the dial count stays 1)
- a `checkReason` table
- `TestSSHAuth_dial_stages`, using the in-process SSH server in `conn_test.go`
- three new rows in the tracing table
**Monitor:**
- Setup: the fake client is built with a status subresource and a RESTMapper that
registers ProviderHealth as cluster-scoped.
- Tests:
- `buildStatus` table
- objects are created with the label, and existing ones are patched
- the timeout bounds a checker that blocks
- a nil checker gives Unknown
- only transitions are logged (`funcr` logger)
- cleanup deletes only labelled objects that aren't configured
- the disabled path cleans up and returns
- an invalid name skips the writes
- `ValidateSchedule` table
- An envtest round-trip, skipped under `-short`, which proves the generated schema is
accepted by a real apiserver.
**Metrics:** `TestRegisterProviderTargets_scrape`, using a fresh registry and
`GatherAndCompare`; an empty snapshot produces no series.
**e2e (kind):** `providerhealth kubernetes` reaches `Ready=True`, which proves the
generated RBAC works end to end.
## Verification
**Automated (Claude runs):** `go vet ./...`, `go test -short ./...`, `make test` (envtest
with `-race`) and `make test-e2e`.
**Against the real hypervisor (handed to the user, not run by Claude)**, with beelink as
the host:
```bash
KVM_TEST_HOST=beelink.tail8c69c.ts.net:22 KVM_TEST_USER=novakj KVM_TEST_SSH_KEY=~/.ssh/id_ed25519 \
KVM_TEST_KNOWN_HOSTS=~/.ssh/known_hosts KVM_TEST_BRIDGE=<bridge> KVM_TEST_IMAGE_URL=<image-url> \
go test -tags=integration -count=1 -v -run TestKVMCheck ./internal/provider/kvm/
make install && go run ./cmd/main.go --providers-config <scratch copy of kvm-home config, local key paths> \
--metrics-bind-address :8080 --metrics-secure=false --provider-check-interval=15s
kubectl get ph -w
kubectl get ph kvm-home -o yaml
curl -s localhost:8080/metrics | grep -E 'proxy_operator_provider_(target_up|check_)'
```
Each negative case is a one-line edit to the scratch config; nothing changes on the host.
| Edit | Expected reason |
|---|---|
| `pool: nope` | `PoolNotFound` |
| `address: …:2` | `Unreachable` |
| `knownHostsFile: /dev/null` or `user: nobody` | `SSHHandshakeFailed` |
| `sshKeyFile: /nonexistent` | `SSHMaterialUnavailable` |
| `socket: /nonexistent` | `LibvirtConnectFailed` |
## Out of scope
- Gating Proxy creates on a host that is known to be down.
- Tying `/readyz` to provider health.
- Bridge checks.
- SelfSubjectAccessReview checks of create/delete rights.
- Hot-reloading the providers config.
- Picking up a rotated `kvm-ssh` Secret without a restart. This is an existing gap; it
gets documented.
@@ -0,0 +1,119 @@
# Plan: Provider blueprint (docs/providers.md)
**Created:** 2026-09-16 08:41
## Context
The provider contract is spread across `docs/prompts/__initial-prompt.md` §5,
the comments in `internal/provider/{provider,errors,check}.go`, `docs/architecture.md`
§5 and §11, `docs/kvm-provider.md`, and plan Step 2/3/8. Many rules exist only in
code: providerID must be self-contained, UID/CreatedAt must be read back, Create
must survive a crash partway through, calls must stop when ctx is cancelled,
decorators are wired in `cmd/main.go`, and so on. With "~5 providers soon", the
next provider author needs one blueprint. It should say what a provider MUST,
SHOULD or MAY do, show which features each existing provider (kubernetes, gcp,
kvm) actually implements (with file:line evidence), give a checklist for adding
one, and list the gaps.
Decisions (user): store in `docs/providers.md`; scope = current state plus a
separate "Gaps and proposed features" section (nothing implemented); wait for
`feat/provider-health` to merge, then branch off `main`.
## Step 0 — Save the plan; wait for the merge
- Right after ExitPlanMode, copy this plan to `docs/plans/2026-09-16-0841-provider-blueprint.md`
and commit it on its own. The current branch is `feat/provider-health`, so the
plan reaches `main` with that merge. CLAUDE.md requires the plan commit to
come before any branching.
- Stop and wait for the user to confirm that `feat/provider-health` is merged.
Then `git switch main && git pull`, and `git switch -c feat/provider-blueprint`.
## Step 1 — Collect evidence (read-only, in parallel)
Run 3 Explore agents at the same time, one each for `internal/provider/kubernetes`,
`internal/provider/gcp` and `internal/provider/kvm`, including their tests. Each
answers the **same questionnaire**, citing `file:line` for every answer and writing
"not implemented" explicitly when that's the case:
1. **Identity**: providerID format and parsing; what it carries (zone, host, namespace).
2. **Create**: how it stays idempotent (name lookup, 409, DomainLookupByName);
whether it returns before the VM is up or waits; what happens when an earlier
attempt crashed partway; which Placement fields are used or ignored; how
cloud-init is delivered; how the port is used; how GC labels/UID are attached.
3. **Get**: native state → InstanceState mapping; IP discovery rules; UID and
CreatedAt readback; how NotFound is detected.
4. **Delete**: idempotency; what gets removed (disks, volumes, seeds); returns
before or after the resource is gone; unknown/malformed providerID behaviour.
5. **ListByTag**: scope (namespaces, zones, hosts); partial-failure behaviour; how
managed resources are recognised.
6. **Error classification**: table of backend error → NotFound/Quota/Transient/
Permanent; whether quota is detected at all.
7. **Health check** (`Checker`): targets, provider-specific Reasons, Details keys.
8. **Config**: YAML section, defaults, validation (`internal/provider/config*.go`),
credentials/Secrets, which settings are read once per process.
9. **Connections and cancellation**: pooling, reconnects, timeouts, ctx honouring.
10. **Observability**: log keys and verbosity levels, spans (`kvm/trace.go`, gcp
`wirelog.go`), anything beyond the shared metrics/tracing decorators.
11. **Testing**: fakes, envtest/kind, `//go:build integration` tests, what is
not covered.
12. **Known gaps**: TODOs, stale comments, documented limitations.
Meanwhile, read the shared layer myself: `internal/provider/{provider,errors,check,
name,config,metrics,tracing}.go`, `internal/provider/registry/registry.go`,
`cmd/main.go` (registry wiring, decorator order, Checker collection), and the
places that consume the contract, so every MUST rule traces to a caller that depends on it:
`internal/controller` (error-class branching, ErrNotFound → replace/adopt),
`internal/gc` (UID, CreatedAt age skip, ListByTag errors),
`internal/providerhealth`. Then cross-check against the existing docs listed in Context.
## Step 2 — Write `docs/providers.md`
Sections:
1. **What a provider is**: a few sentences plus links to architecture.md §5/§11.
2. **The contract**: per method (Create/Get/Delete/ListByTag), MUST/SHOULD/MAY
rules, each followed by *why*: the reconciler or GC behaviour it protects,
linked to code.
3. **Error taxonomy**: what each class makes the reconciler do; how to map backend
errors; "unclassified = transient" pitfall.
4. **Identity and GC labels**: providerID guidance, `LabelManaged`/`LabelUID`,
CreatedAt, deterministic `NameFromUID`.
5. **Optional interfaces**: `Checker` (shared reasons, never fails as a whole,
collected before decorators).
6. **Feature matrix**: one row per questionnaire item, columns kubernetes/gcp/kvm,
cells `yes`/`partial`/`no` plus a short note. Put the file:line evidence in a
compact per-provider notes subsection under the table, not in the cells.
7. **Adding a provider: checklist**: package layout, config section and validation,
registry constructor, `cmd/main.go` wiring, Checker, error mapping with tests,
`name`/label tests, integration test behind a build tag, provider doc page,
architecture.md rows, demo script (optional).
8. **Gaps and proposed features**: current-state gaps per provider (e.g. no kvm
quota detection, orphan volumes invisible to GC, Secret rotation needs a
restart, ListByTag partial-failure inconsistency, stale `mock` references in
`provider.go` comments), plus proposals that are clearly marked as not
implemented. Only list them; don't design them.
Also: link it from `docs/architecture.md` (Components table → provider contract row)
and from the CLAUDE.md "Architecture" section. Leave per-provider docs where they are.
Follow the plan-execution convention: create `docs/plans-executions/2026-09-16-0841-provider-blueprint.md`
with the status checklist and step entries, and commit it with each step.
## Step 3 — Verify, then open the MR
- A verification agent (Explore) re-checks each matrix cell and each MUST rule
against the cited file:line and reports any mismatch. Fix what it finds.
- Check that every relative link in the new and edited docs resolves (grep the
`](` targets and `ls` each one).
- `go build ./... && go test -short ./...` as a sanity check (docs-only change,
expected no-op).
- Push `feat/provider-blueprint` and open the MR against `main` via the Gitea API,
following the project's push/MR memory note. After the user confirms, add a CHANGELOG.md entry.
## Critical files
- New: `docs/providers.md`, `docs/plans/2026-09-16-0841-provider-blueprint.md`,
`docs/plans-executions/2026-09-16-0841-provider-blueprint.md`
- Edited: `docs/architecture.md` (link), `CLAUDE.md` (link)
- Read-only sources: `internal/provider/**`, `cmd/main.go`, `internal/controller`,
`internal/gc`, `internal/providerhealth`, `docs/prompts/__initial-prompt.md`,
`docs/kvm-provider.md`, `docs/gcp-vm-validation.md`
+730
View File
@@ -0,0 +1,730 @@
# Providers: the blueprint
This page is for whoever writes the next provider. It explains the contract a
provider must meet, which caller depends on each rule, how the three existing
providers (kubernetes, gcp, kvm) meet it, and what is still missing.
Where the rest lives:
- **Data flow:** the calls each provider makes are in
[architecture.md §5](architecture.md#5-what-provider-calls-do-in-the-outside-world).
- **Health checks:** the monitor that runs them is in
[architecture.md §11](architecture.md#11-provider-health-checks-internalproviderhealth).
- **Per-provider guides:** [kvm-provider.md](kvm-provider.md) and
[gcp-vm-validation.md](gcp-vm-validation.md).
- **Code:** the contract itself is
[`internal/provider/provider.go`](../internal/provider/provider.go),
[`errors.go`](../internal/provider/errors.go) and
[`check.go`](../internal/provider/check.go).
Rule keywords: **MUST** means breaking the rule breaks a caller. **SHOULD**
means every current provider does it and a new one needs a reason not to.
**MAY** means it's allowed.
Code references are `path:line` at the time of writing. When a line has moved,
look for the function named next to it.
## 1. What a provider is
A provider is a Go type implementing `provider.Provider`: four methods over one
VM-like instance, identified by an opaque `providerID` string. Nothing else in
the operator knows how a backend works:
- **The reconciler** (`internal/controller`) calls `Create`, `Get` and `Delete`
and branches only on the error class and `Instance.State`.
- **Orphan GC** (`internal/gc`) calls `ListByTag` and `Delete`.
- **The provider health monitor** (`internal/providerhealth`) calls the optional
`Checker`.
The reconciler never watches backend resources. Everything it knows comes from
polling `Get` (`ProvisioningPoll` and `DeletionPoll` 10s, `RequeueNow` 1s,
`QuotaRetry` 5m, `internal/controller/proxy_controller.go:452-467`). Up to 3
Proxies reconcile at once (`MaxConcurrentReconciles: 3`, `:442`).
Providers are named instances of a type. `gcp-eu` and `gcp-us` can both have
type `gcp`. The name is what `Proxy.spec.provider` refers to. `spec.provider` is
immutable, so a Proxy never moves between providers.
**What the instance is for.** It is a forward proxy with its own egress path.
Two parties connect to `http://<Instance.IP>:<spec.port>`:
- The health engine, from the operator pod, sends a real `CONNECT` through it
(`internal/health/engine.go:226`).
- Crawler clients, which get `ip` and `port` from the discovery API
(`internal/discovery/handlers.go:37`).
So the IP a provider reports MUST be reachable on that port from the operator's
pod network and from the clients. That may require firewalls, security groups
or routing: gcp needs a manual firewall rule whose source range covers the
cluster's egress IP ([gcp-vm-validation.md](gcp-vm-validation.md)).
Providers don't know which proxy software runs. The proxy is installed and
configured by the Proxy's `cloudInit`, and building install logic into the
operator is an explicit non-goal (`docs/prompts/__initial-prompt.md` §12).
kubernetes is the exception: a Pod has no cloud-init, so it generates a Squid
config from `Port` itself.
**Who calls, and from where.**
- **Leader only:** the reconciler, orphan GC, the health engine and the provider
health monitor.
- **Every replica, at startup:** the constructors run in `cmd/main.go` before
the manager starts.
- **Concurrently:** up to 3 reconciles, a GC sweep and a health-check round can
call the same provider at the same time. A provider **MUST be safe for
concurrent use**, including two calls for the same instance name. Nothing
above the provider serialises them.
**Where it runs.** The manager container runs:
- as non-root, with a **read-only root filesystem**
(`config/manager/manager.yaml:57`, `:122`);
- with a 64Mi `emptyDir` at `/tmp` (`:176`);
- with a **128Mi memory limit** (`:144`).
So a provider:
- MUST stream large payloads rather than buffering them (kvm streams image
downloads, `image.go:182`);
- MUST write scratch files only under `/tmp`;
- gets credentials either as mounted files (the providers ConfigMap at
`/etc/proxy-operator`, optional Secret mounts such as `kvm-ssh`) or from the
environment (ADC, the in-cluster service account).
## 2. The contract, method by method
### Create(ctx, CreateRequest) (providerID, error)
`CreateRequest` carries:
- `Name`: from `NameFromUID`, 22 characters, `proxy-` plus base32.
- `UID`: the owning Proxy's UID.
- `Namespace` and `ProxyName`.
- `Placement`: `Region`, `Zone`, `MachineType`, `Image`.
- `CloudInit`: already resolved from inline text or a Secret.
- `Port`.
Rules:
- **MUST be idempotent by `req.Name`.** A repeat call for a name that already
exists returns that instance's providerID with a nil error. It must not return
an error and must not create a duplicate.
*Why:* the reconciler records the ID in status only after `Create` returns
(`proxy_controller.go:155-172`). A crash between the two, or a status write
that is lost, means `Create` runs again for the same Proxy. `replaceInstance`
also depends on this to recover from a crash between its two writes
(`:240-244`).
- **MUST mark the instance as managed when it is created**, with
`LabelManaged=true` and `LabelUID=<req.UID>` or the backend's equivalent. kvm
uses domain `<metadata>` because libvirt has no labels.
*Why:* orphan GC finds instances only through that mark (`internal/gc/gc.go:105`),
and never deletes one with an empty UID (`gc.go:113-117`). An instance that
exists for a while without the mark leaks if the operator crashes.
- **MUST return a providerID that is self-contained.** `Get` and `Delete`
receive only the ID, and the config or spec may have changed since `Create`,
for example a zone edit during a replacement. So put the location in the ID:
- kubernetes: `<namespace>/<name>`
- gcp: `zones/<zone>/instances/<name>`
- kvm: `hosts/<host>/domains/<name>`
Validate any config values that end up in the ID so they can't contain the
separator (`config_kvm.go:108-114`).
- **MUST honour `ctx`.** The reconcile context is cancelled on shutdown and when
leadership is lost.
- **SHOULD return once the request is submitted, not once the instance is
running.** Readiness is discovered by polling `Get`. A blocking `Create` holds
one of the 3 reconcile workers for its whole duration. kvm breaks this rule on
purpose, see §9.
- **SHOULD reject a request it can never satisfy** with `ErrPermanent` before
touching the backend: a missing zone, machine type or image, or a name that is
not in its catalogue (gcp `gcp.go:137-141`, kvm `kvm.go:126-144`, `:160-164`).
The CRD can't validate `placement`, because it is provider-agnostic, so this
error is the user's only feedback. The message SHOULD name the offending field
and value.
- **Catalogue changes don't replace instances.** The spec hash covers
`placement`, the resolved `cloudInit` and `port` (`spechash.go:15-19`), but not
the provider config. When a placement value names a config entry (kvm's
`images` and `machineTypes`), changing what that entry means affects only new
instances. For kvm it can even corrupt running VMs
([kvm-provider.md § The image cache](kvm-provider.md#the-image-cache)). Document
this, and tell users to add a new key rather than edit one in use.
- **MAY ignore `Placement` fields that don't apply** (`provider.go:56-57`). If a
field is ignored, the provider's doc page MUST say so, because the fields are
still part of the spec hash. Changing an ignored field replaces the instance
and has no other effect.
- **SHOULD deliver `CloudInit` as user-data.** A provider that can't (kubernetes)
documents that it ignores the field.
- **MUST NOT log `CloudInit` contents.** It can hold secrets. Log its byte count
instead. gcp and kvm test this with a sentinel string (`kvm_test.go:975-995`,
`gcp_test.go:330-407`).
### Get(ctx, providerID) (*Instance, error)
- **MUST return `ErrNotFound` when, and only when, the instance is gone.**
*Why:* `ErrNotFound` triggers three actions:
- In normal reconciling, the reconciler clears the ID and creates again
(`proxy_controller.go:197-202`).
- During replacement, it is the signal to create the new instance
(`:237-251`).
- During deletion, it releases the finalizer (`:287-291`).
A false NotFound during an outage creates a duplicate or loses the instance.
"Unreachable", "wrong credentials" and "can't tell" are Transient or
Permanent, never NotFound.
- **MUST keep returning the instance, not NotFound, while a delete is in
progress** and the name is still taken.
*Why:* the replacement instance gets the same name. The reconciler waits for
NotFound before creating it (`:228-233` comment, `:237-251`). A NotFound too
early would make `Create` adopt the dying instance.
- **MUST report `Running` only with a non-empty `IP`.** Running without an
address yet is `Provisioning`.
*Why:* the IP from a `Running` answer goes straight into status and the health
engine (`:212-215`). All three providers do this.
- **MUST report `Stopped` or `Terminated` only when the instance will not
recover by itself.** SHOULD map states it doesn't recognise to `Stopped`.
*Why:* the reconciler deletes and recreates on either state (`:216-226`); the
instances are treated as disposable. The two states cause exactly the same
reaction.
- **MUST fill `UID` and `CreatedAt` by reading them back from the backend**,
never from memory.
*Why:* GC compares `UID` with the live Proxies. It deletes anything whose
`CreatedAt` is older than `MinAge` (10m), and **a zero `CreatedAt` counts as
very old** (`gc.go:120`). A provider that forgets `CreatedAt` makes instances
still being created collectable. An unparsable timestamp gives the zero time
in both gcp and kvm. That is accepted because it never shields an orphan
forever (`gcp.go:253-256`).
- **SHOULD return `ErrPermanent` for an ID it can never resolve**, such as a
malformed ID or a host no longer in the config.
*Why:* the deletion path then tries `Delete` anyway and releases the
finalizer if that works (`:292-302`). If the provider returned Transient,
deletion would poll forever.
### Delete(ctx, providerID) error
- **MUST be idempotent.** Deleting something that is already gone returns nil.
*Why:* GC and the reconciler can race, and a crash can repeat a call.
- **MUST accept being called again while an earlier delete is still in
progress.** Treat "already deleting" as success, and SHOULD keep that cheap.
*Why:* while `Get` still returns the instance, the reconciler calls `Delete`
again on every `DeletionPoll` (10s) until `Get` returns NotFound (`:305-311`).
gcp submits a new delete operation each time.
- **MUST remove everything `Create` made**: disks, volumes, seeds, addresses.
Either delete them explicitly or rely on the backend doing it (gcp
`AutoDelete` on the boot disk).
*Why:* GC sees only what `ListByTag` returns, so secondary resources have no
other cleanup path.
- **MAY return before the instance is gone** (gcp, kubernetes), as long as `Get`
keeps showing it until it is (see Get).
- **SHOULD succeed for an ID whose `Get` fails permanently** when there is
nothing left it could reach. Example: kvm on a host that was removed from the
config returns nil (`kvm.go:466-476`).
*Why:* on the deletion path a permanent error isn't latched. It keeps
retrying (`deletionFailure`, `:349-359`), so a Proxy is only released through
the `crawl.example.com/skip-cleanup` annotation.
### ListByTag(ctx) ([]Instance, error)
- **MUST return only instances that carry the managed mark**, each with an `ID`
that `Delete` accepts, plus `UID` and `CreatedAt`.
*Why:* GC deletes every listed instance whose UID isn't live and whose age is
past `MinAge` (`gc.go:111-131`). Listing something without the mark is how an
operator deletes a VM it doesn't own.
- **MUST return an error rather than a list it can't vouch for.** An error makes
GC skip that provider for the sweep (`gc.go:105-110`).
- **MAY leave out instances it couldn't read.** Leaving something out is always
safe, because GC never deletes what isn't listed. The current providers make
different choices here:
- gcp silently leaves out unreachable zones (`ReturnPartialSuccess`,
`gcp.go:229`).
- kvm fails the whole list when one host is down but skips a single unreadable
domain (`kvm.go:540-565`, `:580-596`).
- kubernetes makes a single call.
- **SHOULD cover the whole scope the provider can create in**: every namespace,
zone or host. Anything outside that scope is never collected.
- **Scope caveat:** GC protects instances by UID, not by operator install. Two
installs, or two clusters, sharing one GCP project or hypervisor will delete
each other's instances ([gcp-vm-validation.md](gcp-vm-validation.md#gotchas)).
Two provider entries of one install that share a scope are safe, because they
share the live-UID set.
## 3. Error taxonomy
Every error a provider method returns MUST be built with
`provider.Wrap(class, op, providerName, id, err)`. `op` is one of `create`,
`get`, `delete` or `list`, and `providerName` is the **configured name**
(`cfg.Name`), not the type.
`Wrap` keeps both `errors.Is(err, class)` and `errors.As` to the SDK error
working (`errors.go:53-62`). The shared metrics and tracing decorators label
results with the class (`metrics.go:61-74`, `tracing.go:67-80`).
| Class | Use it for | During normal reconciling | While deleting |
|---|---|---|---|
| `ErrNotFound` | the instance doesn't exist | Get: recreate or advance (§2); other ops: returned as the error, so exponential backoff | Get: release the finalizer |
| `ErrQuotaExceeded` | quota or rate limit | `Provisioned=False` with reason `QuotaExceeded`, requeue after `QuotaRetry` (5m), not counted as an error (`:337-340`) | requeue after `QuotaRetry` |
| `ErrTransient` | network, 5xx, timeouts, cancellation, anything unknown | error returned, so workqueue exponential backoff | same |
| `ErrPermanent` | bad config, bad argument, auth or permission denied, malformed ID | **latched**: `Provisioned=False` with reason `PermanentError`, no more provider calls until `metadata.generation` changes (`:133-139`, `:341-343`) | error returned and retried forever; Get Permanent also triggers `Delete` (`:292-302`) |
Pitfalls:
- **Unclassified means Transient.** `provider.Class` returns `ErrTransient` for
any error not built with `Wrap` (`errors.go:71-91`). That is the safe default,
but a permanent failure that was never classified retries forever, and a quota
error backs off far too fast.
- **Permanent is sticky.** Use it only for errors that editing the Proxy spec or
fixing the config can resolve. A 403 that could be a quota problem is the hard
case: gcp checks the error reason (`gcp/errors.go:31-34`), while kubernetes
can't tell quota from RBAC and picks Permanent (`kubernetes.go:201-206`).
- **Don't return NotFound from `Create`.** The reconciler treats it as Transient
there, but it misleads the metrics and anyone reading logs. kvm can currently
do this (§9).
- **Classify in one `classify` function** with a table-driven test. All three
providers do.
## 4. Identity, names and GC marks
- **Names:** `provider.NameFromUID(uid)` gives `proxy-` plus 16 lowercase base32
characters (`name.go:29-33`). It is valid as a GCP name, a DNS label and a
libvirt domain name. A backend with stricter rules SHOULD map it
deterministically rather than invent a new scheme, because idempotency depends
on the name being the same every time.
- **providerID:** opaque to everyone except the provider. Keep it
`<location>/<name>`-shaped, parse it strictly, and return `ErrPermanent` for
a malformed one (§2).
- **Marks:** `LabelManaged = "proxy-operator-managed"` with value `"true"`, and
`LabelUID = "proxy-operator-uid"` (`provider.go:28-32`). Use native labels or
tags where the backend has them. Otherwise store them in metadata that is
written in the same call that creates the instance (kvm: `metadata.go`).
- **`CreatedAt`:** read it from the backend's own creation timestamp. If the
backend has none, write it into the metadata at creation (kvm:
`kvm.go:277-298`).
- **Provider names:** keep a configured `name` a lowercase DNS-1123 label.
`spec.provider` allows at most 63 characters (`proxy_types.go:236`). The
`ProviderHealth` object is named after the provider, and the monitor skips
writing it for a name that isn't a valid object name (`monitor.go:141`,
`:248`). The metrics still work. Config loading doesn't check this yet (§9).
## 5. Optional interfaces: `Checker`
`provider.Checker` is `Check(ctx) []TargetStatus` (`check.go:37-46`).
`cmd/main.go` type-asserts it on the **unwrapped** provider (`cmd/main.go:216`)
before adding the decorators (`:219-222`). The monitor treats a nil Checker as
unsupported, not as failed (`monitor.go:212`).
- **MUST never fail as a whole.** Every failure is a `TargetStatus` with
`Ready=false`.
- **MUST return promptly once `ctx` ends.** The monitor gives the whole round
one timeout (`monitor.go:180`).
- **MUST return one entry per target, in a stable order.** Targets are what can
fail independently: the apiserver, a project, each hypervisor.
- **MUST NOT disrupt reconciles.** A check that shares a connection with
provisioning must not tear that connection down when it times out. kvm uses
`hostPool.probe` for this, not `call` (`kvm/check.go:65`, `kvm/conn.go:213-242`).
- **SHOULD use the shared reasons** (`Reachable`, `Timeout`, `Unreachable`,
`Unauthorized`, `Forbidden`, `APIError`). Add provider-specific reasons only
for failures an admin would fix differently: kvm has `SSHHandshakeFailed`,
`PoolNotFound` and others.
- **SHOULD check with the same credentials, verb and scope that `ListByTag` and
`Create` use**, and be cheap: one page, one object.
- **MAY fill `Details`** with facts useful when triaging, such as versions or
capacity. Only kvm does today.
## 6. Observability: tracing, logs and metrics
Tracing is OpenTelemetry, set up in `internal/tracing` from the standard
`OTEL_*` environment variables. With no exporter configured, spans are no-ops
and logs are unchanged ([architecture.md §10](architecture.md#10-tracing-internaltracing)).
A provider has three layers to think about; the first comes for free.
**1. Operation spans and metrics: the decorators, nothing to write.**
`cmd/main.go:219-222` wraps every provider in
`WithTracing(WithMetrics(...))`:
- `provider.WithTracing` opens one client span per call, `provider.create`,
`provider.get`, `provider.delete` or `provider.list`, with `provider.name`,
`provider.id` and `provider.result` (the error class) as attributes
(`internal/provider/tracing.go:27-80`). `ErrNotFound` keeps the span status
Ok, because polling to NotFound is expected.
- `provider.WithMetrics` counts each call as
`proxy_operator_provider_requests_total{provider,op,result}`
(`metrics.go:15-74`, `internal/metrics/metrics.go:40-43`). A `Checker` feeds
`proxy_operator_provider_target_up`, `_check_duration_seconds` and
`_check_timestamp_seconds`, labelled `{provider,target}`.
- The spans sit under `Reconcile Proxy → reconcile.managed |
reconcile.replaceInstance | reconcile.delete`, or under `gc.sweep`.
Rules:
- **MUST NOT** open its own per-operation span or count its own requests. They
would duplicate the decorators.
- **MUST pass the caller's `ctx`** to every backend call and log through
`logf.FromContext(ctx)` **at call time**. The decorator starts the span with
`tracing.StartSpan`, which puts a logger carrying `traceID` and `spanID` into
`ctx` (`internal/tracing/logger.go:32-44`). A logger cached at construction,
or a backend call made with `context.Background()`, loses the trace.
- **Health checks bypass the decorators** (`cmd/main.go:211-218`). A
`Checker` runs under the monitor's `providerhealth.check` span
(`monitor.go:176`), and the same `ctx` rules apply to it.
**2. Child spans for backend calls: SHOULD.**
One operation span says that `Create` took 40 seconds. Child spans say which
backend call took them. Every provider SHOULD produce one child span per backend
request, under the operation span:
- **HTTP backends:** wrap the SDK's transport with `otelhttp`. Gate it on an
existing parent span, as `tracing.RestConfigWrapper` does
(`internal/tracing/transport.go:12-51`). Otherwise background traffic such as
watches, token refreshes or keepalives becomes root spans. Check first whether
the SDK already does this: Google's does.
- **Non-HTTP backends** (RPC over SSH, gRPC without interceptors, CLIs): start
child spans yourself with `tracing.Start(ctx, "<type>.<call>")`, around each
RPC and around long steps such as a dial or an image download. Use a
decorator over the SDK seam, not call-site code.
- **MUST NOT** put secrets or CloudInit contents in span attributes. Put
identifiers, sizes and error codes there.
**3. Logs.** V(1): one line per operation outcome and per failed backend call.
V(2): curated detail. V(5): raw wire, if the backend has one. Log lines are
matched to traces by the `traceID` key from layer 1, which is how kvm's
per-RPC log lines join a trace today.
What each provider does now:
| Layer | kubernetes | gcp | kvm |
|---|---|---|---|
| 1. Operation spans and metrics | yes: decorators | yes: decorators | yes: decorators |
| 2. Child spans per backend call | yes: `tracing.RestConfigWrapper()` on its `rest.Config` (`cmd/main.go:196-200`, `kubernetes.go:58-60`), parent-gated | yes: the Google SDK adds `otelhttp` itself unless telemetry is disabled (`google.golang.org/api/transport/http/dial.go:308-312`); not parent-gated, but every call site has a parent | **no**: `kvm/trace.go` is a log decorator, not spans |
| 3. Logs carry `traceID`/`spanID` | n/a: the package doesn't log | yes, including the V(5) wire log (`gcp/wirelog.go:54-59`) | yes: every `libvirt call` line at V(1) |
| Tested against a real collector | yes: kind e2e checks the `provider.create` and `provider.delete` traces in Tempo (`test/e2e/tracing_test.go:200-238`), operation spans only | no | no |
## 7. Feature matrix
Legend: **yes** / **partial** / **no**, with a short note. The evidence is in
the per-provider notes below the table.
| Feature | kubernetes | gcp | kvm |
|---|---|---|---|
| providerID self-contained | yes: `ns/name` | yes: `zones/z/instances/n`; project from config | yes: `hosts/h/domains/n` |
| Create idempotent | yes: AlreadyExists = success | yes: 409 = success | yes: lookup by name; overlay reused, seed rebuilt |
| Create verifies adopted instance's UID | no | no | no |
| Create non-blocking | yes | yes: operation not awaited | **no**: waits for `DomainCreate`; first use of an image downloads it |
| Placement used | none | Zone, MachineType, Image | Zone (host), MachineType, Image |
| CloudInit delivered | **no**: ignored | yes: `user-data` metadata | yes: NoCloud seed ISO |
| Port used | yes: Squid config and containerPort | no: firewall is manual | no |
| GC marks on create | yes: Pod labels | yes: instance labels | yes: domain `<metadata>`; volumes unmarked |
| States produced | Provisioning, Running, Terminated | all four | Provisioning, Running, Stopped |
| IP source | PodIP | external NAT IP only | ARP, then guest agent; first IPv4 |
| UID / CreatedAt readback | yes | yes | yes |
| Delete idempotent | yes | yes | yes, at each step |
| Delete waits until gone | no | no | yes |
| Delete cleans secondary resources | n/a | yes: boot disk auto-deleted | yes: overlay and seed; base image kept |
| ListByTag scope | cluster-wide | one project, all zones | configured hosts |
| ListByTag on partial failure | n/a: single call | leaves out failed zones | whole list fails if a host is down; unreadable domain skipped |
| Quota detection | no | yes: 429 and 403 quota reasons | no |
| Error.Provider = configured name | **no**: hardcoded `"kubernetes"` | yes | yes |
| Checker | yes: `apiserver` | yes: the project | yes: each host |
| Checker Details | no | no | yes |
| Credentials | in-cluster config / kubeconfig | ADC | SSH key + known_hosts files, usually a Secret |
| Credential rotation without restart | n/a | via ADC | no: cached after first load |
| Connection handling | client-go | SDK client; SDK retries on Get and List | one pooled SSH+libvirt connection per host, keepalive |
| Logging | none in the package | V(1) ops, V(2) detail, wire log | V(1) ops and RPCs, V(2) detail |
| Tracing: operation spans (§6) | yes | yes | yes |
| Tracing: child spans per backend call | yes: parent-gated transport | yes: SDK's otelhttp | **no** |
| Unit tests with a fake | yes: controller-runtime fake client | yes: `fakeAPI` + httptest | yes: `fakeAPI` + in-process SSH server |
| Real-backend test | kind e2e | manual runbook | `//go:build integration` |
| Proxy software | generated Squid config (`pod.go:60-75`) | from `cloudInit`; image must run cloud-init (debian-cloud doesn't) | from `cloudInit`, plus the guest-agent vendor-data |
| IP reachable from the operator | Pod network, so in-cluster only (not from `make run-dev` on a laptop) | external IP, once a firewall rule allows the cluster's egress IP | LAN address on the host's bridge; the operator needs a route to that LAN |
| Distinct egress IP per proxy | no: the cluster's egress | yes: ephemeral external IP | whatever the host's LAN routes through; not per VM by itself |
### kubernetes notes (`internal/provider/kubernetes`)
- **ID:** `kubernetes.go:152-165`. An empty name part is not rejected.
- **Create:** `kubernetes.go:90-100` and `pod.go:23-50`. Only `Name`,
`Namespace`, `UID` and `Port` are used. Squid config is generated from the
port and passed through `SQUID_CONF` (`pod.go:38-75`).
- **State:** `stateFromPod`, `kubernetes.go:185-199`. Pending, or Running with
no PodIP, is Provisioning. Succeeded, Failed and Unknown are Terminated. A Pod
being deleted still reports its phase.
- **Delete:** `kubernetes.go:118-127`. ListByTag: `kubernetes.go:134-146`.
- **Errors:** `classify`, `kubernetes.go:201-218`. 429 and timeouts are
Transient. Forbidden, Invalid, BadRequest and Unauthorized are Permanent.
Nothing is Quota.
- **Check:** `check.go:21-53`. It lists at most one managed Pod cluster-wide, so
it proves read and list access, not create.
- **Config:** only `kubernetes.image` (`config.go:45-48`, default at
`kubernetes.go:28`).
### gcp notes (`internal/provider/gcp`)
- **ID:** `gcp.go:289-299`.
- **Create:** `gcp.go:135-167` and `insert.go:34-74`. `Operation.Wait` is never
called. There is one NIC with an ephemeral external IP, and the boot disk has
`AutoDelete`.
- **State:** `mapState`, `gcp.go:271-287`. RUNNING without a NAT IP is
Provisioning. Unknown statuses are Stopped.
- **Get:** `gcp.go:172-192`. Delete: `gcp.go:196-219`. ListByTag:
`gcp.go:224-244`.
- **Errors:** `errors.go:20-39`. Non-googleapi errors are Transient. 404 is
NotFound. 429 and 403 with reason `quotaExceeded` or `rateLimitExceeded` are
Quota. 400, 401 and other 403s are Permanent.
- **Check:** `check.go:30-78`. One `AggregatedList` page with `maxResults=1`.
- **Config:** `project` is required. `network`, `networkTag` and `diskSizeGb`
have defaults (`config.go:51-61`, `insert.go:12-30`).
- **Logging:** V(1) and V(2) tiers in `gcp.go`. The wire log in `wirelog.go`
runs at V(5), with payloads elided unless `--gcp-wire-log-full-payloads` is
set.
### kvm notes (`internal/provider/kvm`)
- **ID:** `id.go:11-21`. Host names are restricted by `config_kvm.go:108-114`.
- **Create:** `kvm.go:154-221`:
- Looks up the domain by name and starts it if needed (`ensureRunning`,
`:300-318`).
- Otherwise: base image (`image.go`), then overlay (reused if it exists,
`:225-245`), then seed (always rebuilt, `:249-275`), then define with
metadata (`:277-298`), then start.
- **Get:** `kvm.go:323-356`. States come from `mapState` (`:442-453`): nostate
is Provisioning, running or blocked without an IP is Provisioning, everything
else is Stopped.
- **Addresses:** `address` (`:397-428`) tries ARP first, then the guest agent.
- **Delete:** `kvm.go:461-538`. Destroy, undefine (with NVRAM, managed save and
snapshot metadata), then remove the overlay and seed by name. A host that is no
longer configured is a success.
- **ListByTag:** `kvm.go:545-596`.
- **Errors:** `errors.go:21-42`. No domain or no volume is NotFound. XML,
unsupported, invalid-argument, auth and access errors, and a missing pool, are
Permanent. Everything else is Transient, including SSH handshake and auth
failures. Nothing is Quota.
- **Check:** `check.go:41-139`. Runs concurrently per host through
`hostPool.probe`. Reasons are tagged by the dial stage.
- **Connections:** `conn.go`. One connection per host with a 15s keepalive.
The pool dials on first use; with provider checks on (the default), the
leader's first health check opens it at startup. On cancel, `call` closes the shared connection; `probe` abandons
the RPC and leaves the connection open.
- **Full guide:** [kvm-provider.md](kvm-provider.md).
## 8. Adding a provider: checklist
1. **Package:** create `internal/provider/<type>/` with the provider type and a
compile-time interface assertions: `var _ provider.Checker = (*Provider)(nil)`
as in `kubernetes/check.go:15`, and the same for `provider.Provider`, which
the existing providers don't assert yet.
- Put the backend SDK behind a small interface (gcp `instancesAPI`, kvm
`libvirtAPI`) so a fake can stand in for it in tests.
- Build the logger with
`logf.FromContext(ctx).WithName("<type>").WithValues("provider", name)`.
2. **Config:** in `internal/provider/config.go`:
- Add a `<Type>Config` block and a `ProviderConfig` field.
- Add the block to `setBlocks`.
- Add the type to the `switch p.Type` allow-list (`config.go:102-108`).
- Add required-field validation, in its own `config_<type>.go` if it is
large.
- Add rows to `config_test.go`.
The file is parsed with `UnmarshalStrict`, so unknown keys already fail.
3. **Constructor:** `New(ctx, provider.ProviderConfig) (provider.Provider,
error)`, registered in the `registry.Build` map in `cmd/main.go:195-205`.
- A constructor error stops the manager.
- So fail only on config errors, and **don't dial the backend here**: one
unreachable hypervisor or API must not take down the other providers (kvm
logs a warning and connects lazily, `kvm.go:89-101`).
4. **Decorators:** nothing to do. `cmd/main.go:219-222` wraps every provider in
`WithTracing(WithMetrics(...))`. Don't count requests or open per-operation
spans yourself.
5. **Contract:** implement each rule in §2, and test each one against the fake:
- repeat `Create`
- `Get` after `Delete`
- `Delete` twice
- malformed ID
- unknown state mapped to `Stopped`
- `Running` requires an IP
- `UID` and `CreatedAt` read back
- `ListByTag` excludes unmarked instances
6. **Errors:** one `classify` plus a table test covering every class, including
"not an SDK error → Transient". Use `cfg.Name` in `provider.Wrap`.
7. **Checker:** shared reasons first, a table test for the reasons, and a test
that `Check` returns promptly with an expired `ctx`.
8. **Tracing and logging** (§6):
- Child spans for backend calls: an existing SDK instrumentation, a
parent-gated transport wrapper, or a span decorator over the SDK seam.
Add a test with an in-memory span recorder
(`sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(tracetest.NewSpanRecorder()))`,
as in `internal/provider/tracing_test.go`) asserting the spans are
children of the operation span.
- Take the logger from `ctx` on every call so log lines carry `traceID`.
- V(1): one line per operation outcome, plus a `logAPIError`-style line with
the raw backend error.
- V(2): request and response detail.
- Never put CloudInit contents in logs or span attributes; add a sentinel
test for it.
9. **Real-backend test:** behind `//go:build integration`, skipped when its
environment variables are unset. It runs the whole lifecycle: create, create
again, poll until Running, list, delete, delete again, and confirm it is gone
and cleaned up. See `kvm/kvm_integration_test.go` and
[testing.md](testing.md).
10. **RBAC:** if the provider talks to the cluster, add `+kubebuilder:rbac`
markers in their own comment block, not inside a doc comment. The kubernetes
provider's Pod markers live next to the reconciler's
(`internal/controller/proxy_controller.go:86-88`); `controller-gen` scans
`./...`, so either place works. Then run `make manifests` and check
`git diff config/rbac/role.yaml`.
11. **Deployment:**
- A credentials mount in `config/manager/manager.yaml`, marked
`optional: true` so the manager still starts without it (see `kvm-ssh`).
- A commented entry in `config/manager/providers_config.yaml` and
`config/samples/providers-config.yaml`.
- A `config/samples/proxy_<type>.yaml` with a cloud-init that actually
installs a proxy on the sample image.
- Optionally, a Makefile helper for the Secret (`make kvm-ssh-secret`).
12. **Docs:**
- A `docs/<type>-provider.md` covering prerequisites, config reference,
placement mapping, what lands on the backend, the health check, and known
gaps.
- A row in the architecture.md components table, and a block in §5 and
§11. If the provider adds spans, update the span topology in §10.
- A column in the §7 matrix above.
- A row in README.md "Implemented providers".
13. **Optional:** a demo script under `docs/demo/`.
## 9. Gaps and proposed features
These are listed only. Nothing below is designed or implemented.
### Current gaps
**Across providers and the contract**
- **Adoption isn't verified.** `Create` adopts any existing instance with the
right name without checking its `LabelUID`. None of the three providers check
it.
- **The Create comment doesn't match kvm.** The contract says `Create` returns
as soon as the request is submitted (`provider.go:90-94`). kvm blocks until
the domain has started, and on first use of an image, for the whole download.
- **`Stopped` and `Terminated` are indistinguishable to the reconciler.**
kubernetes never produces Stopped, and kvm never produces Terminated.
- **Different ListByTag partial-failure policies.** gcp leaves failed parts out,
kvm fails the whole list. Both are safe for GC; the contract doesn't say which
to prefer.
- **Config is read once per process.** Adding a provider or host, or rotating
credentials, needs a manager restart.
- **Stale comments.** `provider.go:3-4` and `registry/registry.go:5-12` still
describe a `mock` subpackage, which was replaced by the kubernetes provider.
`spec.provider`'s doc (`proxy_types.go:232`, and so the generated CRD) still
gives `"mock"` as an example.
- **GC doesn't know about installations.** Two clusters sharing one GCP project
or hypervisor delete each other's instances, because GC only knows its own
live UIDs. This was a merge-blocker in the 2026-08-10 PR review
([docs/reviews](reviews/2026-08-10-1134-proxy-operator-pr-review.md)) and is
still open.
- **An unconfigured provider latches the Proxy.** A Proxy naming a provider
missing from the config is latched `PermanentError`. Adding the provider and
restarting doesn't change `metadata.generation`, and `spec.provider` is
immutable, so the Proxy stays latched until it is recreated (same review).
- **Provider names aren't validated** as DNS-1123 at config load (§4).
- **The type list exists twice:** the config allow-list (`config.go:102-108`)
and the constructor map in `cmd/main.go:195-205` must be kept in step by hand.
- **Delete is resubmitted on every deletion poll** (§2). For gcp that is a new
API operation every 10s until the VM is gone.
**kubernetes**
- **Wrong provider name in errors.** `Error.Provider` is hardcoded to
`"kubernetes"` (`kubernetes.go:105-216`), so errors and logs show the wrong
name when the configured name differs.
- **Spec fields ignored.** `CloudInit` and all of `Placement` are ignored.
- **No quota class.** A quota 403 is Permanent.
- **Deleting Pods look healthy.** A Pod being deleted still reports its phase,
so it can read as `Running`.
- **Unguarded Pods.** No securityContext, no resources, and
`http_access allow all` (`pod.go`).
- **No logging** in the package, and **no Checker `Details`**.
- **No `OwnerReference` on Pods.** A bypassed finalizer leaves the Pod behind.
- **Pod RBAC is always installed:** cluster-wide `pods` create/delete ships even
when no kubernetes provider is configured (`proxy_controller.go:86-88`).
**Tracing**
- **kvm has no child spans.** A kvm trace shows `provider.create` and nothing
under it: no libvirt RPCs, no SSH dial, no image download, although that is
where the time goes. Correlation works only through the `traceID` on its V(1)
log lines, which needs debug logging on.
- **Only kubernetes is tested against a real collector**, and that test checks
only the operation spans. Nothing checks that gcp's SDK spans or the
kubernetes API spans actually nest under `provider.*`.
**gcp**
- **Asynchronous insert failures are never read.** An insert that the API
accepts but that fails later (quota, bad image) is never seen. The next `Get`
returns NotFound and the reconciler creates it again, so the reason never
reaches status. [gcp-vm-validation.md](gcp-vm-validation.md#gotchas) says
such errors do reach status.
- **External IP only.** A VM without an external NAT IP stays `Provisioning`
forever.
- **No firewall management.** The rule for the proxy port is a manual
prerequisite, and `req.Port` is unused.
- **The SDK client is never closed**, and `Placement.Region` is unused.
- **Wrong label value in the docs.** `gcp-vm-validation.md` filters on
`labels.proxy-operator-managed=yes`; the value is `true`.
- **The sample can't work.** `config/samples/proxy_gcp.yaml` uses a
`debian-cloud` image, which has no cloud-init, so its Squid install never runs
([gcp-in-specific-project.md](gcp-in-specific-project.md) uses Ubuntu for that
reason).
**kvm** (see also [kvm-provider.md § Known gaps](kvm-provider.md#known-gaps))
- **No quota or capacity detection.**
- **Create blocks**, including the first image download on each host.
- **GC blind spots.** Volumes without a domain are invisible to GC, and so are
VMs on a host removed from the config. `Delete` treats the latter as done.
- **SSH material is cached after the first successful load**, so rotating keys
needs a restart.
- **Cancelling one operation cuts off the others on that host.** A cancelled
`call` closes the shared per-host connection, aborting other operations in
flight on that host (`conn.go:199-211`).
- **SSH auth failures retry forever.** SSH authentication and host-key failures
are Transient, while libvirt auth errors are Permanent.
- **`Create` can return `ErrNotFound`** when a volume or domain disappears
during the call.
- **No reused-overlay check.** An existing overlay is reused without checking
its backing image or size.
- **Sample mismatch.** `config/samples/proxy_kvm.yaml` places on host `nas1`,
while both providers configs define only `beelink`.
- **Doc drift in kvm-provider.md:**
- It says ARP is the fallback; the code tries ARP first.
- It says `host` and `name` are on every log line; the check and list paths
lack them.
### Proposed features (not implemented)
- **Child spans in kvm:** extend the `tracingAPI` decorator in `kvm/trace.go`,
which already wraps every libvirt call, to also start a span per RPC, and add
spans for the SSH dial and the image download. The seam's methods take no
`ctx` today (`hostPool.call` hands `fn` only the API), so the span context has
to be plumbed in.
- **Tempo e2e checks for child spans**, and for gcp and kvm through their real
backend runbooks.
- **A shared conformance suite:** for example `internal/provider/providertest`,
which runs the §2 rules against any provider plus a fake backend, so a new
provider gets the contract tests for free.
- **Verify the UID when `Create` adopts an instance.** Return `ErrPermanent` on
a mismatch instead of adopting an instance that belongs to someone else.
- **Report asynchronous create failures:** gcp keeps the operation name, and
`Get` reports the failure instead of NotFound.
- **Capacity or quota signals** for kvm, such as pool free space and host
memory, mapped to `ErrQuotaExceeded`.
- **GC for secondary resources:** mark volumes and similar resources, and let
`ListByTag` or a second listing method cover them.
- **Reload credentials and config without a restart.**
- **Checker `Details`** for kubernetes (server version) and gcp (quota
headroom).
- **Internal-IP mode** for gcp, and a firewall rule managed per provider.
- **An installation-identity mark**, set on create and filtered on in
`ListByTag`, so installations sharing a backend don't collect each other's
instances.
- **A permanent-failure latch that sees provider availability**, so adding a
missing provider releases the Proxies waiting on it.
- **Validate provider names at load**, and derive the config type allow-list
from the constructor map.
- **Per-provider RBAC** as an optional kustomize component, and an
`OwnerReference` on kubernetes Pods.
+28 -6
View File
@@ -53,18 +53,23 @@ manual runs too.
| | `config_test.go` | providers-config parsing + fail-fast validation | pure | | | `config_test.go` | providers-config parsing + fail-fast validation | pure |
| | `metrics_test.go` | `WithMetrics` decorator: classified result labels, error passthrough | fake recorder + static provider | | | `metrics_test.go` | `WithMetrics` decorator: classified result labels, error passthrough | fake recorder + static provider |
| `internal/provider/kubernetes` | `kubernetes_test.go`, `pod_test.go` | Create/Get/Delete/ListByTag over real Pod objects; pure `buildPod`/`squidConf` incl. the `max_filedescriptors` OOM-regression assertion | `client/fake` (real `corev1.Pod`s, no kubelet) | | `internal/provider/kubernetes` | `kubernetes_test.go`, `pod_test.go` | Create/Get/Delete/ListByTag over real Pod objects; pure `buildPod`/`squidConf` incl. the `max_filedescriptors` OOM-regression assertion | `client/fake` (real `corev1.Pod`s, no kubelet) |
| | `check_test.go` | health check: a single cluster-wide list with `Limit=1` and the managed-label selector; the reason for every API error class | `client/fake` + interceptor |
| `internal/provider/gcp` | `insert_test.go`, `errors_test.go`, `gcp_test.go` | field-by-field `buildInsertRequest`; HTTP-code classification table; zone-qualified IDs, 409-is-success, 9-row state mapping, ListByTag filter + `ReturnPartialSuccess` | pure builder needs **no fake**; the rest uses the flattened `instancesAPI` seam | | `internal/provider/gcp` | `insert_test.go`, `errors_test.go`, `gcp_test.go` | field-by-field `buildInsertRequest`; HTTP-code classification table; zone-qualified IDs, 409-is-success, 9-row state mapping, ListByTag filter + `ReturnPartialSuccess` | pure builder needs **no fake**; the rest uses the flattened `instancesAPI` seam |
| | `check_test.go` | health check: request shape and the reason per HTTP code or network error; the **real SDK client** makes exactly one request with `maxResults=1`, even when a next page exists | fake seam + `httptest` |
| `internal/provider/kvm` | `domainxml_test.go`, `volume_test.go`, `seed_test.go`, `id_test.go`, `metadata_test.go` | pure builders: domain/volume XML unmarshalled back and checked field by field, seed ISO reopened with the same library, providerID round-trip, metadata in both namespace spellings | pure | | `internal/provider/kvm` | `domainxml_test.go`, `volume_test.go`, `seed_test.go`, `id_test.go`, `metadata_test.go` | pure builders: domain/volume XML unmarshalled back and checked field by field, seed ISO reopened with the same library, providerID round-trip, metadata in both namespace spellings | pure |
| | `errors_test.go`, `conn_test.go` | libvirt-code classification table; connection pool (reuse, redial, drop on transport error, cancel releases a blocked RPC, one dial per host under concurrency); SSH dial, host-key policy, keepalive | fake dialer; a **real in-process `x/crypto/ssh` server** forwarding a unix socket | | | `errors_test.go`, `conn_test.go` | libvirt-code classification table; connection pool (reuse, redial, drop on transport error, cancel releases a blocked RPC, one dial per host under concurrency); SSH dial, host-key policy, keepalive | fake dialer; a **real in-process `x/crypto/ssh` server** forwarding a unix socket |
| | `image_test.go`, `kvm_test.go` | image cache (fresh fetch, ready, marker missing, checksum change/mismatch, 4xx/5xx, concurrent callers); Create resume after every partial step, leftover truncated seed, state/IP mapping, every Delete already-gone case, ListByTag across hosts / one down / unreadable domain; logging tiers (V0 silent, V1 one line per RPC, V2 XML) with cloud-init never logged | `httptest` image server + in-memory `fakeAPI` behind the `libvirtAPI` seam | | | `image_test.go`, `kvm_test.go` | image cache (fresh fetch, ready, marker missing, checksum change/mismatch, 4xx/5xx, concurrent callers); Create resume after every partial step, leftover truncated seed, state/IP mapping, every Delete already-gone case, ListByTag across hosts / one down / unreadable domain; logging tiers (V0 silent, V1 one line per RPC, V2 XML) with cloud-init never logged | `httptest` image server + in-memory `fakeAPI` behind the `libvirtAPI` seam |
| | `trace_test.go`, `conn_test.go`, `errors_test.go` | every `libvirtAPI` method traced exactly once (guarded by reflection on the interface), results and errors pass through, libvirt error fields, detail keys only at V2, upload content never logged; pool connect/drop lines, no wrapping below V1 | `fakeAPI` + `funcr` capture | | | `trace_test.go`, `conn_test.go`, `errors_test.go` | every `libvirtAPI` method traced exactly once (guarded by reflection on the interface), results and errors pass through, libvirt error fields, detail keys only at V2, upload content never logged; pool connect/drop lines, no wrapping below V1 | `fakeAPI` + `funcr` capture |
| | `kvm_integration_test.go` (`-tags=integration`) | one real VM: image fetch, Create, idempotent re-Create, Running with IP, ListByTag, Delete, pool state afterwards | a **real hypervisor** — see below | | | `check_test.go`, `conn_test.go` | health check: details, per-host verdicts in config order, hosts checked concurrently, a timeout that keeps the connection and refuses a second check while the first still waits, no waiting behind a reconcile's dial, the reason table; `probe` abandons without closing, but still drops on a transport error; waiting for a host's slot ends with the caller's deadline; an inactive or missing pool and daemon errors keep the connection; every dial stage tagged through the real SSH server, with classification unchanged | `fakeAPI`, fake dialers, in-process SSH server |
| | `kvm_integration_test.go` (`-tags=integration`) | one real VM: image fetch, Create, idempotent re-Create, Running with IP, ListByTag, Delete, pool state afterwards; `TestKVMCheck_realHost` runs the health check against the host and creates nothing | a **real hypervisor** — see below |
| `internal/health` | `probe_test.go` | probes through a **real CONNECT-capable proxy stub** to a real TLS server: tunnel success, refused CONNECT, wrong status, dead proxy, plain-http forward | `httptest` + hijacked bidirectional tunnel | | `internal/health` | `probe_test.go` | probes through a **real CONNECT-capable proxy stub** to a real TLS server: tunnel success, refused CONNECT, wrong status, dead proxy, plain-http forward | `httptest` + hijacked bidirectional tunnel |
| | `engine_test.go` | thresholds, first-verdict, latency suppression matrix, dropped-event retry, stale-probe UID guard, tick scheduling/seeding/pruning, `Start` end-to-end | fake `client.Reader` + injected `probeFn` | | | `engine_test.go` | thresholds, first-verdict, latency suppression matrix, dropped-event retry, stale-probe UID guard, tick scheduling/seeding/pruning, `Start` end-to-end | fake `client.Reader` + injected `probeFn` |
| `internal/lease` | `store_test.go` | capacity, `MaxLeases=0`, all three selection tie-breaks, target-scoped vs global cooldowns, expiry via fake clock, report-on-retained-lease, idempotent release, **40 concurrent acquires never exceeding capacity** | injectable clock; the concurrency case is why `-race` matters | | `internal/lease` | `store_test.go` | capacity, `MaxLeases=0`, all three selection tie-breaks, target-scoped vs global cooldowns, expiry via fake clock, report-on-retained-lease, idempotent release, **40 concurrent acquires never exceeding capacity** | injectable clock; the concurrency case is why `-race` matters |
| `internal/discovery` | `server_test.go` | auth matrix (incl. `/healthz` exemption), list filtering, grant shape, full 409 arithmetic, invalid TTL/body/result, idempotent release, report→cooldown→409 round trip, `Start`/shutdown | `httptest` over the real handler chain, fake cache reader, **real** `lease.Store` | | `internal/discovery` | `server_test.go` | auth matrix (incl. `/healthz` exemption), list filtering, grant shape, full 409 arithmetic, invalid TTL/body/result, idempotent release, report→cooldown→409 round trip, `Start`/shutdown | `httptest` over the real handler chain, fake cache reader, **real** `lease.Store` |
| `internal/gc` | `gc_test.go` | the true-orphan matrix (live/deleting/young/unlabelled all kept), broken-provider isolation, list-failure skips sweep, namespace guard, sweep loop | fake reader + canned-list provider | | `internal/gc` | `gc_test.go` | the true-orphan matrix (live/deleting/young/unlabelled all kept), broken-provider isolation, list-failure skips sweep, namespace guard, sweep loop | fake reader + canned-list provider |
| `internal/metrics` | `metrics_test.go` | scrape-time collectors (`GatherAndCompare`), per-proxy series cleanup via `ForgetProxy`, label counts, fresh-registry-per-test property | `prometheus/client_golang/testutil` | | `internal/providerhealth` | `status_test.go`, `monitor_test.go` | `buildStatus` (the Ready condition, per-target transition times, truncation); objects created with the label and patched, the round timeout, no-checker Unknown, invalid names, transition-only logging, cleanup of labelled unconfigured objects, the checks-off path, `ValidateSchedule` | fake client with a cluster-scoped RESTMapper + `funcr` capture |
| | `envtest_test.go` | the generated `ProviderHealth` schema against a real API server: cluster scope, status subresource, required fields, list-map keys, the 1024-character message limit | envtest (skipped under `-short`) |
| `internal/metrics` | `metrics_test.go` | scrape-time collectors (`GatherAndCompare`) incl. the provider-target gauges, per-proxy series cleanup via `ForgetProxy`, label counts, fresh-registry-per-test property | `prometheus/client_golang/testutil` |
Conventions (from `CLAUDE.md`): stdlib `testing`, table-driven by default, Conventions (from `CLAUDE.md`): stdlib `testing`, table-driven by default,
`t.Parallel()` where safe, tests next to source, ginkgo only in the `t.Parallel()` where safe, tests next to source, ginkgo only in the
@@ -89,7 +94,7 @@ envtest suite the scaffold generated.
- **Live GCP**: no test talks to Google. The seam boundary - **Live GCP**: no test talks to Google. The seam boundary
(`buildInsertRequest` + classification) is tested exhaustively instead. (`buildInsertRequest` + classification) is tested exhaustively instead.
- **Live libvirt, by default**: the KVM unit tests run over an in-memory - **Live libvirt, by default**: the KVM unit tests run over an in-memory
fake of the 17 libvirt calls the provider uses. Whether libvirt accepts fake of the 20 libvirt calls the provider uses. Whether libvirt accepts
the domain XML, whether a raw qcow2 upload plus pool refresh yields a the domain XML, whether a raw qcow2 upload plus pool refresh yields a
usable backing image, and whether the guest actually boots and reports usable backing image, and whether the guest actually boots and reports
an address are provable only on a real host — that is the opt-in an address are provable only on a real host — that is the opt-in
@@ -109,13 +114,19 @@ had.
export KVM_TEST_HOST=nas1.home.lan:22 # SSH endpoint export KVM_TEST_HOST=nas1.home.lan:22 # SSH endpoint
export KVM_TEST_USER=virt # must be in the libvirt group export KVM_TEST_USER=virt # must be in the libvirt group
export KVM_TEST_SSH_KEY=$HOME/.ssh/id_ed25519 # private key for that user export KVM_TEST_SSH_KEY=$HOME/.ssh/id_ed25519 # private key for that user
export KVM_TEST_KNOWN_HOSTS=$HOME/.ssh/kvm_known_hosts # the hypervisors' host keys
export KVM_TEST_BRIDGE=br0 # bridge to a network with DHCP export KVM_TEST_BRIDGE=br0 # bridge to a network with DHCP
export KVM_TEST_IMAGE_URL=https://cloud.debian.org/images/cloud/trixie/latest/debian-13-generic-amd64.qcow2 export KVM_TEST_IMAGE_URL=https://cloud.debian.org/images/cloud/trixie/latest/debian-13-generic-amd64.qcow2
make test-integration-kvm make test-integration-kvm
``` ```
Optional: `KVM_TEST_KNOWN_HOSTS` (without it host keys are **not** `KVM_TEST_KNOWN_HOSTS` is technically optional, but leaving it out turns
verified — the log says so), `KVM_TEST_IMAGE_SHA256`, `KVM_TEST_POOL` host-key verification **off** for the run (the log says so); keep a
`kvm_known_hosts` holding just the hypervisors rather than pointing the
test at your everyday `known_hosts`, so a re-keyed hypervisor is a
one-line fix in a file nothing else uses.
Also optional: `KVM_TEST_IMAGE_SHA256`, `KVM_TEST_POOL`
(default `default`), `KVM_TEST_SOCKET`, `KVM_TEST_TIMEOUT` (default (default `default`), `KVM_TEST_SOCKET`, `KVM_TEST_TIMEOUT` (default
`20m`, must cover the first-time image download), `KVM_TEST_KEEP` `20m`, must cover the first-time image download), `KVM_TEST_KEEP`
to leave the VM behind for `virsh console` when something goes wrong, to leave the VM behind for `virsh console` when something goes wrong,
@@ -139,13 +150,24 @@ but no per-VM overlay or seed. It also tries a TCP connect to port 22 on
the guest, but only logs the outcome — the machine running the test may the guest, but only logs the outcome — the machine running the test may
not route to the bridge network. not route to the bridge network.
`TestKVMCheck_realHost` runs the provider health check
([kvm-provider.md](kvm-provider.md#health-check)) against the same host.
It must find the host reachable and the pool running, and it creates
nothing. It needs the same variables; to run only this test:
```sh
go test -tags=integration -count=1 -v -run TestKVMCheck ./internal/provider/kvm/
```
The test is not part of `make test` and not run in CI. Keep it compiling The test is not part of `make test` and not run in CI. Keep it compiling
with `go vet -tags=integration ./...` when the provider's internals move. with `go vet -tags=integration ./...` when the provider's internals move.
## The scaffolded `make test-e2e` suite ## The scaffolded `make test-e2e` suite
`test/e2e/` is the kubebuilder-generated smoke suite (build image → kind `test/e2e/` is the kubebuilder-generated smoke suite (build image → kind
cluster → deploy → assert the manager pod runs and serves metrics). It cluster → deploy → assert the manager pod runs, publishes a Ready
`ProviderHealth` for the `kubernetes` provider, and serves metrics that
include that provider's `proxy_operator_provider_target_up` series). It
compiles only under `-tags=e2e`, manages its own kind cluster compiles only under `-tags=e2e`, manages its own kind cluster
(`make test-e2e` / `make cleanup-test-e2e`), and has been kept compiling (`make test-e2e` / `make cleanup-test-e2e`), and has been kept compiling
(`go vet -tags=e2e ./...` is part of the routine) but is **not part of (`go vet -tags=e2e ./...` is part of the routine) but is **not part of
+48
View File
@@ -117,3 +117,51 @@ func (c *constCollector) Collect(ch chan<- prometheus.Metric) {
ch <- prometheus.MustNewConstMetric(c.desc, prometheus.GaugeValue, float64(value), label) ch <- prometheus.MustNewConstMetric(c.desc, prometheus.GaugeValue, float64(value), label)
} }
} }
// ProviderTarget is one provider health-check target as the gauges see it.
type ProviderTarget struct {
Provider, Target string
Up bool
Duration time.Duration
CheckedAt time.Time
}
// RegisterProviderTargets registers the provider-check gauges. Like the
// other scrape-time collectors they are read from the source at every
// scrape, so a target that goes away stops being reported rather than
// leaving a stale series behind.
func RegisterProviderTargets(reg prometheus.Registerer, read func() []ProviderTarget) error {
labels := []string{"provider", "target"}
return reg.Register(&providerTargetCollector{
up: prometheus.NewDesc("proxy_operator_provider_target_up",
"Whether the target passed its latest provider health check (1) or not (0).", labels, nil),
duration: prometheus.NewDesc("proxy_operator_provider_check_duration_seconds",
"Duration of the target's latest provider health check.", labels, nil),
checked: prometheus.NewDesc("proxy_operator_provider_check_timestamp_seconds",
"Unix time of the target's latest provider health check.", labels, nil),
read: read,
})
}
type providerTargetCollector struct {
up, duration, checked *prometheus.Desc
read func() []ProviderTarget
}
func (c *providerTargetCollector) Describe(ch chan<- *prometheus.Desc) {
ch <- c.up
ch <- c.duration
ch <- c.checked
}
func (c *providerTargetCollector) Collect(ch chan<- prometheus.Metric) {
for _, t := range c.read() {
up := 0.0
if t.Up {
up = 1
}
ch <- prometheus.MustNewConstMetric(c.up, prometheus.GaugeValue, up, t.Provider, t.Target)
ch <- prometheus.MustNewConstMetric(c.duration, prometheus.GaugeValue, t.Duration.Seconds(), t.Provider, t.Target)
ch <- prometheus.MustNewConstMetric(c.checked, prometheus.GaugeValue, float64(t.CheckedAt.UnixNano())/1e9, t.Provider, t.Target)
}
}
+42
View File
@@ -111,3 +111,45 @@ func TestRegister_freshRegistryPerTest(t *testing.T) {
register(t, m1, nil, 0) register(t, m1, nil, 0)
register(t, m2, nil, 0) register(t, m2, nil, 0)
} }
func TestRegisterProviderTargets_scrape(t *testing.T) {
t.Parallel()
checked := time.Unix(1757602800, 0)
targets := []ProviderTarget{
{Provider: "kvm-home", Target: "nas1", Up: true, Duration: 84 * time.Millisecond, CheckedAt: checked},
{Provider: "kvm-home", Target: "nas2", Up: false, Duration: 15 * time.Second, CheckedAt: checked},
}
reg := prometheus.NewRegistry()
if err := RegisterProviderTargets(reg, func() []ProviderTarget { return targets }); err != nil {
t.Fatalf("RegisterProviderTargets: %v", err)
}
expected := `
# HELP proxy_operator_provider_check_duration_seconds Duration of the target's latest provider health check.
# TYPE proxy_operator_provider_check_duration_seconds gauge
proxy_operator_provider_check_duration_seconds{provider="kvm-home",target="nas1"} 0.084
proxy_operator_provider_check_duration_seconds{provider="kvm-home",target="nas2"} 15
# HELP proxy_operator_provider_check_timestamp_seconds Unix time of the target's latest provider health check.
# TYPE proxy_operator_provider_check_timestamp_seconds gauge
proxy_operator_provider_check_timestamp_seconds{provider="kvm-home",target="nas1"} 1.7576028e+09
proxy_operator_provider_check_timestamp_seconds{provider="kvm-home",target="nas2"} 1.7576028e+09
# HELP proxy_operator_provider_target_up Whether the target passed its latest provider health check (1) or not (0).
# TYPE proxy_operator_provider_target_up gauge
proxy_operator_provider_target_up{provider="kvm-home",target="nas1"} 1
proxy_operator_provider_target_up{provider="kvm-home",target="nas2"} 0
`
if err := testutil.GatherAndCompare(reg, strings.NewReader(expected)); err != nil {
t.Error(err)
}
}
func TestRegisterProviderTargets_emptySnapshotHasNoSeries(t *testing.T) {
t.Parallel()
reg := prometheus.NewRegistry()
if err := RegisterProviderTargets(reg, func() []ProviderTarget { return nil }); err != nil {
t.Fatalf("RegisterProviderTargets: %v", err)
}
if n, err := testutil.GatherAndCount(reg); err != nil || n != 0 {
t.Errorf("series before the first check = %d (%v), want 0", n, err)
}
}
+46
View File
@@ -0,0 +1,46 @@
package provider
import (
"context"
"time"
)
// Reasons shared by every Checker, so a listing reads the same across
// providers. Providers add their own for failures only they can tell
// apart (KVM's SSHHandshakeFailed or PoolNotActive, say).
const (
ReasonReachable = "Reachable"
ReasonTimeout = "Timeout"
ReasonUnreachable = "Unreachable"
ReasonUnauthorized = "Unauthorized"
ReasonForbidden = "Forbidden"
ReasonAPIError = "APIError"
)
// TargetStatus is one target's verdict from a Checker.
type TargetStatus struct {
// Name identifies the target within its provider: a KVM host name,
// "apiserver", a GCP project.
Name string
// Address is what the check connected to, for humans.
Address string
Ready bool
// Reason is ReasonReachable when Ready, otherwise why not.
Reason string
// Message is human-readable detail, usually the error.
Message string
Latency time.Duration
// Details are provider-specific facts gathered by the check.
Details map[string]string
}
// Checker is the optional interface of a provider that can verify its
// backend is reachable and usable without touching any instance. The
// provider health monitor type-asserts for it on the unwrapped providers,
// so it stays out of the Provider contract.
type Checker interface {
// Check returns one verdict per target, in a stable order. It never
// fails as a whole — a failure is a verdict — and must return promptly
// once ctx ends.
Check(ctx context.Context) []TargetStatus
}
+78
View File
@@ -0,0 +1,78 @@
package gcp
import (
"context"
"errors"
"net/http"
"time"
"cloud.google.com/go/compute/apiv1/computepb"
"google.golang.org/api/googleapi"
"google.golang.org/api/iterator"
"google.golang.org/protobuf/proto"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
var _ provider.Checker = (*Provider)(nil)
// computeAddress is the endpoint the SDK talks to, shown only to humans.
const computeAddress = "compute.googleapis.com"
// managedFilter selects the instances this operator created.
const managedFilter = "labels." + provider.LabelManaged + " = " + provider.LabelManagedYes
// Check asks the project for one page holding at most one managed
// instance. That exercises the credentials, the Compute API and the
// compute.instances.list permission orphan GC depends on, in a single
// request whatever the project's size. The target is the project: the
// provider config names no zones.
func (p *Provider) Check(ctx context.Context) []provider.TargetStatus {
start := time.Now()
err := p.api.Probe(ctx, &computepb.AggregatedListInstancesRequest{
Project: p.cfg.Project,
Filter: proto.String(managedFilter),
MaxResults: proto.Uint32(1),
ReturnPartialSuccess: proto.Bool(true),
})
ts := provider.TargetStatus{Name: p.cfg.Project, Address: computeAddress, Latency: time.Since(start)}
if err != nil {
logAPIError(p.logger(ctx), "check", err)
ts.Reason = checkReason(ctx, err)
ts.Message = err.Error()
return []provider.TargetStatus{ts}
}
ts.Ready = true
ts.Reason = provider.ReasonReachable
return []provider.TargetStatus{ts}
}
// Probe fetches the first page of an aggregated list and discards it. The
// SDK treats MaxResults as the page size, so AggregatedList, which drains
// the iterator, would still walk every page of the project.
func (r *realInstances) Probe(ctx context.Context, req *computepb.AggregatedListInstancesRequest) error {
_, err := r.client.AggregatedList(ctx, req).Next()
if err == iterator.Done {
return nil
}
return err
}
// checkReason maps a failed probe to a reason. A token that cannot be
// obtained never reaches the API, so it has no status code and lands on
// Unreachable, with the detail in the message.
func checkReason(ctx context.Context, err error) string {
var gerr *googleapi.Error
switch {
case ctx.Err() != nil, errors.Is(err, context.DeadlineExceeded):
return provider.ReasonTimeout
case !errors.As(err, &gerr):
return provider.ReasonUnreachable
case gerr.Code == http.StatusUnauthorized:
return provider.ReasonUnauthorized
case gerr.Code == http.StatusForbidden:
return provider.ReasonForbidden
default:
return provider.ReasonAPIError
}
}
+165
View File
@@ -0,0 +1,165 @@
package gcp
import (
"context"
"errors"
"net/http"
"net/http/httptest"
"net/url"
"sync"
"syscall"
"testing"
compute "cloud.google.com/go/compute/apiv1"
"cloud.google.com/go/compute/apiv1/computepb"
"google.golang.org/api/googleapi"
"google.golang.org/api/option"
"google.golang.org/protobuf/proto"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
func TestCheck_reachable(t *testing.T) {
t.Parallel()
api := &fakeAPI{}
p := newTestProvider(api)
got := p.Check(context.Background())
if len(got) != 1 {
t.Fatalf("Check() returned %d targets, want 1", len(got))
}
ts := got[0]
if !ts.Ready || ts.Reason != provider.ReasonReachable || ts.Message != "" {
t.Errorf("Check() = %+v, want ready and Reachable", ts)
}
if ts.Name != "my-project" || ts.Address != "compute.googleapis.com" {
t.Errorf("target = %q at %q, want my-project at compute.googleapis.com", ts.Name, ts.Address)
}
want := &computepb.AggregatedListInstancesRequest{
Project: "my-project",
Filter: proto.String("labels.proxy-operator-managed = true"),
MaxResults: proto.Uint32(1),
ReturnPartialSuccess: proto.Bool(true),
}
if !proto.Equal(api.probeReq, want) {
t.Errorf("probe request = %v, want %v", api.probeReq, want)
}
}
func TestCheck_failureReasons(t *testing.T) {
t.Parallel()
ended, cancel := context.WithCancel(context.Background())
cancel()
tests := []struct {
name string
ctx context.Context
err error
want string
}{
{"unauthorized", context.Background(), gerr(401), provider.ReasonUnauthorized},
{"forbidden", context.Background(), gerr(403, "forbidden"), provider.ReasonForbidden},
{"api disabled", context.Background(), gerr(403, "accessNotConfigured"), provider.ReasonForbidden},
{"server error", context.Background(), gerr(500), provider.ReasonAPIError},
{"deadline", context.Background(), &url.Error{
Op: "Get", URL: "https://compute.googleapis.com/", Err: context.DeadlineExceeded,
}, provider.ReasonTimeout},
{"context ended", ended, context.Canceled, provider.ReasonTimeout},
{"connection refused", context.Background(), &url.Error{
Op: "Get", URL: "https://compute.googleapis.com/", Err: syscall.ECONNREFUSED,
}, provider.ReasonUnreachable},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
p := newTestProvider(&fakeAPI{probeErr: tc.err})
got := p.Check(tc.ctx)
if len(got) != 1 {
t.Fatalf("Check() returned %d targets, want 1", len(got))
}
if got[0].Ready || got[0].Reason != tc.want || got[0].Message != tc.err.Error() {
t.Errorf("Check() = %+v, want not ready, reason %s, message %q", got[0], tc.want, tc.err.Error())
}
})
}
}
// TestRealInstances_Probe runs the real SDK client against a local server,
// since the point of Probe is how many requests the SDK makes.
func TestRealInstances_Probe(t *testing.T) {
t.Parallel()
tests := []struct {
name string
status int
body string
wantCode int
}{
{
name: "fetches only the first page",
status: http.StatusOK,
body: `{"items":{"zones/europe-west1-b":{"instances":[{"name":"proxy-a"}]}},"nextPageToken":"page-2"}`,
},
{
name: "empty project",
status: http.StatusOK,
body: `{}`,
},
{
name: "api error keeps its status",
status: http.StatusForbidden,
body: `{"error":{"code":403,"message":"Compute Engine API has not been used","errors":[{"reason":"accessNotConfigured"}]}}`,
wantCode: http.StatusForbidden,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
var (
mu sync.Mutex
reqs []*url.URL
)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
reqs = append(reqs, r.URL)
mu.Unlock()
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(tc.status)
_, _ = w.Write([]byte(tc.body))
}))
t.Cleanup(srv.Close)
ctx := context.Background()
c, err := compute.NewInstancesRESTClient(ctx, option.WithEndpoint(srv.URL), option.WithoutAuthentication())
if err != nil {
t.Fatalf("NewInstancesRESTClient: %v", err)
}
t.Cleanup(func() { _ = c.Close() })
err = (&realInstances{client: c}).Probe(ctx, &computepb.AggregatedListInstancesRequest{
Project: "my-project",
MaxResults: proto.Uint32(1),
})
var gerr *googleapi.Error
switch {
case tc.wantCode == 0 && err != nil:
t.Fatalf("Probe() = %v, want nil", err)
case tc.wantCode != 0 && (!errors.As(err, &gerr) || gerr.Code != tc.wantCode):
t.Fatalf("Probe() = %v, want a googleapi error with code %d", err, tc.wantCode)
}
mu.Lock()
defer mu.Unlock()
if len(reqs) != 1 {
t.Fatalf("server saw %d requests, want 1", len(reqs))
}
if got := reqs[0].Query().Get("maxResults"); got != "1" {
t.Errorf("maxResults = %q, want 1", got)
}
if want := "/compute/v1/projects/my-project/aggregated/instances"; reqs[0].Path != want {
t.Errorf("path = %q, want %q", reqs[0].Path, want)
}
})
}
}
+4 -3
View File
@@ -33,6 +33,8 @@ type instancesAPI interface {
Get(ctx context.Context, req *computepb.GetInstanceRequest) (*computepb.Instance, error) Get(ctx context.Context, req *computepb.GetInstanceRequest) (*computepb.Instance, error)
Delete(ctx context.Context, req *computepb.DeleteInstanceRequest) (opName string, err error) Delete(ctx context.Context, req *computepb.DeleteInstanceRequest) (opName string, err error)
AggregatedList(ctx context.Context, req *computepb.AggregatedListInstancesRequest) ([]*computepb.Instance, error) AggregatedList(ctx context.Context, req *computepb.AggregatedListInstancesRequest) ([]*computepb.Instance, error)
// Probe fetches only the first page of an aggregated list; see Check.
Probe(ctx context.Context, req *computepb.AggregatedListInstancesRequest) error
} }
// realInstances adapts *compute.InstancesClient to the seam. // realInstances adapts *compute.InstancesClient to the seam.
@@ -221,17 +223,16 @@ func (p *Provider) Delete(ctx context.Context, providerID string) error {
// entire GC sweep. // entire GC sweep.
func (p *Provider) ListByTag(ctx context.Context) ([]provider.Instance, error) { func (p *Provider) ListByTag(ctx context.Context) ([]provider.Instance, error) {
log := p.logger(ctx) log := p.logger(ctx)
filter := fmt.Sprintf("labels.%s = %s", provider.LabelManaged, provider.LabelManagedYes)
instances, err := p.api.AggregatedList(ctx, &computepb.AggregatedListInstancesRequest{ instances, err := p.api.AggregatedList(ctx, &computepb.AggregatedListInstancesRequest{
Project: p.cfg.Project, Project: p.cfg.Project,
Filter: proto.String(filter), Filter: proto.String(managedFilter),
ReturnPartialSuccess: proto.Bool(true), ReturnPartialSuccess: proto.Bool(true),
}) })
if err != nil { if err != nil {
logAPIError(log, "list", err) logAPIError(log, "list", err)
return nil, p.wrapErr("list", "", err) return nil, p.wrapErr("list", "", err)
} }
log.V(1).Info("GCP instances listed", "filter", filter, "count", len(instances)) log.V(1).Info("GCP instances listed", "filter", managedFilter, "count", len(instances))
out := make([]provider.Instance, 0, len(instances)) out := make([]provider.Instance, 0, len(instances))
for _, inst := range instances { for _, inst := range instances {
conv := toInstance(inst, lastPathSegment(inst.GetZone())) conv := toInstance(inst, lastPathSegment(inst.GetZone()))
+8
View File
@@ -31,6 +31,14 @@ type fakeAPI struct {
listReq *computepb.AggregatedListInstancesRequest listReq *computepb.AggregatedListInstancesRequest
listInsts []*computepb.Instance listInsts []*computepb.Instance
listErr error listErr error
probeReq *computepb.AggregatedListInstancesRequest
probeErr error
}
func (f *fakeAPI) Probe(_ context.Context, req *computepb.AggregatedListInstancesRequest) error {
f.probeReq = req
return f.probeErr
} }
func (f *fakeAPI) Insert(_ context.Context, req *computepb.InsertInstanceRequest) (string, error) { func (f *fakeAPI) Insert(_ context.Context, req *computepb.InsertInstanceRequest) (string, error) {
+53
View File
@@ -0,0 +1,53 @@
package kubernetes
import (
"context"
"errors"
"time"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"sigs.k8s.io/controller-runtime/pkg/client"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
var _ provider.Checker = (*Provider)(nil)
// Check lists at most one managed Pod across the cluster: the same client,
// verb and scope ListByTag and orphan GC rely on, so a pass means the API
// server answers and the operator's RBAC still covers them. It does not
// prove create or delete rights.
func (p *Provider) Check(ctx context.Context) []provider.TargetStatus {
start := time.Now()
var pods corev1.PodList
err := p.client.List(ctx, &pods, client.MatchingLabels{
provider.LabelManaged: provider.LabelManagedYes,
}, client.Limit(1))
ts := provider.TargetStatus{Name: "apiserver", Address: p.apiHost, Latency: time.Since(start)}
if err != nil {
ts.Reason = checkReason(ctx, err)
ts.Message = err.Error()
return []provider.TargetStatus{ts}
}
ts.Ready = true
ts.Reason = provider.ReasonReachable
return []provider.TargetStatus{ts}
}
func checkReason(ctx context.Context, err error) string {
var status apierrors.APIStatus
switch {
case ctx.Err() != nil, errors.Is(err, context.DeadlineExceeded),
apierrors.IsTimeout(err), apierrors.IsServerTimeout(err):
return provider.ReasonTimeout
case apierrors.IsUnauthorized(err):
return provider.ReasonUnauthorized
case apierrors.IsForbidden(err):
return provider.ReasonForbidden
case errors.As(err, &status):
return provider.ReasonAPIError
default:
return provider.ReasonUnreachable
}
}
+114
View File
@@ -0,0 +1,114 @@
package kubernetes
import (
"context"
"errors"
"net/url"
"syscall"
"testing"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
type listFunc func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error
func newCheckProvider(list listFunc) *Provider {
scheme := runtime.NewScheme()
if err := corev1.AddToScheme(scheme); err != nil {
panic(err)
}
c := fake.NewClientBuilder().WithScheme(scheme).WithInterceptorFuncs(interceptor.Funcs{List: list}).Build()
p := newWithClient(c, provider.ProviderConfig{Name: "kubernetes"})
p.apiHost = "https://10.96.0.1:443"
return p
}
func TestCheck_reachable(t *testing.T) {
t.Parallel()
p := newTestProvider(testPod("ns", "a", "uid-a", corev1.PodRunning, "10.0.0.1"))
p.apiHost = "https://10.96.0.1:443"
got := p.Check(context.Background())
if len(got) != 1 {
t.Fatalf("Check() returned %d targets, want 1", len(got))
}
ts := got[0]
if !ts.Ready || ts.Reason != provider.ReasonReachable || ts.Message != "" {
t.Errorf("Check() = %+v, want ready and Reachable", ts)
}
if ts.Name != "apiserver" || ts.Address != "https://10.96.0.1:443" {
t.Errorf("target = %q at %q, want apiserver at https://10.96.0.1:443", ts.Name, ts.Address)
}
}
func TestCheck_listOptions(t *testing.T) {
t.Parallel()
var got client.ListOptions
p := newCheckProvider(func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error {
got.ApplyOptions(opts)
return c.List(ctx, list, opts...)
})
p.Check(context.Background())
if got.Limit != 1 {
t.Errorf("Limit = %d, want 1", got.Limit)
}
if got.Namespace != "" {
t.Errorf("Namespace = %q, want cluster-wide", got.Namespace)
}
want := provider.LabelManaged + "=" + provider.LabelManagedYes
if got.LabelSelector == nil || got.LabelSelector.String() != want {
t.Errorf("LabelSelector = %v, want %s", got.LabelSelector, want)
}
}
func TestCheck_failureReasons(t *testing.T) {
t.Parallel()
pods := schema.GroupResource{Resource: "pods"}
ended, cancel := context.WithCancel(context.Background())
cancel()
tests := []struct {
name string
ctx context.Context
err error
want string
}{
{"forbidden", context.Background(), apierrors.NewForbidden(pods, "", errors.New("rbac denied")), provider.ReasonForbidden},
{"unauthorized", context.Background(), apierrors.NewUnauthorized("token expired"), provider.ReasonUnauthorized},
{"api timeout", context.Background(), apierrors.NewTimeoutError("slow etcd", 1), provider.ReasonTimeout},
{"server timeout", context.Background(), apierrors.NewServerTimeout(pods, "list", 1), provider.ReasonTimeout},
{"context ended", ended, context.Canceled, provider.ReasonTimeout},
{"internal error", context.Background(), apierrors.NewInternalError(errors.New("etcd down")), provider.ReasonAPIError},
{"connection refused", context.Background(), &url.Error{
Op: "Get", URL: "https://10.96.0.1:443/api/v1/pods", Err: syscall.ECONNREFUSED,
}, provider.ReasonUnreachable},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
p := newCheckProvider(func(context.Context, client.WithWatch, client.ObjectList, ...client.ListOption) error {
return tc.err
})
got := p.Check(tc.ctx)
if len(got) != 1 {
t.Fatalf("Check() returned %d targets, want 1", len(got))
}
if got[0].Ready || got[0].Reason != tc.want || got[0].Message != tc.err.Error() {
t.Errorf("Check() = %+v, want not ready, reason %s, message %q", got[0], tc.want, tc.err.Error())
}
})
}
}
+5 -1
View File
@@ -36,6 +36,8 @@ const defaultImage = "ubuntu/squid:6.6-24.04_edge"
type Provider struct { type Provider struct {
client client.Client client client.Client
image string image string
// apiHost is the API server URL, shown by Check; empty in tests.
apiHost string
} }
// New builds a kubernetes Provider from its config block. Satisfies // New builds a kubernetes Provider from its config block. Satisfies
@@ -64,7 +66,9 @@ func NewWithTransportWrapper(_ context.Context, cfg provider.ProviderConfig, wra
if err != nil { if err != nil {
return nil, fmt.Errorf("kubernetes provider %q: building client: %w", cfg.Name, err) return nil, fmt.Errorf("kubernetes provider %q: building client: %w", cfg.Name, err)
} }
return newWithClient(c, cfg), nil p := newWithClient(c, cfg)
p.apiHost = restCfg.Host
return p, nil
} }
// newWithClient builds a Provider around an already-constructed client, // newWithClient builds a Provider around an already-constructed client,
+166
View File
@@ -0,0 +1,166 @@
package kvm
import (
"context"
"errors"
"fmt"
"sync"
"time"
libvirt "github.com/digitalocean/go-libvirt"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
var _ provider.Checker = (*Provider)(nil)
// Reasons only the KVM check can tell apart; the rest are provider-wide.
const (
reasonSSHMaterialUnavailable = "SSHMaterialUnavailable"
reasonSSHHandshakeFailed = "SSHHandshakeFailed"
reasonLibvirtConnectFailed = "LibvirtConnectFailed"
reasonLibvirtError = "LibvirtError"
reasonPoolNotFound = "PoolNotFound"
reasonPoolNotActive = "PoolNotActive"
)
// hostFacts is what a successful check learns about a host.
type hostFacts struct {
version uint64
hostname string
poolState libvirt.StoragePoolState
capacity uint64
available uint64
}
// Check verifies every configured host concurrently, in kvm.hosts order.
// It goes through the same pooled connection Create and Get use, so the
// SSH and libvirt handshakes only happen on the first check or after the
// link dropped; then two cheap RPCs prove the daemon answers, and the
// storage pool VMs are created in must exist and be running.
func (p *Provider) Check(ctx context.Context) []provider.TargetStatus {
out := make([]provider.TargetStatus, len(p.cfg.Hosts))
var wg sync.WaitGroup
for i, h := range p.cfg.Hosts {
wg.Go(func() { out[i] = p.checkHost(ctx, h) })
}
wg.Wait()
return out
}
// checkHost goes through probe, so a check that runs out of time abandons
// its RPCs rather than closing the connection the reconciles share. fn
// returns the RPC errors untouched, since a transport error still drops the
// link; a pool that exists but is stopped says nothing about the link, so
// that verdict is reached afterwards. facts is read only after probe
// returned nil, since an abandoned fn may still write it.
func (p *Provider) checkHost(ctx context.Context, host provider.KVMHost) provider.TargetStatus {
ts := provider.TargetStatus{
Name: host.Name,
Address: host.Address,
Details: map[string]string{"user": host.User, "pool": host.Pool},
}
start := time.Now()
var facts hostFacts
err := p.hosts.probe(ctx, p.logger(ctx), host, func(api libvirtAPI) error {
version, err := api.ConnectGetLibVersion()
if err != nil {
return err
}
hostname, err := api.ConnectGetHostname()
if err != nil {
return err
}
pool, err := api.StoragePoolLookupByName(host.Pool)
if err != nil {
return err
}
state, capacity, _, available, err := api.StoragePoolGetInfo(pool)
if err != nil {
return err
}
facts = hostFacts{
version: version,
hostname: hostname,
poolState: libvirt.StoragePoolState(state),
capacity: capacity,
available: available,
}
return nil
})
ts.Latency = time.Since(start)
if err != nil {
ts.Reason = checkReason(err)
ts.Message = err.Error()
return ts
}
ts.Details["hostname"] = facts.hostname
ts.Details["libvirtVersion"] = formatLibVersion(facts.version)
ts.Details["poolState"] = poolStateName(facts.poolState)
ts.Details["poolCapacity"] = formatGiB(facts.capacity)
ts.Details["poolAvailable"] = formatGiB(facts.available)
if facts.poolState != libvirt.StoragePoolRunning {
ts.Reason = reasonPoolNotActive
ts.Message = fmt.Sprintf("storage pool %q is %s", host.Pool, poolStateName(facts.poolState))
return ts
}
ts.Ready = true
ts.Reason = provider.ReasonReachable
return ts
}
// checkReason names where a failed check broke. The dial stage wins over
// a timeout: "the SSH handshake did not finish" says more than "the check
// ran out of time", and the message still carries the deadline.
func checkReason(err error) string {
var de *dialError
if errors.As(err, &de) {
switch de.stage {
case stageSSHMaterial:
return reasonSSHMaterialUnavailable
case stageSSHHandshake:
return reasonSSHHandshakeFailed
case stageLibvirtConnect:
return reasonLibvirtConnectFailed
default:
return provider.ReasonUnreachable
}
}
switch {
case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled), errors.Is(err, errProbeBusy):
return provider.ReasonTimeout
case hasCode(err, libvirt.ErrNoStoragePool):
return reasonPoolNotFound
case isLibvirtError(err):
return reasonLibvirtError
default:
return provider.ReasonUnreachable
}
}
// formatLibVersion decodes libvirt's major*1e6 + minor*1e3 + micro.
func formatLibVersion(v uint64) string {
return fmt.Sprintf("%d.%d.%d", v/1_000_000, v/1_000%1_000, v%1_000)
}
// poolStateName uses virsh's words for a pool state.
func poolStateName(s libvirt.StoragePoolState) string {
switch s {
case libvirt.StoragePoolInactive:
return "inactive"
case libvirt.StoragePoolBuilding:
return "building"
case libvirt.StoragePoolRunning:
return "running"
case libvirt.StoragePoolDegraded:
return "degraded"
case libvirt.StoragePoolInaccessible:
return "inaccessible"
default:
return fmt.Sprintf("state %d", s)
}
}
func formatGiB(bytes uint64) string {
return fmt.Sprintf("%.1fGiB", float64(bytes)/(1<<30))
}
+319
View File
@@ -0,0 +1,319 @@
package kvm
import (
"context"
"errors"
"fmt"
"io"
"maps"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
libvirt "github.com/digitalocean/go-libvirt"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
// withoutLatency zeroes the one field a test cannot predict.
func withoutLatency(in []provider.TargetStatus) []provider.TargetStatus {
out := make([]provider.TargetStatus, len(in))
for i, ts := range in {
ts.Latency = 0
out[i] = ts
}
return out
}
func TestCheck_reachable(t *testing.T) {
t.Parallel()
api := newFakeAPI("default")
api.poolCapacity = 10 << 30
api.pools["default"]["disk.qcow2"] = &fakeVol{capacity: 4 << 30}
p := newTestProvider(hostFakes{"nas1": api}, "nas1")
got := withoutLatency(p.Check(context.Background()))
want := provider.TargetStatus{
Name: "nas1",
Address: "nas1:22",
Ready: true,
Reason: provider.ReasonReachable,
Details: map[string]string{
"user": "virt",
"pool": "default",
"hostname": "fake-host",
"libvirtVersion": "10.2.0",
"poolState": "running",
"poolCapacity": "10.0GiB",
"poolAvailable": "6.0GiB",
},
}
if len(got) != 1 {
t.Fatalf("Check() returned %d targets, want 1", len(got))
}
if got[0].Name != want.Name || got[0].Address != want.Address || got[0].Ready != want.Ready ||
got[0].Reason != want.Reason || got[0].Message != "" || !maps.Equal(got[0].Details, want.Details) {
t.Errorf("Check() = %+v, want %+v", got[0], want)
}
}
func TestCheck_perHostVerdicts(t *testing.T) {
t.Parallel()
p := newTestProvider(hostFakes{"nas1": newFakeAPI("default"), "nas3": newFakeAPI("default")}, "nas1", "nas2", "nas3")
got := p.Check(context.Background())
if len(got) != 3 {
t.Fatalf("Check() returned %d targets, want 3", len(got))
}
for i, want := range []struct {
name string
ready bool
}{{"nas1", true}, {"nas2", false}, {"nas3", true}} {
if got[i].Name != want.name || got[i].Ready != want.ready {
t.Errorf("target %d = %s ready=%v, want %s ready=%v", i, got[i].Name, got[i].Ready, want.name, want.ready)
}
}
if got[1].Reason != provider.ReasonUnreachable || !strings.Contains(got[1].Message, errHostOff.Error()) {
t.Errorf("unreachable host = %s %q, want Unreachable carrying %q", got[1].Reason, got[1].Message, errHostOff)
}
if got[1].Details["pool"] != "default" || got[1].Details["hostname"] != "" {
t.Errorf("unreachable host details = %v, want the configured pool and nothing learned", got[1].Details)
}
}
// TestCheck_libvirtVerdicts also checks twice per case: none of these
// verdicts is evidence against the link, so the connection must survive.
func TestCheck_libvirtVerdicts(t *testing.T) {
t.Parallel()
tests := []struct {
name string
setup func(api *fakeAPI)
pools []string
wantReason string
wantMessage string
}{
{
name: "pool inactive",
pools: []string{"default"},
setup: func(api *fakeAPI) { api.poolStates["default"] = libvirt.StoragePoolInactive },
wantReason: reasonPoolNotActive,
wantMessage: `storage pool "default" is inactive`,
},
{
name: "pool missing",
pools: []string{"other"},
wantReason: reasonPoolNotFound,
wantMessage: "no storage pool with matching name 'default'",
},
{
name: "daemon error",
pools: []string{"default"},
setup: func(api *fakeAPI) { api.fail["ConnectGetHostname"] = lvError(libvirt.ErrInternalError, "boom") },
wantReason: reasonLibvirtError,
wantMessage: "boom",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
api := newFakeAPI(tc.pools...)
if tc.setup != nil {
tc.setup(api)
}
var dials atomic.Int32
dial := func(ctx context.Context, h provider.KVMHost) (*hostClient, error) {
dials.Add(1)
return hostFakes{"nas1": api}.dial(ctx, h)
}
pc := testConfig("nas1")
p := newWithDialer(pc.Name, *pc.KVM, dial, nil)
for range 2 {
got := p.Check(context.Background())
if got[0].Ready || got[0].Reason != tc.wantReason || !strings.Contains(got[0].Message, tc.wantMessage) {
t.Errorf("Check() = %+v, want not ready, %s, message containing %q", got[0], tc.wantReason, tc.wantMessage)
}
}
if n := dials.Load(); n != 1 {
t.Errorf("dialed %d times over two checks, want 1", n)
}
})
}
}
func TestCheck_hostsCheckedConcurrently(t *testing.T) {
t.Parallel()
nas2Dialing := make(chan struct{})
dial := func(ctx context.Context, h provider.KVMHost) (*hostClient, error) {
switch h.Name {
case "nas1":
select {
case <-nas2Dialing:
case <-ctx.Done():
return nil, ctx.Err()
}
case "nas2":
close(nas2Dialing)
}
return &hostClient{api: newFakeAPI("default"), close: func() {}}, nil
}
pc := testConfig("nas1", "nas2")
p := newWithDialer(pc.Name, *pc.KVM, dial, nil)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
got := p.Check(ctx)
for _, ts := range got {
if !ts.Ready {
t.Errorf("%s = %s %q, want ready: a sequential check would time out nas1", ts.Name, ts.Reason, ts.Message)
}
}
}
// A daemon that stopped answering times the check out, but the connection
// the reconciles share stays up, and the next check does not stack a
// second hung RPC on top of the first.
func TestCheck_timeoutKeepsTheConnection(t *testing.T) {
t.Parallel()
api := newFakeAPI("default")
block := make(chan struct{})
t.Cleanup(func() { close(block) })
blocked := &blockingAPI{fakeAPI: api, block: block}
var closed atomic.Bool
dial := func(context.Context, provider.KVMHost) (*hostClient, error) {
return &hostClient{api: blocked, close: func() { closed.Store(true) }}, nil
}
pc := testConfig("nas1")
p := newWithDialer(pc.Name, *pc.KVM, dial, nil)
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
got := p.Check(ctx)
if got[0].Ready || got[0].Reason != provider.ReasonTimeout {
t.Errorf("Check() = %+v, want Timeout", got[0])
}
if closed.Load() {
t.Error("the timed-out check closed the connection the reconciles share")
}
again := p.Check(context.Background())
if again[0].Reason != provider.ReasonTimeout || !strings.Contains(again[0].Message, "still waiting") {
t.Errorf("second Check() = %+v, want an immediate Timeout while the first still waits", again[0])
}
}
// A reconcile redialing a dead host holds the host's slot for minutes; the
// check must give up at its own deadline rather than wait that out.
func TestCheck_doesNotWaitBehindAReconcileDial(t *testing.T) {
t.Parallel()
release := make(chan struct{})
dialing := make(chan struct{})
var once sync.Once
dial := func(context.Context, provider.KVMHost) (*hostClient, error) {
once.Do(func() { close(dialing) })
<-release
return nil, errHostOff
}
pc := testConfig("nas1")
p := newWithDialer(pc.Name, *pc.KVM, dial, nil)
reconcile := make(chan error, 1)
go func() {
_, err := p.Get(context.Background(), formatProviderID("nas1", testName))
reconcile <- err
}()
<-dialing
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
start := time.Now()
got := p.Check(ctx)
if waited := time.Since(start); waited > time.Second {
t.Errorf("Check() took %s behind the reconcile's dial, want about 50ms", waited)
}
if got[0].Reason != provider.ReasonTimeout {
t.Errorf("Check() = %+v, want Timeout", got[0])
}
close(release)
<-reconcile
}
// blockingAPI hangs in ConnectGetLibVersion until block closes, like a
// daemon that stopped answering.
type blockingAPI struct {
*fakeAPI
block chan struct{}
}
func (b *blockingAPI) ConnectGetLibVersion() (uint64, error) {
<-b.block
return 0, io.EOF
}
func TestCheckReason(t *testing.T) {
t.Parallel()
tests := []struct {
name string
err error
want string
}{
{"ssh material", &dialError{stage: stageSSHMaterial, err: fmt.Errorf("%w: parsing key", provider.ErrPermanent)}, reasonSSHMaterialUnavailable},
{"tcp", &dialError{stage: stageTCP, err: errors.New("ssh dial: connection refused")}, provider.ReasonUnreachable},
{"ssh handshake", &dialError{stage: stageSSHHandshake, err: errors.New("ssh handshake: unable to authenticate")}, reasonSSHHandshakeFailed},
{"libvirt connect", &dialError{stage: stageLibvirtConnect, err: errors.New("libvirt connect via /sock: EOF")}, reasonLibvirtConnectFailed},
{"stage survives the pool's wrapping", fmt.Errorf("connecting to host nas1 (nas1:22): %w",
&dialError{stage: stageSSHHandshake, err: context.DeadlineExceeded}), reasonSSHHandshakeFailed},
{"deadline", context.DeadlineExceeded, provider.ReasonTimeout},
{"cancelled", context.Canceled, provider.ReasonTimeout},
{"previous check still waiting", errProbeBusy, provider.ReasonTimeout},
{"no pool", lvError(libvirt.ErrNoStoragePool, "no pool"), reasonPoolNotFound},
{"other libvirt error", lvError(libvirt.ErrInternalError, "boom"), reasonLibvirtError},
{"link died mid-call", io.EOF, provider.ReasonUnreachable},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
if got := checkReason(tc.err); got != tc.want {
t.Errorf("checkReason(%v) = %s, want %s", tc.err, got, tc.want)
}
})
}
}
func TestDialError_keepsClassification(t *testing.T) {
t.Parallel()
err := fmt.Errorf("connecting to host nas1: %w",
&dialError{stage: stageSSHMaterial, err: fmt.Errorf("%w: parsing kvm.sshKeyFile: bad", provider.ErrPermanent)})
if got := classify(err); !errors.Is(got, provider.ErrPermanent) {
t.Errorf("classify() = %v, want ErrPermanent through the dialError", got)
}
if !isClassified(err) {
t.Error("isClassified() = false, want true")
}
if want := "connecting to host nas1: permanent: parsing kvm.sshKeyFile: bad"; !strings.HasSuffix(err.Error(), "parsing kvm.sshKeyFile: bad") {
t.Errorf("message = %q, want it unchanged, like %q", err.Error(), want)
}
}
func TestFormatLibVersion(t *testing.T) {
t.Parallel()
tests := []struct {
in uint64
want string
}{
{10002000, "10.2.0"},
{11006001, "11.6.1"},
{9000000, "9.0.0"},
}
for _, tc := range tests {
if got := formatLibVersion(tc.in); got != tc.want {
t.Errorf("formatLibVersion(%d) = %s, want %s", tc.in, got, tc.want)
}
}
}
+111 -16
View File
@@ -9,6 +9,7 @@ import (
"net" "net"
"os" "os"
"sync" "sync"
"sync/atomic"
"time" "time"
libvirt "github.com/digitalocean/go-libvirt" libvirt "github.com/digitalocean/go-libvirt"
@@ -44,6 +45,9 @@ type libvirtAPI interface {
StorageVolUpload(vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error StorageVolUpload(vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error
StorageVolDownload(vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error StorageVolDownload(vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error
StorageVolDelete(vol libvirt.StorageVol, flags libvirt.StorageVolDeleteFlags) error StorageVolDelete(vol libvirt.StorageVol, flags libvirt.StorageVolDeleteFlags) error
ConnectGetLibVersion() (uint64, error)
ConnectGetHostname() (string, error)
StoragePoolGetInfo(pool libvirt.StoragePool) (state uint8, capacity, allocation, available uint64, err error)
} }
// hostClient is one live connection to a hypervisor. // hostClient is one live connection to a hypervisor.
@@ -68,6 +72,27 @@ func (c *hostClient) alive() bool {
// real one; tests inject their own. // real one; tests inject their own.
type hostDialer func(ctx context.Context, host provider.KVMHost) (*hostClient, error) type hostDialer func(ctx context.Context, host provider.KVMHost) (*hostClient, error)
// dialStage names the step of a connection attempt that failed, so the
// health check can say why a host is unreachable without parsing messages.
type dialStage string
const (
stageSSHMaterial dialStage = "ssh-material"
stageTCP dialStage = "tcp"
stageSSHHandshake dialStage = "ssh-handshake"
stageLibvirtConnect dialStage = "libvirt-connect"
)
// dialError tags a dial failure with its stage. It keeps the message and
// unwraps to the cause, so classify sees exactly what it saw before.
type dialError struct {
stage dialStage
err error
}
func (e *dialError) Error() string { return e.err.Error() }
func (e *dialError) Unwrap() error { return e.err }
// hostPool caches one connection per host, dialing lazily and redialing // hostPool caches one connection per host, dialing lazily and redialing
// after the connection drops. Dials for different hosts don't block each // after the connection drops. Dials for different hosts don't block each
// other; concurrent callers for the same host share a single dial. // other; concurrent callers for the same host share a single dial.
@@ -78,9 +103,16 @@ type hostPool struct {
entries map[string]*hostEntry entries map[string]*hostEntry
} }
// hostEntry is one host's slot in the pool. lock is a one-slot semaphore
// rather than a mutex so that waiting for it can end with the caller's
// context: whoever holds it may be dialing a dead host for as long as the
// dial takes.
type hostEntry struct { type hostEntry struct {
mu sync.Mutex lock chan struct{}
client *hostClient client *hostClient
// probing is set while a probe's fn runs, including one whose caller
// has already given up on it.
probing atomic.Bool
} }
func newHostPool(dial hostDialer) *hostPool { func newHostPool(dial hostDialer) *hostPool {
@@ -92,16 +124,37 @@ func (p *hostPool) entry(name string) *hostEntry {
defer p.mu.Unlock() defer p.mu.Unlock()
e, ok := p.entries[name] e, ok := p.entries[name]
if !ok { if !ok {
e = &hostEntry{} e = &hostEntry{lock: make(chan struct{}, 1)}
p.entries[name] = e p.entries[name] = e
} }
return e return e
} }
// acquire takes e.lock. Only waiting is bounded by ctx: a free slot is
// taken even when ctx has already ended, so an uncontended caller behaves
// as it did with a mutex, and a contended one gives up with ctx's error.
func (e *hostEntry) acquire(ctx context.Context) error {
select {
case e.lock <- struct{}{}:
return nil
default:
}
select {
case e.lock <- struct{}{}:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (e *hostEntry) release() { <-e.lock }
func (p *hostPool) get(ctx context.Context, log logr.Logger, host provider.KVMHost) (*hostClient, error) { func (p *hostPool) get(ctx context.Context, log logr.Logger, host provider.KVMHost) (*hostClient, error) {
e := p.entry(host.Name) e := p.entry(host.Name)
e.mu.Lock() if err := e.acquire(ctx); err != nil {
defer e.mu.Unlock() return nil, fmt.Errorf("waiting for the connection to host %s (%s): %w", host.Name, host.Address, err)
}
defer e.release()
if e.client != nil { if e.client != nil {
if e.client.alive() { if e.client.alive() {
return e.client, nil return e.client, nil
@@ -125,8 +178,8 @@ func (p *hostPool) get(ctx context.Context, log logr.Logger, host provider.KVMHo
// connection that was already replaced by a redial is left alone. // connection that was already replaced by a redial is left alone.
func (p *hostPool) drop(host string, c *hostClient) { func (p *hostPool) drop(host string, c *hostClient) {
e := p.entry(host) e := p.entry(host)
e.mu.Lock() e.lock <- struct{}{}
defer e.mu.Unlock() defer e.release()
if e.client == c { if e.client == c {
e.client = nil e.client = nil
} }
@@ -149,19 +202,61 @@ func (p *hostPool) call(ctx context.Context, log logr.Logger, host provider.KVMH
if err != nil { if err != nil {
return err return err
} }
api := c.api err = awaitOrCancel(ctx, func() error { return fn(traced(c, log)) }, func() {
if log.V(1).Enabled() {
api = &tracingAPI{inner: c.api, log: log}
}
err = awaitOrCancel(ctx, func() error { return fn(api) }, func() {
log.V(1).Info("libvirt connection dropped", "reason", "context cancelled", "error", ctx.Err().Error()) log.V(1).Info("libvirt connection dropped", "reason", "context cancelled", "error", ctx.Err().Error())
p.drop(host.Name, c) p.drop(host.Name, c)
}) })
p.dropOnTransportError(ctx, log, host, c, err)
return err
}
// errProbeBusy refuses a probe while the previous one on the same host has
// not come back yet.
var errProbeBusy = errors.New("the previous check of this host is still waiting for libvirt to answer")
// probe is call for the health check, which must never cost the reconciles
// their connection: when ctx ends first, fn is abandoned and the connection
// left alone. A link that is really dead is closed by the keepalive, which
// ends fn too. At most one probe per host runs at a time, an abandoned one
// included, so a daemon that stopped answering cannot pile up goroutines;
// until it answers, further probes get errProbeBusy.
func (p *hostPool) probe(ctx context.Context, log logr.Logger, host provider.KVMHost, fn func(api libvirtAPI) error) error {
log = log.WithName("libvirt")
e := p.entry(host.Name)
if !e.probing.CompareAndSwap(false, true) {
return errProbeBusy
}
c, err := p.get(ctx, log, host)
if err != nil {
e.probing.Store(false)
return err
}
err = awaitOrCancel(ctx, func() error {
defer e.probing.Store(false)
return fn(traced(c, log))
}, func() {
log.V(1).Info("libvirt probe abandoned, connection kept", "error", ctx.Err().Error())
})
p.dropOnTransportError(ctx, log, host, c, err)
return err
}
// traced returns c's API, logging every RPC when log has V(1) enabled.
func traced(c *hostClient, log logr.Logger) libvirtAPI {
if log.V(1).Enabled() {
return &tracingAPI{inner: c.api, log: log}
}
return c.api
}
// dropOnTransportError drops c when err is evidence against the link:
// neither a libvirt error, nor a verdict already classified, nor the
// caller's own context ending.
func (p *hostPool) dropOnTransportError(ctx context.Context, log logr.Logger, host provider.KVMHost, c *hostClient, err error) {
if err != nil && !isLibvirtError(err) && !isClassified(err) && !errors.Is(err, ctx.Err()) { if err != nil && !isLibvirtError(err) && !isClassified(err) && !errors.Is(err, ctx.Err()) {
log.V(1).Info("libvirt connection dropped", "reason", "transport error", "error", err.Error()) log.V(1).Info("libvirt connection dropped", "reason", "transport error", "error", err.Error())
p.drop(host.Name, c) p.drop(host.Name, c)
} }
return err
} }
// awaitOrCancel runs fn in the background and returns its result, unless ctx // awaitOrCancel runs fn in the background and returns its result, unless ctx
@@ -254,7 +349,7 @@ func (l *lazySSH) load() (*sshAuth, error) {
func (l *lazySSH) dial(ctx context.Context, host provider.KVMHost) (*hostClient, error) { func (l *lazySSH) dial(ctx context.Context, host provider.KVMHost) (*hostClient, error) {
a, err := l.load() a, err := l.load()
if err != nil { if err != nil {
return nil, err return nil, &dialError{stage: stageSSHMaterial, err: err}
} }
return a.dial(ctx, host) return a.dial(ctx, host)
} }
@@ -285,7 +380,7 @@ func (a *sshAuth) dialSSH(ctx context.Context, host provider.KVMHost) (*ssh.Clie
var d net.Dialer var d net.Dialer
nc, err := d.DialContext(ctx, "tcp", host.Address) nc, err := d.DialContext(ctx, "tcp", host.Address)
if err != nil { if err != nil {
return nil, fmt.Errorf("ssh dial: %w", err) return nil, &dialError{stage: stageTCP, err: fmt.Errorf("ssh dial: %w", err)}
} }
_ = nc.SetDeadline(deadline) _ = nc.SetDeadline(deadline)
conn, chans, reqs, err := ssh.NewClientConn(nc, host.Address, &ssh.ClientConfig{ conn, chans, reqs, err := ssh.NewClientConn(nc, host.Address, &ssh.ClientConfig{
@@ -295,7 +390,7 @@ func (a *sshAuth) dialSSH(ctx context.Context, host provider.KVMHost) (*ssh.Clie
}) })
if err != nil { if err != nil {
nc.Close() nc.Close()
return nil, fmt.Errorf("ssh handshake: %w", err) return nil, &dialError{stage: stageSSHHandshake, err: fmt.Errorf("ssh handshake: %w", err)}
} }
_ = nc.SetDeadline(time.Time{}) _ = nc.SetDeadline(time.Time{})
return ssh.NewClient(conn, chans, reqs), nil return ssh.NewClient(conn, chans, reqs), nil
@@ -319,7 +414,7 @@ func (u unixForward) Dial() (net.Conn, error) {
func connectLibvirt(ctx context.Context, client *ssh.Client, socket string) (*hostClient, error) { func connectLibvirt(ctx context.Context, client *ssh.Client, socket string) (*hostClient, error) {
lv := libvirt.NewWithDialer(unixForward{client: client, path: socket}) lv := libvirt.NewWithDialer(unixForward{client: client, path: socket})
if err := awaitOrCancel(ctx, lv.Connect, func() { client.Close() }); err != nil { if err := awaitOrCancel(ctx, lv.Connect, func() { client.Close() }); err != nil {
return nil, fmt.Errorf("libvirt connect via %s: %w", socket, err) return nil, &dialError{stage: stageLibvirtConnect, err: fmt.Errorf("libvirt connect via %s: %w", socket, err)}
} }
stop := make(chan struct{}) stop := make(chan struct{})
var once sync.Once var once sync.Once
+171
View File
@@ -232,6 +232,97 @@ func TestHostPool_call_hostsAreIndependent(t *testing.T) {
} }
} }
// A dial to a dead host holds the host's slot for as long as it takes; a
// caller with a deadline must not wait past it.
func TestHostPool_get_waitingRespectsContext(t *testing.T) {
t.Parallel()
release := make(chan struct{})
dialing := make(chan struct{})
var once sync.Once
dial := func(context.Context, provider.KVMHost) (*hostClient, error) {
once.Do(func() { close(dialing) })
<-release
return nil, errors.New("ssh dial: connection timed out")
}
p := newHostPool(dial)
slow := make(chan error, 1)
go func() { slow <- p.call(context.Background(), discard, testHost, noop) }()
<-dialing
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
start := time.Now()
err := p.call(ctx, discard, testHost, noop)
if !errors.Is(err, context.DeadlineExceeded) {
t.Errorf("call() error = %v, want the caller's deadline", err)
}
if waited := time.Since(start); waited > time.Second {
t.Errorf("call() waited %s behind another caller's dial, want about 50ms", waited)
}
close(release)
if err := <-slow; err == nil {
t.Error("the blocked dial succeeded, want its error")
}
}
func TestHostPool_probe_abandonsWithoutClosing(t *testing.T) {
t.Parallel()
d := &fakeDialer{}
p := newHostPool(d.dial)
unblock := make(chan struct{})
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
err := p.probe(ctx, discard, testHost, func(libvirtAPI) error {
<-unblock // a daemon that stopped answering
return nil
})
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("probe() error = %v, want DeadlineExceeded", err)
}
if d.clients[0].closed.Load() {
t.Error("probe closed the shared connection when it gave up")
}
if err := p.probe(context.Background(), discard, testHost, noop); !errors.Is(err, errProbeBusy) {
t.Errorf("probe() while the abandoned one still waits = %v, want errProbeBusy", err)
}
if err := p.call(context.Background(), discard, testHost, noop); err != nil {
t.Errorf("call() on the kept connection = %v", err)
}
close(unblock)
deadline := time.Now().Add(time.Second)
for {
err := p.probe(context.Background(), discard, testHost, noop)
if err == nil {
break
}
if !errors.Is(err, errProbeBusy) || time.Now().After(deadline) {
t.Fatalf("probe() after the abandoned one returned = %v, want nil", err)
}
time.Sleep(time.Millisecond)
}
if d.dialCount() != 1 {
t.Errorf("dials = %d, want 1: nothing should have redialed", d.dialCount())
}
}
func TestHostPool_probe_dropsConnectionOnTransportError(t *testing.T) {
t.Parallel()
d := &fakeDialer{}
p := newHostPool(d.dial)
err := p.probe(context.Background(), discard, testHost, func(libvirtAPI) error { return io.ErrUnexpectedEOF })
if !errors.Is(err, io.ErrUnexpectedEOF) {
t.Fatalf("probe() error = %v, want io.ErrUnexpectedEOF passed through", err)
}
if !d.clients[0].closed.Load() {
t.Error("a transport error during a probe left the connection cached")
}
}
// --- SSH dial against an in-process server --- // --- SSH dial against an in-process server ---
func newKeyPair(t *testing.T) (ssh.Signer, []byte) { func newKeyPair(t *testing.T) (ssh.Signer, []byte) {
@@ -614,6 +705,86 @@ func TestNew_missingSSHMaterialStillBuilds(t *testing.T) {
} }
} }
// The health check reports where a connection broke, so every stage of a
// real dial must come back tagged.
func TestLazySSH_dial_tagsTheFailingStage(t *testing.T) {
t.Parallel()
tests := []struct {
name string
mutate func(t *testing.T, f *sshFixture)
want dialStage
wantReason string
}{
{
name: "key material missing",
mutate: func(t *testing.T, f *sshFixture) { f.cfg.SSHKeyFile = filepath.Join(t.TempDir(), "absent") },
want: stageSSHMaterial,
wantReason: reasonSSHMaterialUnavailable,
},
{
name: "nothing listening",
mutate: func(t *testing.T, f *sshFixture) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
f.host.Address = ln.Addr().String()
ln.Close()
},
want: stageTCP,
wantReason: provider.ReasonUnreachable,
},
{
name: "host key not in known_hosts",
mutate: func(t *testing.T, f *sshFixture) {
impostor, _ := newKeyPair(t)
line := knownhosts.Line([]string{f.server.addr}, impostor.PublicKey())
if err := os.WriteFile(f.cfg.KnownHostsFile, []byte(line+"\n"), 0o600); err != nil {
t.Fatal(err)
}
},
want: stageSSHHandshake,
wantReason: reasonSSHHandshakeFailed,
},
{
name: "client key rejected",
mutate: func(t *testing.T, f *sshFixture) {
_, otherPEM := newKeyPair(t)
if err := os.WriteFile(f.keyFile, otherPEM, 0o600); err != nil {
t.Fatal(err)
}
},
want: stageSSHHandshake,
wantReason: reasonSSHHandshakeFailed,
},
{
name: "libvirt socket missing",
mutate: func(_ *testing.T, f *sshFixture) { f.host.Socket = "/nonexistent/libvirt-sock" },
want: stageLibvirtConnect,
wantReason: reasonLibvirtConnectFailed,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
f := newSSHFixture(t, true)
tc.mutate(t, f)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err := newLazySSH(f.cfg).dial(ctx, f.host)
var de *dialError
if !errors.As(err, &de) || de.stage != tc.want {
t.Fatalf("dial() error = %v, want a %s dialError", err, tc.want)
}
if got := checkReason(err); got != tc.wantReason {
t.Errorf("checkReason() = %s, want %s", got, tc.wantReason)
}
})
}
}
func TestKeepalive_closesClientWhenServerStalls(t *testing.T) { func TestKeepalive_closesClientWhenServerStalls(t *testing.T) {
t.Parallel() t.Parallel()
f := newSSHFixture(t, false) f := newSSHFixture(t, false)
+54
View File
@@ -28,6 +28,12 @@ type fakeAPI struct {
// uploadHook, when set, sees every StorageVolUpload before the fake // uploadHook, when set, sees every StorageVolUpload before the fake
// reads the stream; it can return an error to simulate a broken upload. // reads the stream; it can return an error to simulate a broken upload.
uploadHook func(vol string, r io.Reader) error uploadHook func(vol string, r io.Reader) error
// hostname, libVersion, poolCapacity and poolStates answer the health
// check's RPCs. A pool missing from poolStates is running.
hostname string
libVersion uint64
poolCapacity uint64
poolStates map[string]libvirt.StoragePoolState
} }
type fakeVol struct { type fakeVol struct {
@@ -53,6 +59,11 @@ func newFakeAPI(pools ...string) *fakeAPI {
domains: map[string]*fakeDomain{}, domains: map[string]*fakeDomain{},
fail: map[string]error{}, fail: map[string]error{},
failOnce: map[string]error{}, failOnce: map[string]error{},
hostname: "fake-host",
libVersion: 10002000,
poolCapacity: 100 << 30,
poolStates: map[string]libvirt.StoragePoolState{},
} }
for _, p := range pools { for _, p := range pools {
f.pools[p] = map[string]*fakeVol{} f.pools[p] = map[string]*fakeVol{}
@@ -409,3 +420,46 @@ func (f *fakeAPI) StorageVolDelete(vol libvirt.StorageVol, _ libvirt.StorageVolD
delete(f.pools[vol.Pool], vol.Name) delete(f.pools[vol.Pool], vol.Name)
return nil return nil
} }
// --- host ---
func (f *fakeAPI) ConnectGetLibVersion() (uint64, error) {
f.mu.Lock()
defer f.mu.Unlock()
if err := f.enter("ConnectGetLibVersion"); err != nil {
return 0, err
}
return f.libVersion, nil
}
func (f *fakeAPI) ConnectGetHostname() (string, error) {
f.mu.Lock()
defer f.mu.Unlock()
if err := f.enter("ConnectGetHostname"); err != nil {
return "", err
}
return f.hostname, nil
}
// StoragePoolGetInfo reports the pool's state and poolCapacity minus the
// volumes' capacities as available space.
func (f *fakeAPI) StoragePoolGetInfo(pool libvirt.StoragePool) (uint8, uint64, uint64, uint64, error) {
f.mu.Lock()
defer f.mu.Unlock()
if err := f.enter("StoragePoolGetInfo"); err != nil {
return 0, 0, 0, 0, err
}
vols, ok := f.pools[pool.Name]
if !ok {
return 0, 0, 0, 0, lvError(libvirt.ErrNoStoragePool, "no pool %s", pool.Name)
}
var used uint64
for _, v := range vols {
used += uint64(v.capacity)
}
state := libvirt.StoragePoolRunning
if s, ok := f.poolStates[pool.Name]; ok {
state = s
}
return uint8(state), f.poolCapacity, used, f.poolCapacity - used, nil
}
@@ -347,6 +347,36 @@ func TestKVM_lifecycle(t *testing.T) {
env.keep = false env.keep = false
} }
// TestKVMCheck_realHost runs the provider health check against the host.
// It creates nothing: it must reach the host and find the pool running.
func TestKVMCheck_realHost(t *testing.T) {
env := loadIntegrationEnv(t)
log := integrationLogger(t, env.logLevel)
ctx, cancel := context.WithTimeout(logr.NewContext(context.Background(), log), time.Minute)
defer cancel()
prov, err := New(ctx, env.cfg)
if err != nil {
t.Fatalf("New: %v", err)
}
got := prov.(*Provider).Check(ctx)
if len(got) != 1 {
t.Fatalf("Check returned %d targets, want 1", len(got))
}
ts := got[0]
t.Logf("%s (%s): ready=%v reason=%s latency=%s details=%v",
ts.Name, ts.Address, ts.Ready, ts.Reason, ts.Latency.Round(time.Millisecond), ts.Details)
if !ts.Ready {
t.Fatalf("check failed: %s: %s", ts.Reason, ts.Message)
}
for _, key := range []string{"hostname", "libvirtVersion", "poolState", "poolAvailable"} {
if ts.Details[key] == "" {
t.Errorf("details[%s] is empty, want what the host reported", key)
}
}
}
// waitForRunning polls Get until the instance is Running with an IP or ctx // waitForRunning polls Get until the instance is Running with an IP or ctx
// ends. Transient errors are retried; anything else fails the test. // ends. Transient errors are retried; anything else fails the test.
func waitForRunning(ctx context.Context, t *testing.T, p *Provider, id string) *provider.Instance { func waitForRunning(ctx context.Context, t *testing.T, p *Provider, id string) *provider.Instance {
+21
View File
@@ -209,3 +209,24 @@ func (t *tracingAPI) StorageVolDelete(vol libvirt.StorageVol, flags libvirt.Stor
c.done(err) c.done(err)
return err return err
} }
func (t *tracingAPI) ConnectGetLibVersion() (uint64, error) {
c := t.begin("ConnectGetLibVersion")
v, err := t.inner.ConnectGetLibVersion()
c.done(err, "version", formatLibVersion(v))
return v, err
}
func (t *tracingAPI) ConnectGetHostname() (string, error) {
c := t.begin("ConnectGetHostname")
h, err := t.inner.ConnectGetHostname()
c.done(err, "hostname", h)
return h, err
}
func (t *tracingAPI) StoragePoolGetInfo(pool libvirt.StoragePool) (uint8, uint64, uint64, uint64, error) {
c := t.begin("StoragePoolGetInfo", "pool", pool.Name)
state, capacity, allocation, available, err := t.inner.StoragePoolGetInfo(pool)
c.done(err, "state", poolStateName(libvirt.StoragePoolState(state)), "capacity", capacity, "available", available)
return state, capacity, allocation, available, err
}
+3
View File
@@ -103,6 +103,9 @@ func TestTracingAPI_logsEveryMethod(t *testing.T) {
return api.StorageVolDownload(overlay, io.Discard, 0, 0, 0) return api.StorageVolDownload(overlay, io.Discard, 0, 0, 0)
}}, }},
{"StorageVolDelete", func(api libvirtAPI) error { return api.StorageVolDelete(overlay, 0) }}, {"StorageVolDelete", func(api libvirtAPI) error { return api.StorageVolDelete(overlay, 0) }},
{"ConnectGetLibVersion", func(api libvirtAPI) error { _, err := api.ConnectGetLibVersion(); return err }},
{"ConnectGetHostname", func(api libvirtAPI) error { _, err := api.ConnectGetHostname(); return err }},
{"StoragePoolGetInfo", func(api libvirtAPI) error { _, _, _, _, err := api.StoragePoolGetInfo(pool); return err }},
} }
if n := reflect.TypeOf((*libvirtAPI)(nil)).Elem().NumMethod(); n != len(cases) { if n := reflect.TypeOf((*libvirtAPI)(nil)).Elem().NumMethod(); n != len(cases) {
t.Fatalf("libvirtAPI has %d methods, this test covers %d; trace the new one", n, len(cases)) t.Fatalf("libvirtAPI has %d methods, this test covers %d; trace the new one", n, len(cases))
+100
View File
@@ -0,0 +1,100 @@
package providerhealth
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
"time"
"unicode/utf8"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/envtest"
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
// TestMonitor_envtestRoundTrip writes through a real API server, which is
// the only thing that validates the generated schema: cluster scope, the
// status subresource, required fields, list-map keys and the message limit.
func TestMonitor_envtestRoundTrip(t *testing.T) {
if testing.Short() {
t.Skip("envtest skipped under -short")
}
env := &envtest.Environment{
CRDDirectoryPaths: []string{filepath.Join("..", "..", "config", "crd", "bases")},
ErrorIfCRDPathMissing: true,
BinaryAssetsDirectory: envtestBinaryDir(),
}
cfg, err := env.Start()
if err != nil {
t.Fatalf("starting envtest (run `make setup-envtest` first): %v", err)
}
t.Cleanup(func() { _ = env.Stop() })
scheme := runtime.NewScheme()
if err := crawlv1alpha1.AddToScheme(scheme); err != nil {
t.Fatal(err)
}
c, err := client.New(cfg, client.Options{Scheme: scheme})
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
m := &Monitor{Client: c, Reader: c}
first := Result{Provider: "kvm-home", Type: "kvm", Supported: true, CheckedAt: t0, Targets: []provider.TargetStatus{
{Name: "nas1", Address: "nas1:22", Ready: true, Reason: provider.ReasonReachable,
Latency: 42 * time.Millisecond, Details: map[string]string{"libvirtVersion": "10.2.0"}},
{Name: "nas2", Address: "nas2:22", Reason: provider.ReasonUnreachable, Message: strings.Repeat("x", 2000)},
}}
if err := m.publish(ctx, first); err != nil {
t.Fatalf("first publish: %v", err)
}
got := getHealth(t, c, "kvm-home")
if got.Status.TargetsReady != "1/2" || got.Status.Targets[0].Details["libvirtVersion"] != "10.2.0" ||
got.Status.Targets[0].LatencyMillis != 42 || utf8.RuneCountInString(got.Status.Targets[1].Message) != maxMessageLen {
t.Errorf("status after the first round = %+v", got.Status)
}
second := Result{Provider: "kvm-home", Type: "kvm", Supported: true, CheckedAt: t1,
Targets: []provider.TargetStatus{up("nas1"), up("nas2")}}
if err := m.publish(ctx, second); err != nil {
t.Fatalf("second publish: %v", err)
}
got = getHealth(t, c, "kvm-home")
if cond := readyCondition(got); cond == nil || cond.Reason != reasonAllTargetsReady {
t.Errorf("Ready after recovery = %+v, want %s", cond, reasonAllTargetsReady)
}
if lt := got.Status.Targets[0].LastTransitionTime.Time; !lt.Equal(t0) {
t.Errorf("nas1 lastTransitionTime = %s, want %s", lt, t0)
}
if lt := got.Status.Targets[1].LastTransitionTime.Time; !lt.Equal(t1) {
t.Errorf("nas2 lastTransitionTime = %s, want %s", lt, t1)
}
m.cleanup(ctx, nil)
var gone crawlv1alpha1.ProviderHealth
if err := c.Get(ctx, client.ObjectKey{Name: "kvm-home"}, &gone); !apierrors.IsNotFound(err) {
t.Errorf("Get after cleanup = %v, want NotFound", err)
}
}
// envtestBinaryDir finds the binaries `make setup-envtest` downloads, so
// the test also runs outside make (which sets KUBEBUILDER_ASSETS instead).
func envtestBinaryDir() string {
base := filepath.Join("..", "..", "bin", "k8s")
entries, err := os.ReadDir(base)
if err != nil {
return ""
}
for _, e := range entries {
if e.IsDir() {
return filepath.Join(base, e.Name())
}
}
return ""
}
+296
View File
@@ -0,0 +1,296 @@
// Package providerhealth checks every configured provider on an interval
// and publishes the results: one ProviderHealth object per provider, a log
// line whenever a target goes down or comes back, and a snapshot the
// metrics collector reads at scrape time.
package providerhealth
import (
"context"
"fmt"
"slices"
"sync"
"time"
"github.com/go-logr/logr"
"go.opentelemetry.io/otel/attribute"
"k8s.io/apimachinery/pkg/api/equality"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/validation"
"sigs.k8s.io/controller-runtime/pkg/client"
logf "sigs.k8s.io/controller-runtime/pkg/log"
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
)
const (
// labelManagedBy marks the ProviderHealth objects this operator owns;
// cleanup never deletes an object without it.
labelManagedBy = "app.kubernetes.io/managed-by"
managedByValue = "egress-proxies-operator"
minInterval = 10 * time.Second
defaultTimeout = 15 * time.Second
)
// Target is one configured provider. Checker is nil for a provider that
// cannot check itself.
type Target struct {
Name string
Type string
Checker provider.Checker
}
// Result is the outcome of one provider's latest check.
type Result struct {
Provider string
Type string
// Supported is false when the provider has no Checker.
Supported bool
Targets []provider.TargetStatus
CheckedAt time.Time
}
// controller-gen reads RBAC markers only at package level, so they stay
// out of Monitor's doc comment.
// +kubebuilder:rbac:groups=crawl.example.com,resources=providerhealths,verbs=get;list;create;delete
// +kubebuilder:rbac:groups=crawl.example.com,resources=providerhealths/status,verbs=get;patch
// Monitor is the manager Runnable that checks the providers.
type Monitor struct {
// Client writes the ProviderHealth objects.
Client client.Client
// Reader reads them; pass the manager's API reader. The objects are
// cluster-scoped, and reading them through the cache would start a
// cluster-wide informer needing watch rights, whatever --proxy-namespace
// says, for a handful of reads a minute.
Reader client.Reader
// Targets are the configured providers, in config order.
Targets []Target
// Interval between rounds of checks; 0 turns checking off. The first
// round runs at start: nothing races it, and early visibility is the
// point.
Interval time.Duration
// Timeout bounds each round; 0 means 15s, or the interval when that is
// shorter.
Timeout time.Duration
now func() time.Time
mu sync.Mutex
latest []Result
// prev holds each target's previous verdict for transition logging.
// Only the loop goroutine touches it.
prev map[string]provider.TargetStatus
}
// NeedLeaderElection is true: only the leader dials the hypervisors and
// writes the objects, and only its gauges describe the checks.
func (m *Monitor) NeedLeaderElection() bool { return true }
// ValidateSchedule checks --provider-check-interval and
// --provider-check-timeout. The interval is 0 (off) or at least 10s, which
// is what keeps one status write per provider per round cheap. The timeout
// is 0 (derived from the interval, see Monitor.Timeout) or positive and no
// longer than the interval.
func ValidateSchedule(interval, timeout time.Duration) error {
switch {
case interval == 0:
return nil
case interval < minInterval:
return fmt.Errorf("--provider-check-interval must be 0 (off) or at least %s, got %s", minInterval, interval)
case timeout < 0 || timeout > interval:
return fmt.Errorf("--provider-check-timeout must be at most the interval (%s), got %s", interval, timeout)
}
return nil
}
func (m *Monitor) setDefaults() {
if m.Timeout == 0 {
m.Timeout = min(defaultTimeout, m.Interval)
}
if m.now == nil {
m.now = time.Now
}
}
// Start deletes the objects of providers that are no longer configured,
// then checks every Interval until ctx ends. The providers config is read
// once per process, so cleaning up once is enough. With checking off,
// every object goes, so none is left claiming a stale Ready.
func (m *Monitor) Start(ctx context.Context) error {
m.setDefaults()
log := logf.FromContext(ctx).WithName("provider-health")
ctx = logf.IntoContext(ctx, log)
keep := map[string]bool{}
if m.Interval > 0 {
for _, t := range m.Targets {
keep[t.Name] = true
}
}
m.cleanup(ctx, keep)
if m.Interval == 0 {
log.Info("Provider checks are off (--provider-check-interval=0)")
return nil
}
for _, t := range m.Targets {
if errs := validation.IsDNS1123Subdomain(t.Name); len(errs) > 0 {
log.Info("WARNING: provider name is not a valid object name; its ProviderHealth object is not written, its metrics are",
"provider", t.Name, "problems", errs)
}
}
m.checkOnce(ctx)
ticker := time.NewTicker(m.Interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return nil
case <-ticker.C:
m.checkOnce(ctx)
}
}
}
// Snapshot returns the latest results in config order.
func (m *Monitor) Snapshot() []Result {
m.mu.Lock()
defer m.mu.Unlock()
out := make([]Result, len(m.latest))
for i, r := range m.latest {
r.Targets = slices.Clone(r.Targets)
out[i] = r
}
return out
}
// checkOnce runs every provider's check concurrently under one timeout,
// then records, logs and publishes the results provider by provider. A
// failed write is logged and retried by the next round.
func (m *Monitor) checkOnce(ctx context.Context) {
ctx, span := tracing.Start(ctx, "providerhealth.check")
defer span.End()
log := logf.FromContext(ctx)
checkCtx, cancel := context.WithTimeout(ctx, m.Timeout)
defer cancel()
results := make([]Result, len(m.Targets))
var wg sync.WaitGroup
for i, t := range m.Targets {
wg.Go(func() { results[i] = m.run(checkCtx, t) })
}
wg.Wait()
m.mu.Lock()
m.latest = results
m.mu.Unlock()
notReady := 0
for _, r := range results {
m.logTransitions(log, r)
for _, ts := range r.Targets {
if !ts.Ready {
notReady++
}
}
if err := m.publish(ctx, r); err != nil {
log.Error(err, "Writing provider health", "provider", r.Provider)
}
}
span.SetAttributes(
attribute.Int("providerhealth.providers", len(results)),
attribute.Int("providerhealth.targets_not_ready", notReady),
)
}
func (m *Monitor) run(ctx context.Context, t Target) Result {
r := Result{Provider: t.Name, Type: t.Type, Supported: t.Checker != nil}
if t.Checker != nil {
r.Targets = t.Checker.Check(ctx)
}
r.CheckedAt = m.now()
return r
}
// logTransitions logs a target going down, staying down for a different
// reason, or coming back. Steady states stay quiet; a target that is ready
// the first time it is seen only logs at V(1).
func (m *Monitor) logTransitions(log logr.Logger, r Result) {
if m.prev == nil {
m.prev = map[string]provider.TargetStatus{}
}
for _, ts := range r.Targets {
key := r.Provider + "/" + ts.Name
prev, seen := m.prev[key]
m.prev[key] = ts
kv := []any{"provider", r.Provider, "target", ts.Name, "address", ts.Address, "reason", ts.Reason}
switch {
case !ts.Ready && (!seen || prev.Ready || prev.Reason != ts.Reason):
log.Info("WARNING: provider target not ready", append(kv, "message", ts.Message)...)
case ts.Ready && seen && !prev.Ready:
log.Info("Provider target ready", append(kv, "latency", ts.Latency.Round(time.Millisecond))...)
case ts.Ready && !seen:
log.V(1).Info("Provider target ready", append(kv, "latency", ts.Latency.Round(time.Millisecond))...)
}
}
}
// publish creates the provider's ProviderHealth object when missing and
// patches its status. The status changes on every round, since
// lastCheckTime moves; the comparison only skips the patch when a clock
// did not advance.
func (m *Monitor) publish(ctx context.Context, r Result) error {
if len(validation.IsDNS1123Subdomain(r.Provider)) > 0 {
return nil
}
var ph crawlv1alpha1.ProviderHealth
err := m.Reader.Get(ctx, client.ObjectKey{Name: r.Provider}, &ph)
switch {
case apierrors.IsNotFound(err):
ph = crawlv1alpha1.ProviderHealth{ObjectMeta: metav1.ObjectMeta{
Name: r.Provider,
Labels: map[string]string{labelManagedBy: managedByValue},
}}
if err := m.Client.Create(ctx, &ph); err != nil {
return fmt.Errorf("creating ProviderHealth %s: %w", r.Provider, err)
}
case err != nil:
return fmt.Errorf("reading ProviderHealth %s: %w", r.Provider, err)
}
base := ph.DeepCopy()
ph.Status = buildStatus(ph.Status, r)
if equality.Semantic.DeepEqual(base.Status, ph.Status) {
return nil
}
if err := m.Client.Status().Patch(ctx, &ph, client.MergeFrom(base)); err != nil {
return fmt.Errorf("patching ProviderHealth %s status: %w", r.Provider, err)
}
return nil
}
// cleanup deletes this operator's ProviderHealth objects whose name is not
// in keep.
func (m *Monitor) cleanup(ctx context.Context, keep map[string]bool) {
log := logf.FromContext(ctx)
var list crawlv1alpha1.ProviderHealthList
if err := m.Reader.List(ctx, &list, client.MatchingLabels{labelManagedBy: managedByValue}); err != nil {
log.Error(err, "Listing provider health objects; stale ones stay until the next start")
return
}
for i := range list.Items {
ph := &list.Items[i]
if keep[ph.Name] {
continue
}
if err := m.Client.Delete(ctx, ph); client.IgnoreNotFound(err) != nil {
log.Error(err, "Deleting provider health of an unconfigured provider", "name", ph.Name)
continue
}
log.Info("Deleted provider health of an unconfigured provider", "name", ph.Name)
}
}
+377
View File
@@ -0,0 +1,377 @@
package providerhealth
import (
"context"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/go-logr/logr"
"github.com/go-logr/logr/funcr"
apierrors "k8s.io/apimachinery/pkg/api/errors"
apimeta "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
var (
t0 = time.Date(2026, 9, 11, 15, 0, 0, 0, time.UTC)
t1 = t0.Add(time.Minute)
)
type checkFunc func(ctx context.Context) []provider.TargetStatus
func (f checkFunc) Check(ctx context.Context) []provider.TargetStatus { return f(ctx) }
func verdicts(ts ...provider.TargetStatus) checkFunc {
return func(context.Context) []provider.TargetStatus { return ts }
}
func up(name string) provider.TargetStatus {
return provider.TargetStatus{Name: name, Address: name + ":22", Ready: true, Reason: provider.ReasonReachable}
}
func down(name, reason string) provider.TargetStatus {
return provider.TargetStatus{Name: name, Address: name + ":22", Reason: reason, Message: name + " is down"}
}
func managed(name string) *crawlv1alpha1.ProviderHealth {
return &crawlv1alpha1.ProviderHealth{ObjectMeta: metav1.ObjectMeta{
Name: name,
Labels: map[string]string{labelManagedBy: managedByValue},
}}
}
// newFakeClient registers ProviderHealth as cluster-scoped; the fake's
// default mapper would treat it as namespaced.
func newFakeClient(t *testing.T, objs ...client.Object) client.WithWatch {
t.Helper()
scheme := runtime.NewScheme()
if err := crawlv1alpha1.AddToScheme(scheme); err != nil {
t.Fatal(err)
}
mapper := apimeta.NewDefaultRESTMapper([]schema.GroupVersion{crawlv1alpha1.GroupVersion})
mapper.Add(crawlv1alpha1.GroupVersion.WithKind("ProviderHealth"), apimeta.RESTScopeRoot)
return fake.NewClientBuilder().
WithScheme(scheme).
WithRESTMapper(mapper).
WithStatusSubresource(&crawlv1alpha1.ProviderHealth{}).
WithObjects(objs...).
Build()
}
func newTestMonitor(c client.Client, targets ...Target) *Monitor {
return &Monitor{
Client: c,
Reader: c,
Targets: targets,
Interval: time.Minute,
Timeout: time.Second,
now: func() time.Time { return t1 },
}
}
func getHealth(t *testing.T, c client.Reader, name string) *crawlv1alpha1.ProviderHealth {
t.Helper()
var ph crawlv1alpha1.ProviderHealth
if err := c.Get(context.Background(), client.ObjectKey{Name: name}, &ph); err != nil {
t.Fatalf("Get ProviderHealth %s: %v", name, err)
}
return &ph
}
func readyCondition(ph *crawlv1alpha1.ProviderHealth) *metav1.Condition {
return apimeta.FindStatusCondition(ph.Status.Conditions, crawlv1alpha1.ConditionReady)
}
func TestMonitor_checkOnce_createsLabelledObjects(t *testing.T) {
t.Parallel()
c := newFakeClient(t)
m := newTestMonitor(c,
Target{Name: "kvm-home", Type: "kvm", Checker: verdicts(up("nas1"), down("nas2", provider.ReasonUnreachable))},
Target{Name: "local", Type: "kubernetes", Checker: verdicts(up("apiserver"))},
)
m.checkOnce(context.Background())
kvm := getHealth(t, c, "kvm-home")
if kvm.Labels[labelManagedBy] != managedByValue {
t.Errorf("labels = %v, want %s=%s", kvm.Labels, labelManagedBy, managedByValue)
}
if cond := readyCondition(kvm); cond == nil || cond.Status != metav1.ConditionFalse || cond.Reason != reasonTargetsNotReady {
t.Errorf("kvm-home Ready = %+v, want False/%s", cond, reasonTargetsNotReady)
}
if kvm.Status.Type != "kvm" || kvm.Status.TargetsReady != "1/2" || kvm.Status.LastCheckTime == nil || !kvm.Status.LastCheckTime.Time.Equal(t1) {
t.Errorf("kvm-home status = %+v, want type kvm, 1/2 ready, checked at %s", kvm.Status, t1)
}
local := getHealth(t, c, "local")
if cond := readyCondition(local); cond == nil || cond.Status != metav1.ConditionTrue {
t.Errorf("local Ready = %+v, want True", cond)
}
}
func TestMonitor_checkOnce_patchesExisting(t *testing.T) {
t.Parallel()
existing := managed("local")
existing.Status = crawlv1alpha1.ProviderHealthStatus{
Type: "kubernetes",
Targets: []crawlv1alpha1.ProviderTarget{readyAt("apiserver", t0)},
}
c := newFakeClient(t, existing)
m := newTestMonitor(c, Target{Name: "local", Type: "kubernetes", Checker: verdicts(up("apiserver"))})
m.checkOnce(context.Background())
got := getHealth(t, c, "local")
if got.Status.LastCheckTime == nil || !got.Status.LastCheckTime.Time.Equal(t1) {
t.Errorf("lastCheckTime = %v, want %s", got.Status.LastCheckTime, t1)
}
if lt := got.Status.Targets[0].LastTransitionTime.Time; !lt.Equal(t0) {
t.Errorf("apiserver lastTransitionTime = %s, want %s kept: it was ready before too", lt, t0)
}
}
func TestMonitor_checkOnce_timeoutBoundsEachRound(t *testing.T) {
t.Parallel()
c := newFakeClient(t)
var hadDeadline atomic.Bool
hanging := checkFunc(func(ctx context.Context) []provider.TargetStatus {
_, ok := ctx.Deadline()
hadDeadline.Store(ok)
<-ctx.Done()
return []provider.TargetStatus{down("nas1", provider.ReasonTimeout)}
})
m := newTestMonitor(c,
Target{Name: "kvm-home", Type: "kvm", Checker: hanging},
Target{Name: "local", Type: "kubernetes", Checker: verdicts(up("apiserver"))},
)
m.Timeout = 50 * time.Millisecond
done := make(chan struct{})
go func() {
m.checkOnce(context.Background())
close(done)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("checkOnce still running 5s after a 50ms timeout")
}
if !hadDeadline.Load() {
t.Error("the checker got a context without a deadline")
}
if cond := readyCondition(getHealth(t, c, "kvm-home")); cond == nil || cond.Status != metav1.ConditionFalse {
t.Errorf("kvm-home Ready = %+v, want False", cond)
}
if cond := readyCondition(getHealth(t, c, "local")); cond == nil || cond.Status != metav1.ConditionTrue {
t.Errorf("local Ready = %+v, want True despite the slow neighbour", cond)
}
}
func TestMonitor_checkOnce_noCheckerIsUnknown(t *testing.T) {
t.Parallel()
c := newFakeClient(t)
m := newTestMonitor(c, Target{Name: "gcp-eu", Type: "gcp"})
m.checkOnce(context.Background())
got := getHealth(t, c, "gcp-eu")
if cond := readyCondition(got); cond == nil || cond.Status != metav1.ConditionUnknown || cond.Reason != reasonChecksNotImplemented {
t.Errorf("Ready = %+v, want Unknown/%s", cond, reasonChecksNotImplemented)
}
}
func TestMonitor_checkOnce_invalidNameSkipsTheObject(t *testing.T) {
t.Parallel()
c := newFakeClient(t)
m := newTestMonitor(c, Target{Name: "Home_KVM", Type: "kvm", Checker: verdicts(up("nas1"))})
m.checkOnce(context.Background())
var list crawlv1alpha1.ProviderHealthList
if err := c.List(context.Background(), &list); err != nil {
t.Fatal(err)
}
if len(list.Items) != 0 {
t.Errorf("wrote %d objects, want none for an invalid name", len(list.Items))
}
snap := m.Snapshot()
if len(snap) != 1 || snap[0].Provider != "Home_KVM" || !snap[0].Targets[0].Ready {
t.Errorf("Snapshot() = %+v, want the result kept for metrics", snap)
}
}
func TestMonitor_logsOnlyTransitions(t *testing.T) {
t.Parallel()
var mu sync.Mutex
var lines []string
log := funcr.New(func(_, args string) {
mu.Lock()
defer mu.Unlock()
lines = append(lines, args)
}, funcr.Options{})
ctx := logr.NewContext(context.Background(), log)
rounds := [][]provider.TargetStatus{
{up("nas1")},
{up("nas1")},
{down("nas1", provider.ReasonUnreachable)},
{down("nas1", provider.ReasonUnreachable)},
{down("nas1", provider.ReasonTimeout)},
{up("nas1")},
}
round := 0
m := newTestMonitor(newFakeClient(t), Target{Name: "kvm-home", Type: "kvm", Checker: checkFunc(
func(context.Context) []provider.TargetStatus { return rounds[round] })})
for round = range rounds {
m.checkOnce(ctx)
}
mu.Lock()
defer mu.Unlock()
want := [][]string{
{`"msg"="WARNING: provider target not ready"`, `"reason"="Unreachable"`},
{`"msg"="WARNING: provider target not ready"`, `"reason"="Timeout"`},
{`"msg"="Provider target ready"`},
}
if len(lines) != len(want) {
t.Fatalf("logged %d lines, want %d:\n%s", len(lines), len(want), strings.Join(lines, "\n"))
}
for i, subs := range want {
for _, sub := range subs {
if !strings.Contains(lines[i], sub) {
t.Errorf("line %d missing %s:\n%s", i, sub, lines[i])
}
}
}
}
func TestMonitor_cleanup_deletesOnlyLabelledUnconfigured(t *testing.T) {
t.Parallel()
foreign := &crawlv1alpha1.ProviderHealth{ObjectMeta: metav1.ObjectMeta{Name: "foreign"}}
c := newFakeClient(t, managed("kvm-home"), managed("retired"), foreign)
m := newTestMonitor(c)
m.cleanup(context.Background(), map[string]bool{"kvm-home": true})
getHealth(t, c, "kvm-home")
getHealth(t, c, "foreign")
var gone crawlv1alpha1.ProviderHealth
if err := c.Get(context.Background(), client.ObjectKey{Name: "retired"}, &gone); !apierrors.IsNotFound(err) {
t.Errorf("Get retired = %v, want NotFound", err)
}
}
func TestMonitor_Start_offCleansUpAndReturns(t *testing.T) {
t.Parallel()
c := newFakeClient(t, managed("kvm-home"))
var checked atomic.Bool
m := newTestMonitor(c, Target{Name: "kvm-home", Type: "kvm", Checker: checkFunc(
func(context.Context) []provider.TargetStatus { checked.Store(true); return nil })})
m.Interval = 0
errc := make(chan error, 1)
go func() { errc <- m.Start(context.Background()) }()
select {
case err := <-errc:
if err != nil {
t.Fatalf("Start() = %v, want nil", err)
}
case <-time.After(5 * time.Second):
t.Fatal("Start did not return with checking off")
}
if checked.Load() {
t.Error("a check ran with checking off")
}
var gone crawlv1alpha1.ProviderHealth
if err := c.Get(context.Background(), client.ObjectKey{Name: "kvm-home"}, &gone); !apierrors.IsNotFound(err) {
t.Errorf("Get kvm-home = %v, want NotFound so no stale Ready is left behind", err)
}
}
func TestMonitor_Start_checksAtOnceAndStops(t *testing.T) {
t.Parallel()
c := newFakeClient(t)
m := newTestMonitor(c, Target{Name: "local", Type: "kubernetes", Checker: verdicts(up("apiserver"))})
m.Interval = time.Hour
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
errc := make(chan error, 1)
go func() { errc <- m.Start(ctx) }()
deadline := time.Now().Add(5 * time.Second)
for len(m.Snapshot()) == 0 {
if time.Now().After(deadline) {
t.Fatal("no check within 5s of Start, want one right away")
}
time.Sleep(10 * time.Millisecond)
}
cancel()
select {
case err := <-errc:
if err != nil {
t.Errorf("Start() = %v, want nil", err)
}
case <-time.After(5 * time.Second):
t.Fatal("Start did not return after its context ended")
}
}
func TestValidateSchedule(t *testing.T) {
t.Parallel()
tests := []struct {
name string
interval, timeout time.Duration
wantErr bool
}{
{"off", 0, 0, false},
{"defaults", time.Minute, 15 * time.Second, false},
{"shortest interval", 10 * time.Second, 10 * time.Second, false},
{"too frequent", 5 * time.Second, time.Second, true},
{"negative interval", -time.Minute, time.Second, true},
{"timeout left to derive, short interval", 12 * time.Second, 0, false},
{"negative timeout", time.Minute, -time.Second, true},
{"timeout longer than interval", time.Minute, 2 * time.Minute, true},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
if err := ValidateSchedule(tc.interval, tc.timeout); (err != nil) != tc.wantErr {
t.Errorf("ValidateSchedule(%s, %s) = %v, want error %v", tc.interval, tc.timeout, err, tc.wantErr)
}
})
}
}
func TestMonitor_setDefaults_timeout(t *testing.T) {
t.Parallel()
tests := []struct {
name string
interval, timeout, want time.Duration
}{
{"default", time.Minute, 0, 15 * time.Second},
{"interval shorter than the default", 12 * time.Second, 0, 12 * time.Second},
{"explicit", time.Minute, 20 * time.Second, 20 * time.Second},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
m := &Monitor{Interval: tc.interval, Timeout: tc.timeout}
m.setDefaults()
if m.Timeout != tc.want {
t.Errorf("Timeout = %s, want %s", m.Timeout, tc.want)
}
})
}
}
+98
View File
@@ -0,0 +1,98 @@
package providerhealth
import (
"fmt"
"slices"
"strings"
"unicode/utf8"
apimeta "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
)
// Reasons on the Ready condition.
const (
reasonAllTargetsReady = "AllTargetsReady"
reasonTargetsNotReady = "TargetsNotReady"
reasonChecksNotImplemented = "ChecksNotImplemented"
reasonNoTargets = "NoTargets"
)
// maxMessageLen matches the CRD's limit on a target's message.
const maxMessageLen = 1024
// buildStatus derives a ProviderHealth status from the previous one and a
// fresh result. Every timestamp comes from r.CheckedAt, and prev is left
// untouched. A target keeps its lastTransitionTime while ready does not
// flip, which is how several hosts each show when they went down.
func buildStatus(prev crawlv1alpha1.ProviderHealthStatus, r Result) crawlv1alpha1.ProviderHealthStatus {
now := metav1.NewTime(r.CheckedAt)
out := crawlv1alpha1.ProviderHealthStatus{
Type: r.Type,
Conditions: slices.Clone(prev.Conditions),
LastCheckTime: &now,
}
before := make(map[string]crawlv1alpha1.ProviderTarget, len(prev.Targets))
for _, t := range prev.Targets {
before[t.Name] = t
}
ready := 0
var failing []string
for _, ts := range r.Targets {
t := crawlv1alpha1.ProviderTarget{
Name: ts.Name,
Address: ts.Address,
Ready: ts.Ready,
Reason: ts.Reason,
Message: truncate(ts.Message, maxMessageLen),
LatencyMillis: ts.Latency.Milliseconds(),
LastTransitionTime: now,
Details: ts.Details,
}
if b, ok := before[ts.Name]; ok && b.Ready == ts.Ready {
t.LastTransitionTime = b.LastTransitionTime
}
if ts.Ready {
ready++
} else {
failing = append(failing, ts.Name+": "+ts.Reason)
}
out.Targets = append(out.Targets, t)
}
if r.Supported {
out.TargetsReady = fmt.Sprintf("%d/%d", ready, len(r.Targets))
}
cond := metav1.Condition{Type: crawlv1alpha1.ConditionReady, LastTransitionTime: now}
switch {
case !r.Supported:
cond.Status = metav1.ConditionUnknown
cond.Reason = reasonChecksNotImplemented
cond.Message = fmt.Sprintf("%s providers have no health check", r.Type)
case len(r.Targets) == 0:
cond.Status = metav1.ConditionUnknown
cond.Reason = reasonNoTargets
cond.Message = "the provider has nothing to check"
case len(failing) == 0:
cond.Status = metav1.ConditionTrue
cond.Reason = reasonAllTargetsReady
cond.Message = fmt.Sprintf("%d of %d targets ready", ready, len(r.Targets))
default:
cond.Status = metav1.ConditionFalse
cond.Reason = reasonTargetsNotReady
cond.Message = truncate(strings.Join(failing, "; "), maxMessageLen)
}
apimeta.SetStatusCondition(&out.Conditions, cond)
return out
}
// truncate cuts s to at most n characters, which is how the API server
// counts maxLength.
func truncate(s string, n int) string {
if utf8.RuneCountInString(s) <= n {
return s
}
return string([]rune(s)[:n-1]) + "…"
}
+135
View File
@@ -0,0 +1,135 @@
package providerhealth
import (
"strings"
"testing"
"time"
"unicode/utf8"
apimeta "k8s.io/apimachinery/pkg/api/meta"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
func readyAt(name string, at time.Time) crawlv1alpha1.ProviderTarget {
return crawlv1alpha1.ProviderTarget{Name: name, Ready: true, Reason: provider.ReasonReachable, LastTransitionTime: metav1.NewTime(at)}
}
func TestBuildStatus(t *testing.T) {
t.Parallel()
tests := []struct {
name string
prev crawlv1alpha1.ProviderHealthStatus
result Result
wantStatus metav1.ConditionStatus
wantReason string
wantMessage string
wantTargetsReady string
// wantTransitions is each target's lastTransitionTime, in order.
wantTransitions []time.Time
}{
{
name: "all ready",
result: Result{Type: "kvm", Supported: true, Targets: []provider.TargetStatus{up("nas1"), up("nas2")}, CheckedAt: t1},
wantStatus: metav1.ConditionTrue,
wantReason: reasonAllTargetsReady,
wantMessage: "2 of 2 targets ready",
wantTargetsReady: "2/2",
wantTransitions: []time.Time{t1, t1},
},
{
name: "one target down",
result: Result{Type: "kvm", Supported: true, Targets: []provider.TargetStatus{up("nas1"), down("nas2", provider.ReasonUnreachable)}, CheckedAt: t1},
wantStatus: metav1.ConditionFalse,
wantReason: reasonTargetsNotReady,
wantMessage: "nas2: Unreachable",
wantTargetsReady: "1/2",
wantTransitions: []time.Time{t1, t1},
},
{
name: "only a flip moves a target's transition time",
prev: crawlv1alpha1.ProviderHealthStatus{Targets: []crawlv1alpha1.ProviderTarget{readyAt("nas1", t0), readyAt("nas2", t0)}},
result: Result{Type: "kvm", Supported: true, CheckedAt: t1,
Targets: []provider.TargetStatus{up("nas1"), down("nas2", provider.ReasonTimeout)}},
wantStatus: metav1.ConditionFalse,
wantReason: reasonTargetsNotReady,
wantMessage: "nas2: Timeout",
wantTargetsReady: "1/2",
wantTransitions: []time.Time{t0, t1},
},
{
name: "no checker",
result: Result{Type: "gcp", CheckedAt: t1},
wantStatus: metav1.ConditionUnknown,
wantReason: reasonChecksNotImplemented,
},
{
name: "nothing to check",
result: Result{Type: "kvm", Supported: true, CheckedAt: t1},
wantStatus: metav1.ConditionUnknown,
wantReason: reasonNoTargets,
wantTargetsReady: "0/0",
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
got := buildStatus(tc.prev, tc.result)
cond := apimeta.FindStatusCondition(got.Conditions, crawlv1alpha1.ConditionReady)
if cond == nil || cond.Status != tc.wantStatus || cond.Reason != tc.wantReason {
t.Fatalf("Ready = %+v, want %s/%s", cond, tc.wantStatus, tc.wantReason)
}
if tc.wantMessage != "" && cond.Message != tc.wantMessage {
t.Errorf("Ready message = %q, want %q", cond.Message, tc.wantMessage)
}
if got.TargetsReady != tc.wantTargetsReady {
t.Errorf("targetsReady = %q, want %q", got.TargetsReady, tc.wantTargetsReady)
}
if got.Type != tc.result.Type || got.LastCheckTime == nil || !got.LastCheckTime.Time.Equal(t1) {
t.Errorf("type/lastCheckTime = %q/%v, want %q/%s", got.Type, got.LastCheckTime, tc.result.Type, t1)
}
if len(got.Targets) != len(tc.wantTransitions) {
t.Fatalf("got %d targets, want %d", len(got.Targets), len(tc.wantTransitions))
}
for i, want := range tc.wantTransitions {
if lt := got.Targets[i].LastTransitionTime.Time; !lt.Equal(want) {
t.Errorf("target %s lastTransitionTime = %s, want %s", got.Targets[i].Name, lt, want)
}
}
})
}
}
func TestBuildStatus_conditionKeepsTransitionWhileUnchanged(t *testing.T) {
t.Parallel()
first := buildStatus(crawlv1alpha1.ProviderHealthStatus{},
Result{Type: "kvm", Supported: true, Targets: []provider.TargetStatus{up("nas1")}, CheckedAt: t0})
firstCond := apimeta.FindStatusCondition(first.Conditions, crawlv1alpha1.ConditionReady)
second := buildStatus(first, Result{Type: "kvm", Supported: true, Targets: []provider.TargetStatus{up("nas1")}, CheckedAt: t1})
cond := apimeta.FindStatusCondition(second.Conditions, crawlv1alpha1.ConditionReady)
if !cond.LastTransitionTime.Time.Equal(t0) {
t.Errorf("Ready lastTransitionTime = %s, want %s from the first round", cond.LastTransitionTime, t0)
}
if !firstCond.LastTransitionTime.Time.Equal(t0) || len(first.Conditions) != 1 {
t.Errorf("buildStatus modified the previous status: %+v", first.Conditions)
}
}
func TestBuildStatus_truncatesLongMessages(t *testing.T) {
t.Parallel()
long := strings.Repeat("é", 3000)
got := buildStatus(crawlv1alpha1.ProviderHealthStatus{}, Result{Type: "kvm", Supported: true, CheckedAt: t1,
Targets: []provider.TargetStatus{{Name: "nas1", Reason: provider.ReasonUnreachable, Message: long}}})
if n := utf8.RuneCountInString(got.Targets[0].Message); n != maxMessageLen {
t.Errorf("message has %d characters, want %d", n, maxMessageLen)
}
if !strings.HasSuffix(got.Targets[0].Message, "…") {
t.Error("truncated message does not end with an ellipsis")
}
}
+16
View File
@@ -173,6 +173,20 @@ var _ = Describe("Manager", Ordered, func() {
Eventually(verifyControllerUp).Should(Succeed()) Eventually(verifyControllerUp).Should(Succeed())
}) })
// Runs before the metrics spec on purpose: once a ProviderHealth is
// Ready, the first round of checks is done and its gauges exist.
It("should publish provider health", func() {
By("waiting for the kubernetes provider's ProviderHealth to report Ready")
verifyProviderReady := func(g Gomega) {
cmd := exec.Command("kubectl", "get", "providerhealth", "kubernetes",
"-o", `jsonpath={.status.conditions[?(@.type=="Ready")].status}`)
output, err := utils.Run(cmd)
g.Expect(err).NotTo(HaveOccurred())
g.Expect(output).To(Equal("True"), "kubernetes provider not Ready")
}
Eventually(verifyProviderReady).Should(Succeed())
})
It("should ensure the metrics endpoint is serving metrics", func() { It("should ensure the metrics endpoint is serving metrics", func() {
By("creating a ClusterRoleBinding for the service account to allow access to metrics") By("creating a ClusterRoleBinding for the service account to allow access to metrics")
cmd := exec.Command("kubectl", "create", "clusterrolebinding", metricsRoleBindingName, cmd := exec.Command("kubectl", "create", "clusterrolebinding", metricsRoleBindingName,
@@ -264,6 +278,8 @@ var _ = Describe("Manager", Ordered, func() {
g.Expect(err).NotTo(HaveOccurred(), "Failed to retrieve logs from curl pod") g.Expect(err).NotTo(HaveOccurred(), "Failed to retrieve logs from curl pod")
g.Expect(metricsOutput).NotTo(BeEmpty()) g.Expect(metricsOutput).NotTo(BeEmpty())
g.Expect(metricsOutput).To(ContainSubstring("< HTTP/1.1 200 OK")) g.Expect(metricsOutput).To(ContainSubstring("< HTTP/1.1 200 OK"))
g.Expect(metricsOutput).To(ContainSubstring(
`proxy_operator_provider_target_up{provider="kubernetes",target="apiserver"} 1`))
} }
Eventually(verifyMetricsAvailable, 2*time.Minute).Should(Succeed()) Eventually(verifyMetricsAvailable, 2*time.Minute).Should(Succeed())
}) })