coosld/kubo
1
0
Fork 0
mirror of https://github.com/ipfs/kubo.git synced 2026-02-21 10:27:46 +08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

cmds: use Executors

Browse Source 
- some fixes for cmds1.0
- reinsert plugin loading code, pretty print wrapper

TODO:
if plugin loading fails it only calls log.Warning.
returning an error would be better but that would have
to happen after PreRun, which is not possible atm.

License: MIT
Signed-off-by: keks <keks@cryptoscope.co>
This commit is contained in:
keks 2017-12-07 19:33:25 +01:00 committed by Jeromy
parent 84101c8a2a
commit feef5c3415
9 changed files with 144 additions and 281 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

326
cmd/ipfs/main.go
View File

@ -22,7 +22,7 @@ import (
core "github.com/ipfs/go-ipfs/core"
coreCmds "github.com/ipfs/go-ipfs/core/commands"
corehttp "github.com/ipfs/go-ipfs/core/corehttp"
"github.com/ipfs/go-ipfs/plugin/loader"
loader "github.com/ipfs/go-ipfs/plugin/loader"
repo "github.com/ipfs/go-ipfs/repo"
config "github.com/ipfs/go-ipfs/repo/config"
fsrepo "github.com/ipfs/go-ipfs/repo/fsrepo"
@ -31,7 +31,6 @@ import (
manet "gx/ipfs/QmSGL5Uoa6gKHgBBwQG8u1CWKUC8ZnwaZiLgFVTFBR2bxr/go-multiaddr-net"
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
loggables "gx/ipfs/QmSvcDkiRwB8LuMhUtnvhum2C851Mproo75ZDD19jx43tD/go-libp2p-loggables"
"gx/ipfs/QmVD1W3MC8Hk1WZgFQPWWmBECJ3X72BgUYf9eCQ4PGzPps/go-ipfs-cmdkit"
ma "gx/ipfs/QmW8s4zTsUoX1Q6CeYxVKPyqSKbF7H1YDUyTostBtZ8DaG/go-multiaddr"
osh "gx/ipfs/QmXuBJ7DR6k3rmUEKtvVMhwjmXDuJgXXPUt4LQXKBMsU93/go-os-helper"
"gx/ipfs/QmYopJAcV7R9SbxiPBCvqhnt8EusQpWPHewoZakCMt8hps/go-ipfs-cmds"
@ -90,96 +89,64 @@ func mainRet() int {
}
defer stopFunc() // to be executed as late as possible
var invoc cmdInvocation
defer invoc.close()
// this is a local helper to print out help text.
// there's some considerations that this makes easier.
printHelp := func(long bool, w io.Writer) {
helpFunc := cli.ShortHelp
if long {
helpFunc = cli.LongHelp
}
var p []string
if invoc.req != nil {
p = invoc.req.Path
}
helpFunc("ipfs", Root, p, w)
}
// this is a message to tell the user how to get the help text
printMetaHelp := func(w io.Writer) {
cmdPath := strings.Join(invoc.req.Path, " ")
fmt.Fprintf(w, "Use 'ipfs %s --help' for information about this command\n", cmdPath)
}
intrh, ctx := setupInterruptHandler(ctx)
defer intrh.Close()
// Handle `ipfs help'
if len(os.Args) == 2 {
if os.Args[1] == "help" {
printHelp(false, os.Stdout)
return 0
os.Args[1] = "-h"
} else if os.Args[1] == "--version" {
os.Args[1] = "version"
}
}
intrh, ctx := invoc.SetupInterruptHandler(ctx)
defer intrh.Close()
// parse the commandline into a command invocation
parseErr := invoc.Parse(ctx, os.Args[1:])
// BEFORE handling the parse error, if we have enough information
// AND the user requested help, print it out and exit
if invoc.req != nil {
longH, shortH, err := invoc.requestedHelp()
buildEnv := func(ctx context.Context, req *cmds.Request) (interface{}, error) {
repoPath, err := getRepoPath(req)
if err != nil {
printErr(err)
return 1
}
if longH || shortH {
printHelp(longH, os.Stdout)
return 0
return nil, err
}
log.Debugf("config path is %s", repoPath)
// this sets up the function that will initialize the config lazily.
// this sets up the function that will initialize the node
// this is so that we can construct the node lazily.
return &oldcmds.Context{
ConfigRoot: repoPath,
LoadConfig: loadConfig,
ReqLog: &oldcmds.ReqLog{},
ConstructNode: func() (n *core.IpfsNode, err error) {
if req == nil {
return nil, errors.New("constructing node without a request")
}
r, err := fsrepo.Open(repoPath)
if err != nil { // repo is owned by the node
return nil, err
}
// ok everything is good. set it on the invocation (for ownership)
// and return it.
n, err = core.NewNode(ctx, &core.BuildCfg{
// TODO(keks) figure out how Online was set before. I think it was set to
// a value that always is the zero value so we can just drop it, but
// I'll have to check that.
Repo: r,
})
if err != nil {
return nil, err
}
n.SetLocal(true)
return n, nil
},
}, nil
}
// ok now handle parse error (which means cli input was wrong,
// e.g. incorrect number of args, or nonexistent subcommand)
if parseErr != nil {
printErr(parseErr)
// this was a user error, print help.
if invoc.req != nil && invoc.req.Command != nil {
// we need a newline space.
fmt.Fprintf(os.Stderr, "\n")
printHelp(false, os.Stderr)
}
return 1
}
// here we handle the cases where
// - commands with no Run func are invoked directly.
// - the main command is invoked.
if invoc.req == nil || invoc.req.Command == nil || invoc.req.Command.Run == nil {
printHelp(false, os.Stdout)
return 0
}
// ok, finally, run the command invocation.
err = invoc.Run(ctx)
err = cli.Run(ctx, Root, os.Args, os.Stdin, os.Stdout, os.Stderr, buildEnv, makeExecutor)
if err != nil {
if code, ok := err.(exitErr); ok {
return int(code)
}
printErr(err)
// if this error was a client error, print short help too.
if isClientError(err) {
printMetaHelp(os.Stderr)
}
return 1
}
@ -187,10 +154,9 @@ func mainRet() int {
return 0
}
func (i *cmdInvocation) Run(ctx context.Context) error {
func checkDebug(req *cmds.Request) {
// check if user wants to debug. option OR env var.
debug, _ := i.req.Options["debug"].(bool)
debug, _ := req.Options["debug"].(bool)
if debug || os.Getenv("IPFS_LOGGING") == "debug" {
u.Debug = true
logging.SetDebugLogging()
@ -198,193 +164,33 @@ func (i *cmdInvocation) Run(ctx context.Context) error {
if u.GetenvBool("DEBUG") {
u.Debug = true
}
return callCommand(ctx, i.req, Root, i.ctx)
}
func (i *cmdInvocation) constructNodeFunc(ctx context.Context) func() (*core.IpfsNode, error) {
return func() (n *core.IpfsNode, err error) {
if i.req == nil {
return nil, errors.New("constructing node without a request")
}
r, err := fsrepo.Open(i.ctx.ConfigRoot)
if err != nil { // repo is owned by the node
return nil, err
}
// ok everything is good. set it on the invocation (for ownership)
// and return it.
n, err = core.NewNode(ctx, &core.BuildCfg{
Online: i.ctx.Online,
Repo: r,
})
if err != nil {
return nil, err
}
n.SetLocal(true)
i.node = n
return i.node, nil
}
}
func (i *cmdInvocation) close() {
// let's not forget teardown. If a node was initialized, we must close it.
// Note that this means the underlying req.Context().Node variable is exposed.
// this is gross, and should be changed when we extract out the exec Context.
if i.node != nil {
log.Info("Shutting down node...")
i.node.Close()
}
}
func (i *cmdInvocation) Parse(ctx context.Context, args []string) error {
var err error
i.req, err = cli.Parse(args, os.Stdin, Root)
func makeExecutor(req *cmds.Request, env interface{}) (cmds.Executor, error) {
checkDebug(req)
details, err := commandDetails(req.Path, Root)
if err != nil {
return err
return nil, err
}
//TODO remove this
//fmt.Printf("%#v\n", i.req)
// TODO(keks): pass this as arg to cli.Parse()
i.req.Context = ctx
repoPath, err := getRepoPath(i.req)
client, err := commandShouldRunOnDaemon(*details, req, Root, env.(*oldcmds.Context))
if err != nil {
return err
}
log.Debugf("config path is %s", repoPath)
// this sets up the function that will initialize the config lazily.
if i.ctx == nil {
i.ctx = &oldcmds.Context{}
}
i.ctx.ConfigRoot = repoPath
i.ctx.LoadConfig = loadConfig
// this sets up the function that will initialize the node
// this is so that we can construct the node lazily.
i.ctx.ConstructNode = i.constructNodeFunc(ctx)
// if no encoding was specified by user, default to plaintext encoding
// (if command doesn't support plaintext, use JSON instead)
if enc := i.req.Options[cmds.EncLong]; enc == "" {
if i.req.Command.Encoders != nil && i.req.Command.Encoders[cmds.Text] != nil {
i.req.SetOption(cmds.EncLong, cmds.Text)
} else {
i.req.SetOption(cmds.EncLong, cmds.JSON)
}
return nil, err
}
return nil
}
func (i *cmdInvocation) requestedHelp() (short bool, long bool, err error) {
longHelp, _ := i.req.Options["help"].(bool)
shortHelp, _ := i.req.Options["h"].(bool)
return longHelp, shortHelp, nil
}
func callPreCommandHooks(ctx context.Context, details cmdDetails, req *cmds.Request, root *cmds.Command) error {
log.Event(ctx, "callPreCommandHooks", &details)
log.Debug("calling pre-command hooks...")
return nil
}
func callCommand(ctx context.Context, req *cmds.Request, root *cmds.Command, cctx *oldcmds.Context) error {
log.Info(config.EnvDir, " ", cctx.ConfigRoot)
cmd := req.Command
details, err := commandDetails(req.Path, root)
if err != nil {
return err
}
client, err := commandShouldRunOnDaemon(*details, req, root, cctx)
if err != nil {
return err
}
err = callPreCommandHooks(ctx, *details, req, root)
if err != nil {
return err
}
encTypeStr, _ := req.Options[cmds.EncLong].(string)
encType := cmds.EncodingType(encTypeStr)
var (
re cmds.ResponseEmitter
exitCh <-chan int
)
// first if condition checks the command's encoder map, second checks global encoder map (cmd vs. cmds)
if enc, ok := cmd.Encoders[encType]; ok {
re, exitCh = cli.NewResponseEmitter(os.Stdout, os.Stderr, enc, req)
} else if enc, ok := cmds.Encoders[encType]; ok {
re, exitCh = cli.NewResponseEmitter(os.Stdout, os.Stderr, enc, req)
var exctr cmds.Executor
if client != nil && !req.Command.External {
exctr = client.(cmds.Executor)
} else {
return fmt.Errorf("could not find matching encoder for enctype %#v", encType)
}
if cmd.PreRun != nil {
err = cmd.PreRun(req, cctx)
if err != nil {
return err
}
}
if cmd.PostRun != nil && cmd.PostRun[cmds.CLI] != nil {
re = cmd.PostRun[cmds.CLI](req, re)
}
if client != nil && !cmd.External {
log.Debug("executing command via API")
res, err := client.Send(req)
if err != nil {
if isConnRefused(err) {
err = repo.ErrApiNotRunning
}
return wrapContextCanceled(err)
}
go func() {
err := cmds.Copy(re, res)
if err != nil {
err = re.Emit(cmdkit.Error{err.Error(), cmdkit.ErrNormal | cmdkit.ErrFatal})
if err != nil {
log.Error(err)
}
}
}()
} else {
log.Debug("executing command locally")
pluginpath := filepath.Join(cctx.ConfigRoot, "plugins")
pluginpath := filepath.Join(env.(*oldcmds.Context).ConfigRoot, "plugins")
if _, err := loader.LoadPlugins(pluginpath); err != nil {
return err
log.Warning("error loading plugins: ", err)
}
// Okay!!!!! NOW we can call the command.
go func() {
err := root.Call(req, re, cctx)
if err != nil {
re.SetError(err, cmdkit.ErrNormal)
}
}()
exctr = cmds.NewExecutor(req.Root)
}
if returnCode := <-exitCh; returnCode != 0 {
err = exitErr(returnCode)
}
return err
return exctr, nil
}
// commandDetails returns a command's details for the command given by |path|
@ -468,14 +274,6 @@ func commandShouldRunOnDaemon(details cmdDetails, req *cmds.Request, root *cmds.
return nil, nil
}
func isClientError(err error) bool {
if e, ok := err.(*cmdkit.Error); ok {
return e.Code == cmdkit.ErrClient
}
return false
}
func getRepoPath(req *cmds.Request) (string, error) {
repoOpt, found := req.Options["config"].(string)
if found && repoOpt != "" {
@ -496,7 +294,6 @@ func loadConfig(path string) (*config.Config, error) {
// startProfiling begins CPU profiling and returns a `stop` function to be
// executed as late as possible. The stop function captures the memprofile.
func startProfiling() (func(), error) {
// start CPU profiling as early as possible
ofi, err := os.Create(cpuProfile)
if err != nil {
@ -566,8 +363,7 @@ func (ih *IntrHandler) Handle(handler func(count int, ih *IntrHandler), sigs ...
}()
}
func (i *cmdInvocation) SetupInterruptHandler(ctx context.Context) (io.Closer, context.Context) {
func setupInterruptHandler(ctx context.Context) (io.Closer, context.Context) {
intrh := NewIntrHandler()
ctx, cancelFunc := context.WithCancel(ctx)

27
commands/legacy/request.go
View File

@ -20,6 +20,14 @@ type requestWrapper struct {
ctx *oldcmds.Context
}
func (r *requestWrapper) String() string {
return fmt.Sprintf("{%v, %v}", r.req, r.ctx)
}
func (r *requestWrapper) GoString() string {
return fmt.Sprintf("lgc.Request{%#v, %#v}", r.req, r.ctx)
}
// InvocContext retuns the invocation context of the oldcmds.Request.
// It is faked using OldContext().
func (r *requestWrapper) InvocContext() *oldcmds.Context {
@ -36,6 +44,19 @@ func (r *requestWrapper) SetInvocContext(ctx oldcmds.Context) {
func (r *requestWrapper) Command() *oldcmds.Command { return nil }
func (r *requestWrapper) Arguments() []string {
cmdArgs := r.req.Command.Arguments
reqArgs := r.req.Arguments
// TODO figure out the exaclt policy for when to use these automatically
// TODO once that's done, change the log.Debug below to log.Error
// read arguments from body if we don't have all of them or the command has variadic arguemnts
if len(reqArgs) < len(cmdArgs) ||
len(cmdArgs) > 0 && cmdArgs[len(cmdArgs)-1].Variadic {
err := r.req.ParseBodyArgs()
if err != nil {
log.Debug("error reading arguments from stdin: ", err)
}
}
return r.req.Arguments
}
@ -54,7 +75,11 @@ func (r *requestWrapper) Files() files.File {
func (r *requestWrapper) Option(name string) *cmdkit.OptionValue {
var option cmdkit.Option
for _, def := range r.req.Command.Options {
optDefs, err := r.req.Root.GetOptions(r.req.Path)
if err != nil {
return &cmdkit.OptionValue{nil, false, nil}
}
for _, def := range optDefs {
for _, optName := range def.Names() {
if name == optName {
option = def

29
commands/request.go
View File

@ -9,6 +9,7 @@ import (
"os"
"reflect"
"strconv"
"strings"
"time"
"github.com/ipfs/go-ipfs/core"
@ -17,6 +18,7 @@ import (
"gx/ipfs/QmVD1W3MC8Hk1WZgFQPWWmBECJ3X72BgUYf9eCQ4PGzPps/go-ipfs-cmdkit"
"gx/ipfs/QmVD1W3MC8Hk1WZgFQPWWmBECJ3X72BgUYf9eCQ4PGzPps/go-ipfs-cmdkit/files"
"gx/ipfs/QmYopJAcV7R9SbxiPBCvqhnt8EusQpWPHewoZakCMt8hps/go-ipfs-cmds"
)
type Context struct {
@ -74,6 +76,33 @@ func (c *Context) RootContext() context.Context {
return n.Context()
}
func (c *Context) LogRequest(req *cmds.Request) func() {
rle := &ReqLogEntry{
StartTime: time.Now(),
Active: true,
Command: strings.Join(req.Path, "/"),
Options: req.Options,
Args: req.Arguments,
ID: c.ReqLog.nextID,
log: c.ReqLog,
}
c.ReqLog.AddEntry(rle)
return func() {
c.ReqLog.Finish(rle)
}
}
func (c *Context) Close() {
// let's not forget teardown. If a node was initialized, we must close it.
// Note that this means the underlying req.Context().Node variable is exposed.
// this is gross, and should be changed when we extract out the exec Context.
if c.node != nil {
log.Info("Shutting down node...")
c.node.Close()
}
}
// Request represents a call to a command from a consumer
type Request interface {
Path() []string

2
core/commands/add.go
View File

@ -429,7 +429,7 @@ You can now check what blocks have been created by:
bar.ShowTimeLeft = true
}
case <-req.Context.Done():
re.SetError(req.Context.Err(), cmdkit.ErrNormal)
//re.SetError(req.Context.Err(), cmdkit.ErrNormal)
return
}
}

3
core/commands/dag/dag.go
View File

@ -13,12 +13,15 @@ import (
path "github.com/ipfs/go-ipfs/path"
pin "github.com/ipfs/go-ipfs/pin"
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
cmdkit "gx/ipfs/QmVD1W3MC8Hk1WZgFQPWWmBECJ3X72BgUYf9eCQ4PGzPps/go-ipfs-cmdkit"
files "gx/ipfs/QmVD1W3MC8Hk1WZgFQPWWmBECJ3X72BgUYf9eCQ4PGzPps/go-ipfs-cmdkit/files"
mh "gx/ipfs/QmYeKnKpubCMRiq3PGZcTREErthbb5Q9cXsCoSkD9bjEBd/go-multihash"
cid "gx/ipfs/QmeSrf6pzut73u6zLQkRFQ3ygt3k6XFT2kjdYP8Tnkwwyg/go-cid"
)
var log = logging.Logger("cmds/files")
var DagCmd = &cmds.Command{
Helptext: cmdkit.HelpText{
Tagline: "Interact with ipld dag objects.",

20
core/commands/files/files.go
View File

@ -67,6 +67,12 @@ var hashOption = cmdkit.StringOption("hash", "Hash function to use. Will set Cid
var formatError = errors.New("Format was set by multiple options. Only one format option is allowed")
const defaultStatFormat = `<hash>
Size: <size>
CumulativeSize: <cumulsize>
ChildBlocks: <childs>
Type: <type>`
var FilesStatCmd = &cmds.Command{
Helptext: cmdkit.HelpText{
Tagline: "Display file status.",
@ -77,12 +83,7 @@ var FilesStatCmd = &cmds.Command{
},
Options: []cmdkit.Option{
cmdkit.StringOption("format", "Print statistics in given format. Allowed tokens: "+
"<hash> <size> <cumulsize> <type> <childs>. Conflicts with other format options.").WithDefault(
`<hash>
Size: <size>
CumulativeSize: <cumulsize>
ChildBlocks: <childs>
Type: <type>`),
"<hash> <size> <cumulsize> <type> <childs>. Conflicts with other format options.").WithDefault(defaultStatFormat),
cmdkit.BoolOption("hash", "Print only hash. Implies '--format=<hash>'. Conflicts with other format options."),
cmdkit.BoolOption("size", "Print only size. Implies '--format=<cumulsize>'. Conflicts with other format options."),
},
@ -154,9 +155,9 @@ func statGetFormatOptions(req cmds.Request) (string, error) {
hash, _, _ := req.Option("hash").Bool()
size, _, _ := req.Option("size").Bool()
format, found, _ := req.Option("format").String()
format, _, _ := req.Option("format").String()
if moreThanOne(hash, size, found) {
if moreThanOne(hash, size, format != defaultStatFormat) {
return "", formatError
}
@ -235,6 +236,7 @@ var FilesCpCmd = &cmds.Command{
}
flush, _, _ := req.Option("flush").Bool()
fmt.Println("flush:", flush)
src, err := checkPath(req.Arguments()[0])
if err != nil {
@ -636,7 +638,7 @@ stat' on the file or any of its ancestors.
hashOption,
},
Run: func(req cmds.Request, res cmds.Response) {
path, err := checkPath(req.Arguments()[0])
path, err := checkPath(req.StringArguments()[0])
if err != nil {
res.SetError(err, cmdkit.ErrNormal)
return

6
core/commands/get.go
View File

@ -257,13 +257,13 @@ func (gw *getWriter) writeExtracted(r io.Reader, fpath string) error {
func getCompressOptions(req *cmds.Request) (int, error) {
cmprs, _ := req.Options["compress"].(bool)
cmplvl, cmplvlFound := req.Options["compression-level"].(int)
cmplvl, _ := req.Options["compression-level"].(int)
switch {
case !cmprs:
return gzip.NoCompression, nil
case cmprs && !cmplvlFound:
case cmprs && cmplvl == -1:
return gzip.DefaultCompression, nil
case cmprs && cmplvlFound && (cmplvl < 1 || cmplvl > 9):
case cmprs && (cmplvl < 1 || cmplvl > 9):
return gzip.NoCompression, ErrInvalidCompressionLevel
}
return cmplvl, nil

7
core/commands/object/patch.go
View File

@ -13,9 +13,12 @@ import (
path "github.com/ipfs/go-ipfs/path"
ft "github.com/ipfs/go-ipfs/unixfs"
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
cmdkit "gx/ipfs/QmVD1W3MC8Hk1WZgFQPWWmBECJ3X72BgUYf9eCQ4PGzPps/go-ipfs-cmdkit"
)
var log = logging.Logger("core/commands/object")
var ObjectPatchCmd = &cmds.Command{
Helptext: cmdkit.HelpText{
Tagline: "Create a new merkledag object based on an existing one.",
@ -74,7 +77,7 @@ the limit will not be respected by the network.
return
}
root, err := path.ParsePath(req.Arguments()[0])
root, err := path.ParsePath(req.StringArguments()[0])
if err != nil {
res.SetError(err, cmdkit.ErrNormal)
return
@ -142,7 +145,7 @@ Example:
return
}
rp, err := path.ParsePath(req.Arguments()[0])
rp, err := path.ParsePath(req.StringArguments()[0])
if err != nil {
res.SetError(err, cmdkit.ErrNormal)
return

5
repo/config/init.go
View File

@ -25,6 +25,11 @@ func Init(out io.Writer, nBitsForKeypair int) (*Config, error) {
datastore := DefaultDatastoreConfig()
conf := &Config{
API: API{
HTTPHeaders: map[string][]string{
"Server": {"go-ipfs/" + CurrentVersionNumber},
},
},
// setup the node's default addresses.
// NOTE: two swarm listen addrs, one tcp, one utp.