From 7834cab97f73672f9c1ee484c0337ac722e2251d Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Sun, 27 Sep 2026 19:16:48 +0000 Subject: [PATCH] =?UTF-8?q?refactor:=20=E9=9B=86=E4=B8=AD=E6=98=BE?= =?UTF-8?q?=E5=BC=8F=E6=B3=A8=E5=85=A5=E5=B9=B6=E5=BD=92=E4=BD=8D=E5=87=AD?= =?UTF-8?q?=E6=8D=AE=E9=A2=86=E5=9F=9F=E8=A7=84=E5=88=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/architecture/overview.md | 11 ++ docs/database/README.md | 4 + internal/bootstrap/database.go | 20 +-- internal/bootstrap/database_wiring.go | 28 ++++ internal/bootstrap/database_wiring_test.go | 74 ++++++++++ .../bootstrap/manager_integration_test.go | 6 +- internal/bootstrap/run.go | 4 +- .../adapter/kubernetes/binding_mapping.go | 8 -- .../kubernetes/credential_conditions.go | 28 ++++ .../kubernetes/credential_resources.go | 45 +++--- .../adapter/openbao/credential_store.go | 14 +- .../database/adapter/openbao/credentials.go | 24 ++-- .../adapter/openbao/credentials_test.go | 7 +- ...reparation_concurrency_integration_test.go | 2 +- .../preparation_fixture_integration_test.go | 5 +- .../openbao/preparation_integration_test.go | 18 +-- .../preparation_pruning_integration_test.go | 3 +- .../preparation_watch_integration_test.go | 20 ++- .../instance_controller_integration_test.go | 4 +- .../application/credential_preparation.go | 119 ++++++---------- .../credential_preparation_test.go | 24 ++-- .../database/application/credential_target.go | 27 ---- .../database/controller/binding_controller.go | 20 ++- .../controller/binding_controller_test.go | 25 ++-- internal/database/controller/binding_setup.go | 15 +- .../controller/credential_controller.go | 23 ++-- .../controller/instance_controller.go | 19 ++- .../controller/instance_controller_test.go | 7 +- .../database/controller/instance_setup.go | 8 +- .../credential/credential.go} | 2 +- .../credential/credential_test.go} | 17 +-- .../database/domain/credential/preparation.go | 129 ++++++++++++++++++ .../domain/credential/preparation_test.go | 59 ++++++++ .../database/domain/credential/target_test.go | 52 +++++++ 34 files changed, 612 insertions(+), 259 deletions(-) create mode 100644 internal/bootstrap/database_wiring.go create mode 100644 internal/bootstrap/database_wiring_test.go create mode 100644 internal/database/adapter/kubernetes/credential_conditions.go delete mode 100644 internal/database/application/credential_target.go rename internal/database/{application/application_credential.go => domain/credential/credential.go} (99%) rename internal/database/{application/application_credential_test.go => domain/credential/credential_test.go} (77%) create mode 100644 internal/database/domain/credential/preparation.go create mode 100644 internal/database/domain/credential/preparation_test.go create mode 100644 internal/database/domain/credential/target_test.go diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 666a65c..b7b43e5 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -56,6 +56,17 @@ Makefile 与 Dockerfile 均继续构建 `cmd/main.go`。 条件分支，也不为此引入插件注册框架。组件启动失败时释放已装配资源，正常退出则先停止 manager worker，再释放连接。 +Database 使用 Wire 式的显式构造函数注入，目前手写装配，不引入 Wire 生成器、Dig/Fx +容器或运行时服务查找。`database_wiring.go` 集中展示 repository → 用例 → controller 的 +对象构造；`registerDatabaseControllers` 负责注册，返回的 `closeDatabaseConnections` +只在 worker 停止后关闭 Instance 连接。controller 在启动时接收完整依赖，不在 Reconcile +或 SetupWithManager 中补建 adapter/service；缺失依赖在注册时失败。 + +domain 持有凭据值对象、准备资格与创建恢复规则；application 只组织读写、调用领域判断 +和保存结果。用例需要的 repository 接口与并发快照仍由消费方定义，不把所有类型都强塞 +进领域。Kubernetes adapter 将领域阶段映射为已有 Conditions，领域不依赖其字符串协议。 +单测检查 domain/controller 的依赖边界，并验证共享 writer、直连 reader 和服务的注入。 + 基础设施能力属于整个 controller-manager，不因首个消费者是 Database 就归入该领域。 `internal/infra/openbao` 管理官方 SDK client 的 TLS 配置、Kubernetes 认证及 token 生命周期， 不依赖 Database 或其他产品领域。Bao client 默认禁用自动重试，写入结果不确定时由用例处理； diff --git a/docs/database/README.md b/docs/database/README.md index 5df95c0..547b91f 100644 --- a/docs/database/README.md +++ b/docs/database/README.md @@ -183,6 +183,10 @@ Database 已有 `status.credentialRef` 和 `status.credentialVersion` 的字段 顺序为固定位置 → 确认后端尚无凭据 → 保存 CreationStarted → 创建并回读 → 保存确认版本。 `CredentialReconciler` 只负责 Database/Tenant/Instance watch 和 30 秒依赖重查； Kubernetes repository 负责直接读取及有 resourceVersion 保护的状态更新。 +三类 controller 的依赖均在 bootstrap 显式组装，不在 Reconcile 中构造服务。 +`domain/credential` 保存应用凭据值对象、准备资格、固定位置与未确认创建的恢复规则； +application 保留 I/O 顺序、消费方接口及并发快照，不再复用绑定用例的快照类型。 +领域阶段与 `CredentialsReady`/Reason 的转换由 Kubernetes adapter 负责，已有 API 保持兼容。 外部操作前后回查目标：Database 必须仍是同一 UID/resourceVersion，Tenant/Instance 必须 保持绑定、spec generation、删除状态、finalizer 保护与有效 Ready；无关 Conditions 刷新 diff --git a/internal/bootstrap/database.go b/internal/bootstrap/database.go index d4d207a..d94af8f 100644 --- a/internal/bootstrap/database.go +++ b/internal/bootstrap/database.go @@ -37,25 +37,25 @@ func (o databaseOptions) configureManager(options *ctrl.Options) { } } -// setupDatabase 封装 Database 的内部装配，并返回在 manager 停止后执行的清理。 -func setupDatabase(ctx context.Context, manager ctrl.Manager, options databaseOptions, baoClient *bao.Client) (func(), error) { - cleanup := func() {} +// registerDatabaseControllers 封装 Database 的内部装配，并返回在 manager 停止后执行的清理。 +func registerDatabaseControllers(ctx context.Context, manager ctrl.Manager, options databaseOptions, baoClient *bao.Client) (func(), error) { + closeDatabaseConnections := func() {} if options.secretNamespace != "" { service, err := setupInstanceObservation(manager, options.secretNamespace, options.rootCert) if err != nil { return nil, fmt.Errorf("set up Instance observation: %w", err) } - cleanup = service.Close + closeDatabaseConnections = service.Close } - if err := (&databasecontroller.BindingReconciler{}).SetupWithManager(ctx, manager); err != nil { - cleanup() + if err := wireBindingController(manager.GetClient(), manager.GetAPIReader()).SetupWithManager(ctx, manager); err != nil { + closeDatabaseConnections() return nil, fmt.Errorf("set up Database binding controller: %w", err) } if err := setupCredentialPreparation(manager, options, baoClient); err != nil { - cleanup() + closeDatabaseConnections() return nil, fmt.Errorf("set up Database credential preparation: %w", err) } - return cleanup, nil + return closeDatabaseConnections, nil } func setupCredentialPreparation(manager ctrl.Manager, options databaseOptions, baoClient *bao.Client) error { @@ -69,7 +69,7 @@ func setupCredentialPreparation(manager ctrl.Manager, options databaseOptions, b if err != nil { return err } - return (&databasecontroller.CredentialReconciler{Store: store}).SetupWithManager(manager) + return wireCredentialController(manager.GetClient(), manager.GetAPIReader(), store).SetupWithManager(manager) } func setupInstanceObservation(manager ctrl.Manager, namespace, rootCert string) (*application.InstanceService, error) { @@ -81,7 +81,7 @@ func setupInstanceObservation(manager ctrl.Manager, namespace, rootCert string) if err != nil { return nil, err } - reconciler := &databasecontroller.InstanceReconciler{Observer: service, SecretNamespace: namespace} + reconciler := wireInstanceController(manager.GetClient(), manager.GetAPIReader(), service, namespace) if err := reconciler.SetupWithManager(manager); err != nil { service.Close() return nil, err diff --git a/internal/bootstrap/database_wiring.go b/internal/bootstrap/database_wiring.go new file mode 100644 index 0000000..e445ba9 --- /dev/null +++ b/internal/bootstrap/database_wiring.go @@ -0,0 +1,28 @@ +package bootstrap + +import ( + "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes" + "git.ddupan.top/panxiao81/ayatori/internal/database/application" + databasecontroller "git.ddupan.top/panxiao81/ayatori/internal/database/controller" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// 本文件是 Database 的显式依赖图：adapter → 用例 → controller。 +// 构造只在启动时执行；用例和 controller 都不持有容器或动态查找依赖。 +func wireBindingController(writer client.Client, reader client.Reader) *databasecontroller.BindingReconciler { + resources := &kubernetes.BindingResources{Client: writer, Reader: reader} + service := &application.BindingService{Resources: resources} + return databasecontroller.NewBindingReconciler(writer, service, resources) +} + +func wireCredentialController(writer client.Client, reader client.Reader, store application.CredentialStore) *databasecontroller.CredentialReconciler { + resources := &kubernetes.CredentialResources{Client: writer, Reader: reader} + service := &application.CredentialPreparation{Resources: resources, Store: store} + return databasecontroller.NewCredentialReconciler(writer, service) +} + +func wireInstanceController(writer client.Client, reader client.Reader, observer application.InstanceObserver, namespace string) *databasecontroller.InstanceReconciler { + resources := &kubernetes.InstanceResources{Client: writer, Reader: reader} + service := &application.InstanceReconciliation{Resources: resources, Observer: observer} + return databasecontroller.NewInstanceReconciler(writer, service, resources, namespace) +} diff --git a/internal/bootstrap/database_wiring_test.go b/internal/bootstrap/database_wiring_test.go new file mode 100644 index 0000000..05ab9ac --- /dev/null +++ b/internal/bootstrap/database_wiring_test.go @@ -0,0 +1,74 @@ +package bootstrap + +import ( + "go/parser" + "go/token" + "io/fs" + "path/filepath" + "strconv" + "strings" + "testing" + + "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes" + "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/postgresql" + "git.ddupan.top/panxiao81/ayatori/internal/database/application" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +const controllerLayer = "controller" + +func TestDatabaseExplicitWiring(t *testing.T) { + writer := fake.NewClientBuilder().Build() + directReader := fake.NewClientBuilder().Build() + binder := wireBindingController(writer, directReader) + bindingResources, ok := binder.Service.Resources.(*kubernetes.BindingResources) + if !ok || binder.Client != writer || bindingResources.Reader != directReader || bindingResources.Client != writer || binder.Presenter != bindingResources { + t.Fatal("绑定用例没有共享显式注入的 writer、直连 reader 与 presenter") + } + credentials, err := kubernetes.NewSecretCredentials(directReader, "wiring-tests") + if err != nil { + t.Fatal(err) + } + observer, err := application.NewInstanceService(credentials, postgresql.Connector{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(observer.Close) + reconciler := wireInstanceController(writer, directReader, observer, "wiring-tests") + resources, ok := reconciler.Service.Resources.(*kubernetes.InstanceResources) + if !ok || resources.Reader != directReader || resources.Client != writer || reconciler.Service.Observer != observer || reconciler.Presenter != resources { + t.Fatal("Instance 的服务或读取边界未按依赖图注入") + } +} + +// 防止领域重新依赖用例/存储，也防止 controller 再次私自构造具体 adapter。 +func TestDatabaseLayerImports(t *testing.T) { + for _, layer := range []string{"domain", controllerLayer} { + err := filepath.WalkDir(filepath.Join("../database", layer), func(path string, entry fs.DirEntry, err error) error { + if err != nil { + return err + } + if entry.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { + return nil + } + file, err := parser.ParseFile(token.NewFileSet(), path, nil, parser.ImportsOnly) + if err != nil { + return err + } + for _, dependency := range file.Imports { + name, err := strconv.Unquote(dependency.Path.Value) + if err != nil { + return err + } + if strings.Contains(name, "/database/adapter/") || (layer == "domain" && + (strings.Contains(name, "/database/application") || strings.Contains(name, "k8s.io/") || strings.Contains(name, "/internal/infra/"))) { + t.Errorf("%s 不得导入 %s", path, name) + } + } + return nil + }) + if err != nil { + t.Fatal(err) + } + } +} diff --git a/internal/bootstrap/manager_integration_test.go b/internal/bootstrap/manager_integration_test.go index cfde582..337afe2 100644 --- a/internal/bootstrap/manager_integration_test.go +++ b/internal/bootstrap/manager_integration_test.go @@ -54,11 +54,11 @@ func TestBootstrapWithRealAPIServer(t *testing.T) { } ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) defer cancel() - cleanup, err := setupDatabase(ctx, manager, options.database, baoClient) + closeDatabaseConnections, err := registerDatabaseControllers(ctx, manager, options.database, baoClient) if err != nil { t.Fatal(err) } - defer cleanup() + defer closeDatabaseConnections() // 空 API 中没有供应目标；此处验证启用路径确实注册 controller，不访问外部 Bao。 fixtureConfig := bao.NewConfig() fixtureConfig.Address = "http://127.0.0.1:1" @@ -80,7 +80,7 @@ func TestBootstrapWithRealAPIServer(t *testing.T) { t.Error(err) } case <-time.After(20 * time.Second): - t.Error("manager did not stop before component cleanup") + t.Error("manager did not stop before component closeDatabaseConnections") } }() if !manager.GetCache().WaitForCacheSync(ctx) { diff --git a/internal/bootstrap/run.go b/internal/bootstrap/run.go index 77f7575..c5b8fe2 100644 --- a/internal/bootstrap/run.go +++ b/internal/bootstrap/run.go @@ -27,12 +27,12 @@ func Run() error { if err != nil { return fmt.Errorf("set up OpenBao authentication: %w", err) } - cleanup, err := setupDatabase(context.Background(), manager, options.database, baoClient) + closeDatabaseConnections, err := registerDatabaseControllers(context.Background(), manager, options.database, baoClient) if err != nil { return err } // manager 的 worker 完全停止后才释放组件持有的资源。 - defer cleanup() + defer closeDatabaseConnections() setupLog.Info("Starting manager") if err := manager.Start(ctrl.SetupSignalHandler()); err != nil { diff --git a/internal/database/adapter/kubernetes/binding_mapping.go b/internal/database/adapter/kubernetes/binding_mapping.go index 5349a39..2f90dd6 100644 --- a/internal/database/adapter/kubernetes/binding_mapping.go +++ b/internal/database/adapter/kubernetes/binding_mapping.go @@ -51,11 +51,3 @@ func tenantReference(tenant binding.TenantIdentity) *databasev1alpha1.TenantRefe Namespace: tenant.Namespace, Name: databasev1alpha1.ObjectName(tenant.Name), UID: types.UID(tenant.UID), } } - -// BindingTargetName 供 informer 索引使用；不把无效请求丢出事件映射。 -func BindingTargetName(tenant *databasev1alpha1.PostgreSQLTenant) string { - if tenant.Spec.DatabaseRef != nil { - return string(tenant.Spec.DatabaseRef.Name) - } - return binding.DynamicDatabaseName(string(tenant.UID)) -} diff --git a/internal/database/adapter/kubernetes/credential_conditions.go b/internal/database/adapter/kubernetes/credential_conditions.go new file mode 100644 index 0000000..48a4aa1 --- /dev/null +++ b/internal/database/adapter/kubernetes/credential_conditions.go @@ -0,0 +1,28 @@ +package kubernetes + +import credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" + +const credentialsReadyCondition = "CredentialsReady" + +// 持久化 API 的 Reason 保持兼容，领域内部只使用准备阶段。 +var credentialReasons = map[credentialdomain.Phase]string{ + credentialdomain.Pending: "PreparationPending", + credentialdomain.Pinned: "LocationPinned", + credentialdomain.Creating: "CreationStarted", + credentialdomain.Prepared: "CredentialPrepared", + credentialdomain.Conflict: "Conflict", + credentialdomain.Unavailable: "DependencyUnavailable", + credentialdomain.Stopped: "PreparationStopped", + credentialdomain.InvalidTarget: "InvalidTarget", +} + +func credentialReason(phase credentialdomain.Phase) string { return credentialReasons[phase] } + +func credentialPhase(reason string) credentialdomain.Phase { + for phase, value := range credentialReasons { + if value == reason { + return phase + } + } + return credentialdomain.Pending +} diff --git a/internal/database/adapter/kubernetes/credential_resources.go b/internal/database/adapter/kubernetes/credential_resources.go index c7eb0ca..a65f5e1 100644 --- a/internal/database/adapter/kubernetes/credential_resources.go +++ b/internal/database/adapter/kubernetes/credential_resources.go @@ -4,6 +4,8 @@ import ( "context" "fmt" + credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" + databasev1alpha1 "git.ddupan.top/panxiao81/ayatori/api/database/v1alpha1" "git.ddupan.top/panxiao81/ayatori/internal/database/application" "git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding" @@ -30,8 +32,9 @@ func (r *CredentialResources) Load(ctx context.Context, name string) (*applicati return nil, client.IgnoreNotFound(err) } record := &application.CredentialRecord{ - Database: *bindingDatabase(database), DatabaseProtected: controllerutil.ContainsFinalizer(database, DatabaseFinalizer), - Status: credentialStatus(database), + Database: bindingDatabase(database).Database, Revision: database.ResourceVersion, + DatabaseProtected: controllerutil.ContainsFinalizer(database, DatabaseFinalizer), + Status: credentialStatus(database), } if database.Spec.Source != "Provision" { return record, nil @@ -43,7 +46,8 @@ func (r *CredentialResources) Load(ctx context.Context, name string) (*applicati return nil, err } if err == nil { - record.Tenant = bindingTenant(tenant) + record.Tenant = &bindingTenant(tenant).Tenant + record.TenantGeneration = tenant.Generation record.TenantProtected = controllerutil.ContainsFinalizer(tenant, TenantFinalizer) } } @@ -56,29 +60,29 @@ func (r *CredentialResources) Load(ctx context.Context, name string) (*applicati if err != nil { return nil, err } - record.Instance = &application.CredentialInstance{ + record.Instance = &credentialdomain.Instance{ Identity: binding.Identity{Name: instance.Name, UID: string(instance.UID)}, Deleting: !instance.DeletionTimestamp.IsZero(), Ready: currentReady(instance.Generation, instance.Status.Conditions), - Generation: instance.Generation, Endpoint: observed.Target.Definition().Endpoint(), + Endpoint: observed.Target.Definition().Endpoint(), } + record.InstanceGeneration = instance.Generation return record, nil } -func credentialStatus(database *databasev1alpha1.PostgreSQLDatabase) application.CredentialStatus { - status := application.CredentialStatus{Version: database.Status.CredentialVersion} +func credentialStatus(database *databasev1alpha1.PostgreSQLDatabase) credentialdomain.State { + status := credentialdomain.State{Version: database.Status.CredentialVersion} if ref := database.Status.CredentialRef; ref != nil { - status.Location = &application.CredentialLocation{Mount: ref.Mount, Path: ref.Path} + status.Location = &credentialdomain.Location{Mount: ref.Mount, Path: ref.Path} } - if condition := meta.FindStatusCondition(database.Status.Conditions, application.CredentialsReady); condition != nil { - status.Ready = condition.Status == metav1.ConditionTrue && condition.ObservedGeneration == database.Generation - status.Reason, status.Message = condition.Reason, condition.Message + if condition := meta.FindStatusCondition(database.Status.Conditions, credentialsReadyCondition); condition != nil { + status.Phase, status.Message = credentialPhase(condition.Reason), condition.Message } return status } -func (r *CredentialResources) Save(ctx context.Context, record *application.CredentialRecord, status application.CredentialStatus) (*application.CredentialRecord, error) { +func (r *CredentialResources) Save(ctx context.Context, record *application.CredentialRecord, status credentialdomain.State) (*application.CredentialRecord, error) { bindingResources := &BindingResources{Client: r.Client, Reader: r.Reader} - database, err := bindingResources.databaseAtVersion(ctx, &record.Database) + database, err := bindingResources.databaseAtVersion(ctx, &application.BindingDatabase{Database: record.Database, Revision: record.Revision}) if err != nil { return nil, err } @@ -88,11 +92,11 @@ func (r *CredentialResources) Save(ctx context.Context, record *application.Cred } database.Status.CredentialVersion = status.Version conditionStatus := metav1.ConditionFalse - if status.Ready { + if status.Phase == credentialdomain.Prepared { conditionStatus = metav1.ConditionTrue } meta.SetStatusCondition(&database.Status.Conditions, metav1.Condition{ - Type: application.CredentialsReady, Status: conditionStatus, Reason: status.Reason, Message: status.Message, + Type: credentialsReadyCondition, Status: conditionStatus, Reason: credentialReason(status.Phase), Message: status.Message, ObservedGeneration: database.Generation, }) meta.SetStatusCondition(&database.Status.Conditions, metav1.Condition{ @@ -106,7 +110,8 @@ func (r *CredentialResources) Save(ctx context.Context, record *application.Cred } } updated := *record - updated.Database = *bindingDatabase(database) + updated.Database = bindingDatabase(database).Database + updated.Revision = database.ResourceVersion updated.Status = credentialStatus(database) // 旧 CRD 会裁剪未知 status 字段。不能把 HTTP 成功当作位置/版本已保存后继续写后端。 if !equality.Semantic.DeepEqual(updated.Status.Location, status.Location) || updated.Status.Version != status.Version { @@ -120,7 +125,7 @@ func (r *CredentialResources) CheckCurrent(ctx context.Context, record *applicat if err != nil { return err } - if current == nil || current.Database.Identity != record.Database.Identity || current.Database.Revision != record.Database.Revision || + if current == nil || current.Database.Identity != record.Database.Identity || current.Revision != record.Revision || !sameCredentialDependencies(current, record) { return apierrors.NewConflict(databasev1alpha1.GroupVersion.WithResource("postgresqldatabases").GroupResource(), record.Database.Identity.Name, fmt.Errorf("凭据准备的资源快照已变化；停止本轮操作并重新观察")) @@ -134,9 +139,9 @@ func sameCredentialDependencies(current, previous *application.CredentialRecord) } // Tenant Ready 的诊断变化、Instance 对同一 generation 的观测刷新不改变写入目标。 // 仍检查申请 spec generation、完整绑定身份、删除状态、保护和当前 Instance Ready。 - return current.Tenant.Generation == previous.Tenant.Generation && - equality.Semantic.DeepEqual(current.Tenant.Tenant, previous.Tenant.Tenant) && + return current.TenantGeneration == previous.TenantGeneration && + equality.Semantic.DeepEqual(current.Tenant, previous.Tenant) && current.TenantProtected == previous.TenantProtected && current.DatabaseProtected == previous.DatabaseProtected && - current.Instance.Instance == previous.Instance.Instance && current.Instance.Generation == previous.Instance.Generation && + current.Instance.Instance == previous.Instance.Instance && current.InstanceGeneration == previous.InstanceGeneration && current.Instance.Endpoint == previous.Instance.Endpoint } diff --git a/internal/database/adapter/openbao/credential_store.go b/internal/database/adapter/openbao/credential_store.go index 75e24bb..2c58f8d 100644 --- a/internal/database/adapter/openbao/credential_store.go +++ b/internal/database/adapter/openbao/credential_store.go @@ -3,22 +3,24 @@ package openbao import ( "context" + credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" + "git.ddupan.top/panxiao81/ayatori/internal/database/application" ) var _ application.CredentialStore = (*Credentials)(nil) -func (c *Credentials) ProvisionLocation(uid string) (application.CredentialLocation, error) { +func (c *Credentials) ProvisionLocation(uid string) (credentialdomain.Location, error) { path, err := c.ProvisionPath(uid) if err != nil { - return application.CredentialLocation{}, err + return credentialdomain.Location{}, err } - return application.CredentialLocation{Mount: c.mount, Path: path}, nil + return credentialdomain.Location{Mount: c.mount, Path: path}, nil } -func (c *Credentials) ReadCredential(ctx context.Context, location application.CredentialLocation, version int64) (application.ApplicationCredential, error) { +func (c *Credentials) ReadCredential(ctx context.Context, location credentialdomain.Location, version int64) (credentialdomain.ApplicationCredential, error) { if location.Mount != c.mount { - return application.ApplicationCredential{}, ErrInvalidLocation + return credentialdomain.ApplicationCredential{}, ErrInvalidLocation } if version == 0 { return c.Read(ctx, location.Path) @@ -26,7 +28,7 @@ func (c *Credentials) ReadCredential(ctx context.Context, location application.C return c.ReadConfirmed(ctx, location.Path, version) } -func (c *Credentials) CreateCredential(ctx context.Context, location application.CredentialLocation, credential application.ApplicationCredential) (int64, error) { +func (c *Credentials) CreateCredential(ctx context.Context, location credentialdomain.Location, credential credentialdomain.ApplicationCredential) (int64, error) { if location.Mount != c.mount { return 0, ErrInvalidLocation } diff --git a/internal/database/adapter/openbao/credentials.go b/internal/database/adapter/openbao/credentials.go index 465eca9..52de4c8 100644 --- a/internal/database/adapter/openbao/credentials.go +++ b/internal/database/adapter/openbao/credentials.go @@ -26,6 +26,8 @@ import ( "slices" "strings" + credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" + bao "github.com/openbao/openbao/api/v2" "git.ddupan.top/panxiao81/ayatori/internal/database/application" @@ -80,33 +82,33 @@ func (c *Credentials) accepts(path string) bool { } // Read 只读取调用方已确认关联的路径；成功读取不构成对既有凭据的自动认领。 -func (c *Credentials) Read(ctx context.Context, path string) (application.ApplicationCredential, error) { +func (c *Credentials) Read(ctx context.Context, path string) (credentialdomain.ApplicationCredential, error) { secret, err := c.read(ctx, path) if err != nil { - return application.ApplicationCredential{}, err + return credentialdomain.ApplicationCredential{}, err } - return application.ParseApplicationCredential(secret.Data) + return credentialdomain.ParseApplicationCredential(secret.Data) } // ReadConfirmed 读取最新值并核对已持久化的确认版本，不回退读取历史版本。 // 确认后的删除或改写需要人工处理，不能因此重新生成密码。 -func (c *Credentials) ReadConfirmed(ctx context.Context, path string, version int64) (application.ApplicationCredential, error) { +func (c *Credentials) ReadConfirmed(ctx context.Context, path string, version int64) (credentialdomain.ApplicationCredential, error) { if version < 1 { - return application.ApplicationCredential{}, ErrConflict + return credentialdomain.ApplicationCredential{}, ErrConflict } secret, err := c.read(ctx, path) if errors.Is(err, ErrNotFound) { - return application.ApplicationCredential{}, ErrConflict + return credentialdomain.ApplicationCredential{}, ErrConflict } if err != nil { - return application.ApplicationCredential{}, err + return credentialdomain.ApplicationCredential{}, err } if secret.VersionMetadata == nil || int64(secret.VersionMetadata.Version) != version { - return application.ApplicationCredential{}, ErrConflict + return credentialdomain.ApplicationCredential{}, ErrConflict } - credential, err := application.ParseApplicationCredential(secret.Data) + credential, err := credentialdomain.ParseApplicationCredential(secret.Data) if err != nil { - return application.ApplicationCredential{}, ErrConflict + return credentialdomain.ApplicationCredential{}, ErrConflict } return credential, nil } @@ -130,7 +132,7 @@ func (c *Credentials) read(ctx context.Context, path string) (*bao.KVSecret, err // Create 只创建从未存在过的路径，并验证回读七键与提交值完全一致。 // 任何不确定写入都不返回凭据；上层必须停止供应并持久化冲突，不能重新生成密码。 -func (c *Credentials) Create(ctx context.Context, path string, credential application.ApplicationCredential) error { +func (c *Credentials) Create(ctx context.Context, path string, credential credentialdomain.ApplicationCredential) error { if !c.accepts(path) { return ErrInvalidLocation } diff --git a/internal/database/adapter/openbao/credentials_test.go b/internal/database/adapter/openbao/credentials_test.go index c065716..819bfbf 100644 --- a/internal/database/adapter/openbao/credentials_test.go +++ b/internal/database/adapter/openbao/credentials_test.go @@ -25,10 +25,11 @@ import ( "sync/atomic" "testing" + credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" + bao "github.com/openbao/openbao/api/v2" "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/openbao" - "git.ddupan.top/panxiao81/ayatori/internal/database/application" ) const ( @@ -39,9 +40,9 @@ const ( kvVersionKey = "version" ) -func fixtureCredential(t *testing.T) application.ApplicationCredential { +func fixtureCredential(t *testing.T) credentialdomain.ApplicationCredential { t.Helper() - credential, err := application.ParseApplicationCredential(map[string]any{ + credential, err := credentialdomain.ParseApplicationCredential(map[string]any{ "username": "app_owner", "password": fixturePassword, "database": "app", "host": "postgres.example", "hostaddr": "192.0.2.1", "port": "5432", "sslmode": "verify-full", }) diff --git a/internal/database/adapter/openbao/preparation_concurrency_integration_test.go b/internal/database/adapter/openbao/preparation_concurrency_integration_test.go index ebed94b..0c8cb2c 100644 --- a/internal/database/adapter/openbao/preparation_concurrency_integration_test.go +++ b/internal/database/adapter/openbao/preparation_concurrency_integration_test.go @@ -54,7 +54,7 @@ func testPreparationConcurrency(t *testing.T, f *preparationFixture) { if succeeded != 1 || conflicted != 1 { t.Fatal("同一快照只能有一个用例成功固定位置并继续创建") } - f.status(t, database, 1, application.CredentialPrepared) + f.status(t, database, 1, "CredentialPrepared") stored, err := f.bao.KVv2("secret").Get(t.Context(), database.Status.CredentialRef.Path) if err != nil || stored.VersionMetadata.Version != 1 { t.Fatal("并发准备用例只能产生一个凭据版本") diff --git a/internal/database/adapter/openbao/preparation_fixture_integration_test.go b/internal/database/adapter/openbao/preparation_fixture_integration_test.go index dbf2a86..20a44d1 100644 --- a/internal/database/adapter/openbao/preparation_fixture_integration_test.go +++ b/internal/database/adapter/openbao/preparation_fixture_integration_test.go @@ -80,7 +80,8 @@ func (f *preparationFixture) bound(t *testing.T, name string) (*databasev1alpha1 if err := f.api.Create(t.Context(), tenant); err != nil { t.Fatal(err) } - binder := &databasecontroller.BindingReconciler{Client: f.api, Reader: f.api} + resources := &kubernetes.BindingResources{Client: f.api, Reader: f.api} + binder := databasecontroller.NewBindingReconciler(f.api, &application.BindingService{Resources: resources}, resources) if _, err := binder.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(tenant)}); err != nil { t.Fatal(err) } @@ -106,7 +107,7 @@ func (f *preparationFixture) status(t *testing.T, database *databasev1alpha1.Pos if err := f.api.Get(t.Context(), client.ObjectKeyFromObject(database), database); err != nil { t.Fatal(err) } - condition := meta.FindStatusCondition(database.Status.Conditions, application.CredentialsReady) + condition := meta.FindStatusCondition(database.Status.Conditions, "CredentialsReady") if database.Status.CredentialVersion != version || condition == nil || condition.Reason != reason { t.Fatalf("凭据版本或条件不符：version=%d，期望 reason=%s", database.Status.CredentialVersion, reason) } diff --git a/internal/database/adapter/openbao/preparation_integration_test.go b/internal/database/adapter/openbao/preparation_integration_test.go index 8f5dd0a..79f2242 100644 --- a/internal/database/adapter/openbao/preparation_integration_test.go +++ b/internal/database/adapter/openbao/preparation_integration_test.go @@ -8,6 +8,8 @@ import ( "maps" "testing" + credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" + bao "github.com/openbao/openbao/api/v2" "sigs.k8s.io/controller-runtime/pkg/client" @@ -36,7 +38,7 @@ func testPreparationRestart(t *testing.T, f *preparationFixture) { if err := f.service(t).Reconcile(t.Context(), database.Name); err != nil { t.Fatal(err) } - f.status(t, database, 1, application.CredentialPrepared) + f.status(t, database, 1, "CredentialPrepared") path := database.Status.CredentialRef.Path before, err := f.bao.KVv2("secret").Get(t.Context(), path) if err != nil { @@ -51,12 +53,12 @@ func testPreparationRestart(t *testing.T, f *preparationFixture) { t.Fatal(err) } } - f.status(t, database, 1, application.CredentialPrepared) + f.status(t, database, 1, "CredentialPrepared") revision := database.ResourceVersion if err := f.service(t).Reconcile(t.Context(), database.Name); err != nil { t.Fatal(err) } - f.status(t, database, 1, application.CredentialPrepared) + f.status(t, database, 1, "CredentialPrepared") if database.ResourceVersion != revision { t.Fatal("幂等重试不应改写 status") } @@ -135,7 +137,7 @@ func testPreparationLostConfirmation(t *testing.T, f *preparationFixture) { if err := service.Reconcile(t.Context(), database.Name); err == nil { t.Fatal("确认写入失败应返回 API 错误") } - f.status(t, database, 0, application.CredentialCreationStarted) + f.status(t, database, 0, "CreationStarted") if err := f.service(t).Reconcile(t.Context(), database.Name); err != nil { t.Fatal(err) } @@ -151,7 +153,7 @@ type afterCreateStore struct { after func() error } -func (s afterCreateStore) CreateCredential(ctx context.Context, location application.CredentialLocation, credential application.ApplicationCredential) (int64, error) { +func (s afterCreateStore) CreateCredential(ctx context.Context, location credentialdomain.Location, credential credentialdomain.ApplicationCredential) (int64, error) { version, err := s.CredentialStore.CreateCredential(ctx, location, credential) if err != nil { return 0, err @@ -185,7 +187,7 @@ func testPreparationChangedBinding(t *testing.T, f *preparationFixture) { if err := service.Reconcile(t.Context(), database.Name); err == nil { t.Fatal("中途删除 Tenant 后不得确认凭据") } - f.status(t, database, 0, application.CredentialCreationStarted) + f.status(t, database, 0, "CreationStarted") if err := f.service(t).Reconcile(t.Context(), database.Name); err != nil { t.Fatal(err) } @@ -205,7 +207,7 @@ func testPreparationDependencies(t *testing.T, f *preparationFixture) { if err := f.service(t).Reconcile(t.Context(), database.Name); err != nil { t.Fatal(err) } - f.status(t, database, 1, application.CredentialPrepared) + f.status(t, database, 1, "CredentialPrepared") if err := service.Reconcile(t.Context(), database.Name); err != nil { t.Fatal(err) } @@ -213,7 +215,7 @@ func testPreparationDependencies(t *testing.T, f *preparationFixture) { if err := f.service(t).Reconcile(t.Context(), database.Name); err != nil { t.Fatal(err) } - f.status(t, database, 1, application.CredentialPrepared) + f.status(t, database, 1, "CredentialPrepared") moved, err := openbao.NewCredentials(f.bao, "other", "elsewhere") if err != nil { t.Fatal(err) diff --git a/internal/database/adapter/openbao/preparation_pruning_integration_test.go b/internal/database/adapter/openbao/preparation_pruning_integration_test.go index 9e7e022..396c63e 100644 --- a/internal/database/adapter/openbao/preparation_pruning_integration_test.go +++ b/internal/database/adapter/openbao/preparation_pruning_integration_test.go @@ -12,7 +12,6 @@ import ( databasev1alpha1 "git.ddupan.top/panxiao81/ayatori/api/database/v1alpha1" "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes" - "git.ddupan.top/panxiao81/ayatori/internal/database/application" ) // 在 API 边界模拟旧 schema 裁剪位置字段；剩余写入仍由真实 API server 处理。 @@ -48,5 +47,5 @@ func testPreparationPruning(t *testing.T, f *preparationFixture) { if err := f.service(t).Reconcile(t.Context(), database.Name); err != nil { t.Fatal(err) } - f.status(t, database, 1, application.CredentialPrepared) + f.status(t, database, 1, "CredentialPrepared") } diff --git a/internal/database/adapter/openbao/preparation_watch_integration_test.go b/internal/database/adapter/openbao/preparation_watch_integration_test.go index bf2e886..2bcf5bb 100644 --- a/internal/database/adapter/openbao/preparation_watch_integration_test.go +++ b/internal/database/adapter/openbao/preparation_watch_integration_test.go @@ -7,6 +7,9 @@ import ( "testing" "time" + "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes" + "git.ddupan.top/panxiao81/ayatori/internal/database/application" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ctrl "sigs.k8s.io/controller-runtime" @@ -15,7 +18,6 @@ import ( metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" databasev1alpha1 "git.ddupan.top/panxiao81/ayatori/api/database/v1alpha1" - "git.ddupan.top/panxiao81/ayatori/internal/database/application" databasecontroller "git.ddupan.top/panxiao81/ayatori/internal/database/controller" ) @@ -26,7 +28,7 @@ func testPreparationWatch(t *testing.T, f *preparationFixture) { return } if err := f.api.Get(context.Background(), client.ObjectKeyFromObject(database), database); err == nil { - if condition := meta.FindStatusCondition(database.Status.Conditions, application.CredentialsReady); condition != nil { + if condition := meta.FindStatusCondition(database.Status.Conditions, "CredentialsReady"); condition != nil { t.Logf("失败时凭据条件: %s: %s", condition.Reason, condition.Message) } } @@ -50,7 +52,7 @@ func testPreparationWatch(t *testing.T, f *preparationFixture) { if err := f.api.Get(t.Context(), client.ObjectKeyFromObject(database), database); err != nil { return false } - condition := meta.FindStatusCondition(database.Status.Conditions, application.CredentialsReady) + condition := meta.FindStatusCondition(database.Status.Conditions, "CredentialsReady") return condition != nil && condition.Reason == "DependencyUnavailable" }) if database.Status.CredentialRef != nil { @@ -69,7 +71,7 @@ func testPreparationWatch(t *testing.T, f *preparationFixture) { if err := f.api.Get(t.Context(), client.ObjectKeyFromObject(database), database); err != nil { return false } - condition := meta.FindStatusCondition(database.Status.Conditions, application.CredentialsReady) + condition := meta.FindStatusCondition(database.Status.Conditions, "CredentialsReady") return condition != nil && condition.Status == metav1.ConditionFalse && condition.Reason == "DependencyUnavailable" }) setReady(metav1.ConditionTrue) @@ -77,7 +79,7 @@ func testPreparationWatch(t *testing.T, f *preparationFixture) { if err := f.api.Get(t.Context(), client.ObjectKeyFromObject(database), database); err != nil { return false } - condition := meta.FindStatusCondition(database.Status.Conditions, application.CredentialsReady) + condition := meta.FindStatusCondition(database.Status.Conditions, "CredentialsReady") return condition != nil && condition.Status == metav1.ConditionTrue && database.Status.CredentialVersion == 1 }) stored, err := f.bao.KVv2("secret").Get(t.Context(), database.Status.CredentialRef.Path) @@ -96,10 +98,14 @@ func startPreparationManager(t *testing.T, f *preparationFixture) func() { if err != nil { t.Fatal(err) } - if err := (&databasecontroller.BindingReconciler{}).SetupWithManager(t.Context(), manager); err != nil { + bindingResources := &kubernetes.BindingResources{Client: manager.GetClient(), Reader: manager.GetAPIReader()} + binder := databasecontroller.NewBindingReconciler(manager.GetClient(), &application.BindingService{Resources: bindingResources}, bindingResources) + if err := binder.SetupWithManager(t.Context(), manager); err != nil { t.Fatal(err) } - if err := (&databasecontroller.CredentialReconciler{Store: fixtureStore(t, f.bao)}).SetupWithManager(manager); err != nil { + credentialResources := &kubernetes.CredentialResources{Client: manager.GetClient(), Reader: manager.GetAPIReader()} + preparation := &application.CredentialPreparation{Resources: credentialResources, Store: fixtureStore(t, f.bao)} + if err := databasecontroller.NewCredentialReconciler(manager.GetClient(), preparation).SetupWithManager(manager); err != nil { t.Fatal(err) } ctx, cancel := context.WithCancel(t.Context()) diff --git a/internal/database/adapter/postgresql/instance_controller_integration_test.go b/internal/database/adapter/postgresql/instance_controller_integration_test.go index c6e9d57..a21588d 100644 --- a/internal/database/adapter/postgresql/instance_controller_integration_test.go +++ b/internal/database/adapter/postgresql/instance_controller_integration_test.go @@ -68,7 +68,9 @@ func TestInstanceControllerWithRealPostgreSQL(t *testing.T) { if err != nil { t.Fatal(err) } - reconciler := &databasecontroller.InstanceReconciler{Observer: service, SecretNamespace: controllerNamespace} + resources := &secretadapter.InstanceResources{Client: manager.GetClient(), Reader: manager.GetAPIReader()} + usecase := &application.InstanceReconciliation{Resources: resources, Observer: service} + reconciler := databasecontroller.NewInstanceReconciler(manager.GetClient(), usecase, resources, controllerNamespace) if err := reconciler.SetupWithManager(manager); err != nil { t.Fatal(err) } diff --git a/internal/database/application/credential_preparation.go b/internal/database/application/credential_preparation.go index 7444c2a..4aef435 100644 --- a/internal/database/application/credential_preparation.go +++ b/internal/database/application/credential_preparation.go @@ -5,8 +5,7 @@ import ( "errors" "fmt" - "git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding" - "git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance" + credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" ) var ( @@ -17,54 +16,28 @@ var ( ErrCredentialUncertain = errors.New("credential creation outcome is uncertain; manual resolution required") ) -const ( - CredentialsReady = "CredentialsReady" - CredentialCreationStarted = "CreationStarted" - CredentialPrepared = "CredentialPrepared" -) - -type CredentialLocation struct { - Mount string - Path string -} - // CredentialStore 只表达本用例需要的凭据操作，不提供覆盖或删除。 // version=0 的读取只用于观察是否已有值，成功不能作为认领依据。 type CredentialStore interface { - ProvisionLocation(string) (CredentialLocation, error) - ReadCredential(context.Context, CredentialLocation, int64) (ApplicationCredential, error) - CreateCredential(context.Context, CredentialLocation, ApplicationCredential) (int64, error) -} - -type CredentialInstance struct { - binding.Instance - Generation int64 - Endpoint instance.Endpoint + ProvisionLocation(string) (credentialdomain.Location, error) + ReadCredential(context.Context, credentialdomain.Location, int64) (credentialdomain.ApplicationCredential, error) + CreateCredential(context.Context, credentialdomain.Location, credentialdomain.ApplicationCredential) (int64, error) } // CredentialRecord 是同一轮观察的事实，状态中永远不保存密码。 type CredentialRecord struct { - Database BindingDatabase - Tenant *BindingTenant - Instance *CredentialInstance - DatabaseProtected bool - TenantProtected bool - Status CredentialStatus -} - -type CredentialStatus struct { - Location *CredentialLocation - Version int64 - Ready bool - Reason string - Message string + credentialdomain.Target + Revision string + TenantGeneration int64 + InstanceGeneration int64 + Status credentialdomain.State } // CredentialResources 的写入必须检查 Database UID/resourceVersion，保留其他状态。 // CheckCurrent 在外部操作前后回读本轮三个资源，拒绝陈旧快照；它不是跨系统事务。 type CredentialResources interface { Load(context.Context, string) (*CredentialRecord, error) - Save(context.Context, *CredentialRecord, CredentialStatus) (*CredentialRecord, error) + Save(context.Context, *CredentialRecord, credentialdomain.State) (*CredentialRecord, error) CheckCurrent(context.Context, *CredentialRecord) error } @@ -75,37 +48,35 @@ type CredentialPreparation struct { func (s CredentialPreparation) Reconcile(ctx context.Context, name string) error { record, err := s.Resources.Load(ctx, name) - if err != nil || record == nil || record.Database.Source != "Provision" { + if err != nil || record == nil || !record.RequiresPreparation() { return err } - // 未完成创建的重入不猜测后端结果。即使进程在实际发请求前退出，也需要人工核实。 - if record.Status.Version == 0 && record.Status.Reason == binding.Conflict { - return nil // 保留首次冲突的具体原因，不因后端恢复而重入创建。 + if state, canContinue := record.Status.Resume(); !canContinue { + if state == record.Status { + return nil + } + return s.report(ctx, record, state.Phase, state.Message) } - if record.Status.Version == 0 && record.Status.Reason == CredentialCreationStarted { - return s.report(ctx, record, binding.Conflict, - "凭据创建未留下成功确认；请核对固定位置与后端历史并人工处理，未重新生成密码") - } - if issue := record.check(); issue != nil { - return s.report(ctx, record, issue.Reason, issue.Message) + if issue := record.Check(); issue != nil { + return s.report(ctx, record, issue.Phase, issue.Message) } location, err := s.Store.ProvisionLocation(record.Database.Identity.UID) if err != nil { - return s.report(ctx, record, binding.DependencyUnavailable, "凭据存储位置配置无效，未执行外部写入") + return s.report(ctx, record, credentialdomain.Unavailable, "凭据存储位置配置无效，未执行外部写入") + } + if issue := record.Status.CheckLocation(location); issue != nil { + return s.report(ctx, record, issue.Phase, issue.Message) } if record.Status.Location == nil { status := record.Status status.Location = &location - status.Ready, status.Reason, status.Message = false, "LocationPinned", "凭据位置已固定，等待创建" + status = status.WithPhase(credentialdomain.Pinned, "凭据位置已固定，等待创建") record, err = s.Resources.Save(ctx, record, status) if err != nil { return err } - } else if *record.Status.Location != location { - return s.report(ctx, record, binding.DependencyUnavailable, - "部署配置与固定凭据位置不一致；请恢复原 mount/path 配置，未迁移或改密") } - if record.Status.Version > 0 { + if record.Status.Confirmed() { return s.observe(ctx, record) } return s.create(ctx, record) @@ -113,19 +84,18 @@ func (s CredentialPreparation) Reconcile(ctx context.Context, name string) error func (s CredentialPreparation) create(ctx context.Context, record *CredentialRecord) error { _, err := s.Store.ReadCredential(ctx, *record.Status.Location, 0) - if err == nil || errors.Is(err, ErrApplicationCredentialInvalid) { - return s.report(ctx, record, binding.Conflict, "固定位置已有未确认的凭据；请人工核实，未认领或覆盖") + if err == nil || errors.Is(err, credentialdomain.ErrApplicationCredentialInvalid) { + return s.report(ctx, record, credentialdomain.Conflict, "固定位置已有未确认的凭据；请人工核实，未认领或覆盖") } if !errors.Is(err, ErrCredentialNotFound) { - return s.report(ctx, record, binding.DependencyUnavailable, "创建前无法确认凭据位置是否为空，等待依赖恢复") + return s.report(ctx, record, credentialdomain.Unavailable, "创建前无法确认凭据位置是否为空，等待依赖恢复") } - credential, err := GenerateApplicationCredential(record.Database.LoginRole, record.Database.Name, record.Instance.Endpoint) + credential, err := credentialdomain.GenerateApplicationCredential(record.Database.LoginRole, record.Database.Name, record.Instance.Endpoint) if err != nil { - return s.report(ctx, record, "InvalidTarget", "应用凭据目标无效，未执行外部写入") + return s.report(ctx, record, credentialdomain.InvalidTarget, "应用凭据目标无效，未执行外部写入") } status := record.Status - status.Ready, status.Reason = false, CredentialCreationStarted - status.Message = "凭据创建已开始；尚无成功确认时不得重入创建" + status = status.WithPhase(credentialdomain.Creating, "凭据创建已开始；尚无成功确认时不得重入创建") record, err = s.Resources.Save(ctx, record, status) if err != nil { return err @@ -136,46 +106,47 @@ func (s CredentialPreparation) create(ctx context.Context, record *CredentialRec version, err := s.Store.CreateCredential(ctx, *record.Status.Location, credential) if errors.Is(err, ErrCredentialUnavailable) { // 适配器只在明确未执行写入（认证拒绝或请求前取消）时返回此错误。 - return s.report(ctx, record, binding.DependencyUnavailable, "凭据创建在执行前被拒绝，等待认证或权限恢复") + return s.report(ctx, record, credentialdomain.Unavailable, "凭据创建在执行前被拒绝，等待认证或权限恢复") } - if err != nil || version != 1 { - return s.report(ctx, record, binding.Conflict, + if err != nil { + return s.report(ctx, record, credentialdomain.Conflict, "凭据创建冲突或结果不确定；请核对固定位置的版本历史，未认领、覆盖或重新生成密码") } + confirmed, issue := record.Status.Created(version) + if issue != nil { + return s.report(ctx, record, issue.Phase, issue.Message) + } if err := s.Resources.CheckCurrent(ctx, record); err != nil { return err } - status = record.Status - status.Version, status.Ready, status.Reason = version, true, CredentialPrepared - status.Message = "凭据已创建并回读确认；尚未创建 PostgreSQL 资源或交付给 Tenant" - _, err = s.Resources.Save(ctx, record, status) + _, err = s.Resources.Save(ctx, record, confirmed) return err } func (s CredentialPreparation) observe(ctx context.Context, record *CredentialRecord) error { credential, err := s.Store.ReadCredential(ctx, *record.Status.Location, record.Status.Version) if errors.Is(err, ErrCredentialConflict) || errors.Is(err, ErrCredentialNotFound) { - return s.report(ctx, record, binding.Conflict, "已确认凭据消失、版本变化或内容无效；请人工核实，未生成替代密码") + return s.report(ctx, record, credentialdomain.Conflict, "已确认凭据消失、版本变化或内容无效；请人工核实，未生成替代密码") } if err != nil { - return s.report(ctx, record, binding.DependencyUnavailable, "已确认凭据暂时无法读取；保留确认版本，等待依赖恢复") + return s.report(ctx, record, credentialdomain.Unavailable, "已确认凭据暂时无法读取；保留确认版本，等待依赖恢复") } - if !credential.MatchesTarget(record.Database.LoginRole, record.Database.Name, record.Instance.Endpoint) { - return s.report(ctx, record, binding.Conflict, "已确认凭据与当前 Instance/database/loginRole 不一致；请人工核实，未修改凭据") + if issue := record.CheckCredential(credential); issue != nil { + return s.report(ctx, record, issue.Phase, issue.Message) } if err := s.Resources.CheckCurrent(ctx, record); err != nil { return err } status := record.Status - status.Ready, status.Reason = true, CredentialPrepared + status.Phase = credentialdomain.Prepared status.Message = "已确认凭据可读取；尚未验证 PostgreSQL 资源或完成 Tenant 交付" _, err = s.Resources.Save(ctx, record, status) return err } -func (s CredentialPreparation) report(ctx context.Context, record *CredentialRecord, reason, message string) error { +func (s CredentialPreparation) report(ctx context.Context, record *CredentialRecord, phase credentialdomain.Phase, message string) error { status := record.Status - status.Ready, status.Reason = false, reason + status.Phase = phase status.Message = fmt.Sprintf("Database %s：%s", record.Database.Identity.Name, message) _, err := s.Resources.Save(ctx, record, status) return err diff --git a/internal/database/application/credential_preparation_test.go b/internal/database/application/credential_preparation_test.go index 7906374..8d79644 100644 --- a/internal/database/application/credential_preparation_test.go +++ b/internal/database/application/credential_preparation_test.go @@ -5,6 +5,8 @@ import ( "errors" "testing" + credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" + "git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding" "git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance" ) @@ -25,14 +27,14 @@ func preparationRecord(t *testing.T) *CredentialRecord { t.Fatal(err) } return &CredentialRecord{ - Database: BindingDatabase{Database: binding.Database{ + Database: binding.Database{ Identity: database, Instance: preparationInstanceName, InstanceUID: preparationInstanceUID, Name: bindingTestName, LoginRole: bindingTestName, Source: "Provision", Tenant: &tenant, - }}, - Tenant: &BindingTenant{ + }, + Tenant: &binding.Tenant{ Identity: tenant, Phase: binding.Bound, Database: &database, Request: binding.Request{Provision: &binding.ProvisionRequest{Instance: preparationInstanceName}}, }, - Instance: &CredentialInstance{Identity: binding.Identity{Name: preparationInstanceName, UID: preparationInstanceUID}, Ready: true, Endpoint: endpoint}, + Instance: &credentialdomain.Instance{Identity: binding.Identity{Name: preparationInstanceName, UID: preparationInstanceUID}, Ready: true, Endpoint: endpoint}, DatabaseProtected: true, TenantProtected: true, } } @@ -48,7 +50,7 @@ func (r *memoryCredentialResources) Load(context.Context, string) (*CredentialRe return ©, nil } -func (r *memoryCredentialResources) Save(_ context.Context, record *CredentialRecord, status CredentialStatus) (*CredentialRecord, error) { +func (r *memoryCredentialResources) Save(_ context.Context, record *CredentialRecord, status credentialdomain.State) (*CredentialRecord, error) { if r.saveError != nil { return nil, r.saveError } @@ -69,16 +71,16 @@ type preparationStore struct { createError error } -func (*preparationStore) ProvisionLocation(uid string) (CredentialLocation, error) { - return CredentialLocation{Mount: "applications", Path: "database/" + uid}, nil +func (*preparationStore) ProvisionLocation(uid string) (credentialdomain.Location, error) { + return credentialdomain.Location{Mount: "applications", Path: "database/" + uid}, nil } -func (s *preparationStore) ReadCredential(context.Context, CredentialLocation, int64) (ApplicationCredential, error) { +func (s *preparationStore) ReadCredential(context.Context, credentialdomain.Location, int64) (credentialdomain.ApplicationCredential, error) { s.reads++ - return ApplicationCredential{}, s.readError + return credentialdomain.ApplicationCredential{}, s.readError } -func (s *preparationStore) CreateCredential(context.Context, CredentialLocation, ApplicationCredential) (int64, error) { +func (s *preparationStore) CreateCredential(context.Context, credentialdomain.Location, credentialdomain.ApplicationCredential) (int64, error) { s.creates++ if s.createError != nil { return 0, s.createError @@ -159,7 +161,7 @@ func TestCredentialPreparationWriteBoundary(t *testing.T) { if err := service.Reconcile(t.Context(), resources.record.Database.Identity.Name); err != nil { t.Fatal(err) } - if store.creates != test.wantCreates || resources.record.Status.Reason != binding.Conflict { + if store.creates != test.wantCreates || resources.record.Status.Phase != credentialdomain.Conflict { t.Fatal("未确认创建重入时不得生成替代密码") } } diff --git a/internal/database/application/credential_target.go b/internal/database/application/credential_target.go deleted file mode 100644 index 02ca549..0000000 --- a/internal/database/application/credential_target.go +++ /dev/null @@ -1,27 +0,0 @@ -package application - -import "git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding" - -func (r *CredentialRecord) check() *binding.Issue { - database := r.Database - if database.Deleting || database.Phase == binding.Deleting || database.Phase == "Released" { - return &binding.Issue{Reason: "PreparationStopped", Message: "Database 正在删除或已释放；保留凭据与 finalizer，不执行供应或清理"} - } - if database.Tenant == nil || r.Tenant == nil || r.Tenant.Database == nil { - return &binding.Issue{Reason: binding.DependencyUnavailable, Message: "等待 Database 与 Tenant 双向绑定完成"} - } - if *database.Tenant != r.Tenant.Identity || *r.Tenant.Database != database.Identity { - return &binding.Issue{Reason: binding.Conflict, Message: "双向绑定的名称或 UID 不匹配，未创建凭据"} - } - if r.Tenant.Deleting || r.Tenant.Phase != binding.Bound || !r.DatabaseProtected || !r.TenantProtected { - return &binding.Issue{Reason: "PreparationStopped", Message: "Tenant 未完成绑定、正在删除或缺少 finalizer 保护，未创建凭据"} - } - target, err := r.Tenant.Request.Resolve(r.Tenant.Identity) - if err != nil || (target.Provision != nil && !database.MatchesProvision(target, r.Tenant.Identity)) || target.Name != database.Identity.Name { - return &binding.Issue{Reason: binding.Conflict, Message: "Tenant 申请与 Database 目标不一致，未创建凭据"} - } - if r.Instance == nil || database.InstanceUID == "" { - return &binding.Issue{Reason: binding.DependencyUnavailable, Message: "等待 Instance 与已记录的实例身份"} - } - return r.Instance.Check(&database.Database) -} diff --git a/internal/database/controller/binding_controller.go b/internal/database/controller/binding_controller.go index 0dff332..e4a2f3c 100644 --- a/internal/database/controller/binding_controller.go +++ b/internal/database/controller/binding_controller.go @@ -5,7 +5,6 @@ import ( "context" "time" - "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes" "git.ddupan.top/panxiao81/ayatori/internal/database/application" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" @@ -14,8 +13,17 @@ import ( const dependencyRetry = 30 * time.Second type BindingReconciler struct { - Client client.Client - Reader client.Reader + Client client.Client + Service *application.BindingService + Presenter BindingPresenter +} + +type BindingPresenter interface { + Present(context.Context, application.BindingResult) error +} + +func NewBindingReconciler(cache client.Client, service *application.BindingService, presenter BindingPresenter) *BindingReconciler { + return &BindingReconciler{Client: cache, Service: service, Presenter: presenter} } // +kubebuilder:rbac:groups=database.ayatori.ddupan.top,resources=postgresqltenants,verbs=get;list;watch;update;patch @@ -27,13 +35,11 @@ type BindingReconciler struct { // +kubebuilder:rbac:groups=database.ayatori.ddupan.top,resources=postgresqlinstances,verbs=get;list;watch func (r *BindingReconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.Result, error) { - resources := &kubernetes.BindingResources{Client: r.Client, Reader: r.Reader} - service := application.BindingService{Resources: resources} - result, err := service.Reconcile(ctx, request.Namespace, request.Name) + result, err := r.Service.Reconcile(ctx, request.Namespace, request.Name) if err != nil { return ctrl.Result{}, err } - if err := resources.Present(ctx, result); err != nil { + if err := r.Presenter.Present(ctx, result); err != nil { return ctrl.Result{}, err } if result.RetrySoon { diff --git a/internal/database/controller/binding_controller_test.go b/internal/database/controller/binding_controller_test.go index cf1ac91..ced0eca 100644 --- a/internal/database/controller/binding_controller_test.go +++ b/internal/database/controller/binding_controller_test.go @@ -42,6 +42,11 @@ func targetDatabaseName(tenant *databasev1alpha1.PostgreSQLTenant) string { return "tenant-" + string(tenant.UID) } +func bindingTestReconciler(writer client.Client, reader client.Reader) *BindingReconciler { + resources := &kubernetes.BindingResources{Client: writer, Reader: reader} + return NewBindingReconciler(writer, &application.BindingService{Resources: resources}, resources) +} + func tenantReference(tenant *databasev1alpha1.PostgreSQLTenant) *databasev1alpha1.TenantReference { return &databasev1alpha1.TenantReference{ Namespace: tenant.Namespace, Name: databasev1alpha1.ObjectName(tenant.Name), UID: tenant.UID, @@ -103,7 +108,7 @@ func testDynamicBinding(t *testing.T, apiClient client.Client) { instance := readyInstance(t, apiClient, "dynamic-instance") tenant := provisionTenant("dynamic", instance.Name) requireCreate(t, apiClient, tenant) - reconciler := &BindingReconciler{Client: apiClient, Reader: apiClient} + reconciler := bindingTestReconciler(apiClient, apiClient) reconcileOK(t, reconciler, tenant) reload(t, apiClient, tenant) if tenant.Status.DatabaseRef == nil || tenant.Status.Phase != phaseBound { @@ -154,7 +159,7 @@ func testBindingRestart(t *testing.T, apiClient client.Client) { instance := readyInstance(t, apiClient, "restart-instance") tenant := provisionTenant("restart", instance.Name) requireCreate(t, apiClient, tenant) - first := &BindingReconciler{Client: &failedTenantStatusClient{Client: apiClient}, Reader: apiClient} + first := bindingTestReconciler(&failedTenantStatusClient{Client: apiClient}, apiClient) if _, err := first.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(tenant)}); err == nil { t.Fatal("预期第二次绑定写入失败") } @@ -169,7 +174,7 @@ func testBindingRestart(t *testing.T, apiClient client.Client) { t.Fatal("失败后资源侧绑定不应回滚") } // 新建 reconciler，无旧内存，只从 API 中读取进度。 - restarted := &BindingReconciler{Client: apiClient, Reader: apiClient} + restarted := bindingTestReconciler(apiClient, apiClient) reconcileOK(t, restarted, tenant) reload(t, apiClient, tenant) if tenant.Status.DatabaseRef == nil || tenant.Status.DatabaseRef.UID != database.UID { @@ -190,7 +195,7 @@ func testConcurrentBinding(t *testing.T, apiClient client.Client) { results := make(chan error, len(tenants)) for _, tenant := range tenants { workers.Go(func() { - reconciler := &BindingReconciler{Client: apiClient, Reader: apiClient} + reconciler := bindingTestReconciler(apiClient, apiClient) _, err := reconciler.Reconcile(t.Context(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(tenant)}) results <- err }) @@ -202,7 +207,7 @@ func testConcurrentBinding(t *testing.T, apiClient client.Client) { t.Fatalf("并发协调出现非版本冲突错误: %v", err) } } - reconciler := &BindingReconciler{Client: apiClient, Reader: apiClient} + reconciler := bindingTestReconciler(apiClient, apiClient) bound := 0 for _, tenant := range tenants { reconcileOK(t, reconciler, tenant) @@ -233,7 +238,7 @@ func testBindingIdentity(t *testing.T, apiClient client.Client) { } tenant := existingTenant("identity", database.Name) requireCreate(t, apiClient, tenant) - reconciler := &BindingReconciler{Client: apiClient, Reader: apiClient} + reconciler := bindingTestReconciler(apiClient, apiClient) reconcileOK(t, reconciler, tenant) reload(t, apiClient, tenant) assertNotReady(t, tenant, reasonConflict) @@ -256,7 +261,7 @@ func testBindingIdentity(t *testing.T, apiClient client.Client) { func testBindingProtection(t *testing.T, apiClient client.Client) { tenant := provisionTenant("protection", "missing-instance") requireCreate(t, apiClient, tenant) - reconciler := &BindingReconciler{Client: apiClient, Reader: apiClient} + reconciler := bindingTestReconciler(apiClient, apiClient) reconcileOK(t, reconciler, tenant) reload(t, apiClient, tenant) assertNotReady(t, tenant, reasonDependency) @@ -292,7 +297,7 @@ func testStaleObservation(t *testing.T, apiClient client.Client) { } tenant := existingTenant("stale", database.Name) requireCreate(t, apiClient, tenant) - reconciler := &BindingReconciler{Client: apiClient, Reader: apiClient} + reconciler := bindingTestReconciler(apiClient, apiClient) reconcileOK(t, reconciler, tenant) reload(t, apiClient, tenant) assertNotReady(t, tenant, reasonDependency) @@ -318,7 +323,7 @@ func testBindingWatch(t *testing.T, apiClient client.Client, config *rest.Config if err != nil { t.Fatal(err) } - reconciler := &BindingReconciler{} + reconciler := bindingTestReconciler(manager.GetClient(), manager.GetAPIReader()) if err := reconciler.SetupWithManager(t.Context(), manager); err != nil { t.Fatal(err) } @@ -373,7 +378,7 @@ func testPresentationVersion(t *testing.T, apiClient client.Client) { if err := resources.Present(t.Context(), result); !apierrors.IsConflict(err) { t.Fatalf("过期结果呈现 = %v, want Conflict", err) } - reconcileOK(t, &BindingReconciler{Client: apiClient, Reader: apiClient}, tenant) + reconcileOK(t, bindingTestReconciler(apiClient, apiClient), tenant) reload(t, apiClient, tenant) if tenant.Status.Phase != phaseBound || tenant.Spec.SecretName != "updated-delivery" || tenant.Annotations["example.test/keep"] != "preserved" { diff --git a/internal/database/controller/binding_setup.go b/internal/database/controller/binding_setup.go index 6fa70f1..ac2148f 100644 --- a/internal/database/controller/binding_setup.go +++ b/internal/database/controller/binding_setup.go @@ -2,9 +2,10 @@ package controller import ( "context" + "errors" databasev1alpha1 "git.ddupan.top/panxiao81/ayatori/api/database/v1alpha1" - "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes" + "git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding" "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" @@ -14,16 +15,16 @@ import ( const targetDatabaseIndex = "database.bindingTarget" func (r *BindingReconciler) SetupWithManager(ctx context.Context, manager ctrl.Manager) error { - if r.Client == nil { - r.Client = manager.GetClient() - } - if r.Reader == nil { - r.Reader = manager.GetAPIReader() + if r.Client == nil || r.Service == nil || r.Presenter == nil { + return errors.New("binding controller requires injected client, use case and presenter") } if err := manager.GetFieldIndexer().IndexField(ctx, &databasev1alpha1.PostgreSQLTenant{}, targetDatabaseIndex, func(object client.Object) []string { tenant := object.(*databasev1alpha1.PostgreSQLTenant) - return []string{kubernetes.BindingTargetName(tenant)} + if tenant.Spec.DatabaseRef != nil { + return []string{string(tenant.Spec.DatabaseRef.Name)} + } + return []string{binding.DynamicDatabaseName(string(tenant.UID))} }); err != nil { return err } diff --git a/internal/database/controller/credential_controller.go b/internal/database/controller/credential_controller.go index ec87f4f..a3d66a7 100644 --- a/internal/database/controller/credential_controller.go +++ b/internal/database/controller/credential_controller.go @@ -2,9 +2,9 @@ package controller import ( "context" + "errors" databasev1alpha1 "git.ddupan.top/panxiao81/ayatori/api/database/v1alpha1" - "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes" "git.ddupan.top/panxiao81/ayatori/internal/database/application" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" @@ -13,16 +13,16 @@ import ( // CredentialReconciler 只连接事件、用例和重试，不在控制器中编排凭据写入。 type CredentialReconciler struct { - Client client.Client - Reader client.Reader - Store application.CredentialStore + Client client.Client + Service *application.CredentialPreparation +} + +func NewCredentialReconciler(cache client.Client, service *application.CredentialPreparation) *CredentialReconciler { + return &CredentialReconciler{Client: cache, Service: service} } func (r *CredentialReconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.Result, error) { - service := application.CredentialPreparation{ - Resources: &kubernetes.CredentialResources{Client: r.Client, Reader: r.Reader}, Store: r.Store, - } - if err := service.Reconcile(ctx, request.Name); err != nil { + if err := r.Service.Reconcile(ctx, request.Name); err != nil { return ctrl.Result{}, err } // Bao 的可用性和版本变化没有 Kubernetes watch；与已有依赖重查保持一致。 @@ -30,11 +30,8 @@ func (r *CredentialReconciler) Reconcile(ctx context.Context, request ctrl.Reque } func (r *CredentialReconciler) SetupWithManager(manager ctrl.Manager) error { - if r.Client == nil { - r.Client = manager.GetClient() - } - if r.Reader == nil { - r.Reader = manager.GetAPIReader() + if r.Client == nil || r.Service == nil { + return errors.New("credential controller requires injected client and use case") } return ctrl.NewControllerManagedBy(manager). Named("database-credentials").For(&databasev1alpha1.PostgreSQLDatabase{}). diff --git a/internal/database/controller/instance_controller.go b/internal/database/controller/instance_controller.go index ff20446..e2c6115 100644 --- a/internal/database/controller/instance_controller.go +++ b/internal/database/controller/instance_controller.go @@ -4,7 +4,6 @@ import ( "context" "time" - "git.ddupan.top/panxiao81/ayatori/internal/database/adapter/kubernetes" "git.ddupan.top/panxiao81/ayatori/internal/database/application" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" @@ -12,27 +11,33 @@ import ( type InstanceReconciler struct { Client client.Client - Reader client.Reader - Observer application.InstanceObserver + Service *application.InstanceReconciliation + Presenter InstancePresenter SecretNamespace string } +type InstancePresenter interface { + PresentInstance(context.Context, application.InstanceResult) error +} + +func NewInstanceReconciler(cache client.Client, service *application.InstanceReconciliation, presenter InstancePresenter, namespace string) *InstanceReconciler { + return &InstanceReconciler{Client: cache, Service: service, Presenter: presenter, SecretNamespace: namespace} +} + // +kubebuilder:rbac:groups=database.ayatori.ddupan.top,resources=postgresqlinstances,verbs=get;list;watch;update;patch // +kubebuilder:rbac:groups=database.ayatori.ddupan.top,resources=postgresqlinstances/status,verbs=get;update;patch // +kubebuilder:rbac:groups=database.ayatori.ddupan.top,resources=postgresqlinstances/finalizers,verbs=update // Secret 权限单独声明为 namespace Role，不放入生成的 ClusterRole。 func (r *InstanceReconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.Result, error) { - resources := &kubernetes.InstanceResources{Client: r.Client, Reader: r.Reader} - service := application.InstanceReconciliation{Resources: resources, Observer: r.Observer} observationContext, cancel := context.WithTimeout(ctx, 15*time.Second) defer cancel() - result, err := service.Reconcile(observationContext, request.Name) + result, err := r.Service.Reconcile(observationContext, request.Name) if err != nil { return ctrl.Result{}, err } // 查询超时后仍用 worker context 保存安全失败结果；manager 停止时不强行写入。 - if err := resources.PresentInstance(ctx, result); err != nil { + if err := r.Presenter.PresentInstance(ctx, result); err != nil { return ctrl.Result{}, err } if result.Record == nil || result.RemoveProtection { diff --git a/internal/database/controller/instance_controller_test.go b/internal/database/controller/instance_controller_test.go index 1c4479e..489a0c9 100644 --- a/internal/database/controller/instance_controller_test.go +++ b/internal/database/controller/instance_controller_test.go @@ -52,7 +52,8 @@ func newInstanceReconciler(t *testing.T, apiClient client.Client, backend *insta t.Fatal(err) } t.Cleanup(service.Close) - return &InstanceReconciler{Client: apiClient, Reader: apiClient, Observer: service} + resources := &kubernetes.InstanceResources{Client: apiClient, Reader: apiClient} + return NewInstanceReconciler(apiClient, &application.InstanceReconciliation{Resources: resources, Observer: service}, resources, "") } func reconcileInstance(t *testing.T, reconciler *InstanceReconciler, object *databasev1alpha1.PostgreSQLInstance) { @@ -164,11 +165,11 @@ func TestInstanceDeletionProtection(t *testing.T) { t.Fatal(err) } backend.inspect = func() { t.Fatal("删除中不应连接 PostgreSQL") } - reconciler.Reader = &failedReferenceReader{Reader: apiClient} + reconciler.Service.Resources.(*kubernetes.InstanceResources).Reader = &failedReferenceReader{Reader: apiClient} reconcileInstance(t, reconciler, object) reload(t, apiClient, object) assertInstanceReason(t, object, reasonDependency) - reconciler.Reader = apiClient + reconciler.Service.Resources.(*kubernetes.InstanceResources).Reader = apiClient reconcileInstance(t, reconciler, object) reload(t, apiClient, object) assertInstanceReason(t, object, "InstanceInUse") diff --git a/internal/database/controller/instance_setup.go b/internal/database/controller/instance_setup.go index fe9097f..a99eca1 100644 --- a/internal/database/controller/instance_setup.go +++ b/internal/database/controller/instance_setup.go @@ -22,15 +22,9 @@ func InstanceCacheOptions(namespace string) cache.Options { } func (r *InstanceReconciler) SetupWithManager(manager ctrl.Manager) error { - if r.Observer == nil || len(validation.IsDNS1123Label(r.SecretNamespace)) != 0 { + if r.Client == nil || r.Service == nil || r.Presenter == nil || len(validation.IsDNS1123Label(r.SecretNamespace)) != 0 { return errors.New("instance observer and valid management Secret namespace required") } - if r.Client == nil { - r.Client = manager.GetClient() - } - if r.Reader == nil { - r.Reader = manager.GetAPIReader() - } return ctrl.NewControllerManagedBy(manager). Named("database-instance"). For(&databasev1alpha1.PostgreSQLInstance{}). diff --git a/internal/database/application/application_credential.go b/internal/database/domain/credential/credential.go similarity index 99% rename from internal/database/application/application_credential.go rename to internal/database/domain/credential/credential.go index 2eb7634..506933d 100644 --- a/internal/database/application/application_credential.go +++ b/internal/database/domain/credential/credential.go @@ -14,7 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package application +package credential import ( "crypto/rand" diff --git a/internal/database/application/application_credential_test.go b/internal/database/domain/credential/credential_test.go similarity index 77% rename from internal/database/application/application_credential_test.go rename to internal/database/domain/credential/credential_test.go index c0e5f0b..8022fcf 100644 --- a/internal/database/application/application_credential_test.go +++ b/internal/database/domain/credential/credential_test.go @@ -14,7 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. */ -package application_test +package credential_test import ( "encoding/json" @@ -23,7 +23,8 @@ import ( "strings" "testing" - "git.ddupan.top/panxiao81/ayatori/internal/database/application" + credentialdomain "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" + "git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance" ) @@ -35,11 +36,11 @@ func TestApplicationCredential(t *testing.T) { if err != nil { t.Fatal(err) } - first, err := application.GenerateApplicationCredential("owner", "app", endpoint) + first, err := credentialdomain.GenerateApplicationCredential("owner", "app", endpoint) if err != nil { t.Fatal(err) } - second, err := application.GenerateApplicationCredential("owner", "app", endpoint) + second, err := credentialdomain.GenerateApplicationCredential("owner", "app", endpoint) if err != nil { t.Fatal(err) } @@ -47,7 +48,7 @@ func TestApplicationCredential(t *testing.T) { if len(data) != 7 || data["password"] == second.SecretData()["password"] || len(data["password"].(string)) != 43 { t.Fatal("expected seven keys and independent 256-bit passwords") } - parsed, err := application.ParseApplicationCredential(data) + parsed, err := credentialdomain.ParseApplicationCredential(data) if err != nil || !maps.Equal(parsed.SecretData(), data) { t.Fatal("credential did not round trip") } @@ -67,15 +68,15 @@ func TestApplicationCredential(t *testing.T) { for key := range data { invalid := maps.Clone(data) delete(invalid, key) - if _, err := application.ParseApplicationCredential(invalid); err == nil { + if _, err := credentialdomain.ParseApplicationCredential(invalid); err == nil { t.Fatalf("accepted missing %s", key) } invalid[key] = 42 - if _, err := application.ParseApplicationCredential(invalid); err == nil { + if _, err := credentialdomain.ParseApplicationCredential(invalid); err == nil { t.Fatalf("accepted non-string %s", key) } } - if (application.ApplicationCredential{}).Validate() == nil { + if (credentialdomain.ApplicationCredential{}).Validate() == nil { t.Fatal("accepted zero credential") } } diff --git a/internal/database/domain/credential/preparation.go b/internal/database/domain/credential/preparation.go new file mode 100644 index 0000000..12750a1 --- /dev/null +++ b/internal/database/domain/credential/preparation.go @@ -0,0 +1,129 @@ +package credential + +import ( + "git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding" + "git.ddupan.top/panxiao81/ayatori/internal/database/domain/instance" +) + +type Location struct { + Mount string + Path string +} + +// Phase 表达凭据准备进度，不依赖 Kubernetes Condition 的类型或 Reason。 +type Phase uint8 + +const ( + Pending Phase = iota + Pinned + Creating + Prepared + Conflict + Unavailable + Stopped + InvalidTarget +) + +type State struct { + Location *Location + Version int64 + Phase Phase + Message string +} + +func (s State) WithPhase(phase Phase, message string) State { + s.Phase, s.Message = phase, message + return s +} + +func (s State) Confirmed() bool { return s.Version > 0 } + +// Created 只接受首次创建并回读得到的版本，不能把后续写入认作首次供应。 +func (s State) Created(version int64) (State, *Issue) { + if version != 1 { + return s, &Issue{Conflict, "凭据创建冲突或结果不确定；请核对固定位置的版本历史，未认领、覆盖或重新生成密码"} + } + s.Version = version + return s.WithPhase(Prepared, "凭据已创建并回读确认；尚未创建 PostgreSQL 资源或交付给 Tenant"), nil +} + +// Resume 决定新一轮是否可以继续。未确认的创建不能靠读取成功认领。 +func (s State) Resume() (State, bool) { + if s.Version != 0 { + return s, true + } + switch s.Phase { + case Conflict: + return s, false + case Creating: + return s.WithPhase(Conflict, "凭据创建未留下成功确认；请核对固定位置与后端历史并人工处理，未重新生成密码"), false + default: + return s, true + } +} + +func (s State) CheckLocation(configured Location) *Issue { + if s.Location != nil && *s.Location != configured { + return &Issue{Unavailable, "部署配置与固定凭据位置不一致；请恢复原 mount/path 配置，未迁移或改密"} + } + return nil +} + +type Instance struct { + binding.Instance + Endpoint instance.Endpoint +} + +// Target 只包含供应资格所需事实，不含 resourceVersion、Conditions 或 repository 对象。 +type Target struct { + Database binding.Database + Tenant *binding.Tenant + Instance *Instance + DatabaseProtected bool + TenantProtected bool +} + +type Issue struct { + Phase Phase + Message string +} + +func (t Target) RequiresPreparation() bool { return t.Database.Source == "Provision" } + +func (t Target) CheckCredential(value ApplicationCredential) *Issue { + if t.Instance == nil || !value.MatchesTarget(t.Database.LoginRole, t.Database.Name, t.Instance.Endpoint) { + return &Issue{Conflict, "已确认凭据与当前 Instance/database/loginRole 不一致；请人工核实，未修改凭据"} + } + return nil +} + +func (t Target) Check() *Issue { + database := t.Database + if database.Deleting || database.Phase == binding.Deleting || database.Phase == "Released" { + return &Issue{Stopped, "Database 正在删除或已释放；保留凭据与 finalizer，不执行供应或清理"} + } + if database.Tenant == nil || t.Tenant == nil || t.Tenant.Database == nil { + return &Issue{Unavailable, "等待 Database 与 Tenant 双向绑定完成"} + } + if *database.Tenant != t.Tenant.Identity || *t.Tenant.Database != database.Identity { + return &Issue{Conflict, "双向绑定的名称或 UID 不匹配，未创建凭据"} + } + if t.Tenant.Deleting || t.Tenant.Phase != binding.Bound || !t.DatabaseProtected || !t.TenantProtected { + return &Issue{Stopped, "Tenant 未完成绑定、正在删除或缺少 finalizer 保护，未创建凭据"} + } + request, err := t.Tenant.Request.Resolve(t.Tenant.Identity) + if err != nil || (request.Provision != nil && !database.MatchesProvision(request, t.Tenant.Identity)) || request.Name != database.Identity.Name { + return &Issue{Conflict, "Tenant 申请与 Database 目标不一致，未创建凭据"} + } + if t.Instance == nil || database.InstanceUID == "" { + return &Issue{Unavailable, "等待 Instance 与已记录的实例身份"} + } + if issue := t.Instance.Check(&database); issue != nil { + phase := Unavailable + if issue.Reason == binding.Conflict { + phase = Conflict + } + return &Issue{phase, issue.Message} + } + return nil +} diff --git a/internal/database/domain/credential/preparation_test.go b/internal/database/domain/credential/preparation_test.go new file mode 100644 index 0000000..021714e --- /dev/null +++ b/internal/database/domain/credential/preparation_test.go @@ -0,0 +1,59 @@ +package credential_test + +import ( + "testing" + + credential "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" +) + +func TestPreparationResume(t *testing.T) { + for _, test := range []struct { + name string + phase credential.Phase + version int64 + continueAllowed bool + result credential.Phase + }{ + {"尚未创建", credential.Pinned, 0, true, credential.Pinned}, + {"依赖恢复", credential.Unavailable, 0, true, credential.Unavailable}, + {"中断创建", credential.Creating, 0, false, credential.Conflict}, + {"未确认冲突", credential.Conflict, 0, false, credential.Conflict}, + {"已确认后读取失败", credential.Unavailable, 1, true, credential.Unavailable}, + {"已确认后冲突重验", credential.Conflict, 1, true, credential.Conflict}, + } { + t.Run(test.name, func(t *testing.T) { + original := credential.State{Phase: test.phase, Version: test.version, Message: "保留原诊断"} + state, allowed := original.Resume() + if allowed != test.continueAllowed || state.Phase != test.result || state.Version != original.Version { + t.Fatal("恢复判定或确认版本发生变化") + } + if test.phase == credential.Conflict && state.Message != original.Message { + t.Fatal("冲突重入应保留原诊断") + } + }) + } +} + +func TestPreparationLocationAndConfirmation(t *testing.T) { + location := credential.Location{Mount: "applications", Path: "database/uid"} + state := credential.State{Location: &location, Phase: credential.Creating} + if issue := state.CheckLocation(location); issue != nil { + t.Fatal("固定位置不应被拒绝") + } + if issue := state.CheckLocation(credential.Location{Mount: "other", Path: location.Path}); issue == nil || issue.Phase != credential.Unavailable { + t.Fatal("配置变化必须停止，不迁移已固定位置") + } + for _, version := range []int64{0, -1, 2} { + result, issue := state.Created(version) + if issue == nil || issue.Phase != credential.Conflict || result.Confirmed() { + t.Fatal("错误版本不得确认创建") + } + } + result, issue := state.Created(1) + if issue != nil || !result.Confirmed() || result.Phase != credential.Prepared || result.Location != state.Location { + t.Fatal("首次写入回读应确认并保留位置") + } + if state.Version != 0 { + t.Fatal("领域判定不得修改调用方的旧状态") + } +} diff --git a/internal/database/domain/credential/target_test.go b/internal/database/domain/credential/target_test.go new file mode 100644 index 0000000..b461322 --- /dev/null +++ b/internal/database/domain/credential/target_test.go @@ -0,0 +1,52 @@ +package credential_test + +import ( + "testing" + + "git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding" + credential "git.ddupan.top/panxiao81/ayatori/internal/database/domain/credential" +) + +const ( + targetTestApplication = "sampleapp" + targetTestInstance = "test-instance" +) + +func TestPreparationTarget(t *testing.T) { + for _, test := range []struct { + name string + change func(*credential.Target) + want credential.Phase + }{ + {"完整绑定", func(*credential.Target) {}, credential.Pending}, + {"单向绑定", func(target *credential.Target) { target.Tenant.Database = nil }, credential.Unavailable}, + {"旧租户身份", func(target *credential.Target) { target.Tenant.Identity.UID = "new" }, credential.Conflict}, + {"旧实例身份", func(target *credential.Target) { target.Instance.Identity.UID = "new" }, credential.Conflict}, + {"资源删除", func(target *credential.Target) { target.Database.Deleting = true }, credential.Stopped}, + {"申请删除", func(target *credential.Target) { target.Tenant.Deleting = true }, credential.Stopped}, + {"Released", func(target *credential.Target) { target.Database.Phase = "Released" }, credential.Stopped}, + {"缺少保护", func(target *credential.Target) { target.DatabaseProtected = false }, credential.Stopped}, + {"实例未就绪", func(target *credential.Target) { target.Instance.Ready = false }, credential.Unavailable}, + {"目标变化", func(target *credential.Target) { target.Database.LoginRole = "other" }, credential.Conflict}, + } { + t.Run(test.name, func(t *testing.T) { + tenantID := binding.TenantIdentity{Namespace: "apps", Name: targetTestApplication, UID: "tenant"} + databaseID := binding.Identity{Name: binding.DynamicDatabaseName(tenantID.UID), UID: "database"} + target := credential.Target{ + Database: binding.Database{Identity: databaseID, Tenant: &tenantID, Instance: targetTestInstance, InstanceUID: "instance-id", Name: targetTestApplication, LoginRole: targetTestApplication, Source: "Provision"}, + Tenant: &binding.Tenant{Identity: tenantID, Database: &databaseID, Phase: binding.Bound, Request: binding.Request{Provision: &binding.ProvisionRequest{Instance: targetTestInstance}}}, + Instance: &credential.Instance{Identity: binding.Identity{Name: targetTestInstance, UID: "instance-id"}, Ready: true}, + DatabaseProtected: true, TenantProtected: true, + } + test.change(&target) + issue := target.Check() + if test.want == credential.Pending { + if issue != nil { + t.Fatalf("有效绑定被拒绝: %s", issue.Message) + } + } else if issue == nil || issue.Phase != test.want { + t.Fatal("领域资格判定不符") + } + }) + } +}