106 lines
3.5 KiB
Go
106 lines
3.5 KiB
Go
/*
|
|
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}
|
|
}
|