Similarityoung commented on code in PR #1042:
URL: https://github.com/apache/dubbo-go-pixiu/pull/1042#discussion_r4072315204
##########
admin/core/xds.go:
##########
@@ -129,168 +131,89 @@ func runXDSServer(ctx context.Context, srv
envoyServer.Server, port uint) error
)
grpcServer := grpc.NewServer(grpcOptions...)
- lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
- if err != nil {
- return err
- }
-
registerServer(grpcServer, srv)
- logger.Infof("management server listening on %d\n", port)
- if err = grpcServer.Serve(lis); err != nil {
- return nil
+ logger.Infof("management server listening on %s", lis.Addr())
+ if err := grpcServer.Serve(lis); err != nil {
+ return err
}
return nil
}
-func watchConfigAndReload() {
- ch, err :=
adminconfig.Client.WatchWithPrefix(adminconfig.Bootstrap.EtcdConfig.Path)
-
- if err != nil {
- logger.Errorf("watch config error %q", err)
- panic(err)
- }
-
- for range ch {
- logger.Info("get etcd config change")
- // Create the config that we'll serve to Envoy
- config := GenerateSnapshotPixiu()
- if err := config.Consistent(); err != nil {
- logger.Errorf("config inconsistency: %+v\n%+v", config,
err)
- os.Exit(1)
- }
-
- // Add the config to the snaphost
- if err := snaphost.SetSnapshot(context.Background(), nodeID,
config); err != nil {
- logger.Errorf("config error %q for %+v", err, config)
- os.Exit(1)
- }
- }
+type snapshotPublisher interface {
+ Publish(ctx context.Context) error
}
-// makeHTTPFilter returns a handler for the given resource.
-func makeHTTPFilter(listener config.Listener) *model.FilterChain {
- var filters, routes []any
-
- for _, f := range listener.HTTPFilters {
- filters = append(filters, map[string]any{
- "name": f.Name,
- "config": f.Config,
- })
- }
-
- for _, r := range listener.RouteConfig.Routes {
- routes = append(filters, map[string]any{
- "match": map[string]any{
- "prefix": r.Match.Prefix,
- },
- "route": map[string]any{
- "cluster":
r.Route.Cluster,
- "cluster_not_found_response_code":
r.Route.ClusterNotFoundResponseCode,
- },
- })
- }
-
- return &model.FilterChain{
- Filters: []*model.NetworkFilter{
- {
- Name: constant.HTTPConnectManagerFilter,
- Config: &model.NetworkFilter_Struct{
- Struct: func() *structpb.Struct {
- v, err :=
structpb.NewStruct(map[string]any{
- "route_config":
map[string]any{
- "routes":
routes,
- },
- "http_filters": filters,
- })
- if err != nil {
- panic(err)
- }
- return v
- }(),
- },
- },
- },
- }
+type configWatcher interface {
+ WatchWithPrefix(key string) (clientv3.WatchChan, error)
}
-func makeListeners() *model.PixiuExtensionListeners {
- listeners, err := logic.BizGetListeners()
- if err != nil {
- logger.Errorf("get listeners error %q", err)
- return nil
- }
-
- if len(listeners) == 0 {
- return nil
+func watchConfigAndReload(ctx context.Context, publisher snapshotPublisher) {
+ if adminconfig.Client == nil {
+ err := fmt.Errorf("watch xDS configuration: etcd client is not
initialized")
+ adminxds.DefaultStatusStore.RecordError(err)
+ logger.Error(err)
+ return
}
-
- pbListeners := &model.PixiuExtensionListeners{}
- for _, listener := range listeners {
- pbListeners.Listeners = append(pbListeners.Listeners,
&model.Listener{
- Name: listener.Name,
- Address: &model.Address{
- SocketAddress: &model.SocketAddress{
- Address:
listener.Address.SocketAddress.Address,
- Port:
int64(listener.Address.SocketAddress.Port),
- },
- Name: listener.Address.Name,
- },
- FilterChain: makeHTTPFilter(listener),
- })
- }
- return pbListeners
+ watchConfigWithRetry(ctx, adminconfig.Client,
adminconfig.Bootstrap.EtcdConfig.Path, publisher, configWatchRetryDelay)
}
-func makeClusters() *model.PixiuExtensionClusters {
- clusters, err := logic.BizGetClusters()
- if err != nil {
- logger.Errorf("get clusters error %q", err)
- return nil
- }
-
- if len(clusters) == 0 {
- return nil
- }
-
- pbCluster := &model.PixiuExtensionClusters{}
+func watchConfigWithRetry(ctx context.Context, watcher configWatcher, path
string, publisher snapshotPublisher, retryDelay time.Duration) {
+ for {
+ ch, err := watcher.WatchWithPrefix(path)
+ if err == nil {
+ // Establish the watch before rebuilding on both
startup and reconnect.
+ // Events that happen during the rebuild remain queued
on the channel.
+ if publishErr := publisher.Publish(ctx); publishErr !=
nil {
Review Comment:
当首次建立 watch 后这里的初始 失败时，只记录日志并继续进入 。如果后续没有新的配置事件或 watch 关闭，初始快照就不会再次构建，xDS
可能一直没有可用配置。这里需要在首次发布失败时重试或触发 watch 重建，不能把这次失败留给后续事件来碰运气。
##########
pkg/common/extension/filter/filter_manager.go:
##########
@@ -58,20 +58,55 @@ func NewEmptyFilterManager() *FilterManager {
return &FilterManager{filters: make(map[string]HttpFilterFactory)}
}
+// CreateFilterChain preserves the original best-effort public API: factories
+// that fail are logged and skipped, while filters prepared successfully by
+// other factories remain in the returned chain. Configuration publication uses
+// CreateFilterChainChecked so errors can reject the complete resource.
func (fm *FilterManager) CreateFilterChain(ctx *http.HttpContext) FilterChain {
- chain := NewDefaultFilterChain()
+ chain, factories := fm.newLeasedFilterChain()
+ for index, f := range factories {
+ if f == nil || *f == nil {
+ logger.Errorf("create HTTP filter chain: HTTP filter
factory %d is nil", index)
+ continue
+ }
+ if err := (*f).PrepareFilterChain(ctx, chain); err != nil {
+ logger.Errorf("create HTTP filter chain: prepare HTTP
filter %d: %v", index, err)
+ }
+ }
+ return chain
+}
+
+func (fm *FilterManager) CreateFilterChainChecked(ctx *http.HttpContext)
(FilterChain, error) {
+ chain, factories := fm.newLeasedFilterChain()
+ for index, f := range factories {
+ if f == nil || *f == nil {
+ releaseFilterChain(chain)
+ return nil, errors.Errorf("HTTP filter factory %d is
nil", index)
+ }
+ if err := (*f).PrepareFilterChain(ctx, chain); err != nil {
+ releaseFilterChain(chain)
+ return nil, errors.Wrapf(err, "prepare HTTP filter %d",
index)
+ }
+ }
+ return chain, nil
+}
+func (fm *FilterManager) newLeasedFilterChain() (FilterChain,
[]*HttpFilterFactory) {
+ chain := NewDefaultFilterChain()
fm.mu.RLock()
- defer fm.mu.RUnlock()
factories := append([]*HttpFilterFactory(nil), fm.filtersArray...)
fm.leaseFactories(factories)
Review Comment:
为返回的 chain 持有所有 factory 的 lease，但 返回的 接口没有 契约；按原有公开 API 使用的调用方无法可靠释放这些
lease。配置 reload 后旧 factory 可能因此一直保留， 永远不会执行。需要让该 API
的生命周期契约可被调用方使用，或避免在无释放契约的路径上持有 lease。
--
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]
