diff --git a/CHANGELOG.md b/CHANGELOG.md index be4b3f33a5..501bf57048 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,9 @@ Changelog for NeoFS Node ### Added ### 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/config.go b/cmd/neofs-node/config.go index cd52ddd0b4..12bbe4c32b 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) + } } } @@ -626,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) @@ -639,74 +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) - oldGRPC := writeGRPCConfig(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, oldGRPC); 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 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 { 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()