diff --git a/config/parsing/service/parse.go b/config/parsing/service/parse.go index 6ae9eec..9c69ed1 100644 --- a/config/parsing/service/parse.go +++ b/config/parsing/service/parse.go @@ -2,6 +2,7 @@ package service import ( "fmt" + "io" "runtime" "strings" "time" @@ -209,9 +210,11 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { } lnRouter := xchain.NewRouter(routerOpts...) - // The routers may hold a chainGroup running probe goroutines; make sure - // they are closed if service construction fails after this point. + // The routers may hold a chainGroup running probe goroutines, and the + // forwarder hop may run probe goroutines of its own; make sure they are + // closed if service construction fails after this point. var hRouter *xchain.Router + var fwdCloser io.Closer success := false defer func() { if !success { @@ -219,6 +222,9 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { if hRouter != nil { hRouter.Close() } + if fwdCloser != nil { + fwdCloser.Close() + } } }() @@ -422,10 +428,11 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { } if forwarder, ok := h.(handler.Forwarder); ok { - hop, err := parseForwarder(cfg.Forwarder, log) + hop, closer, err := parseForwarder(cfg.Forwarder, log) if err != nil { return nil, err } + fwdCloser = closer forwarder.Forward(hop) } @@ -438,6 +445,10 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { return nil, err } + closers := []io.Closer{lnRouter, hRouter} + if fwdCloser != nil { + closers = append(closers, fwdCloser) + } s := xservice.NewService(cfg.Name, ln, h, xservice.AdmissionOption(xadmission.AdmissionGroup(admissions...)), xservice.PreUpOption(preUp), @@ -450,7 +461,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { xservice.ObserverPeriodOption(observerPeriod), xservice.LoggerOption(serviceLogger), xservice.LabelsOption(labels), - xservice.ClosersOption(lnRouter, hRouter), + xservice.ClosersOption(closers...), ) success = true @@ -458,14 +469,23 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { return s, nil } -func parseForwarder(cfg *config.ForwarderConfig, log logger.Logger) (hop.Hop, error) { +// parseForwarder builds the forwarder hop for a service. The returned +// io.Closer is non-nil only when the hop is owned by the service (inline +// nodes or a hop group) and must be closed on service shutdown; hops from +// the HopRegistry are closed by the registry on unregister. +func parseForwarder(cfg *config.ForwarderConfig, log logger.Logger) (hop.Hop, io.Closer, error) { if cfg == nil { - return nil, nil + return nil, nil, nil } // HopGroup takes precedence over Hop / Name / Nodes. if cfg.HopGroup != nil { - return parseHopGroup(cfg.HopGroup, log) + group, err := parseHopGroup(cfg.HopGroup, log) + if err != nil || group == nil { + return nil, nil, err + } + closer, _ := group.(io.Closer) + return group, closer, nil } hopName := cfg.Hop @@ -473,7 +493,7 @@ func parseForwarder(cfg *config.ForwarderConfig, log logger.Logger) (hop.Hop, er hopName = cfg.Name } if hopName != "" { - return registry.HopRegistry().Get(hopName), nil + return registry.HopRegistry().Get(hopName), nil, nil } hc := config.HopConfig{ @@ -519,7 +539,13 @@ func parseForwarder(cfg *config.ForwarderConfig, log logger.Logger) (hop.Hop, er Metadata: node.Metadata, }) } - return hop_parser.ParseHop(&hc, log) + // Inline nodes build a per-service hop, owned (and closed) by the service. + h, err := hop_parser.ParseHop(&hc, log) + if err != nil || h == nil { + return nil, nil, err + } + closer, _ := h.(io.Closer) + return h, closer, nil } func parseHopGroup(cfg *config.ForwardHopGroupConfig, log logger.Logger) (hop.Hop, error) { diff --git a/hop/group.go b/hop/group.go index 590148c..221536d 100644 --- a/hop/group.go +++ b/hop/group.go @@ -187,14 +187,11 @@ func (g *hopGroup) Nodes() []*chain.Node { return nodes } -// Close stops all probe goroutines and closes each hop entry. +// Close stops all probe goroutines. It does NOT close the underlying hops: +// they come from the HopRegistry and are shared — the registry closes them +// on unregister. func (g *hopGroup) Close() error { g.cancelFunc() - for _, e := range g.entries { - if closer, ok := e.hop.(interface{ Close() error }); ok { - closer.Close() - } - } return nil }