Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0c8edaa8de
|
||
|
|
61985acb32
|
||
|
|
305e778203
|
||
|
|
26d4528357
|
+1
-1
@@ -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
|
# 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,
|
# 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.
|
# 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
|
# Use distroless as minimal base image to package the manager binary
|
||||||
# Refer to https://github.com/GoogleContainerTools/distroless for more details
|
# Refer to https://github.com/GoogleContainerTools/distroless for more details
|
||||||
|
|||||||
@@ -132,11 +132,11 @@ lint-config: golangci-lint ## Verify golangci-lint linter configuration
|
|||||||
|
|
||||||
.PHONY: build
|
.PHONY: build
|
||||||
build: manifests generate fmt vet ## Build manager binary.
|
build: manifests generate fmt vet ## Build manager binary.
|
||||||
go build -o bin/manager cmd/main.go
|
go build -o bin/manager ./cmd
|
||||||
|
|
||||||
.PHONY: run
|
.PHONY: run
|
||||||
run: manifests generate fmt vet ## Run a controller from your host.
|
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.
|
# 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.
|
# (i.e. docker build --platform linux/arm64). However, you must enable docker buildKit for it.
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
+15
-28
@@ -20,7 +20,6 @@ import (
|
|||||||
"crypto/tls"
|
"crypto/tls"
|
||||||
"flag"
|
"flag"
|
||||||
"os"
|
"os"
|
||||||
"time"
|
|
||||||
|
|
||||||
// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
|
// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
|
||||||
// to ensure that exec-entrypoint and run can make use of them.
|
// 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"
|
databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
|
||||||
"git.ddupan.top/panxiao81/postgresql-tenant-operator/internal/controller"
|
"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
|
// +kubebuilder:scaffold:imports
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -63,10 +63,7 @@ func main() {
|
|||||||
var probeAddr string
|
var probeAddr string
|
||||||
var secureMetrics bool
|
var secureMetrics bool
|
||||||
var enableHTTP2 bool
|
var enableHTTP2 bool
|
||||||
var openBaoAddress, openBaoConsumerAddress, openBaoAuthMount, openBaoAuthRole string
|
var dependencies dependencyOptions
|
||||||
var openBaoKVMount, openBaoTenantBasePath, openBaoServiceAccountTokenPath string
|
|
||||||
var externalSecretStoreName, postgreSQLCABundlePath string
|
|
||||||
var reconcileTimeout time.Duration
|
|
||||||
var tlsOpts []func(*tls.Config)
|
var tlsOpts []func(*tls.Config)
|
||||||
flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
|
flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
|
||||||
"Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.")
|
"Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.")
|
||||||
@@ -85,20 +82,8 @@ func main() {
|
|||||||
flag.StringVar(&metricsCertKey, "metrics-cert-key", "tls.key", "The name of the metrics server key file.")
|
flag.StringVar(&metricsCertKey, "metrics-cert-key", "tls.key", "The name of the metrics server key file.")
|
||||||
flag.BoolVar(&enableHTTP2, "enable-http2", false,
|
flag.BoolVar(&enableHTTP2, "enable-http2", false,
|
||||||
"If set, HTTP/2 will be enabled for the metrics and webhook servers")
|
"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.")
|
dependencies.bindFlags(flag.CommandLine)
|
||||||
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.")
|
|
||||||
opts := zap.Options{
|
opts := zap.Options{
|
||||||
Development: true,
|
Development: true,
|
||||||
}
|
}
|
||||||
@@ -107,17 +92,15 @@ func main() {
|
|||||||
|
|
||||||
ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
|
ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
|
||||||
|
|
||||||
instanceInitializer, err := instanceinitializer.New(instanceinitializer.Config{
|
credentials, err := dependencies.credentials()
|
||||||
OpenBaoAddress: openBaoAddress, OpenBaoConsumerAddress: openBaoConsumerAddress,
|
|
||||||
OpenBaoAuthMount: openBaoAuthMount, OpenBaoAuthRole: openBaoAuthRole,
|
|
||||||
ServiceAccountTokenPath: openBaoServiceAccountTokenPath, OpenBaoKVMount: openBaoKVMount,
|
|
||||||
OpenBaoTenantBasePath: openBaoTenantBasePath, ExternalSecretStoreName: externalSecretStoreName,
|
|
||||||
PostgreSQLCABundlePath: postgreSQLCABundlePath, Timeout: reconcileTimeout,
|
|
||||||
})
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
setupLog.Error(err, "Invalid controller dependency configuration")
|
setupLog.Error(err, "Invalid controller dependency configuration")
|
||||||
os.Exit(1)
|
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
|
// if the enable-http2 flag is false (the default), http/2 should be disabled
|
||||||
// due to its vulnerabilities. More specifically, disabling http/2 will
|
// due to its vulnerabilities. More specifically, disabling http/2 will
|
||||||
@@ -211,7 +194,10 @@ func main() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if err := (&controller.PostgreSQLInstanceReconciler{
|
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 {
|
}).SetupWithManager(mgr); err != nil {
|
||||||
setupLog.Error(err, "Failed to create controller", "controller", "postgresqlinstance")
|
setupLog.Error(err, "Failed to create controller", "controller", "postgresqlinstance")
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
@@ -237,6 +223,7 @@ func main() {
|
|||||||
setupLog.Info("Starting manager")
|
setupLog.Info("Starting manager")
|
||||||
if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
|
if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
|
||||||
setupLog.Error(err, "Failed to run manager")
|
setupLog.Error(err, "Failed to run manager")
|
||||||
|
instances.Close()
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -63,9 +63,6 @@ spec:
|
|||||||
args:
|
args:
|
||||||
- --leader-elect
|
- --leader-elect
|
||||||
- --health-probe-bind-address=:8081
|
- --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
|
image: controller:latest
|
||||||
name: manager
|
name: manager
|
||||||
ports:
|
ports:
|
||||||
|
|||||||
+20
-8
@@ -25,8 +25,8 @@
|
|||||||
|
|
||||||
## Controller 配置合同
|
## Controller 配置合同
|
||||||
|
|
||||||
controller 使用以下 CLI flags。必填项缺失、路径无效或 duration 不为正数时,进程必须
|
controller 使用以下 CLI flags。启动入口加载配置、检查本切片必需项并创建共享依赖,
|
||||||
在启动 manager 前失败;不得等到 reconcile 时才逐个资源报告配置错误。
|
配置错误在启动 manager 前失败。外部服务暂时不可用由 reconcile 重试,不阻止进程启动。
|
||||||
|
|
||||||
| CLI flag | 必填/默认 | 说明 |
|
| CLI flag | 必填/默认 | 说明 |
|
||||||
| --- | --- | --- |
|
| --- | --- | --- |
|
||||||
@@ -35,16 +35,28 @@ controller 使用以下 CLI flags。必填项缺失、路径无效或 duration
|
|||||||
| `--openbao-auth-mount` | `kubernetes` | Kubernetes auth mount 名称 |
|
| `--openbao-auth-mount` | `kubernetes` | Kubernetes auth mount 名称 |
|
||||||
| `--openbao-auth-role` | 必填 | controller ServiceAccount 对应 role |
|
| `--openbao-auth-role` | 必填 | controller ServiceAccount 对应 role |
|
||||||
| `--openbao-kv-mount` | `kv` | KV v2 mount;开发可显式用 `secret` |
|
| `--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 前缀 |
|
| `--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,不含私钥 |
|
| `--postgresql-ca-bundle-path` | PostgreSQL TLS 模式必填 | 只读 PEM trust bundle,不含私钥 |
|
||||||
| `--reconcile-timeout` | `30s` | 单轮 reconcile 中外部操作的总期限,必须大于零 |
|
| `--reconcile-timeout` | `30s` | 单轮 reconcile 中外部操作的总期限,必须大于零 |
|
||||||
|
|
||||||
address 必须是绝对 `http` 或 `https` URL,不允许 userinfo、query 或 fragment,末尾 `/`
|
连接地址由 OpenBao SDK 解析,移除末尾 `/`;配置错误不得携带原始地址中的认证信息。
|
||||||
在规范化后移除。mount、auth mount 和 base path 都使用 mount-relative path 语义,不以
|
生产环境使用 HTTPS;HTTP 只用于开发 fixture。consumer URL 的认证信息限制和 Tenant
|
||||||
`/` 开头,不含空段、`.` 或 `..`;base path 还不得编码 KV v2 的 `data`/`metadata`
|
派生路径限制由对应输出边界负责,Instance 服务不校验尚未使用的 Tenant 配置。
|
||||||
API 层。生产环境的 `--openbao-address` 必须使用 HTTPS;HTTP 只用于明确的开发 fixture。
|
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 路径固定推导为 `<base-path>/<namespace>/<metadata.name>`。namespace/name 都已通过
|
Tenant 路径固定推导为 `<base-path>/<namespace>/<metadata.name>`。namespace/name 都已通过
|
||||||
Kubernetes 名称校验,因此不再允许 CR 提供任意路径。KV v2 API URL 使用 consumer
|
Kubernetes 名称校验,因此不再允许 CR 提供任意路径。KV v2 API URL 使用 consumer
|
||||||
|
|||||||
@@ -38,6 +38,12 @@ controller 自动重建/纠正。
|
|||||||
修复依赖后让正常 reconcile 自动重试。不要通过删除/重建 CR 规避 Conflict;新 UID 只会
|
修复依赖后让正常 reconcile 自动重试。不要通过删除/重建 CR 规避 Conflict;新 UID 只会
|
||||||
使已有保留资源继续冲突。
|
使已有保留资源继续冲突。
|
||||||
|
|
||||||
|
Instance 管理凭据在首次装配连接时读取并缓存在进程内;连接故障重试不会重新读取 Bao
|
||||||
|
或自动更换密码。Bao 暂时不可用不会影响已经装配的 PostgreSQL 连接,但会阻止新 Instance
|
||||||
|
或新凭据引用的装配。手工更换 Bao 同一路径下的管理密码后,需要重启 controller,
|
||||||
|
或将 Instance 改为引用新的凭据路径。修改 endpoint/凭据引用会关闭旧连接并重新装配;
|
||||||
|
只修改 allowlist 不会刷新凭据。
|
||||||
|
|
||||||
## Retain 后的资源
|
## Retain 后的资源
|
||||||
|
|
||||||
Retain 删除完成后,database、role、OpenBao record 和 registry 所有权记录仍存在但标记
|
Retain 删除完成后,database、role、OpenBao record 和 registry 所有权记录仍存在但标记
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ package controller
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
"reflect"
|
"reflect"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -34,14 +35,15 @@ import (
|
|||||||
type PostgreSQLInstanceReconciler struct {
|
type PostgreSQLInstanceReconciler struct {
|
||||||
client.Client
|
client.Client
|
||||||
Scheme *runtime.Scheme
|
Scheme *runtime.Scheme
|
||||||
Initializer PostgreSQLInstanceInitializer
|
Instances PostgreSQLInstances
|
||||||
Timeout time.Duration
|
Timeout time.Duration
|
||||||
}
|
}
|
||||||
|
|
||||||
// PostgreSQLInstanceInitializer is the external dependency boundary used by the Instance state machine.
|
// PostgreSQLInstances is the external dependency boundary used by the Instance state machine.
|
||||||
type PostgreSQLInstanceInitializer interface {
|
type PostgreSQLInstances interface {
|
||||||
Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error)
|
Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error)
|
||||||
InitializeRegistry(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
|
// +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)
|
logger := logf.FromContext(ctx)
|
||||||
instance := &databasev1alpha1.PostgreSQLInstance{}
|
instance := &databasev1alpha1.PostgreSQLInstance{}
|
||||||
if err := r.Get(ctx, req.NamespacedName, instance); err != nil {
|
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)
|
return ctrl.Result{}, client.IgnoreNotFound(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
before := instance.DeepCopy()
|
before := instance.DeepCopy()
|
||||||
|
operationCtx := ctx
|
||||||
if r.Timeout > 0 {
|
if r.Timeout > 0 {
|
||||||
var cancel context.CancelFunc
|
var cancel context.CancelFunc
|
||||||
ctx, cancel = context.WithTimeout(ctx, r.Timeout)
|
operationCtx, cancel = context.WithTimeout(ctx, r.Timeout)
|
||||||
defer cancel()
|
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 !reflect.DeepEqual(before.Status, instance.Status) {
|
||||||
if err := r.Status().Patch(ctx, instance, client.MergeFrom(before)); err != nil {
|
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.
|
// SetupWithManager sets up the controller with the Manager.
|
||||||
func (r *PostgreSQLInstanceReconciler) SetupWithManager(mgr ctrl.Manager) error {
|
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).
|
return ctrl.NewControllerManagedBy(mgr).
|
||||||
For(&databasev1alpha1.PostgreSQLInstance{}).
|
For(&databasev1alpha1.PostgreSQLInstance{}).
|
||||||
Named("postgresqlinstance").
|
Named("postgresqlinstance").
|
||||||
|
|||||||
@@ -79,6 +79,7 @@ var _ = Describe("PostgreSQLInstance Controller", func() {
|
|||||||
controllerReconciler := &PostgreSQLInstanceReconciler{
|
controllerReconciler := &PostgreSQLInstanceReconciler{
|
||||||
Client: k8sClient,
|
Client: k8sClient,
|
||||||
Scheme: k8sClient.Scheme(),
|
Scheme: k8sClient.Scheme(),
|
||||||
|
Instances: fakeInstances{validateVersion: testPostgreSQLVersion, registryVersion: testPostgreSQLVersion},
|
||||||
}
|
}
|
||||||
|
|
||||||
_, err := controllerReconciler.Reconcile(ctx, reconcile.Request{
|
_, err := controllerReconciler.Reconcile(ctx, reconcile.Request{
|
||||||
@@ -95,6 +96,14 @@ var _ = Describe("PostgreSQLInstance Controller", func() {
|
|||||||
HaveField("ObservedGeneration", actual.Generation),
|
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
|
resourceVersion := actual.ResourceVersion
|
||||||
_, err = controllerReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName})
|
_, err = controllerReconciler.Reconcile(ctx, reconcile.Request{NamespacedName: typeNamespacedName})
|
||||||
Expect(err).NotTo(HaveOccurred())
|
Expect(err).NotTo(HaveOccurred())
|
||||||
|
|||||||
@@ -28,12 +28,12 @@ import (
|
|||||||
type instancePhaseHandler func(context.Context, *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error)
|
type instancePhaseHandler func(context.Context, *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error)
|
||||||
|
|
||||||
type instanceStateMachine struct {
|
type instanceStateMachine struct {
|
||||||
initializer PostgreSQLInstanceInitializer
|
instances PostgreSQLInstances
|
||||||
handlers map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler
|
handlers map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler
|
||||||
}
|
}
|
||||||
|
|
||||||
func newInstanceStateMachine(initializer PostgreSQLInstanceInitializer) *instanceStateMachine {
|
func newInstanceStateMachine(instances PostgreSQLInstances) *instanceStateMachine {
|
||||||
m := &instanceStateMachine{initializer: initializer}
|
m := &instanceStateMachine{instances: instances}
|
||||||
m.handlers = map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler{
|
m.handlers = map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler{
|
||||||
databasev1alpha1.PostgreSQLInstancePhasePending: m.pending,
|
databasev1alpha1.PostgreSQLInstancePhasePending: m.pending,
|
||||||
databasev1alpha1.PostgreSQLInstancePhaseValidating: m.validate,
|
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) {
|
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
|
phase := instance.Status.Phase
|
||||||
if !instance.DeletionTimestamp.IsZero() {
|
if !instance.DeletionTimestamp.IsZero() {
|
||||||
phase = databasev1alpha1.PostgreSQLInstancePhaseDeleting
|
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) {
|
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,
|
return advanceInstance(instance, databasev1alpha1.PostgreSQLInstancePhaseValidating,
|
||||||
"instance dependencies are being validated"), nil
|
"instance dependencies are being validated"), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *instanceStateMachine) validate(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
|
func (m *instanceStateMachine) validate(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
|
||||||
if m.initializer == nil {
|
version, err := m.instances.Validate(ctx, instance)
|
||||||
return ctrl.Result{}, nil
|
|
||||||
}
|
|
||||||
version, err := m.initializer.Validate(ctx, instance)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return ctrl.Result{}, err
|
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) {
|
func (m *instanceStateMachine) initializeRegistry(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
|
||||||
if m.initializer == nil {
|
version, err := m.instances.InitializeRegistry(ctx, instance)
|
||||||
return ctrl.Result{}, nil
|
|
||||||
}
|
|
||||||
version, err := m.initializer.InitializeRegistry(ctx, instance)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return ctrl.Result{}, err
|
return ctrl.Result{}, err
|
||||||
}
|
}
|
||||||
@@ -105,10 +102,7 @@ func (m *instanceStateMachine) ready(ctx context.Context, instance *databasev1al
|
|||||||
if instance.Status.ObservedGeneration != instance.Generation {
|
if instance.Status.ObservedGeneration != instance.Generation {
|
||||||
return m.pending(ctx, instance)
|
return m.pending(ctx, instance)
|
||||||
}
|
}
|
||||||
if m.initializer == nil {
|
version, err := m.instances.Validate(ctx, instance)
|
||||||
return ctrl.Result{}, nil
|
|
||||||
}
|
|
||||||
version, err := m.initializer.Validate(ctx, instance)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseValidating
|
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseValidating
|
||||||
return ctrl.Result{}, err
|
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) {
|
func (m *instanceStateMachine) deleting(_ context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
|
||||||
|
m.instances.Forget(instance.Name)
|
||||||
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseDeleting
|
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseDeleting
|
||||||
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, "instance deletion is reconciling")
|
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, "instance deletion is reconciling")
|
||||||
return ctrl.Result{}, nil
|
return ctrl.Result{}, nil
|
||||||
|
|||||||
@@ -28,7 +28,7 @@ import (
|
|||||||
databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
|
databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
|
||||||
)
|
)
|
||||||
|
|
||||||
type fakeInstanceInitializer struct {
|
type fakeInstances struct {
|
||||||
validateVersion string
|
validateVersion string
|
||||||
registryVersion string
|
registryVersion string
|
||||||
err error
|
err error
|
||||||
@@ -36,18 +36,20 @@ type fakeInstanceInitializer struct {
|
|||||||
|
|
||||||
const testPostgreSQLVersion = "17.6"
|
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
|
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
|
return f.registryVersion, f.err
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ = Describe("phase handler state machines", func() {
|
var _ = Describe("phase handler state machines", func() {
|
||||||
It("persists validation intent before touching unavailable dependencies", func() {
|
It("persists validation intent before touching unavailable dependencies", func() {
|
||||||
instance := &databasev1alpha1.PostgreSQLInstance{}
|
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)
|
result, err := machine.reconcile(context.Background(), instance)
|
||||||
Expect(err).NotTo(HaveOccurred())
|
Expect(err).NotTo(HaveOccurred())
|
||||||
Expect(result.RequeueAfter).To(BeNumerically(">", 0))
|
Expect(result.RequeueAfter).To(BeNumerically(">", 0))
|
||||||
@@ -62,7 +64,7 @@ var _ = Describe("phase handler state machines", func() {
|
|||||||
Phase: databasev1alpha1.PostgreSQLInstancePhaseReady, ObservedGeneration: 3,
|
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)
|
_, err := machine.reconcile(context.Background(), instance)
|
||||||
Expect(err).NotTo(HaveOccurred())
|
Expect(err).NotTo(HaveOccurred())
|
||||||
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
|
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
|
||||||
@@ -76,12 +78,12 @@ var _ = Describe("phase handler state machines", func() {
|
|||||||
Phase: databasev1alpha1.PostgreSQLInstancePhaseReady, ObservedGeneration: 3,
|
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)
|
_, err := machine.reconcile(context.Background(), instance)
|
||||||
Expect(err).To(MatchError("unavailable"))
|
Expect(err).To(MatchError("unavailable"))
|
||||||
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
|
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
|
||||||
Expect(instance.Status.Conditions[0].Status).To(Equal(metav1.ConditionFalse))
|
Expect(instance.Status.Conditions[0].Status).To(Equal(metav1.ConditionFalse))
|
||||||
machine.initializer = fakeInstanceInitializer{
|
machine.instances = fakeInstances{
|
||||||
validateVersion: testPostgreSQLVersion,
|
validateVersion: testPostgreSQLVersion,
|
||||||
registryVersion: testPostgreSQLVersion,
|
registryVersion: testPostgreSQLVersion,
|
||||||
}
|
}
|
||||||
@@ -102,7 +104,7 @@ var _ = Describe("phase handler state machines", func() {
|
|||||||
ObjectMeta: metav1.ObjectMeta{Generation: 3},
|
ObjectMeta: metav1.ObjectMeta{Generation: 3},
|
||||||
Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating},
|
Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating},
|
||||||
}
|
}
|
||||||
machine := newInstanceStateMachine(fakeInstanceInitializer{
|
machine := newInstanceStateMachine(fakeInstances{
|
||||||
validateVersion: testPostgreSQLVersion,
|
validateVersion: testPostgreSQLVersion,
|
||||||
registryVersion: testPostgreSQLVersion,
|
registryVersion: testPostgreSQLVersion,
|
||||||
})
|
})
|
||||||
@@ -124,7 +126,7 @@ var _ = Describe("phase handler state machines", func() {
|
|||||||
ObjectMeta: metav1.ObjectMeta{Generation: 2},
|
ObjectMeta: metav1.ObjectMeta{Generation: 2},
|
||||||
Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating},
|
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)
|
_, err := machine.reconcile(context.Background(), instance)
|
||||||
Expect(err).To(MatchError("unavailable"))
|
Expect(err).To(MatchError("unavailable"))
|
||||||
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
|
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
|
||||||
@@ -135,7 +137,7 @@ var _ = Describe("phase handler state machines", func() {
|
|||||||
|
|
||||||
DescribeTable("dispatches Instance phases",
|
DescribeTable("dispatches Instance phases",
|
||||||
func(instance *databasev1alpha1.PostgreSQLInstance, expected databasev1alpha1.PostgreSQLInstancePhase, hasMessage bool) {
|
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(err).NotTo(HaveOccurred())
|
||||||
Expect(instance.Status.Phase).To(Equal(expected))
|
Expect(instance.Status.Phase).To(Equal(expected))
|
||||||
if hasMessage {
|
if hasMessage {
|
||||||
@@ -149,12 +151,12 @@ var _ = Describe("phase handler state machines", func() {
|
|||||||
databasev1alpha1.PostgreSQLInstancePhaseValidating,
|
databasev1alpha1.PostgreSQLInstancePhaseValidating,
|
||||||
true,
|
true,
|
||||||
),
|
),
|
||||||
Entry("keeps an active phase until its handler can observe dependencies",
|
Entry("initializes the registry and reaches Ready",
|
||||||
&databasev1alpha1.PostgreSQLInstance{Status: databasev1alpha1.PostgreSQLInstanceStatus{
|
&databasev1alpha1.PostgreSQLInstance{Status: databasev1alpha1.PostgreSQLInstanceStatus{
|
||||||
Phase: databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry,
|
Phase: databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry,
|
||||||
}},
|
}},
|
||||||
databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry,
|
databasev1alpha1.PostgreSQLInstancePhaseReady,
|
||||||
false,
|
true,
|
||||||
),
|
),
|
||||||
Entry("routes deletion independently of the previous phase",
|
Entry("routes deletion independently of the previous phase",
|
||||||
deletingInstance(),
|
deletingInstance(),
|
||||||
|
|||||||
@@ -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
|
|
||||||
}
|
|
||||||
@@ -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")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -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}
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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}
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user