[#1639] node: Support active RPC limiting

- Allow configuration of active RPC limits for method groups
- Apply RPC limiting for all services except the control service

Signed-off-by: Aleksey Savchuk <a.savchuk@yadro.com>

cmd/frostfs-node/config.go

@@ -29,6 +29,7 @@ import (
 	nodeconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/node"
 	objectconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/object"
 	replicatorconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/replicator"
+	rpcconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/rpc"
 	tracingconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/tracing"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/metrics"
@@ -69,6 +70,7 @@ import (
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/state"
 	"git.frostfs.info/TrueCloudLab/frostfs-observability/logging/lokicore"
 	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
+	"git.frostfs.info/TrueCloudLab/frostfs-qos/limiting"
 	netmapV2 "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/netmap"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
 	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
@@ -528,6 +530,8 @@ type cfgGRPC struct {
 	maxChunkSize     uint64
 	maxAddrAmount    uint64
 	reconnectTimeout time.Duration
+
+	limiter limiting.Limiter
 }
 
 func (c *cfgGRPC) append(e string, l net.Listener, s *grpc.Server) {
@@ -717,7 +721,7 @@ func initCfg(appCfg *config.Config) *cfg {
 
 	c.cfgNetmap = initNetmap(appCfg, netState, relayOnly)
 
-	c.cfgGRPC = initCfgGRPC()
+	c.cfgGRPC = initCfgGRPC(appCfg)
 
 	c.cfgMorph = cfgMorph{
 		proxyScriptHash: contractsconfig.Proxy(appCfg),
@@ -848,13 +852,22 @@ func initFrostfsID(appCfg *config.Config) cfgFrostfsID {
 	}
 }
 
-func initCfgGRPC() cfgGRPC {
+func initCfgGRPC(appCfg *config.Config) cfgGRPC {
 	maxChunkSize := uint64(maxMsgSize) * 3 / 4  // 25% to meta, 75% to payload
 	maxAddrAmount := maxChunkSize / addressSize // each address is about 72 bytes
 
+	var limits []limiting.KeyLimit
+	for _, l := range rpcconfig.Limits(appCfg) {
+		limits = append(limits, limiting.KeyLimit{Keys: l.Methods, Limit: l.MaxOps})
+	}
+
+	limiter, err := limiting.NewSemaphoreLimiter(limits)
+	fatalOnErr(err)
+
 	return cfgGRPC{
 		maxChunkSize:  maxChunkSize,
 		maxAddrAmount: maxAddrAmount,
+		limiter:       limiter,
 	}
 }
 

cmd/frostfs-node/config/rpc/config.go

@@ -0,0 +1,43 @@
+package rpcconfig
+
+import (
+	"strconv"
+
+	"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config"
+)
+
+const (
+	subsection       = "rpc"
+	limitsSubsection = "limits"
+)
+
+type LimitConfig struct {
+	Methods []string
+	MaxOps  int64
+}
+
+func Limits(c *config.Config) []LimitConfig {
+	c = c.Sub(subsection).Sub(limitsSubsection)
+
+	var limits []LimitConfig
+
+	var i uint64
+	for ; ; i++ {
+		si := strconv.FormatUint(i, 10)
+		sc := c.Sub(si)
+
+		methods := config.StringSliceSafe(sc, "methods")
+		if len(methods) == 0 {
+			break
+		}
+
+		maxOps := config.IntSafe(sc, "max_ops")
+		if maxOps == 0 {
+			panic("no max operations for method group")
+		}
+
+		limits = append(limits, LimitConfig{methods, maxOps})
+	}
+
+	return limits
+}

cmd/frostfs-node/config/rpc/config_test.go

@@ -0,0 +1,53 @@
+package rpcconfig
+
+import (
+	"testing"
+
+	"git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config"
+	configtest "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/test"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRPCSection(t *testing.T) {
+	t.Run("defaults", func(t *testing.T) {
+		require.Empty(t, Limits(configtest.EmptyConfig()))
+	})
+
+	t.Run("correct config", func(t *testing.T) {
+		const path = "../../../../config/example/node"
+
+		fileConfigTest := func(c *config.Config) {
+			limits := Limits(c)
+			require.Len(t, limits, 2)
+
+			limit0 := limits[0]
+			limit1 := limits[1]
+
+			require.ElementsMatch(t, limit0.Methods, []string{"/neo.fs.v2.object.ObjectService/PutSingle", "/neo.fs.v2.object.ObjectService/Put"})
+			require.Equal(t, limit0.MaxOps, int64(1000))
+
+			require.ElementsMatch(t, limit1.Methods, []string{"/neo.fs.v2.object.ObjectService/Get"})
+			require.Equal(t, limit1.MaxOps, int64(10000))
+		}
+
+		configtest.ForEachFileType(path, fileConfigTest)
+
+		t.Run("ENV", func(t *testing.T) {
+			configtest.ForEnvFileType(t, path, fileConfigTest)
+		})
+	})
+
+	t.Run("no max operations", func(t *testing.T) {
+		const path = "testdata/node"
+
+		fileConfigTest := func(c *config.Config) {
+			require.Panics(t, func() { _ = Limits(c) })
+		}
+
+		configtest.ForEachFileType(path, fileConfigTest)
+
+		t.Run("ENV", func(t *testing.T) {
+			configtest.ForEnvFileType(t, path, fileConfigTest)
+		})
+	})
+}

cmd/frostfs-node/config/rpc/testdata/node.env

@@ -0,0 +1,3 @@
+FROSTFS_RPC_LIMITS_0_METHODS="/neo.fs.v2.object.ObjectService/PutSingle /neo.fs.v2.object.ObjectService/Put"
+FROSTFS_RPC_LIMITS_1_METHODS="/neo.fs.v2.object.ObjectService/Get"
+FROSTFS_RPC_LIMITS_1_MAX_OPS=10000

cmd/frostfs-node/config/rpc/testdata/node.json

@@ -0,0 +1,18 @@
+{
+  "rpc": {
+    "limits": [
+      {
+        "methods": [
+          "/neo.fs.v2.object.ObjectService/PutSingle",
+          "/neo.fs.v2.object.ObjectService/Put"
+        ]
+      },
+      {
+        "methods": [
+          "/neo.fs.v2.object.ObjectService/Get"
+        ],
+        "max_ops": 10000
+      }
+    ]
+  }
+}

cmd/frostfs-node/config/rpc/testdata/node.yaml

@@ -0,0 +1,8 @@
+rpc:
+  limits:
+    - methods:
+      - /neo.fs.v2.object.ObjectService/PutSingle
+      - /neo.fs.v2.object.ObjectService/Put
+    - methods:
+      - /neo.fs.v2.object.ObjectService/Get
+      max_ops: 10000

cmd/frostfs-node/grpc.go

@@ -9,6 +9,7 @@ import (
 
 	grpcconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/grpc"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
+	qosInternal "git.frostfs.info/TrueCloudLab/frostfs-node/internal/qos"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
 	metrics "git.frostfs.info/TrueCloudLab/frostfs-observability/metrics/grpc"
 	tracing "git.frostfs.info/TrueCloudLab/frostfs-observability/tracing/grpc"
@@ -134,11 +135,13 @@ func getGrpcServerOpts(ctx context.Context, c *cfg, sc *grpcconfig.Config) ([]gr
 			qos.NewUnaryServerInterceptor(),
 			metrics.NewUnaryServerInterceptor(),
 			tracing.NewUnaryServerInterceptor(),
+			qosInternal.NewMaxActiveRPCLimiterUnaryServerInterceptor(c.cfgGRPC.limiter),
 		),
 		grpc.ChainStreamInterceptor(
 			qos.NewStreamServerInterceptor(),
 			metrics.NewStreamServerInterceptor(),
 			tracing.NewStreamServerInterceptor(),
+			qosInternal.NewMaxActiveRPCLimiterStreamServerInterceptor(c.cfgGRPC.limiter),
 		),
 	}
 

config/example/node.env

@@ -91,6 +91,11 @@ FROSTFS_OBJECT_PUT_SKIP_SESSION_TOKEN_ISSUER_VERIFICATION=true
 FROSTFS_OBJECT_DELETE_TOMBSTONE_LIFETIME=10
 FROSTFS_OBJECT_GET_PRIORITY="$attribute:ClusterName $attribute:UN-LOCODE"
 
+FROSTFS_RPC_LIMITS_0_METHODS="/neo.fs.v2.object.ObjectService/PutSingle /neo.fs.v2.object.ObjectService/Put"
+FROSTFS_RPC_LIMITS_0_MAX_OPS=1000
+FROSTFS_RPC_LIMITS_1_METHODS="/neo.fs.v2.object.ObjectService/Get"
+FROSTFS_RPC_LIMITS_1_MAX_OPS=10000
+
 # Storage engine section
 FROSTFS_STORAGE_SHARD_POOL_SIZE=15
 FROSTFS_STORAGE_SHARD_RO_ERROR_THRESHOLD=100

config/example/node.json

@@ -140,6 +140,23 @@
       "priority": ["$attribute:ClusterName", "$attribute:UN-LOCODE"]
     }
   },
+  "rpc": {
+    "limits": [
+      {
+        "methods": [
+          "/neo.fs.v2.object.ObjectService/PutSingle",
+          "/neo.fs.v2.object.ObjectService/Put"
+        ],
+        "max_ops": 1000
+      },
+      {
+        "methods": [
+          "/neo.fs.v2.object.ObjectService/Get"
+        ],
+        "max_ops": 10000
+      }
+    ]
+  },
   "storage": {
     "shard_pool_size": 15,
     "shard_ro_error_threshold": 100,

config/example/node.yaml

@@ -123,6 +123,16 @@ object:
       - $attribute:ClusterName
       - $attribute:UN-LOCODE
 
+rpc:
+  limits:
+    - methods:
+      - /neo.fs.v2.object.ObjectService/PutSingle
+      - /neo.fs.v2.object.ObjectService/Put
+      max_ops: 1000
+    - methods:
+      - /neo.fs.v2.object.ObjectService/Get
+      max_ops: 10000
+
 storage:
   # note: shard configuration can be omitted for relay node (see `node.relay`)
   shard_pool_size: 15 # size of per-shard worker pools used for PUT operations

docs/storage-node-configuration.md

@@ -416,6 +416,27 @@ object:
 | `delete.tombstone_lifetime` | `int`      | `5`           | Tombstone lifetime for removed objects in epochs.                                              |
 | `get.priority`              | `[]string` |               | List of metrics of nodes for prioritization. Used for computing response on GET requests.      |
 
+
+# `rpc` section
+Contains limits on the number of active RPC for specified method(s).
+
+```yaml
+rpc:
+  limits:
+    - methods:
+      - /neo.fs.v2.object.ObjectService/PutSingle
+      - /neo.fs.v2.object.ObjectService/Put
+      max_ops: 1000
+    - methods:
+      - /neo.fs.v2.object.ObjectService/Get
+      max_ops: 10000
+```
+
+| Parameter        | Type       | Default value | Description                                                  |
+|------------------|------------|---------------|--------------------------------------------------------------|
+| `limits.max_ops` | `int`      |               | Maximum number of active RPC allowed for the given method(s) |
+| `limits.methods` | `[]string` |               | List of RPC methods sharing the given limit                  |
+
 # `runtime` section
 Contains runtime parameters.