panxiao81/postgresql-tenant-operator
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: wire shared Instance dependencies at startup

Browse Source
This commit is contained in:
panxiao81
2026-09-11 16:10:36 +00:00
parent 305e778203
commit 61985acb32
13 changed files with 221 additions and 337 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/controller/postgresqlinstance_controller.go
+16 -7
View File
@@ -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").
internal/controller/postgresqlinstance_controller_test.go
+11 -2
View File
@@ -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())
internal/controller/postgresqlinstance_state_machine.go
+11 -16
View File
@@ -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
internal/controller/state_machine_test.go
+15 -13
View File
@@ -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(),
internal/instance/initializer.go
-216
View File
@@ -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
}
internal/instance/initializer_test.go
-44
View File
@@ -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")
}
}