diff --git a/Dockerfile b/Dockerfile index 5b59f51..22c5041 100644 --- a/Dockerfile +++ b/Dockerfile @@ -19,7 +19,7 @@ COPY . . # was called. For example, if we call make docker-build in a local env which has the Apple Silicon M1 SO # the docker BUILDPLATFORM arg will be linux/arm64 when for Apple x86 it will be linux/amd64. Therefore, # by leaving it empty we can ensure that the container and binary shipped on it will have the same platform. -RUN CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH} go build -a -o manager cmd/main.go +RUN CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH} go build -a -o manager ./cmd # Use distroless as minimal base image to package the manager binary # Refer to https://github.com/GoogleContainerTools/distroless for more details diff --git a/Makefile b/Makefile index 998a3f2..0696602 100644 --- a/Makefile +++ b/Makefile @@ -132,11 +132,11 @@ lint-config: golangci-lint ## Verify golangci-lint linter configuration .PHONY: build build: manifests generate fmt vet ## Build manager binary. - go build -o bin/manager cmd/main.go + go build -o bin/manager ./cmd .PHONY: run run: manifests generate fmt vet ## Run a controller from your host. - go run ./cmd/main.go + go run ./cmd # If you wish to build the manager image targeting other platforms you can use the --platform flag. # (i.e. docker build --platform linux/arm64). However, you must enable docker buildKit for it. diff --git a/cmd/config.go b/cmd/config.go new file mode 100644 index 0000000..c0a487d --- /dev/null +++ b/cmd/config.go @@ -0,0 +1,71 @@ +/* +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 main + +import ( + "errors" + "flag" + "strings" + "time" + + "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/openbao" + bao "github.com/openbao/openbao/api/v2" +) + +// dependencyOptions belongs to process startup, never to an Instance. +// Configuration is immutable after assembly; no configuration hot reload is enabled. +type dependencyOptions struct { + address string + consumerAddress string + authMount string + authRole string + kvMount string + tenantBasePath string + externalSecretStoreName string + postgreSQLCABundlePath string + timeout time.Duration +} + +func (o *dependencyOptions) bindFlags(flags *flag.FlagSet) { + flags.StringVar(&o.address, "openbao-address", "", "OpenBao API address.") + flags.StringVar(&o.consumerAddress, "openbao-consumer-address", "", "Consumer OpenBao API address.") + flags.StringVar(&o.authMount, "openbao-auth-mount", "kubernetes", "OpenBao Kubernetes auth mount.") + flags.StringVar(&o.authRole, "openbao-auth-role", "", "OpenBao Kubernetes auth role.") + flags.StringVar(&o.kvMount, "openbao-kv-mount", "kv", "OpenBao KV v2 mount.") + flags.StringVar(&o.tenantBasePath, "openbao-tenant-base-path", "postgresql-tenants", "Tenant credential prefix.") + flags.StringVar(&o.externalSecretStoreName, "external-secret-store-name", "", "ESO ClusterSecretStore name.") + flags.StringVar(&o.postgreSQLCABundlePath, "postgresql-ca-bundle-path", "", "PostgreSQL CA bundle path.") + flags.DurationVar(&o.timeout, "reconcile-timeout", 30*time.Second, "External operation deadline.") +} + +func (o dependencyOptions) credentials() (*openbao.Credentials, error) { + if o.address == "" || o.authRole == "" || o.authMount == "" || o.kvMount == "" { + return nil, errors.New("OpenBao address, auth role, auth mount and KV mount are required") + } + if o.timeout <= 0 { + return nil, errors.New("reconcile timeout must be positive") + } + // Delegate connection address parsing to the SDK, and avoid implicit BAO_* overrides. + config := bao.NewConfig() + config.Address = strings.TrimRight(o.address, "/") + config.Timeout = o.timeout + client, err := bao.NewClient(config) + if err != nil { + return nil, errors.New("invalid OpenBao client configuration") + } + return openbao.NewCredentials(client, o.kvMount, o.authMount, o.authRole), nil +} diff --git a/cmd/config_test.go b/cmd/config_test.go new file mode 100644 index 0000000..7f491f1 --- /dev/null +++ b/cmd/config_test.go @@ -0,0 +1,56 @@ +/* +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 main + +import ( + "flag" + "io" + "testing" +) + +const testAddressFlag = "--openbao-address=http://bao:8200" + +func TestDependencyConfigurationFailsBeforeAssembly(t *testing.T) { + for _, args := range [][]string{ + {}, + {testAddressFlag}, + {testAddressFlag, "--openbao-auth-role=controller", "--reconcile-timeout=0s"}, + } { + var options dependencyOptions + flags := flag.NewFlagSet("test", flag.ContinueOnError) + flags.SetOutput(io.Discard) + options.bindFlags(flags) + if err := flags.Parse(args); err != nil { + t.Fatal(err) + } + if _, err := options.credentials(); err == nil { + t.Fatal("invalid configuration was accepted") + } + } +} + +func TestAssemblyDoesNotRequireLiveDependenciesOrTenantConfiguration(t *testing.T) { + var options dependencyOptions + flags := flag.NewFlagSet("test", flag.ContinueOnError) + options.bindFlags(flags) + if err := flags.Parse([]string{testAddressFlag, "--openbao-auth-role=controller"}); err != nil { + t.Fatal(err) + } + if _, err := options.credentials(); err != nil { + t.Fatal(err) + } +} diff --git a/cmd/main.go b/cmd/main.go index 81416db..6de778a 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -20,7 +20,6 @@ import ( "crypto/tls" "flag" "os" - "time" // Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.) // to ensure that exec-entrypoint and run can make use of them. @@ -38,7 +37,8 @@ import ( databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/controller" - instanceinitializer "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/instance" + "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/instance" + "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/postgresql" // +kubebuilder:scaffold:imports ) @@ -63,10 +63,7 @@ func main() { var probeAddr string var secureMetrics bool var enableHTTP2 bool - var openBaoAddress, openBaoConsumerAddress, openBaoAuthMount, openBaoAuthRole string - var openBaoKVMount, openBaoTenantBasePath, openBaoServiceAccountTokenPath string - var externalSecretStoreName, postgreSQLCABundlePath string - var reconcileTimeout time.Duration + var dependencies dependencyOptions var tlsOpts []func(*tls.Config) 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.") @@ -85,20 +82,8 @@ func main() { flag.StringVar(&metricsCertKey, "metrics-cert-key", "tls.key", "The name of the metrics server key file.") flag.BoolVar(&enableHTTP2, "enable-http2", false, "If set, HTTP/2 will be enabled for the metrics and webhook servers") - flag.StringVar(&openBaoAddress, "openbao-address", "", "OpenBao API address used by the controller.") - flag.StringVar(&openBaoConsumerAddress, "openbao-consumer-address", "", "OpenBao API address exposed to consumers.") - flag.StringVar(&openBaoAuthMount, "openbao-auth-mount", "kubernetes", "OpenBao Kubernetes auth mount.") - flag.StringVar(&openBaoAuthRole, "openbao-auth-role", "", "OpenBao Kubernetes auth role.") - flag.StringVar(&openBaoKVMount, "openbao-kv-mount", "kv", "OpenBao KV v2 mount.") - flag.StringVar(&openBaoServiceAccountTokenPath, "openbao-service-account-token-path", - "/var/run/secrets/kubernetes.io/serviceaccount/token", - "Projected service account token used for OpenBao authentication.") - flag.StringVar(&openBaoTenantBasePath, "openbao-tenant-base-path", "postgresql-tenants", - "Tenant credential base path.") - flag.StringVar(&externalSecretStoreName, "external-secret-store-name", "", "ESO ClusterSecretStore name.") - flag.StringVar(&postgreSQLCABundlePath, "postgresql-ca-bundle-path", "", "PostgreSQL CA bundle path.") - flag.DurationVar(&reconcileTimeout, "reconcile-timeout", 30*time.Second, - "Deadline for external operations in one reconcile.") + dependencies.bindFlags(flag.CommandLine) + opts := zap.Options{ Development: true, } @@ -107,17 +92,15 @@ func main() { ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts))) - instanceInitializer, err := instanceinitializer.New(instanceinitializer.Config{ - OpenBaoAddress: openBaoAddress, OpenBaoConsumerAddress: openBaoConsumerAddress, - OpenBaoAuthMount: openBaoAuthMount, OpenBaoAuthRole: openBaoAuthRole, - ServiceAccountTokenPath: openBaoServiceAccountTokenPath, OpenBaoKVMount: openBaoKVMount, - OpenBaoTenantBasePath: openBaoTenantBasePath, ExternalSecretStoreName: externalSecretStoreName, - PostgreSQLCABundlePath: postgreSQLCABundlePath, Timeout: reconcileTimeout, - }) + credentials, err := dependencies.credentials() if err != nil { setupLog.Error(err, "Invalid controller dependency configuration") os.Exit(1) } + instances := instance.NewService(credentials, postgresql.Connector{ + CABundlePath: dependencies.postgreSQLCABundlePath, + }) + defer instances.Close() // if the enable-http2 flag is false (the default), http/2 should be disabled // due to its vulnerabilities. More specifically, disabling http/2 will @@ -211,7 +194,10 @@ func main() { } if err := (&controller.PostgreSQLInstanceReconciler{ - Client: mgr.GetClient(), Scheme: mgr.GetScheme(), Initializer: instanceInitializer, Timeout: reconcileTimeout, + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + Instances: instances, + Timeout: dependencies.timeout, }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "Failed to create controller", "controller", "postgresqlinstance") os.Exit(1) @@ -237,6 +223,7 @@ func main() { setupLog.Info("Starting manager") if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil { setupLog.Error(err, "Failed to run manager") + instances.Close() os.Exit(1) } } diff --git a/config/manager/manager.yaml b/config/manager/manager.yaml index 602c828..139304b 100644 --- a/config/manager/manager.yaml +++ b/config/manager/manager.yaml @@ -63,9 +63,6 @@ spec: args: - --leader-elect - --health-probe-bind-address=:8081 - - --openbao-address=https://openbao.openbao.svc:8200 - - --openbao-auth-role=postgresql-tenant-operator - - --external-secret-store-name=openbao image: controller:latest name: manager ports: diff --git a/docs/deployment.md b/docs/deployment.md index c4de169..b5dfa80 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -25,8 +25,8 @@ ## Controller 配置合同 -controller 使用以下 CLI flags。必填项缺失、路径无效或 duration 不为正数时，进程必须 -在启动 manager 前失败；不得等到 reconcile 时才逐个资源报告配置错误。 +controller 使用以下 CLI flags。启动入口加载配置、检查本切片必需项并创建共享依赖， +配置错误在启动 manager 前失败。外部服务暂时不可用由 reconcile 重试，不阻止进程启动。 | CLI flag | 必填/默认 | 说明 | | --- | --- | --- | @@ -35,16 +35,28 @@ controller 使用以下 CLI flags。必填项缺失、路径无效或 duration | `--openbao-auth-mount` | `kubernetes` | Kubernetes auth mount 名称 | | `--openbao-auth-role` | 必填 | controller ServiceAccount 对应 role | | `--openbao-kv-mount` | `kv` | KV v2 mount；开发可显式用 `secret` | -| `--openbao-service-account-token-path` | `/var/run/secrets/kubernetes.io/serviceaccount/token` | Kubernetes auth 使用的投射 token 文件 | | `--openbao-tenant-base-path` | 默认 `postgresql-tenants` | controller 专属 mount-relative 前缀 | -| `--external-secret-store-name` | 必填 | controller 创建的 ExternalSecret 固定引用 | +| `--external-secret-store-name` | Tenant 投射时必填 | controller 创建的 ExternalSecret 固定引用；Instance Ready 不依赖此项 | | `--postgresql-ca-bundle-path` | PostgreSQL TLS 模式必填 | 只读 PEM trust bundle，不含私钥 | | `--reconcile-timeout` | `30s` | 单轮 reconcile 中外部操作的总期限，必须大于零 | -address 必须是绝对 `http` 或 `https` URL，不允许 userinfo、query 或 fragment，末尾 `/` -在规范化后移除。mount、auth mount 和 base path 都使用 mount-relative path 语义，不以 -`/` 开头，不含空段、`.` 或 `..`；base path 还不得编码 KV v2 的 `data`/`metadata` -API 层。生产环境的 `--openbao-address` 必须使用 HTTPS；HTTP 只用于明确的开发 fixture。 +连接地址由 OpenBao SDK 解析，移除末尾 `/`；配置错误不得携带原始地址中的认证信息。 +生产环境使用 HTTPS；HTTP 只用于开发 fixture。consumer URL 的认证信息限制和 Tenant +派生路径限制由对应输出边界负责，Instance 服务不校验尚未使用的 Tenant 配置。 +Kubernetes auth 使用 SDK 默认的 ServiceAccount token 挂载路径，不单独暴露路径参数。 + +启动时创建一个共享 OpenBao client 和凭据源，再注入 Instance 服务。凭据源在有读取需求且 +token 即将过期时重新登录；每次登录重新读取投射的 ServiceAccount token。token 被撤销导致 +403 时最多重新认证一次，持续 policy 拒绝仍返回错误。 + +每个 Instance 首次使用时读取管理 username/password 并创建自己的 PostgreSQL 连接池。 +后续 reconcile 复用连接池和内存中的凭据，不做自动 PostgreSQL 密码轮换。UID、endpoint +或管理凭据引用改变时释放旧连接并重新装配；只修改 extension allowlist 不重建连接。 +删除 Instance 或 controller 正常退出时关闭连接池；重启后按需重新读取 Bao。 +仅修改 Bao 中原路径的密码不会触发刷新，管理员需要重启 controller 或修改凭据引用。 + +基础 manager manifest 不携带 OpenBao 地址、role 等环境值。实际部署通过环境专属 +Kustomize overlay 注入上述 args；E2E 由测试 fixture 注入一次性环境配置。 Tenant 路径固定推导为 `//`。namespace/name 都已通过 Kubernetes 名称校验，因此不再允许 CR 提供任意路径。KV v2 API URL 使用 consumer diff --git a/docs/operations.md b/docs/operations.md index e98d45b..839d798 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -38,6 +38,12 @@ controller 自动重建/纠正。 修复依赖后让正常 reconcile 自动重试。不要通过删除/重建 CR 规避 Conflict；新 UID 只会 使已有保留资源继续冲突。 +Instance 管理凭据在首次装配连接时读取并缓存在进程内；连接故障重试不会重新读取 Bao +或自动更换密码。Bao 暂时不可用不会影响已经装配的 PostgreSQL 连接，但会阻止新 Instance +或新凭据引用的装配。手工更换 Bao 同一路径下的管理密码后，需要重启 controller， +或将 Instance 改为引用新的凭据路径。修改 endpoint/凭据引用会关闭旧连接并重新装配； +只修改 allowlist 不会刷新凭据。 + ## Retain 后的资源 Retain 删除完成后，database、role、OpenBao record 和 registry 所有权记录仍存在但标记 diff --git a/internal/controller/postgresqlinstance_controller.go b/internal/controller/postgresqlinstance_controller.go index 7c4dd9f..d51f6b6 100644 --- a/internal/controller/postgresqlinstance_controller.go +++ b/internal/controller/postgresqlinstance_controller.go @@ -18,6 +18,7 @@ package controller import ( "context" + "fmt" "reflect" "time" @@ -33,15 +34,16 @@ import ( // PostgreSQLInstanceReconciler reconciles a PostgreSQLInstance object type PostgreSQLInstanceReconciler struct { client.Client - Scheme *runtime.Scheme - Initializer PostgreSQLInstanceInitializer - Timeout time.Duration + Scheme *runtime.Scheme + Instances PostgreSQLInstances + Timeout time.Duration } -// PostgreSQLInstanceInitializer is the external dependency boundary used by the Instance state machine. -type PostgreSQLInstanceInitializer interface { +// PostgreSQLInstances is the external dependency boundary used by the Instance state machine. +type PostgreSQLInstances interface { Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) InitializeRegistry(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) + Forget(string) } // +kubebuilder:rbac:groups=database.ddupan.top,resources=postgresqlinstances,verbs=get;list;watch;create;update;patch;delete @@ -61,16 +63,20 @@ func (r *PostgreSQLInstanceReconciler) Reconcile(ctx context.Context, req ctrl.R logger := logf.FromContext(ctx) instance := &databasev1alpha1.PostgreSQLInstance{} if err := r.Get(ctx, req.NamespacedName, instance); err != nil { + if apierrors.IsNotFound(err) { + r.Instances.Forget(req.Name) + } return ctrl.Result{}, client.IgnoreNotFound(err) } before := instance.DeepCopy() + operationCtx := ctx if r.Timeout > 0 { var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, r.Timeout) + operationCtx, cancel = context.WithTimeout(ctx, r.Timeout) defer cancel() } - result, reconcileErr := newInstanceStateMachine(r.Initializer).reconcile(ctx, instance) + result, reconcileErr := newInstanceStateMachine(r.Instances).reconcile(operationCtx, instance) if !reflect.DeepEqual(before.Status, instance.Status) { if err := r.Status().Patch(ctx, instance, client.MergeFrom(before)); err != nil { @@ -86,6 +92,9 @@ func (r *PostgreSQLInstanceReconciler) Reconcile(ctx context.Context, req ctrl.R // SetupWithManager sets up the controller with the Manager. func (r *PostgreSQLInstanceReconciler) SetupWithManager(mgr ctrl.Manager) error { + if r.Instances == nil { + return fmt.Errorf("instance service must be injected before controller setup") + } return ctrl.NewControllerManagedBy(mgr). For(&databasev1alpha1.PostgreSQLInstance{}). Named("postgresqlinstance"). diff --git a/internal/controller/postgresqlinstance_controller_test.go b/internal/controller/postgresqlinstance_controller_test.go index 515c7d6..572e734 100644 --- a/internal/controller/postgresqlinstance_controller_test.go +++ b/internal/controller/postgresqlinstance_controller_test.go @@ -77,8 +77,9 @@ var _ = Describe("PostgreSQLInstance Controller", func() { It("should successfully reconcile the resource", func() { By("Reconciling the created resource") controllerReconciler := &PostgreSQLInstanceReconciler{ - Client: k8sClient, - Scheme: k8sClient.Scheme(), + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Instances: fakeInstances{validateVersion: testPostgreSQLVersion, registryVersion: testPostgreSQLVersion}, } _, err := controllerReconciler.Reconcile(ctx, reconcile.Request{ @@ -95,6 +96,14 @@ var _ = Describe("PostgreSQLInstance Controller", func() { HaveField("ObservedGeneration", actual.Generation), ))) + // Advance to Ready using an injected fake, then prove an unchanged + // readiness observation does not patch status. + for range 2 { + _, err = controllerReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) + Expect(err).NotTo(HaveOccurred()) + } + Expect(k8sClient.Get(ctx, typeNamespacedName, actual)).To(Succeed()) + Expect(actual.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseReady)) resourceVersion := actual.ResourceVersion _, err = controllerReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName}) Expect(err).NotTo(HaveOccurred()) diff --git a/internal/controller/postgresqlinstance_state_machine.go b/internal/controller/postgresqlinstance_state_machine.go index 4e08995..9a75528 100644 --- a/internal/controller/postgresqlinstance_state_machine.go +++ b/internal/controller/postgresqlinstance_state_machine.go @@ -28,12 +28,12 @@ import ( type instancePhaseHandler func(context.Context, *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) type instanceStateMachine struct { - initializer PostgreSQLInstanceInitializer - handlers map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler + instances PostgreSQLInstances + handlers map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler } -func newInstanceStateMachine(initializer PostgreSQLInstanceInitializer) *instanceStateMachine { - m := &instanceStateMachine{initializer: initializer} +func newInstanceStateMachine(instances PostgreSQLInstances) *instanceStateMachine { + m := &instanceStateMachine{instances: instances} m.handlers = map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler{ databasev1alpha1.PostgreSQLInstancePhasePending: m.pending, databasev1alpha1.PostgreSQLInstancePhaseValidating: m.validate, @@ -45,6 +45,8 @@ func newInstanceStateMachine(initializer PostgreSQLInstanceInitializer) *instanc } func (m *instanceStateMachine) reconcile(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) { + // Dispatch exactly one handler from the persisted checkpoint. A handler performs + // its state action and returns the next checkpoint for the Reconciler to save. phase := instance.Status.Phase if !instance.DeletionTimestamp.IsZero() { phase = databasev1alpha1.PostgreSQLInstancePhaseDeleting @@ -69,15 +71,13 @@ func (m *instanceStateMachine) reconcile(ctx context.Context, instance *database } func (m *instanceStateMachine) pending(_ context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) { + // Persist intent before the next reconcile opens external connections. return advanceInstance(instance, databasev1alpha1.PostgreSQLInstancePhaseValidating, "instance dependencies are being validated"), nil } func (m *instanceStateMachine) validate(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) { - if m.initializer == nil { - return ctrl.Result{}, nil - } - version, err := m.initializer.Validate(ctx, instance) + version, err := m.instances.Validate(ctx, instance) if err != nil { return ctrl.Result{}, err } @@ -87,10 +87,7 @@ func (m *instanceStateMachine) validate(ctx context.Context, instance *databasev } func (m *instanceStateMachine) initializeRegistry(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) { - if m.initializer == nil { - return ctrl.Result{}, nil - } - version, err := m.initializer.InitializeRegistry(ctx, instance) + version, err := m.instances.InitializeRegistry(ctx, instance) if err != nil { return ctrl.Result{}, err } @@ -105,10 +102,7 @@ func (m *instanceStateMachine) ready(ctx context.Context, instance *databasev1al if instance.Status.ObservedGeneration != instance.Generation { return m.pending(ctx, instance) } - if m.initializer == nil { - return ctrl.Result{}, nil - } - version, err := m.initializer.Validate(ctx, instance) + version, err := m.instances.Validate(ctx, instance) if err != nil { instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseValidating return ctrl.Result{}, err @@ -118,6 +112,7 @@ func (m *instanceStateMachine) ready(ctx context.Context, instance *databasev1al } func (m *instanceStateMachine) deleting(_ context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) { + m.instances.Forget(instance.Name) instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseDeleting setReconcilingCondition(&instance.Status.Conditions, instance.Generation, "instance deletion is reconciling") return ctrl.Result{}, nil diff --git a/internal/controller/state_machine_test.go b/internal/controller/state_machine_test.go index 4b54995..f4d7368 100644 --- a/internal/controller/state_machine_test.go +++ b/internal/controller/state_machine_test.go @@ -28,7 +28,7 @@ import ( databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" ) -type fakeInstanceInitializer struct { +type fakeInstances struct { validateVersion string registryVersion string err error @@ -36,18 +36,20 @@ type fakeInstanceInitializer struct { const testPostgreSQLVersion = "17.6" -func (f fakeInstanceInitializer) Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) { +func (f fakeInstances) Forget(string) {} + +func (f fakeInstances) Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) { return f.validateVersion, f.err } -func (f fakeInstanceInitializer) InitializeRegistry(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) { +func (f fakeInstances) InitializeRegistry(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) { return f.registryVersion, f.err } var _ = Describe("phase handler state machines", func() { It("persists validation intent before touching unavailable dependencies", func() { instance := &databasev1alpha1.PostgreSQLInstance{} - machine := newInstanceStateMachine(fakeInstanceInitializer{err: errors.New("must not be called")}) + machine := newInstanceStateMachine(fakeInstances{err: errors.New("must not be called")}) result, err := machine.reconcile(context.Background(), instance) Expect(err).NotTo(HaveOccurred()) Expect(result.RequeueAfter).To(BeNumerically(">", 0)) @@ -62,7 +64,7 @@ var _ = Describe("phase handler state machines", func() { Phase: databasev1alpha1.PostgreSQLInstancePhaseReady, ObservedGeneration: 3, }, } - machine := newInstanceStateMachine(fakeInstanceInitializer{err: errors.New("must not be called")}) + machine := newInstanceStateMachine(fakeInstances{err: errors.New("must not be called")}) _, err := machine.reconcile(context.Background(), instance) Expect(err).NotTo(HaveOccurred()) Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating)) @@ -76,12 +78,12 @@ var _ = Describe("phase handler state machines", func() { Phase: databasev1alpha1.PostgreSQLInstancePhaseReady, ObservedGeneration: 3, }, } - machine := newInstanceStateMachine(fakeInstanceInitializer{err: errors.New("unavailable")}) + machine := newInstanceStateMachine(fakeInstances{err: errors.New("unavailable")}) _, err := machine.reconcile(context.Background(), instance) Expect(err).To(MatchError("unavailable")) Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating)) Expect(instance.Status.Conditions[0].Status).To(Equal(metav1.ConditionFalse)) - machine.initializer = fakeInstanceInitializer{ + machine.instances = fakeInstances{ validateVersion: testPostgreSQLVersion, registryVersion: testPostgreSQLVersion, } @@ -102,7 +104,7 @@ var _ = Describe("phase handler state machines", func() { ObjectMeta: metav1.ObjectMeta{Generation: 3}, Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating}, } - machine := newInstanceStateMachine(fakeInstanceInitializer{ + machine := newInstanceStateMachine(fakeInstances{ validateVersion: testPostgreSQLVersion, registryVersion: testPostgreSQLVersion, }) @@ -124,7 +126,7 @@ var _ = Describe("phase handler state machines", func() { ObjectMeta: metav1.ObjectMeta{Generation: 2}, Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating}, } - machine := newInstanceStateMachine(fakeInstanceInitializer{err: errors.New("unavailable")}) + machine := newInstanceStateMachine(fakeInstances{err: errors.New("unavailable")}) _, err := machine.reconcile(context.Background(), instance) Expect(err).To(MatchError("unavailable")) Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating)) @@ -135,7 +137,7 @@ var _ = Describe("phase handler state machines", func() { DescribeTable("dispatches Instance phases", func(instance *databasev1alpha1.PostgreSQLInstance, expected databasev1alpha1.PostgreSQLInstancePhase, hasMessage bool) { - _, err := newInstanceStateMachine(nil).reconcile(context.Background(), instance) + _, err := newInstanceStateMachine(fakeInstances{}).reconcile(context.Background(), instance) Expect(err).NotTo(HaveOccurred()) Expect(instance.Status.Phase).To(Equal(expected)) if hasMessage { @@ -149,12 +151,12 @@ var _ = Describe("phase handler state machines", func() { databasev1alpha1.PostgreSQLInstancePhaseValidating, true, ), - Entry("keeps an active phase until its handler can observe dependencies", + Entry("initializes the registry and reaches Ready", &databasev1alpha1.PostgreSQLInstance{Status: databasev1alpha1.PostgreSQLInstanceStatus{ Phase: databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry, }}, - databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry, - false, + databasev1alpha1.PostgreSQLInstancePhaseReady, + true, ), Entry("routes deletion independently of the previous phase", deletingInstance(), diff --git a/internal/instance/errors.go b/internal/instance/errors.go new file mode 100644 index 0000000..8a82d25 --- /dev/null +++ b/internal/instance/errors.go @@ -0,0 +1,27 @@ +/* +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 instance + +// Failure contains only a safe operation message and a Condition reason. +// Raw driver errors can contain credentials and must not escape through wrapping. +type Failure struct { + Reason string + Operation string +} + +func (e Failure) Error() string { return e.Operation } +func (e Failure) ConditionReason() string { return e.Reason } diff --git a/internal/instance/initializer.go b/internal/instance/initializer.go deleted file mode 100644 index 076a285..0000000 --- a/internal/instance/initializer.go +++ /dev/null @@ -1,216 +0,0 @@ -/* -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 instance validates PostgreSQLInstance dependencies and initializes -// the controller registry without exposing administrative credentials. -package instance - -import ( - "context" - "errors" - "net" - "net/url" - "path/filepath" - "strconv" - "strings" - "time" - - "github.com/jackc/pgx/v5/pgxpool" - kubernetesauth "github.com/openbao/openbao/api/auth/kubernetes/v2" - openbao "github.com/openbao/openbao/api/v2" - "k8s.io/apimachinery/pkg/util/validation" - - databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" - "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/postgresql/registry" -) - -// Config contains the deployment-level settings needed by Instance readiness. -type Config struct { - OpenBaoAddress string - OpenBaoConsumerAddress string - OpenBaoAuthMount string - OpenBaoAuthRole string - ServiceAccountTokenPath string - OpenBaoKVMount string - OpenBaoTenantBasePath string - ExternalSecretStoreName string - PostgreSQLCABundlePath string - Timeout time.Duration -} - -// Initializer implements the Instance validation and registry phases. -type Initializer struct{ config Config } - -type failure struct { - reason string - message string -} - -func (e failure) Error() string { return e.message } -func (e failure) ConditionReason() string { return e.reason } - -func dependencyFailure(message string) error { - return failure{reason: databasev1alpha1.ReasonDependencyUnavailable, message: message} -} - -func authenticationFailure(message string) error { - return failure{reason: databasev1alpha1.ReasonAuthenticationFailed, message: message} -} - -func privilegeFailure(message string) error { - return failure{reason: databasev1alpha1.ReasonInsufficientPrivileges, message: message} -} - -// New validates config and constructs an Initializer. -func New(config Config) (*Initializer, error) { - if config.OpenBaoAddress == "" || config.OpenBaoAuthRole == "" || config.OpenBaoAuthMount == "" || - config.ServiceAccountTokenPath == "" || config.OpenBaoKVMount == "" || config.OpenBaoTenantBasePath == "" || - config.ExternalSecretStoreName == "" || config.Timeout <= 0 { - return nil, errors.New("initialize instance dependencies: required configuration is missing") - } - if err := validateAddress(config.OpenBaoAddress); err != nil { - return nil, err - } - if config.OpenBaoConsumerAddress == "" { - config.OpenBaoConsumerAddress = config.OpenBaoAddress - } - if err := validateAddress(config.OpenBaoConsumerAddress); err != nil { - return nil, err - } - if !filepath.IsAbs(config.ServiceAccountTokenPath) || validateRelativePath(config.OpenBaoAuthMount, false) != nil || - validateRelativePath(config.OpenBaoKVMount, false) != nil || validateRelativePath(config.OpenBaoTenantBasePath, true) != nil || - len(validation.IsDNS1123Subdomain(config.ExternalSecretStoreName)) != 0 { - return nil, errors.New("initialize instance dependencies: invalid path or resource name configuration") - } - return &Initializer{config: config}, nil -} - -// Validate authenticates to OpenBao, reads the administrative credential, and -// verifies that PostgreSQL accepts it. It returns only public server metadata. -func (i *Initializer) Validate(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (string, error) { - pool, err := i.connect(ctx, instance) - if err != nil { - return "", err - } - defer pool.Close() - - var version string - if err := pool.QueryRow(ctx, "SHOW server_version").Scan(&version); err != nil { - return "", errors.New("validate PostgreSQL server metadata") - } - return version, nil -} - -// InitializeRegistry repeats dependency validation and applies registry migrations. -func (i *Initializer) InitializeRegistry(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (string, error) { - pool, err := i.connect(ctx, instance) - if err != nil { - return "", err - } - defer pool.Close() - - if err := registry.NewStore(pool).Bootstrap(ctx); err != nil { - return "", privilegeFailure("initialize PostgreSQL registry") - } - var version string - if err := pool.QueryRow(ctx, "SHOW server_version").Scan(&version); err != nil { - return "", errors.New("validate PostgreSQL server metadata") - } - return version, nil -} - -func (i *Initializer) connect(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (*pgxpool.Pool, error) { - if instance.Spec.Endpoint.SSLMode != databasev1alpha1.PostgreSQLSSLModeDisable && i.config.PostgreSQLCABundlePath == "" { - return nil, errors.New("configure PostgreSQL TLS: CA bundle path is required") - } - ctx, cancel := context.WithTimeout(ctx, i.config.Timeout) - defer cancel() - - clientConfig := openbao.DefaultConfig() - clientConfig.Address = i.config.OpenBaoAddress - clientConfig.Timeout = i.config.Timeout - clientConfig.DisableEnvironment = true - client, err := openbao.NewClient(clientConfig) - if err != nil { - return nil, dependencyFailure("create OpenBao client") - } - auth, err := kubernetesauth.NewKubernetesAuth( - i.config.OpenBaoAuthRole, - kubernetesauth.WithMountPath(i.config.OpenBaoAuthMount), - kubernetesauth.WithServiceAccountTokenPath(i.config.ServiceAccountTokenPath), - ) - if err != nil { - return nil, errors.New("configure OpenBao Kubernetes authentication") - } - if secret, err := client.Auth().Login(ctx, auth); err != nil || secret == nil || secret.Auth == nil { - return nil, authenticationFailure("authenticate to OpenBao") - } - - secret, err := client.KVv2(i.config.OpenBaoKVMount).Get(ctx, instance.Spec.AdminCredentialRef.Path) - if err != nil { - return nil, privilegeFailure("read PostgreSQL administrative credential") - } - username, usernameOK := secret.Data[instance.Spec.AdminCredentialRef.UsernameKey].(string) - password, passwordOK := secret.Data[instance.Spec.AdminCredentialRef.PasswordKey].(string) - if !usernameOK || !passwordOK || username == "" || password == "" { - return nil, errors.New("read PostgreSQL administrative credential fields") - } - - connectionURL := &url.URL{ - Scheme: "postgresql", - User: url.UserPassword(username, password), - Host: net.JoinHostPort(instance.Spec.Endpoint.Host, strconv.Itoa(int(instance.Spec.Endpoint.Port))), - Path: instance.Spec.Endpoint.Database, - } - query := connectionURL.Query() - query.Set("sslmode", string(instance.Spec.Endpoint.SSLMode)) - if instance.Spec.Endpoint.SSLMode != databasev1alpha1.PostgreSQLSSLModeDisable { - query.Set("sslrootcert", i.config.PostgreSQLCABundlePath) - } - connectionURL.RawQuery = query.Encode() - poolConfig, err := pgxpool.ParseConfig(connectionURL.String()) - if err != nil { - return nil, errors.New("configure PostgreSQL connection") - } - pool, err := pgxpool.NewWithConfig(ctx, poolConfig) - if err != nil { - return nil, errors.New("connect to PostgreSQL") - } - if err := pool.Ping(ctx); err != nil { - pool.Close() - return nil, authenticationFailure("authenticate to PostgreSQL") - } - return pool, nil -} - -func validateAddress(value string) error { - parsed, err := url.Parse(value) - if err != nil || !parsed.IsAbs() || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" || - parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" { - return errors.New("initialize instance dependencies: invalid OpenBao address") - } - return nil -} - -func validateRelativePath(value string, rejectAPILayer bool) error { - for index, segment := range strings.Split(value, "/") { - if segment == "" || segment == "." || segment == ".." || - (rejectAPILayer && index == 0 && (segment == "data" || segment == "metadata")) { - return errors.New("invalid mount-relative path") - } - } - return nil -} diff --git a/internal/instance/initializer_test.go b/internal/instance/initializer_test.go deleted file mode 100644 index e09d6a6..0000000 --- a/internal/instance/initializer_test.go +++ /dev/null @@ -1,44 +0,0 @@ -/* -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 instance - -import ( - "testing" - "time" -) - -func TestNewValidatesDeploymentConfiguration(t *testing.T) { - valid := Config{ - OpenBaoAddress: "http://openbao:8200", OpenBaoAuthMount: "kubernetes", OpenBaoAuthRole: "controller", - ServiceAccountTokenPath: "/var/run/secrets/token", OpenBaoKVMount: "secret", - OpenBaoTenantBasePath: "postgresql-tenants", ExternalSecretStoreName: "openbao", Timeout: 30 * time.Second, - } - if _, err := New(valid); err != nil { - t.Fatalf("New(valid) error = %v", err) - } - - invalid := valid - invalid.OpenBaoTenantBasePath = "metadata/tenants" - if _, err := New(invalid); err == nil { - t.Fatal("New() accepted a KV v2 API-layer base path") - } - invalid = valid - invalid.OpenBaoAddress = "http://user:password@openbao:8200?token=secret" - if _, err := New(invalid); err == nil { - t.Fatal("New() accepted credentials in the OpenBao address") - } -} diff --git a/internal/instance/service.go b/internal/instance/service.go new file mode 100644 index 0000000..ba435a7 --- /dev/null +++ b/internal/instance/service.go @@ -0,0 +1,152 @@ +/* +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 instance manages the lifetime and capabilities of external PostgreSQL instances. +package instance + +import ( + "context" + "errors" + "sync" + + api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" + "k8s.io/apimachinery/pkg/types" +) + +// Credentials are supplied by the credential source, never by the database adapter. +type Credentials struct{ Username, Password string } + +func (Credentials) String() string { return "[redacted credentials]" } +func (Credentials) GoString() string { return "[redacted credentials]" } + +// CredentialSource resolves administrative credentials when an Instance is assembled. +type CredentialSource interface { + Read(context.Context, api.OpenBaoSecretReference) (Credentials, error) +} + +// Database exposes the capabilities of an assembled Instance. +type Database interface { + Version(context.Context) (string, error) + EnsureRegistry(context.Context) error + Close() +} + +// Connector creates a database without knowing where credentials came from. +type Connector interface { + Connect(context.Context, api.PostgreSQLEndpoint, Credentials) (Database, error) +} + +type identity struct { + uid types.UID + endpoint api.PostgreSQLEndpoint + reference api.OpenBaoSecretReference +} + +type entry struct { + identity identity + credentials *Credentials + database Database +} + +// Service owns Instance connections. Reconciles reuse credentials and pools until +// the UID, endpoint or credential reference changes; allowlist changes do not rotate them. +// Operations are serialized to prevent Close racing with an active database operation. +type Service struct { + mu sync.Mutex + source CredentialSource + connector Connector + entries map[string]*entry + closed bool +} + +func NewService(source CredentialSource, connector Connector) *Service { + return &Service{source: source, connector: connector, entries: make(map[string]*entry)} +} + +func (s *Service) Validate(ctx context.Context, resource *api.PostgreSQLInstance) (string, error) { + return s.observe(ctx, resource, false) +} + +func (s *Service) InitializeRegistry(ctx context.Context, resource *api.PostgreSQLInstance) (string, error) { + return s.observe(ctx, resource, true) +} + +func (s *Service) observe(ctx context.Context, resource *api.PostgreSQLInstance, ensureRegistry bool) (string, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return "", errors.New("instance service is closed") + } + db, err := s.database(ctx, resource) + if err != nil { + return "", err + } + if ensureRegistry { + if err := db.EnsureRegistry(ctx); err != nil { + return "", err + } + } + return db.Version(ctx) +} + +func (s *Service) database(ctx context.Context, resource *api.PostgreSQLInstance) (Database, error) { + key := identity{uid: resource.UID, endpoint: resource.Spec.Endpoint, reference: resource.Spec.AdminCredentialRef} + current := s.entries[resource.Name] + if current == nil || current.identity != key { + s.release(resource.Name) + current = &entry{identity: key} + s.entries[resource.Name] = current + } + if current.credentials == nil { + credentials, err := s.source.Read(ctx, key.reference) + if err != nil { + return nil, err + } + current.credentials = &credentials + } + if current.database == nil { + database, err := s.connector.Connect(ctx, key.endpoint, *current.credentials) + if err != nil { + return nil, err + } + current.database = database + } + return current.database, nil +} + +// Forget releases local resources only. It never deletes the external registry. +func (s *Service) Forget(name string) { + s.mu.Lock() + defer s.mu.Unlock() + s.release(name) +} + +func (s *Service) release(name string) { + if current := s.entries[name]; current != nil && current.database != nil { + current.database.Close() + } + delete(s.entries, name) +} + +// Close is called after manager workers stop, and is safe to repeat. +func (s *Service) Close() { + s.mu.Lock() + defer s.mu.Unlock() + s.closed = true + for name := range s.entries { + s.release(name) + } +} diff --git a/internal/instance/service_test.go b/internal/instance/service_test.go new file mode 100644 index 0000000..022e10c --- /dev/null +++ b/internal/instance/service_test.go @@ -0,0 +1,156 @@ +/* +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 instance + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + + api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +type sourceStub struct { + reads int + err error +} + +func (s *sourceStub) Read(context.Context, api.OpenBaoSecretReference) (Credentials, error) { + s.reads++ + return Credentials{Username: "admin", Password: "canary-secret"}, s.err +} + +type databaseStub struct { + closes, migrations int + err error +} + +func (d *databaseStub) Version(context.Context) (string, error) { return "17", d.err } +func (d *databaseStub) EnsureRegistry(context.Context) error { d.migrations++; return d.err } +func (d *databaseStub) Close() { d.closes++ } + +type connectorStub struct { + databases []*databaseStub + err error +} + +func (c *connectorStub) Connect(context.Context, api.PostgreSQLEndpoint, Credentials) (Database, error) { + if c.err != nil { + return nil, c.err + } + database := &databaseStub{} + c.databases = append(c.databases, database) + return database, nil +} + +func TestInstanceConnectionLifetime(t *testing.T) { + source, connector := &sourceStub{}, &connectorStub{} + service := NewService(source, connector) + defer service.Close() + resource := &api.PostgreSQLInstance{ObjectMeta: metav1.ObjectMeta{Name: "shared", UID: "uid-1"}} + ctx := context.Background() + check := func() { + t.Helper() + if _, err := service.Validate(ctx, resource); err != nil { + t.Fatalf("validate: %v", err) + } + } + check() + if _, err := service.InitializeRegistry(ctx, resource); err != nil { + t.Fatal(err) + } + resource.Generation++ + resource.Spec.AllowedExtensions = []string{"pg_trgm"} + check() + if source.reads != 1 || len(connector.databases) != 1 || connector.databases[0].migrations != 1 { + t.Fatal("reconciliation or allowlist update unnecessarily reassembled dependencies") + } + connector.databases[0].err = errors.New("network outage") + if _, err := service.Validate(ctx, resource); err == nil { + t.Fatal("outage went unnoticed") + } + connector.databases[0].err = nil + check() + if source.reads != 1 { + t.Fatal("network recovery reread the password") + } + + changes := []func(){ + func() { resource.Spec.Endpoint.Host = "new.example" }, + func() { resource.Spec.AdminCredentialRef.Path = "new/admin" }, + func() { resource.UID = "uid-2" }, + } + for index, change := range changes { + change() + check() + if connector.databases[index].closes != 1 { + t.Fatal("replaced pool was not closed") + } + } + if source.reads != 4 { + t.Fatal("changed connection identity did not reload credentials") + } + service.Forget(resource.Name) + service.Forget(resource.Name) + if connector.databases[3].closes != 1 { + t.Fatal("deletion did not close pool exactly once") + } + check() + service.Close() + service.Close() + if connector.databases[4].closes != 1 { + t.Fatal("shutdown did not close pool exactly once") + } + if _, err := service.Validate(ctx, resource); err == nil { + t.Fatal("closed service accepted work") + } +} + +func TestAssemblyRetriesWithoutPasswordRotation(t *testing.T) { + source := &sourceStub{err: errors.New("bao unavailable")} + connector := &connectorStub{err: errors.New("connection unavailable")} + service := NewService(source, connector) + defer service.Close() + resource := &api.PostgreSQLInstance{ObjectMeta: metav1.ObjectMeta{Name: "shared"}} + ctx := context.Background() + if _, err := service.Validate(ctx, resource); err == nil { + t.Fatal("expected source failure") + } + source.err = nil + if _, err := service.Validate(ctx, resource); err == nil { + t.Fatal("expected connection failure") + } + connector.err = nil + if _, err := service.Validate(ctx, resource); err != nil { + t.Fatal(err) + } + if source.reads != 2 { + t.Fatal("retry reread successfully cached credentials") + } +} + +func TestCredentialsFormattingIsRedacted(t *testing.T) { + value := Credentials{Username: "canary-user", Password: "canary-secret"} + for _, format := range []string{"%v", "%+v", "%#v"} { + if strings.Contains(fmt.Sprintf(format, value), "canary") { + t.Fatal("credential formatting leaked data") + } + } +} diff --git a/internal/openbao/credentials.go b/internal/openbao/credentials.go new file mode 100644 index 0000000..688f006 --- /dev/null +++ b/internal/openbao/credentials.go @@ -0,0 +1,133 @@ +/* +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 openbao supplies credentials using one shared client and Kubernetes identity. +package openbao + +import ( + "context" + "errors" + "sync" + "time" + + api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" + "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/instance" + kubernetesauth "github.com/openbao/openbao/api/auth/kubernetes/v2" + bao "github.com/openbao/openbao/api/v2" +) + +// Credentials owns authentication; database connections never depend on this client. +type Credentials struct { + client *bao.Client + mount string + authenticate func(context.Context) (*bao.Secret, error) + mu sync.Mutex + expires time.Time +} + +func NewCredentials(client *bao.Client, mount, authMount, role string) *Credentials { + source := &Credentials{client: client, mount: mount} + source.authenticate = func(ctx context.Context) (*bao.Secret, error) { + // Constructing auth here rereads the projected JWT after Kubernetes rotates it. + auth, err := kubernetesauth.NewKubernetesAuth(role, kubernetesauth.WithMountPath(authMount)) + if err != nil { + return nil, instance.Failure{Reason: api.ReasonAuthenticationFailed, Operation: "read Kubernetes identity"} + } + return client.Auth().Login(ctx, auth) + } + return source +} + +func (s *Credentials) login(ctx context.Context) error { + secret, err := s.authenticate(ctx) + if err != nil { + return classify("authenticate to OpenBao", err, true) + } + if secret == nil || secret.Auth == nil || secret.Auth.ClientToken == "" || secret.Auth.LeaseDuration <= 0 { + return instance.Failure{Reason: api.ReasonAuthenticationFailed, Operation: "OpenBao returned no leased identity"} + } + s.client.SetToken(secret.Auth.ClientToken) + ttl := time.Duration(secret.Auth.LeaseDuration) * time.Second + s.expires = time.Now().Add(ttl - ttl/10) + return nil +} + +func (s *Credentials) Read(ctx context.Context, reference api.OpenBaoSecretReference) (instance.Credentials, error) { + // Login and reads share a lock so concurrent first-use requests cannot race token replacement. + s.mu.Lock() + defer s.mu.Unlock() + if !time.Now().Before(s.expires) { + if err := s.login(ctx); err != nil { + return instance.Credentials{}, err + } + } + secret, err := s.client.KVv2(s.mount).Get(ctx, reference.Path) + var response *bao.ResponseError + if errors.As(err, &response) && response.StatusCode == 403 { + // A token may be revoked before its TTL. Reauthenticate once, never loop on denied policies. + s.expires = time.Time{} + if err := s.login(ctx); err != nil { + return instance.Credentials{}, err + } + secret, err = s.client.KVv2(s.mount).Get(ctx, reference.Path) + } + if err != nil { + return instance.Credentials{}, classify("read administrative credential", err, false) + } + usernameKey, passwordKey := reference.UsernameKey, reference.PasswordKey + if usernameKey == "" { + usernameKey = "username" + } + if passwordKey == "" { + passwordKey = "password" + } + if secret == nil { + return instance.Credentials{}, invalidCredential() + } + username, usernameOK := secret.Data[usernameKey].(string) + password, passwordOK := secret.Data[passwordKey].(string) + if !usernameOK || !passwordOK || username == "" || password == "" { + return instance.Credentials{}, invalidCredential() + } + return instance.Credentials{Username: username, Password: password}, nil +} + +func invalidCredential() error { + return instance.Failure{Reason: api.ReasonAuthenticationFailed, Operation: "administrative credential fields are missing"} +} + +func classify(operation string, err error, login bool) error { + reason := api.ReasonDependencyUnavailable + var safe instance.Failure + if errors.As(err, &safe) { + return safe + } + var response *bao.ResponseError + if errors.As(err, &response) { + switch response.StatusCode { + case 403: + reason = api.ReasonInsufficientPrivileges + if login { + reason = api.ReasonAuthenticationFailed + } + case 400, 401: + reason = api.ReasonAuthenticationFailed + case 404: + reason = api.ReasonInvalidSpec + } + } + return instance.Failure{Reason: reason, Operation: operation} +} diff --git a/internal/openbao/credentials_test.go b/internal/openbao/credentials_test.go new file mode 100644 index 0000000..62fba9a --- /dev/null +++ b/internal/openbao/credentials_test.go @@ -0,0 +1,125 @@ +/* +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 openbao + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" + bao "github.com/openbao/openbao/api/v2" +) + +const testToken = "test-token" +const testAdmin = "admin" + +func TestSharedIdentityReauthenticatesOnExpiryAndRevocation(t *testing.T) { + var logins atomic.Int32 + var revoked atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/secret/data/admin" { + http.NotFound(w, r) + return + } + if revoked.Swap(false) { + w.WriteHeader(http.StatusForbidden) + return + } + if r.Header.Get("X-Vault-Token") != testToken { + w.WriteHeader(http.StatusForbidden) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "data": map[string]any{ + "data": map[string]any{"username": testAdmin, "password": "test-password"}, + "metadata": map[string]any{"version": 1}, + }, + }) + })) + defer server.Close() + config := bao.NewConfig() + config.Address = server.URL + config.MaxRetries = 0 + client, err := bao.NewClient(config) + if err != nil { + t.Fatal(err) + } + source := NewCredentials(client, "secret", "kubernetes", "controller") + source.authenticate = func(context.Context) (*bao.Secret, error) { + logins.Add(1) + return &bao.Secret{Auth: &bao.SecretAuth{ClientToken: testToken, LeaseDuration: 300}}, nil + } + reference := api.OpenBaoSecretReference{Path: testAdmin} + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { + if _, err := source.Read(context.Background(), reference); err != nil { + t.Error(err) + } + }) + } + wg.Wait() + if logins.Load() != 1 { + t.Fatal("concurrent reads repeatedly authenticated") + } + source.expires = time.Time{} + if _, err := source.Read(context.Background(), reference); err != nil { + t.Fatal(err) + } + if logins.Load() != 2 { + t.Fatal("expired identity was not refreshed") + } + revoked.Store(true) + if _, err := source.Read(context.Background(), reference); err != nil { + t.Fatal(err) + } + if logins.Load() != 3 { + t.Fatal("revoked identity was not refreshed") + } +} + +func TestDeniedPolicyRetriesAuthenticationOnlyOnce(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusForbidden) + })) + defer server.Close() + config := bao.NewConfig() + config.Address = server.URL + config.MaxRetries = 0 + client, err := bao.NewClient(config) + if err != nil { + t.Fatal(err) + } + source := NewCredentials(client, "secret", "kubernetes", "controller") + logins := 0 + source.authenticate = func(context.Context) (*bao.Secret, error) { + logins++ + return &bao.Secret{Auth: &bao.SecretAuth{ClientToken: testToken, LeaseDuration: 300}}, nil + } + if _, err := source.Read(context.Background(), api.OpenBaoSecretReference{Path: testAdmin}); err == nil { + t.Fatal("denied policy was accepted") + } + if logins != 2 { + t.Fatal("authentication did not stop after one retry") + } +} diff --git a/internal/postgresql/connection.go b/internal/postgresql/connection.go new file mode 100644 index 0000000..69c7bca --- /dev/null +++ b/internal/postgresql/connection.go @@ -0,0 +1,105 @@ +/* +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 postgresql + +import ( + "context" + "crypto/x509" + "errors" + "net" + "net/url" + "strconv" + + api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" + "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/instance" + "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/postgresql/registry" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +// Connector is configured once with the deployment trust bundle. +type Connector struct{ CABundlePath string } + +func (c Connector) Connect(ctx context.Context, endpoint api.PostgreSQLEndpoint, credentials instance.Credentials) (instance.Database, error) { + if endpoint.SSLMode != api.PostgreSQLSSLModeDisable && c.CABundlePath == "" { + return nil, instance.Failure{Reason: api.ReasonInvalidSpec, Operation: "PostgreSQL TLS requires a CA bundle"} + } + address := &url.URL{ + Scheme: "postgresql", + Host: net.JoinHostPort(endpoint.Host, strconv.Itoa(int(endpoint.Port))), + Path: endpoint.Database, + } + query := address.Query() + query.Set("sslmode", string(endpoint.SSLMode)) + if endpoint.SSLMode != api.PostgreSQLSSLModeDisable { + query.Set("sslrootcert", c.CABundlePath) + } + address.RawQuery = query.Encode() + config, err := pgxpool.ParseConfig(address.String()) + if err != nil { + return nil, classify("configure PostgreSQL connection", err) + } + // Avoid embedding credentials in the driver's original connection string. + config.ConnConfig.User = credentials.Username + config.ConnConfig.Password = credentials.Password + config.MaxConns = 2 + pool, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + return nil, classify("create PostgreSQL pool", err) + } + // Pool construction is lazy. Version performs the first authenticated query. + return &Database{pool: pool}, nil +} + +type Database struct{ pool *pgxpool.Pool } + +func (d *Database) Version(ctx context.Context) (string, error) { + var version string + if err := d.pool.QueryRow(ctx, "SHOW server_version").Scan(&version); err != nil { + return "", classify("read PostgreSQL server version", err) + } + return version, nil +} + +func (d *Database) EnsureRegistry(ctx context.Context) error { + if err := registry.NewStore(d.pool).Bootstrap(ctx); err != nil { + return classify("initialize PostgreSQL registry", err) + } + return nil +} + +func (d *Database) Close() { d.pool.Close() } + +func classify(operation string, err error) error { + reason := api.ReasonDependencyUnavailable + var pgError *pgconn.PgError + var unknownCA x509.UnknownAuthorityError + var hostname x509.HostnameError + var invalidCertificate x509.CertificateInvalidError + switch { + case errors.As(err, &pgError): + switch pgError.Code { + case "28P01", "28000": + reason = api.ReasonAuthenticationFailed + case "42501": + reason = api.ReasonInsufficientPrivileges + } + case errors.As(err, &unknownCA), errors.As(err, &hostname), errors.As(err, &invalidCertificate): + reason = api.ReasonAuthenticationFailed + } + return instance.Failure{Reason: reason, Operation: operation} +} diff --git a/internal/postgresql/connection_test.go b/internal/postgresql/connection_test.go new file mode 100644 index 0000000..df510ef --- /dev/null +++ b/internal/postgresql/connection_test.go @@ -0,0 +1,53 @@ +/* +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 postgresql + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + + api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" + "git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/instance" + "github.com/jackc/pgx/v5/pgconn" +) + +const testCanary = "canary-secret" + +func TestDatabaseFailuresDoNotLeakDriverMessages(t *testing.T) { + tests := []struct { + err error + reason string + }{ + {context.DeadlineExceeded, api.ReasonDependencyUnavailable}, + {&pgconn.PgError{Code: "28P01", Message: testCanary}, api.ReasonAuthenticationFailed}, + {&pgconn.PgError{Code: "42501", Message: testCanary}, api.ReasonInsufficientPrivileges}, + {&pgconn.PgError{Code: "53300", Message: testCanary}, api.ReasonDependencyUnavailable}, + } + for _, tt := range tests { + err := classify("database operation", fmt.Errorf("driver canary-secret: %w", tt.err)) + var failure instance.Failure + if !errors.As(err, &failure) || failure.Reason != tt.reason { + t.Fatal("wrong error category") + } + if strings.Contains(fmt.Sprintf("%+v %#v", err, err), "canary") { + t.Fatal("driver message escaped") + } + } +}