contextual logging

clog/clog.go

@@ -0,0 +1,185 @@
+/*
+Package clog provides Context with logging information.
+*/
+package clog
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+
+	"github.com/golang/glog"
+)
+
+// unique type to prevent assignment.
+type clogContextKeyT struct{}
+
+var clogContextKey = clogContextKeyT{}
+
+const (
+	// standard keys
+	manifestID    = "manifestID"
+	sessionID     = "sessionID"
+	nonce         = "nonce"
+	seqNo         = "seqNo"
+	orchSessionID = "orchSessionID" // session id generated on orchestrator for broadcaster
+)
+
+// Verbose is a boolean type that implements Infof (like Printf) etc.
+type Verbose bool
+
+var stdKeys map[string]bool
+var stdKeysOrder = []string{manifestID, sessionID, nonce, seqNo, orchSessionID}
+
+func init() {
+	stdKeys = make(map[string]bool)
+	for _, key := range stdKeysOrder {
+		stdKeys[key] = true
+	}
+}
+
+type values struct {
+	mu   sync.RWMutex
+	vals map[string]string
+}
+
+func newValues() *values {
+	return &values{
+		vals: make(map[string]string),
+	}
+}
+
+// Clone creates new context with parentCtx as parent and
+// logging details from logCtx
+func Clone(parentCtx, logCtx context.Context) context.Context {
+	cmap, _ := logCtx.Value(clogContextKey).(*values)
+	newCmap := newValues()
+	if cmap != nil {
+		cmap.mu.RLock()
+		for k, v := range cmap.vals {
+			newCmap.vals[k] = v
+		}
+		cmap.mu.RUnlock()
+	}
+	return context.WithValue(parentCtx, clogContextKey, newCmap)
+}
+
+func AddManifestID(ctx context.Context, val string) context.Context {
+	return AddVal(ctx, manifestID, val)
+}
+
+func AddSessionID(ctx context.Context, val string) context.Context {
+	return AddVal(ctx, sessionID, val)
+}
+
+func AddNonce(ctx context.Context, val uint64) context.Context {
+	return AddVal(ctx, nonce, strconv.FormatUint(val, 10))
+}
+
+func AddSeqNo(ctx context.Context, val uint64) context.Context {
+	return AddVal(ctx, seqNo, strconv.FormatUint(val, 10))
+}
+
+func AddOrchSessionID(ctx context.Context, val string) context.Context {
+	return AddVal(ctx, orchSessionID, val)
+}
+
+func AddVal(ctx context.Context, key, val string) context.Context {
+	cmap, _ := ctx.Value(clogContextKey).(*values)
+	if cmap == nil {
+		cmap = newValues()
+		ctx = context.WithValue(ctx, clogContextKey, cmap)
+	}
+	cmap.mu.Lock()
+	cmap.vals[key] = val
+	cmap.mu.Unlock()
+	return ctx
+}
+
+func Warningf(ctx context.Context, format string, args ...interface{}) {
+	glog.WarningDepth(1, formatMessage(ctx, false, format, args...))
+}
+
+func Errorf(ctx context.Context, format string, args ...interface{}) {
+	glog.ErrorDepth(1, formatMessage(ctx, false, format, args...))
+}
+
+func Fatalf(ctx context.Context, format string, args ...interface{}) {
+	glog.FatalDepth(1, formatMessage(ctx, false, format, args...))
+}
+
+func Infof(ctx context.Context, format string, args ...interface{}) {
+	infof(ctx, false, format, args...)
+}
+
+// Infofe if last argument is not nil it will be printed as " err=%q"
+func Infofe(ctx context.Context, format string, args ...interface{}) {
+	infof(ctx, true, format, args...)
+}
+
+func V(level glog.Level) Verbose {
+	return Verbose(glog.V(level))
+}
+
+// Infof is equivalent to the global Infof function, guarded by the value of v.
+// See the documentation of V for usage.
+func (v Verbose) Infof(ctx context.Context, format string, args ...interface{}) {
+	if v {
+		infof(ctx, false, format, args...)
+	}
+}
+
+func (v Verbose) Infofe(ctx context.Context, format string, args ...interface{}) {
+	if v {
+		infof(ctx, true, format, args...)
+	}
+}
+
+func infof(ctx context.Context, lastErr bool, format string, args ...interface{}) {
+	glog.InfoDepth(2, formatMessage(ctx, lastErr, format, args...))
+}
+
+func messageFromContext(ctx context.Context, sb *strings.Builder) {
+	if ctx == nil {
+		return
+	}
+	cmap, _ := ctx.Value(clogContextKey).(*values)
+	if cmap == nil {
+		return
+	}
+	cmap.mu.RLock()
+	for _, key := range stdKeysOrder {
+		if val, ok := cmap.vals[key]; ok {
+			sb.WriteString(key)
+			sb.WriteString("=")
+			sb.WriteString(val)
+			sb.WriteString(" ")
+		}
+	}
+	for key, val := range cmap.vals {
+		if _, ok := stdKeys[key]; !ok {
+			sb.WriteString(key)
+			sb.WriteString("=")
+			sb.WriteString(val)
+			sb.WriteString(" ")
+		}
+	}
+	cmap.mu.RUnlock()
+}
+
+func formatMessage(ctx context.Context, lastErr bool, format string, args ...interface{}) string {
+	var sb strings.Builder
+	messageFromContext(ctx, &sb)
+	var err interface{}
+	if lastErr && len(args) > 0 {
+		err = args[len(args)-1]
+		args = args[:len(args)-1]
+	}
+	sb.WriteString(fmt.Sprintf(format, args...))
+	if err != nil {
+		sb.WriteString(fmt.Sprintf(" err=%q", err))
+	}
+	return sb.String()
+}

clog/clog_test.go

@@ -0,0 +1,39 @@
+package clog
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStdKeys(t *testing.T) {
+	assert := assert.New(t)
+	ctx := AddManifestID(context.Background(), "manID")
+	ctx = AddSessionID(ctx, "sessionID")
+	ctx = AddNonce(ctx, 1038)
+	ctx = AddOrchSessionID(ctx, "orchID")
+	ctx = AddSeqNo(ctx, 9427)
+	ctx = AddVal(ctx, "customKey", "customVal")
+	msg := formatMessage(ctx, false, "testing message num=%d", 452)
+	assert.Equal("manifestID=manID sessionID=sessionID nonce=1038 seqNo=9427 orchSessionID=orchID customKey=customVal testing message num=452", msg)
+	ctxCloned := Clone(context.Background(), ctx)
+	ctxCloned = AddManifestID(ctxCloned, "newManifest")
+	msgCloned := formatMessage(ctxCloned, false, "testing message num=%d", 4521)
+	assert.Equal("manifestID=newManifest sessionID=sessionID nonce=1038 seqNo=9427 orchSessionID=orchID customKey=customVal testing message num=4521", msgCloned)
+	// old context shouldn't change
+	msg = formatMessage(ctx, false, "testing message num=%d", 452)
+	assert.Equal("manifestID=manID sessionID=sessionID nonce=1038 seqNo=9427 orchSessionID=orchID customKey=customVal testing message num=452", msg)
+}
+
+func TestLastErr(t *testing.T) {
+	assert := assert.New(t)
+	ctx := AddManifestID(context.Background(), "manID")
+	var err error
+	msg := formatMessage(ctx, true, "testing message num=%d", 452, err)
+	assert.Equal("manifestID=manID testing message num=452", msg)
+	err = errors.New("test error")
+	msg = formatMessage(ctx, true, "testing message num=%d", 452, err)
+	assert.Equal("manifestID=manID testing message num=452 err=\"test error\"", msg)
+}

cmd/livepeer/livepeer.go

@@ -276,7 +276,7 @@ func main() {
 			if *testTranscoder {
 				err := core.TestNvidiaTranscoder(devices)
 				if err != nil {
-					glog.Fatalf("Unable to transcode using Nvidia gpu=%s err=%v", strings.Join(devices, ","), err)
+					glog.Fatalf("Unable to transcode using Nvidia gpu=%q err=%q", strings.Join(devices, ","), err)
 				}
 			}
 			// FIXME: Short-term hack to pre-load the detection models on every device
@@ -1079,19 +1079,19 @@ func getServiceURI(n *core.LivepeerNode, serviceAddr string) (*url.URL, error) {
 	// TODO probably should put this (along w wizard GETs) into common code
 	resp, err := http.Get("https://api.ipify.org?format=text")
 	if err != nil {
-		glog.Errorf("Could not look up public IP err=%v", err)
+		glog.Errorf("Could not look up public IP err=%q", err)
 		return nil, err
 	}
 	defer resp.Body.Close()
 	body, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
-		glog.Errorf("Could not look up public IP err=%v", err)
+		glog.Errorf("Could not look up public IP err=%q", err)
 		return nil, err
 	}
 	addr := "https://" + strings.TrimSpace(string(body)) + ":" + RpcPort
 	inferredUri, err := url.ParseRequestURI(addr)
 	if err != nil {
-		glog.Errorf("Could not look up public IP err=%v", err)
+		glog.Errorf("Could not look up public IP err=%q", err)
 		return nil, err
 	}
 	if n.Eth == nil {
@@ -1102,12 +1102,12 @@ func getServiceURI(n *core.LivepeerNode, serviceAddr string) (*url.URL, error) {
 	// On-chain lookup and matching with inferred public address
 	addr, err = n.Eth.GetServiceURI(n.Eth.Account().Address)
 	if err != nil {
-		glog.Errorf("Could not get service URI; orchestrator may be unreachable err=%v", err)
+		glog.Errorf("Could not get service URI; orchestrator may be unreachable err=%q", err)
 		return nil, err
 	}
 	ethUri, err := url.ParseRequestURI(addr)
 	if err != nil {
-		glog.Errorf("Could not parse service URI; orchestrator may be unreachable err=%v", err)
+		glog.Errorf("Could not parse service URI; orchestrator may be unreachable err=%q", err)
 		ethUri, _ = url.ParseRequestURI("http://127.0.0.1:" + RpcPort)
 	}
 	if ethUri.Hostname() != inferredUri.Hostname() || ethUri.Port() != inferredUri.Port() {

common/db.go

@@ -766,7 +766,7 @@ func (db *DB) SelectEarliestWinningTicket(sender ethcommon.Address, minCreationR
 	)
 	if err := row.Scan(&senderString, &recipient, &faceValue, &winProb, &senderNonce, &recipientRand, &recipientRandHash, &sig, &creationRound, &creationRoundBlockHash, &paramsExpirationBlock); err != nil {
 		if err.Error() != "sql: no rows in result set" {
-			return nil, fmt.Errorf("could not retrieve earliest ticket err=%v", err)
+			return nil, fmt.Errorf("could not retrieve earliest ticket err=%q", err)
 		}
 		// If there is no result return no error, just nil value
 		return nil, nil

common/types.go

@@ -1,6 +1,7 @@
 package common
 
 import (
+	"context"
 	"encoding/json"
 	"math/big"
 	"net/url"
@@ -74,7 +75,7 @@ type OrchestratorPool interface {
 	// GetInfo gets info for specific orchestrator
 	GetInfo(uri string) OrchestratorLocalInfo
 	GetInfos() []OrchestratorLocalInfo
-	GetOrchestrators(int, Suspender, CapabilityComparator, ScorePred) ([]*net.OrchestratorInfo, error)
+	GetOrchestrators(context.Context, int, Suspender, CapabilityComparator, ScorePred) ([]*net.OrchestratorInfo, error)
 	Size() int
 	SizeWith(ScorePred) int
 }

core/capabilities_test.go

@@ -282,7 +282,7 @@ func (os *stubOS) GetInfo() *net.OSInfo {
 	return &net.OSInfo{StorageType: net.OSInfo_StorageType(os.storageType)}
 }
 func (os *stubOS) EndSession() {}
-func (os *stubOS) SaveData(string, []byte, map[string]string, time.Duration) (string, error) {
+func (os *stubOS) SaveData(context.Context, string, []byte, map[string]string, time.Duration) (string, error) {
 	return "", nil
 }
 func (os *stubOS) IsExternal() bool      { return false }

core/core_test.go

@@ -57,14 +57,14 @@ func TestTranscode(t *testing.T) {
 	md := &SegTranscodingMetadata{Profiles: videoProfiles, AuthToken: stubAuthToken()}
 
 	// Check nil transcoder.
-	tr, err := n.sendToTranscodeLoop(md, ss)
+	tr, err := n.sendToTranscodeLoop(context.TODO(), md, ss)
 	if err != ErrTranscoderAvail {
 		t.Error("Error transcoding ", err)
 	}
 
 	// Sanity check full flow.
 	n.Transcoder = NewLocalTranscoder(tmp)
-	tr, err = n.sendToTranscodeLoop(md, ss)
+	tr, err = n.sendToTranscodeLoop(context.TODO(), md, ss)
 	if err != nil {
 		t.Error("Error transcoding ", err)
 	}
@@ -106,18 +106,18 @@ func TestTranscodeSeg(t *testing.T) {
 
 	// Test offchain mode
 	require.Nil(n.Eth) // sanity check the offchain precondition of a nil eth
-	res := n.transcodeSeg(conf, seg, md)
+	res := n.transcodeSeg(context.TODO(), conf, seg, md)
 	assert.Nil(res.Err)
 	assert.Nil(res.Sig)
 	// sanity check results
-	resBytes, _ := n.Transcoder.Transcode(md)
+	resBytes, _ := n.Transcoder.Transcode(context.TODO(), md)
 	for i, trData := range res.TranscodeData.Segments {
 		assert.Equal(resBytes.Segments[i].Data, trData.Data)
 	}
 
 	// Test onchain mode
 	n.Eth = &eth.StubClient{}
-	res = n.transcodeSeg(conf, seg, md)
+	res = n.transcodeSeg(context.TODO(), conf, seg, md)
 	assert.Nil(res.Err)
 	assert.NotNil(res.Sig)
 	// check sig
@@ -147,7 +147,7 @@ func TestTranscodeLoop_GivenNoSegmentsPastTimeout_CleansSegmentChan(t *testing.T
 
 	mid := ManifestID(md.AuthToken.SessionId)
 
-	_, err := n.sendToTranscodeLoop(md, ss)
+	_, err := n.sendToTranscodeLoop(context.TODO(), md, ss)
 	require.Nil(err)
 	segChan := getSegChan(n, mid)
 	require.NotNil(segChan)