Add active bundles in the heartbeat.

Author: shunping

sdks/go/pkg/beam/runners/prism/internal/execute.go

@@ -396,7 +396,8 @@ func executePipeline(ctx context.Context, wks map[string]*worker.W, j *jobservic
 					j.Logger.Info("worker status",
 						slog.String("workerID", wk.ID),
 						slog.String("envID", envID),
-						slog.Duration("uptime", wk.Uptime()))
+						slog.Duration("uptime", wk.Uptime()),
+						slog.Any("active_bundles", wk.ActiveBundles()))
 				}
 			}
 		}

sdks/go/pkg/beam/runners/prism/internal/worker/worker.go

@@ -303,6 +303,19 @@ func (wk *W) Uptime() time.Duration {
 	return time.Since(wk.StartTime)
 }
 
+// ActiveBundles returns a list of active bundles currently processing on this worker.
+func (wk *W) ActiveBundles() []string {
+	wk.mu.Lock()
+	defer wk.mu.Unlock()
+	var bundles []string
+	for id, responder := range wk.activeInstructions {
+		if b, ok := responder.(*B); ok {
+			bundles = append(bundles, fmt.Sprintf("%s (%s)", id, b.PBDID))
+		}
+	}
+	return bundles
+}
+
 // Control relays instructions to SDKs and back again, coordinated via unique instructionIDs.
 //
 // Requests come from the runner, and are sent to the client in the SDK.