From b3c19f8be14b5e929f76098c2ea61b1d9a6908b7 Mon Sep 17 00:00:00 2001 From: Andrey Butusov Date: Fri, 31 Jul 2026 18:35:58 +0300 Subject: [PATCH 1/3] node: prepare gRPC servers before SIGHUP reload Build replacement gRPC servers and bind newly added endpoints before stopping existing servers. This keeps the current Public API available when the new TLS configuration is invalid or a new endpoint is occupied. Track the configuration snapshot associated with running servers. Update it only after a successful reload, so a failed rebind is retried on the next SIGHUP instead of reusing a stopped server. Signed-off-by: Andrey Butusov --- CHANGELOG.md | 1 + cmd/neofs-node/config.go | 8 ++- cmd/neofs-node/grpc.go | 116 ++++++++++++++++++++++++++++++++------- 3 files changed, 101 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b87bede4e..663d61d58b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ Changelog for NeoFS Node ### Fixed - SN could panic on forwarding GET/HEAD/RANGE request (#4120) +- gRPC SIGHUP reload no longer stops a running public API before replacement configuration is verified (#4113) ### Changed - SNs exchange TLS certificates on inter-node connections (#4097) diff --git a/cmd/neofs-node/config.go b/cmd/neofs-node/config.go index e20fa3451e..ac7d31c615 100644 --- a/cmd/neofs-node/config.go +++ b/cmd/neofs-node/config.go @@ -237,6 +237,7 @@ type cfgGRPC struct { listeners []net.Listener servers []*grpc.Server + config grpcConfigSnapshot // serviceRegistrators stores functions that register gRPC service // implementations into a gRPC server. @@ -251,7 +252,9 @@ func (g *cfgGRPC) registerService(f func(*grpc.Server)) { g.serviceRegistrators = append(g.serviceRegistrators, f) for _, srv := range g.servers { - f(srv) + if srv != nil { + f(srv) + } } } @@ -642,7 +645,6 @@ func (c *cfg) configWatcher(ctx context.Context) { oldMetrics := writeMetricConfig(c.appCfg) oldProfiler := writeProfilerConfig(c.appCfg) - oldGRPC := writeGRPCConfig(c.appCfg) c.appCfg, err = config.New(config.WithConfigFile(c.appCfg.Path())) if err != nil { @@ -694,7 +696,7 @@ func (c *cfg) configWatcher(ctx context.Context) { // gRPC - if err = reloadGRPC(c, oldGRPC); err != nil { + if err = reloadGRPC(c); err != nil { c.log.Error("gRPC configuration reload", zap.Error(err)) continue } diff --git a/cmd/neofs-node/grpc.go b/cmd/neofs-node/grpc.go index ac862da20e..2c7576a677 100644 --- a/cmd/neofs-node/grpc.go +++ b/cmd/neofs-node/grpc.go @@ -101,6 +101,7 @@ func initGRPC(c *cfg) { if err := buildGRPCServers(c, maxRecvMsgSizeOpt); err != nil { fatalOnErr(err) } + c.cfgGRPC.config = writeGRPCConfig(c.appCfg) // register a single shutdown hook that stops whatever servers are current // at the time of shutdown (including those created by reload). @@ -110,7 +111,9 @@ func initGRPC(c *cfg) { copy(srvs, c.cfgGRPC.servers) c.cfgGRPC.mu.Unlock() for _, srv := range srvs { - stopGRPC("NeoFS Public API", srv, c.log) + if srv != nil { + stopGRPC("NeoFS Public API", srv, c.log) + } } }) } @@ -169,6 +172,23 @@ func buildGRPCServers(c *cfg, maxRecvMsgSizeOpt grpc.ServerOption) error { } func buildSingleGRPCServer(c *cfg, sc grpcconfig.GRPC, maxRecvMsgSizeOpt grpc.ServerOption) (*grpc.Server, net.Listener, error) { + srv, err := newGRPCServer(c, sc, maxRecvMsgSizeOpt) + if err != nil { + return nil, nil, err + } + + lis, err := listenGRPC(sc) + if err != nil { + return nil, nil, err + } + + return srv, lis, nil +} + +// newGRPCServer creates a server without binding its listener. Keeping these +// operations separate lets reload validate a replacement before stopping the +// running server on the same endpoint. +func newGRPCServer(c *cfg, sc grpcconfig.GRPC, maxRecvMsgSizeOpt grpc.ServerOption) (*grpc.Server, error) { serverOpts := []grpc.ServerOption{ grpc.MaxSendMsgSize(maxMsgSize), grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ @@ -192,7 +212,7 @@ func buildSingleGRPCServer(c *cfg, sc grpcconfig.GRPC, maxRecvMsgSizeOpt grpc.Se if _, err := tls.LoadX509KeyPair(certFile, keyFile); err != nil { c.log.Error("could not read certificate from file", zap.Error(err)) - return nil, nil, err + return nil, err } // read certificate from disk on each handshake to pick up renewals automatically. @@ -212,23 +232,26 @@ func buildSingleGRPCServer(c *cfg, sc grpcconfig.GRPC, maxRecvMsgSizeOpt grpc.Se serverOpts = append(serverOpts, grpc.Creds(creds)) } + return grpc.NewServer(serverOpts...), nil +} + +func listenGRPC(sc grpcconfig.GRPC) (net.Listener, error) { lis, err := net.Listen("tcp", sc.Endpoint) if err != nil { - c.log.Error("can't listen gRPC endpoint", zap.Error(err)) - return nil, nil, err + return nil, err } if connLimit := sc.ConnLimit; connLimit > 0 { lis = netutil.LimitListener(lis, connLimit) } - return grpc.NewServer(serverOpts...), lis, nil + return lis, nil } // reloadGRPC performs a fine-grained reload: only gRPC servers whose // configuration or TLS certificate has actually changed are stopped and // re-created; the rest continue serving without interruption. -func reloadGRPC(c *cfg, oldCfg grpcConfigSnapshot) error { +func reloadGRPC(c *cfg) error { newCfg := writeGRPCConfig(c.appCfg) maxRecvMsgSizeOpt, err := getMaxRecvMsgSizeOpt(c) @@ -239,14 +262,23 @@ func reloadGRPC(c *cfg, oldCfg grpcConfigSnapshot) error { c.cfgGRPC.mu.Lock() defer c.cfgGRPC.mu.Unlock() + oldCfg := c.cfgGRPC.config + type serverEntry struct { srv *grpc.Server lis net.Listener snap grpcServerSnapshot } + type pendingServer struct { + index int + srv *grpc.Server + snap grpcServerSnapshot + replaced *serverEntry + } + oldByEndpoint := make(map[string]serverEntry, len(c.cfgGRPC.servers)) for i, srv := range c.cfgGRPC.servers { - if i < len(oldCfg) { + if i < len(oldCfg) && srv != nil && c.cfgGRPC.listeners[i] != nil { oldByEndpoint[oldCfg[i].Endpoint] = serverEntry{ srv: srv, lis: c.cfgGRPC.listeners[i], @@ -255,37 +287,78 @@ func reloadGRPC(c *cfg, oldCfg grpcConfigSnapshot) error { } } - newServers := make([]*grpc.Server, 0, len(newCfg)) - newListeners := make([]net.Listener, 0, len(newCfg)) - // freshServers/freshListeners hold only newly created servers that need - // service registration and must start serving. + newServers := make([]*grpc.Server, len(newCfg)) + newListeners := make([]net.Listener, len(newCfg)) var freshServers []*grpc.Server var freshListeners []net.Listener + var pending []pendingServer + + closeFreshListeners := func() { + for _, lis := range freshListeners { + _ = lis.Close() + } + } - for _, newSnap := range newCfg { + // Construct every replacement before stopping any existing server. New + // endpoints are also bound here, so an invalid TLS configuration or an + // occupied new endpoint leaves the current API untouched. + for i, newSnap := range newCfg { if old, ok := oldByEndpoint[newSnap.Endpoint]; ok { delete(oldByEndpoint, newSnap.Endpoint) if old.snap.unchanged(newSnap) { - newServers = append(newServers, old.srv) - newListeners = append(newListeners, old.lis) + newServers[i] = old.srv + newListeners[i] = old.lis continue } - stopGRPC("NeoFS Public API", old.srv, c.log) + + srv, err := newGRPCServer(c, newSnap.GRPC, maxRecvMsgSizeOpt) + if err != nil { + closeFreshListeners() + return fmt.Errorf("build gRPC server for %q: %w", newSnap.Endpoint, err) + } + pending = append(pending, pendingServer{index: i, srv: srv, snap: newSnap, replaced: &old}) + continue } srv, lis, err := buildSingleGRPCServer(c, newSnap.GRPC, maxRecvMsgSizeOpt) if err != nil { - c.log.Error("failed to start gRPC server", - zap.String("endpoint", newSnap.Endpoint), zap.Error(err)) - continue + closeFreshListeners() + return fmt.Errorf("build gRPC server for %q: %w", newSnap.Endpoint, err) } - newServers = append(newServers, srv) - newListeners = append(newListeners, lis) + newServers[i] = srv + newListeners[i] = lis freshServers = append(freshServers, srv) freshListeners = append(freshListeners, lis) } - // stop servers that were removed from the config entirely + // A listener on an unchanged endpoint cannot be created before its old + // server is stopped. All configuration-dependent work was preflighted + // above, so the only remaining failure is a race for the TCP port. + for _, entry := range pending { + stopGRPC("NeoFS Public API", entry.replaced.srv, c.log) + } + for _, entry := range pending { + lis, err := listenGRPC(entry.snap.GRPC) + if err != nil { + closeFreshListeners() + for _, pendingEntry := range pending { + for i, srv := range c.cfgGRPC.servers { + if srv == pendingEntry.replaced.srv { + c.cfgGRPC.servers[i] = nil + c.cfgGRPC.listeners[i] = nil + } + } + } + return fmt.Errorf("listen gRPC endpoint %q: %w", entry.snap.Endpoint, err) + } + newServers[entry.index] = entry.srv + newListeners[entry.index] = lis + freshServers = append(freshServers, entry.srv) + freshListeners = append(freshListeners, lis) + } + + // Removed endpoints are still serving while replacement endpoints are + // rebound above. Do not take them down if rebinding fails. for _, entry := range oldByEndpoint { stopGRPC("NeoFS Public API", entry.srv, c.log) } @@ -296,6 +369,7 @@ func reloadGRPC(c *cfg, oldCfg grpcConfigSnapshot) error { c.cfgGRPC.servers = newServers c.cfgGRPC.listeners = newListeners + c.cfgGRPC.config = newCfg for _, reg := range c.cfgGRPC.serviceRegistrators { for _, srv := range freshServers { From eb729908de4534b4ad885f26e5aa54a55923dddd Mon Sep 17 00:00:00 2001 From: Andrey Butusov Date: Fri, 31 Jul 2026 18:35:58 +0300 Subject: [PATCH 2/3] node: validate and apply configuration on SIGHUP Validate the newly read configuration before applying it. Invalid configuration now aborts reload before services are modified. Stop the node when configuration reload fails, preventing it from running with a runtime state that can differ from the configuration file or environment. Systemd restarts the node according to the service restart policy. Signed-off-by: Andrey Butusov --- CHANGELOG.md | 1 + cmd/neofs-node/config.go | 101 ++++++++++++++++++++------------------- 2 files changed, 54 insertions(+), 48 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 663d61d58b..bf40a51da1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ Changelog for NeoFS Node ### Fixed - SN could panic on forwarding GET/HEAD/RANGE request (#4120) - gRPC SIGHUP reload no longer stops a running public API before replacement configuration is verified (#4113) +- Storage node shuts down when SIGHUP configuration reload fails (#4113) ### Changed - SNs exchange TLS certificates on inter-node connections (#4097) diff --git a/cmd/neofs-node/config.go b/cmd/neofs-node/config.go index ac7d31c615..cd27e12235 100644 --- a/cmd/neofs-node/config.go +++ b/cmd/neofs-node/config.go @@ -630,7 +630,6 @@ func (c *cfg) needBootstrap() bool { } func (c *cfg) configWatcher(ctx context.Context) { - var err error ch := make(chan os.Signal, 1) signal.Notify(ch, syscall.SIGHUP) @@ -643,73 +642,79 @@ func (c *cfg) configWatcher(ctx context.Context) { c.log.Warn("failed to notify systemd about reloading", zap.Error(err)) } - oldMetrics := writeMetricConfig(c.appCfg) - oldProfiler := writeProfilerConfig(c.appCfg) + if err := c.reloadConfig(); err != nil { + c.internalErr <- fmt.Errorf("configuration reload: %w", err) + return + } - c.appCfg, err = config.New(config.WithConfigFile(c.appCfg.Path())) - if err != nil { - c.log.Error("configuration reading", zap.Error(err)) - continue + c.log.Info("configuration has been reloaded successfully") + + if err := sdnotify.Send(sdnotify.Ready); err != nil { + c.log.Warn("failed to notify systemd about readiness after reload", zap.Error(err)) } + case <-ctx.Done(): + return + } + } +} - // Prometheus and pprof +//nolint:contextcheck // Reloading HTTP services does not receive a request context. +func (c *cfg) reloadConfig() error { + oldCfg := c.appCfg + oldMetrics := writeMetricConfig(oldCfg) + oldProfiler := writeProfilerConfig(oldCfg) - // nolint:contextcheck - c.reloadMetricsAndPprof(oldMetrics, oldProfiler) + newCfg, err := config.New(config.WithConfigFile(oldCfg.Path())) + if err != nil { + return fmt.Errorf("read configuration: %w", err) + } + if err := validateConfig(newCfg); err != nil { + return fmt.Errorf("validate configuration: %w", err) + } + c.appCfg = newCfg - // Logger + // Prometheus and pprof - err = c.logLevel.UnmarshalText([]byte(c.appCfg.Logger.Level)) - if err != nil { - c.log.Error("invalid logger level configuration", zap.Error(err)) - continue - } + c.reloadMetricsAndPprof(oldMetrics, oldProfiler) - // Policer + // Logger - c.policer.Reload(c.policerOpts()...) + if err := c.logLevel.UnmarshalText([]byte(c.appCfg.Logger.Level)); err != nil { + return fmt.Errorf("set logger level: %w", err) + } - // Storage Engine + // Policer - var rcfg engine.ReConfiguration - for _, optsWithID := range c.shardOpts() { - rcfg.AddShard(optsWithID.configID, optsWithID.shOpts) - } + c.policer.Reload(c.policerOpts()...) - err = c.cfgObject.cfgLocalStorage.localStorage.Reload(rcfg) - if err != nil { - c.log.Error("storage engine configuration update", zap.Error(err)) - continue - } + // Storage Engine - // Morph + var rcfg engine.ReConfiguration + for _, optsWithID := range c.shardOpts() { + rcfg.AddShard(optsWithID.configID, optsWithID.shOpts) + } - c.cli.Reload(client.WithEndpoints(c.appCfg.FSChain.Endpoints)) + if err := c.cfgObject.cfgLocalStorage.localStorage.Reload(rcfg); err != nil { + return fmt.Errorf("update storage engine configuration: %w", err) + } - // Node + // Morph - err = c.reloadNodeAttributes() - if err != nil { - c.log.Error("invalid node attributes configuration", zap.Error(err)) - continue - } + c.cli.Reload(client.WithEndpoints(c.appCfg.FSChain.Endpoints)) - // gRPC + // Node - if err = reloadGRPC(c); err != nil { - c.log.Error("gRPC configuration reload", zap.Error(err)) - continue - } + if err := c.reloadNodeAttributes(); err != nil { + return fmt.Errorf("update node attributes: %w", err) + } - c.log.Info("configuration has been reloaded successfully") + // gRPC - if err := sdnotify.Send(sdnotify.Ready); err != nil { - c.log.Warn("failed to notify systemd about readiness after reload", zap.Error(err)) - } - case <-ctx.Done(): - return - } + if err := reloadGRPC(c); err != nil { + return fmt.Errorf("reload gRPC configuration: %w", err) } + + return nil } // writeSystemAttributes writes app version as defined at compilation From 290f5ffa1ddf1543b56903cb8ae0615aa1ce8a6f Mon Sep 17 00:00:00 2001 From: Andrey Butusov Date: Fri, 31 Jul 2026 18:35:58 +0300 Subject: [PATCH 3/3] node: handle invalid node attributes on SIGHUP Return attribute parsing errors to the configuration reloader instead of terminating the process. Restore the previous node attributes when the updated attributes cannot be parsed. Signed-off-by: Andrey Butusov --- CHANGELOG.md | 1 + cmd/neofs-node/attributes.go | 14 +++++++++----- cmd/neofs-node/netmap.go | 7 +++++-- 3 files changed, 15 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf40a51da1..c756ac0877 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ Changelog for NeoFS Node - SN could panic on forwarding GET/HEAD/RANGE request (#4120) - gRPC SIGHUP reload no longer stops a running public API before replacement configuration is verified (#4113) - Storage node shuts down when SIGHUP configuration reload fails (#4113) +- Invalid node attributes on SIGHUP no longer terminate a storage node (#4113) ### Changed - SNs exchange TLS certificates on inter-node connections (#4097) diff --git a/cmd/neofs-node/attributes.go b/cmd/neofs-node/attributes.go index 2a3546eeb3..802a67b8f9 100644 --- a/cmd/neofs-node/attributes.go +++ b/cmd/neofs-node/attributes.go @@ -8,24 +8,26 @@ import ( "go.uber.org/zap" ) -func parseAttributes(c *cfg) { +func parseAttributes(c *cfg) error { if c.appCfg.Node.Relay { - return + return nil } - fatalOnErr(attributes.ReadNodeAttributes(&c.cfgNodeInfo.localInfo, c.appCfg.Node.Attributes)) + if err := attributes.ReadNodeAttributes(&c.cfgNodeInfo.localInfo, c.appCfg.Node.Attributes); err != nil { + return err + } // expand UN/LOCODE attribute if any found; keep user's attributes // if any conflicts appear locAttr := c.cfgNodeInfo.localInfo.LOCODE() if locAttr == "" { - return + return nil } record, err := getRecord(locAttr) if err != nil { - fatalOnErr(fmt.Errorf("could not get locode record from DB: %w", err)) + return fmt.Errorf("could not get locode record from DB: %w", err) } countryCode := locAttr[:locodedb.CountryCodeLen] @@ -84,6 +86,8 @@ func parseAttributes(c *cfg) { } else { setIfNotEmpty(n.SetSubdivisionName, record.SubDivName) } + + return nil } func getRecord(lc string) (locodedb.Record, error) { diff --git a/cmd/neofs-node/netmap.go b/cmd/neofs-node/netmap.go index 0a9f5e1cb9..36b98722f0 100644 --- a/cmd/neofs-node/netmap.go +++ b/cmd/neofs-node/netmap.go @@ -141,7 +141,7 @@ func initNetmapService(c *cfg) { network.WriteToNodeInfo(c.localAddr, &c.cfgNodeInfo.localInfo) c.cfgNodeInfo.localInfo.SetPublicKey(c.key.PublicKey().Bytes()) - parseAttributes(c) + fatalOnErr(parseAttributes(c)) c.cfgNodeInfo.localInfo.SetOffline() c.cfgNodeInfo.localInfoLock.Unlock() @@ -478,11 +478,14 @@ func (c *cfg) reloadNodeAttributes() error { c.cfgNodeInfo.localInfo.SetAttributes(nil) err := writeSystemAttributes(c) + if err == nil { + err = parseAttributes(c) + } if err != nil { + c.cfgNodeInfo.localInfo.SetAttributes(oldAttrs) c.cfgNodeInfo.localInfoLock.Unlock() return err } - parseAttributes(c) newAttrs := c.cfgNodeInfo.localInfo.GetAttributes()