diff --git a/docs/plans-executions/2026-09-11-1526-provider-health-checks.md b/docs/plans-executions/2026-09-11-1526-provider-health-checks.md index 67362fa..9ba5a38 100644 --- a/docs/plans-executions/2026-09-11-1526-provider-health-checks.md +++ b/docs/plans-executions/2026-09-11-1526-provider-health-checks.md @@ -3,7 +3,7 @@ Plan: [docs/plans/2026-09-11-1526-provider-health-checks.md](../plans/2026-09-11 - [x] Step 0 — Save the plan - [x] Step 1 — ProviderHealth API and CRD -- [ ] Step 2 — Contract plus the kubernetes and GCP checks +- [x] Step 2 — Contract plus the kubernetes and GCP checks - [ ] Step 3 — KVM check - [ ] Step 4 — Monitor - [ ] Step 5 — Metrics and wiring @@ -58,3 +58,40 @@ Worth noting: 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. diff --git a/internal/provider/check.go b/internal/provider/check.go new file mode 100644 index 0000000..b2fe4d3 --- /dev/null +++ b/internal/provider/check.go @@ -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 +} diff --git a/internal/provider/gcp/check.go b/internal/provider/gcp/check.go new file mode 100644 index 0000000..3a5f976 --- /dev/null +++ b/internal/provider/gcp/check.go @@ -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 + } +} diff --git a/internal/provider/gcp/check_test.go b/internal/provider/gcp/check_test.go new file mode 100644 index 0000000..faae156 --- /dev/null +++ b/internal/provider/gcp/check_test.go @@ -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) + } + }) + } +} diff --git a/internal/provider/gcp/gcp.go b/internal/provider/gcp/gcp.go index a951f0b..57b8acf 100644 --- a/internal/provider/gcp/gcp.go +++ b/internal/provider/gcp/gcp.go @@ -33,6 +33,8 @@ type instancesAPI interface { Get(ctx context.Context, req *computepb.GetInstanceRequest) (*computepb.Instance, error) Delete(ctx context.Context, req *computepb.DeleteInstanceRequest) (opName string, err 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. @@ -221,17 +223,16 @@ func (p *Provider) Delete(ctx context.Context, providerID string) error { // entire GC sweep. func (p *Provider) ListByTag(ctx context.Context) ([]provider.Instance, error) { log := p.logger(ctx) - filter := fmt.Sprintf("labels.%s = %s", provider.LabelManaged, provider.LabelManagedYes) instances, err := p.api.AggregatedList(ctx, &computepb.AggregatedListInstancesRequest{ Project: p.cfg.Project, - Filter: proto.String(filter), + Filter: proto.String(managedFilter), ReturnPartialSuccess: proto.Bool(true), }) if err != nil { logAPIError(log, "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)) for _, inst := range instances { conv := toInstance(inst, lastPathSegment(inst.GetZone())) diff --git a/internal/provider/gcp/gcp_test.go b/internal/provider/gcp/gcp_test.go index 60e526e..a2e3d00 100644 --- a/internal/provider/gcp/gcp_test.go +++ b/internal/provider/gcp/gcp_test.go @@ -31,6 +31,14 @@ type fakeAPI struct { listReq *computepb.AggregatedListInstancesRequest listInsts []*computepb.Instance 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) { diff --git a/internal/provider/kubernetes/check.go b/internal/provider/kubernetes/check.go new file mode 100644 index 0000000..4943b05 --- /dev/null +++ b/internal/provider/kubernetes/check.go @@ -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 + } +} diff --git a/internal/provider/kubernetes/check_test.go b/internal/provider/kubernetes/check_test.go new file mode 100644 index 0000000..b2e80f0 --- /dev/null +++ b/internal/provider/kubernetes/check_test.go @@ -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()) + } + }) + } +} diff --git a/internal/provider/kubernetes/kubernetes.go b/internal/provider/kubernetes/kubernetes.go index a2688b0..72c6fd6 100644 --- a/internal/provider/kubernetes/kubernetes.go +++ b/internal/provider/kubernetes/kubernetes.go @@ -36,6 +36,8 @@ const defaultImage = "ubuntu/squid:6.6-24.04_edge" type Provider struct { client client.Client 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 @@ -64,7 +66,9 @@ func NewWithTransportWrapper(_ context.Context, cfg provider.ProviderConfig, wra if err != nil { 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,