@cryptotaxi247 / kubo / commits / ca4f48678

refactor: simplify logic for MFS pinning (#10506)

Andrew Gillis committed Sep 27, 2024 at 06:37 UTC ca4f486781a6ddc3c3aa98ae78314d06b11a1bb0
4 files changed +141 -129
cmd/ipfs/kubo/daemon.go
+1 -1
@@ -586,7 +586,7 @@ take effect.
586 prometheus.MustRegister(&corehttp.IpfsNodeCollector{Node: node})
587
588 // start MFS pinning thread
589 - startPinMFS(daemonConfigPollInterval, cctx, &ipfsPinMFSNode{node})
589 + startPinMFS(cctx, daemonConfigPollInterval, &ipfsPinMFSNode{node})
590
591 // The daemon is *finally* ready.
592 fmt.Printf("Daemon is ready\n")
cmd/ipfs/kubo/pinmfs.go
+65 -102
@@ -12,7 +12,7 @@ import (
12 pinclient "github.com/ipfs/boxo/pinning/remote/client"
13 cid "github.com/ipfs/go-cid"
14 ipld "github.com/ipfs/go-ipld-format"
15 - logging "github.com/ipfs/go-log"
15 + logging "github.com/ipfs/go-log/v2"
16
17 config "github.com/ipfs/kubo/config"
18 "github.com/ipfs/kubo/core"
@@ -40,6 +40,7 @@ func init() {
40 d, err := time.ParseDuration(pollDurStr)
41 if err != nil {
42 mfslog.Error("error parsing MFS_PIN_POLL_INTERVAL, using default:", err)
43 + return
44 }
45 daemonConfigPollInterval = d
46 }
@@ -74,56 +75,28 @@ func (x *ipfsPinMFSNode) PeerHost() host.Host {
75 return x.node.PeerHost
76 }
77
77 -func startPinMFS(configPollInterval time.Duration, cctx pinMFSContext, node pinMFSNode) {
78 - errCh := make(chan error)
79 - go pinMFSOnChange(configPollInterval, cctx, node, errCh)
80 - go func() {
81 - for {
82 - select {
83 - case err, isOpen := <-errCh:
84 - if !isOpen {
85 - return
86 - }
87 - mfslog.Errorf("%v", err)
88 - case <-cctx.Context().Done():
89 - return
90 - }
91 - }
92 - }()
78 +func startPinMFS(cctx pinMFSContext, configPollInterval time.Duration, node pinMFSNode) {
79 + go pinMFSOnChange(cctx, configPollInterval, node)
80 }
81
95 -func pinMFSOnChange(configPollInterval time.Duration, cctx pinMFSContext, node pinMFSNode, errCh chan<- error) {
96 - defer close(errCh)
97 -
98 - var tmo *time.Timer
99 - defer func() {
100 - if tmo != nil {
101 - tmo.Stop()
102 - }
103 - }()
82 +func pinMFSOnChange(cctx pinMFSContext, configPollInterval time.Duration, node pinMFSNode) {
83 + tmo := time.NewTimer(configPollInterval)
84 + defer tmo.Stop()
85
86 lastPins := map[string]lastPin{}
87 for {
88 // polling sleep
108 - if tmo == nil {
109 - tmo = time.NewTimer(configPollInterval)
110 - } else {
111 - tmo.Reset(configPollInterval)
112 - }
89 select {
90 case <-cctx.Context().Done():
91 return
92 case <-tmo.C:
93 + tmo.Reset(configPollInterval)
94 }
95
96 // reread the config, which may have changed in the meantime
97 cfg, err := cctx.GetConfig()
98 if err != nil {
122 - select {
123 - case errCh <- fmt.Errorf("pinning reading config (%v)", err):
124 - case <-cctx.Context().Done():
125 - return
126 - }
99 + mfslog.Errorf("pinning reading config (%v)", err)
100 continue
101 }
102 mfslog.Debugf("pinning loop is awake, %d remote services", len(cfg.Pinning.RemoteServices))
@@ -131,30 +104,29 @@ func pinMFSOnChange(configPollInterval time.Duration, cctx pinMFSContext, node p
104 // get the most recent MFS root cid
105 rootNode, err := node.RootNode()
106 if err != nil {
134 - select {
135 - case errCh <- fmt.Errorf("pinning reading MFS root (%v)", err):
136 - case <-cctx.Context().Done():
137 - return
138 - }
107 + mfslog.Errorf("pinning reading MFS root (%v)", err)
108 continue
109 }
141 - rootCid := rootNode.Cid()
110
111 // pin to all remote services in parallel
144 - pinAllMFS(cctx.Context(), node, cfg, rootCid, lastPins, errCh)
112 + pinAllMFS(cctx.Context(), node, cfg, rootNode.Cid(), lastPins)
113 }
114 }
115
116 // pinAllMFS pins on all remote services in parallel to overcome DoS attacks.
149 -func pinAllMFS(ctx context.Context, node pinMFSNode, cfg *config.Config, rootCid cid.Cid, lastPins map[string]lastPin, errCh chan<- error) {
150 - ch := make(chan lastPin, len(cfg.Pinning.RemoteServices))
151 - for svcName_, svcConfig_ := range cfg.Pinning.RemoteServices {
117 +func pinAllMFS(ctx context.Context, node pinMFSNode, cfg *config.Config, rootCid cid.Cid, lastPins map[string]lastPin) {
118 + ch := make(chan lastPin)
119 + var started int
120 +
121 + for svcName, svcConfig := range cfg.Pinning.RemoteServices {
122 + if ctx.Err() != nil {
123 + break
124 + }
125 +
126 // skip services where MFS is not enabled
153 - svcName, svcConfig := svcName_, svcConfig_
127 mfslog.Debugf("pinning MFS root considering service %q", svcName)
128 if !svcConfig.Policies.MFS.Enable {
129 mfslog.Debugf("pinning service %q is not enabled", svcName)
157 - ch <- lastPin{}
130 continue
131 }
132 // read mfs pin interval for this service
@@ -165,11 +137,7 @@ func pinAllMFS(ctx context.Context, node pinMFSNode, cfg *config.Config, rootCid
137 var err error
138 repinInterval, err = time.ParseDuration(svcConfig.Policies.MFS.RepinInterval)
139 if err != nil {
168 - select {
169 - case errCh <- fmt.Errorf("remote pinning service %q has invalid MFS.RepinInterval (%v)", svcName, err):
170 - case <-ctx.Done():
171 - }
172 - ch <- lastPin{}
140 + mfslog.Errorf("remote pinning service %q has invalid MFS.RepinInterval (%v)", svcName, err)
141 continue
142 }
143 }
@@ -182,38 +150,30 @@ func pinAllMFS(ctx context.Context, node pinMFSNode, cfg *config.Config, rootCid
150 } else {
151 mfslog.Debugf("pinning MFS root to %q: skipped due to MFS.RepinInterval=%s (remaining: %s)", svcName, repinInterval.String(), (repinInterval - time.Since(last.Time)).String())
152 }
185 - ch <- lastPin{}
153 continue
154 }
155 }
156
157 mfslog.Debugf("pinning MFS root %q to %q", rootCid, svcName)
191 - go func() {
192 - if r, err := pinMFS(ctx, node, rootCid, svcName, svcConfig); err != nil {
193 - select {
194 - case errCh <- fmt.Errorf("pinning MFS root %q to %q (%v)", rootCid, svcName, err):
195 - case <-ctx.Done():
196 - }
197 - ch <- lastPin{}
198 - } else {
199 - ch <- r
158 + go func(svcName string, svcConfig config.RemotePinningService) {
159 + r, err := pinMFS(ctx, node, rootCid, svcName, svcConfig)
160 + if err != nil {
161 + mfslog.Errorf("pinning MFS root %q to %q (%v)", rootCid, svcName, err)
162 }
201 - }()
163 + ch <- r
164 + }(svcName, svcConfig)
165 + started++
166 }
203 - for i := 0; i < len(cfg.Pinning.RemoteServices); i++ {
167 +
168 + // Collect results from all started goroutines.
169 + for i := 0; i < started; i++ {
170 if x := <-ch; x.IsValid() {
171 lastPins[x.ServiceName] = x
172 }
173 }
174 }
175
210 -func pinMFS(
211 - ctx context.Context,
212 - node pinMFSNode,
213 - cid cid.Cid,
214 - svcName string,
215 - svcConfig config.RemotePinningService,
216 -) (lastPin, error) {
176 +func pinMFS(ctx context.Context, node pinMFSNode, cid cid.Cid, svcName string, svcConfig config.RemotePinningService) (lastPin, error) {
177 c := pinclient.NewClient(svcConfig.API.Endpoint, svcConfig.API.Key)
178
179 pinName := svcConfig.Policies.MFS.PinName
@@ -243,43 +203,46 @@ func pinMFS(
203 }
204 for range lsPinCh { // in case the prior loop exits early
205 }
246 - if err := <-lsErrCh; err != nil {
206 + err := <-lsErrCh
207 + if err != nil {
208 return lastPin{}, fmt.Errorf("error while listing remote pins: %v", err)
209 }
210
250 - // CID of the current MFS root is already being pinned, nothing to do
251 - if pinning {
252 - mfslog.Debugf("pinning MFS to %q: pin for %q exists since %s, skipping", svcName, cid, pinTime.String())
253 - return lastPin{Time: pinTime, ServiceName: svcName, ServiceConfig: svcConfig, CID: cid}, nil
254 - }
255 -
256 - // Prepare Pin.name
257 - addOpts := []pinclient.AddOption{pinclient.PinOpts.WithName(pinName)}
211 + if !pinning {
212 + // Prepare Pin.name
213 + addOpts := []pinclient.AddOption{pinclient.PinOpts.WithName(pinName)}
214
259 - // Prepare Pin.origins
260 - // Add own multiaddrs to the 'origins' array, so Pinning Service can
261 - // use that as a hint and connect back to us (if possible)
262 - if node.PeerHost() != nil {
263 - addrs, err := peer.AddrInfoToP2pAddrs(host.InfoFromHost(node.PeerHost()))
264 - if err != nil {
265 - return lastPin{}, err
215 + // Prepare Pin.origins
216 + // Add own multiaddrs to the 'origins' array, so Pinning Service can
217 + // use that as a hint and connect back to us (if possible)
218 + if node.PeerHost() != nil {
219 + addrs, err := peer.AddrInfoToP2pAddrs(host.InfoFromHost(node.PeerHost()))
220 + if err != nil {
221 + return lastPin{}, err
222 + }
223 + addOpts = append(addOpts, pinclient.PinOpts.WithOrigins(addrs...))
224 }
267 - addOpts = append(addOpts, pinclient.PinOpts.WithOrigins(addrs...))
268 - }
225
270 - // Create or replace pin for MFS root
271 - if existingRequestID != "" {
272 - mfslog.Debugf("pinning to %q: replacing existing MFS root pin with %q", svcName, cid)
273 - _, err := c.Replace(ctx, existingRequestID, cid, addOpts...)
274 - if err != nil {
275 - return lastPin{}, err
226 + // Create or replace pin for MFS root
227 + if existingRequestID != "" {
228 + mfslog.Debugf("pinning to %q: replacing existing MFS root pin with %q", svcName, cid)
229 + if _, err = c.Replace(ctx, existingRequestID, cid, addOpts...); err != nil {
230 + return lastPin{}, err
231 + }
232 + } else {
233 + mfslog.Debugf("pinning to %q: creating a new MFS root pin for %q", svcName, cid)
234 + if _, err = c.Add(ctx, cid, addOpts...); err != nil {
235 + return lastPin{}, err
236 + }
237 }
238 } else {
278 - mfslog.Debugf("pinning to %q: creating a new MFS root pin for %q", svcName, cid)
279 - _, err := c.Add(ctx, cid, addOpts...)
280 - if err != nil {
281 - return lastPin{}, err
282 - }
239 + mfslog.Debugf("pinning MFS to %q: pin for %q exists since %s, skipping", svcName, cid, pinTime.String())
240 }
284 - return lastPin{Time: pinTime, ServiceName: svcName, ServiceConfig: svcConfig, CID: cid}, nil
241 +
242 + return lastPin{
243 + Time: pinTime,
244 + ServiceName: svcName,
245 + ServiceConfig: svcConfig,
246 + CID: cid,
247 + }, nil
248 }
cmd/ipfs/kubo/pinmfs_test.go
+66 -24
@@ -1,14 +1,19 @@
1 package kubo
2
3 import (
4 + "bufio"
5 "context"
6 + "encoding/json"
7 + "errors"
8 "fmt"
9 + "io"
10 "strings"
11 "testing"
12 "time"
13
14 merkledag "github.com/ipfs/boxo/ipld/merkledag"
15 ipld "github.com/ipfs/go-ipld-format"
16 + logging "github.com/ipfs/go-log/v2"
17 config "github.com/ipfs/kubo/config"
18 "github.com/libp2p/go-libp2p/core/host"
19 peer "github.com/libp2p/go-libp2p/core/peer"
@@ -60,25 +65,37 @@ func isErrorSimilar(e1, e2 error) bool {
65 }
66
67 func TestPinMFSConfigError(t *testing.T) {
63 - ctx := &testPinMFSContext{
64 - ctx: context.Background(),
68 + ctx, cancel := context.WithTimeout(context.Background(), 2*testConfigPollInterval)
69 + defer cancel()
70 +
71 + cctx := &testPinMFSContext{
72 + ctx: ctx,
73 cfg: nil,
74 err: fmt.Errorf("couldn't read config"),
75 }
76 node := &testPinMFSNode{}
69 - errCh := make(chan error)
70 - go pinMFSOnChange(testConfigPollInterval, ctx, node, errCh)
71 - if !isErrorSimilar(<-errCh, ctx.err) {
72 - t.Errorf("error did not propagate")
77 +
78 + logReader := logging.NewPipeReader()
79 + go func() {
80 + pinMFSOnChange(cctx, testConfigPollInterval, node)
81 + logReader.Close()
82 + }()
83 +
84 + level, msg := readLogLine(t, logReader)
85 + if level != "error" {
86 + t.Error("expected error to be logged")
87 }
74 - if !isErrorSimilar(<-errCh, ctx.err) {
88 + if !isErrorSimilar(errors.New(msg), cctx.err) {
89 t.Errorf("error did not propagate")
90 }
91 }
92
93 func TestPinMFSRootNodeError(t *testing.T) {
80 - ctx := &testPinMFSContext{
81 - ctx: context.Background(),
94 + ctx, cancel := context.WithTimeout(context.Background(), 2*testConfigPollInterval)
95 + defer cancel()
96 +
97 + cctx := &testPinMFSContext{
98 + ctx: ctx,
99 cfg: &config.Config{
100 Pinning: config.Pinning{},
101 },
@@ -87,12 +104,16 @@ func TestPinMFSRootNodeError(t *testing.T) {
104 node := &testPinMFSNode{
105 err: fmt.Errorf("cannot create root node"),
106 }
90 - errCh := make(chan error)
91 - go pinMFSOnChange(testConfigPollInterval, ctx, node, errCh)
92 - if !isErrorSimilar(<-errCh, node.err) {
93 - t.Errorf("error did not propagate")
107 + logReader := logging.NewPipeReader()
108 + go func() {
109 + pinMFSOnChange(cctx, testConfigPollInterval, node)
110 + logReader.Close()
111 + }()
112 + level, msg := readLogLine(t, logReader)
113 + if level != "error" {
114 + t.Error("expected error to be logged")
115 }
95 - if !isErrorSimilar(<-errCh, node.err) {
116 + if !isErrorSimilar(errors.New(msg), node.err) {
117 t.Errorf("error did not propagate")
118 }
119 }
@@ -155,7 +176,8 @@ func TestPinMFSService(t *testing.T) {
176 }
177
178 func testPinMFSServiceWithError(t *testing.T, cfg *config.Config, expectedErrorPrefix string) {
158 - goctx, cancel := context.WithCancel(context.Background())
179 + goctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
180 + defer cancel()
181 ctx := &testPinMFSContext{
182 ctx: goctx,
183 cfg: cfg,
@@ -164,16 +186,36 @@ func testPinMFSServiceWithError(t *testing.T, cfg *config.Config, expectedErrorP
186 node := &testPinMFSNode{
187 err: nil,
188 }
167 - errCh := make(chan error)
168 - go pinMFSOnChange(testConfigPollInterval, ctx, node, errCh)
169 - defer cancel()
170 - // first pass through the pinning loop
171 - err := <-errCh
172 - if !strings.Contains((err).Error(), expectedErrorPrefix) {
173 - t.Errorf("expecting error containing %q", expectedErrorPrefix)
189 + logReader := logging.NewPipeReader()
190 + go func() {
191 + pinMFSOnChange(ctx, testConfigPollInterval, node)
192 + logReader.Close()
193 + }()
194 + level, msg := readLogLine(t, logReader)
195 + if level != "error" {
196 + t.Error("expected error to be logged")
197 }
175 - // second pass through the pinning loop
176 - if !strings.Contains((err).Error(), expectedErrorPrefix) {
198 + if !strings.Contains(msg, expectedErrorPrefix) {
199 t.Errorf("expecting error containing %q", expectedErrorPrefix)
200 }
201 }
202 +
203 +func readLogLine(t *testing.T, logReader io.Reader) (string, string) {
204 + t.Helper()
205 +
206 + r := bufio.NewReader(logReader)
207 + data, err := r.ReadBytes('\n')
208 + if err != nil {
209 + t.Fatal(err)
210 + }
211 +
212 + logInfo := struct {
213 + Level string `json:"level"`
214 + Msg string `json:"msg"`
215 + }{}
216 + err = json.Unmarshal(data, &logInfo)
217 + if err != nil {
218 + t.Fatal(err)
219 + }
220 + return logInfo.Level, logInfo.Msg
221 +}
core/commands/pin/remotepin.go
+9 -2
@@ -221,6 +221,8 @@ NOTE: a comma-separated notation is supported in CLI for convenience:
221
222 // Block unless --background=true is passed
223 if !req.Options[pinBackgroundOptionName].(bool) {
224 + const pinWaitTime = 500 * time.Millisecond
225 + var timer *time.Timer
226 requestID := ps.GetRequestId()
227 for {
228 ps, err = c.GetStatusByID(ctx, requestID)
@@ -237,10 +239,15 @@ NOTE: a comma-separated notation is supported in CLI for convenience:
239 if s == pinclient.StatusFailed {
240 return fmt.Errorf("remote service failed to pin requestid=%q", requestID)
241 }
240 - tmr := time.NewTimer(time.Second / 2)
242 + if timer == nil {
243 + timer = time.NewTimer(pinWaitTime)
244 + } else {
245 + timer.Reset(pinWaitTime)
246 + }
247 select {
242 - case <-tmr.C:
248 + case <-timer.C:
249 case <-ctx.Done():
250 + timer.Stop()
251 return fmt.Errorf("waiting for pin interrupted, requestid=%q remains on remote service", requestID)
252 }
253 }