diff --git a/common/inspect/managed_config_inspections.go b/common/inspect/managed_config_inspections.go new file mode 100644 index 000000000..77992b788 --- /dev/null +++ b/common/inspect/managed_config_inspections.go @@ -0,0 +1,56 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://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 inspect + +const ( + RouterConfigRegistryKey = "router-config-registry" +) + +// RouterConfigRegistryState is the snapshot returned by the router config +// registry's Inspect method, suitable for JSON serialization (e.g. +// `ziti fabric inspect router-config-registry`). +type RouterConfigRegistryState struct { + Sealed bool `json:"sealed"` + Closed bool `json:"closed"` + Handlers []RouterConfigHandlerDetail `json:"handlers"` +} + +// RouterConfigHandlerDetail describes one registered handler: its base, the +// versions it understands, what data is currently known from each source, and +// what is currently applied. +type RouterConfigHandlerDetail struct { + BaseType string `json:"baseType"` + SupportedVersions []int `json:"supportedVersions"` + ControllerConfigs []RouterConfigVersionDetail `json:"controllerConfigs"` + LocalConfig *RouterConfigVersionDetail `json:"localConfig,omitempty"` + Applied *RouterConfigAppliedDetail `json:"applied,omitempty"` +} + +// RouterConfigVersionDetail describes a single version-of-data the registry +// knows about. Data is the parsed JSON payload, inlined for readability when +// the inspect output is itself JSON-encoded. +type RouterConfigVersionDetail struct { + Version int `json:"version"` + Data any `json:"data"` +} + +// RouterConfigAppliedDetail describes the currently-applied config for a +// handler. +type RouterConfigAppliedDetail struct { + Source string `json:"source"` + Version int `json:"version"` +} diff --git a/common/router_data_model.go b/common/router_data_model.go index 6a94a4e3a..1b5293ead 100644 --- a/common/router_data_model.go +++ b/common/router_data_model.go @@ -626,6 +626,11 @@ type RouterDataModel struct { // time. Empty for non-receiver views (controller-side, validate parsing). selfRouterId string + // routerConfigSubscriber receives notifications for Config events whose + // ConfigType.Target is "router". Set via SetRouterConfigSubscriber after + // construction; a nil-pointer Load means no dispatch. + routerConfigSubscriber atomic.Pointer[RouterConfigEventSubscriber] + lock sync.Mutex index uint64 } @@ -1003,10 +1008,16 @@ func (rdm *RouterDataModel) HandleRouterEvent(event *edge_ctrl_pb.DataState_Even if _, retained := keep[oldId]; retained { continue } - if !rdm.Configs.Has(oldId) { + cfg, ok := rdm.Configs.Get(oldId) + if !ok { continue } rdm.Configs.Remove(oldId) + // Dispatch the remove so the managed-config subscriber tears + // down the listener / dialer for this config. Without this, + // removing a Config from a router's Configs list would silently + // leave the router-side listener bound. + rdm.dispatchRouterConfigRemove(cfg.TypeId) } } @@ -1334,8 +1345,10 @@ func (rdm *RouterDataModel) HandleConfigTypeEvent(index uint64, event *edge_ctrl // during startup. func (rdm *RouterDataModel) HandleConfigEvent(index uint64, event *edge_ctrl_pb.DataState_Event, model *edge_ctrl_pb.DataState_Event_Config) { if event.Action == edge_ctrl_pb.DataState_Delete { + var removedTypeId string rdm.Configs.RemoveCb(model.Config.Id, func(key string, v *Config, exists bool) bool { if v != nil { + removedTypeId = v.TypeId v.services.IterCb(func(serviceId string, _ struct{}) { rdm.NotifyServiceOfConfigChange(serviceId, index) }) @@ -1357,7 +1370,11 @@ func (rdm *RouterDataModel) HandleConfigEvent(index uint64, event *edge_ctrl_pb. } return exists }) + if removedTypeId != "" { + rdm.dispatchRouterConfigRemove(removedTypeId) + } } else { + changed := false rdm.Configs.Upsert(model.Config.Id, nil, func(exist bool, valueInMap *Config, newValue *Config) *Config { result := &Config{ Id: model.Config.Id, @@ -1375,6 +1392,7 @@ func (rdm *RouterDataModel) HandleConfigEvent(index uint64, event *edge_ctrl_pb. } if !result.Equals(valueInMap) { + changed = true result.services.IterCb(func(serviceId string, _ struct{}) { rdm.NotifyServiceOfConfigChange(serviceId, index) }) @@ -1388,9 +1406,64 @@ func (rdm *RouterDataModel) HandleConfigEvent(index uint64, event *edge_ctrl_pb. return result }) + if changed { + rdm.dispatchRouterConfigApply(model.Config.TypeId, model.Config.DataJson) + } } } +// SetRouterConfigSubscriber registers (or clears, when s is nil) the subscriber +// that receives router-target Config events. Subsequent HandleConfigEvent calls +// dispatch through s. Replaces any previous subscriber. This sets only the +// hook; bootstrap and diff-on-resync are handled by the state manager that +// owns the subscriber lifecycle. +func (rdm *RouterDataModel) SetRouterConfigSubscriber(s RouterConfigEventSubscriber) { + if s == nil { + rdm.routerConfigSubscriber.Store(nil) + return + } + rdm.routerConfigSubscriber.Store(&s) +} + +// RouterConfigSubscriber returns the currently-registered subscriber, or nil. +func (rdm *RouterDataModel) RouterConfigSubscriber() RouterConfigEventSubscriber { + if p := rdm.routerConfigSubscriber.Load(); p != nil { + return *p + } + return nil +} + +// dispatchRouterConfigApply notifies the subscriber if the typeId resolves to +// a router-target ConfigType. No-op when no subscriber is set or the type +// isn't router-target. +func (rdm *RouterDataModel) dispatchRouterConfigApply(typeId, data string) { + sub := rdm.RouterConfigSubscriber() + if sub == nil { + return + } + ct, ok := rdm.ConfigTypes.Get(typeId) + if !ok || ct == nil || ct.Target != ConfigTypeTargetRouter { + return + } + sub.OnRouterConfigApplied(ct.Name, data) +} + +// dispatchRouterConfigRemove notifies the subscriber if the typeId resolves to +// a router-target ConfigType. ConfigType records outlive the Config records +// that reference them, so the lookup succeeds even after the Config has been +// removed. +func (rdm *RouterDataModel) dispatchRouterConfigRemove(typeId string) { + sub := rdm.RouterConfigSubscriber() + if sub == nil { + return + } + ct, ok := rdm.ConfigTypes.Get(typeId) + if !ok || ct == nil || ct.Target != ConfigTypeTargetRouter { + return + } + sub.OnRouterConfigRemoved(ct.Name) +} + func (rdm *RouterDataModel) applyUpdateServicePolicyEvent(index uint64, model *edge_ctrl_pb.DataState_Event_ServicePolicy) { servicePolicy := model.ServicePolicy rdm.ServicePolicies.Upsert(servicePolicy.Id, nil, func(exist bool, valueInMap *ServicePolicy, newValue *ServicePolicy) *ServicePolicy { diff --git a/common/router_data_model_test.go b/common/router_data_model_test.go index 1cc5761a0..cc33011b0 100644 --- a/common/router_data_model_test.go +++ b/common/router_data_model_test.go @@ -75,6 +75,39 @@ func Test_HandleRouterEvent_GCsRouterTargetOrphansOnSelf(t *testing.T) { req.True(rdm.Configs.Has("scfg-1")) } +// Test_HandleRouterEvent_GcDispatchesSubscriberRemove ensures the +// router-event-driven orphan GC dispatches OnRouterConfigRemoved through +// the subscriber. Without this, removing a Config from a router's +// Configs list would silently leave the router-side listener bound. +func Test_HandleRouterEvent_GcDispatchesSubscriberRemove(t *testing.T) { + req := require.New(t) + rdm := NewBareRouterDataModel("r1") + + seedConfigType(rdm, "router-link", ConfigTypeTargetRouter) + rdm.ConfigTypes.Set("router-link", &ConfigType{Id: "router-link", Name: "router.link.v1", Target: ConfigTypeTargetRouter}) + seedConfig(rdm, "rcfg-A", "router-link") + + rec := &recordingRouterConfigSubscriber{} + rdm.SetRouterConfigSubscriber(rec) + + rdm.Routers.Set("r1", &edge_ctrl_pb.DataState_Router{ + Id: "r1", + Configs: []string{"rcfg-A"}, + }) + + // Drop the config from the router's Configs list. + rdm.HandleRouterEvent( + &edge_ctrl_pb.DataState_Event{Action: edge_ctrl_pb.DataState_Update}, + &edge_ctrl_pb.DataState_Event_Router{Router: &edge_ctrl_pb.DataState_Router{ + Id: "r1", + Configs: []string{}, + }}, + ) + + req.False(rdm.Configs.Has("rcfg-A"), "orphaned config should be GC'd") + req.Equal([]string{"router.link.v1"}, rec.removed, "GC must dispatch a remove event to the subscriber") +} + func Test_HandleRouterEvent_NoGCForOtherRouter(t *testing.T) { req := require.New(t) rdm := NewBareRouterDataModel("r1") @@ -199,3 +232,129 @@ func Test_HandleRouterEvent_DeleteSelf(t *testing.T) { req.False(rdm.Routers.Has("r1")) req.True(rdm.Configs.Has("rcfg-A")) } + +// --- Router config subscriber dispatch --------------------------------------- + +type recordingRouterConfigSubscriber struct { + applied []appliedEntry + removed []string +} + +type appliedEntry struct { + configType string + data string +} + +func (r *recordingRouterConfigSubscriber) OnRouterConfigApplied(configType string, data string) { + r.applied = append(r.applied, appliedEntry{configType: configType, data: data}) +} + +func (r *recordingRouterConfigSubscriber) OnRouterConfigRemoved(configType string) { + r.removed = append(r.removed, configType) +} + +// configEvent builds a Config Create/Update event suitable for HandleConfigEvent. +func configEvent(action edge_ctrl_pb.DataState_Action, configId, typeId, dataJson string) (*edge_ctrl_pb.DataState_Event, *edge_ctrl_pb.DataState_Event_Config) { + model := &edge_ctrl_pb.DataState_Event_Config{ + Config: &edge_ctrl_pb.DataState_Config{ + Id: configId, + Name: configId, + TypeId: typeId, + DataJson: dataJson, + }, + } + return &edge_ctrl_pb.DataState_Event{Action: action, Model: model}, model +} + +func Test_HandleConfigEvent_DispatchesRouterTargetApply(t *testing.T) { + req := require.New(t) + rdm := NewBareRouterDataModel("r1") + rdm.ConfigTypes.Set("router-link", &ConfigType{Id: "router-link", Name: "router.link.v1", Target: ConfigTypeTargetRouter}) + rec := &recordingRouterConfigSubscriber{} + rdm.SetRouterConfigSubscriber(rec) + + event, model := configEvent(edge_ctrl_pb.DataState_Create, "cfg-A", "router-link", `{"k":"v"}`) + rdm.HandleConfigEvent(1, event, model) + + req.Len(rec.applied, 1) + req.Equal("router.link.v1", rec.applied[0].configType) + req.Equal(`{"k":"v"}`, rec.applied[0].data) + req.Empty(rec.removed) +} + +func Test_HandleConfigEvent_DispatchesRouterTargetRemove(t *testing.T) { + req := require.New(t) + rdm := NewBareRouterDataModel("r1") + rdm.ConfigTypes.Set("router-link", &ConfigType{Id: "router-link", Name: "router.link.v1", Target: ConfigTypeTargetRouter}) + seedConfig(rdm, "cfg-A", "router-link") + rec := &recordingRouterConfigSubscriber{} + rdm.SetRouterConfigSubscriber(rec) + + event, model := configEvent(edge_ctrl_pb.DataState_Delete, "cfg-A", "router-link", "") + rdm.HandleConfigEvent(1, event, model) + + req.Empty(rec.applied) + req.Equal([]string{"router.link.v1"}, rec.removed) +} + +func Test_HandleConfigEvent_NonRouterTargetNotDispatched(t *testing.T) { + req := require.New(t) + rdm := NewBareRouterDataModel("r1") + rdm.ConfigTypes.Set("service-cfg", &ConfigType{Id: "service-cfg", Name: "service.cfg.v1", Target: "service"}) + rec := &recordingRouterConfigSubscriber{} + rdm.SetRouterConfigSubscriber(rec) + + event, model := configEvent(edge_ctrl_pb.DataState_Create, "cfg-X", "service-cfg", `{}`) + rdm.HandleConfigEvent(1, event, model) + + req.Empty(rec.applied) + req.Empty(rec.removed) +} + +func Test_HandleConfigEvent_NoSubscriberNoPanic(t *testing.T) { + rdm := NewBareRouterDataModel("r1") + rdm.ConfigTypes.Set("router-link", &ConfigType{Id: "router-link", Name: "router.link.v1", Target: ConfigTypeTargetRouter}) + + event, model := configEvent(edge_ctrl_pb.DataState_Create, "cfg-A", "router-link", `{}`) + // Must not panic with no subscriber set. + rdm.HandleConfigEvent(1, event, model) + + delEvent, delModel := configEvent(edge_ctrl_pb.DataState_Delete, "cfg-A", "router-link", "") + rdm.HandleConfigEvent(2, delEvent, delModel) +} + +func Test_HandleConfigEvent_UnknownConfigTypeNotDispatched(t *testing.T) { + req := require.New(t) + rdm := NewBareRouterDataModel("r1") + // No ConfigType registered: dispatcher must skip silently. + rec := &recordingRouterConfigSubscriber{} + rdm.SetRouterConfigSubscriber(rec) + + event, model := configEvent(edge_ctrl_pb.DataState_Create, "cfg-A", "missing-type", `{}`) + rdm.HandleConfigEvent(1, event, model) + + req.Empty(rec.applied) + req.Empty(rec.removed) +} + +func Test_HandleConfigEvent_NoChangeNotDispatched(t *testing.T) { + req := require.New(t) + rdm := NewBareRouterDataModel("r1") + rdm.ConfigTypes.Set("router-link", &ConfigType{Id: "router-link", Name: "router.link.v1", Target: ConfigTypeTargetRouter}) + rec := &recordingRouterConfigSubscriber{} + rdm.SetRouterConfigSubscriber(rec) + + event, model := configEvent(edge_ctrl_pb.DataState_Create, "cfg-A", "router-link", `{"k":"v"}`) + rdm.HandleConfigEvent(1, event, model) + req.Len(rec.applied, 1, "first event should dispatch") + + // Identical second event must not redispatch. + rdm.HandleConfigEvent(2, event, model) + req.Len(rec.applied, 1, "identical-data event should not redispatch") + + // Update with different data dispatches. + changedEvent, changedModel := configEvent(edge_ctrl_pb.DataState_Update, "cfg-A", "router-link", `{"k":"v2"}`) + rdm.HandleConfigEvent(3, changedEvent, changedModel) + req.Len(rec.applied, 2, "changed-data event should dispatch") + req.Equal(`{"k":"v2"}`, rec.applied[1].data) +} diff --git a/common/subscriber.go b/common/subscriber.go index a8ee6deca..ca72b448a 100644 --- a/common/subscriber.go +++ b/common/subscriber.go @@ -477,6 +477,23 @@ type IdentityEventSubscriber interface { NotifyServiceChange(state *IdentityState, previousService, service *IdentityService, eventType ServiceEventType) } +// RouterConfigEventSubscriber receives notifications about router-managed +// configuration changes. The router-side RDM dispatches to a single subscriber +// for any Config whose ConfigType.Target is "router". Configs targeted at +// services or other entities do not flow through this interface. +// +// OnRouterConfigApplied is called for both Create and Update events; consumers +// (e.g. the managedconfig.Registry) handle the same-data no-op case +// themselves. OnRouterConfigRemoved is called for Delete events and for +// configs that disappear during a full-state resync. +// +// configType is the ConfigType.Name (e.g. "router.link.v1"); data is the +// raw JSON payload from the controller. +type RouterConfigEventSubscriber interface { + OnRouterConfigApplied(configType string, data string) + OnRouterConfigRemoved(configType string) +} + // subscriberEvent is an internal interface for events that need to be processed to update // identity subscriptions. These events are queued and processed asynchronously. type subscriberEvent interface { diff --git a/go.mod b/go.mod index 2ee7ab42d..eb5b91fa4 100644 --- a/go.mod +++ b/go.mod @@ -66,7 +66,7 @@ require ( github.com/openziti/channel/v5 v5.0.15 github.com/openziti/cobra-to-md v1.0.1 github.com/openziti/edge-api v0.32.0 - github.com/openziti/foundation/v2 v2.0.95 + github.com/openziti/foundation/v2 v2.0.99 github.com/openziti/identity v1.0.133 github.com/openziti/jwks v1.0.6 github.com/openziti/metrics v1.4.5 diff --git a/go.sum b/go.sum index b2fc0f8d3..6fc285918 100644 --- a/go.sum +++ b/go.sum @@ -543,8 +543,8 @@ github.com/openziti/cobra-to-md v1.0.1 h1:WRinNoIRmwWUSJm+pSNXMjOrtU48oxXDZgeCYQ github.com/openziti/cobra-to-md v1.0.1/go.mod h1:FjCpk/yzHF7/r28oSTNr5P57yN5VolpdAtS/g7KNi2c= github.com/openziti/edge-api v0.32.0 h1:+hnr0kzk5/jUzMPWl0TViyROUahccKnDyxlFCkAJqfM= github.com/openziti/edge-api v0.32.0/go.mod h1:wFExSB9pO7yEwVVuz4AatZeU++dgcjy9pkzeCFLXeno= -github.com/openziti/foundation/v2 v2.0.95 h1:ZB6AeclGCfuI2l3c3PN7stKLp6Ck+P0wBuokY4pp7qo= -github.com/openziti/foundation/v2 v2.0.95/go.mod h1:yhdRzmEzrHxqG8frg2fMmSpR5+iZI1tL/teeaAoWl7U= +github.com/openziti/foundation/v2 v2.0.99 h1:Yhhfv8Zl5Yu0S5EjM6dbJmyRVbEexLJ7iyZo41mvbxE= +github.com/openziti/foundation/v2 v2.0.99/go.mod h1:yhdRzmEzrHxqG8frg2fMmSpR5+iZI1tL/teeaAoWl7U= github.com/openziti/go-term-markdown v1.0.1 h1:9uzMpK4tav6OtvRxRt99WwPTzAzCh+Pj9zWU2FBp3Qg= github.com/openziti/go-term-markdown v1.0.1/go.mod h1:aIhR12zlROVqr7x51DtBALaQZQ3FofBuKDWqCfaz3oo= github.com/openziti/identity v1.0.133 h1:rOFsV+sQVpy3FJqtShF7NYQxR+678FDQfFUS3ziGNeo= @@ -555,8 +555,6 @@ github.com/openziti/metrics v1.4.5 h1:p51HYSQyqaDafizPWluhYjU/rVfY52snER8/BHpfPr github.com/openziti/metrics v1.4.5/go.mod h1:MOLcoTxhPNla6+NWUCMVTnl1PNqTU40qrbKVa/lVVgg= github.com/openziti/runzmd v1.0.90 h1:fasGlaq9xV+zohEGDC7Q0nOLA0n8Kpfccribc+VGQVw= github.com/openziti/runzmd v1.0.90/go.mod h1:ma3b7UdVYAC9ZCVUSjevUH/7yz9asI537XPsEh7+j14= -github.com/openziti/sdk-golang/v2 v2.0.0-pre1 h1:oB875uZ+oRaIR5s4CmGH116fo1L7klb+LaHlhA7V60k= -github.com/openziti/sdk-golang/v2 v2.0.0-pre1/go.mod h1:y0Kvj1jQ6FITI64T5mHAx64rtrZaXLCY1NaFrSRUboI= github.com/openziti/sdk-golang/v2 v2.0.0-pre1.0.20260624192439-7081a4d51c11 h1:KfLYEDZfsHVuVdlKvmphoWWpbZNea1RZn93JbZenSBo= github.com/openziti/sdk-golang/v2 v2.0.0-pre1.0.20260624192439-7081a4d51c11/go.mod h1:y0Kvj1jQ6FITI64T5mHAx64rtrZaXLCY1NaFrSRUboI= github.com/openziti/secretstream v0.1.51 h1:j/rMfIzBNqZD5a1EKV8J4Z5QeaoSK56s/zxvvB08eSA= @@ -565,8 +563,6 @@ github.com/openziti/transport/v2 v2.0.216 h1:/2ALqUaeDzOfvZwzGSPNtWqMu8srgA1AERF github.com/openziti/transport/v2 v2.0.216/go.mod h1:Kh4dP6FtAUbiJvTmRcMJvLVCBmpQOpvUT1jMUCLDxQg= github.com/openziti/x509-claims v1.0.3 h1:HNdQ8Nf1agB3lBs1gahcO6zfkeS4S5xoQ2/PkY4HRX0= github.com/openziti/x509-claims v1.0.3/go.mod h1:Z0WIpBm6c4ecrpRKrou6Gk2wrLWxJO/+tuUwKh8VewE= -github.com/openziti/xweb/v3 v3.0.4 h1:sKQJOJDQiWcTkuc37L7oc0Z0jQ5Poh2zaa5raiP0GLc= -github.com/openziti/xweb/v3 v3.0.4/go.mod h1:xb9m4ySnKilHM4wbUC2MnGfilDu/kKw97DWL13ZTvqk= github.com/openziti/xweb/v3 v3.0.5-0.20260618140905-54cc19903f1c h1:ItIhxwbkKVrDrqJlPrb0RMwEYircOSyDkQWo/KcVjx4= github.com/openziti/xweb/v3 v3.0.5-0.20260618140905-54cc19903f1c/go.mod h1:xb9m4ySnKilHM4wbUC2MnGfilDu/kKw97DWL13ZTvqk= github.com/orcaman/concurrent-map/v2 v2.0.1 h1:jOJ5Pg2w1oeB6PeDurIYf6k9PQ+aTITr/6lP/L/zp6c= diff --git a/router/env/config.go b/router/env/config.go index 0bbf3a878..c50adb74a 100644 --- a/router/env/config.go +++ b/router/env/config.go @@ -151,9 +151,9 @@ type Config struct { Listeners []*CtrlListenerConfig } Link struct { - Listeners []map[interface{}]interface{} - Dialers []map[interface{}]interface{} - Heartbeats channel.HeartbeatOptions + Listeners []map[interface{}]interface{} + Dialers []map[interface{}]interface{} + Heartbeats channel.HeartbeatOptions PayloadSenderQueueSize int AckSenderQueueSize int } @@ -184,6 +184,7 @@ type Config struct { Plugins []string Edge *EdgeConfig IfaceDiscovery InterfaceDiscoveryConfig + ManagedConfig ManagedConfigOptions Src map[interface{}]interface{} path string } @@ -1012,6 +1013,10 @@ func LoadConfigWithOptions(path string, loadIdentity bool) (*Config, error) { } } + if err = LoadManagedConfigFromMap(cfgmap, &cfg.ManagedConfig); err != nil { + return nil, err + } + return cfg, nil } diff --git a/router/env/config_managed.go b/router/env/config_managed.go new file mode 100644 index 000000000..92950056a --- /dev/null +++ b/router/env/config_managed.go @@ -0,0 +1,121 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://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 env + +import ( + "fmt" + "strings" +) + +// ManagedConfigAllowAll is the special allow-list entry that accepts every +// config type the controller sends. +const ManagedConfigAllowAll = "all" + +// ManagedConfigOptions holds the local router's allow-list for +// controller-managed config types. See doc/design/ctrl-managed-router-config.md +// "Local Config Type Allow-list" for the contract. +type ManagedConfigOptions struct { + // Allow is the operator's allow-list, exactly as written in YAML. + // - empty/nil -> controller-managed config is disabled + // - ["all"] -> every config type is accepted + // - otherwise -> only matching entries are accepted (see IsAllowed) + Allow []string +} + +// LoadManagedConfigFromMap parses the optional `managedConfig` section out of +// a router YAML config map and writes the result into options. Resets +// options.Allow first, so the function is safe to call repeatedly on the same +// struct. An absent section leaves options.Allow nil (controller-managed +// config disabled, which is the safe default). +func LoadManagedConfigFromMap(cfgmap map[interface{}]interface{}, options *ManagedConfigOptions) error { + options.Allow = nil + value, found := cfgmap["managedConfig"] + if !found || value == nil { + return nil + } + submap, ok := value.(map[interface{}]interface{}) + if !ok { + return fmt.Errorf("[managedConfig] must be a map, got %T", value) + } + allowVal, found := submap["allow"] + if !found || allowVal == nil { + return nil + } + allowList, ok := allowVal.([]interface{}) + if !ok { + return fmt.Errorf("[managedConfig/allow] must be a list, got %T", allowVal) + } + for i, item := range allowList { + // This is a security boundary, so reject non-string entries rather than + // coercing values like `true` or `123` into config type names. + s, ok := item.(string) + if !ok { + return fmt.Errorf("[managedConfig/allow][%d] must be a string, got %T", i, item) + } + options.Allow = append(options.Allow, s) + } + return nil +} + +// IsAllowed reports whether the given config type name should be accepted from +// the controller. Returns false when the allow-list is empty (managed config +// disabled). Returns true if the list contains "all". Otherwise returns true +// when the type matches an entry exactly OR is a versioned child of an entry, +// i.e. the type equals entry + ".v" + . No other descendants match. +// +// Examples (with Allow = ["router.link", "router.xgress.proxy"]): +// +// "router.link" -> true (exact) +// "router.link.v1" -> true (versioned child) +// "router.link.v2" -> true (versioned child) +// "router.link.v1.x" -> false (suffix after .v must be digits only) +// "router.link.subpath" -> false (not a versioned child) +// "router.xgress.proxy.v1" -> true +// "router.xgress.tunnel.v1" -> false (not under any allowed entry) +// "router.linker.v1" -> false (not a child; the trailing dot matters) +func (m *ManagedConfigOptions) IsAllowed(configType string) bool { + if m == nil || len(m.Allow) == 0 { + return false + } + for _, entry := range m.Allow { + if entry == ManagedConfigAllowAll { + return true + } + if entry == configType { + return true + } + if isVersionedChild(configType, entry) { + return true + } + } + return false +} + +// isVersionedChild reports whether configType is of the form entry + ".v" + N, +// where N is one or more decimal digits. +func isVersionedChild(configType, entry string) bool { + suffix, ok := strings.CutPrefix(configType, entry+".v") + if !ok || suffix == "" { + return false + } + for _, r := range suffix { + if r < '0' || r > '9' { + return false + } + } + return true +} diff --git a/router/env/config_managed_test.go b/router/env/config_managed_test.go new file mode 100644 index 000000000..032e0e708 --- /dev/null +++ b/router/env/config_managed_test.go @@ -0,0 +1,167 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://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 env + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func loadManaged(t *testing.T, src map[interface{}]interface{}) *ManagedConfigOptions { + t.Helper() + opts := &ManagedConfigOptions{} + err := LoadManagedConfigFromMap(src, opts) + require.NoError(t, err) + return opts +} + +func Test_ManagedConfig_AbsentSection(t *testing.T) { + req := require.New(t) + opts := loadManaged(t, map[interface{}]interface{}{}) + req.Empty(opts.Allow) + req.False(opts.IsAllowed("router.link.v1")) + req.False(opts.IsAllowed("anything")) +} + +func Test_ManagedConfig_NilSection(t *testing.T) { + req := require.New(t) + opts := loadManaged(t, map[interface{}]interface{}{"managedConfig": nil}) + req.Empty(opts.Allow) + req.False(opts.IsAllowed("router.link.v1")) +} + +func Test_ManagedConfig_EmptyAllow(t *testing.T) { + req := require.New(t) + opts := loadManaged(t, map[interface{}]interface{}{ + "managedConfig": map[interface{}]interface{}{ + "allow": []interface{}{}, + }, + }) + req.Empty(opts.Allow) + req.False(opts.IsAllowed("router.link.v1")) +} + +func Test_ManagedConfig_RejectsNonStringAllowEntry(t *testing.T) { + req := require.New(t) + // The allow-list is a security boundary; non-string YAML values must be + // rejected rather than coerced into config type names. + for _, bad := range []interface{}{true, 123} { + opts := &ManagedConfigOptions{} + err := LoadManagedConfigFromMap(map[interface{}]interface{}{ + "managedConfig": map[interface{}]interface{}{ + "allow": []interface{}{"router.link", bad}, + }, + }, opts) + req.Error(err) + req.Contains(err.Error(), "managedConfig/allow") + } +} + +func Test_ManagedConfig_AllKeyword(t *testing.T) { + req := require.New(t) + opts := loadManaged(t, map[interface{}]interface{}{ + "managedConfig": map[interface{}]interface{}{ + "allow": []interface{}{"all"}, + }, + }) + req.True(opts.IsAllowed("router.link.v1")) + req.True(opts.IsAllowed("router.xgress.tunnel.v1")) + req.True(opts.IsAllowed("anything-at-all")) +} + +func Test_ManagedConfig_ExactMatch(t *testing.T) { + req := require.New(t) + opts := loadManaged(t, map[interface{}]interface{}{ + "managedConfig": map[interface{}]interface{}{ + "allow": []interface{}{"router.link.v1"}, + }, + }) + req.True(opts.IsAllowed("router.link.v1")) + req.False(opts.IsAllowed("router.link.v2")) + req.False(opts.IsAllowed("router.forwarder")) +} + +func Test_ManagedConfig_FamilyPrefix(t *testing.T) { + req := require.New(t) + opts := loadManaged(t, map[interface{}]interface{}{ + "managedConfig": map[interface{}]interface{}{ + "allow": []interface{}{"router.link", "router.xgress.proxy"}, + }, + }) + req.True(opts.IsAllowed("router.link.v1")) + req.True(opts.IsAllowed("router.link.v2")) + req.True(opts.IsAllowed("router.link.v42")) + req.True(opts.IsAllowed("router.xgress.proxy.v1")) + req.False(opts.IsAllowed("router.xgress.tunnel.v1")) + req.False(opts.IsAllowed("router.linker.v1"), "trailing dot should prevent false-prefix match") + req.True(opts.IsAllowed("router.link"), "an exact match against the bare family entry is also allowed") +} + +func Test_ManagedConfig_VersionSuffixIsStrict(t *testing.T) { + req := require.New(t) + opts := loadManaged(t, map[interface{}]interface{}{ + "managedConfig": map[interface{}]interface{}{ + "allow": []interface{}{"router.link"}, + }, + }) + // only entry + ".v" + digits matches; other descendants do not + req.False(opts.IsAllowed("router.link.subpath"), "non-version descendants must not match") + req.False(opts.IsAllowed("router.link.v1.x"), "version suffix must be the trailing segment") + req.False(opts.IsAllowed("router.link.v"), "empty version number must not match") + req.False(opts.IsAllowed("router.link.va"), "version suffix must be digits") + req.False(opts.IsAllowed("router.link.v1a"), "trailing non-digit must not match") + req.False(opts.IsAllowed("router.link.V1"), "version marker is case-sensitive") +} + +func Test_ManagedConfig_MixedAllAndExplicit(t *testing.T) { + req := require.New(t) + // "all" anywhere short-circuits; the rest of the list is irrelevant. + opts := loadManaged(t, map[interface{}]interface{}{ + "managedConfig": map[interface{}]interface{}{ + "allow": []interface{}{"router.link", "all"}, + }, + }) + req.True(opts.IsAllowed("router.xgress.tunnel.v1")) + req.True(opts.IsAllowed("anything")) +} + +func Test_ManagedConfig_InvalidShape(t *testing.T) { + req := require.New(t) + + // managedConfig must be a map + err := LoadManagedConfigFromMap(map[interface{}]interface{}{ + "managedConfig": "not a map", + }, &ManagedConfigOptions{}) + req.Error(err) + req.Contains(err.Error(), "[managedConfig]") + + // allow must be a list + err = LoadManagedConfigFromMap(map[interface{}]interface{}{ + "managedConfig": map[interface{}]interface{}{ + "allow": "not a list", + }, + }, &ManagedConfigOptions{}) + req.Error(err) + req.Contains(err.Error(), "[managedConfig/allow]") +} + +func Test_ManagedConfig_NilReceiver_IsAllowed(t *testing.T) { + req := require.New(t) + var opts *ManagedConfigOptions + req.False(opts.IsAllowed("anything"), "nil receiver should be the disabled state") +} diff --git a/router/env/env.go b/router/env/env.go index e6dfc2b35..7eb160506 100644 --- a/router/env/env.go +++ b/router/env/env.go @@ -28,6 +28,7 @@ import ( "github.com/openziti/ziti/v2/common" "github.com/openziti/ziti/v2/common/pb/ctrl_pb" "github.com/openziti/ziti/v2/common/servermetrics" + "github.com/openziti/ziti/v2/router/managedconfig" "github.com/openziti/ziti/v2/router/xgress_router" "github.com/openziti/ziti/v2/router/xlink" ) @@ -49,6 +50,7 @@ type RouterEnv interface { GetVersionInfo() versions.VersionProvider GetRouterDataModel() *common.RouterDataModel WithRouterDataModel(f func(*common.RouterDataModel) error) error + GetRouterConfigRegistry() *managedconfig.Registry GetConnectEventsConfig() *ConnectEventsConfig IsRouterDataModelRequired() bool MarkRouterDataModelRequired() diff --git a/router/inspect/inspect.go b/router/inspect/inspect.go index 0daee0bff..8d9177d1e 100644 --- a/router/inspect/inspect.go +++ b/router/inspect/inspect.go @@ -144,6 +144,8 @@ func (context *inspectRequestContext) processLocal() { } else if lc == "router-data-model" { result := context.handler.env.GetRouterDataModel() context.handleJsonResponse(requested, result.ToMap()) + } else if lc == inspect.RouterConfigRegistryKey { + context.handleJsonResponse(requested, context.handler.env.GetRouterConfigRegistry().Inspect()) } else if lc == "router-data-model-index" { idx := context.handler.env.GetRouterDataModel().CurrentIndex() data := map[string]any{ diff --git a/router/managedconfig/handler.go b/router/managedconfig/handler.go new file mode 100644 index 000000000..72cc06ad2 --- /dev/null +++ b/router/managedconfig/handler.go @@ -0,0 +1,56 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://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 managedconfig contains the router-side machinery for receiving +// controller-managed configuration: a handler interface that subsystems +// implement and a registry that routes Config events from the RDM, picks the +// highest-version a handler supports, and drives Apply/Remove with rollback +// semantics. See doc/design/ctrl-managed-router-config.md for the full design. +// +// Config types follow the convention `.v` where N is a positive +// integer (e.g. "router.link.v2"). The registry parses incoming names, keys +// state by base type, and selects the highest version that is both available +// from the controller and supported by the registered handler. +package managedconfig + +// ConfigHandler is implemented by router subsystems that accept controller- +// managed configuration. The registry routes Config events to the handler +// whose BaseType matches. +// +// Implementations must be safe to call from a single goroutine; the registry +// serializes Apply / Remove for a given handler. +type ConfigHandler interface { + // BaseType returns the un-versioned config type family this handler owns, + // e.g. "router.link". Every config type whose name parses to this base + // will be routed to this handler. + BaseType() string + + // SupportedVersions returns the integer versions of BaseType this handler + // can apply. Order is not significant; the registry picks max(supported ∩ + // available) on every reconcile. + SupportedVersions() []int + + // Apply is called when the registry has selected an active version for + // this handler. data is the raw JSON payload from the controller, + // untransformed. Returning an error triggers rollback (or, if no previous + // config exists, Remove). + Apply(version int, data string) error + + // Remove is called when no version of this handler's BaseType is + // currently available, or when both a fresh Apply and rollback Apply have + // failed and the registry is forcing the subsystem offline. + Remove() error +} diff --git a/router/managedconfig/registry.go b/router/managedconfig/registry.go new file mode 100644 index 000000000..cf1ddd352 --- /dev/null +++ b/router/managedconfig/registry.go @@ -0,0 +1,623 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://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 managedconfig + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/michaelquigley/pfxlog" + "github.com/openziti/foundation/v2/goroutines" + "github.com/openziti/ziti/v2/common/inspect" +) + +const ( + // reconcilePoolIdleTime is how long a handler's single reconcile worker + // lives while idle before exiting. With one worker per handler this just + // avoids parking a goroutine between infrequent config changes. + reconcilePoolIdleTime = 10 * time.Second + + // closeDrainTimeout bounds how long Close waits for an in-flight reconcile + // to finish, so a wedged handler can't block router shutdown indefinitely. + closeDrainTimeout = 10 * time.Second + + // waitForIdleTimeout bounds WaitForIdle so a stuck reconcile surfaces as a + // failed assertion rather than a hung test. + waitForIdleTimeout = 30 * time.Second +) + +// ConfigSource identifies where a Config event originated. The registry +// tracks data per source and resolves precedence at reconcile time: local +// wins entirely at the base level (if the operator set anything locally for +// a given base, the controller's versions are ignored for that base). +type ConfigSource int + +const ( + // SourceController means the data came from the controller via the RDM. + SourceController ConfigSource = iota + // SourceLocal means the data came from the router's local config file. + SourceLocal +) + +// String returns the lowercase name of the source, suitable for diagnostics. +func (s ConfigSource) String() string { + switch s { + case SourceController: + return "controller" + case SourceLocal: + return "local" + default: + return fmt.Sprintf("unknown(%d)", int(s)) + } +} + +// AlertCallback is invoked when the registry encounters a non-recoverable +// outcome (Apply fails AND rollback fails, or Remove fails). Phase 3c will +// wire this to a controller-alerting transport; the default implementation +// logs. +type AlertCallback func(baseType, detail string) + +// ErrHandlerAlreadyRegistered is returned by Register when a handler is +// already registered for the same BaseType. +var ErrHandlerAlreadyRegistered = errors.New("a handler is already registered for this base type") + +// ErrNoHandlerRegistered is returned by Apply/Remove when no handler has been +// registered for the requested config type's base. +var ErrNoHandlerRegistered = errors.New("no handler registered for config type's base") + +// ParseConfigType parses a versioned config type name like "router.link.v2" +// into ("router.link", 2). Returns an error if the name doesn't end in +// ".v". +func ParseConfigType(name string) (baseType string, version int, err error) { + idx := strings.LastIndex(name, ".v") + if idx < 1 { + return "", 0, fmt.Errorf("config type %q is not of the form .v", name) + } + baseType = name[:idx] + versionStr := name[idx+2:] + if versionStr == "" { + return "", 0, fmt.Errorf("config type %q has empty version", name) + } + version, err = strconv.Atoi(versionStr) + if err != nil { + return "", 0, fmt.Errorf("config type %q has non-integer version %q", name, versionStr) + } + if version <= 0 { + return "", 0, fmt.Errorf("config type %q has non-positive version %d", name, version) + } + return baseType, version, nil +} + +// appliedState records what a handler currently has active. version == 0 means +// nothing is applied; source is meaningful only when version > 0. +type appliedState struct { + source ConfigSource + version int + data string +} + +// localEntry is the local config currently in effect for a handler base. +// Stored as a pointer on handlerEntry so nil unambiguously means "no local +// config." version is the JSON schema version the YAML translator emitted +// (typically the newest the build supports); data is the raw JSON. +type localEntry struct { + version int + data string +} + +// handlerEntry binds a registered handler with the per-handler pool that +// serializes reconciles for that handler family, the set of currently-known +// data versions, and the currently-applied state. All fields except `pool` +// are guarded by Registry.mu; pool is set once at registration and read-only +// thereafter (it has its own internal synchronization). +// +// Controller data can carry multiple versions simultaneously (e.g. v1 and v2 +// both flowing during a rollout). Local data is always at most one +// (version, data) pair — the local YAML file expresses one effective config +// per subsystem, not a multi-version set — so we store it as a single +// pointer rather than a map. +type handlerEntry struct { + handler ConfigHandler + controllerVersions map[int]string // version -> data + local *localEntry // nil means no local config set + applied appliedState + + // pool runs this handler's reconciles. It has one worker, so reconciles for + // the handler never overlap, and a queue of one, so a burst of events + // collapses to at most one running plus one pending pass (further + // submissions are dropped, since each pass reads the latest state). Handlers + // have independent pools and reconcile in parallel. + pool goroutines.Pool +} + +// Registry holds router subsystems that accept controller-managed +// configuration and routes Config events to them. It implements multi-version +// selection (highest int wins among versions a handler supports), source-aware +// precedence (local config wins over controller), and the rollback contract +// from doc/design/ctrl-managed-router-config.md. +// +// Lifecycle: +// +// 1. Construct with NewRegistry. +// 2. Subsystems Register their handlers. +// 3. Caller invokes Seal. After Seal, Register panics. Apply / Remove +// before Seal also panic. +// 4. Config events arrive via ApplyController / ApplyLocal (and the +// matching Remove* methods). +// 5. Close drains in-flight reconciles and stops the registry. +// +// Apply / Remove return as soon as the registry has updated its shared state; +// the actual handler.Apply / handler.Remove call runs on the handler's reconcile +// pool so slow handlers don't back up the caller. A handler's reconciles are +// serialized by its single-worker pool; different handlers reconcile in parallel. +type Registry struct { + mu sync.Mutex + handlers map[string]*handlerEntry // baseType -> entry + alert AlertCallback + + sealed atomic.Bool + closed atomic.Bool +} + +// NewRegistry creates a new Registry. If alert is nil, a logging alerter is +// used. +func NewRegistry(alert AlertCallback) *Registry { + if alert == nil { + alert = defaultAlert + } + return &Registry{ + handlers: map[string]*handlerEntry{}, + alert: alert, + } +} + +func defaultAlert(baseType, detail string) { + pfxlog.Logger().WithField("baseType", baseType).Warn(detail) +} + +// Register associates the handler with its BaseType. Returns +// ErrHandlerAlreadyRegistered if a different handler is already registered +// for that base. Panics if called after Seal. +func (self *Registry) Register(handler ConfigHandler) error { + if self.sealed.Load() { + panic(fmt.Sprintf("managedconfig.Registry.Register called after Seal for base %q", handler.BaseType())) + } + + base := handler.BaseType() + + self.mu.Lock() + defer self.mu.Unlock() + + if existing, ok := self.handlers[base]; ok && existing.handler != handler { + return fmt.Errorf("%w: %s", ErrHandlerAlreadyRegistered, base) + } + if _, ok := self.handlers[base]; !ok { + pool, err := newReconcilePool() + if err != nil { + return err + } + self.handlers[base] = &handlerEntry{ + handler: handler, + controllerVersions: map[int]string{}, + pool: pool, + } + } + return nil +} + +// newReconcilePool builds the per-handler pool: a single worker (so a handler's +// reconciles never overlap) fed by a one-deep queue (so bursts coalesce), with +// the worker spun up on demand and reaped when idle. +func newReconcilePool() (goroutines.Pool, error) { + return goroutines.NewPool(goroutines.PoolConfig{ + QueueSize: 1, + MinWorkers: 0, + MaxWorkers: 1, + IdleTime: reconcilePoolIdleTime, + }) +} + +// Seal marks the registration phase complete. After Seal, calls to Register +// panic. Apply / Remove may only be called after Seal. +func (self *Registry) Seal() { + self.sealed.Store(true) +} + +// Handler returns the handler registered for the BaseType extracted from +// configType, or nil if none is registered. +func (self *Registry) Handler(configType string) ConfigHandler { + base, _, err := ParseConfigType(configType) + if err != nil { + return nil + } + self.mu.Lock() + defer self.mu.Unlock() + if entry := self.handlers[base]; entry != nil { + return entry.handler + } + return nil +} + +// ApplyController records the controller's most recent data for configType +// and spawns a goroutine to reconcile the owning handler. Returns parse +// errors synchronously, ErrNoHandlerRegistered when no handler owns the +// base, or nil. Panics if called pre-Seal. +func (self *Registry) ApplyController(configType string, data string) error { + if !self.sealed.Load() { + panic("managedconfig.Registry.ApplyController called before Seal") + } + base, version, err := ParseConfigType(configType) + if err != nil { + return err + } + + self.mu.Lock() + entry, ok := self.handlers[base] + if !ok { + self.mu.Unlock() + return fmt.Errorf("%w: %s", ErrNoHandlerRegistered, base) + } + entry.controllerVersions[version] = data + self.mu.Unlock() + + self.queueReconcile(entry) + return nil +} + +// RemoveController drops a specific (base, version) entry from the +// controller-source set. Other controller versions for the same base +// remain. If local data is set for the base, the handler is unaffected +// (local was already winning). Otherwise the handler reconciles to whatever +// is left. +func (self *Registry) RemoveController(configType string) error { + if !self.sealed.Load() { + panic("managedconfig.Registry.RemoveController called before Seal") + } + base, version, err := ParseConfigType(configType) + if err != nil { + return err + } + + self.mu.Lock() + entry, ok := self.handlers[base] + if !ok { + self.mu.Unlock() + return fmt.Errorf("%w: %s", ErrNoHandlerRegistered, base) + } + delete(entry.controllerVersions, version) + self.mu.Unlock() + + self.queueReconcile(entry) + return nil +} + +// ApplyLocal records the local config file's data for configType and spawns +// a goroutine to reconcile the owning handler. Local config is always a +// single (version, data) pair per base — repeated calls replace any prior +// local entry. Local takes precedence over controller versions for the +// same base, so as long as a local entry exists, the controller's data is +// ignored. Panics if called pre-Seal. +func (self *Registry) ApplyLocal(configType string, data string) error { + if !self.sealed.Load() { + panic("managedconfig.Registry.ApplyLocal called before Seal") + } + base, version, err := ParseConfigType(configType) + if err != nil { + return err + } + + self.mu.Lock() + entry, ok := self.handlers[base] + if !ok { + self.mu.Unlock() + return fmt.Errorf("%w: %s", ErrNoHandlerRegistered, base) + } + entry.local = &localEntry{version: version, data: data} + self.mu.Unlock() + + self.queueReconcile(entry) + return nil +} + +// RemoveLocal clears the local config for the given base. Takes a base type +// rather than a configType because the version is meaningless for local +// removal — there's at most one local entry per base, regardless of which +// version it was set at. When local is cleared, the controller's highest +// supported version becomes effective. Panics if called pre-Seal. +func (self *Registry) RemoveLocal(baseType string) error { + if !self.sealed.Load() { + panic("managedconfig.Registry.RemoveLocal called before Seal") + } + if baseType == "" { + return errors.New("RemoveLocal requires a non-empty base type") + } + + self.mu.Lock() + entry, ok := self.handlers[baseType] + if !ok { + self.mu.Unlock() + return fmt.Errorf("%w: %s", ErrNoHandlerRegistered, baseType) + } + entry.local = nil + self.mu.Unlock() + + self.queueReconcile(entry) + return nil +} + +// Close marks the registry shut down and blocks until every in-flight +// reconcile has finished, bounded by closeDrainTimeout per handler. After +// Close, future Apply / Remove calls still update state but do not run new +// reconciles. +func (self *Registry) Close() { + self.closed.Store(true) + + pools := self.snapshotPools() + + // Stop every pool first so in-flight reconciles across all handlers drain + // in parallel, then wait for each to finish. + for _, pool := range pools { + pool.Shutdown() + } + for _, pool := range pools { + if err := pool.ShutdownAndWait(closeDrainTimeout); err != nil { + pfxlog.Logger().WithError(err).Warn("timed out draining managed-config reconcile pool during close") + } + } +} + +// WaitForIdle blocks until every handler's reconcile pool has no outstanding +// work. Useful in tests to assert handler effects synchronously. +func (self *Registry) WaitForIdle() { + for _, pool := range self.snapshotPools() { + if err := pool.AwaitIdle(waitForIdleTimeout); err != nil { + pfxlog.Logger().WithError(err).Warn("timed out waiting for managed-config reconcile pool to become idle") + } + } +} + +func (self *Registry) snapshotPools() []goroutines.Pool { + self.mu.Lock() + defer self.mu.Unlock() + pools := make([]goroutines.Pool, 0, len(self.handlers)) + for _, entry := range self.handlers { + pools = append(pools, entry.pool) + } + return pools +} + +// queueReconcile submits a reconcile pass for the handler to its pool. The pool +// coalesces: while a reconcile runs, at most one more is queued and any further +// submissions return QueueFullError and are dropped, which is safe because the +// queued pass reads the latest state when it runs. A PoolStoppedError means +// Close has run. Either way there's nothing more to do, so the error is ignored. +func (self *Registry) queueReconcile(entry *handlerEntry) { + _ = entry.pool.QueueOrError(func() { + self.reconcile(entry) + }) +} + +func (self *Registry) reconcile(entry *handlerEntry) { + handler := entry.handler + + self.mu.Lock() + prev := entry.applied + nextSource, nextVersion, nextData, hasNext := self.findEffectiveLocked(entry) + self.mu.Unlock() + + base := handler.BaseType() + + switch { + case prev.version == 0 && !hasNext: + // nothing applied, nothing available + + case prev.version == 0 && hasNext: + if err := handler.Apply(nextVersion, nextData); err != nil { + self.alert(base, fmt.Sprintf("v%d (%s) initial apply failed: %v", nextVersion, nextSource, err)) + if rmErr := handler.Remove(); rmErr != nil { + self.alert(base, fmt.Sprintf("v%d (%s) initial apply failed and Remove also failed: %v", nextVersion, nextSource, rmErr)) + } + self.setApplied(entry, appliedState{}) + return + } + self.setApplied(entry, appliedState{source: nextSource, version: nextVersion, data: nextData}) + + case prev.version != 0 && !hasNext: + if err := handler.Remove(); err != nil { + self.alert(base, fmt.Sprintf("v%d (%s) Remove failed: %v; subsystem state unchanged", prev.version, prev.source, err)) + return + } + self.setApplied(entry, appliedState{}) + + case prev.version != 0 && hasNext: + if prev.source == nextSource && prev.version == nextVersion && prev.data == nextData { + return + } + if err := handler.Apply(nextVersion, nextData); err != nil { + self.alert(base, fmt.Sprintf("v%d (%s) apply failed (%v); rolling back to v%d (%s)", nextVersion, nextSource, err, prev.version, prev.source)) + if rbErr := handler.Apply(prev.version, prev.data); rbErr != nil { + self.alert(base, fmt.Sprintf("rollback to v%d (%s) also failed (%v); calling Remove", prev.version, prev.source, rbErr)) + if rmErr := handler.Remove(); rmErr != nil { + self.alert(base, fmt.Sprintf("Remove also failed: %v; subsystem state unknown", rmErr)) + } + self.setApplied(entry, appliedState{}) + return + } + // rollback succeeded; applied stays at prev + return + } + self.setApplied(entry, appliedState{source: nextSource, version: nextVersion, data: nextData}) + } +} + +// findEffectiveLocked computes the effective config for the handler. +// +// Strict local-wins at the base level: if local is set, the controller's +// data is entirely ignored. If local's version is one the handler supports, +// that's the effective config. If local's version is NOT supported (e.g. +// after an upgrade that drops the version), nothing applies — the operator +// must fix their YAML. We deliberately don't fall back to the controller's +// data in that case because the operator's intent ("use my local config") +// should not be silently overridden. +// +// If local isn't set, the effective config is the highest controller +// version the handler supports. Caller holds Registry.mu. +func (self *Registry) findEffectiveLocked(entry *handlerEntry) (source ConfigSource, version int, data string, found bool) { + if entry.local != nil { + for _, v := range entry.handler.SupportedVersions() { + if v == entry.local.version { + return SourceLocal, entry.local.version, entry.local.data, true + } + } + pfxlog.Logger().WithField("baseType", entry.handler.BaseType()).WithField("version", entry.local.version).WithField("supported", entry.handler.SupportedVersions()).Error("local config at unsupported version; nothing applied (likely a programming error in the YAML translator)") + return 0, 0, "", false + } + + bestVersion := 0 + var bestData string + for _, v := range entry.handler.SupportedVersions() { + if d, ok := entry.controllerVersions[v]; ok { + if v > bestVersion { + bestVersion = v + bestData = d + } + } + } + if bestVersion > 0 { + return SourceController, bestVersion, bestData, true + } + return 0, 0, "", false +} + +func (self *Registry) setApplied(entry *handlerEntry, state appliedState) { + self.mu.Lock() + entry.applied = state + self.mu.Unlock() +} + +// AppliedVersion returns the version of configType currently applied for the +// handler whose BaseType matches the parsed configType, or 0 if nothing is +// applied. Source-agnostic; use Applied for the full state. +func (self *Registry) AppliedVersion(configType string) int { + _, version, _ := self.Applied(configType) + return version +} + +// Inspect returns a snapshot of the registry's state, intended for diagnostics +// via `ziti fabric inspect router-config-registry`. Handlers are returned in +// BaseType order for deterministic output. +func (self *Registry) Inspect() inspect.RouterConfigRegistryState { + self.mu.Lock() + defer self.mu.Unlock() + + result := inspect.RouterConfigRegistryState{ + Sealed: self.sealed.Load(), + Closed: self.closed.Load(), + Handlers: make([]inspect.RouterConfigHandlerDetail, 0, len(self.handlers)), + } + + bases := make([]string, 0, len(self.handlers)) + for base := range self.handlers { + bases = append(bases, base) + } + sort.Strings(bases) + + for _, base := range bases { + result.Handlers = append(result.Handlers, self.handlers[base].inspect()) + } + return result +} + +// inspect returns a snapshot of this handler's registry state. Caller must +// hold Registry.mu. +func (self *handlerEntry) inspect() inspect.RouterConfigHandlerDetail { + detail := inspect.RouterConfigHandlerDetail{ + BaseType: self.handler.BaseType(), + SupportedVersions: self.handler.SupportedVersions(), + } + versions := make([]int, 0, len(self.controllerVersions)) + for v := range self.controllerVersions { + versions = append(versions, v) + } + sort.Ints(versions) + for _, v := range versions { + detail.ControllerConfigs = append(detail.ControllerConfigs, inspect.RouterConfigVersionDetail{ + Version: v, + Data: parseInspectData(self.controllerVersions[v]), + }) + } + if self.local != nil { + local := self.local.inspect() + detail.LocalConfig = &local + } + detail.Applied = self.applied.inspect() + return detail +} + +// inspect returns a version detail for this local entry. +func (self *localEntry) inspect() inspect.RouterConfigVersionDetail { + return inspect.RouterConfigVersionDetail{ + Version: self.version, + Data: parseInspectData(self.data), + } +} + +// parseInspectData decodes a stored config payload into the parsed structure +// the inspect output should display. If the payload isn't valid JSON, the raw +// string is returned so the diagnostic still shows what the registry holds. +func parseInspectData(data string) any { + var parsed any + if err := json.Unmarshal([]byte(data), &parsed); err != nil { + return data + } + return parsed +} + +// inspect returns the applied detail, or nil if nothing is applied. +func (self appliedState) inspect() *inspect.RouterConfigAppliedDetail { + if self.version == 0 { + return nil + } + return &inspect.RouterConfigAppliedDetail{ + Source: self.source.String(), + Version: self.version, + } +} + +// Applied returns the source and version currently applied for the handler +// owning configType. found is false if nothing is applied or no handler is +// registered for the base. +func (self *Registry) Applied(configType string) (source ConfigSource, version int, found bool) { + base, _, err := ParseConfigType(configType) + if err != nil { + return 0, 0, false + } + self.mu.Lock() + defer self.mu.Unlock() + entry := self.handlers[base] + if entry == nil || entry.applied.version == 0 { + return 0, 0, false + } + return entry.applied.source, entry.applied.version, true +} diff --git a/router/managedconfig/registry_test.go b/router/managedconfig/registry_test.go new file mode 100644 index 000000000..980b6b620 --- /dev/null +++ b/router/managedconfig/registry_test.go @@ -0,0 +1,872 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://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 managedconfig + +import ( + "encoding/json" + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type applyCall struct { + version int + data string +} + +// fakeHandler is a scriptable ConfigHandler for tests. ApplyErrs and +// RemoveErrs are consumed in order; nil means succeed. After exhausting the +// scripted errors, all subsequent calls succeed. Methods are safe to call +// from any goroutine. +type fakeHandler struct { + mu sync.Mutex + base string + versions []int + applies []applyCall + removes int + applyErrs []error + removeErrs []error +} + +func (f *fakeHandler) BaseType() string { return f.base } +func (f *fakeHandler) SupportedVersions() []int { return f.versions } + +func (f *fakeHandler) Apply(version int, data string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.applies = append(f.applies, applyCall{version, data}) + if len(f.applyErrs) == 0 { + return nil + } + err := f.applyErrs[0] + f.applyErrs = f.applyErrs[1:] + return err +} + +func (f *fakeHandler) Remove() error { + f.mu.Lock() + defer f.mu.Unlock() + f.removes++ + if len(f.removeErrs) == 0 { + return nil + } + err := f.removeErrs[0] + f.removeErrs = f.removeErrs[1:] + return err +} + +func (f *fakeHandler) snapshot() (applies []applyCall, removes int) { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]applyCall, len(f.applies)) + copy(out, f.applies) + return out, f.removes +} + +type alertEntry struct { + baseType string + detail string +} + +type alertRecorder struct { + mu sync.Mutex + entries []alertEntry +} + +func (a *alertRecorder) record(baseType, detail string) { + a.mu.Lock() + defer a.mu.Unlock() + a.entries = append(a.entries, alertEntry{baseType, detail}) +} + +func (a *alertRecorder) all() []alertEntry { + a.mu.Lock() + defer a.mu.Unlock() + out := make([]alertEntry, len(a.entries)) + copy(out, a.entries) + return out +} + +func newRecordingRegistry() (*Registry, *alertRecorder) { + rec := &alertRecorder{} + r := NewRegistry(rec.record) + return r, rec +} + +// newSealedRegistry returns a recording registry with the given handlers +// already registered and Seal() called. The standard test fixture for any +// test that needs to invoke Apply / Remove. +func newSealedRegistry(t *testing.T, handlers ...ConfigHandler) (*Registry, *alertRecorder) { + t.Helper() + r, rec := newRecordingRegistry() + for _, h := range handlers { + require.NoError(t, r.Register(h)) + } + r.Seal() + return r, rec +} + +// --- ParseConfigType --------------------------------------------------------- + +func Test_ParseConfigType_Valid(t *testing.T) { + req := require.New(t) + cases := []struct { + in string + wantBaseType string + wantVersion int + }{ + {"router.link.v1", "router.link", 1}, + {"router.link.v2", "router.link", 2}, + {"router.link.v42", "router.link", 42}, + {"router.xgress.proxy.v1", "router.xgress.proxy", 1}, + } + for _, c := range cases { + base, ver, err := ParseConfigType(c.in) + req.NoError(err, c.in) + req.Equal(c.wantBaseType, base, c.in) + req.Equal(c.wantVersion, ver, c.in) + } +} + +func Test_ParseConfigType_Invalid(t *testing.T) { + req := require.New(t) + cases := []string{ + "router.link", + "router.link.v", + "router.link.va", + "router.link.v1a", + "router.link.v0", + "router.link.v-1", + ".v1", + "v1", + } + for _, c := range cases { + _, _, err := ParseConfigType(c) + req.Error(err, c) + } +} + +// --- Register & lookup ------------------------------------------------------- + +func Test_Register_Single(t *testing.T) { + req := require.New(t) + r, _ := newRecordingRegistry() + + h := &fakeHandler{base: "router.link", versions: []int{1}} + req.NoError(r.Register(h)) + req.Same(h, r.Handler("router.link.v1")) + req.Nil(r.Handler("other.family.v1")) +} + +func Test_Register_Handler_OwnsBase(t *testing.T) { + req := require.New(t) + r, _ := newRecordingRegistry() + + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + req.NoError(r.Register(h)) + req.Same(h, r.Handler("router.link.v1")) + req.Same(h, r.Handler("router.link.v2")) + req.Same(h, r.Handler("router.link.v99")) +} + +func Test_Register_Duplicate(t *testing.T) { + req := require.New(t) + r, _ := newRecordingRegistry() + + h1 := &fakeHandler{base: "router.link", versions: []int{1}} + req.NoError(r.Register(h1)) + + h2 := &fakeHandler{base: "router.link", versions: []int{2}} + err := r.Register(h2) + req.Error(err) + req.ErrorIs(err, ErrHandlerAlreadyRegistered) +} + +// --- Seal lifecycle ---------------------------------------------------------- + +func Test_Seal_PanicsOnLateRegister(t *testing.T) { + req := require.New(t) + r, _ := newRecordingRegistry() + r.Seal() + + defer func() { + req.NotNil(recover(), "Register after Seal should panic") + }() + h := &fakeHandler{base: "router.link", versions: []int{1}} + _ = r.Register(h) +} + +func Test_Seal_ApplyStillWorks(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{}`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 1) +} + +func Test_Seal_PanicsOnApplyBeforeSeal(t *testing.T) { + req := require.New(t) + r, _ := newRecordingRegistry() + defer func() { + req.NotNil(recover(), "Apply before Seal should panic") + }() + _ = r.ApplyController("router.link.v1", `{}`) +} + +func Test_Seal_PanicsOnRemoveBeforeSeal(t *testing.T) { + req := require.New(t) + r, _ := newRecordingRegistry() + defer func() { + req.NotNil(recover(), "Remove before Seal should panic") + }() + _ = r.RemoveController("router.link.v1") +} + +// --- Apply / Remove parse errors -------------------------------------------- + +func Test_Apply_InvalidConfigType_ReturnsError(t *testing.T) { + req := require.New(t) + r, _ := newSealedRegistry(t) + err := r.ApplyController("router.link", `{}`) + req.Error(err) +} + +func Test_Remove_InvalidConfigType_ReturnsError(t *testing.T) { + req := require.New(t) + r, _ := newSealedRegistry(t) + err := r.RemoveController("router.link.va") + req.Error(err) +} + +// --- Apply / Remove without a handler --------------------------------------- + +func Test_Apply_NoHandler_ReturnsError(t *testing.T) { + req := require.New(t) + r, _ := newSealedRegistry(t) + err := r.ApplyController("router.unknown.v1", `{}`) + req.Error(err) + req.ErrorIs(err, ErrNoHandlerRegistered) +} + +func Test_Remove_NoHandler_ReturnsError(t *testing.T) { + req := require.New(t) + r, _ := newSealedRegistry(t) + err := r.RemoveController("router.unknown.v1") + req.Error(err) + req.ErrorIs(err, ErrNoHandlerRegistered) +} + +// --- Single-version flow ----------------------------------------------------- + +func Test_Apply_FirstTime(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{"k":"v"}`)) + r.WaitForIdle() + + applies, removes := h.snapshot() + req.Len(applies, 1) + req.Equal(1, applies[0].version) + req.Equal(0, removes) + req.Equal(1, r.AppliedVersion("router.link.v1")) +} + +func Test_Apply_Update(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{"k":"a"}`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v1", `{"k":"b"}`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 2) + req.Equal(`{"k":"a"}`, string(applies[0].data)) + req.Equal(`{"k":"b"}`, string(applies[1].data)) +} + +func Test_Apply_NoOpWhenIdentical(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{"k":"v"}`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v1", `{"k":"v"}`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 1) +} + +func Test_Apply_FirstFailure_TriggersRemove(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}, applyErrs: []error{errors.New("bad")}} + r, alerts := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{}`)) + r.WaitForIdle() + + applies, removes := h.snapshot() + req.Len(applies, 1) + req.Equal(1, removes) + req.Equal(0, r.AppliedVersion("router.link.v1")) + req.NotEmpty(alerts.all()) +} + +func Test_Apply_UpdateFailure_RollbackSucceeds(t *testing.T) { + req := require.New(t) + h := &fakeHandler{ + base: "router.link", + versions: []int{1}, + applyErrs: []error{nil, errors.New("bad-update"), nil}, + } + r, alerts := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{"k":"a"}`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v1", `{"k":"b"}`)) + r.WaitForIdle() + + applies, removes := h.snapshot() + req.Len(applies, 3) + req.Equal(`{"k":"a"}`, string(applies[0].data)) + req.Equal(`{"k":"b"}`, string(applies[1].data)) + req.Equal(`{"k":"a"}`, string(applies[2].data)) + req.Equal(0, removes) + req.Equal(1, r.AppliedVersion("router.link.v1")) + req.NotEmpty(alerts.all()) +} + +func Test_Apply_UpdateFailure_RollbackFails(t *testing.T) { + req := require.New(t) + h := &fakeHandler{ + base: "router.link", + versions: []int{1}, + applyErrs: []error{nil, errors.New("bad-update"), errors.New("bad-rollback")}, + } + r, alerts := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{"k":"a"}`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v1", `{"k":"b"}`)) + r.WaitForIdle() + + applies, removes := h.snapshot() + req.Len(applies, 3) + req.Equal(1, removes) + req.Equal(0, r.AppliedVersion("router.link.v1")) + req.GreaterOrEqual(len(alerts.all()), 2) +} + +// --- Multi-version flow ------------------------------------------------------ + +func Test_MultiVersion_HighestWins(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `v1data`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v2", `v2data`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 2) + req.Equal(1, applies[0].version) + req.Equal(2, applies[1].version) + req.Equal(2, r.AppliedVersion("router.link.v1")) +} + +func Test_MultiVersion_HandlerSupportsSubset(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v2", `v2`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v1", `v1`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 1) + req.Equal(1, applies[0].version) +} + +func Test_MultiVersion_FallbackOnRemove(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `v1data`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v2", `v2data`)) + r.WaitForIdle() + req.NoError(r.RemoveController("router.link.v2")) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 3) + req.Equal(1, applies[2].version) + req.Equal(`v1data`, string(applies[2].data)) + req.Equal(1, r.AppliedVersion("router.link.v1")) +} + +func Test_MultiVersion_FallbackOnApplyFailure(t *testing.T) { + req := require.New(t) + h := &fakeHandler{ + base: "router.link", + versions: []int{1, 2}, + applyErrs: []error{nil, errors.New("v2 broken"), nil}, + } + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `v1data`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v2", `v2data`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 3) + req.Equal(1, applies[0].version) + req.Equal(2, applies[1].version) + req.Equal(1, applies[2].version) + req.Equal(1, r.AppliedVersion("router.link.v1")) +} + +func Test_MultiVersion_RemoveLastAvailable(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `v1`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v2", `v2`)) + r.WaitForIdle() + req.NoError(r.RemoveController("router.link.v2")) + r.WaitForIdle() + req.NoError(r.RemoveController("router.link.v1")) + r.WaitForIdle() + + _, removes := h.snapshot() + req.Equal(1, removes) + req.Equal(0, r.AppliedVersion("router.link.v1")) +} + +func Test_MultiVersion_OutOfOrderArrival(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `v1`)) + r.WaitForIdle() + req.Equal(1, r.AppliedVersion("router.link.v1")) + req.NoError(r.ApplyController("router.link.v2", `v2`)) + r.WaitForIdle() + req.Equal(2, r.AppliedVersion("router.link.v1")) + + applies, _ := h.snapshot() + req.Len(applies, 2) +} + +// --- Remove flow ------------------------------------------------------------- + +func Test_Remove_HandlerError(t *testing.T) { + req := require.New(t) + h := &fakeHandler{ + base: "router.link", + versions: []int{1}, + removeErrs: []error{errors.New("remove broken")}, + } + r, alerts := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{}`)) + r.WaitForIdle() + req.NoError(r.RemoveController("router.link.v1")) + r.WaitForIdle() + + _, removes := h.snapshot() + req.Equal(1, removes) + req.Equal(1, r.AppliedVersion("router.link.v1"), + "applied should stay at previous when Remove fails") + req.NotEmpty(alerts.all()) +} + +// --- Source precedence ------------------------------------------------------ + +func Test_Source_LocalApplies(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyLocal("router.link.v1", `local-data`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 1) + req.Equal(`local-data`, string(applies[0].data)) + + src, ver, ok := r.Applied("router.link.v1") + req.True(ok) + req.Equal(SourceLocal, src) + req.Equal(1, ver) +} + +func Test_Source_LocalBeatsController(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + // Controller arrives first with v2, then local arrives with v1. + // Local-wins is at the base level, so local v1 should beat controller v2. + req.NoError(r.ApplyController("router.link.v2", `ctrl-v2`)) + r.WaitForIdle() + req.Equal(2, r.AppliedVersion("router.link.v1")) + + req.NoError(r.ApplyLocal("router.link.v1", `local-v1`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 2) + req.Equal(2, applies[0].version) // ctrl v2 came first + req.Equal(1, applies[1].version) // then local v1 took over + + src, ver, _ := r.Applied("router.link.v1") + req.Equal(SourceLocal, src) + req.Equal(1, ver) +} + +func Test_Source_ControllerIgnoredWhileLocalSet(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + // Local v1 first, then controller tries v2. Controller should be ignored. + req.NoError(r.ApplyLocal("router.link.v1", `local-v1`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v2", `ctrl-v2`)) + r.WaitForIdle() + + applies, _ := h.snapshot() + req.Len(applies, 1, "controller event should not produce a handler call when local is set") + req.Equal(1, applies[0].version) + + src, _, _ := r.Applied("router.link.v1") + req.Equal(SourceLocal, src) +} + +func Test_Source_RemoveLocalFallsBackToController(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + // Both sources have data; local wins. + req.NoError(r.ApplyController("router.link.v2", `ctrl-v2`)) + r.WaitForIdle() + req.NoError(r.ApplyLocal("router.link.v1", `local-v1`)) + r.WaitForIdle() + src, _, _ := r.Applied("router.link.v1") + req.Equal(SourceLocal, src) + + // Drop local; controller becomes effective. + req.NoError(r.RemoveLocal("router.link")) + r.WaitForIdle() + + src, ver, _ := r.Applied("router.link.v1") + req.Equal(SourceController, src) + req.Equal(2, ver) + + applies, _ := h.snapshot() + // initial ctrl v2 + local v1 + fallback to ctrl v2 = 3 handler calls + req.Len(applies, 3) + req.Equal(2, applies[2].version) +} + +func Test_Source_RemoveControllerWhileLocalSet_NoChange(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v2", `ctrl-v2`)) + r.WaitForIdle() + req.NoError(r.ApplyLocal("router.link.v1", `local-v1`)) + r.WaitForIdle() + appliesBefore, _ := h.snapshot() + + req.NoError(r.RemoveController("router.link.v2")) + r.WaitForIdle() + + appliesAfter, _ := h.snapshot() + req.Len(appliesAfter, len(appliesBefore), + "controller removal should not trigger handler when local was already winning") + + src, ver, _ := r.Applied("router.link.v1") + req.Equal(SourceLocal, src) + req.Equal(1, ver) +} + +func Test_Source_LocalMultiVersion(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + // Both local versions present; highest wins within the source. + req.NoError(r.ApplyLocal("router.link.v1", `local-v1`)) + r.WaitForIdle() + req.NoError(r.ApplyLocal("router.link.v2", `local-v2`)) + r.WaitForIdle() + + src, ver, _ := r.Applied("router.link.v1") + req.Equal(SourceLocal, src) + req.Equal(2, ver) +} + +func Test_Source_LocalUnsupportedVersion_NothingApplies(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + // Controller has v2 that the handler supports. + req.NoError(r.ApplyController("router.link.v2", `ctrl-v2`)) + r.WaitForIdle() + req.Equal(2, r.AppliedVersion("router.link.v1")) + + // Operator sets local at v3 — handler doesn't support v3. Local-wins is + // strict: controller's v2 must NOT silently take over. Subsystem should + // reconcile to "nothing applied," surfacing the problem. + req.NoError(r.ApplyLocal("router.link.v3", `local-v3`)) + r.WaitForIdle() + + _, _, found := r.Applied("router.link.v1") + req.False(found, "local-but-unsupported should not silently fall back to controller") +} + +func Test_ConfigSource_String(t *testing.T) { + req := require.New(t) + req.Equal("controller", SourceController.String()) + req.Equal("local", SourceLocal.String()) +} + +// --- Inspect ---------------------------------------------------------------- + +func Test_Inspect_Empty(t *testing.T) { + req := require.New(t) + r, _ := newSealedRegistry(t) + snap := r.Inspect() + req.True(snap.Sealed) + req.False(snap.Closed) + req.Empty(snap.Handlers) +} + +func Test_Inspect_HandlersSortedByBase(t *testing.T) { + req := require.New(t) + h1 := &fakeHandler{base: "router.xgress.proxy", versions: []int{1}} + h2 := &fakeHandler{base: "router.link", versions: []int{1, 2}} + h3 := &fakeHandler{base: "router.forwarder", versions: []int{1}} + r, _ := newSealedRegistry(t, h1, h2, h3) + + snap := r.Inspect() + req.Len(snap.Handlers, 3) + req.Equal("router.forwarder", snap.Handlers[0].BaseType) + req.Equal("router.link", snap.Handlers[1].BaseType) + req.Equal("router.xgress.proxy", snap.Handlers[2].BaseType) +} + +func Test_Inspect_ReportsControllerAndLocalAndApplied(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1, 2}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyController("router.link.v1", `{"src":"ctrl","v":1}`)) + r.WaitForIdle() + req.NoError(r.ApplyController("router.link.v2", `{"src":"ctrl","v":2}`)) + r.WaitForIdle() + req.NoError(r.ApplyLocal("router.link.v1", `{"src":"local","v":1}`)) + r.WaitForIdle() + + snap := r.Inspect() + req.Len(snap.Handlers, 1) + hi := snap.Handlers[0] + req.Equal("router.link", hi.BaseType) + req.Equal([]int{1, 2}, hi.SupportedVersions) + req.Len(hi.ControllerConfigs, 2) + req.Equal(1, hi.ControllerConfigs[0].Version) + req.Equal(map[string]any{"src": "ctrl", "v": float64(1)}, hi.ControllerConfigs[0].Data) + req.Equal(2, hi.ControllerConfigs[1].Version) + req.Equal(map[string]any{"src": "ctrl", "v": float64(2)}, hi.ControllerConfigs[1].Data) + req.NotNil(hi.LocalConfig) + req.Equal(1, hi.LocalConfig.Version) + req.Equal(map[string]any{"src": "local", "v": float64(1)}, hi.LocalConfig.Data) + req.NotNil(hi.Applied) + req.Equal("local", hi.Applied.Source) + req.Equal(1, hi.Applied.Version) +} + +func Test_Inspect_LocalConfigJsonKeyIsCamelCase(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + + req.NoError(r.ApplyLocal("router.link.v1", `{"src":"local","v":1}`)) + r.WaitForIdle() + + buf, err := json.Marshal(r.Inspect().Handlers[0]) + req.NoError(err) + req.Contains(string(buf), `"localConfig"`) + req.NotContains(string(buf), `"localconfig"`) +} + +func Test_Inspect_JSON(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + req.NoError(r.ApplyController("router.link.v1", `{"hello":"world"}`)) + r.WaitForIdle() + + b, err := json.Marshal(r.Inspect()) + req.NoError(err) + // Spot-check the marshaled output for the expected keys; not a full + // schema validation, just enough to confirm the struct tags work and that + // the config payload is inlined as parsed JSON rather than a quoted string. + out := string(b) + req.Contains(out, `"sealed":true`) + req.Contains(out, `"baseType":"router.link"`) + req.Contains(out, `"applied":{"source":"controller","version":1}`) + req.Contains(out, `"data":{"hello":"world"}`) +} + +// --- Concurrency ------------------------------------------------------------- + +func Test_DifferentHandlersReconcileInParallel(t *testing.T) { + req := require.New(t) + slowGate := make(chan struct{}) + slow := &slowHandler{base: "router.slow", versions: []int{1}, gate: slowGate} + fast := &fakeHandler{base: "router.fast", versions: []int{1}} + r, _ := newSealedRegistry(t, slow, fast) + + req.NoError(r.ApplyController("router.slow.v1", `a`)) + req.NoError(r.ApplyController("router.fast.v1", `b`)) + + pollUntil(t, func() bool { + applies, _ := fast.snapshot() + return len(applies) == 1 + }) + + close(slowGate) + r.WaitForIdle() + + slowApplies, _ := slow.snapshot() + req.Len(slowApplies, 1) +} + +type slowHandler struct { + mu sync.Mutex + base string + versions []int + applies []applyCall + gate chan struct{} +} + +func (s *slowHandler) BaseType() string { return s.base } +func (s *slowHandler) SupportedVersions() []int { return s.versions } +func (s *slowHandler) Apply(version int, data string) error { + <-s.gate + s.mu.Lock() + defer s.mu.Unlock() + s.applies = append(s.applies, applyCall{version, data}) + return nil +} +func (s *slowHandler) Remove() error { return nil } +func (s *slowHandler) snapshot() (applies []applyCall, removes int) { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]applyCall, len(s.applies)) + copy(out, s.applies) + return out, 0 +} + +func pollUntil(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(time.Millisecond) + } + t.Fatal("condition not met within 5s") +} + +// --- Default alert ----------------------------------------------------------- + +func Test_Default_AlertLogs(t *testing.T) { + r := NewRegistry(nil) + h := &fakeHandler{base: "x", versions: []int{1}, applyErrs: []error{errors.New("nope")}} + require.NoError(t, r.Register(h)) + r.Seal() + require.NoError(t, r.ApplyController("x.v1", `{}`)) + r.WaitForIdle() +} + +// --- Close ------------------------------------------------------------------- + +func Test_Close_DrainsInFlight(t *testing.T) { + req := require.New(t) + gate := make(chan struct{}) + slow := &slowHandler{base: "router.slow", versions: []int{1}, gate: gate} + r, _ := newSealedRegistry(t, slow) + + req.NoError(r.ApplyController("router.slow.v1", `a`)) + + closeDone := make(chan struct{}) + go func() { + r.Close() + close(closeDone) + }() + + select { + case <-closeDone: + t.Fatal("Close returned before in-flight reconcile completed") + default: + } + + close(gate) + <-closeDone +} + +func Test_Close_PreventsNewSpawns(t *testing.T) { + req := require.New(t) + h := &fakeHandler{base: "router.link", versions: []int{1}} + r, _ := newSealedRegistry(t, h) + r.Close() + + // After Close, Apply still records data but does not spawn a reconcile. + req.NoError(r.ApplyController("router.link.v1", `{}`)) + applies, _ := h.snapshot() + req.Empty(applies, "no apply should be observed after Close") +} diff --git a/router/router.go b/router/router.go index 26a734023..28f1423d2 100644 --- a/router/router.go +++ b/router/router.go @@ -69,6 +69,7 @@ import ( "github.com/openziti/ziti/v2/router/inspect" "github.com/openziti/ziti/v2/router/interfaces" "github.com/openziti/ziti/v2/router/link" + "github.com/openziti/ziti/v2/router/managedconfig" routerMetrics "github.com/openziti/ziti/v2/router/metrics" "github.com/openziti/ziti/v2/router/state" "github.com/openziti/ziti/v2/router/xgress_edge" @@ -123,6 +124,7 @@ type Router struct { xgMetrics *routerMetrics.XgressMetrics healthChecker gosundheit.Health alertReporter *alert.Reporter + configRegistry *managedconfig.Registry inspectHandler channel.ContentTypeReceiver } @@ -214,6 +216,11 @@ func (self *Router) GetRouterDataModel() *common.RouterDataModel { return self.stateManager.RouterDataModel() } +// GetRouterConfigRegistry returns the controller-managed config registry. +func (self *Router) GetRouterConfigRegistry() *managedconfig.Registry { + return self.configRegistry +} + // WithRouterDataModel passes the current router data model into the provide function func (self *Router) WithRouterDataModel(f func(*common.RouterDataModel) error) error { return self.stateManager.WithRouterDataModel(f) @@ -337,6 +344,7 @@ func Create(cfg *env.Config, versionProvider versions.VersionProvider) *Router { router.stateManager = state.NewManager(router) router.certManager = state.NewCertExpirationChecker(router, true) router.alertReporter = alert.NewAlertReporter(router.ctrls, cfg.Id.Token, 1000, 10) + router.configRegistry = managedconfig.NewRegistry(nil) router.xlinkRegistry = link.NewLinkRegistry(router) router.faulter = forwarder.NewFaulter(router, cfg.Forwarder.FaultTxInterval) @@ -499,6 +507,15 @@ func (self *Router) Start() error { go web.Run() } + // Seal the managed-config registry. All subsystem handler registration + // must be complete before this point; subsequent ApplyController / + // RemoveController calls from RDM events will route to registered + // handlers. Wire the router-side subscriber into the state manager so + // Config events arriving from the controller dispatch through the + // allow-list and into the registry. + self.configRegistry.Seal() + self.stateManager.SetRouterConfigSubscriber(state.NewRouterConfigSubscriber(self)) + // Start control plane (must be last) if err := self.startControlPlane(); err != nil { return err @@ -522,6 +539,12 @@ func (self *Router) Shutdown() error { } } + // Drain in-flight managed-config reconciles before tearing down the + // subsystems their handlers manage. ctrls.Close above has already + // stopped new Config events from arriving, so this only blocks on work + // already spawned. + self.configRegistry.Close() + for _, xlinkListener := range self.xlinkListeners { if err := xlinkListener.Close(); err != nil { errs = append(errs, err) diff --git a/router/state/manager.go b/router/state/manager.go index 2e54313a4..bbd298306 100644 --- a/router/state/manager.go +++ b/router/state/manager.go @@ -231,6 +231,16 @@ type Manager interface { // optionally resetting the controller subscription. SetRouterDataModel(model *common.RouterDataModel, resetSubscription bool) + // SetRouterConfigSubscriber registers the subscriber that receives Config + // events for router-target ConfigTypes. Called once during router startup, + // after handler registration and managedconfig.Registry.Seal. The manager + // attaches the subscriber to the current RDM (if any), bootstraps by + // dispatching Applied for every router-target Config already loaded, and + // re-attaches the subscriber to any subsequent RDM provided via + // SetRouterDataModel — with a remove/apply diff against the prior RDM so + // configs that vanish during a full-state resync are surfaced as removals. + SetRouterConfigSubscriber(s common.RouterConfigEventSubscriber) + // GetRouterDataModelPool returns the goroutine pool used for processing // router data model events and updates. GetRouterDataModelPool() goroutines.Pool @@ -534,6 +544,12 @@ type ManagerImpl struct { dataModelSubscription concurrenz.AtomicValue[DataModelSubscription] dataModelSubTimeout time.Time + // routerConfigSubscriber is the bridge between RDM Config events and the + // managedconfig.Registry. Stored on the manager (not just the RDM) so it + // survives full-state resyncs that swap the RDM out from under us. Read/ + // written under rdmLock so attach is serialized with SetRouterDataModel. + routerConfigSubscriber common.RouterConfigEventSubscriber + postureCache *posture.Cache connectionTracker ConnectionTracker @@ -1051,6 +1067,13 @@ func (self *ManagerImpl) SetRouterDataModel(model *common.RouterDataModel, reset logger = logger.WithField("existingIndex", existingIndex) } + // Attach the subscriber to the new model before the swap so any + // HandleConfigEvent on the new RDM dispatches correctly the moment it + // becomes reachable. + if self.routerConfigSubscriber != nil { + model.SetRouterConfigSubscriber(self.routerConfigSubscriber) + } + self.routerDataModel.Store(model) // Clear the replace-in-progress flag before syncing subscribers. The model is @@ -1067,6 +1090,14 @@ func (self *ManagerImpl) SetRouterDataModel(model *common.RouterDataModel, reset model.SyncAllSubscribers() + // Diff router-target configs between old and new now that the new model + // is the authoritative one. Subscribers that query RouterDataModel() + // during dispatch see the post-swap state — matching the contract of + // HandleConfigEvent, which also fires after rdm.Configs is updated. + if self.routerConfigSubscriber != nil { + dispatchRouterConfigDiff(existing, model, self.routerConfigSubscriber) + } + if resetSubscription { // notify subscription manager code to resubscribe with the updated model and index select { @@ -1078,6 +1109,84 @@ func (self *ManagerImpl) SetRouterDataModel(model *common.RouterDataModel, reset logger.Infof("router data model replacement complete, old: %p, new: %p", existing, model) } +// SetRouterConfigSubscriber registers the subscriber and bootstraps it +// against the current RDM. Called once at router startup, after the registry +// has been Sealed and all subsystem handlers have been registered. +func (self *ManagerImpl) SetRouterConfigSubscriber(s common.RouterConfigEventSubscriber) { + self.rdmLock.Lock() + defer self.rdmLock.Unlock() + + self.routerConfigSubscriber = s + current := self.routerDataModel.Load() + if current != nil { + current.SetRouterConfigSubscriber(s) + } + if s == nil { + return + } + // Bootstrap: dispatch Applied for every router-target config currently in + // the RDM. Treats "before the subscriber was set" as an empty prior state, + // so the diff is just "everything in current." + dispatchRouterConfigDiff(nil, current, s) +} + +// dispatchRouterConfigDiff walks router-target Configs in oldRdm and newRdm, +// dispatching OnRouterConfigRemoved for types present in old but not new and +// OnRouterConfigApplied for types whose data has actually changed (new or +// different from old). Either RDM may be nil. +// +// Unchanged entries are skipped: the registry would no-op them anyway, but +// each dispatched Apply still spawns a reconcile goroutine and takes locks, +// so skipping is meaningfully cheaper for the common case of a full-state +// resync where most configs are stable. +// +// The fast path through HandleConfigEvent dispatches via the subscriber +// already attached to the RDM. This function is used at the transition +// points (subscriber attach, RDM swap) where events that *would have* +// fired through HandleConfigEvent are replayed against the latest state. +func dispatchRouterConfigDiff(oldRdm, newRdm *common.RouterDataModel, sub common.RouterConfigEventSubscriber) { + if sub == nil { + return + } + oldTypes := collectRouterConfigTypes(oldRdm) + newTypes := collectRouterConfigTypes(newRdm) + + for typeName := range oldTypes { + if _, stillPresent := newTypes[typeName]; !stillPresent { + sub.OnRouterConfigRemoved(typeName) + } + } + for typeName, newData := range newTypes { + if oldData, ok := oldTypes[typeName]; ok && oldData == newData { + continue + } + sub.OnRouterConfigApplied(typeName, newData) + } +} + +// collectRouterConfigTypes scans rdm.Configs for entries whose ConfigType has +// Target == "router" and returns a map of ConfigType.Name -> DataJson. Empty +// map when rdm is nil. If multiple Configs share a ConfigType (which the +// design doesn't currently allow, but is cheap to be safe about), the last +// one wins; an alert here would surface a controller bug. +func collectRouterConfigTypes(rdm *common.RouterDataModel) map[string]string { + out := map[string]string{} + if rdm == nil { + return out + } + rdm.Configs.IterCb(func(_ string, cfg *common.Config) { + if cfg == nil { + return + } + ct, ok := rdm.ConfigTypes.Get(cfg.TypeId) + if !ok || ct == nil || ct.Target != common.ConfigTypeTargetRouter { + return + } + out[ct.Name] = cfg.DataJson + }) + return out +} + func (self *ManagerImpl) ResyncRouterDataModel() { self.resyncRouterDataModel.Store(true) self.dataModelSubscription.Store(DataModelSubscription{}) diff --git a/router/state/router_config_subscriber.go b/router/state/router_config_subscriber.go new file mode 100644 index 000000000..18c63cb7c --- /dev/null +++ b/router/state/router_config_subscriber.go @@ -0,0 +1,90 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://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 state + +import ( + "github.com/michaelquigley/pfxlog" + "github.com/openziti/ziti/v2/common" + "github.com/openziti/ziti/v2/router/env" + "github.com/openziti/ziti/v2/router/managedconfig" +) + +// configAllowList is the slice of ManagedConfigOptions the subscriber relies +// on. Narrowing it to an interface keeps the subscriber testable without +// pulling in the full RouterEnv. +type configAllowList interface { + IsAllowed(configType string) bool +} + +// RouterConfigSubscriber is the bridge from RDM Config events to the +// managedconfig.Registry. It filters incoming events through the operator's +// allow-list before dispatching to the registry, so a router never applies a +// config type the operator hasn't opted into. +// +// Errors from ApplyController/RemoveController are logged; the subscriber +// has no way to push back on the RDM (events are already committed). The +// registry's own alert callback covers handler-level failures. +type RouterConfigSubscriber struct { + allow configAllowList + registry *managedconfig.Registry +} + +// NewRouterConfigSubscriber constructs a subscriber bound to the given +// RouterEnv. Pulls the allow-list and registry off env at construction time +// so OnRouterConfig* don't re-resolve them per call. +func NewRouterConfigSubscriber(routerEnv env.RouterEnv) *RouterConfigSubscriber { + return &RouterConfigSubscriber{ + allow: &routerEnv.GetConfig().ManagedConfig, + registry: routerEnv.GetRouterConfigRegistry(), + } +} + +// newRouterConfigSubscriberFromParts is the testable constructor; production +// code should use NewRouterConfigSubscriber. +func newRouterConfigSubscriberFromParts(allow configAllowList, registry *managedconfig.Registry) *RouterConfigSubscriber { + return &RouterConfigSubscriber{allow: allow, registry: registry} +} + +// OnRouterConfigApplied implements common.RouterConfigEventSubscriber. Drops +// configs whose type isn't allowed by the local managed-config policy; logs +// rejections at info so operators can see why a controller config didn't +// land. +func (self *RouterConfigSubscriber) OnRouterConfigApplied(configType string, data string) { + if self.allow == nil || !self.allow.IsAllowed(configType) { + pfxlog.Logger().WithField("configType", configType).Info("router config not in local allow-list; ignoring controller apply") + return + } + if err := self.registry.ApplyController(configType, data); err != nil { + pfxlog.Logger().WithField("configType", configType).WithError(err).Warn("registry rejected ApplyController") + } +} + +// OnRouterConfigRemoved implements common.RouterConfigEventSubscriber. Remove +// also goes through the allow-list: if the operator has since revoked a type +// it accepted earlier, the registry already lost ownership of it; we don't +// need to forward the removal. +func (self *RouterConfigSubscriber) OnRouterConfigRemoved(configType string) { + if self.allow == nil || !self.allow.IsAllowed(configType) { + return + } + if err := self.registry.RemoveController(configType); err != nil { + pfxlog.Logger().WithField("configType", configType).WithError(err).Warn("registry rejected RemoveController") + } +} + +// ensure interface compliance +var _ common.RouterConfigEventSubscriber = (*RouterConfigSubscriber)(nil) diff --git a/router/state/router_config_subscriber_test.go b/router/state/router_config_subscriber_test.go new file mode 100644 index 000000000..b3d297411 --- /dev/null +++ b/router/state/router_config_subscriber_test.go @@ -0,0 +1,276 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://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 state + +import ( + "sort" + "testing" + + "github.com/openziti/ziti/v2/common" + "github.com/openziti/ziti/v2/router/managedconfig" + "github.com/stretchr/testify/require" +) + +// --- allow-list subscriber ----------------------------------------------- + +type fakeAllow struct { + allowed map[string]bool +} + +func (f *fakeAllow) IsAllowed(configType string) bool { return f.allowed[configType] } + +type fakeRegistryHandler struct { + base string + versions []int + applies []applyRec + removes int +} + +type applyRec struct { + version int + data string +} + +func (f *fakeRegistryHandler) BaseType() string { return f.base } +func (f *fakeRegistryHandler) SupportedVersions() []int { return f.versions } +func (f *fakeRegistryHandler) Apply(v int, data string) error { f.applies = append(f.applies, applyRec{v, data}); return nil } +func (f *fakeRegistryHandler) Remove() error { f.removes++; return nil } + +func newSealedRegistry(t *testing.T, h *fakeRegistryHandler) *managedconfig.Registry { + t.Helper() + r := managedconfig.NewRegistry(nil) + require.NoError(t, r.Register(h)) + r.Seal() + return r +} + +func Test_RouterConfigSubscriber_AppliedAllowedReachesRegistry(t *testing.T) { + req := require.New(t) + h := &fakeRegistryHandler{base: "router.link", versions: []int{1}} + r := newSealedRegistry(t, h) + allow := &fakeAllow{allowed: map[string]bool{"router.link.v1": true}} + + sub := newRouterConfigSubscriberFromParts(allow, r) + sub.OnRouterConfigApplied("router.link.v1", `{"k":"v"}`) + r.WaitForIdle() + + req.Len(h.applies, 1) + req.Equal(1, h.applies[0].version) + req.Equal(`{"k":"v"}`, h.applies[0].data) +} + +func Test_RouterConfigSubscriber_AppliedDisallowedDropped(t *testing.T) { + req := require.New(t) + h := &fakeRegistryHandler{base: "router.link", versions: []int{1}} + r := newSealedRegistry(t, h) + allow := &fakeAllow{allowed: map[string]bool{}} // empty: nothing allowed + + sub := newRouterConfigSubscriberFromParts(allow, r) + sub.OnRouterConfigApplied("router.link.v1", `{}`) + r.WaitForIdle() + + req.Empty(h.applies) +} + +func Test_RouterConfigSubscriber_RemovedAllowedReachesRegistry(t *testing.T) { + req := require.New(t) + h := &fakeRegistryHandler{base: "router.link", versions: []int{1}} + r := newSealedRegistry(t, h) + allow := &fakeAllow{allowed: map[string]bool{"router.link.v1": true}} + + sub := newRouterConfigSubscriberFromParts(allow, r) + sub.OnRouterConfigApplied("router.link.v1", `{"k":"v"}`) + r.WaitForIdle() + sub.OnRouterConfigRemoved("router.link.v1") + r.WaitForIdle() + + req.Equal(1, h.removes) +} + +func Test_RouterConfigSubscriber_RemovedDisallowedDropped(t *testing.T) { + req := require.New(t) + h := &fakeRegistryHandler{base: "router.link", versions: []int{1}} + r := newSealedRegistry(t, h) + allow := &fakeAllow{allowed: map[string]bool{}} + + sub := newRouterConfigSubscriberFromParts(allow, r) + sub.OnRouterConfigRemoved("router.link.v1") + r.WaitForIdle() + + req.Equal(0, h.removes) +} + +// --- diff helper tests --------------------------------------------------- + +// recordingSub captures subscriber notifications for assertions. Order is +// preserved for applies; removes are sorted before comparison since map +// iteration is non-deterministic. +type recordingSub struct { + applied []string + removed []string +} + +func (r *recordingSub) OnRouterConfigApplied(configType string, data string) { + r.applied = append(r.applied, configType+"="+data) +} + +func (r *recordingSub) OnRouterConfigRemoved(configType string) { + r.removed = append(r.removed, configType) +} + +func newRdmWithConfigs(routerId string, configs map[string]string, types map[string]string) *common.RouterDataModel { + rdm := common.NewBareRouterDataModel(routerId) + for typeId, typeName := range types { + target := common.ConfigTypeTargetRouter + // If typeName starts with "service.", treat as service target. + if len(typeName) >= 8 && typeName[:8] == "service." { + target = "service" + } + rdm.ConfigTypes.Set(typeId, &common.ConfigType{Id: typeId, Name: typeName, Target: target}) + } + for cfgId, typeId := range configs { + rdm.Configs.Set(cfgId, &common.Config{Id: cfgId, Name: cfgId, TypeId: typeId, DataJson: cfgId + "-data"}) + } + return rdm +} + +func sortedCopy(s []string) []string { + out := append([]string(nil), s...) + sort.Strings(out) + return out +} + +func Test_DispatchRouterConfigDiff_BothNilNoOp(t *testing.T) { + rec := &recordingSub{} + dispatchRouterConfigDiff(nil, nil, rec) + require.Empty(t, rec.applied) + require.Empty(t, rec.removed) +} + +func Test_DispatchRouterConfigDiff_NilSubscriberNoPanic(t *testing.T) { + rdm := newRdmWithConfigs("r1", + map[string]string{"cfg-A": "rt"}, + map[string]string{"rt": "router.link.v1"}, + ) + dispatchRouterConfigDiff(nil, rdm, nil) + // no panic = pass +} + +func Test_DispatchRouterConfigDiff_OnlyNewState_AllApplied(t *testing.T) { + req := require.New(t) + rec := &recordingSub{} + newRdm := newRdmWithConfigs("r1", + map[string]string{"cfg-A": "rt", "cfg-B": "rt2"}, + map[string]string{"rt": "router.link.v1", "rt2": "router.xgress.v1"}, + ) + dispatchRouterConfigDiff(nil, newRdm, rec) + + req.Empty(rec.removed) + req.ElementsMatch([]string{"router.link.v1=cfg-A-data", "router.xgress.v1=cfg-B-data"}, rec.applied) +} + +func Test_DispatchRouterConfigDiff_OnlyOldState_AllRemoved(t *testing.T) { + req := require.New(t) + rec := &recordingSub{} + oldRdm := newRdmWithConfigs("r1", + map[string]string{"cfg-A": "rt", "cfg-B": "rt2"}, + map[string]string{"rt": "router.link.v1", "rt2": "router.xgress.v1"}, + ) + dispatchRouterConfigDiff(oldRdm, nil, rec) + + req.Empty(rec.applied) + req.ElementsMatch([]string{"router.link.v1", "router.xgress.v1"}, sortedCopy(rec.removed)) +} + +func Test_DispatchRouterConfigDiff_MixedAddRemoveKeep(t *testing.T) { + req := require.New(t) + rec := &recordingSub{} + + oldRdm := newRdmWithConfigs("r1", + map[string]string{"cfg-keep": "rt-link", "cfg-gone": "rt-gone"}, + map[string]string{"rt-link": "router.link.v1", "rt-gone": "router.gone.v1"}, + ) + newRdm := newRdmWithConfigs("r1", + map[string]string{"cfg-keep": "rt-link", "cfg-new": "rt-new"}, + map[string]string{"rt-link": "router.link.v1", "rt-new": "router.new.v1"}, + ) + + dispatchRouterConfigDiff(oldRdm, newRdm, rec) + + req.ElementsMatch([]string{"router.gone.v1"}, rec.removed) + // cfg-keep is unchanged (same data) so it should NOT be re-applied; only + // the new config dispatches. + req.ElementsMatch([]string{"router.new.v1=cfg-new-data"}, rec.applied) +} + +func Test_DispatchRouterConfigDiff_UnchangedConfigsSkipped(t *testing.T) { + req := require.New(t) + rec := &recordingSub{} + + // Both RDMs have the same configs with the same data. Diff must be empty. + oldRdm := newRdmWithConfigs("r1", + map[string]string{"cfg-A": "rt-1", "cfg-B": "rt-2"}, + map[string]string{"rt-1": "router.a.v1", "rt-2": "router.b.v1"}, + ) + newRdm := newRdmWithConfigs("r1", + map[string]string{"cfg-A": "rt-1", "cfg-B": "rt-2"}, + map[string]string{"rt-1": "router.a.v1", "rt-2": "router.b.v1"}, + ) + + dispatchRouterConfigDiff(oldRdm, newRdm, rec) + + req.Empty(rec.removed) + req.Empty(rec.applied) +} + +func Test_DispatchRouterConfigDiff_ChangedDataReapplied(t *testing.T) { + req := require.New(t) + rec := &recordingSub{} + + oldRdm := common.NewBareRouterDataModel("r1") + oldRdm.ConfigTypes.Set("rt", &common.ConfigType{Id: "rt", Name: "router.link.v1", Target: common.ConfigTypeTargetRouter}) + oldRdm.Configs.Set("cfg", &common.Config{Id: "cfg", Name: "cfg", TypeId: "rt", DataJson: `{"old":true}`}) + + newRdm := common.NewBareRouterDataModel("r1") + newRdm.ConfigTypes.Set("rt", &common.ConfigType{Id: "rt", Name: "router.link.v1", Target: common.ConfigTypeTargetRouter}) + newRdm.Configs.Set("cfg", &common.Config{Id: "cfg", Name: "cfg", TypeId: "rt", DataJson: `{"new":true}`}) + + dispatchRouterConfigDiff(oldRdm, newRdm, rec) + + req.Empty(rec.removed) + req.Equal([]string{`router.link.v1={"new":true}`}, rec.applied) +} + +func Test_DispatchRouterConfigDiff_SkipsNonRouterTarget(t *testing.T) { + req := require.New(t) + rec := &recordingSub{} + newRdm := newRdmWithConfigs("r1", + map[string]string{"cfg-svc": "svc-type", "cfg-router": "rt"}, + map[string]string{"svc-type": "service.intercept.v1", "rt": "router.link.v1"}, + ) + dispatchRouterConfigDiff(nil, newRdm, rec) + + req.ElementsMatch([]string{"router.link.v1=cfg-router-data"}, rec.applied) + req.Empty(rec.removed) +} + +func Test_CollectRouterConfigTypes_NilRdm(t *testing.T) { + req := require.New(t) + out := collectRouterConfigTypes(nil) + req.NotNil(out) + req.Empty(out) +} diff --git a/ziti/cmd/fabric/inspect.go b/ziti/cmd/fabric/inspect.go index e9f655348..0dbb8dccf 100644 --- a/ziti/cmd/fabric/inspect.go +++ b/ziti/cmd/fabric/inspect.go @@ -40,6 +40,7 @@ func NewInspectCmd(p common.OptionsProvider) *cobra.Command { cmd.AddCommand(action.newInspectSubCmd(p, "router-messaging", "gets information about pending router peer updates and terminator validations")) cmd.AddCommand(action.newInspectSubCmd(p, "router-data-model", "gets information about the router data model")) cmd.AddCommand(action.newInspectSubCmd(p, "router-data-model-index", "gets current index of the router data model")) + cmd.AddCommand(action.newInspectSubCmd(p, inspectCommon.RouterConfigRegistryKey, "gets the router's managed-config registry state (handlers, controller/local versions, applied state)")) cmd.AddCommand(action.newInspectSubCmd(p, "data-model-index", "gets current index of the controller data model")) cmd.AddCommand(action.newInspectSubCmd(p, "router-controllers", "gets information about the state of a router's connections to its controllers")) cmd.AddCommand(action.newInspectSubCmd(p, "terminator-costs", "gets information about terminator dynamic costs"))