Alanxtl commented on code in PR #1046: URL: https://github.com/apache/dubbo-go-pixiu/pull/1046#discussion_r4109957924
########## admin/logic/route_binding.go: ########## @@ -0,0 +1,1245 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package logic + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/url" + "reflect" + "sort" + "strconv" + "strings" + "time" +) + +import ( + clientv3 "go.etcd.io/etcd/client/v3" +) + +import ( + adminconfig "github.com/apache/dubbo-go-pixiu/admin/config" + commonyaml "github.com/apache/dubbo-go-pixiu/pkg/common/yaml" + legacyconfig "github.com/apache/dubbo-go-pixiu/pkg/config" + "github.com/apache/dubbo-go-pixiu/pkg/config/schema" +) + +const ( + routeBindingPathSegment = "route-bindings" + routeBindingRevisionSegment = "route-bindings-revision" + routeBindingRuntimeResourceDir = "resources" + maxRouteBindingIDRetries = 16 +) + +var ( + ErrRouteBindingNotFound = errors.New("admin route binding not found") + ErrRouteBindingAlreadyExists = errors.New("admin route binding already exists") + ErrRouteBindingNameImmutable = errors.New("admin route binding metadata.name is immutable") + ErrRouteBindingConflict = errors.New("admin route binding was modified concurrently") + ErrRouteBindingPublishConflict = errors.New("admin route binding publish conflict") + ErrRouteBindingRuntimeConflict = errors.New("admin route binding conflicts with an existing runtime route") +) + +// routeBindingKV is the small part of clientv3.KV needed by the route store. +// Keeping this boundary narrow makes the planning and transaction logic easy +// to exercise without coupling it to the gost wrapper. +type routeBindingKV interface { + Get(context.Context, string, ...clientv3.OpOption) (*clientv3.GetResponse, error) + Txn(context.Context) clientv3.Txn +} + +// RouteBinding is the Admin-facing response shape. ResourceID and MethodID +// are stable runtime identities used by Pixiu's legacy etcd watcher; they are +// not part of the user-editable AdminRouteBinding object. +type RouteBinding struct { + Object schema.AdminObject `json:"object"` + ResourceID int `json:"resourceId"` + MethodID int `json:"methodId"` + Revision int64 `json:"revision"` + PublishStatus *RouteBindingPublishStatus `json:"publishStatus,omitempty"` +} + +// RouteBindingPublishResult describes one atomic publish transaction. +type RouteBindingPublishResult struct { + Name string `json:"name"` + Revision int64 `json:"revision"` + DraftRevision int64 `json:"draftRevision"` + PublishedRevision int64 `json:"publishedRevision"` + PublishedCount int `json:"publishedCount"` + DeletedCount int `json:"deletedCount"` +} + +// RouteBindingPublishStatus exposes one route's draft and published state. +// Revisions are the etcd key mod revisions for that route, not a global +// configuration revision. +type RouteBindingPublishStatus struct { + Name string `json:"name"` + DraftRevision int64 `json:"draftRevision"` + PublishedRevision int64 `json:"publishedRevision"` + DraftExists bool `json:"draftExists"` + PublishedExists bool `json:"publishedExists"` + Dirty bool `json:"dirty"` +} + +type RouteBindingDiffChange struct { + Path string `json:"path"` + Before any `json:"before"` + After any `json:"after"` +} + +// RouteBindingDiff compares one route's draft object with its published +// object. A nil side represents a route that only exists in the other scope. +type RouteBindingDiff struct { + Name string `json:"name"` + Draft *RouteBinding `json:"draft,omitempty"` + Published *RouteBinding `json:"published,omitempty"` + Changes []RouteBindingDiffChange `json:"changes"` +} + +type routeBindingRecord struct { + Object schema.AdminObject `json:"object"` + ResourceID int `json:"resourceId"` + MethodID int `json:"methodId"` +} + +type routeBindingEntry struct { + record routeBindingRecord + key string + revision int64 +} + +// RouteBindingStore owns the AdminRouteBinding persistence and publish +// boundary. Published runtime data is deliberately written in the existing +// Resource/Method layout so the current Pixiu watcher can consume it. +type RouteBindingStore struct { + kv routeBindingKV + root string + registry *schema.Registry +} + +// NewRouteBindingStore creates a store over an etcd v3 client. +func NewRouteBindingStore(kv routeBindingKV, root string, registry *schema.Registry) (*RouteBindingStore, error) { + if kv == nil { + return nil, errors.New("route binding etcd client is nil") + } + root = strings.TrimRight(strings.TrimSpace(root), "/") + if root == "" { + return nil, errors.New("route binding etcd root path is empty") + } + if registry == nil { + var err error + registry, err = schema.NewBuiltinRegistry() + if err != nil { + return nil, fmt.Errorf("create route binding schema registry: %w", err) + } + } + return &RouteBindingStore{kv: kv, root: root, registry: registry}, nil +} + +// NewAdminRouteBindingStore creates the production store from the Admin +// process' configured etcd client and root path. +func NewAdminRouteBindingStore() (*RouteBindingStore, error) { + if adminconfig.Bootstrap == nil { + return nil, errors.New("admin bootstrap is nil") + } + if adminconfig.Client == nil { + return nil, errors.New("admin etcd client is nil") + } + rawClient := adminconfig.Client.GetRawClient() + if rawClient == nil { + return nil, errors.New("admin raw etcd client is nil") + } + return NewRouteBindingStore(rawClient, adminconfig.Bootstrap.GetPath(), nil) +} + +func routeBindingContext(ctx context.Context) context.Context { + if ctx == nil { + return context.Background() + } + return ctx +} + +// Registry returns the schema registry used by this store. +func (s *RouteBindingStore) Registry() *schema.Registry { + return s.registry +} + +// List returns high-level route bindings from either the draft or published +// namespace. Empty namespaces are returned as an empty slice. +func (s *RouteBindingStore) List(ctx context.Context, unpublished bool) ([]RouteBinding, error) { + ctx = routeBindingContext(ctx) + entries, err := s.listEntries(ctx, s.bindingPrefix(unpublished)) + if err != nil { + return nil, err + } + var statuses map[string]RouteBindingPublishStatus + if unpublished { + publishedEntries, err := s.listEntries(ctx, s.bindingPrefix(false)) + if err != nil { + return nil, err + } + statuses = routeBindingStatuses(entries, publishedEntries) + } + result := make([]RouteBinding, 0, len(entries)) + for _, entry := range entries { + view := routeBindingView(entry) + if status, exists := statuses[entry.record.Object.Metadata.Name]; exists { + view.PublishStatus = &status + } + result = append(result, view) + } + return result, nil +} + +func routeBindingStatuses(draftEntries, publishedEntries []routeBindingEntry) map[string]RouteBindingPublishStatus { + publishedByName := make(map[string]routeBindingEntry, len(publishedEntries)) + for _, entry := range publishedEntries { + publishedByName[entry.record.Object.Metadata.Name] = entry + } + + statuses := make(map[string]RouteBindingPublishStatus, len(draftEntries)) + for _, draft := range draftEntries { + name := draft.record.Object.Metadata.Name + published, publishedExists := publishedByName[name] + status := RouteBindingPublishStatus{ + Name: name, + DraftRevision: draft.revision, + DraftExists: true, + PublishedExists: publishedExists, + PublishedRevision: published.revision, + Dirty: !publishedExists || !reflect.DeepEqual(draft.record.Object, published.record.Object), + } + statuses[name] = status + } + return statuses +} + +// Get returns one high-level route binding from either the draft or published +// namespace. +func (s *RouteBindingStore) Get(ctx context.Context, name string, unpublished bool) (RouteBinding, error) { + ctx = routeBindingContext(ctx) + name, err := normalizeRouteBindingName(name) + if err != nil { + return RouteBinding{}, err + } + entry, exists, _, err := s.getEntry(ctx, s.bindingPrefix(unpublished), name) + if err != nil { + return RouteBinding{}, err + } + if !exists { + return RouteBinding{}, fmt.Errorf("%w: %s", ErrRouteBindingNotFound, name) + } + return routeBindingView(entry), nil +} + +// Normalize validates an object and applies all registered schema defaults. +// It is useful to the controller's validate and preview endpoints and does +// not require an etcd write. +func (s *RouteBindingStore) Normalize(object schema.AdminObject) (schema.AdminObject, error) { + normalized, err := s.registry.Normalize(object) + if err != nil { + return normalized, err + } + name, err := normalizeRouteBindingName(normalized.Metadata.Name) + if err != nil { + return normalized, err + } + normalized.Metadata.Name = name + return normalized, nil +} + +// Preview validates and compiles an Admin object without writing it. +func (s *RouteBindingStore) Preview(object schema.AdminObject) (schema.AdminObject, []byte, error) { + normalized, err := s.Normalize(object) + if err != nil { + return normalized, nil, err + } + compiled, err := schema.CompileAdminRouteBinding(s.registry, normalized) + if err != nil { + return normalized, nil, err + } + preview, err := compiled.PreviewYAML() + if err != nil { + return normalized, nil, err + } + return normalized, preview, nil +} + +// SaveDraft validates and atomically stores one AdminRouteBinding in the +// draft namespace. POST callers use create=true; PUT callers use create=false. +// expectedRevision is the draft key's mod revision and may be zero to allow +// an unconditional write. +func (s *RouteBindingStore) SaveDraft(ctx context.Context, object schema.AdminObject, create bool, expectedRevision int64) (RouteBinding, error) { + ctx = routeBindingContext(ctx) + if err := validateSaveDraftRequest(create, expectedRevision); err != nil { + return RouteBinding{}, err + } + normalized, err := s.Normalize(object) + if err != nil { + return RouteBinding{}, err + } + name := normalized.Metadata.Name + draftPrefix := s.bindingPrefix(true) + draftEntry, draftExists, draftRevision, err := s.getEntry(ctx, draftPrefix, name) + if err != nil { + return RouteBinding{}, err + } + if err := validateDraftSaveState(name, create, expectedRevision, draftExists, draftRevision); err != nil { + return RouteBinding{}, err + } + + resourceID, methodID, err := s.resolveRuntimeIdentity(ctx, name, draftEntry, draftExists) + if err != nil { + return RouteBinding{}, err + } + + record := routeBindingRecord{ + Object: normalized, + ResourceID: resourceID, + MethodID: methodID, + } + value, err := json.Marshal(record) + if err != nil { + return RouteBinding{}, fmt.Errorf("encode route binding %q: %w", name, err) + } + if err := s.commitDraft(ctx, name, draftPrefix, value, create, expectedRevision); err != nil { + return RouteBinding{}, err + } + + saved, exists, revision, err := s.getEntry(ctx, draftPrefix, name) + if err != nil { + return RouteBinding{}, err + } + if !exists { + return RouteBinding{}, fmt.Errorf("route binding %q disappeared after save", name) + } + saved.revision = revision + return routeBindingView(saved), nil +} + +// UpdateDraft updates one existing route binding draft without allowing its +// metadata.name to change. The original name is the stable identity supplied +// by the caller, while the object name is user-editable in the YAML editor. +func (s *RouteBindingStore) UpdateDraft(ctx context.Context, originalName string, object schema.AdminObject, expectedRevision int64) (RouteBinding, error) { + originalName, err := normalizeRouteBindingName(originalName) + if err != nil { + return RouteBinding{}, err + } + normalized, err := s.Normalize(object) + if err != nil { + return RouteBinding{}, err + } + if normalized.Metadata.Name != originalName { + return RouteBinding{}, fmt.Errorf("%w: %q cannot be changed to %q", ErrRouteBindingNameImmutable, originalName, normalized.Metadata.Name) + } + return s.SaveDraft(ctx, normalized, false, expectedRevision) +} + +func validateSaveDraftRequest(create bool, expectedRevision int64) error { + if expectedRevision < 0 { + return errors.New("expected revision must not be negative") + } + if create && expectedRevision > 0 { + return errors.New("create route binding does not accept an expected revision") + } + return nil +} + +func validateDraftSaveState(name string, create bool, expectedRevision int64, exists bool, revision int64) error { + if create && exists { + return fmt.Errorf("%w: %s", ErrRouteBindingAlreadyExists, name) + } + if expectedRevision > 0 && (!exists || revision != expectedRevision) { + return fmt.Errorf("%w: %s", ErrRouteBindingConflict, name) + } + return nil +} + +func (s *RouteBindingStore) resolveRuntimeIdentity(ctx context.Context, name string, draft routeBindingEntry, draftExists bool) (int, int, error) { + resourceID, methodID := draft.record.ResourceID, draft.record.MethodID + if !draftExists { + published, exists, _, err := s.getEntry(ctx, s.bindingPrefix(false), name) + if err != nil { + return 0, 0, err + } + if exists { + resourceID, methodID = published.record.ResourceID, published.record.MethodID + } + } + if resourceID <= 0 { + var err error + resourceID, err = s.allocateRuntimeID(ctx) + if err != nil { + return 0, 0, err + } + } + if methodID <= 0 { + // One AdminRouteBinding currently compiles to exactly one Method. Method + // IDs are scoped below the resource, so sharing the stable route ID is + // safe and avoids a second allocation transaction. + methodID = resourceID + } + return resourceID, methodID, nil +} + +func (s *RouteBindingStore) commitDraft(ctx context.Context, name, draftPrefix string, value []byte, create bool, expectedRevision int64) error { + comparisons := make([]clientv3.Cmp, 0, 1) + if create { + comparisons = append(comparisons, clientv3.Compare(clientv3.CreateRevision(s.bindingKey(draftPrefix, name)), "=", 0)) + } else if expectedRevision > 0 { + comparisons = append(comparisons, clientv3.Compare(clientv3.ModRevision(s.bindingKey(draftPrefix, name)), "=", expectedRevision)) + } + operations := []clientv3.Op{ + clientv3.OpPut(s.bindingKey(draftPrefix, name), string(value)), + } + transaction := s.kv.Txn(ctx) + if len(comparisons) > 0 { + transaction = transaction.If(comparisons...) + } + response, err := transaction.Then(operations...).Commit() + if err != nil { + return fmt.Errorf("save route binding %q: %w", name, err) + } + if response.Succeeded { + return nil + } + if create { + return fmt.Errorf("%w: %s", ErrRouteBindingAlreadyExists, name) + } + return fmt.Errorf("%w: %s", ErrRouteBindingConflict, name) +} + +// DeleteAndPublish removes one route draft and its published runtime state in +// the same etcd transaction. This keeps the delete flow atomic for callers: +// there is no intermediate state where the draft is gone but the route still +// remains active in Pixiu. +func (s *RouteBindingStore) DeleteAndPublish(ctx context.Context, name string, expectedRevision int64) (RouteBindingPublishResult, error) { + ctx = routeBindingContext(ctx) + if expectedRevision < 0 { + return RouteBindingPublishResult{}, errors.New("expected revision must not be negative") + } + name, err := normalizeRouteBindingName(name) + if err != nil { + return RouteBindingPublishResult{}, err + } + state, err := s.readPublishState(ctx, name) + if err != nil { + return RouteBindingPublishResult{}, err + } + if expectedRevision > 0 && (!state.draftExists || state.draftRevision != expectedRevision) { + return RouteBindingPublishResult{}, fmt.Errorf("%w: %s", ErrRouteBindingConflict, name) + } + comparisons, err := s.buildPublishComparisons(ctx, state) + if err != nil { + return RouteBindingPublishResult{}, err + } + operations := make([]clientv3.Op, 0, 4) + if state.draftExists { + operations = append(operations, clientv3.OpDelete(s.bindingKey(state.draftPrefix, name))) + } + deletedCount := 0 + if state.publishedExists { + if state.published.record.ResourceID <= 0 || state.published.record.MethodID <= 0 { + return RouteBindingPublishResult{}, fmt.Errorf("published route binding %q has invalid runtime identity", name) + } + operations = append(operations, + clientv3.OpDelete(s.runtimeResourceKey(state.published.record.ResourceID), clientv3.WithPrefix()), + clientv3.OpDelete(s.bindingKey(state.publishedPrefix, name)), + ) + deletedCount = 1 + } + operations = append(operations, + clientv3.OpPut(s.publishedRevisionKey(), strconv.FormatInt(time.Now().UnixNano(), 10))) + response, err := s.kv.Txn(ctx).If(comparisons...).Then(operations...).Commit() + if err != nil { + return RouteBindingPublishResult{}, fmt.Errorf("delete route binding %q: %w", name, err) + } + if !response.Succeeded { + return RouteBindingPublishResult{}, fmt.Errorf("%w: %s", ErrRouteBindingPublishConflict, name) + } + return RouteBindingPublishResult{ + Name: name, + Revision: response.Header.Revision, + DraftRevision: state.draftRevision, + PublishedRevision: 0, + DeletedCount: deletedCount, + }, nil +} + +// Publish publishes exactly one route binding. The generated legacy resource +// and method, the published high-level binding, and the internal generation +// guard are written or deleted in one etcd transaction. Other routes' draft +// and published values are left untouched. +func (s *RouteBindingStore) Publish(ctx context.Context, name string, expectedDraftRevision int64) (RouteBindingPublishResult, error) { + ctx = routeBindingContext(ctx) + if expectedDraftRevision < 0 { + return RouteBindingPublishResult{}, errors.New("expected draft revision must not be negative") + } + name, err := normalizeRouteBindingName(name) + if err != nil { + return RouteBindingPublishResult{}, err + } + state, err := s.readPublishState(ctx, name) + if err != nil { + return RouteBindingPublishResult{}, err + } + if expectedDraftRevision > 0 && (!state.draftExists || state.draftRevision != expectedDraftRevision) { + return RouteBindingPublishResult{}, fmt.Errorf("%w: %s", ErrRouteBindingPublishConflict, name) + } + publishedEntries, err := s.listEntries(ctx, state.publishedPrefix) + if err != nil { + return RouteBindingPublishResult{}, err + } + candidate, err := s.compilePublishCandidate(name, state.draft, state.draftExists, publishedEntries) + if err != nil { + return RouteBindingPublishResult{}, err + } + if err := s.checkRuntimeRouteConflict(ctx, candidate, state); err != nil { + return RouteBindingPublishResult{}, err + } + comparisons, err := s.buildPublishComparisons(ctx, state) + if err != nil { + return RouteBindingPublishResult{}, err + } + operations, publishedCount, deletedCount, err := s.buildSinglePublishOperations(state, candidate) + if err != nil { + return RouteBindingPublishResult{}, err + } + operations = append(operations, + clientv3.OpPut(s.publishedRevisionKey(), strconv.FormatInt(time.Now().UnixNano(), 10))) + response, err := s.kv.Txn(ctx).If(comparisons...).Then(operations...).Commit() + if err != nil { + return RouteBindingPublishResult{}, fmt.Errorf("publish route binding %q: %w", name, err) + } + if !response.Succeeded { + return RouteBindingPublishResult{}, fmt.Errorf("%w: %s", ErrRouteBindingPublishConflict, name) + } + publishedRevision, err := s.publishedRouteRevision(ctx, state.publishedPrefix, name, candidate) + if err != nil { + return RouteBindingPublishResult{}, err + } + return RouteBindingPublishResult{ + Name: name, + Revision: response.Header.Revision, + DraftRevision: state.draftRevision, + PublishedRevision: publishedRevision, + PublishedCount: publishedCount, + DeletedCount: deletedCount, + }, nil +} + +type routeBindingPublishState struct { + draftPrefix string + publishedPrefix string + draft routeBindingEntry + draftExists bool + draftRevision int64 + published routeBindingEntry + publishedExists bool + publishedRevision int64 +} + +func (s *RouteBindingStore) readPublishState(ctx context.Context, name string) (routeBindingPublishState, error) { + state := routeBindingPublishState{ + draftPrefix: s.bindingPrefix(true), + publishedPrefix: s.bindingPrefix(false), + } + var err error + state.draft, state.draftExists, state.draftRevision, err = s.getEntry(ctx, state.draftPrefix, name) + if err != nil { + return routeBindingPublishState{}, err + } + state.published, state.publishedExists, state.publishedRevision, err = s.getEntry(ctx, state.publishedPrefix, name) + if err != nil { + return routeBindingPublishState{}, err + } + if !state.draftExists && !state.publishedExists { + return routeBindingPublishState{}, fmt.Errorf("%w: %s", ErrRouteBindingNotFound, name) + } + return state, nil +} + +func (s *RouteBindingStore) compilePublishCandidate(name string, draft routeBindingEntry, draftExists bool, published []routeBindingEntry) (*compiledRouteEntry, error) { + if !draftExists { + return nil, nil + } + entries := make([]routeBindingEntry, 0, len(published)+1) + for _, entry := range published { + if entry.record.Object.Metadata.Name != name { + entries = append(entries, entry) + } + } + entries = append(entries, draft) + compiled, err := s.compileSnapshot(entries) + if err != nil { + return nil, err + } + for index := range compiled { + if compiled[index].object.Metadata.Name == name { + return &compiled[index], nil + } + } + return nil, fmt.Errorf("route binding %q was not compiled", name) +} + +func (s *RouteBindingStore) buildPublishComparisons(ctx context.Context, state routeBindingPublishState) ([]clientv3.Cmp, error) { + comparisons := make([]clientv3.Cmp, 0, 2) + if state.draftExists { + // Always compare the value observed above, even when the caller omitted + // expectedDraftRevision. This closes the read/modify/write race. + comparisons = append(comparisons, + clientv3.Compare(clientv3.ModRevision(s.bindingKey(state.draftPrefix, state.draft.record.Object.Metadata.Name)), "=", state.draftRevision)) + if state.publishedExists { + comparisons = append(comparisons, + clientv3.Compare(clientv3.ModRevision(s.bindingKey(state.publishedPrefix, state.published.record.Object.Metadata.Name)), "=", state.publishedRevision)) + } else { + comparisons = append(comparisons, + clientv3.Compare(clientv3.CreateRevision(s.bindingKey(state.publishedPrefix, state.draft.record.Object.Metadata.Name)), "=", 0)) + } + } else { + comparisons = append(comparisons, + clientv3.Compare(clientv3.ModRevision(s.bindingKey(state.publishedPrefix, state.published.record.Object.Metadata.Name)), "=", state.publishedRevision)) + } + publishedMarkerRevision, publishedMarkerExists, err := s.markerRevision(ctx, s.publishedRevisionKey()) + if err != nil { + return nil, err + } + if publishedMarkerExists { + comparisons = append(comparisons, + clientv3.Compare(clientv3.ModRevision(s.publishedRevisionKey()), "=", publishedMarkerRevision)) + } else { + comparisons = append(comparisons, + clientv3.Compare(clientv3.CreateRevision(s.publishedRevisionKey()), "=", 0)) + } + return comparisons, nil +} + +func (s *RouteBindingStore) buildSinglePublishOperations(state routeBindingPublishState, candidate *compiledRouteEntry) ([]clientv3.Op, int, int, error) { + if candidate == nil { + return []clientv3.Op{ + clientv3.OpDelete(s.runtimeResourceKey(state.published.record.ResourceID), clientv3.WithPrefix()), + clientv3.OpDelete(s.bindingKey(state.publishedPrefix, state.published.record.Object.Metadata.Name)), + }, 0, 1, nil + } + compiled := candidate.legacy + operations := make([]clientv3.Op, 0, 4) + if state.publishedExists && state.published.record.ResourceID != compiled.Resource.ID { + operations = append(operations, + clientv3.OpDelete(s.runtimeResourceKey(state.published.record.ResourceID), clientv3.WithPrefix())) + } else if state.publishedExists && state.published.record.MethodID != compiled.Method.ID { + operations = append(operations, + clientv3.OpDelete(s.runtimeMethodKey(state.published.record.ResourceID, state.published.record.MethodID))) + } + resourceValue, methodValue, err := marshalCompiledRuntimeRoute(compiled) + if err != nil { + return nil, 0, 0, fmt.Errorf("encode route binding %q for runtime: %w", candidate.object.Metadata.Name, err) + } + recordValue, err := json.Marshal(routeBindingRecord{ + Object: candidate.object, + ResourceID: compiled.Resource.ID, + MethodID: compiled.Method.ID, + }) + if err != nil { + return nil, 0, 0, fmt.Errorf("encode published route binding %q: %w", candidate.object.Metadata.Name, err) + } + operations = append(operations, + clientv3.OpPut(s.runtimeResourceKey(compiled.Resource.ID), string(resourceValue)), + clientv3.OpPut(s.runtimeMethodKey(compiled.Resource.ID, compiled.Method.ID), string(methodValue)), + clientv3.OpPut(s.bindingKey(state.publishedPrefix, candidate.object.Metadata.Name), string(recordValue))) + return operations, 1, 0, nil +} + +func (s *RouteBindingStore) publishedRouteRevision(ctx context.Context, prefix, name string, candidate *compiledRouteEntry) (int64, error) { + if candidate == nil { + return 0, nil + } + _, exists, revision, err := s.getEntry(ctx, prefix, name) + if err != nil { + return 0, err + } + if !exists { + return 0, fmt.Errorf("route binding %q disappeared after publish", name) + } + return revision, nil +} + +// checkRuntimeRouteConflict protects the breaking migration boundary between +// the legacy Resource/Method keys and AdminRouteBinding. Existing runtime +// method keys are still consumed by Pixiu's watcher, even when they are not +// visible through the new Admin API. Publishing a new route with the same +// HTTP method and path would otherwise commit successfully and leave the +// watcher with duplicate runtime routes. +func (s *RouteBindingStore) checkRuntimeRouteConflict(ctx context.Context, candidate *compiledRouteEntry, state routeBindingPublishState) error { + if candidate == nil { + return nil + } + prefix := s.runtimeResourcePrefix() + response, err := s.kv.Get(ctx, prefix, clientv3.WithPrefix()) + if err != nil { + return fmt.Errorf("list existing runtime methods: %w", err) + } + candidatePath := candidate.legacy.Resource.Path + candidateVerb := strings.TrimSpace(candidate.legacy.Method.HTTPVerb) + for _, kv := range response.Kvs { + resourceID, methodID, ok, err := runtimeMethodIDs(prefix, string(kv.Key)) + if err != nil { + return err + } + if !ok { + continue + } + // A republish of the same published AdminRouteBinding owns this runtime + // identity; a new route must not bypass the check by reusing an orphaned + // legacy method key with the same numeric IDs. + if state.publishedExists && resourceID == state.published.record.ResourceID && methodID == state.published.record.MethodID { + continue + } + var method legacyconfig.Method + if err := commonyaml.UnmarshalYML(kv.Value, &method); err != nil { + return fmt.Errorf("decode existing runtime method %q: %w", string(kv.Key), err) + } + if method.ResourcePath == candidatePath && strings.EqualFold(method.HTTPVerb, candidateVerb) { + return fmt.Errorf("%w: %s %s already exists at %q", ErrRouteBindingRuntimeConflict, candidateVerb, candidatePath, string(kv.Key)) Review Comment: 旧运行时路由冲突检查不完整。checkRuntimeRouteConflict 只比较 `Method.ResourcePath`。旧配置中该字段可能为空，嵌套路由的实际路径也来自 Resource 层级；另外运行时 Router 会先把路径转小写 route.go:123。因此新路由可能发布成功，但 watcher 注册时才因重复路径失败，导致 Admin 显示已发布而实际未生效。建议按 Resource 的有效路径并使用与 Router 一致的规范化规则检查，并补充空 `ResourcePath`、大小写差异、嵌套路由测试。 ########## pkg/config/schema/README.md: ########## @@ -0,0 +1,43 @@ +# Admin route binding model + +This package is an initial code-level model for replacing raw Resource and +Method YAML editing with one Admin-facing route binding. + +The supported flow is: + +```text +AdminRouteBinding + -> schema defaults and validation + -> legacy Resource + Method compilation + -> api_config.yaml preview + -> future draft and publish storage +``` + +## Scope + +The first version deliberately covers one HTTP entry mapped to one +registry-backed Dubbo method. Listener, Cluster, PluginGroup, direct provider +URLs, and the actual etcd publish transaction are outside this package. + +`Resource` and `Method` are generated Pixiu runtime structures. They are not +registered as Admin form objects. + +## Object boundary + +`AdminObject` keeps only a stable `kind`, `metadata`, and dynamic +`spec map[string]any`. `ObjectSchema` and `FieldSchema` define the form, +defaults, and validation rules for the dynamic values. + +The built-in `AdminRouteBinding` schema has five user-facing sections: + +- `entry`: HTTP protocol, path, and method. +- `target`: Dubbo application, interface, method, version, group, and cluster. +- `params`: ordered HTTP-source to Dubbo-argument mappings. +- `enabled`: whether the published route accepts requests; it defaults to `true`. +- `publish`: Admin control-plane intent; it is never emitted into Pixiu YAML. Review Comment: schema README 仍把已移除的 spec.publish 写成用户配置项，这里应改为 extensions 或明确标注已废弃。 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
