server: Add orch info on AI events

CHANGELOG_PENDING.md

@@ -9,6 +9,7 @@
 #### General
 
 #### Broadcaster
+- [#3321](https://github.com/livepeer/go-livepeer/pull/3321) Add orchestrator info on live AI monitoring events
 
 #### Orchestrator
 

server/ai_live_video.go

@@ -199,7 +199,7 @@ func startControlPublish(control *url.URL, params aiRequestParams) {
 	}()
 }
 
-func startEventsSubscribe(ctx context.Context, url *url.URL, params aiRequestParams) {
+func startEventsSubscribe(ctx context.Context, url *url.URL, params aiRequestParams, sess *AISession) {
 	subscriber := trickle.NewTrickleSubscriber(url.String())
 	stream := params.liveParams.stream
 	streamId := params.liveParams.streamID
@@ -233,6 +233,12 @@ func startEventsSubscribe(ctx context.Context, url *url.URL, params aiRequestPar
 			event["stream_id"] = streamId
 			event["request_id"] = params.liveParams.requestID
 			event["pipeline_id"] = params.liveParams.pipelineID
+			if sess != nil {
+				event["orchestrator_info"] = map[string]interface{}{
+					"address": sess.Address(),
+					"url":     sess.Transcoder(),
+				}
+			}
 
 			clog.Infof(ctx, "Received event for stream=%s event=%+v", stream, event)
 

server/ai_process.go

@@ -1073,7 +1073,7 @@ func submitLiveVideoToVideo(ctx context.Context, params aiRequestParams, sess *A
 	startControlPublish(control, params)
 	startTricklePublish(ctx, pub, params, sess)
 	startTrickleSubscribe(ctx, sub, params)
-	startEventsSubscribe(ctx, events, params)
+	startEventsSubscribe(ctx, events, params, sess)
 	return resp, nil
 }