Add a key ordering to clog logging (#3424)

* Add a key ordering to clog logging

* update log line

* Update test

Author: mjh1

clog/clog.go

@@ -52,13 +52,15 @@ func init() {
 }
 
 type values struct {
-	mu   sync.RWMutex
-	vals map[string]string
+	mu        sync.RWMutex
+	keysOrder []string
+	vals      map[string]string
 }
 
 func newValues() *values {
 	return &values{
-		vals: make(map[string]string),
+		vals:      make(map[string]string),
+		keysOrder: []string{},
 	}
 }
 
@@ -71,6 +73,7 @@ func Clone(parentCtx, logCtx context.Context) context.Context {
 		cmap.mu.RLock()
 		for k, v := range cmap.vals {
 			newCmap.vals[k] = v
+			newCmap.keysOrder = append(newCmap.keysOrder, k)
 		}
 		cmap.mu.RUnlock()
 	}
@@ -109,6 +112,10 @@ func AddVal(ctx context.Context, key, val string) context.Context {
 		ctx = context.WithValue(ctx, clogContextKey, cmap)
 	}
 	cmap.mu.Lock()
+	// add to keysOrder only if the key is not already present to avoid duplicate fields
+	if _, ok := cmap.vals[key]; !ok {
+		cmap.keysOrder = append(cmap.keysOrder, key)
+	}
 	cmap.vals[key] = val
 	cmap.mu.Unlock()
 	return ctx
@@ -203,11 +210,11 @@ func messageFromContext(ctx context.Context, sb *strings.Builder) {
 			sb.WriteString(" ")
 		}
 	}
-	for key, val := range cmap.vals {
+	for _, key := range cmap.keysOrder {
 		if _, ok := stdKeys[key]; !ok {
 			sb.WriteString(key)
 			sb.WriteString("=")
-			sb.WriteString(val)
+			sb.WriteString(cmap.vals[key])
 			sb.WriteString(" ")
 		}
 	}

clog/clog_test.go

@@ -22,8 +22,9 @@ func TestStdKeys(t *testing.T) {
 	assert.Equal("manifestID=manID sessionID=sessionID nonce=1038 seqNo=9427 orchSessionID=orchID ethaddress=0x0 orchestrator=http://127.0.0.1:8935 customKey=customVal testing message num=452", msg)
 	ctxCloned := Clone(context.Background(), ctx)
 	ctxCloned = AddManifestID(ctxCloned, "newManifest")
+	ctxCloned = AddVal(ctxCloned, "customKey2", "customVal2")
 	msgCloned, _ := formatMessage(ctxCloned, false, false, "testing message num=%d", 4521)
-	assert.Equal("manifestID=newManifest sessionID=sessionID nonce=1038 seqNo=9427 orchSessionID=orchID ethaddress=0x0 orchestrator=http://127.0.0.1:8935 customKey=customVal testing message num=4521", msgCloned)
+	assert.Equal("manifestID=newManifest sessionID=sessionID nonce=1038 seqNo=9427 orchSessionID=orchID ethaddress=0x0 orchestrator=http://127.0.0.1:8935 customKey=customVal customKey2=customVal2 testing message num=4521", msgCloned)
 	// old context shouldn't change
 	msg, _ = formatMessage(ctx, false, false, "testing message num=%d", 452)
 	assert.Equal("manifestID=manID sessionID=sessionID nonce=1038 seqNo=9427 orchSessionID=orchID ethaddress=0x0 orchestrator=http://127.0.0.1:8935 customKey=customVal testing message num=452", msg)

server/ai_mediaserver.go

@@ -540,7 +540,7 @@ func (ls *LivepeerServer) StartLiveVideo() http.Handler {
 			ms := media.MediaSegmenter{Workdir: ls.LivepeerNode.WorkDir, MediaMTXClient: mediaMTXClient}
 			ms.RunSegmentation(ctx, fmt.Sprintf("rtmp://%s/%s%s", remoteHost, mediaMTXStreamPrefix, streamName), ssr.Read)
 			ssr.Close()
-			ls.cleanupLive(streamName)
+			ls.cleanupLive(ctx, streamName)
 		}()
 
 		sendErrorEvent := LiveErrorEventSender(ctx, streamID, map[string]string{
@@ -557,7 +557,7 @@ func (ls *LivepeerServer) StartLiveVideo() http.Handler {
 			if err == nil {
 				return
 			}
-			clog.Errorf(ctx, "Live video pipeline stopping: %s", err)
+			clog.Errorf(ctx, "Live video pipeline finished with error: %s", err)
 
 			sendErrorEvent(err)
 
@@ -687,7 +687,8 @@ func (ls *LivepeerServer) GetLiveVideoToVideoStatus() http.Handler {
 	})
 }
 
-func (ls *LivepeerServer) cleanupLive(stream string) {
+func (ls *LivepeerServer) cleanupLive(ctx context.Context, stream string) {
+	clog.Infof(ctx, "Live video pipeline finished")
 	ls.LivepeerNode.LiveMu.Lock()
 	pub, ok := ls.LivepeerNode.LivePipelines[stream]
 	delete(ls.LivepeerNode.LivePipelines, stream)