feat: reconcile PostgreSQLInstance dependencies
This commit is contained in:
@@ -18,7 +18,9 @@ package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"reflect"
|
||||
"time"
|
||||
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
@@ -32,7 +34,15 @@ import (
|
||||
// PostgreSQLInstanceReconciler reconciles a PostgreSQLInstance object
|
||||
type PostgreSQLInstanceReconciler struct {
|
||||
client.Client
|
||||
Scheme *runtime.Scheme
|
||||
Scheme *runtime.Scheme
|
||||
Initializer PostgreSQLInstanceInitializer
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
// PostgreSQLInstanceInitializer is the external dependency boundary used by the Instance state machine.
|
||||
type PostgreSQLInstanceInitializer interface {
|
||||
Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error)
|
||||
InitializeRegistry(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error)
|
||||
}
|
||||
|
||||
// +kubebuilder:rbac:groups=database.ddupan.top,resources=postgresqlinstances,verbs=get;list;watch;create;update;patch;delete
|
||||
@@ -62,6 +72,12 @@ func (r *PostgreSQLInstanceReconciler) Reconcile(ctx context.Context, req ctrl.R
|
||||
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, phaseResult.reconcilingMessage)
|
||||
}
|
||||
|
||||
var reconcileErr error
|
||||
if r.Initializer != nil && instance.DeletionTimestamp.IsZero() &&
|
||||
before.Status.Phase != "" && before.Status.Phase != databasev1alpha1.PostgreSQLInstancePhasePending {
|
||||
reconcileErr = r.reconcileDependencies(ctx, instance)
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(before.Status, instance.Status) {
|
||||
if err := r.Status().Patch(ctx, instance, client.MergeFrom(before)); err != nil {
|
||||
if apierrors.IsConflict(err) {
|
||||
@@ -71,7 +87,56 @@ func (r *PostgreSQLInstanceReconciler) Reconcile(ctx context.Context, req ctrl.R
|
||||
}
|
||||
}
|
||||
|
||||
return ctrl.Result{}, nil
|
||||
return ctrl.Result{}, reconcileErr
|
||||
}
|
||||
|
||||
func (r *PostgreSQLInstanceReconciler) reconcileDependencies(
|
||||
ctx context.Context,
|
||||
instance *databasev1alpha1.PostgreSQLInstance,
|
||||
) error {
|
||||
if r.Timeout > 0 {
|
||||
var cancel context.CancelFunc
|
||||
ctx, cancel = context.WithTimeout(ctx, r.Timeout)
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
if instance.Status.Phase == databasev1alpha1.PostgreSQLInstancePhaseReady &&
|
||||
instance.Status.ObservedGeneration != instance.Generation {
|
||||
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseValidating
|
||||
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, "instance dependencies are being validated")
|
||||
return nil
|
||||
}
|
||||
|
||||
var version string
|
||||
var err error
|
||||
switch instance.Status.Phase {
|
||||
case databasev1alpha1.PostgreSQLInstancePhaseValidating:
|
||||
version, err = r.Initializer.Validate(ctx, instance)
|
||||
if err == nil {
|
||||
instance.Status.PostgreSQLVersion = version
|
||||
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry
|
||||
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, "PostgreSQL registry is being initialized")
|
||||
}
|
||||
case databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry:
|
||||
version, err = r.Initializer.InitializeRegistry(ctx, instance)
|
||||
if err == nil {
|
||||
instance.Status.PostgreSQLVersion = version
|
||||
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseReady
|
||||
instance.Status.ObservedGeneration = instance.Generation
|
||||
setReadyCondition(&instance.Status.Conditions, instance.Generation, "instance dependencies are ready")
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
instance.Status.ObservedGeneration = instance.Generation
|
||||
reason := databasev1alpha1.ReasonDependencyUnavailable
|
||||
var categorized interface{ ConditionReason() string }
|
||||
if errors.As(err, &categorized) {
|
||||
reason = categorized.ConditionReason()
|
||||
}
|
||||
setFailedCondition(&instance.Status.Conditions, instance.Generation, reason,
|
||||
"instance dependency validation failed")
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// SetupWithManager sets up the controller with the Manager.
|
||||
|
||||
@@ -17,6 +17,8 @@ limitations under the License.
|
||||
package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
@@ -26,7 +28,53 @@ import (
|
||||
databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
|
||||
)
|
||||
|
||||
type fakeInstanceInitializer struct {
|
||||
validateVersion string
|
||||
registryVersion string
|
||||
err error
|
||||
}
|
||||
|
||||
func (f fakeInstanceInitializer) Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) {
|
||||
return f.validateVersion, f.err
|
||||
}
|
||||
|
||||
func (f fakeInstanceInitializer) InitializeRegistry(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) {
|
||||
return f.registryVersion, f.err
|
||||
}
|
||||
|
||||
var _ = Describe("phase handler state machines", func() {
|
||||
It("advances an Instance through external validation and registry initialization", func() {
|
||||
instance := &databasev1alpha1.PostgreSQLInstance{
|
||||
ObjectMeta: metav1.ObjectMeta{Generation: 3},
|
||||
Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating},
|
||||
}
|
||||
reconciler := &PostgreSQLInstanceReconciler{Initializer: fakeInstanceInitializer{
|
||||
validateVersion: "17.6", registryVersion: "17.6",
|
||||
}}
|
||||
Expect(reconciler.reconcileDependencies(context.Background(), instance)).To(Succeed())
|
||||
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry))
|
||||
Expect(reconciler.reconcileDependencies(context.Background(), instance)).To(Succeed())
|
||||
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseReady))
|
||||
Expect(instance.Status.PostgreSQLVersion).To(Equal("17.6"))
|
||||
Expect(instance.Status.ObservedGeneration).To(Equal(int64(3)))
|
||||
Expect(instance.Status.Conditions).To(ConsistOf(And(
|
||||
HaveField("Status", metav1.ConditionTrue), HaveField("Reason", databasev1alpha1.ReasonReady),
|
||||
)))
|
||||
})
|
||||
|
||||
It("keeps the safe phase when a dependency is unavailable", func() {
|
||||
instance := &databasev1alpha1.PostgreSQLInstance{
|
||||
ObjectMeta: metav1.ObjectMeta{Generation: 2},
|
||||
Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating},
|
||||
}
|
||||
reconciler := &PostgreSQLInstanceReconciler{Initializer: fakeInstanceInitializer{err: errors.New("unavailable")}}
|
||||
Expect(reconciler.reconcileDependencies(context.Background(), instance)).To(MatchError("unavailable"))
|
||||
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
|
||||
Expect(instance.Status.Conditions).To(ConsistOf(And(
|
||||
HaveField("Status", metav1.ConditionFalse), HaveField("Reason", databasev1alpha1.ReasonDependencyUnavailable),
|
||||
)))
|
||||
})
|
||||
|
||||
DescribeTable("dispatches Instance phases",
|
||||
func(instance *databasev1alpha1.PostgreSQLInstance, expected databasev1alpha1.PostgreSQLInstancePhase, hasMessage bool) {
|
||||
result := newInstanceStateMachine().reconcile(instance)
|
||||
|
||||
@@ -32,3 +32,17 @@ func setReconcilingCondition(conditions *[]metav1.Condition, generation int64, m
|
||||
Message: message,
|
||||
})
|
||||
}
|
||||
|
||||
func setReadyCondition(conditions *[]metav1.Condition, generation int64, message string) {
|
||||
apiMeta.SetStatusCondition(conditions, metav1.Condition{
|
||||
Type: databasev1alpha1.ConditionTypeReady, Status: metav1.ConditionTrue,
|
||||
ObservedGeneration: generation, Reason: databasev1alpha1.ReasonReady, Message: message,
|
||||
})
|
||||
}
|
||||
|
||||
func setFailedCondition(conditions *[]metav1.Condition, generation int64, reason, message string) {
|
||||
apiMeta.SetStatusCondition(conditions, metav1.Condition{
|
||||
Type: databasev1alpha1.ConditionTypeReady, Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: generation, Reason: reason, Message: message,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
/*
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
/*
|
||||
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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user