Merge pull request #2277 from ipfs/feat/active-cmds

commands diagnostics

commands/http/handler.go

@@ -96,9 +96,16 @@ func NewHandler(ctx cmds.Context, root *cmds.Command, cfg *ServerConfig) http.Ha
 		panic("must provide a valid ServerConfig")
 	}
 
+	// setup request logger
+	ctx.ReqLog = new(cmds.ReqLog)
+
 	// Wrap the internal handler with CORS handling-middleware.
 	// Create a handler for the API.
-	internal := internalHandler{ctx, root, cfg}
+	internal := internalHandler{
+		ctx:  ctx,
+		root: root,
+		cfg:  cfg,
+	}
 	c := cors.New(*cfg.cORSOpts)
 	return &Handler{internal, c.Handler(internal)}
 }
@@ -158,6 +165,9 @@ func (i internalHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	rlog := i.ctx.ReqLog.Add(req)
+	defer rlog.Finish()
+
 	//ps: take note of the name clash - commands.Context != context.Context
 	req.SetInvocContext(i.ctx)
 
@@ -201,8 +211,8 @@ func guessMimeType(res cmds.Response) (string, error) {
 func sendResponse(w http.ResponseWriter, r *http.Request, res cmds.Response, req cmds.Request) {
 	h := w.Header()
 	// Expose our agent to allow identification
-	h.Set("Server", "go-ipfs/" + config.CurrentVersionNumber)
-	
+	h.Set("Server", "go-ipfs/"+config.CurrentVersionNumber)
+
 	mime, err := guessMimeType(res)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)

commands/reqlog.go

@@ -0,0 +1,109 @@
+package commands
+
+import (
+	"strings"
+	"sync"
+	"time"
+)
+
+type ReqLogEntry struct {
+	StartTime time.Time
+	EndTime   time.Time
+	Active    bool
+	Command   string
+	Options   map[string]interface{}
+	Args      []string
+	ID        int
+
+	req Request
+	log *ReqLog
+}
+
+func (r *ReqLogEntry) Finish() {
+	log := r.log
+	log.lock.Lock()
+	defer log.lock.Unlock()
+
+	r.Active = false
+	r.EndTime = time.Now()
+	r.log.maybeCleanup()
+
+	// remove references to save memory
+	r.req = nil
+	r.log = nil
+
+}
+
+func (r *ReqLogEntry) Copy() *ReqLogEntry {
+	out := *r
+	out.log = nil
+	return &out
+}
+
+type ReqLog struct {
+	Requests []*ReqLogEntry
+	nextID   int
+	lock     sync.Mutex
+}
+
+func (rl *ReqLog) Add(req Request) *ReqLogEntry {
+	rl.lock.Lock()
+	defer rl.lock.Unlock()
+
+	rle := &ReqLogEntry{
+		StartTime: time.Now(),
+		Active:    true,
+		Command:   strings.Join(req.Path(), "/"),
+		Options:   req.Options(),
+		Args:      req.Arguments(),
+		ID:        rl.nextID,
+		req:       req,
+		log:       rl,
+	}
+
+	rl.nextID++
+	rl.Requests = append(rl.Requests, rle)
+	return rle
+}
+
+func (rl *ReqLog) maybeCleanup() {
+	// only do it every so often or it might
+	// become a perf issue
+	if len(rl.Requests) == 0 {
+		rl.cleanup()
+	}
+}
+
+func (rl *ReqLog) cleanup() {
+	var i int
+	// drop all logs at are inactive and more than an hour old
+	for ; i < len(rl.Requests); i++ {
+		req := rl.Requests[i]
+		if req.Active || req.EndTime.Add(time.Hour).After(time.Now()) {
+			break
+		}
+	}
+
+	if i > 0 {
+		var j int
+		for i < len(rl.Requests) {
+			rl.Requests[j] = rl.Requests[i]
+			j++
+			i++
+		}
+		rl.Requests = rl.Requests[:len(rl.Requests)-i]
+	}
+}
+
+// Report generates a copy of all the entries in the requestlog
+func (rl *ReqLog) Report() []*ReqLogEntry {
+	rl.lock.Lock()
+	defer rl.lock.Unlock()
+	out := make([]*ReqLogEntry, len(rl.Requests))
+
+	for i, e := range rl.Requests {
+		out[i] = e.Copy()
+	}
+
+	return out
+}

commands/request.go

@@ -21,6 +21,7 @@ type OptMap map[string]interface{}
 type Context struct {
 	Online     bool
 	ConfigRoot string
+	ReqLog     *ReqLog
 
 	config     *config.Config
 	LoadConfig func(path string) (*config.Config, error)

core/commands/active.go

@@ -0,0 +1,82 @@
+package commands
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"sort"
+	"text/tabwriter"
+	"time"
+
+	cmds "github.com/ipfs/go-ipfs/commands"
+)
+
+var ActiveReqsCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "List commands run on this ipfs node",
+		ShortDescription: `
+Lists running and recently run commands.
+`,
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		res.SetOutput(req.InvocContext().ReqLog.Report())
+	},
+	Options: []cmds.Option{
+		cmds.BoolOption("v", "verbose", "print more verbose output"),
+	},
+	Marshalers: map[cmds.EncodingType]cmds.Marshaler{
+		cmds.Text: func(res cmds.Response) (io.Reader, error) {
+			out, ok := res.Output().(*[]*cmds.ReqLogEntry)
+			if !ok {
+				log.Errorf("%#v", res.Output())
+				return nil, cmds.ErrIncorrectType
+			}
+			buf := new(bytes.Buffer)
+
+			verbose, _, _ := res.Request().Option("v").Bool()
+
+			w := tabwriter.NewWriter(buf, 4, 4, 2, ' ', 0)
+			if verbose {
+				fmt.Fprint(w, "ID\t")
+			}
+			fmt.Fprint(w, "Command\t")
+			if verbose {
+				fmt.Fprint(w, "Arguments\tOptions\t")
+			}
+			fmt.Fprintln(w, "Active\tStartTime\tRunTime")
+
+			for _, req := range *out {
+				if verbose {
+					fmt.Fprintf(w, "%d\t", req.ID)
+				}
+				fmt.Fprintf(w, "%s\t", req.Command)
+				if verbose {
+					fmt.Fprintf(w, "%v\t[", req.Args)
+					var keys []string
+					for k, _ := range req.Options {
+						keys = append(keys, k)
+					}
+					sort.Strings(keys)
+
+					for _, k := range keys {
+						fmt.Fprintf(w, "%s=%v,", k, req.Options[k])
+					}
+					fmt.Fprintf(w, "]\t")
+				}
+
+				var live time.Duration
+				if req.Active {
+					live = time.Now().Sub(req.StartTime)
+				} else {
+					live = req.EndTime.Sub(req.StartTime)
+				}
+				t := req.StartTime.Format(time.Stamp)
+				fmt.Fprintf(w, "%t\t%s\t%s\n", req.Active, t, live)
+			}
+			w.Flush()
+
+			return buf, nil
+		},
+	},
+	Type: []*cmds.ReqLogEntry{},
+}

core/commands/diag.go

@@ -45,8 +45,9 @@ var DiagCmd = &cmds.Command{
 	},
 
 	Subcommands: map[string]*cmds.Command{
-		"net": diagNetCmd,
-		"sys": sysDiagCmd,
+		"net":  diagNetCmd,
+		"sys":  sysDiagCmd,
+		"cmds": ActiveReqsCmd,
 	},
 }
 

test/sharness/t0065-active-requests.sh

@@ -0,0 +1,52 @@
+#!/bin/sh
+#
+# Copyright (c) 2016 Jeromy Johnson
+# MIT Licensed; see the LICENSE file in this repository.
+#
+
+test_description="Test active request commands"
+
+. lib/test-lib.sh
+
+test_init_ipfs
+test_launch_ipfs_daemon
+
+test_expect_success "command works" '
+	ipfs diag cmds > cmd_out
+'
+
+test_expect_success "invoc shows up in output" '
+	grep "diag/cmds" cmd_out > /dev/null
+'
+
+test_expect_success "start longer running command" '
+	ipfs log tail &
+	LOGPID=$!
+'
+
+test_expect_success "long running command shows up" '
+	ipfs diag cmds > cmd_out2
+'
+
+test_expect_success "output looks good" '
+	grep "log/tail" cmd_out2 | grep "true" > /dev/null
+'
+
+test_expect_success "kill log cmd" '
+	kill $LOGPID
+	go-sleep 0.5s
+	kill $LOGPID
+
+	wait $LOGPID || true
+'
+
+test_expect_success "long running command inactive" '
+	ipfs diag cmds > cmd_out3
+'
+
+test_expect_success "command shows up as inactive" '
+	grep "log/tail" cmd_out3 | grep "false"
+'
+
+test_kill_ipfs_daemon
+test_done