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