AlexStocks commented on code in PR #3482:
URL: https://github.com/apache/dubbo-go/pull/3482#discussion_r3610413475
##########
registry/polaris/core.go:
##########
@@ -72,60 +108,266 @@ func (watcher *PolarisServiceWatcher) lazyRun() {
})
}
+func (watcher *PolarisServiceWatcher) addRegistrySubscriber(
+ initialSnapshot []model.Instance,
+ mode initialReconcileMode,
+ subscriber item,
+ fullSnapshotSubscriber func([]model.Instance),
+) {
+ state := &subscriberState{
+ notify: subscriber,
+ notifyFullSnapshot: fullSnapshotSubscriber,
+ initialSnapshot: copyInstances(initialSnapshot),
+ initialMode: mode,
+ kind: registrySubscriber,
+ }
+
+ func() {
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ watcher.subscribers = append(watcher.subscribers, state)
+ if watcher.snapshotReady {
+ watcher.reconcileSubscriberLocked(state,
notifiableInstances(watcher.currentInstances, true))
+ }
+ }()
+
+ // Start only after the subscriber is registered and any available
current
+ // snapshot has been replayed.
+ watcher.lazyRun()
+}
+
+// missingInstances returns previous - current by model.InstanceKey while
+// preserving the order of the previous snapshot.
+func missingInstances(previous, current []model.Instance) []model.Instance {
+ currentInstances := make(map[model.InstanceKey]struct{}, len(current))
+ for _, instance := range current {
+ currentInstances[instance.GetInstanceKey()] = struct{}{}
+ }
+
+ missing := make([]model.Instance, 0, len(previous))
+ for _, instance := range previous {
+ if _, ok := currentInstances[instance.GetInstanceKey()]; !ok {
+ missing = append(missing, instance)
+ }
+ }
+ return missing
+}
+
+func notifiableInstances(instances []model.Instance, reportInvalid bool)
[]model.Instance {
+ valid := make([]model.Instance, 0, len(instances))
+ for _, instance := range instances {
+ validationError := polarisInstanceURLValidationError(instance)
+ if validationError == "" {
+ valid = append(valid, instance)
+ } else if reportInvalid {
+ logger.Errorf("[Registry][Polaris] %s, instance=%+v",
validationError, instance)
+ }
+ }
+ return valid
+}
+
+// handleWatchSnapshot replaces the watcher's current state and reconciles each
+// subscriber's own synchronous-load baseline exactly once.
+func (watcher *PolarisServiceWatcher) handleWatchSnapshot(current
[]model.Instance) {
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ registryCurrent := notifiableInstances(current,
watcher.hasRegistrySubscriberLocked())
+ previous := watcher.currentInstances
+ hadPreviousSnapshot := watcher.snapshotReady
+ next := copyInstances(current)
+ var removedSincePrevious []model.Instance
+ if hadPreviousSnapshot {
+ registryPrevious := notifiableInstances(previous, false)
+ removedSincePrevious = missingInstances(registryPrevious,
registryCurrent)
+ }
+ watcher.currentInstances = next
+ watcher.snapshotReady = true
+ for _, subscriber := range watcher.subscribers {
+ if subscriber.kind == applicationSubscriber {
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeAdd, watcher.currentInstances)
+ continue
+ }
+ if subscriber.reconciled {
+ if len(removedSincePrevious) > 0 {
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeDel, removedSincePrevious)
+ }
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeAdd, registryCurrent)
+ continue
+ }
+ watcher.reconcileSubscriberLocked(subscriber, registryCurrent)
+ }
+}
+
+func (watcher *PolarisServiceWatcher) reconcileSubscriberLocked(
+ subscriber *subscriberState,
+ current []model.Instance,
+) {
+ switch subscriber.initialMode {
+ case reconcileWithBaseline:
+ missing := missingInstances(subscriber.initialSnapshot, current)
+ if len(missing) > 0 {
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeDel, missing)
+ }
+ watcher.notifySubscriberLocked(subscriber,
remoting.EventTypeAdd, current)
+ case reconcileWithFullSnapshot:
+ subscriber.notifyFullSnapshot(copyInstances(current))
+ }
+ subscriber.reconciled = true
+ subscriber.initialSnapshot = nil
+}
+
// startWatch start run work to watch target service by polaris
func (watcher *PolarisServiceWatcher) startWatch() {
for {
- resp, err :=
watcher.consumer.WatchService(watcher.subscribeParam)
- if err != nil {
+ if err := watcher.watchOnce(); err != nil {
time.Sleep(time.Duration(500 * time.Millisecond))
- continue
}
- watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value: resp.GetAllInstancesResp.Instances,
- ConfigType: remoting.EventTypeAdd,
- })
-
- for event := range resp.EventChannel {
- eType := event.GetSubScribeEventType()
- if eType == internalapi.EventInstance {
- insEvent := event.(*model.InstanceEvent)
-
- if insEvent.AddEvent != nil {
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value:
insEvent.AddEvent.Instances,
- ConfigType:
remoting.EventTypeAdd,
- })
- }
- if insEvent.UpdateEvent != nil {
- instances := make([]model.Instance,
len(insEvent.UpdateEvent.UpdateList))
- for i := range
insEvent.UpdateEvent.UpdateList {
- instances[i] =
insEvent.UpdateEvent.UpdateList[i].After
- }
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value: instances,
- ConfigType:
remoting.EventTypeUpdate,
- })
- }
- if insEvent.DeleteEvent != nil {
-
watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
- Value:
insEvent.DeleteEvent.Instances,
- ConfigType:
remoting.EventTypeDel,
- })
- }
+ }
+}
+
+func (watcher *PolarisServiceWatcher) watchOnce() error {
+ resp, err := watcher.consumer.WatchService(watcher.subscribeParam)
+ if err != nil {
+ return err
+ }
+ watcher.handleWatchSnapshot(resp.GetAllInstancesResp.Instances)
+ for event := range resp.EventChannel {
+ if event.GetSubScribeEventType() == internalapi.EventInstance {
+
watcher.handleInstanceEvent(event.(*model.InstanceEvent))
+ }
+ }
+ return nil
+}
+
+func (watcher *PolarisServiceWatcher) handleInstanceEvent(event
*model.InstanceEvent) {
+ if event == nil {
+ return
+ }
+
+ watcher.lock.Lock()
+ defer watcher.lock.Unlock()
+
+ if event.AddEvent != nil {
+ instances := copyInstances(event.AddEvent.Instances)
+ for _, instance := range instances {
+ watcher.upsertCurrentInstanceLocked(instance)
+ }
+
watcher.notifyReconciledSubscribersLocked(remoting.EventTypeAdd, instances)
+ }
+ if event.UpdateEvent != nil {
+ reportInvalid := watcher.hasRegistrySubscriberLocked()
+ applicationUpdates := make([]model.Instance, 0,
len(event.UpdateEvent.UpdateList))
+ registryUpdates := make([]model.Instance, 0,
len(event.UpdateEvent.UpdateList))
+ registryDeletes := make([]model.Instance, 0,
len(event.UpdateEvent.UpdateList))
+ changedKeyBefore := make([]model.Instance, 0,
len(event.UpdateEvent.UpdateList))
+ for _, update := range event.UpdateEvent.UpdateList {
+ if update.Before.GetInstanceKey() !=
update.After.GetInstanceKey() {
+ changedKeyBefore = append(changedKeyBefore,
update.Before)
+ }
+ }
+ watcher.removeCurrentInstancesLocked(changedKeyBefore)
+ for _, update := range event.UpdateEvent.UpdateList {
+ beforeValid :=
polarisInstanceURLValidationError(update.Before) == ""
+ afterValidationError :=
polarisInstanceURLValidationError(update.After)
+ afterValid := afterValidationError == ""
+ keyChanged := update.Before.GetInstanceKey() !=
update.After.GetInstanceKey()
+ watcher.upsertCurrentInstanceLocked(update.After)
+ applicationUpdates = append(applicationUpdates,
update.After)
+ if keyChanged && beforeValid {
+ registryDeletes = append(registryDeletes,
update.Before)
+ }
+ if afterValid {
+ registryUpdates = append(registryUpdates,
update.After)
+ continue
+ }
+ if reportInvalid {
+ logger.Errorf("[Registry][Polaris] %s,
instance=%+v", afterValidationError, update.After)
+ }
+ if !keyChanged && beforeValid {
+ registryDeletes = append(registryDeletes,
update.Before)
}
}
+
watcher.notifyReconciledSubscribersByKindLocked(applicationSubscriber,
remoting.EventTypeUpdate, applicationUpdates)
+
watcher.notifyReconciledSubscribersByKindLocked(registrySubscriber,
remoting.EventTypeDel, registryDeletes)
+
watcher.notifyReconciledSubscribersByKindLocked(registrySubscriber,
remoting.EventTypeUpdate, registryUpdates)
+ }
+ if event.DeleteEvent != nil {
+ instances := copyInstances(event.DeleteEvent.Instances)
+ watcher.removeCurrentInstancesLocked(instances)
+
watcher.notifyReconciledSubscribersLocked(remoting.EventTypeDel, instances)
+ }
+}
+func (watcher *PolarisServiceWatcher) upsertCurrentInstanceLocked(instance
model.Instance) {
+ key := instance.GetInstanceKey()
+ for i := range watcher.currentInstances {
+ if watcher.currentInstances[i].GetInstanceKey() == key {
+ watcher.currentInstances[i] = instance
+ return
+ }
}
+ watcher.currentInstances = append(watcher.currentInstances, instance)
}
-// notifyAllSubscriber notify config_center.ConfigChangeEvent to all subscriber
-func (watcher *PolarisServiceWatcher) notifyAllSubscriber(event
*config_center.ConfigChangeEvent) {
- watcher.lock.RLock()
- defer watcher.lock.RUnlock()
+func (watcher *PolarisServiceWatcher) removeCurrentInstancesLocked(instances
[]model.Instance) {
+ keys := make(map[model.InstanceKey]struct{}, len(instances))
+ for _, instance := range instances {
+ keys[instance.GetInstanceKey()] = struct{}{}
+ }
+
+ old := watcher.currentInstances
+ current := old[:0]
+ for _, instance := range old {
+ if _, remove := keys[instance.GetInstanceKey()]; !remove {
+ current = append(current, instance)
+ }
+ }
+ clear(old[len(current):])
+ watcher.currentInstances = current
+}
- for i := 0; i < len(watcher.subscribers); i++ {
- subscriber := watcher.subscribers[i]
- subscriber(event.ConfigType, event.Value.([]model.Instance))
+func (watcher *PolarisServiceWatcher)
notifyReconciledSubscribersLocked(eventType remoting.EventType, instances
[]model.Instance) {
+ for _, subscriber := range watcher.subscribers {
+ if subscriber.reconciled {
+ watcher.notifySubscriberLocked(subscriber, eventType,
instances)
+ }
}
+}
+
+func (watcher *PolarisServiceWatcher) hasRegistrySubscriberLocked() bool {
+ for _, subscriber := range watcher.subscribers {
+ if subscriber.kind == registrySubscriber {
+ return true
+ }
+ }
+ return false
+}
+
+func (watcher *PolarisServiceWatcher) notifyReconciledSubscribersByKindLocked(
+ kind subscriberKind,
+ eventType remoting.EventType,
+ instances []model.Instance,
+) {
+ if len(instances) == 0 {
+ return
+ }
+ for _, subscriber := range watcher.subscribers {
+ if subscriber.reconciled && subscriber.kind == kind {
+ watcher.notifySubscriberLocked(subscriber, eventType,
instances)
+ }
+ }
+}
+
+func (watcher *PolarisServiceWatcher) notifySubscriberLocked(
+ subscriber *subscriberState,
+ eventType remoting.EventType,
+ instances []model.Instance,
+) {
+ subscriber.notify(eventType, copyInstances(instances))
+}
+func copyInstances(instances []model.Instance) []model.Instance {
+ return append([]model.Instance(nil), instances...)
Review Comment:
[P1] 不要把可变 `model.Instance` 引用作为长期快照共享
这里仅复制了接口切片，元素仍指向 Polaris SDK 的同一实例对象；`InstanceInProto` 嵌入
protobuf，`GetMetadata()` 直接返回底层可写 map。当前 Head 新增的 `currentInstances`
会长期保留这些引用，同时又把相同引用交给 subscriber，因此调用方或任一 subscriber 修改 metadata 后会污染 watcher
的重连基线。WSL overlay 已稳定复现：首个快照发布 A 后删除原实例 metadata 中的
`interface/path`，再处理空重连快照，`notifiableInstances(previous)` 会把被污染的 A
过滤掉，最终没有发出应有的 `DEL(A)`；另一个用例也证明 subscriber 修改回调对象会直接改变
`watcher.currentInstances`。建议为内部状态保存不可变快照（至少独立复制 metadata 和用于 reconciliation 的
key/可通知状态），给每个 subscriber 再提供隔离副本；补充“修改输入实例”和“修改 callback 参数后重连仍发 DEL”的回归测试。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
