feat: 接入 Database 三资源 API 与分层绑定协调
确定单库单账号、集群级 Database、资源侧先写绑定和凭据定位合同。领域层承载纯规则，service 协調流程，Kubernetes adapter 负责资源呈现与版本保护。 验证：全量 make test、三轮 race、真实 API server 并发与重启补写、最小 RBAC/watch、lint 和文档检查通过。供应、凭据交付及删除清理尚未实现，保留 DeletionPending/finalizer 边界。
This commit is contained in:
@@ -0,0 +1,170 @@
|
||||
package application
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding"
|
||||
)
|
||||
|
||||
// 快照版本只用于协调读写，不进入领域规则。Generation 用于确认整个申请未在回读期间变化。
|
||||
type BindingTenant struct {
|
||||
binding.Tenant
|
||||
Revision string
|
||||
Generation int64
|
||||
}
|
||||
|
||||
type BindingDatabase struct {
|
||||
binding.Database
|
||||
Revision string
|
||||
}
|
||||
|
||||
// BindingResources 是这个用例所需的操作，不是通用 Repository 或跨系统事务接口。
|
||||
// 查不到对象时返回 nil；写入必须检查传入快照的版本，不能覆盖并发修改。
|
||||
type BindingResources interface {
|
||||
Tenant(context.Context, string, string) (*BindingTenant, error)
|
||||
Database(context.Context, string) (*BindingDatabase, error)
|
||||
Instance(context.Context, string) (*binding.Instance, error)
|
||||
BeginBinding(context.Context, *BindingTenant, *BindingStatus) (*BindingTenant, error)
|
||||
CreateDatabase(context.Context, binding.Target, binding.TenantIdentity) (*BindingDatabase, error)
|
||||
RecordInstance(context.Context, *BindingDatabase, string) (*BindingDatabase, error)
|
||||
BindDatabase(context.Context, *BindingDatabase, binding.TenantIdentity) (*BindingDatabase, error)
|
||||
}
|
||||
|
||||
// BindingStatus 是用例结果，资源呈现层决定如何写成 Conditions/status。
|
||||
type BindingStatus struct {
|
||||
Phase string
|
||||
Reason string
|
||||
Message string
|
||||
Database *binding.Identity
|
||||
}
|
||||
|
||||
type BindingResult struct {
|
||||
Tenant *BindingTenant
|
||||
Status BindingStatus
|
||||
RetrySoon bool
|
||||
}
|
||||
|
||||
type BindingService struct {
|
||||
Resources BindingResources
|
||||
}
|
||||
|
||||
func (s BindingService) Reconcile(ctx context.Context, namespace, name string) (BindingResult, error) {
|
||||
tenant, err := s.Resources.Tenant(ctx, namespace, name)
|
||||
if err != nil || tenant == nil {
|
||||
return BindingResult{}, err
|
||||
}
|
||||
if tenant.Deleting {
|
||||
return bindingResult(tenant, binding.Deleting, "DeletionPending",
|
||||
"删除清理尚未接入；保留 finalizer 和 Database 绑定，未执行后端删除"), nil
|
||||
}
|
||||
target, err := tenant.Request.Resolve(tenant.Identity)
|
||||
if err != nil {
|
||||
return bindingResult(tenant, tenant.Phase, "InvalidRequest", err.Error()), nil
|
||||
}
|
||||
// 持久固定申请，再创建/绑定资源；不是预先宣告双向绑定成功。
|
||||
var checkpoint *BindingStatus
|
||||
if tenant.Phase != binding.Binding && tenant.Phase != binding.Bound {
|
||||
checkpoint = &BindingStatus{Phase: binding.Binding, Reason: "BindingPending", Message: "申请目标已固定，等待资源侧绑定"}
|
||||
}
|
||||
tenant, err = s.Resources.BeginBinding(ctx, tenant, checkpoint)
|
||||
if err != nil {
|
||||
return BindingResult{}, err
|
||||
}
|
||||
database, issue, err := s.resolveDatabase(ctx, tenant, target)
|
||||
if err != nil {
|
||||
return BindingResult{}, err
|
||||
}
|
||||
if issue != nil {
|
||||
return bindingResult(tenant, tenant.Phase, issue.Reason, issue.Message), nil
|
||||
}
|
||||
if issue := database.CanBind(tenant.Tenant); issue != nil {
|
||||
return bindingResult(tenant, tenant.Phase, issue.Reason, issue.Message), nil
|
||||
}
|
||||
database, err = s.Resources.BindDatabase(ctx, database, tenant.Identity)
|
||||
if err != nil {
|
||||
return BindingResult{}, err
|
||||
}
|
||||
return s.confirmBinding(ctx, tenant, database)
|
||||
}
|
||||
|
||||
func (s BindingService) resolveDatabase(ctx context.Context, tenant *BindingTenant, target binding.Target) (
|
||||
*BindingDatabase, *binding.Issue, error,
|
||||
) {
|
||||
database, err := s.Resources.Database(ctx, target.Name)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if database == nil && target.Provision == nil {
|
||||
return nil, &binding.Issue{Reason: binding.DependencyUnavailable, Message: "指定的 Database 尚不存在，等待资源出现"}, nil
|
||||
}
|
||||
instanceName := ""
|
||||
var observed *binding.Database
|
||||
if database != nil {
|
||||
instanceName, observed = database.Instance, &database.Database
|
||||
}
|
||||
if target.Provision != nil {
|
||||
instanceName = target.Provision.Instance
|
||||
if database != nil && !database.MatchesProvision(target, tenant.Identity) {
|
||||
return nil, &binding.Issue{Reason: binding.Conflict,
|
||||
Message: "动态 Database 名称已存在，但目标或 Tenant UID 不匹配；请核实记录，未自动认领"}, nil
|
||||
}
|
||||
}
|
||||
instance, err := s.Resources.Instance(ctx, instanceName)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if instance == nil {
|
||||
return nil, &binding.Issue{Reason: binding.DependencyUnavailable, Message: "引用的 Instance 尚不存在"}, nil
|
||||
}
|
||||
if issue := instance.Check(observed); issue != nil {
|
||||
return nil, issue, nil
|
||||
}
|
||||
if database == nil {
|
||||
database, err = s.Resources.CreateDatabase(ctx, target, tenant.Identity)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
// 只有首次动态记录可以补入实例身份；导入必须先有资源观察。
|
||||
if database.InstanceUID == "" && target.Provision != nil {
|
||||
database, err = s.Resources.RecordInstance(ctx, database, instance.Identity.UID)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
if database.InstanceUID == "" {
|
||||
return nil, &binding.Issue{Reason: binding.DependencyUnavailable, Message: "Database 尚未完成实例身份验证"}, nil
|
||||
}
|
||||
return database, nil, nil
|
||||
}
|
||||
|
||||
func (s BindingService) confirmBinding(ctx context.Context, tenant *BindingTenant, database *BindingDatabase) (BindingResult, error) {
|
||||
latest, err := s.Resources.Tenant(ctx, tenant.Identity.Namespace, tenant.Identity.Name)
|
||||
if err != nil || latest == nil {
|
||||
return BindingResult{}, err
|
||||
}
|
||||
if latest.Identity != tenant.Identity || latest.Deleting {
|
||||
return BindingResult{}, nil
|
||||
}
|
||||
if latest.Generation != tenant.Generation {
|
||||
return BindingResult{RetrySoon: true}, nil
|
||||
}
|
||||
observed, err := s.Resources.Database(ctx, database.Identity.Name)
|
||||
if err != nil {
|
||||
return BindingResult{}, err
|
||||
}
|
||||
if observed == nil || observed.Identity != database.Identity || observed.Tenant == nil {
|
||||
return bindingResult(latest, latest.Phase, binding.Conflict, "资源侧身份或绑定已变化，未完成申请侧绑定"), nil
|
||||
}
|
||||
if issue := observed.CanBind(latest.Tenant); issue != nil {
|
||||
return bindingResult(latest, latest.Phase, issue.Reason, issue.Message), nil
|
||||
}
|
||||
result := bindingResult(latest, binding.Bound, "BindingComplete",
|
||||
"双向绑定已记录；尚未执行供应、应用登录验证或凭据交付，不能 Ready")
|
||||
result.Status.Database = &observed.Identity
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func bindingResult(tenant *BindingTenant, phase, reason, message string) BindingResult {
|
||||
return BindingResult{Tenant: tenant, Status: BindingStatus{Phase: phase, Reason: reason, Message: message}}
|
||||
}
|
||||
@@ -0,0 +1,181 @@
|
||||
package application
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"git.ddupan.top/panxiao81/ayatori/internal/database/domain/binding"
|
||||
)
|
||||
|
||||
const (
|
||||
bindingReadTenant = "tenant"
|
||||
bindingBegin = "begin"
|
||||
bindingReadInstance = "instance"
|
||||
bindingCreate = "create"
|
||||
bindingRecordInstance = "record-instance"
|
||||
bindingWriteResource = "bind"
|
||||
bindingDatabaseOperation = "database"
|
||||
bindingTestNamespace = "apps"
|
||||
bindingTestName = "app"
|
||||
bindingTestInstance = "shared"
|
||||
)
|
||||
|
||||
func TestBindingServiceOrder(t *testing.T) {
|
||||
resources := bindingFixture()
|
||||
service := BindingService{Resources: resources}
|
||||
result, err := service.Reconcile(t.Context(), bindingTestNamespace, bindingTestName)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []string{bindingReadTenant, bindingBegin, bindingDatabaseOperation, bindingReadInstance, bindingCreate, bindingRecordInstance, bindingWriteResource, bindingReadTenant, bindingDatabaseOperation}
|
||||
if !reflect.DeepEqual(resources.calls, want) {
|
||||
t.Fatalf("协调顺序 = %v, want %v", resources.calls, want)
|
||||
}
|
||||
if result.Status.Phase != binding.Bound || result.Status.Database == nil || result.Status.Database.UID != "database-uid" {
|
||||
t.Fatalf("绑定结果不符: %+v", result.Status)
|
||||
}
|
||||
if resources.tenant.Phase != binding.Binding || resources.tenant.Database != nil {
|
||||
t.Fatal("service 只能返回待呈现结果，不能提前写申请侧绑定")
|
||||
}
|
||||
if resources.database.Tenant == nil || *resources.database.Tenant != resources.tenant.Identity {
|
||||
t.Fatal("返回完成结果之前必须写入资源侧绑定")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBindingServiceStopsOnIOFailure(t *testing.T) {
|
||||
for _, operation := range []string{bindingReadTenant, bindingBegin, bindingDatabaseOperation, bindingReadInstance, bindingCreate, bindingRecordInstance, bindingWriteResource} {
|
||||
t.Run(operation, func(t *testing.T) {
|
||||
resources := bindingFixture()
|
||||
resources.failAt = operation
|
||||
result, err := (BindingService{Resources: resources}).Reconcile(t.Context(), bindingTestNamespace, bindingTestName)
|
||||
if !errors.Is(err, errBindingTest) || result.Status.Database != nil {
|
||||
t.Fatalf("IO 失败不应被转换为绑定成功: result=%+v, err=%v", result, err)
|
||||
}
|
||||
if resources.calls[len(resources.calls)-1] != operation {
|
||||
t.Fatalf("失败后继续执行了操作: %v", resources.calls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBindingServiceDoesNotWriteWhenDeletingOrInvalid(t *testing.T) {
|
||||
for _, deleting := range []bool{false, true} {
|
||||
resources := bindingFixture()
|
||||
resources.tenant.Deleting = deleting
|
||||
resources.tenant.Request = binding.Request{}
|
||||
result, err := (BindingService{Resources: resources}).Reconcile(t.Context(), bindingTestNamespace, bindingTestName)
|
||||
if err != nil || len(resources.calls) != 1 || result.Status.Database != nil {
|
||||
t.Fatalf("删除或无效申请不应触及资源: calls=%v, err=%v", resources.calls, err)
|
||||
}
|
||||
want := "InvalidRequest"
|
||||
if deleting {
|
||||
want = "DeletionPending"
|
||||
}
|
||||
if result.Status.Reason != want {
|
||||
t.Fatalf("Reason = %s, want %s", result.Status.Reason, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBindingServiceConfirmsIdentityAgain(t *testing.T) {
|
||||
resources := bindingFixture()
|
||||
resources.replaceOnReadback = true
|
||||
result, err := (BindingService{Resources: resources}).Reconcile(t.Context(), bindingTestNamespace, bindingTestName)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if result.Status.Reason != binding.Conflict || result.Status.Database != nil {
|
||||
t.Fatal("资源侧写入后发生身份变化时不能返回绑定完成")
|
||||
}
|
||||
}
|
||||
|
||||
// 这里只记录用例操作，不模拟 Kubernetes 校验；真实 IO 契约由 controller envtest 覆盖。
|
||||
type bindingTestResources struct {
|
||||
tenant *BindingTenant
|
||||
database *BindingDatabase
|
||||
instance *binding.Instance
|
||||
calls []string
|
||||
failAt string
|
||||
replaceOnReadback bool
|
||||
bound bool
|
||||
}
|
||||
|
||||
var errBindingTest = errors.New("injected resource operation failure")
|
||||
|
||||
func bindingFixture() *bindingTestResources {
|
||||
tenant := &BindingTenant{Generation: 1}
|
||||
tenant.Tenant = binding.Tenant{
|
||||
Identity: binding.TenantIdentity{Namespace: bindingTestNamespace, Name: bindingTestName, UID: "tenant-uid"},
|
||||
Request: binding.Request{Provision: &binding.ProvisionRequest{Instance: bindingTestInstance}},
|
||||
}
|
||||
return &bindingTestResources{
|
||||
tenant: tenant,
|
||||
instance: &binding.Instance{Identity: binding.Identity{Name: bindingTestInstance, UID: "instance-uid"}, Ready: true},
|
||||
}
|
||||
}
|
||||
|
||||
func (r *bindingTestResources) record(operation string) error {
|
||||
r.calls = append(r.calls, operation)
|
||||
if r.failAt == operation {
|
||||
return errBindingTest
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *bindingTestResources) Tenant(context.Context, string, string) (*BindingTenant, error) {
|
||||
return r.tenant, r.record(bindingReadTenant)
|
||||
}
|
||||
|
||||
func (r *bindingTestResources) Database(context.Context, string) (*BindingDatabase, error) {
|
||||
if r.bound && r.replaceOnReadback {
|
||||
replaced := *r.database
|
||||
replaced.Identity.UID = "replacement"
|
||||
return &replaced, r.record(bindingDatabaseOperation)
|
||||
}
|
||||
return r.database, r.record(bindingDatabaseOperation)
|
||||
}
|
||||
|
||||
func (r *bindingTestResources) Instance(context.Context, string) (*binding.Instance, error) {
|
||||
return r.instance, r.record(bindingReadInstance)
|
||||
}
|
||||
|
||||
func (r *bindingTestResources) BeginBinding(_ context.Context, tenant *BindingTenant, status *BindingStatus) (*BindingTenant, error) {
|
||||
if err := r.record(bindingBegin); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if status != nil {
|
||||
r.tenant.Phase = status.Phase
|
||||
}
|
||||
return tenant, nil
|
||||
}
|
||||
|
||||
func (r *bindingTestResources) CreateDatabase(_ context.Context, target binding.Target, tenant binding.TenantIdentity) (*BindingDatabase, error) {
|
||||
if err := r.record(bindingCreate); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r.database = &BindingDatabase{}
|
||||
r.database.Database = binding.Database{
|
||||
Identity: binding.Identity{Name: target.Name, UID: "database-uid"}, Tenant: &tenant,
|
||||
Instance: target.Provision.Instance, Name: target.Provision.Database, LoginRole: target.Provision.LoginRole, Source: "Provision",
|
||||
}
|
||||
return r.database, nil
|
||||
}
|
||||
|
||||
func (r *bindingTestResources) RecordInstance(_ context.Context, database *BindingDatabase, uid string) (*BindingDatabase, error) {
|
||||
if err := r.record(bindingRecordInstance); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
database.InstanceUID = uid
|
||||
return database, nil
|
||||
}
|
||||
|
||||
func (r *bindingTestResources) BindDatabase(_ context.Context, database *BindingDatabase, tenant binding.TenantIdentity) (*BindingDatabase, error) {
|
||||
if err := r.record(bindingWriteResource); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
database.Tenant = &tenant
|
||||
r.bound = true
|
||||
return database, nil
|
||||
}
|
||||
Reference in New Issue
Block a user