http endpoints: dont print before listen
also splits api, gw and fuse bring up into helper functions
Henry committed
May 24, 2015 at 00:48 UTC
07b3415cdbc4c82cf2204bcb1df28ec65f46cb1e
2 files changed
+187
-71
cmd/ipfs/daemon.go
+173
-64
@@ -3,13 +3,16 @@ package main
3
import (
4
_ "expvar"
5
"fmt"
6
- _ "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/codahale/metrics/runtime"
6
"net/http"
7
_ "net/http/pprof"
8
"os"
9
"strings"
10
+ "sync"
11
12
+ _ "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/codahale/metrics/runtime"
13
ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
14
+ "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
15
+
16
cmds "github.com/ipfs/go-ipfs/commands"
17
"github.com/ipfs/go-ipfs/core"
18
commands "github.com/ipfs/go-ipfs/core/commands"
@@ -192,98 +195,76 @@ func daemonFunc(req cmds.Request, res cmds.Response) {
195
return node, nil
196
}
197
195
- // verify api address is valid multiaddr
196
- apiMaddr, err := ma.NewMultiaddr(cfg.Addresses.API)
198
+ // construct api endpoint - every time
199
+ err, apiErrc := mountHTTPapi(req)
200
if err != nil {
201
res.SetError(err, cmds.ErrNormal)
202
return
203
}
204
202
- var gatewayMaddr ma.Multiaddr
205
+ // construct http gateway - if it is set in the config
206
+ var gwErrc <-chan error
207
if len(cfg.Addresses.Gateway) > 0 {
204
- // ignore error for gateway address
205
- // if there is an error (invalid address), then don't run the gateway
206
- gatewayMaddr, _ = ma.NewMultiaddr(cfg.Addresses.Gateway)
207
- if gatewayMaddr == nil {
208
- log.Errorf("Invalid gateway address: %s", cfg.Addresses.Gateway)
208
+ var err error
209
+ err, gwErrc = mountHTTPgw(req)
210
+ if err != nil {
211
+ res.SetError(err, cmds.ErrNormal)
212
+ return
213
}
214
}
215
212
- // mount if the user provided the --mount flag
216
+ // construct fuse mountpoints - if the user provided the --mount flag
217
mount, _, err := req.Option(mountKwd).Bool()
218
if err != nil {
219
res.SetError(err, cmds.ErrNormal)
220
return
221
}
222
if mount {
219
- fsdir, found, err := req.Option(ipfsMountKwd).String()
220
- if err != nil {
221
- res.SetError(err, cmds.ErrNormal)
222
- return
223
- }
224
- if !found {
225
- fsdir = cfg.Mounts.IPFS
226
- }
227
-
228
- nsdir, found, err := req.Option(ipnsMountKwd).String()
229
- if err != nil {
223
+ if err := mountFuse(req); err != nil {
224
res.SetError(err, cmds.ErrNormal)
225
return
226
}
233
- if !found {
234
- nsdir = cfg.Mounts.IPNS
235
- }
227
+ }
228
237
- err = commands.Mount(node, fsdir, nsdir)
229
+ // collect long-running errors and block for shutdown
230
+ // TODO(cryptix): our fuse currently doesnt follow this pattern for graceful shutdown
231
+ for err := range merge(apiErrc, gwErrc) {
232
if err != nil {
233
res.SetError(err, cmds.ErrNormal)
234
return
235
}
242
- fmt.Printf("IPFS mounted at: %s\n", fsdir)
243
- fmt.Printf("IPNS mounted at: %s\n", nsdir)
236
}
237
+}
238
246
- var rootRedirect corehttp.ServeOption
247
- if len(cfg.Gateway.RootRedirect) > 0 {
248
- rootRedirect = corehttp.RedirectOption("", cfg.Gateway.RootRedirect)
239
+// mountHTTPapi collects options, creates listener, prints status message and starts serving requests
240
+func mountHTTPapi(req cmds.Request) (error, <-chan error) {
241
+ cfg, err := req.Context().GetConfig()
242
+ if err != nil {
243
+ return fmt.Errorf("mountHTTPapi: GetConfig() failed: %s", err), nil
244
}
245
251
- writable, writableOptionFound, err := req.Option(writableKwd).Bool()
246
+ apiMaddr, err := ma.NewMultiaddr(cfg.Addresses.API)
247
if err != nil {
253
- res.SetError(err, cmds.ErrNormal)
254
- return
248
+ return fmt.Errorf("mountHTTPapi: invalid API address: %q (err: %s)", cfg.Addresses.API, err), nil
249
}
256
- if !writableOptionFound {
257
- writable = cfg.Gateway.Writable
250
+
251
+ apiLis, err := manet.Listen(apiMaddr)
252
+ if err != nil {
253
+ return fmt.Errorf("mountHTTPapi: manet.Listen(%s) failed: %s", apiMaddr, err), nil
254
}
255
+ // we might have listened to /tcp/0 - lets see what we are listing on
256
+ apiMaddr = apiLis.Multiaddr()
257
+ fmt.Printf("API server listening on %s\n", apiMaddr)
258
260
- if gatewayMaddr != nil {
261
- go func() {
262
- var opts = []corehttp.ServeOption{
263
- corehttp.VersionOption(),
264
- corehttp.IPNSHostnameOption(),
265
- corehttp.GatewayOption(writable),
266
- }
267
- if rootRedirect != nil {
268
- opts = append(opts, rootRedirect)
269
- }
270
- if writable {
271
- fmt.Printf("Gateway (writable) server listening on %s\n", gatewayMaddr)
272
- } else {
273
- fmt.Printf("Gateway (readonly) server listening on %s\n", gatewayMaddr)
274
- }
275
- err := corehttp.ListenAndServe(node, gatewayMaddr.String(), opts...)
276
- if err != nil {
277
- log.Error(err)
278
- }
279
- }()
259
+ unrestricted, _, err := req.Option(unrestrictedApiAccess).Bool()
260
+ if err != nil {
261
+ return fmt.Errorf("mountHTTPapi: Option(%s) failed: %s", unrestrictedApiAccess, err), nil
262
}
263
282
- gateway := corehttp.NewGateway(corehttp.GatewayConfig{
264
+ apiGw := corehttp.NewGateway(corehttp.GatewayConfig{
265
Writable: true,
266
BlockList: &corehttp.BlockList{
267
Decider: func(s string) bool {
286
- unrestricted, _, _ := req.Option(unrestrictedApiAccess).Bool()
268
if unrestricted {
269
return true
270
}
@@ -300,18 +281,146 @@ func daemonFunc(req cmds.Request, res cmds.Response) {
281
var opts = []corehttp.ServeOption{
282
corehttp.CommandsOption(*req.Context()),
283
corehttp.WebUIOption,
303
- gateway.ServeOption(),
284
+ apiGw.ServeOption(),
285
corehttp.VersionOption(),
286
defaultMux("/debug/vars"),
287
defaultMux("/debug/pprof/"),
288
}
289
309
- if rootRedirect != nil {
310
- opts = append(opts, rootRedirect)
290
+ if len(cfg.Gateway.RootRedirect) > 0 {
291
+ opts = append(opts, corehttp.RedirectOption("", cfg.Gateway.RootRedirect))
292
}
312
- fmt.Printf("API server listening on %s\n", apiMaddr)
313
- if err := corehttp.ListenAndServe(node, apiMaddr.String(), opts...); err != nil {
314
- res.SetError(err, cmds.ErrNormal)
315
- return
293
+
294
+ node, err := req.Context().ConstructNode()
295
+ if err != nil {
296
+ return fmt.Errorf("mountHTTPgw: ConstructNode() failed: %s", err), nil
297
+ }
298
+
299
+ errc := make(chan error)
300
+ go func() {
301
+ errc <- corehttp.Serve(node, apiLis.NetListener(), opts...)
302
+ }()
303
+ return nil, errc
304
+}
305
+
306
+// mountHTTPgw collects options, creates listener, prints status message and starts serving requests
307
+func mountHTTPgw(req cmds.Request) (error, <-chan error) {
308
+ cfg, err := req.Context().GetConfig()
309
+ if err != nil {
310
+ return fmt.Errorf("mountHTTPgw: GetConfig() failed: %s", err), nil
311
+ }
312
+
313
+ gatewayMaddr, err := ma.NewMultiaddr(cfg.Addresses.Gateway)
314
+ if err != nil {
315
+ return fmt.Errorf("mountHTTPgw: invalid gateway address: %q (err: %s)", cfg.Addresses.Gateway, err), nil
316
+ }
317
+
318
+ writable, writableOptionFound, err := req.Option(writableKwd).Bool()
319
+ if err != nil {
320
+ return fmt.Errorf("mountHTTPgw: req.Option(%s) failed: %s", writableKwd, err), nil
321
+ }
322
+ if !writableOptionFound {
323
+ writable = cfg.Gateway.Writable
324
+ }
325
+
326
+ gwLis, err := manet.Listen(gatewayMaddr)
327
+ if err != nil {
328
+ return fmt.Errorf("mountHTTPgw: manet.Listen(%s) failed: %s", gatewayMaddr, err), nil
329
+ }
330
+ // we might have listened to /tcp/0 - lets see what we are listing on
331
+ gatewayMaddr = gwLis.Multiaddr()
332
+
333
+ if writable {
334
+ fmt.Printf("Gateway (writable) server listening on %s\n", gatewayMaddr)
335
+ } else {
336
+ fmt.Printf("Gateway (readonly) server listening on %s\n", gatewayMaddr)
337
+ }
338
+
339
+ var opts = []corehttp.ServeOption{
340
+ corehttp.VersionOption(),
341
+ corehttp.IPNSHostnameOption(),
342
+ corehttp.GatewayOption(writable),
343
}
344
+
345
+ if len(cfg.Gateway.RootRedirect) > 0 {
346
+ opts = append(opts, corehttp.RedirectOption("", cfg.Gateway.RootRedirect))
347
+ }
348
+
349
+ node, err := req.Context().ConstructNode()
350
+ if err != nil {
351
+ return fmt.Errorf("mountHTTPgw: ConstructNode() failed: %s", err), nil
352
+ }
353
+
354
+ errc := make(chan error)
355
+ go func() {
356
+ errc <- corehttp.Serve(node, gwLis.NetListener(), opts...)
357
+ }()
358
+ return nil, errc
359
+}
360
+
361
+//collects options and opens the fuse mountpoint
362
+func mountFuse(req cmds.Request) error {
363
+ cfg, err := req.Context().GetConfig()
364
+ if err != nil {
365
+ return fmt.Errorf("mountFuse: GetConfig() failed: %s", err)
366
+ }
367
+
368
+ fsdir, found, err := req.Option(ipfsMountKwd).String()
369
+ if err != nil {
370
+ return fmt.Errorf("mountFuse: req.Option(%s) failed: %s", ipfsMountKwd, err)
371
+ }
372
+ if !found {
373
+ fsdir = cfg.Mounts.IPFS
374
+ }
375
+
376
+ nsdir, found, err := req.Option(ipnsMountKwd).String()
377
+ if err != nil {
378
+ return fmt.Errorf("mountFuse: req.Option(%s) failed: %s", ipnsMountKwd, err)
379
+ }
380
+ if !found {
381
+ nsdir = cfg.Mounts.IPNS
382
+ }
383
+
384
+ node, err := req.Context().ConstructNode()
385
+ if err != nil {
386
+ return fmt.Errorf("mountFuse: ConstructNode() failed: %s", err)
387
+ }
388
+
389
+ err = commands.Mount(node, fsdir, nsdir)
390
+ if err != nil {
391
+ return err
392
+ }
393
+ fmt.Printf("IPFS mounted at: %s\n", fsdir)
394
+ fmt.Printf("IPNS mounted at: %s\n", nsdir)
395
+ return nil
396
+}
397
+
398
+// merge does fan-in of multiple read-only error channels
399
+// taken from http://blog.golang.org/pipelines
400
+func merge(cs ...<-chan error) <-chan error {
401
+ var wg sync.WaitGroup
402
+ out := make(chan error)
403
+
404
+ // Start an output goroutine for each input channel in cs. output
405
+ // copies values from c to out until c is closed, then calls wg.Done.
406
+ output := func(c <-chan error) {
407
+ for n := range c {
408
+ out <- n
409
+ }
410
+ wg.Done()
411
+ }
412
+ wg.Add(len(cs))
413
+ for _, c := range cs {
414
+ if c != nil {
415
+ go output(c)
416
+ }
417
+ }
418
+
419
+ // Start a goroutine to close out once all the output goroutines are
420
+ // done. This must start after the wg.Add call.
421
+ go func() {
422
+ wg.Wait()
423
+ close(out)
424
+ }()
425
+ return out
426
}
core/corehttp/corehttp.go
+14
-7
@@ -5,6 +5,7 @@ high-level HTTP interfaces to IPFS.
5
package corehttp
6
7
import (
8
+ "fmt"
9
"net"
10
"net/http"
11
"time"
@@ -49,20 +50,26 @@ func ListenAndServe(n *core.IpfsNode, listeningMultiAddr string, options ...Serv
50
if err != nil {
51
return err
52
}
52
- handler, err := makeHandler(n, options...)
53
+
54
+ list, err := manet.Listen(addr)
55
if err != nil {
56
return err
57
}
56
- return listenAndServe(n, addr, handler)
58
+
59
+ // we might have listened to /tcp/0 - lets see what we are listing on
60
+ addr = list.Multiaddr()
61
+ fmt.Printf("API server listening on %s\n", addr)
62
+
63
+ return Serve(n, list.NetListener(), options...)
64
}
65
59
-func listenAndServe(node *core.IpfsNode, addr ma.Multiaddr, handler http.Handler) error {
60
- netarg, host, err := manet.DialArgs(addr)
66
+func Serve(node *core.IpfsNode, lis net.Listener, options ...ServeOption) error {
67
+ handler, err := makeHandler(node, options...)
68
if err != nil {
69
return err
70
}
71
65
- list, err := net.Listen(netarg, host)
72
+ addr, err := manet.FromNetAddr(lis.Addr())
73
if err != nil {
74
return err
75
}
@@ -75,7 +82,7 @@ func listenAndServe(node *core.IpfsNode, addr ma.Multiaddr, handler http.Handler
82
defer node.Children().Done()
83
84
go func() {
78
- serverError = http.Serve(list, handler)
85
+ serverError = http.Serve(lis, handler)
86
close(serverExited)
87
}()
88
@@ -87,7 +94,7 @@ func listenAndServe(node *core.IpfsNode, addr ma.Multiaddr, handler http.Handler
94
case <-node.Closing():
95
log.Infof("server at %s terminating...", addr)
96
90
- list.Close()
97
+ lis.Close()
98
99
outer:
100
for {