From 664c296fee441b0e8a4976ff0c024f1345c37f3f 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 be4b3f33a5..beb5b10976 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ Changelog for NeoFS Node ### Added ### Fixed +- gRPC SIGHUP reload no longer stops a running public API before replacement configuration is verified (#4113) ### Changed diff --git a/cmd/neofs-node/config.go b/cmd/neofs-node/config.go index cd52ddd0b4..05dc293d22 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) + } } } @@ -641,7 +644,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 { @@ -693,7 +695,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 10b4dfeeb5..fb58c0ea52 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. @@ -211,23 +231,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) @@ -238,14 +261,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], @@ -254,37 +286,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) } @@ -295,6 +368,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 f7e7b312288c9de3da9a031b29ff0a7d800d3c3f 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. Notify systemd that the node is ready after both successful and failed reload attempts, preventing a unit from remaining in the reloading state after an error. Signed-off-by: Andrey Butusov --- CHANGELOG.md | 1 + cmd/neofs-node/config.go | 100 ++++++++++++++++++++------------------- 2 files changed, 53 insertions(+), 48 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index beb5b10976..a55a994430 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ Changelog for NeoFS Node ### Fixed - gRPC SIGHUP reload no longer stops a running public API before replacement configuration is verified (#4113) +- Storage node validates configuration and completes systemd reload notification on SIGHUP (#4113) ### Changed diff --git a/cmd/neofs-node/config.go b/cmd/neofs-node/config.go index 05dc293d22..12bbe4c32b 100644 --- a/cmd/neofs-node/config.go +++ b/cmd/neofs-node/config.go @@ -629,7 +629,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) @@ -642,73 +641,78 @@ 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.log.Error("configuration reload", zap.Error(err)) + } else { + c.log.Info("configuration has been reloaded successfully") + } - c.appCfg, err = config.New(config.WithConfigFile(c.appCfg.Path())) - if err != nil { - c.log.Error("configuration reading", zap.Error(err)) - continue + 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 2b1cb443ff3a1b322ae442a8cf0e13a731fe2f6f 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 a55a994430..501bf57048 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ Changelog for NeoFS Node ### Fixed - gRPC SIGHUP reload no longer stops a running public API before replacement configuration is verified (#4113) - Storage node validates configuration and completes systemd reload notification on SIGHUP (#4113) +- Invalid node attributes on SIGHUP no longer terminate a storage node (#4113) ### Changed 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()