Similarityoung commented on code in PR #1042:
URL: https://github.com/apache/dubbo-go-pixiu/pull/1042#discussion_r4143273249
##########
pkg/server/listener_manager.go:
##########
@@ -164,6 +183,272 @@ func (lm *ListenerManager) UpdateListener(m
*model.Listener) error {
return nil
}
+// UpsertXDSListener adds or updates a listener owned by xDS. An xDS resource
+// is not allowed to replace a listener created from static configuration.
+func (lm *ListenerManager) UpsertXDSListener(m *model.Listener) error {
+ if m == nil {
+ return errors.New("xDS listener config is nil")
+ }
+
+ listenerKey := resolveListenerName(m)
+ lm.rwLock.Lock()
+ if active, exists := lm.activeListenerService[listenerKey]; exists {
+ if _, owned := lm.xdsManaged[listenerKey]; !owned {
+ lm.rwLock.Unlock()
+ return errors.Errorf("xDS listener %q conflicts with a
non-xDS listener", listenerKey)
+ }
+ logger.Infof("Update xDS Listener %s (key: %s)", m.Name,
listenerKey)
+ if err := active.Refresh(*m); err != nil {
+ lm.rwLock.Unlock()
+ return errors.Wrapf(err, "refresh xDS listener %q",
listenerKey)
+ }
+ active.config = m
+ lm.rwLock.Unlock()
+ return nil
+ }
+
+ gate := lm.listenerUpdateGateLocked()
+ ls, err := listener.CreateListenerServiceWithUpdateGate(m,
lm.bootstrap, gate)
+ if err != nil {
+ lm.rwLock.Unlock()
+ return err
+ }
+ setListenerActive(ls, false)
+ gate.Lock()
+ if err := startListenerServiceSafely(ls); err != nil {
+ gate.Unlock()
+ lm.rwLock.Unlock()
+ _ = closeListenerServiceSafely(ls)
+ return errors.Wrapf(err, "start xDS listener %q", listenerKey)
+ }
+ lm.activeListenerService[listenerKey] = &wrapListenerService{
+ config: m,
+ ListenerService: ls,
+ }
+ lm.xdsManaged[listenerKey] = struct{}{}
+ setListenerActive(ls, true)
+ gate.Unlock()
+ lm.rwLock.Unlock()
+
+ logger.Infof("Add xDS Listener %s (key: %s)", m.Name, listenerKey)
+ return nil
+}
+
+// ReplaceXDSListeners replaces the complete xDS-owned listener set as one
+// transaction. Every filter chain is fully prepared first; new sockets must
+// bind successfully before the shared request gate publishes all changes.
+// Rejected responses therefore leave the last-good listener set running.
+func (lm *ListenerManager) ReplaceXDSListeners(listeners []*model.Listener)
error {
+ lm.rwLock.Lock()
+ gate := lm.listenerUpdateGateLocked()
+
+ newManaged := make(map[string]struct{}, len(listeners))
+ staged := make(map[string]*wrapListenerService)
+ type refreshTarget struct {
+ key string
+ active *wrapListenerService
+ transactional transactionalListenerService
+ next *model.Listener
+ prepared *listener.PreparedUpdate
+ }
+ refreshes := make([]refreshTarget, 0, len(listeners))
+ closePrepared := func() {
+ for _, target := range refreshes {
+ if err :=
listener.ClosePreparedUpdate(target.prepared); err != nil {
+ logger.Warnf("close rejected xDS listener %s
filter chain: %v", target.key, err)
+ }
+ }
+ }
+ committed := false
+ defer func() {
+ if committed {
+ return
+ }
+ closePrepared()
+ closeStagedListeners(staged)
+ }()
+
+ for _, listenerConfig := range listeners {
+ if listenerConfig == nil {
+ lm.rwLock.Unlock()
+ return errors.New("xDS listener config is nil")
+ }
+ key := resolveListenerName(listenerConfig)
+ if _, duplicate := newManaged[key]; duplicate {
+ lm.rwLock.Unlock()
+ return errors.Errorf("duplicate xDS listener %q", key)
+ }
+ newManaged[key] = struct{}{}
+
+ if active := lm.activeListenerService[key]; active != nil {
+ if _, owned := lm.xdsManaged[key]; !owned {
+ lm.rwLock.Unlock()
+ return errors.Errorf("xDS listener %q conflicts
with a non-xDS listener", key)
+ }
+ transactional, prepared, err :=
prepareListenerRefreshSafely(active.ListenerService, *listenerConfig)
+ if err != nil {
+ lm.rwLock.Unlock()
+ return errors.Wrapf(err, "prepare xDS listener
%q", key)
+ }
+ refreshes = append(refreshes, refreshTarget{key: key,
active: active, transactional: transactional, next: listenerConfig, prepared:
prepared})
+ continue
+ }
+
+ service, err := createListenerServiceSafely(listenerConfig,
lm.bootstrap, gate)
+ if err != nil {
+ lm.rwLock.Unlock()
+ return errors.Wrapf(err, "create xDS listener %q", key)
+ }
+ setListenerActive(service, false)
+ staged[key] = &wrapListenerService{config: listenerConfig,
ListenerService: service}
+ }
+
+ gate.Lock()
+ for key, active := range staged {
+ if err := startListenerServiceSafely(active.ListenerService);
err != nil {
Review Comment:
当 xDS 将同一地址、端口的监听器从 HTTP 改为 HTTP2
时，协议变化使它成为新的管理键。这里先启动新监听器，旧监听器却要到提交后才关闭，因此新监听器绑定端口必然失败，整次有效更新被
NACK。变更前的流程会先移除旧监听器。需要处理这种同端口交接。
--
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]
