master
go 500 lines 16.5 KB
Raw
1 package libp2p
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "os"
9 "path/filepath"
10
11 "github.com/ipfs/kubo/config"
12 "github.com/ipfs/kubo/core/node/helpers"
13 "github.com/ipfs/kubo/core/shutdown"
14 "github.com/ipfs/kubo/repo"
15
16 logging "github.com/ipfs/go-log/v2"
17 "github.com/libp2p/go-libp2p"
18 "github.com/libp2p/go-libp2p/core/network"
19 "github.com/libp2p/go-libp2p/core/peer"
20 "github.com/libp2p/go-libp2p/core/protocol"
21 rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
22 "github.com/multiformats/go-multiaddr"
23 "go.uber.org/fx"
24 )
25
26 var rcmgrLogger = logging.Logger("rcmgr")
27
28 const NetLimitTraceFilename = "rcmgr.json.gz"
29
30 var ErrNoResourceMgr = errors.New("missing ResourceMgr: make sure the daemon is running with Swarm.ResourceMgr.Enabled")
31
32 func ResourceManager(repoPath string, cfg config.SwarmConfig, userResourceOverrides rcmgr.PartialLimitConfig) any {
33 return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, repo repo.Repo) (network.ResourceManager, Libp2pOpts, error) {
34 var manager network.ResourceManager
35 var opts Libp2pOpts
36
37 enabled := cfg.ResourceMgr.Enabled.WithDefault(true)
38
39 // ENV overrides Config (if present)
40 switch os.Getenv("LIBP2P_RCMGR") {
41 case "0", "false":
42 enabled = false
43 case "1", "true":
44 enabled = true
45 }
46
47 if enabled {
48 log.Debug("libp2p resource manager is enabled")
49
50 limitConfig, msg, err := LimitConfig(cfg, userResourceOverrides)
51 if err != nil {
52 return nil, opts, fmt.Errorf("creating final Resource Manager config: %w", err)
53 }
54
55 if !isPartialConfigEmpty(userResourceOverrides) {
56 rcmgrLogger.Info(`
57 libp2p-resource-limit-overrides.json has been loaded, "default" fields will be
58 filled in with autocomputed defaults.`)
59 }
60
61 // We want to see this message on startup, that's why we are using fmt instead of log.
62 rcmgrLogger.Info(msg)
63
64 if err := ensureConnMgrMakeSenseVsResourceMgr(limitConfig, cfg); err != nil {
65 return nil, opts, err
66 }
67
68 str, err := rcmgr.NewStatsTraceReporter()
69 if err != nil {
70 return nil, opts, err
71 }
72
73 ropts := []rcmgr.Option{
74 rcmgr.WithTraceReporter(str),
75 rcmgr.WithLimitPerSubnet(
76 nil,
77 []rcmgr.ConnLimitPerSubnet{
78 {
79 ConnCount: 16,
80 PrefixLength: 56,
81 },
82 {
83 ConnCount: 8 * 16,
84 PrefixLength: 48,
85 },
86 }),
87 }
88
89 if len(cfg.ResourceMgr.Allowlist) > 0 {
90 var mas []multiaddr.Multiaddr
91 for _, maStr := range cfg.ResourceMgr.Allowlist {
92 ma, err := multiaddr.NewMultiaddr(maStr)
93 if err != nil {
94 log.Errorf("failed to parse multiaddr=%v for allowlist, skipping. err=%v", maStr, err)
95 continue
96 }
97 mas = append(mas, ma)
98 }
99 ropts = append(ropts, rcmgr.WithAllowlistedMultiaddrs(mas))
100 log.Infof("Setting allowlist to: %v", mas)
101 }
102
103 if os.Getenv("LIBP2P_DEBUG_RCMGR") != "" {
104 traceFilePath := filepath.Join(repoPath, NetLimitTraceFilename)
105 ropts = append(ropts, rcmgr.WithTrace(traceFilePath))
106 }
107
108 limiter := rcmgr.NewFixedLimiter(limitConfig)
109
110 manager, err = rcmgr.NewResourceManager(limiter, ropts...)
111 if err != nil {
112 return nil, opts, fmt.Errorf("creating libp2p resource manager: %w", err)
113 }
114 lrm := &loggingResourceManager{
115 logger: &logging.Logger("resourcemanager").SugaredLogger,
116 delegate: manager,
117 }
118 lrm.start(helpers.LifecycleCtx(mctx, lc))
119 manager = lrm
120 } else {
121 rcmgrLogger.Info("go-libp2p resource manager protection disabled")
122 manager = &network.NullResourceManager{}
123 }
124
125 opts.Opts = append(opts.Opts, libp2p.ResourceManager(manager))
126
127 lc.Append(fx.Hook{
128 OnStop: func(ctx context.Context) error {
129 return shutdown.CloseWithCtx(ctx, "resource-manager", manager.Close)
130 },
131 })
132
133 return manager, opts, nil
134 }
135 }
136
137 func isPartialConfigEmpty(cfg rcmgr.PartialLimitConfig) bool {
138 var emptyResourceConfig rcmgr.ResourceLimits
139 if cfg.System != emptyResourceConfig ||
140 cfg.Transient != emptyResourceConfig ||
141 cfg.AllowlistedSystem != emptyResourceConfig ||
142 cfg.AllowlistedTransient != emptyResourceConfig ||
143 cfg.ServiceDefault != emptyResourceConfig ||
144 cfg.ServicePeerDefault != emptyResourceConfig ||
145 cfg.ProtocolDefault != emptyResourceConfig ||
146 cfg.ProtocolPeerDefault != emptyResourceConfig ||
147 cfg.PeerDefault != emptyResourceConfig ||
148 cfg.Conn != emptyResourceConfig ||
149 cfg.Stream != emptyResourceConfig {
150 return false
151 }
152 for _, v := range cfg.Service {
153 if v != emptyResourceConfig {
154 return false
155 }
156 }
157 for _, v := range cfg.ServicePeer {
158 if v != emptyResourceConfig {
159 return false
160 }
161 }
162 for _, v := range cfg.Protocol {
163 if v != emptyResourceConfig {
164 return false
165 }
166 }
167 for _, v := range cfg.ProtocolPeer {
168 if v != emptyResourceConfig {
169 return false
170 }
171 }
172 for _, v := range cfg.Peer {
173 if v != emptyResourceConfig {
174 return false
175 }
176 }
177 return true
178 }
179
180 // LimitConfig returns the union of the Computed Default Limits and the User Supplied Override Limits.
181 func LimitConfig(cfg config.SwarmConfig, userResourceOverrides rcmgr.PartialLimitConfig) (limitConfig rcmgr.ConcreteLimitConfig, logMessageForStartup string, err error) {
182 limitConfig, msg, err := createDefaultLimitConfig(cfg)
183 if err != nil {
184 return rcmgr.ConcreteLimitConfig{}, msg, err
185 }
186
187 // The logic for defaults and overriding with specified userResourceOverrides
188 // is documented in docs/libp2p-resource-management.md.
189 // Any changes here should be reflected there.
190
191 // This effectively overrides the computed default LimitConfig with any non-"useDefault" values from the userResourceOverrides file.
192 // Because of how how Build works, any rcmgr.Default value in userResourceOverrides
193 // will be overridden with a computed default value.
194 limitConfig = userResourceOverrides.Build(limitConfig)
195
196 return limitConfig, msg, nil
197 }
198
199 type ResourceLimitsAndUsage struct {
200 // This is duplicated from rcmgr.ResourceResourceLimits but adding *Usage fields.
201 Memory rcmgr.LimitVal64
202 MemoryUsage int64
203 FD rcmgr.LimitVal
204 FDUsage int
205 Conns rcmgr.LimitVal
206 ConnsUsage int
207 ConnsInbound rcmgr.LimitVal
208 ConnsInboundUsage int
209 ConnsOutbound rcmgr.LimitVal
210 ConnsOutboundUsage int
211 Streams rcmgr.LimitVal
212 StreamsUsage int
213 StreamsInbound rcmgr.LimitVal
214 StreamsInboundUsage int
215 StreamsOutbound rcmgr.LimitVal
216 StreamsOutboundUsage int
217 }
218
219 func (u ResourceLimitsAndUsage) ToResourceLimits() rcmgr.ResourceLimits {
220 return rcmgr.ResourceLimits{
221 Memory: u.Memory,
222 FD: u.FD,
223 Conns: u.Conns,
224 ConnsInbound: u.ConnsInbound,
225 ConnsOutbound: u.ConnsOutbound,
226 Streams: u.Streams,
227 StreamsInbound: u.StreamsInbound,
228 StreamsOutbound: u.StreamsOutbound,
229 }
230 }
231
232 type LimitsConfigAndUsage struct {
233 // This is duplicated from rcmgr.ResourceManagerStat but using ResourceLimitsAndUsage
234 // instead of network.ScopeStat.
235 System ResourceLimitsAndUsage
236 Transient ResourceLimitsAndUsage
237 Services map[string]ResourceLimitsAndUsage `json:",omitempty"`
238 Protocols map[protocol.ID]ResourceLimitsAndUsage `json:",omitempty"`
239 Peers map[peer.ID]ResourceLimitsAndUsage `json:",omitempty"`
240 }
241
242 func (u LimitsConfigAndUsage) MarshalJSON() ([]byte, error) {
243 // we want to marshal the encoded peer id
244 encodedPeerMap := make(map[string]ResourceLimitsAndUsage, len(u.Peers))
245 for p, v := range u.Peers {
246 encodedPeerMap[p.String()] = v
247 }
248
249 type Alias LimitsConfigAndUsage
250 return json.Marshal(&struct {
251 *Alias
252 Peers map[string]ResourceLimitsAndUsage `json:",omitempty"`
253 }{
254 Alias: (*Alias)(&u),
255 Peers: encodedPeerMap,
256 })
257 }
258
259 func (u LimitsConfigAndUsage) ToPartialLimitConfig() (result rcmgr.PartialLimitConfig) {
260 result.System = u.System.ToResourceLimits()
261 result.Transient = u.Transient.ToResourceLimits()
262
263 result.Service = make(map[string]rcmgr.ResourceLimits, len(u.Services))
264 for s, l := range u.Services {
265 result.Service[s] = l.ToResourceLimits()
266 }
267 result.Protocol = make(map[protocol.ID]rcmgr.ResourceLimits, len(u.Protocols))
268 for p, l := range u.Protocols {
269 result.Protocol[p] = l.ToResourceLimits()
270 }
271 result.Peer = make(map[peer.ID]rcmgr.ResourceLimits, len(u.Peers))
272 for p, l := range u.Peers {
273 result.Peer[p] = l.ToResourceLimits()
274 }
275
276 return
277 }
278
279 func MergeLimitsAndStatsIntoLimitsConfigAndUsage(l rcmgr.ConcreteLimitConfig, stats rcmgr.ResourceManagerStat) LimitsConfigAndUsage {
280 limits := l.ToPartialLimitConfig()
281
282 return LimitsConfigAndUsage{
283 System: mergeResourceLimitsAndScopeStatToResourceLimitsAndUsage(limits.System, stats.System),
284 Transient: mergeResourceLimitsAndScopeStatToResourceLimitsAndUsage(limits.Transient, stats.Transient),
285 Services: mergeLimitsAndStatsMapIntoLimitsConfigAndUsageMap(limits.Service, stats.Services),
286 Protocols: mergeLimitsAndStatsMapIntoLimitsConfigAndUsageMap(limits.Protocol, stats.Protocols),
287 Peers: mergeLimitsAndStatsMapIntoLimitsConfigAndUsageMap(limits.Peer, stats.Peers),
288 }
289 }
290
291 func mergeLimitsAndStatsMapIntoLimitsConfigAndUsageMap[K comparable](limits map[K]rcmgr.ResourceLimits, stats map[K]network.ScopeStat) map[K]ResourceLimitsAndUsage {
292 r := make(map[K]ResourceLimitsAndUsage, maxInt(len(limits), len(stats)))
293 for p, s := range stats {
294 var l rcmgr.ResourceLimits
295 if limits != nil {
296 if rl, ok := limits[p]; ok {
297 l = rl
298 }
299 }
300 r[p] = mergeResourceLimitsAndScopeStatToResourceLimitsAndUsage(l, s)
301 }
302 for p, s := range limits {
303 if _, ok := stats[p]; ok {
304 continue // we already processed this element in the loop above
305 }
306
307 r[p] = mergeResourceLimitsAndScopeStatToResourceLimitsAndUsage(s, network.ScopeStat{})
308 }
309 return r
310 }
311
312 func maxInt(x, y int) int {
313 if x > y {
314 return x
315 }
316 return y
317 }
318
319 func mergeResourceLimitsAndScopeStatToResourceLimitsAndUsage(rl rcmgr.ResourceLimits, ss network.ScopeStat) ResourceLimitsAndUsage {
320 return ResourceLimitsAndUsage{
321 Memory: rl.Memory,
322 MemoryUsage: ss.Memory,
323 FD: rl.FD,
324 FDUsage: ss.NumFD,
325 Conns: rl.Conns,
326 ConnsUsage: ss.NumConnsOutbound + ss.NumConnsInbound,
327 ConnsOutbound: rl.ConnsOutbound,
328 ConnsOutboundUsage: ss.NumConnsOutbound,
329 ConnsInbound: rl.ConnsInbound,
330 ConnsInboundUsage: ss.NumConnsInbound,
331 Streams: rl.Streams,
332 StreamsUsage: ss.NumStreamsOutbound + ss.NumStreamsInbound,
333 StreamsOutbound: rl.StreamsOutbound,
334 StreamsOutboundUsage: ss.NumStreamsOutbound,
335 StreamsInbound: rl.StreamsInbound,
336 StreamsInboundUsage: ss.NumStreamsInbound,
337 }
338 }
339
340 type ResourceInfos []ResourceInfo
341
342 type ResourceInfo struct {
343 ScopeName string
344 LimitName string
345 LimitValue rcmgr.LimitVal64
346 CurrentUsage int64
347 }
348
349 // LimitConfigsToInfo gets limits and stats and generates a list of scopes and limits to be printed.
350 func LimitConfigsToInfo(stats LimitsConfigAndUsage) ResourceInfos {
351 result := ResourceInfos{}
352
353 result = append(result, resourceLimitsAndUsageToResourceInfo(config.ResourceMgrSystemScope, stats.System)...)
354 result = append(result, resourceLimitsAndUsageToResourceInfo(config.ResourceMgrTransientScope, stats.Transient)...)
355
356 for i, s := range stats.Services {
357 result = append(result, resourceLimitsAndUsageToResourceInfo(
358 config.ResourceMgrServiceScopePrefix+i,
359 s,
360 )...)
361 }
362
363 for i, p := range stats.Protocols {
364 result = append(result, resourceLimitsAndUsageToResourceInfo(
365 config.ResourceMgrProtocolScopePrefix+string(i),
366 p,
367 )...)
368 }
369
370 for i, p := range stats.Peers {
371 result = append(result, resourceLimitsAndUsageToResourceInfo(
372 config.ResourceMgrPeerScopePrefix+i.String(),
373 p,
374 )...)
375 }
376
377 return result
378 }
379
380 const (
381 limitNameMemory = "Memory"
382 limitNameFD = "FD"
383 limitNameConns = "Conns"
384 limitNameConnsInbound = "ConnsInbound"
385 limitNameConnsOutbound = "ConnsOutbound"
386 limitNameStreams = "Streams"
387 limitNameStreamsInbound = "StreamsInbound"
388 limitNameStreamsOutbound = "StreamsOutbound"
389 )
390
391 var limits = []string{
392 limitNameMemory,
393 limitNameFD,
394 limitNameConns,
395 limitNameConnsInbound,
396 limitNameConnsOutbound,
397 limitNameStreams,
398 limitNameStreamsInbound,
399 limitNameStreamsOutbound,
400 }
401
402 func resourceLimitsAndUsageToResourceInfo(scopeName string, stats ResourceLimitsAndUsage) ResourceInfos {
403 result := ResourceInfos{}
404 for _, l := range limits {
405 ri := ResourceInfo{
406 ScopeName: scopeName,
407 }
408 switch l {
409 case limitNameMemory:
410 ri.LimitName = limitNameMemory
411 ri.LimitValue = stats.Memory
412 ri.CurrentUsage = stats.MemoryUsage
413 case limitNameFD:
414 ri.LimitName = limitNameFD
415 ri.LimitValue = rcmgr.LimitVal64(stats.FD)
416 ri.CurrentUsage = int64(stats.FDUsage)
417 case limitNameConns:
418 ri.LimitName = limitNameConns
419 ri.LimitValue = rcmgr.LimitVal64(stats.Conns)
420 ri.CurrentUsage = int64(stats.ConnsUsage)
421 case limitNameConnsInbound:
422 ri.LimitName = limitNameConnsInbound
423 ri.LimitValue = rcmgr.LimitVal64(stats.ConnsInbound)
424 ri.CurrentUsage = int64(stats.ConnsInboundUsage)
425 case limitNameConnsOutbound:
426 ri.LimitName = limitNameConnsOutbound
427 ri.LimitValue = rcmgr.LimitVal64(stats.ConnsOutbound)
428 ri.CurrentUsage = int64(stats.ConnsOutboundUsage)
429 case limitNameStreams:
430 ri.LimitName = limitNameStreams
431 ri.LimitValue = rcmgr.LimitVal64(stats.Streams)
432 ri.CurrentUsage = int64(stats.StreamsUsage)
433 case limitNameStreamsInbound:
434 ri.LimitName = limitNameStreamsInbound
435 ri.LimitValue = rcmgr.LimitVal64(stats.StreamsInbound)
436 ri.CurrentUsage = int64(stats.StreamsInboundUsage)
437 case limitNameStreamsOutbound:
438 ri.LimitName = limitNameStreamsOutbound
439 ri.LimitValue = rcmgr.LimitVal64(stats.StreamsOutbound)
440 ri.CurrentUsage = int64(stats.StreamsOutboundUsage)
441 }
442
443 if ri.LimitValue == rcmgr.Unlimited64 || ri.LimitValue == rcmgr.DefaultLimit64 {
444 // ignore unlimited and unset limits to remove noise from output.
445 continue
446 }
447
448 result = append(result, ri)
449 }
450
451 return result
452 }
453
454 func ensureConnMgrMakeSenseVsResourceMgr(concreteLimits rcmgr.ConcreteLimitConfig, cfg config.SwarmConfig) error {
455 if cfg.ConnMgr.Type.WithDefault(config.DefaultConnMgrType) == "none" || len(cfg.ResourceMgr.Allowlist) != 0 {
456 // no connmgr OR
457 // If an allowlist is set, a user may be enacting some form of DoS defense.
458 // We don't want want to modify the System.ConnsInbound in that case for example
459 // as it may make sense for it to be (and stay) as "blockAll"
460 // so that only connections within the allowlist of multiaddrs get established.
461 return nil
462 }
463
464 rcm := concreteLimits.ToPartialLimitConfig()
465
466 highWater := cfg.ConnMgr.HighWater.WithDefault(config.DefaultConnMgrHighWater)
467 if (rcm.System.Conns > rcmgr.DefaultLimit || rcm.System.Conns == rcmgr.BlockAllLimit) && int64(rcm.System.Conns) <= highWater {
468 // nolint
469 return fmt.Errorf(`
470 Unable to initialize libp2p due to conflicting resource manager limit configuration.
471 resource manager System.Conns (%d) must be bigger than ConnMgr.HighWater (%d)
472 See: https://github.com/ipfs/kubo/blob/master/docs/libp2p-resource-management.md#how-does-the-resource-manager-resourcemgr-relate-to-the-connection-manager-connmgr
473 `, rcm.System.Conns, highWater)
474 }
475 if (rcm.System.ConnsInbound > rcmgr.DefaultLimit || rcm.System.ConnsInbound == rcmgr.BlockAllLimit) && int64(rcm.System.ConnsInbound) <= highWater {
476 // nolint
477 return fmt.Errorf(`
478 Unable to initialize libp2p due to conflicting resource manager limit configuration.
479 resource manager System.ConnsInbound (%d) must be bigger than ConnMgr.HighWater (%d)
480 See: https://github.com/ipfs/kubo/blob/master/docs/libp2p-resource-management.md#how-does-the-resource-manager-resourcemgr-relate-to-the-connection-manager-connmgr
481 `, rcm.System.ConnsInbound, highWater)
482 }
483 if (rcm.System.Streams > rcmgr.DefaultLimit || rcm.System.Streams == rcmgr.BlockAllLimit) && int64(rcm.System.Streams) <= highWater {
484 // nolint
485 return fmt.Errorf(`
486 Unable to initialize libp2p due to conflicting resource manager limit configuration.
487 resource manager System.Streams (%d) must be bigger than ConnMgr.HighWater (%d)
488 See: https://github.com/ipfs/kubo/blob/master/docs/libp2p-resource-management.md#how-does-the-resource-manager-resourcemgr-relate-to-the-connection-manager-connmgr
489 `, rcm.System.Streams, highWater)
490 }
491 if (rcm.System.StreamsInbound > rcmgr.DefaultLimit || rcm.System.StreamsInbound == rcmgr.BlockAllLimit) && int64(rcm.System.StreamsInbound) <= highWater {
492 // nolint
493 return fmt.Errorf(`
494 Unable to initialize libp2p due to conflicting resource manager limit configuration.
495 resource manager System.StreamsInbound (%d) must be bigger than ConnMgr.HighWater (%d)
496 See: https://github.com/ipfs/kubo/blob/master/docs/libp2p-resource-management.md#how-does-the-resource-manager-resourcemgr-relate-to-the-connection-manager-connmgr
497 `, rcm.System.StreamsInbound, highWater)
498 }
499 return nil
500 }