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>
This commit is contained in:
@@ -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 0 — Save the plan
|
||||||
- [x] Step 1 — ProviderHealth API and CRD
|
- [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 3 — KVM check
|
||||||
- [ ] Step 4 — Monitor
|
- [ ] Step 4 — Monitor
|
||||||
- [ ] Step 5 — Metrics and wiring
|
- [ ] Step 5 — Metrics and wiring
|
||||||
@@ -58,3 +58,40 @@ Worth noting:
|
|||||||
round-trip in Step 4.
|
round-trip in Step 4.
|
||||||
- `config/rbac/role.yaml` is unchanged so far. The manager's own permissions arrive in
|
- `config/rbac/role.yaml` is unchanged so far. The manager's own permissions arrive in
|
||||||
Step 4 with the Monitor's RBAC markers.
|
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.
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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()))
|
||||||
|
|||||||
@@ -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) {
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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,
|
||||||
|
|||||||
Reference in New Issue
Block a user