metrics, pprof: support reloading services with SIGHUP

Reload prometheus and pprof services, if the config is updated.
Make metric to work with multithreading.

Closes #1868.

Signed-off-by: Andrey Butusov <[email protected]>

CHANGELOG.md

@@ -20,6 +20,7 @@ attribute, which is used for container domain name in NNS contracts (#2954)
 - `logger.encoding` config option (#2999)
 - Reloading morph endpoints with SIGHUP (#2998)
 - New `peapod-to-fstree` tool providing peapod-to-fstree data migration (#3013)
+- Reloading pprof/metrics services with SIGHUP (#3016)
 
 ### Fixed
 - Do not search for tombstones when handling their expiration, use local indexes instead (#2929)

cmd/neofs-node/config.go

@@ -46,6 +46,7 @@ import (
 	"github.com/nspcc-dev/neofs-node/pkg/network/cache"
 	"github.com/nspcc-dev/neofs-node/pkg/services/control"
 	controlSvc "github.com/nspcc-dev/neofs-node/pkg/services/control/server"
+	objectService "github.com/nspcc-dev/neofs-node/pkg/services/object"
 	getsvc "github.com/nspcc-dev/neofs-node/pkg/services/object/get"
 	"github.com/nspcc-dev/neofs-node/pkg/services/policer"
 	"github.com/nspcc-dev/neofs-node/pkg/services/replicator"
@@ -299,6 +300,8 @@ type internals struct {
 	healthStatus atomic.Int32
 	// is node under maintenance
 	isMaintenance atomic.Bool
+
+	nodeMetrics *metrics.NodeMetrics
 }
 
 // starts node's maintenance.
@@ -378,7 +381,8 @@ type shared struct {
 
 	treeService *tree.Service
 
-	metricsCollector *metrics.NodeMetrics
+	metricsCollectorLock sync.RWMutex
+	metricsCollector     *metrics.NodeMetrics
 
 	control *controlSvc.Server
 }
@@ -488,6 +492,8 @@ type cfgObject struct {
 	tombstoneLifetime uint64
 
 	containerNodes *containerNodes
+
+	metricsCollectorService *objectService.MetricCollector
 }
 
 type cfgLocalStorage struct {
@@ -637,9 +643,15 @@ func initCfg(appCfg *config.Config) *cfg {
 
 	c.ownerIDFromKey = user.NewFromECDSAPublicKey(key.PrivateKey.PublicKey)
 
+	c.nodeMetrics = metrics.NewNodeMetrics(misc.Version)
 	if metricsconfig.Enabled(c.cfgReader) {
-		c.metricsCollector = metrics.NewNodeMetrics(misc.Version)
+		c.metricsCollectorLock.Lock()
+		c.metricsCollector = c.nodeMetrics
+		c.metricsCollectorLock.Unlock()
+
+		c.basics.networkState.metricsLock.Lock()
 		c.basics.networkState.metrics = c.metricsCollector
+		c.basics.networkState.metricsLock.Unlock()
 	}
 
 	c.veryLastClosers = make(map[string]func())
@@ -848,12 +860,20 @@ func (c *cfg) configWatcher(ctx context.Context) {
 		case <-ch:
 			c.log.Info("SIGHUP has been received, rereading configuration...")
 
+			oldMetrics := writeMetricConfig(c.cfgReader)
+			oldProfiler := writeProfilerConfig(c.cfgReader)
+
 			err := c.readConfig(c.cfgReader)
 			if err != nil {
 				c.log.Error("configuration reading", zap.Error(err))
 				continue
 			}
 
+			// Prometheus and pprof
+
+			// nolint:contextcheck
+			c.reloadMetricsAndPprof(oldMetrics, oldProfiler)
+
 			// Logger
 
 			err = c.internals.logLevel.UnmarshalText([]byte(c.logger.level))
@@ -929,3 +949,47 @@ func writeSystemAttributes(c *cfg) error {
 
 	return nil
 }
+
+func (c *cfg) reloadMetricsAndPprof(oldMetrics metricConfig, oldProfiler profilerConfig) {
+	// Metrics
+
+	if oldMetrics.isUpdated(c.cfgReader) {
+		if closer, ok := c.veryLastClosers[metricName]; ok {
+			closer()
+		}
+
+		if metricsconfig.Enabled(c.cfgReader) != oldMetrics.enabled {
+			var nodeMetrics *metrics.NodeMetrics
+			if metricsconfig.Enabled(c.cfgReader) {
+				nodeMetrics = c.nodeMetrics
+				delete(c.veryLastClosers, metricName)
+				c.cfgObject.metricsCollectorService.ReloadMetrics(nodeMetrics)
+				c.cfgObject.cfgLocalStorage.localStorage.ReloadMetrics(nodeMetrics)
+			} else {
+				c.cfgObject.metricsCollectorService.ReloadMetrics(nil)
+				c.cfgObject.cfgLocalStorage.localStorage.ReloadMetrics(nil)
+			}
+
+			c.metricsCollectorLock.Lock()
+			c.metricsCollector = nodeMetrics
+			c.metricsCollectorLock.Unlock()
+
+			c.basics.networkState.metricsLock.Lock()
+			c.basics.networkState.metrics = c.metricsCollector
+			c.basics.networkState.metricsLock.Unlock()
+		}
+
+		preRunAndLog(c, metricName, initMetrics(c))
+	}
+
+	//Profiler
+
+	if oldProfiler.isUpdated(c.cfgReader) {
+		if closer, ok := c.veryLastClosers[profilerName]; ok {
+			closer()
+		}
+		delete(c.veryLastClosers, profilerName)
+
+		preRunAndLog(c, profilerName, initProfiler(c))
+	}
+}

cmd/neofs-node/control.go

@@ -67,9 +67,11 @@ func (c *cfg) NetmapStatus() control.NetmapStatus {
 func (c *cfg) setHealthStatus(st control.HealthStatus) {
 	c.healthStatus.Store(int32(st))
 
+	c.metricsCollectorLock.Lock()
 	if c.metricsCollector != nil {
 		c.metricsCollector.SetHealth(int32(st))
 	}
+	c.metricsCollectorLock.Unlock()
 }
 
 func (c *cfg) HealthStatus() control.HealthStatus {

cmd/neofs-node/metrics.go

@@ -1,6 +1,9 @@
 package main
 
 import (
+	"time"
+
+	"github.com/nspcc-dev/neofs-node/cmd/neofs-node/config"
 	metricsconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/metrics"
 	httputil "github.com/nspcc-dev/neofs-node/pkg/util/http"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -25,3 +28,23 @@ func initMetrics(c *cfg) *httputil.Server {
 
 	return srv
 }
+
+type metricConfig struct {
+	enabled         bool
+	shutdownTimeout time.Duration
+	address         string
+}
+
+func writeMetricConfig(c *config.Config) metricConfig {
+	return metricConfig{
+		enabled:         metricsconfig.Enabled(c),
+		shutdownTimeout: metricsconfig.ShutdownTimeout(c),
+		address:         metricsconfig.Address(c),
+	}
+}
+
+func (m1 metricConfig) isUpdated(c *config.Config) bool {
+	return m1.enabled != metricsconfig.Enabled(c) ||
+		m1.shutdownTimeout != metricsconfig.ShutdownTimeout(c) ||
+		m1.address != metricsconfig.Address(c)
+}

cmd/neofs-node/netmap.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"sync"
 	"sync/atomic"
 
 	netmapGRPC "github.com/nspcc-dev/neofs-api-go/v2/netmap/grpc"
@@ -29,7 +30,8 @@ type networkState struct {
 
 	nodeInfo atomic.Value // *netmapSDK.NodeInfo
 
-	metrics *metrics.NodeMetrics
+	metricsLock sync.RWMutex
+	metrics     *metrics.NodeMetrics
 }
 
 func newNetworkState() *networkState {
@@ -47,6 +49,9 @@ func (s *networkState) CurrentEpoch() uint64 {
 
 func (s *networkState) setCurrentEpoch(v uint64) {
 	s.epoch.Store(v)
+
+	s.metricsLock.Lock()
+	defer s.metricsLock.Unlock()
 	if s.metrics != nil {
 		s.metrics.SetEpoch(v)
 	}

cmd/neofs-node/object.go

@@ -347,12 +347,15 @@ func initObjectService(c *cfg) {
 		respSvc,
 	)
 
-	var firstSvc objectService.ServiceServer = signSvc
-	if c.metricsCollector != nil {
-		firstSvc = objectService.NewMetricCollector(signSvc, c.metricsCollector)
+	c.metricsCollectorLock.Lock()
+	if c.metricsCollector == nil {
+		c.cfgObject.metricsCollectorService = objectService.NewMetricCollector(signSvc, nil)
+	} else {
+		c.cfgObject.metricsCollectorService = objectService.NewMetricCollector(signSvc, c.metricsCollector)
 	}
+	c.metricsCollectorLock.Unlock()
 
-	server := objectTransportGRPC.New(firstSvc, mNumber, objNode, neofsecdsa.SignerRFC6979(c.shared.basics.key.PrivateKey))
+	server := objectTransportGRPC.New(c.cfgObject.metricsCollectorService, mNumber, objNode, neofsecdsa.SignerRFC6979(c.shared.basics.key.PrivateKey))
 
 	for _, srv := range c.cfgGRPC.servers {
 		objectGRPC.RegisterObjectServiceServer(srv, server)

cmd/neofs-node/pprof.go

@@ -1,6 +1,9 @@
 package main
 
 import (
+	"time"
+
+	"github.com/nspcc-dev/neofs-node/cmd/neofs-node/config"
 	profilerconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/profiler"
 	httputil "github.com/nspcc-dev/neofs-node/pkg/util/http"
 )
@@ -24,3 +27,23 @@ func initProfiler(c *cfg) *httputil.Server {
 
 	return srv
 }
+
+type profilerConfig struct {
+	enabled         bool
+	shutdownTimeout time.Duration
+	address         string
+}
+
+func writeProfilerConfig(c *config.Config) profilerConfig {
+	return profilerConfig{
+		enabled:         profilerconfig.Enabled(c),
+		shutdownTimeout: profilerconfig.ShutdownTimeout(c),
+		address:         profilerconfig.Address(c),
+	}
+}
+
+func (m1 profilerConfig) isUpdated(c *config.Config) bool {
+	return m1.enabled != profilerconfig.Enabled(c) ||
+		m1.shutdownTimeout != profilerconfig.ShutdownTimeout(c) ||
+		m1.address != profilerconfig.Address(c)
+}

cmd/neofs-node/storage.go

@@ -92,9 +92,11 @@ func (c *cfg) engineOpts() []engine.Option {
 		opts = append(opts, engine.WithContainersSource(cntClient.AsContainerSource(c.shared.basics.cCli)))
 	}
 
+	c.metricsCollectorLock.Lock()
 	if c.metricsCollector != nil {
 		opts = append(opts, engine.WithMetrics(c.metricsCollector))
 	}
+	c.metricsCollectorLock.Unlock()
 
 	return opts
 }

pkg/local_object_storage/engine/container.go

@@ -16,8 +16,16 @@ import (
 //
 // Returns an error if executions are blocked (see BlockExecution).
 func (e *StorageEngine) ContainerSize(cnr cid.ID) (uint64, error) {
+	var elapsedFn func()
+
+	e.mtx.Lock()
 	if e.metrics != nil {
-		defer elapsed(e.metrics.AddEstimateContainerSizeDuration)()
+		elapsedFn = elapsed(e.metrics.AddEstimateContainerSizeDuration)
+	}
+	e.mtx.Unlock()
+
+	if elapsedFn != nil {
+		defer elapsedFn()
 	}
 
 	e.blockMtx.RLock()
@@ -50,8 +58,16 @@ func (e *StorageEngine) ContainerSize(cnr cid.ID) (uint64, error) {
 //
 // Returns an error if executions are blocked (see BlockExecution).
 func (e *StorageEngine) ListContainers() ([]cid.ID, error) {
+	var elapsedFn func()
+
+	e.mtx.Lock()
 	if e.metrics != nil {
-		defer elapsed(e.metrics.AddListContainersDuration)()
+		elapsedFn = elapsed(e.metrics.AddListContainersDuration)
+	}
+	e.mtx.Unlock()
+
+	if elapsedFn != nil {
+		defer elapsedFn()
 	}
 
 	e.blockMtx.RLock()

pkg/local_object_storage/engine/control.go

@@ -269,6 +269,13 @@ loop:
 	return nil
 }
 
+func (e *StorageEngine) ReloadMetrics(newMetric MetricRegister) {
+	e.mtx.Lock()
+	defer e.mtx.Unlock()
+
+	e.metrics = newMetric
+}
+
 func calculateShardID(info shard.Info) string {
 	// This calculation should be kept in sync with node
 	// configuration parsing during SIGHUP.

pkg/local_object_storage/engine/delete.go

@@ -11,8 +11,16 @@ import (
 // NOTE: This is a forced removal, marks any object to be deleted (despite
 // any prohibitions on operations with that object).
 func (e *StorageEngine) Delete(addr oid.Address) error {
+	var elapsedFn func()
+
+	e.mtx.Lock()
 	if e.metrics != nil {
-		defer elapsed(e.metrics.AddDeleteDuration)()
+		elapsedFn = elapsed(e.metrics.AddDeleteDuration)
+	}
+	e.mtx.Unlock()
+
+	if elapsedFn != nil {
+		defer elapsedFn()
 	}
 
 	e.blockMtx.RLock()

pkg/local_object_storage/engine/get.go

@@ -22,8 +22,16 @@ import (
 //
 // Returns an error if executions are blocked (see BlockExecution).
 func (e *StorageEngine) Get(addr oid.Address) (*objectSDK.Object, error) {
+	var elapsedFn func()
+
+	e.mtx.Lock()
 	if e.metrics != nil {
-		defer elapsed(e.metrics.AddGetDuration)()
+		elapsedFn = elapsed(e.metrics.AddGetDuration)
+	}
+	e.mtx.Unlock()
+
+	if elapsedFn != nil {
+		defer elapsedFn()
 	}
 
 	e.blockMtx.RLock()

pkg/local_object_storage/engine/head.go

@@ -22,8 +22,16 @@ import (
 //
 // Returns an error if executions are blocked (see BlockExecution).
 func (e *StorageEngine) Head(addr oid.Address, raw bool) (*objectSDK.Object, error) {
+	var elapsedFn func()
+
+	e.mtx.Lock()
 	if e.metrics != nil {
-		defer elapsed(e.metrics.AddHeadDuration)()
+		elapsedFn = elapsed(e.metrics.AddHeadDuration)
+	}
+	e.mtx.Unlock()
+
+	if elapsedFn != nil {
+		defer elapsedFn()
 	}
 
 	e.blockMtx.RLock()

pkg/local_object_storage/engine/inhume.go

@@ -24,8 +24,16 @@ var errInhumeFailure = errors.New("inhume operation failed")
 //
 // Returns an error if executions are blocked (see BlockExecution).
 func (e *StorageEngine) Inhume(tombstone oid.Address, tombExpiration uint64, addrs ...oid.Address) error {
+	var elapsedFn func()
+
+	e.mtx.Lock()
 	if e.metrics != nil {
-		defer elapsed(e.metrics.AddInhumeDuration)()
+		elapsedFn = elapsed(e.metrics.AddInhumeDuration)
+	}
+	e.mtx.Unlock()
+
+	if elapsedFn != nil {
+		defer elapsedFn()
 	}
 
 	e.blockMtx.RLock()