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

refactor: assemble and reuse Instance dependencies
E2E Tests / Run on Ubuntu (pull_request) Failing after 59s
Details
Tests / Run on Ubuntu (pull_request) Successful in 3m55s
Details
Lint / Run on Ubuntu (pull_request) Successful in 4m18s
Details

Browse Source
This commit is contained in:
panxiao81
2026-09-11 15:54:05 +00:00
parent 89d9e2316b
commit a4d835e612
21 changed files with 975 additions and 340 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/instance/errors.go
+27
View File
@@ -0,0 +1,27 @@
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package instance
// Failure contains only a safe operation message and a Condition reason.
// Raw driver errors can contain credentials and must not escape through wrapping.
type Failure struct {
Reason string
Operation string
}
func (e Failure) Error() string { return e.Operation }
func (e Failure) ConditionReason() string { return e.Reason }
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")
}
}
internal/instance/service.go
+152
View File
@@ -0,0 +1,152 @@
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package instance manages the lifetime and capabilities of external PostgreSQL instances.
package instance
import (
"context"
"errors"
"sync"
api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
"k8s.io/apimachinery/pkg/types"
)
// Credentials are supplied by the credential source, never by the database adapter.
type Credentials struct{ Username, Password string }
func (Credentials) String() string { return "[redacted credentials]" }
func (Credentials) GoString() string { return "[redacted credentials]" }
// CredentialSource resolves administrative credentials when an Instance is assembled.
type CredentialSource interface {
Read(context.Context, api.OpenBaoSecretReference) (Credentials, error)
}
// Database exposes the capabilities of an assembled Instance.
type Database interface {
Version(context.Context) (string, error)
EnsureRegistry(context.Context) error
Close()
}
// Connector creates a database without knowing where credentials came from.
type Connector interface {
Connect(context.Context, api.PostgreSQLEndpoint, Credentials) (Database, error)
}
type identity struct {
uid types.UID
endpoint api.PostgreSQLEndpoint
reference api.OpenBaoSecretReference
}
type entry struct {
identity identity
credentials *Credentials
database Database
}
// Service owns Instance connections. Reconciles reuse credentials and pools until
// the UID, endpoint or credential reference changes; allowlist changes do not rotate them.
// Operations are serialized to prevent Close racing with an active database operation.
type Service struct {
mu sync.Mutex
source CredentialSource
connector Connector
entries map[string]*entry
closed bool
}
func NewService(source CredentialSource, connector Connector) *Service {
return &Service{source: source, connector: connector, entries: make(map[string]*entry)}
}
func (s *Service) Validate(ctx context.Context, resource *api.PostgreSQLInstance) (string, error) {
return s.observe(ctx, resource, false)
}
func (s *Service) InitializeRegistry(ctx context.Context, resource *api.PostgreSQLInstance) (string, error) {
return s.observe(ctx, resource, true)
}
func (s *Service) observe(ctx context.Context, resource *api.PostgreSQLInstance, ensureRegistry bool) (string, error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.closed {
return "", errors.New("instance service is closed")
}
db, err := s.database(ctx, resource)
if err != nil {
return "", err
}
if ensureRegistry {
if err := db.EnsureRegistry(ctx); err != nil {
return "", err
}
}
return db.Version(ctx)
}
func (s *Service) database(ctx context.Context, resource *api.PostgreSQLInstance) (Database, error) {
key := identity{uid: resource.UID, endpoint: resource.Spec.Endpoint, reference: resource.Spec.AdminCredentialRef}
current := s.entries[resource.Name]
if current == nil || current.identity != key {
s.release(resource.Name)
current = &entry{identity: key}
s.entries[resource.Name] = current
}
if current.credentials == nil {
credentials, err := s.source.Read(ctx, key.reference)
if err != nil {
return nil, err
}
current.credentials = &credentials
}
if current.database == nil {
database, err := s.connector.Connect(ctx, key.endpoint, *current.credentials)
if err != nil {
return nil, err
}
current.database = database
}
return current.database, nil
}
// Forget releases local resources only. It never deletes the external registry.
func (s *Service) Forget(name string) {
s.mu.Lock()
defer s.mu.Unlock()
s.release(name)
}
func (s *Service) release(name string) {
if current := s.entries[name]; current != nil && current.database != nil {
current.database.Close()
}
delete(s.entries, name)
}
// Close is called after manager workers stop, and is safe to repeat.
func (s *Service) Close() {
s.mu.Lock()
defer s.mu.Unlock()
s.closed = true
for name := range s.entries {
s.release(name)
}
}
internal/instance/service_test.go
+156
View File
@@ -0,0 +1,156 @@
/*
Copyright 2026.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package instance
import (
"context"
"errors"
"fmt"
"strings"
"testing"
api "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
type sourceStub struct {
reads int
err error
}
func (s *sourceStub) Read(context.Context, api.OpenBaoSecretReference) (Credentials, error) {
s.reads++
return Credentials{Username: "admin", Password: "canary-secret"}, s.err
}
type databaseStub struct {
closes, migrations int
err error
}
func (d *databaseStub) Version(context.Context) (string, error) { return "17", d.err }
func (d *databaseStub) EnsureRegistry(context.Context) error { d.migrations++; return d.err }
func (d *databaseStub) Close() { d.closes++ }
type connectorStub struct {
databases []*databaseStub
err error
}
func (c *connectorStub) Connect(context.Context, api.PostgreSQLEndpoint, Credentials) (Database, error) {
if c.err != nil {
return nil, c.err
}
database := &databaseStub{}
c.databases = append(c.databases, database)
return database, nil
}
func TestInstanceConnectionLifetime(t *testing.T) {
source, connector := &sourceStub{}, &connectorStub{}
service := NewService(source, connector)
defer service.Close()
resource := &api.PostgreSQLInstance{ObjectMeta: metav1.ObjectMeta{Name: "shared", UID: "uid-1"}}
ctx := context.Background()
check := func() {
t.Helper()
if _, err := service.Validate(ctx, resource); err != nil {
t.Fatalf("validate: %v", err)
}
}
check()
if _, err := service.InitializeRegistry(ctx, resource); err != nil {
t.Fatal(err)
}
resource.Generation++
resource.Spec.AllowedExtensions = []string{"pg_trgm"}
check()
if source.reads != 1 || len(connector.databases) != 1 || connector.databases[0].migrations != 1 {
t.Fatal("reconciliation or allowlist update unnecessarily reassembled dependencies")
}
connector.databases[0].err = errors.New("network outage")
if _, err := service.Validate(ctx, resource); err == nil {
t.Fatal("outage went unnoticed")
}
connector.databases[0].err = nil
check()
if source.reads != 1 {
t.Fatal("network recovery reread the password")
}
changes := []func(){
func() { resource.Spec.Endpoint.Host = "new.example" },
func() { resource.Spec.AdminCredentialRef.Path = "new/admin" },
func() { resource.UID = "uid-2" },
}
for index, change := range changes {
change()
check()
if connector.databases[index].closes != 1 {
t.Fatal("replaced pool was not closed")
}
}
if source.reads != 4 {
t.Fatal("changed connection identity did not reload credentials")
}
service.Forget(resource.Name)
service.Forget(resource.Name)
if connector.databases[3].closes != 1 {
t.Fatal("deletion did not close pool exactly once")
}
check()
service.Close()
service.Close()
if connector.databases[4].closes != 1 {
t.Fatal("shutdown did not close pool exactly once")
}
if _, err := service.Validate(ctx, resource); err == nil {
t.Fatal("closed service accepted work")
}
}
func TestAssemblyRetriesWithoutPasswordRotation(t *testing.T) {
source := &sourceStub{err: errors.New("bao unavailable")}
connector := &connectorStub{err: errors.New("connection unavailable")}
service := NewService(source, connector)
defer service.Close()
resource := &api.PostgreSQLInstance{ObjectMeta: metav1.ObjectMeta{Name: "shared"}}
ctx := context.Background()
if _, err := service.Validate(ctx, resource); err == nil {
t.Fatal("expected source failure")
}
source.err = nil
if _, err := service.Validate(ctx, resource); err == nil {
t.Fatal("expected connection failure")
}
connector.err = nil
if _, err := service.Validate(ctx, resource); err != nil {
t.Fatal(err)
}
if source.reads != 2 {
t.Fatal("retry reread successfully cached credentials")
}
}
func TestCredentialsFormattingIsRedacted(t *testing.T) {
value := Credentials{Username: "canary-user", Password: "canary-secret"}
for _, format := range []string{"%v", "%+v", "%#v"} {
if strings.Contains(fmt.Sprintf(format, value), "canary") {
t.Fatal("credential formatting leaked data")
}
}
}