Merge pull request #343 from stevvooe/tracing-driver

context, storagedriver: trace function calls to Base storage driver

cmd/registry/main.go

@@ -16,7 +16,7 @@ import (
 	"github.com/Sirupsen/logrus/formatters/logstash"
 	"github.com/bugsnag/bugsnag-go"
 	"github.com/docker/distribution/configuration"
-	ctxu "github.com/docker/distribution/context"
+	"github.com/docker/distribution/context"
 	_ "github.com/docker/distribution/health"
 	_ "github.com/docker/distribution/registry/auth/silly"
 	_ "github.com/docker/distribution/registry/auth/token"
@@ -29,7 +29,6 @@ import (
 	"github.com/docker/distribution/version"
 	gorhandlers "github.com/gorilla/handlers"
 	"github.com/yvasiyarov/gorelic"
-	"golang.org/x/net/context"
 )
 
 var showVersion bool
@@ -68,9 +67,9 @@ func main() {
 	}
 
 	if config.HTTP.TLS.Certificate == "" {
-		ctxu.GetLogger(app).Infof("listening on %v", config.HTTP.Addr)
+		context.GetLogger(app).Infof("listening on %v", config.HTTP.Addr)
 		if err := http.ListenAndServe(config.HTTP.Addr, handler); err != nil {
-			ctxu.GetLogger(app).Fatalln(err)
+			context.GetLogger(app).Fatalln(err)
 		}
 	} else {
 		tlsConf := &tls.Config{
@@ -83,23 +82,23 @@ func main() {
 			for _, ca := range config.HTTP.TLS.ClientCAs {
 				caPem, err := ioutil.ReadFile(ca)
 				if err != nil {
-					ctxu.GetLogger(app).Fatalln(err)
+					context.GetLogger(app).Fatalln(err)
 				}
 
 				if ok := pool.AppendCertsFromPEM(caPem); !ok {
-					ctxu.GetLogger(app).Fatalln(fmt.Errorf("Could not add CA to pool"))
+					context.GetLogger(app).Fatalln(fmt.Errorf("Could not add CA to pool"))
 				}
 			}
 
 			for _, subj := range pool.Subjects() {
-				ctxu.GetLogger(app).Debugf("CA Subject: %s", string(subj))
+				context.GetLogger(app).Debugf("CA Subject: %s", string(subj))
 			}
 
 			tlsConf.ClientAuth = tls.RequireAndVerifyClientCert
 			tlsConf.ClientCAs = pool
 		}
 
-		ctxu.GetLogger(app).Infof("listening on %v, tls", config.HTTP.Addr)
+		context.GetLogger(app).Infof("listening on %v, tls", config.HTTP.Addr)
 		server := &http.Server{
 			Addr:      config.HTTP.Addr,
 			Handler:   handler,
@@ -107,7 +106,7 @@ func main() {
 		}
 
 		if err := server.ListenAndServeTLS(config.HTTP.TLS.Certificate, config.HTTP.TLS.Key); err != nil {
-			ctxu.GetLogger(app).Fatalln(err)
+			context.GetLogger(app).Fatalln(err)
 		}
 	}
 }
@@ -191,7 +190,7 @@ func configureLogging(ctx context.Context, config *configuration.Configuration)
 	if config.Log.Level == "" && config.Log.Formatter == "" {
 		// If no config for logging is set, fallback to deprecated "Loglevel".
 		log.SetLevel(logLevel(config.Loglevel))
-		ctx = ctxu.WithLogger(ctx, ctxu.GetLogger(ctx, "version"))
+		ctx = context.WithLogger(ctx, context.GetLogger(ctx, "version"))
 		return ctx, nil
 	}
 
@@ -236,8 +235,8 @@ func configureLogging(ctx context.Context, config *configuration.Configuration)
 			fields = append(fields, k)
 		}
 
-		ctx = ctxu.WithValues(ctx, config.Log.Fields)
+		ctx = context.WithValues(ctx, config.Log.Fields)
-		ctx = ctxu.WithLogger(ctx, ctxu.GetLogger(ctx, fields...))
+		ctx = context.WithLogger(ctx, context.GetLogger(ctx, fields...))
 	}
 
 	return ctx, nil

context/context.go

@@ -1,6 +1,7 @@
 package context
 
 import (
+	"code.google.com/p/go-uuid/uuid"
 	"golang.org/x/net/context"
 )
 
@@ -9,9 +10,31 @@ type Context interface {
 	context.Context
 }
 
-// Background returns a non-nil, empty Context.
+// instanceContext is a context that provides only an instance id. It is
+// provided as the main background context.
+type instanceContext struct {
+	Context
+	id string // id of context, logged as "instance.id"
+}
+
+func (ic *instanceContext) Value(key interface{}) interface{} {
+	if key == "instance.id" {
+		return ic.id
+	}
+
+	return ic.Context.Value(key)
+}
+
+var background = &instanceContext{
+	Context: context.Background(),
+	id:      uuid.New(),
+}
+
+// Background returns a non-nil, empty Context. The background context
+// provides a single key, "instance.id" that is globally unique to the
+// process.
 func Background() Context {
-	return context.Background()
+	return background
 }
 
 // WithValue returns a copy of parent in which the value associated with key is

context/trace.go

@@ -0,0 +1,99 @@
+package context
+
+import (
+	"runtime"
+	"time"
+
+	"code.google.com/p/go-uuid/uuid"
+)
+
+// WithTrace allocates a traced timing span in a new context. This allows a
+// caller to track the time between calling WithTrace and the returned done
+// function. When the done function is called, a log message is emitted with a
+// "trace.duration" field, corresponding to the elapased time and a
+// "trace.func" field, corresponding to the function that called WithTrace.
+//
+// The logging keys "trace.id" and "trace.parent.id" are provided to implement
+// dapper-like tracing. This function should be complemented with a WithSpan
+// method that could be used for tracing distributed RPC calls.
+//
+// The main benefit of this function is to post-process log messages or
+// intercept them in a hook to provide timing data. Trace ids and parent ids
+// can also be linked to provide call tracing, if so required.
+//
+// Here is an example of the usage:
+//
+// 	func timedOperation(ctx Context) {
+// 		ctx, done := WithTrace(ctx)
+// 		defer done("this will be the log message")
+// 		// ... function body ...
+// 	}
+//
+// If the function ran for roughly 1s, such a usage would emit a log message
+// as follows:
+//
+// 	INFO[0001] this will be the log message  trace.duration=1.004575763s trace.func=github.com/docker/distribution/context.traceOperation trace.id=<id> ...
+//
+// Notice that the function name is automatically resolved, along with the
+// package and a trace id is emitted that can be linked with parent ids.
+func WithTrace(ctx Context) (Context, func(format string, a ...interface{})) {
+	if ctx == nil {
+		ctx = Background()
+	}
+
+	pc, file, line, _ := runtime.Caller(1)
+	f := runtime.FuncForPC(pc)
+	ctx = &traced{
+		Context: ctx,
+		id:      uuid.New(),
+		start:   time.Now(),
+		parent:  GetStringValue(ctx, "trace.id"),
+		fnname:  f.Name(),
+		file:    file,
+		line:    line,
+	}
+
+	return ctx, func(format string, a ...interface{}) {
+		GetLogger(ctx, "trace.duration", "trace.id", "trace.parent.id",
+			"trace.func", "trace.file", "trace.line").
+			Infof(format, a...) // info may be too chatty.
+	}
+}
+
+// traced represents a context that is traced for function call timing. It
+// also provides fast lookup for the various attributes that are available on
+// the trace.
+type traced struct {
+	Context
+	id     string
+	parent string
+	start  time.Time
+	fnname string
+	file   string
+	line   int
+}
+
+func (ts *traced) Value(key interface{}) interface{} {
+	switch key {
+	case "trace.start":
+		return ts.start
+	case "trace.duration":
+		return time.Since(ts.start)
+	case "trace.id":
+		return ts.id
+	case "trace.parent.id":
+		if ts.parent == "" {
+			return nil // must return nil to signal no parent.
+		}
+
+		return ts.parent
+	case "trace.func":
+		return ts.fnname
+	case "trace.file":
+		return ts.file
+	case "trace.line":
+		return ts.line
+	}
+
+	return ts.Context.Value(key)
+}

context/trace_test.go

@@ -0,0 +1,85 @@
+package context
+
+import (
+	"runtime"
+	"testing"
+	"time"
+)
+
+// TestWithTrace ensures that tracing has the expected values in the context.
+func TestWithTrace(t *testing.T) {
+	pc, file, _, _ := runtime.Caller(0) // get current caller.
+	f := runtime.FuncForPC(pc)
+
+	base := []valueTestCase{
+		{
+			key:           "trace.id",
+			notnilorempty: true,
+		},
+
+		{
+			key:           "trace.file",
+			expected:      file,
+			notnilorempty: true,
+		},
+		{
+			key:           "trace.line",
+			notnilorempty: true,
+		},
+		{
+			key:           "trace.start",
+			notnilorempty: true,
+		},
+	}
+
+	ctx, done := WithTrace(Background())
+	defer done("this will be emitted at end of test")
+
+	checkContextForValues(t, ctx, append(base, valueTestCase{
+		key:      "trace.func",
+		expected: f.Name(),
+	}))
+
+	traced := func() {
+		parentID := ctx.Value("trace.id") // ensure the parent trace id is correct.
+
+		pc, _, _, _ := runtime.Caller(0) // get current caller.
+		f := runtime.FuncForPC(pc)
+		ctx, done := WithTrace(ctx)
+		defer done("this should be subordinate to the other trace")
+		time.Sleep(time.Second)
+		checkContextForValues(t, ctx, append(base, valueTestCase{
+			key:      "trace.func",
+			expected: f.Name(),
+		}, valueTestCase{
+			key:      "trace.parent.id",
+			expected: parentID,
+		}))
+	}
+	traced()
+
+	time.Sleep(time.Second)
+}
+
+type valueTestCase struct {
+	key           string
+	expected      interface{}
+	notnilorempty bool // just check not empty/not nil
+}
+
+func checkContextForValues(t *testing.T, ctx Context, values []valueTestCase) {
+
+	for _, testcase := range values {
+		v := ctx.Value(testcase.key)
+		if testcase.notnilorempty {
+			if v == nil || v == "" {
+				t.Fatalf("value was nil or empty for %q: %#v", testcase.key, v)
+			}
+			continue
+		}
+
+		if v != testcase.expected {
+			t.Fatalf("unexpected value for key %q: %v != %v", testcase.key, v, testcase.expected)
+		}
+	}
+}

registry/handlers/app.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"time"
 
-	"code.google.com/p/go-uuid/uuid"
 	"github.com/docker/distribution"
 	"github.com/docker/distribution/configuration"
 	ctxu "github.com/docker/distribution/context"
@@ -32,11 +31,8 @@ import (
 // fields should be protected.
 type App struct {
 	context.Context
-	Config configuration.Configuration
 
-	// InstanceID is a unique id assigned to the application on each creation.
+	Config configuration.Configuration
-	// Provides information in the logs and context to identify restarts.
-	InstanceID string
 
 	router           *mux.Router                 // main application router, configured with dispatchers
 	driver           storagedriver.StorageDriver // driver maintains the app global storage driver instance.
@@ -52,17 +48,6 @@ type App struct {
 	redis *redis.Pool
 }
 
-// Value intercepts calls context.Context.Value, returning the current app id,
-// if requested.
-func (app *App) Value(key interface{}) interface{} {
-	switch key {
-	case "app.id":
-		return app.InstanceID
-	}
-
-	return app.Context.Value(key)
-}
-
 // NewApp takes a configuration and returns a configured app, ready to serve
 // requests. The app only implements ServeHTTP and can be wrapped in other
 // handlers accordingly.
@@ -70,11 +55,10 @@ func NewApp(ctx context.Context, configuration configuration.Configuration) *App
 	app := &App{
 		Config:  configuration,
 		Context: ctx,
-		InstanceID: uuid.New(),
 		router:  v2.RouterWithPrefix(configuration.HTTP.Prefix),
 	}
 
-	app.Context = ctxu.WithLogger(app.Context, ctxu.GetLogger(app, "app.id"))
+	app.Context = ctxu.WithLogger(app.Context, ctxu.GetLogger(app, "instance.id"))
 
 	// Register the handler dispatchers.
 	app.register(v2.RouteNameBase, func(ctx *Context, r *http.Request) http.Handler {
@@ -200,7 +184,7 @@ func (app *App) configureEvents(configuration *configuration.Configuration) {
 
 	app.events.source = notifications.SourceRecord{
 		Addr:       hostname,
-		InstanceID: app.InstanceID,
+		InstanceID: ctxu.GetStringValue(app, "instance.id"),
 	}
 }
 

registry/storage/driver/base/base.go

@@ -40,6 +40,7 @@ package base
 import (
 	"io"
 
+	"github.com/docker/distribution/context"
 	storagedriver "github.com/docker/distribution/registry/storage/driver"
 )
 
@@ -51,6 +52,9 @@ type Base struct {
 
 // GetContent wraps GetContent of underlying storage driver.
 func (base *Base) GetContent(path string) ([]byte, error) {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.GetContent")
+
 	if !storagedriver.PathRegexp.MatchString(path) {
 		return nil, storagedriver.InvalidPathError{Path: path}
 	}
@@ -60,6 +64,9 @@ func (base *Base) GetContent(path string) ([]byte, error) {
 
 // PutContent wraps PutContent of underlying storage driver.
 func (base *Base) PutContent(path string, content []byte) error {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.PutContent")
+
 	if !storagedriver.PathRegexp.MatchString(path) {
 		return storagedriver.InvalidPathError{Path: path}
 	}
@@ -69,6 +76,9 @@ func (base *Base) PutContent(path string, content []byte) error {
 
 // ReadStream wraps ReadStream of underlying storage driver.
 func (base *Base) ReadStream(path string, offset int64) (io.ReadCloser, error) {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.ReadStream")
+
 	if offset < 0 {
 		return nil, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
 	}
@@ -82,6 +92,9 @@ func (base *Base) ReadStream(path string, offset int64) (io.ReadCloser, error) {
 
 // WriteStream wraps WriteStream of underlying storage driver.
 func (base *Base) WriteStream(path string, offset int64, reader io.Reader) (nn int64, err error) {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.WriteStream")
+
 	if offset < 0 {
 		return 0, storagedriver.InvalidOffsetError{Path: path, Offset: offset}
 	}
@@ -95,6 +108,9 @@ func (base *Base) WriteStream(path string, offset int64, reader io.Reader) (nn i
 
 // Stat wraps Stat of underlying storage driver.
 func (base *Base) Stat(path string) (storagedriver.FileInfo, error) {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.Stat")
+
 	if !storagedriver.PathRegexp.MatchString(path) {
 		return nil, storagedriver.InvalidPathError{Path: path}
 	}
@@ -104,6 +120,9 @@ func (base *Base) Stat(path string) (storagedriver.FileInfo, error) {
 
 // List wraps List of underlying storage driver.
 func (base *Base) List(path string) ([]string, error) {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.List")
+
 	if !storagedriver.PathRegexp.MatchString(path) && path != "/" {
 		return nil, storagedriver.InvalidPathError{Path: path}
 	}
@@ -113,6 +132,9 @@ func (base *Base) List(path string) ([]string, error) {
 
 // Move wraps Move of underlying storage driver.
 func (base *Base) Move(sourcePath string, destPath string) error {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.Move")
+
 	if !storagedriver.PathRegexp.MatchString(sourcePath) {
 		return storagedriver.InvalidPathError{Path: sourcePath}
 	} else if !storagedriver.PathRegexp.MatchString(destPath) {
@@ -124,6 +146,9 @@ func (base *Base) Move(sourcePath string, destPath string) error {
 
 // Delete wraps Delete of underlying storage driver.
 func (base *Base) Delete(path string) error {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.Move")
+
 	if !storagedriver.PathRegexp.MatchString(path) {
 		return storagedriver.InvalidPathError{Path: path}
 	}
@@ -133,6 +158,9 @@ func (base *Base) Delete(path string) error {
 
 // URLFor wraps URLFor of underlying storage driver.
 func (base *Base) URLFor(path string, options map[string]interface{}) (string, error) {
+	_, done := context.WithTrace(context.Background())
+	defer done("Base.URLFor")
+
 	if !storagedriver.PathRegexp.MatchString(path) {
 		return "", storagedriver.InvalidPathError{Path: path}
 	}