diff --git a/.gitignore b/.gitignore index 4d8c67a..848b48d 100644 --- a/.gitignore +++ b/.gitignore @@ -17,6 +17,7 @@ coverage/ .env.* !.env.example configs/local.yaml +deploy/.control-plane-tls/ *.pem *.key diff --git a/README.md b/README.md index 8656d04..c36d3a2 100644 --- a/README.md +++ b/README.md @@ -128,7 +128,8 @@ flowchart LR Checker 探测和 Observation 状态归并；Provider 连续空结果的代次化自动 Sequential 切换、禁用候选过滤、 末端 `stop` 的 CAS 路由停用和 Snapshot 即时刷新。 -- **部分完成**：Docker Compose/Kubernetes 运行时 mTLS Overlay。 +- **部分完成**：Kubernetes 运行时 mTLS Overlay；Compose 已具备本地的 + Controller/Gateway/Checker mTLS 运行链路。 - **待完成**：故障演练和代表性集群压测；现有 HTTP、CONNECT 长连接和 Extract 场景只提供可复现的负载工具，不构成容量验证结论。 @@ -167,6 +168,19 @@ Redis、PostgreSQL 和 Controller fixture 脚本会使用 Docker 启动隔离依 ./scripts/test-controller.ps1 ``` +本地 Compose 还会启动 Controller、两个固定身份的 Gateway 和一个 Checker。首次启动前 +生成仅用于本机的 7 天 mTLS 证书；输出目录受 `.gitignore` 保护，脚本拒绝写入非空目录： + +```powershell +./scripts/generate-local-controlplane-certs.ps1 +docker compose -f deploy/docker-compose.yml up -d --build +docker compose -f deploy/docker-compose.yml ps +``` + +Gateway 只有取得有效 Snapshot 后才会通过 `/readyz`；Checker 在首次成功领取控制面任务 +批次后才会通过 `/readyz`，空批次也代表连接和身份验证已经成功。后续任务领取失败会立即 +撤销 Checker 的就绪状态。 + PostgreSQL 与 Redis 是 Controller 的启动依赖。当前可运行的 Controller 入口如下， 将 `CONFIG_FILE` 替换为实际配置路径，并确保其中的 PostgreSQL 与 Redis 地址可从 进程所在网络访问： @@ -180,7 +194,7 @@ go run ./cmd/proxy-controller -config CONFIG_FILE 改用宿主机可达的存储地址。 本地配置中的 `.invalid` Provider URL 是故障演示占位，不会提供真实代理。 -Gateway 已提供启动命令；需要先启用 Controller `controlPlane` 并配置匹配的 mTLS +Gateway 也可独立启动；需要启用 Controller `controlPlane` 并配置匹配的 mTLS 证书（回环 fixture 可使用明文），再提供独立的拨号地址和 Worker 身份： ```powershell diff --git a/deploy/README.md b/deploy/README.md index 824f493..ccb14be 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -5,10 +5,18 @@ Grafana 与 Kubernetes。镜像会构建 `proxy-controller`、`proxy-gateway` `proxy-checker`；Controller 已装配 Redis BASIC 检查任务，Checker 可通过认证 控制面执行 HTTP/HTTPS BASIC 探测。 -默认 Compose 和 Kubernetes Base 仍将 `controlPlane.enabled` 保持为 `false`。 -跨节点控制面必须由环境 Overlay 提供 Controller 可拨号地址、独立 Worker/Checker -身份及 mTLS 证书，因此这些默认清单不能视为完整的生产发布配置。`checker.yaml` -保留为 Overlay 模板，尚未加入 Base。 +Compose 使用 `config/local.yaml` 启用 Controller、两个 Gateway 和一个 Checker 的 +mTLS 控制面。`generate-local-controlplane-certs.ps1` 为三个固定工作负载生成不同的 +SPIFFE URI 证书，私钥只写入被忽略的 `deploy/.control-plane-tls/`。运行前先执行： + +```powershell +./scripts/generate-local-controlplane-certs.ps1 +docker compose -f deploy/docker-compose.yml up -d --build +``` + +Kubernetes Base 仍保持 `controlPlane.enabled: false`，并保留 `checker.yaml` 作为 +环境 Overlay 模板。生产 Overlay 必须为每个弹性 Gateway/Checker 工作负载配置唯一 +身份与证书轮换，不能复用 Compose 的固定开发证书。 当前可执行验证： diff --git a/deploy/compose_test.go b/deploy/compose_test.go index d5b1c8d..c35052a 100644 --- a/deploy/compose_test.go +++ b/deploy/compose_test.go @@ -17,12 +17,14 @@ type composeDocument struct { } type composeService struct { - Image string `yaml:"image"` - Command []string `yaml:"command"` - Ports []string `yaml:"ports"` - Volumes []string `yaml:"volumes"` - Tmpfs []string `yaml:"tmpfs"` - DependsOn any `yaml:"depends_on"` + Image string `yaml:"image"` + Command []string `yaml:"command"` + Ports []string `yaml:"ports"` + Expose []string `yaml:"expose"` + Volumes []string `yaml:"volumes"` + Tmpfs []string `yaml:"tmpfs"` + Environment map[string]string `yaml:"environment"` + DependsOn any `yaml:"depends_on"` } func TestPostgresIntegrationFixtureIsIsolatedAndEphemeral(t *testing.T) { @@ -133,6 +135,65 @@ func TestLocalGatewaysDoNotDependOnControlPlaneStorage(t *testing.T) { } } +func TestLocalComposeProvidesAuthenticatedControlPlaneForEveryWorker(t *testing.T) { + document := loadComposeDocument(t) + controller := document.Services["controller"] + if !slices.Contains(controller.Expose, "8443") || !slices.Contains(controller.Volumes, "./.control-plane-tls/controller:/run/proxy-pool-tls/server:ro") { + t.Fatalf("controller does not expose and mount control-plane TLS: expose=%v volumes=%v", controller.Expose, controller.Volumes) + } + for _, expected := range []struct { + name string + identity string + instance string + certificate string + }{ + {name: "gateway-a", identity: "gateway-a", instance: "gateway-a-1", certificate: "gateway-a"}, + {name: "gateway-b", identity: "gateway-b", instance: "gateway-b-1", certificate: "gateway-b"}, + {name: "checker", identity: "checker-a", instance: "checker-a-1", certificate: "checker-a"}, + } { + service, ok := document.Services[expected.name] + if !ok { + t.Fatalf("docker-compose.yml has no %s", expected.name) + } + if service.Environment["PROXY_POOL_CONTROL_PLANE_ADDRESS"] != "controller:8443" || + !slices.Contains(service.Volumes, "./.control-plane-tls/"+expected.certificate+":/run/proxy-pool-tls/client:ro") || + !composeDependsOn(service.DependsOn, "controller") { + t.Errorf("%s control-plane wiring = env=%v volumes=%v dependsOn=%v", expected.name, service.Environment, service.Volumes, service.DependsOn) + } + if expected.name == "checker" { + if service.Environment["PROXY_POOL_CHECKER_ID"] != expected.identity || + service.Environment["PROXY_POOL_CHECKER_INSTANCE_ID"] != expected.instance || + service.Environment["PROXY_POOL_CHECKER_MAX_IN_FLIGHT"] != "200" || !slices.Contains(service.Expose, "9090") { + t.Errorf("checker identity or metrics wiring = env=%v expose=%v", service.Environment, service.Expose) + } + continue + } + if service.Environment["PROXY_POOL_WORKER_ID"] != expected.identity || + service.Environment["PROXY_POOL_INSTANCE_ID"] != expected.instance || + service.Environment["PROXY_POOL_CLUSTER_ID"] != "compose-local" || service.Environment["PROXY_POOL_ZONE"] != "compose-local" { + t.Errorf("%s identity wiring = %v", expected.name, service.Environment) + } + } +} + +func TestLocalControlPlaneCertificateGeneratorUsesDistinctSPIFFERoles(t *testing.T) { + payload, err := os.ReadFile("../scripts/generate-local-controlplane-certs.ps1") + if err != nil { + t.Fatalf("read generate-local-controlplane-certs.ps1: %v", err) + } + script := string(payload) + for _, required := range []string{ + "OutputDirectory must be empty", "DNS:controller", + "spiffe://proxy-pool.local/development/worker/gateway-a", + "spiffe://proxy-pool.local/development/worker/gateway-b", + "spiffe://proxy-pool.local/development/checker/checker-a", + } { + if !strings.Contains(script, required) { + t.Errorf("certificate generator missing %q", required) + } + } +} + func TestDeploymentEntrypointsExistInSource(t *testing.T) { dockerfile, err := os.ReadFile("docker/Dockerfile") if err != nil { @@ -160,13 +221,13 @@ func TestDeploymentEntrypointsExistInSource(t *testing.T) { } } -func TestKubernetesBaseExcludesUnimplementedProcessManifests(t *testing.T) { +func TestKubernetesBaseLeavesControlPlaneClientsForMTLSOverlay(t *testing.T) { payload, err := os.ReadFile("kubernetes/base/kustomization.yaml") if err != nil { t.Fatalf("read kustomization.yaml: %v", err) } if strings.Contains(string(payload), "checker.yaml") { - t.Fatal("kubernetes base includes checker.yaml before proxy-checker exists") + t.Fatal("kubernetes base includes checker.yaml without per-workload mTLS identity overlay") } } diff --git a/deploy/config/local.yaml b/deploy/config/local.yaml index d44a339..26c4c90 100644 --- a/deploy/config/local.yaml +++ b/deploy/config/local.yaml @@ -64,7 +64,30 @@ admin: token: "${PROXY_POOL_ADMIN_TOKEN}" controlPlane: - enabled: false + enabled: true + listen: 0.0.0.0:8443 + protocolVersion: 1 + heartbeatInterval: 10s + sessionTTL: 30s + maxStaleAge: 30s + maxMessageBytes: 4194304 + maxRuntimeCounters: 100000 + maxConcurrentStreams: 1000 + tls: + mode: mtls + certFile: /run/proxy-pool-tls/server/tls.crt + keyFile: /run/proxy-pool-tls/server/tls.key + clientCAFile: /run/proxy-pool-tls/server/ca.crt + trustDomain: proxy-pool.local + environment: development + gatewayTLS: + certFile: /run/proxy-pool-tls/client/tls.crt + keyFile: /run/proxy-pool-tls/client/tls.key + serverCAFile: /run/proxy-pool-tls/client/ca.crt + checkerTLS: + certFile: /run/proxy-pool-tls/client/tls.crt + keyFile: /run/proxy-pool-tls/client/tls.key + serverCAFile: /run/proxy-pool-tls/client/ca.crt metrics: enabled: true diff --git a/deploy/docker-compose.yml b/deploy/docker-compose.yml index 93179b6..e1fe6c6 100644 --- a/deploy/docker-compose.yml +++ b/deploy/docker-compose.yml @@ -15,8 +15,6 @@ x-app: &app image: proxy-pool:local restart: unless-stopped networks: [frontend, backend] - volumes: - - ./config/local.yaml:/etc/proxy-pool/config.yaml:ro environment: *app-environment stop_grace_period: 45s @@ -24,6 +22,19 @@ services: gateway-a: <<: *app command: ["proxy-gateway"] + environment: + <<: *app-environment + PROXY_POOL_CONTROL_PLANE_ADDRESS: controller:8443 + PROXY_POOL_CLUSTER_ID: compose-local + PROXY_POOL_WORKER_ID: gateway-a + PROXY_POOL_INSTANCE_ID: gateway-a-1 + PROXY_POOL_ZONE: compose-local + volumes: + - ./config/local.yaml:/etc/proxy-pool/config.yaml:ro + - ./.control-plane-tls/gateway-a:/run/proxy-pool-tls/client:ro + depends_on: + controller: + condition: service_healthy expose: ["8080", "9090"] healthcheck: test: ["CMD", "curl", "--fail", "--silent", "http://127.0.0.1:9090/readyz"] @@ -35,6 +46,19 @@ services: gateway-b: <<: *app command: ["proxy-gateway"] + environment: + <<: *app-environment + PROXY_POOL_CONTROL_PLANE_ADDRESS: controller:8443 + PROXY_POOL_CLUSTER_ID: compose-local + PROXY_POOL_WORKER_ID: gateway-b + PROXY_POOL_INSTANCE_ID: gateway-b-1 + PROXY_POOL_ZONE: compose-local + volumes: + - ./config/local.yaml:/etc/proxy-pool/config.yaml:ro + - ./.control-plane-tls/gateway-b:/run/proxy-pool-tls/client:ro + depends_on: + controller: + condition: service_healthy expose: ["8080", "9090"] healthcheck: test: ["CMD", "curl", "--fail", "--silent", "http://127.0.0.1:9090/readyz"] @@ -49,12 +73,15 @@ services: environment: <<: *app-environment PROXY_POOL_CONFIG_FINGERPRINT_KEY: ${PROXY_POOL_CONFIG_FINGERPRINT_KEY:?set PROXY_POOL_CONFIG_FINGERPRINT_KEY} + volumes: + - ./config/local.yaml:/etc/proxy-pool/config.yaml:ro + - ./.control-plane-tls/controller:/run/proxy-pool-tls/server:ro depends_on: postgres: condition: service_healthy redis: condition: service_healthy - expose: ["8081", "8082", "9090"] + expose: ["8081", "8082", "8443", "9090"] ports: - "127.0.0.1:8081:8081" - "127.0.0.1:8082:8082" @@ -65,6 +92,29 @@ services: retries: 12 start_period: 15s + checker: + <<: *app + command: ["proxy-checker"] + environment: + <<: *app-environment + PROXY_POOL_CONTROL_PLANE_ADDRESS: controller:8443 + PROXY_POOL_CHECKER_ID: checker-a + PROXY_POOL_CHECKER_INSTANCE_ID: checker-a-1 + PROXY_POOL_CHECKER_MAX_IN_FLIGHT: "200" + volumes: + - ./config/local.yaml:/etc/proxy-pool/config.yaml:ro + - ./.control-plane-tls/checker-a:/run/proxy-pool-tls/client:ro + depends_on: + controller: + condition: service_healthy + expose: ["9090"] + healthcheck: + test: ["CMD", "curl", "--fail", "--silent", "http://127.0.0.1:9090/readyz"] + interval: 5s + timeout: 2s + retries: 12 + start_period: 10s + haproxy: image: haproxy:3.2-alpine restart: unless-stopped diff --git a/deploy/prometheus/prometheus.yml b/deploy/prometheus/prometheus.yml index e979350..9f75f50 100644 --- a/deploy/prometheus/prometheus.yml +++ b/deploy/prometheus/prometheus.yml @@ -14,6 +14,9 @@ scrape_configs: - job_name: proxy-controller static_configs: - targets: [controller:9090] + - job_name: proxy-checker + static_configs: + - targets: [checker:9090] - job_name: haproxy metrics_path: /metrics static_configs: diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 7ba2c74..8d28a66 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -306,9 +306,9 @@ controlPlane: Gateway 连接 Controller 时使用独立启动参数而非 `controlPlane.listen`。至少设置 `PROXY_POOL_CONTROL_PLANE_ADDRESS`、`PROXY_POOL_CLUSTER_ID`、 `PROXY_POOL_WORKER_ID`、`PROXY_POOL_INSTANCE_ID` 和 `PROXY_POOL_ZONE`，详见 -[控制面协议](../api/control-plane.md#9-gateway-启动参数)。基础 Compose/Kubernetes -模板保持 `controlPlane.enabled: false`，环境 Overlay 挂载 mTLS 证书并启用后才可启动 -Gateway。 +[控制面协议](../api/control-plane.md#9-gateway-启动参数)。Compose 本地模板使用固定 +Worker/Checker 身份和被忽略的开发证书目录；Kubernetes Base 保持 +`controlPlane.enabled: false`，必须由具备每工作负载唯一身份的 mTLS Overlay 启用。 Checker 同样使用独立的可拨号地址：`proxy-checker` 的 `-control-plane`、 `-checker-id`、`-instance-id` 和 `-max-in-flight` 可由对应的 diff --git a/docs/operations/runbook.md b/docs/operations/runbook.md index 3f1450e..6c73d77 100644 --- a/docs/operations/runbook.md +++ b/docs/operations/runbook.md @@ -50,6 +50,13 @@ $env:PROVIDER_A_TOKEN = "PROVIDER_A_TOKEN" $env:PROVIDER_B_TOKEN = "PROVIDER_B_TOKEN" ``` +首次启动 Compose 前，生成不入库的本地控制面证书。生成器要求目标目录为空，避免覆写 +已有密钥： + +```powershell +.\scripts\generate-local-controlplane-certs.ps1 +``` + ### 2.3 静态检查 ```powershell @@ -57,6 +64,10 @@ docker compose -f deploy/docker-compose.yml config kubectl kustomize deploy/kubernetes/base > rendered.yaml ``` +Compose 启动后，Gateway 的 `/readyz` 需要先接收有效 Snapshot；Checker 的 `/readyz` +需要成功建立一次任务领取流，并会在后续领取失败时回到未就绪。两者的 `/livez` 只表示 +进程仍在运行。 + 本地 Compose 的 Redis 只作为可重建短效状态 fixture，固定使用 `--appendonly no --save ""`，且不挂载 `/data` 或命名卷。真实 Redis 8.2 契约可 通过 `.\scripts\test-redis.ps1` 执行；脚本使用唯一命名空间并在结束时定向清理， diff --git a/internal/checker/bootstrap/bootstrap.go b/internal/checker/bootstrap/bootstrap.go index 78e7203..f2bfde3 100644 --- a/internal/checker/bootstrap/bootstrap.go +++ b/internal/checker/bootstrap/bootstrap.go @@ -8,9 +8,13 @@ import ( "context" "errors" "fmt" + "net" + "net/http" "reflect" "strings" + "sync/atomic" + "github.com/prometheus/client_golang/prometheus" "google.golang.org/grpc" "google.golang.org/grpc/credentials" @@ -20,11 +24,15 @@ import ( "proxy-pool/internal/config" "proxy-pool/internal/controlplane/clienttransport" "proxy-pool/internal/domain/workerruntime" + "proxy-pool/internal/platform/httpserver" + "proxy-pool/internal/platform/lifecycle" + platformMetrics "proxy-pool/internal/platform/metrics" ) var ( ErrInvalidOptions = errors.New("invalid checker bootstrap options") ErrStartup = errors.New("checker startup failed") + ErrNotReady = errors.New("checker control plane is not ready") ) type Options struct { @@ -36,6 +44,8 @@ type Options struct { MaxInFlight int SupportedLevels []controlplanev1.CheckLevel GRPCTransport credentials.TransportCredentials + MetricsListener net.Listener + HTTP httpserver.Options } func Run(ctx context.Context, options Options) error { @@ -49,11 +59,23 @@ func Run(ctx context.Context, options Options) error { if !configuration.ControlPlane.Enabled { return errors.Join(ErrInvalidOptions, errors.New("controlPlane must be enabled")) } + runtime, err := newRuntime(ctx, configuration, options) + if err != nil { + return fmt.Errorf("%w: %w", ErrStartup, err) + } + defer runtime.Close() + return runtime.Run(ctx) +} + +func newRuntime(ctx context.Context, configuration *config.Config, options Options) (*runtime, error) { + if ctx == nil || configuration == nil { + return nil, ErrInvalidOptions + } transport, err := clienttransport.New( configuration.ControlPlane, options.ControlPlaneAddress, configuration.ControlPlane.CheckerTLS, options.GRPCTransport, "checkerTLS", ) if err != nil { - return fmt.Errorf("%w: %w", ErrInvalidOptions, err) + return nil, fmt.Errorf("%w: %w", ErrInvalidOptions, err) } connection, err := grpc.NewClient(options.ControlPlaneAddress, grpc.WithTransportCredentials(transport), @@ -63,19 +85,40 @@ func Run(ctx context.Context, options Options) error { ), ) if err != nil { - return fmt.Errorf("%w: dial control plane: %w", ErrStartup, err) + return nil, fmt.Errorf("dial control plane: %w", err) } - defer connection.Close() + closeConnection := true + defer func() { + if closeConnection { + _ = connection.Close() + } + }() client := controlplane.NewGeneratedClient(controlplanev1.NewCheckerControlPlaneClient(connection)) + readiness := &controlPlaneReadiness{} runner, err := controlplane.NewRunner(client, probe.NewExecutor(), controlplane.Options{ CheckerID: options.CheckerID, InstanceID: options.InstanceID, MaxInFlight: options.MaxInFlight, SupportedLevels: append([]controlplanev1.CheckLevel(nil), options.SupportedLevels...), - ReportBatchSize: options.MaxInFlight, + ReportBatchSize: options.MaxInFlight, OnSuccessfulPull: readiness.MarkReady, OnFailedPull: readiness.MarkUnavailable, }) if err != nil { - return fmt.Errorf("%w: build checker runner: %w", ErrStartup, err) + return nil, fmt.Errorf("build checker runner: %w", err) } - return runner.Run(ctx) + runners := []lifecycle.Runner{runner} + if configuration.Metrics.Enabled { + metrics, metricsErr := newMetricsRuntime(ctx, configuration.Metrics, options, readiness) + if metricsErr != nil { + return nil, metricsErr + } + runners = append(runners, metrics) + } else if options.MetricsListener != nil { + return nil, ErrInvalidOptions + } + group, err := lifecycle.NewGroup(runners...) + if err != nil { + return nil, err + } + closeConnection = false + return &runtime{connection: connection, group: group}, nil } func validateOptions(ctx context.Context, options Options) error { @@ -88,6 +131,87 @@ func validateOptions(ctx context.Context, options Options) error { return nil } +type runtime struct { + connection *grpc.ClientConn + group *lifecycle.Group +} + +func (runtime *runtime) Run(ctx context.Context) error { + if runtime == nil || runtime.connection == nil || runtime.group == nil || ctx == nil { + return ErrInvalidOptions + } + return runtime.group.Run(ctx) +} + +func (runtime *runtime) Close() { + if runtime != nil && runtime.connection != nil { + _ = runtime.connection.Close() + } +} + +type controlPlaneReadiness struct { + ready atomic.Bool +} + +func (readiness *controlPlaneReadiness) MarkReady() { + if readiness != nil { + readiness.ready.Store(true) + } +} + +func (readiness *controlPlaneReadiness) MarkUnavailable() { + if readiness != nil { + readiness.ready.Store(false) + } +} + +func (readiness *controlPlaneReadiness) Ready(ctx context.Context) error { + if ctx == nil || readiness == nil || !readiness.ready.Load() { + return ErrNotReady + } + return ctx.Err() +} + +type metricsRuntime struct { + listener net.Listener + handler http.Handler + options httpserver.Options +} + +func newMetricsRuntime( + ctx context.Context, + configuration config.Metrics, + options Options, + readiness *controlPlaneReadiness, +) (*metricsRuntime, error) { + if ctx == nil || !configuration.Enabled || readiness == nil { + return nil, ErrInvalidOptions + } + handler, err := platformMetrics.NewHandler(platformMetrics.Dependencies{ + Gatherer: prometheus.DefaultGatherer, Readiness: readiness, + }) + if err != nil { + return nil, fmt.Errorf("build checker metrics handler: %w", err) + } + listener := options.MetricsListener + if listener == nil { + listener, err = (&net.ListenConfig{}).Listen(ctx, "tcp", configuration.Listen) + if err != nil { + return nil, fmt.Errorf("listen checker metrics: %w", err) + } + } + return &metricsRuntime{listener: listener, handler: handler, options: options.HTTP}, nil +} + +func (runtime *metricsRuntime) Run(ctx context.Context) error { + if runtime == nil || runtime.listener == nil || runtime.handler == nil || ctx == nil { + return ErrInvalidOptions + } + return httpserver.Serve(ctx, runtime.options, httpserver.Endpoint{ + Name: "metrics", Listener: runtime.listener, Handler: runtime.handler, + }) +} + func loadConfiguration(ctx context.Context, path string, resolver config.Resolver) (*config.Config, error) { if ctx == nil || strings.TrimSpace(path) != path || path == "" || nilInterface(resolver) { return nil, ErrInvalidOptions diff --git a/internal/checker/bootstrap/bootstrap_test.go b/internal/checker/bootstrap/bootstrap_test.go new file mode 100644 index 0000000..f9ed574 --- /dev/null +++ b/internal/checker/bootstrap/bootstrap_test.go @@ -0,0 +1,55 @@ +package bootstrap + +import ( + "context" + "errors" + "net" + "net/http" + "net/http/httptest" + "testing" + + "proxy-pool/internal/config" +) + +func TestControlPlaneReadinessTransitionsAfterSuccessfulPull(t *testing.T) { + readiness := &controlPlaneReadiness{} + if err := readiness.Ready(context.Background()); !errors.Is(err, ErrNotReady) { + t.Fatalf("Ready() before pull = %v, want ErrNotReady", err) + } + readiness.MarkReady() + if err := readiness.Ready(context.Background()); err != nil { + t.Fatalf("Ready() after successful pull = %v", err) + } + readiness.MarkUnavailable() + if err := readiness.Ready(context.Background()); !errors.Is(err, ErrNotReady) { + t.Fatalf("Ready() after failed pull = %v, want ErrNotReady", err) + } +} + +func TestMetricsRuntimeUsesControlPlaneReadiness(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Listen(): %v", err) + } + defer listener.Close() + readiness := &controlPlaneReadiness{} + runtime, err := newMetricsRuntime(context.Background(), config.Metrics{Enabled: true, Listen: listener.Addr().String()}, Options{ + MetricsListener: listener, + }, readiness) + if err != nil { + t.Fatalf("newMetricsRuntime(): %v", err) + } + assertCheckerProbe(t, runtime.handler, "/livez", http.StatusOK, "live\n") + assertCheckerProbe(t, runtime.handler, "/readyz", http.StatusServiceUnavailable, "unavailable\n") + readiness.MarkReady() + assertCheckerProbe(t, runtime.handler, "/readyz", http.StatusOK, "ready\n") +} + +func assertCheckerProbe(t *testing.T, handler http.Handler, path string, wantStatus int, wantBody string) { + t.Helper() + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, path, nil)) + if recorder.Code != wantStatus || recorder.Body.String() != wantBody { + t.Fatalf("GET %s = %d %q, want %d %q", path, recorder.Code, recorder.Body.String(), wantStatus, wantBody) + } +} diff --git a/internal/checker/controlplane/runner.go b/internal/checker/controlplane/runner.go index cc7d5c6..3825a08 100644 --- a/internal/checker/controlplane/runner.go +++ b/internal/checker/controlplane/runner.go @@ -44,14 +44,16 @@ type Executor interface { } type Options struct { - CheckerID string - InstanceID string - MaxInFlight int - SupportedLevels []controlplanev1.CheckLevel - ReportBatchSize int - PollInterval time.Duration - RetryDelay time.Duration - Now func() time.Time + CheckerID string + InstanceID string + MaxInFlight int + SupportedLevels []controlplanev1.CheckLevel + ReportBatchSize int + PollInterval time.Duration + RetryDelay time.Duration + Now func() time.Time + OnSuccessfulPull func() + OnFailedPull func() } // Runner consumes at most one server-bounded pull per RunOnce. Run adds a @@ -93,6 +95,9 @@ func (runner *Runner) Run(ctx context.Context) error { if err != nil && ctx.Err() != nil { return ctx.Err() } + if err != nil && runner.options.OnFailedPull != nil { + runner.options.OnFailedPull() + } wait := runner.options.PollInterval if err != nil { wait = runner.options.RetryDelay @@ -118,9 +123,13 @@ func (runner *Runner) RunOnce(ctx context.Context) error { return err } tasks, err := collectTasks(stream, runner.options.MaxInFlight) - if err != nil || len(tasks) == 0 { + if err != nil { return err } + if len(tasks) == 0 { + runner.observeSuccessfulPull() + return nil + } observations := runner.executeTasks(ctx, tasks) for start := 0; start < len(observations); start += runner.options.ReportBatchSize { end := start + runner.options.ReportBatchSize @@ -137,9 +146,16 @@ func (runner *Runner) RunOnce(ctx context.Context) error { return ErrObservationsRejected } } + runner.observeSuccessfulPull() return nil } +func (runner *Runner) observeSuccessfulPull() { + if runner != nil && runner.options.OnSuccessfulPull != nil { + runner.options.OnSuccessfulPull() + } +} + func collectTasks(stream TaskStream, maximum int) ([]*controlplanev1.CheckTask, error) { if stream == nil || maximum <= 0 { return nil, ErrInvalidRunner diff --git a/internal/checker/controlplane/runner_test.go b/internal/checker/controlplane/runner_test.go index df8d55b..b947bf5 100644 --- a/internal/checker/controlplane/runner_test.go +++ b/internal/checker/controlplane/runner_test.go @@ -2,8 +2,10 @@ package controlplane import ( "context" + "errors" "io" "sync" + "sync/atomic" "testing" "time" @@ -67,6 +69,51 @@ func TestRunnerOmitsEgressProbeURLFromGlobalObservation(t *testing.T) { } } +func TestRunnerMarksReadyAfterSuccessfulEmptyPull(t *testing.T) { + var successfulPulls atomic.Int64 + runner, err := NewRunner(&clientStub{stream: &taskStreamStub{}}, &executorStub{}, Options{ + CheckerID: "checker-a", InstanceID: "instance-a", MaxInFlight: 1, + SupportedLevels: []controlplanev1.CheckLevel{controlplanev1.CheckLevel_CHECK_LEVEL_BASIC}, + Now: time.Now, OnSuccessfulPull: func() { successfulPulls.Add(1) }, + }) + if err != nil { + t.Fatalf("NewRunner(): %v", err) + } + if err := runner.RunOnce(context.Background()); err != nil { + t.Fatalf("RunOnce(): %v", err) + } + if successfulPulls.Load() != 1 { + t.Fatalf("successful pulls = %d, want 1", successfulPulls.Load()) + } +} + +func TestRunnerMarksUnavailableAfterPullFailure(t *testing.T) { + failed := make(chan struct{}) + var markOnce sync.Once + runner, err := NewRunner(&clientStub{streamErr: errors.New("control plane unavailable")}, &executorStub{}, Options{ + CheckerID: "checker-a", InstanceID: "instance-a", MaxInFlight: 1, + SupportedLevels: []controlplanev1.CheckLevel{controlplanev1.CheckLevel_CHECK_LEVEL_BASIC}, + Now: time.Now, RetryDelay: time.Millisecond, + OnFailedPull: func() { markOnce.Do(func() { close(failed) }) }, + }) + if err != nil { + t.Fatalf("NewRunner(): %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + completed := make(chan error, 1) + go func() { completed <- runner.Run(ctx) }() + select { + case <-failed: + cancel() + case <-time.After(time.Second): + t.Fatal("Runner did not mark failed pull") + } + if err := <-completed; !errors.Is(err, context.Canceled) { + t.Fatalf("Run() = %v, want context.Canceled", err) + } +} + func checkerTask(id string, now time.Time) *controlplanev1.CheckTask { return &controlplanev1.CheckTask{ TaskId: id, ProxyId: id + "-proxy", Protocol: controlplanev1.ProxyProtocol_PROXY_PROTOCOL_HTTP, @@ -76,12 +123,13 @@ func checkerTask(id string, now time.Time) *controlplanev1.CheckTask { } type clientStub struct { - stream TaskStream - batches []*controlplanev1.ObservationBatch + stream TaskStream + streamErr error + batches []*controlplanev1.ObservationBatch } func (stub *clientStub) StreamCheckTasks(context.Context, *controlplanev1.StreamCheckTasksRequest) (TaskStream, error) { - return stub.stream, nil + return stub.stream, stub.streamErr } func (stub *clientStub) ReportObservations(_ context.Context, batch *controlplanev1.ObservationBatch) (*controlplanev1.ReportObservationsResponse, error) { diff --git a/scripts/generate-local-controlplane-certs.ps1 b/scripts/generate-local-controlplane-certs.ps1 new file mode 100644 index 0000000..be459f9 --- /dev/null +++ b/scripts/generate-local-controlplane-certs.ps1 @@ -0,0 +1,138 @@ +param( + [string]$OutputDirectory = (Join-Path (Split-Path -Parent $PSScriptRoot) "deploy/.control-plane-tls") +) + +$ErrorActionPreference = "Stop" + +if ([string]::IsNullOrWhiteSpace($OutputDirectory)) { + throw "OutputDirectory is required" +} + +if (Test-Path -LiteralPath $OutputDirectory) { + $existing = @(Get-ChildItem -Force -LiteralPath $OutputDirectory) + if ($existing.Count -gt 0) { + throw "OutputDirectory must be empty: $OutputDirectory" + } +} +else { + New-Item -ItemType Directory -Path $OutputDirectory | Out-Null +} + +function ConvertTo-Pem { + param( + [string]$Label, + [byte[]]$Bytes + ) + + $base64 = [Convert]::ToBase64String($Bytes, [Base64FormattingOptions]::InsertLineBreaks) + return "-----BEGIN $Label-----`n$base64`n-----END $Label-----`n" +} + +function Write-PemFile { + param( + [string]$Path, + [string]$Label, + [byte[]]$Bytes + ) + + [IO.File]::WriteAllText($Path, (ConvertTo-Pem -Label $Label -Bytes $Bytes), [Text.Encoding]::ASCII) +} + +function New-CertificateRequest { + param( + [string]$CommonName, + [System.Security.Cryptography.RSA]$Key, + [bool]$IsCertificateAuthority, + [string]$ExtendedKeyUsage, + [string]$SubjectAlternativeName + ) + + $request = [System.Security.Cryptography.X509Certificates.CertificateRequest]::new( + "CN=$CommonName", + $Key, + [System.Security.Cryptography.HashAlgorithmName]::SHA256, + [System.Security.Cryptography.RSASignaturePadding]::Pkcs1 + ) + $request.CertificateExtensions.Add([System.Security.Cryptography.X509Certificates.X509BasicConstraintsExtension]::new( + $IsCertificateAuthority, $false, 0, $true + )) + if ($IsCertificateAuthority) { + $request.CertificateExtensions.Add([System.Security.Cryptography.X509Certificates.X509KeyUsageExtension]::new( + [System.Security.Cryptography.X509Certificates.X509KeyUsageFlags]::KeyCertSign -bor + [System.Security.Cryptography.X509Certificates.X509KeyUsageFlags]::CrlSign, + $true + )) + return $request + } + $request.CertificateExtensions.Add([System.Security.Cryptography.X509Certificates.X509KeyUsageExtension]::new( + [System.Security.Cryptography.X509Certificates.X509KeyUsageFlags]::DigitalSignature -bor + [System.Security.Cryptography.X509Certificates.X509KeyUsageFlags]::KeyEncipherment, + $true + )) + $usage = [System.Security.Cryptography.OidCollection]::new() + if ($ExtendedKeyUsage -eq "serverAuth") { + $null = $usage.Add([System.Security.Cryptography.Oid]::new("1.3.6.1.5.5.7.3.1")) + } + else { + $null = $usage.Add([System.Security.Cryptography.Oid]::new("1.3.6.1.5.5.7.3.2")) + } + $request.CertificateExtensions.Add([System.Security.Cryptography.X509Certificates.X509EnhancedKeyUsageExtension]::new($usage, $true)) + $san = [System.Security.Cryptography.X509Certificates.SubjectAlternativeNameBuilder]::new() + if ($SubjectAlternativeName.StartsWith("DNS:")) { + $san.AddDnsName($SubjectAlternativeName.Substring(4)) + } + else { + $san.AddUri([Uri]$SubjectAlternativeName.Substring(4)) + } + $request.CertificateExtensions.Add($san.Build()) + return $request +} + +function New-LeafCertificate { + param( + [string]$Name, + [string]$CommonName, + [string]$ExtendedKeyUsage, + [string]$SubjectAlternativeName + ) + + $directory = Join-Path $OutputDirectory $Name + New-Item -ItemType Directory -Path $directory | Out-Null + $key = [System.Security.Cryptography.RSA]::Create(2048) + try { + $request = New-CertificateRequest -CommonName $CommonName -Key $key -IsCertificateAuthority $false -ExtendedKeyUsage $ExtendedKeyUsage -SubjectAlternativeName $SubjectAlternativeName + $serial = [byte[]]::new(16) + [System.Security.Cryptography.RandomNumberGenerator]::Fill($serial) + $certificate = $request.Create($caCertificate, $notBefore, $notAfter, $serial) + try { + Write-PemFile -Path (Join-Path $directory "tls.crt") -Label "CERTIFICATE" -Bytes $certificate.Export([System.Security.Cryptography.X509Certificates.X509ContentType]::Cert) + Write-PemFile -Path (Join-Path $directory "tls.key") -Label "PRIVATE KEY" -Bytes $key.ExportPkcs8PrivateKey() + } + finally { + $certificate.Dispose() + } + } + finally { + $key.Dispose() + } + Copy-Item -LiteralPath $caCertificatePath -Destination (Join-Path $directory "ca.crt") +} + +$notBefore = [DateTimeOffset]::UtcNow.AddMinutes(-5) +$notAfter = $notBefore.AddDays(7) +$caKey = [System.Security.Cryptography.RSA]::Create(2048) +$caRequest = New-CertificateRequest -CommonName "proxy-pool-local-control-plane-ca" -Key $caKey -IsCertificateAuthority $true -ExtendedKeyUsage "" -SubjectAlternativeName "" +$caCertificate = $caRequest.CreateSelfSigned($notBefore, $notAfter) +$caCertificatePath = Join-Path $OutputDirectory "ca.crt" +Write-PemFile -Path $caCertificatePath -Label "CERTIFICATE" -Bytes $caCertificate.Export([System.Security.Cryptography.X509Certificates.X509ContentType]::Cert) +Write-PemFile -Path (Join-Path $OutputDirectory "ca.key") -Label "PRIVATE KEY" -Bytes $caKey.ExportPkcs8PrivateKey() + +New-LeafCertificate -Name "controller" -CommonName "proxy-pool-controller" -ExtendedKeyUsage "serverAuth" -SubjectAlternativeName "DNS:controller" +New-LeafCertificate -Name "gateway-a" -CommonName "proxy-pool-gateway-a" -ExtendedKeyUsage "clientAuth" -SubjectAlternativeName "URI:spiffe://proxy-pool.local/development/worker/gateway-a" +New-LeafCertificate -Name "gateway-b" -CommonName "proxy-pool-gateway-b" -ExtendedKeyUsage "clientAuth" -SubjectAlternativeName "URI:spiffe://proxy-pool.local/development/worker/gateway-b" +New-LeafCertificate -Name "checker-a" -CommonName "proxy-pool-checker-a" -ExtendedKeyUsage "clientAuth" -SubjectAlternativeName "URI:spiffe://proxy-pool.local/development/checker/checker-a" + +$caCertificate.Dispose() +$caKey.Dispose() + +Write-Host "Generated local control-plane certificates in $OutputDirectory"