[Prism] Improve logging messages and levels. (#36242)

* Improve logging messages. No functional changes.

* Add some more logging messages.

Author: shunping

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

@@ -260,7 +260,7 @@ func NewElementManager(config Config) *ElementManager {
 // AddStage adds a stage to this element manager, connecting it's PCollections and
 // nodes to the watermark propagation graph.
 func (em *ElementManager) AddStage(ID string, inputIDs, outputIDs []string, sides []LinkID) {
-	slog.Debug("AddStage", slog.String("ID", ID), slog.Any("inputs", inputIDs), slog.Any("sides", sides), slog.Any("outputs", outputIDs))
+	slog.Debug("em.AddStage", slog.String("ID", ID), slog.Any("inputs", inputIDs), slog.Any("sides", sides), slog.Any("outputs", outputIDs))
 	ss := makeStageState(ID, inputIDs, outputIDs, sides)
 
 	em.stages[ss.ID] = ss
@@ -504,6 +504,40 @@ func (em *ElementManager) Bundles(ctx context.Context, upstreamCancelFn context.
 	return runStageCh
 }
 
+// DumpStages puts all the stage information into a string and returns it.
+func (em *ElementManager) DumpStages() string {
+	var stageState []string
+	ids := maps.Keys(em.stages)
+	if em.testStreamHandler != nil {
+		stageState = append(stageState, fmt.Sprintf("TestStreamHandler: completed %v, curIndex %v of %v events: %+v, processingTime %v, %v, ptEvents %v \n",
+			em.testStreamHandler.completed, em.testStreamHandler.nextEventIndex, len(em.testStreamHandler.events), em.testStreamHandler.events, em.testStreamHandler.processingTime, mtime.FromTime(em.testStreamHandler.processingTime), em.processTimeEvents))
+	} else {
+		stageState = append(stageState, fmt.Sprintf("ElementManager Now: %v processingTimeEvents: %v injectedBundles: %v\n", em.ProcessingTimeNow(), em.processTimeEvents.events, em.injectedBundles))
+	}
+	sort.Strings(ids)
+	for _, id := range ids {
+		ss := em.stages[id]
+		inW := ss.InputWatermark()
+		outW := ss.OutputWatermark()
+		upPCol, upW := ss.UpstreamWatermark()
+		upS := em.pcolParents[upPCol]
+		if upS == "" {
+			upS = "IMPULSE  " // (extra spaces to allow print to align better.)
+		}
+		stageState = append(stageState, fmt.Sprintln(id, "watermark in", inW, "out", outW, "upstream", upW, "from", upS, "pending", ss.pending, "byKey", ss.pendingByKeys, "inprogressKeys", ss.inprogressKeys, "byBundle", ss.inprogressKeysByBundle, "holds", ss.watermarkHolds.heap, "holdCounts", ss.watermarkHolds.counts, "holdsInBundle", ss.inprogressHoldsByBundle, "pttEvents", ss.processingTimeTimers.toFire, "bundlesToInject", ss.bundlesToInject))
+
+		var outputConsumers, sideConsumers []string
+		for _, col := range ss.outputIDs {
+			outputConsumers = append(outputConsumers, em.consumers[col]...)
+			for _, l := range em.sideConsumers[col] {
+				sideConsumers = append(sideConsumers, l.Global)
+			}
+		}
+		stageState = append(stageState, fmt.Sprintf("\tsideInputs: %v outputCols: %v outputConsumers: %v sideConsumers: %v\n", ss.sides, ss.outputIDs, outputConsumers, sideConsumers))
+	}
+	return strings.Join(stageState, "")
+}
+
 // checkForQuiescence sees if this element manager is no longer able to do any pending work or make progress.
 //
 // Quiescense can happen if there are no inprogress bundles, and there are no further watermark refreshes, which
@@ -524,9 +558,9 @@ func (em *ElementManager) checkForQuiescence(advanced set[string]) error {
 		// If there are changed stages that need a watermarks refresh,
 		// we aren't yet stuck.
 		v := em.livePending.Load()
-		slog.Debug("Bundles: nothing in progress after advance",
-			slog.Any("advanced", advanced),
-			slog.Int("changeCount", len(em.changedStages)),
+		slog.Debug("Bundles: nothing in progress after advance, but some stages need a watermark refresh",
+			slog.Any("mayProgress", advanced),
+			slog.Any("needRefresh", em.changedStages),
 			slog.Int64("pendingElementCount", v),
 		)
 		return nil
@@ -569,36 +603,7 @@ func (em *ElementManager) checkForQuiescence(advanced set[string]) error {
 	// Jobs must never get stuck so this indicates a bug in prism to be investigated.
 
 	slog.Debug("Bundles: nothing in progress and no refreshes", slog.Int64("pendingElementCount", v))
-	var stageState []string
-	ids := maps.Keys(em.stages)
-	if em.testStreamHandler != nil {
-		stageState = append(stageState, fmt.Sprintf("TestStreamHandler: completed %v, curIndex %v of %v events: %+v, processingTime %v, %v, ptEvents %v \n",
-			em.testStreamHandler.completed, em.testStreamHandler.nextEventIndex, len(em.testStreamHandler.events), em.testStreamHandler.events, em.testStreamHandler.processingTime, mtime.FromTime(em.testStreamHandler.processingTime), em.processTimeEvents))
-	} else {
-		stageState = append(stageState, fmt.Sprintf("ElementManager Now: %v processingTimeEvents: %v injectedBundles: %v\n", em.ProcessingTimeNow(), em.processTimeEvents.events, em.injectedBundles))
-	}
-	sort.Strings(ids)
-	for _, id := range ids {
-		ss := em.stages[id]
-		inW := ss.InputWatermark()
-		outW := ss.OutputWatermark()
-		upPCol, upW := ss.UpstreamWatermark()
-		upS := em.pcolParents[upPCol]
-		if upS == "" {
-			upS = "IMPULSE  " // (extra spaces to allow print to align better.)
-		}
-		stageState = append(stageState, fmt.Sprintln(id, "watermark in", inW, "out", outW, "upstream", upW, "from", upS, "pending", ss.pending, "byKey", ss.pendingByKeys, "inprogressKeys", ss.inprogressKeys, "byBundle", ss.inprogressKeysByBundle, "holds", ss.watermarkHolds.heap, "holdCounts", ss.watermarkHolds.counts, "holdsInBundle", ss.inprogressHoldsByBundle, "pttEvents", ss.processingTimeTimers.toFire, "bundlesToInject", ss.bundlesToInject))
-
-		var outputConsumers, sideConsumers []string
-		for _, col := range ss.outputIDs {
-			outputConsumers = append(outputConsumers, em.consumers[col]...)
-			for _, l := range em.sideConsumers[col] {
-				sideConsumers = append(sideConsumers, l.Global)
-			}
-		}
-		stageState = append(stageState, fmt.Sprintf("\tsideInputs: %v outputCols: %v outputConsumers: %v sideConsumers: %v\n", ss.sides, ss.outputIDs, outputConsumers, sideConsumers))
-	}
-	return errors.Errorf("nothing in progress and no refreshes with non zero pending elements: %v\n%v", v, strings.Join(stageState, ""))
+	return errors.Errorf("nothing in progress and no refreshes with non zero pending elements: %v\n%v", v, em.DumpStages())
 }
 
 // InputForBundle returns pre-allocated data for the given bundle, encoding the elements using
@@ -864,7 +869,9 @@ func (em *ElementManager) PersistBundle(rb RunBundle, col2Coders map[string]PCol
 		}
 		consumers := em.consumers[output]
 		sideConsumers := em.sideConsumers[output]
-		slog.Debug("PersistBundle: bundle has downstream consumers.", "bundle", rb, slog.Int("newPending", len(newPending)), "consumers", consumers, "sideConsumers", sideConsumers)
+		slog.Debug("PersistBundle: bundle has downstream consumers.", "bundle", rb,
+			slog.Int("newPending", len(newPending)), "consumers", consumers, "sideConsumers", sideConsumers,
+			"pendingDelta", len(newPending)*len(consumers))
 		for _, sID := range consumers {
 			consumer := em.stages[sID]
 			count := consumer.AddPending(em, newPending)
@@ -1576,6 +1583,7 @@ func (ss *stageState) buildTriggeredBundle(em *ElementManager, key string, win t
 		nil,
 		panesInBundle,
 	)
+	slog.Debug("started a triggered bundle", "stageID", ss.ID, "bundleID", rb.BundleID, "size", len(toProcess))
 
 	ss.bundlesToInject = append(ss.bundlesToInject, rb)
 	// Bundle is marked in progress here to prevent a race condition.
@@ -1688,6 +1696,7 @@ func (ss *stageState) startEventTimeBundle(watermark mtime.Time, genBundID func(
 	}
 
 	bundID := ss.makeInProgressBundle(genBundID, toProcess, minTs, newKeys, holdsInBundle, panesInBundle)
+	slog.Debug("started an event time bundle", "stageID", ss.ID, "bundleID", bundID, "bundleSize", len(toProcess), "upstreamWatermark", watermark)
 
 	return bundID, true, stillSchedulable, accumulatingPendingAdjustment
 }
@@ -1987,6 +1996,8 @@ func (ss *stageState) startProcessingTimeBundle(em *ElementManager, emNow mtime.
 		return "", false, stillSchedulable
 	}
 	bundID := ss.makeInProgressBundle(genBundID, toProcess, minTs, newKeys, holdsInBundle, nil)
+
+	slog.Debug("started a processing time bundle", "stageID", ss.ID, "bundleID", bundID, "size", len(toProcess), "emNow", emNow)
 	return bundID, true, stillSchedulable
 }
 
@@ -2274,8 +2285,7 @@ func (ss *stageState) bundleReady(em *ElementManager, emNow mtime.Time) (mtime.T
 		slog.Debug("bundleReady: unchanged upstream watermark",
 			slog.String("stage", ss.ID),
 			slog.Group("watermark",
-				slog.Any("upstream", upstreamW),
-				slog.Any("input", inputW)))
+				slog.Any("upstream == input == previousInput", inputW)))
 		return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady
 	}
 	ready := true

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

@@ -340,7 +340,6 @@ func executePipeline(ctx context.Context, wks map[string]*worker.W, j *jobservic
 				return fmt.Errorf("prism error building stage %v: \n%w", stage.ID, err)
 			}
 			stages[stage.ID] = stage
-			j.Logger.Debug("pipelineBuild", slog.Group("stage", slog.String("ID", stage.ID), slog.String("transformName", t.GetUniqueName())))
 			outputs := maps.Keys(stage.OutputsToCoders)
 			sort.Strings(outputs)
 			em.AddStage(stage.ID, []string{stage.primaryInput}, outputs, stage.sideInputs)
@@ -381,11 +380,7 @@ func executePipeline(ctx context.Context, wks map[string]*worker.W, j *jobservic
 		case rb, ok := <-bundles:
 			if !ok {
 				err := eg.Wait()
-				var topoAttrs []any
-				for _, s := range topo {
-					topoAttrs = append(topoAttrs, slog.Any(s.ID, s))
-				}
-				j.Logger.Debug("pipeline done!", slog.String("job", j.String()), slog.Any("error", err), slog.Group("topo", topoAttrs...))
+				j.Logger.Debug("pipeline done!", slog.String("job", j.String()), slog.Any("error", err), slog.String("stages", em.DumpStages()))
 				return err
 			}
 			eg.Go(func() error {

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

@@ -182,6 +182,20 @@ func (p *preprocessor) preProcessGraph(comps *pipepb.Components, j *jobservices.
 			return nil
 		}
 	}
+	var stageDetails []any
+	for i, stg := range stages {
+		var transformNames []string
+		for _, tid := range stg.transforms {
+			transformNames = append(transformNames, comps.GetTransforms()[tid].GetUniqueName())
+		}
+		stageDetails = append(stageDetails,
+			slog.Group(fmt.Sprintf("stage-%03d", i),
+				slog.String("environment", stg.envID),
+				slog.Any("transforms", transformNames),
+			),
+		)
+	}
+	slog.Debug("preProcessGraph: all stages and transforms", stageDetails...)
 	return stages
 }
 

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

@@ -174,7 +174,7 @@ func (s *stage) Execute(ctx context.Context, j *jobservices.Job, wk *worker.W, c
 
 		s.prepareSides(b, rb.Watermark)
 
-		slog.Debug("Execute: processing", "bundle", rb)
+		slog.Debug("Execute: sdk worker transform(s)", "bundle", rb)
 		defer b.Cleanup(wk)
 		dataReady = b.ProcessOn(ctx, wk)
 	default:
@@ -354,7 +354,7 @@ progress:
 			slog.Error("SDK Error from bundle finalization", "bundle", rb, "error", err.Error())
 			panic(err)
 		}
-		slog.Info("finalized bundle", "bundle", rb)
+		slog.Debug("finalized bundle", "bundle", rb)
 	}
 	b.OutputData = engine.TentativeData{} // Clear the data.
 	return nil

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

@@ -386,7 +386,7 @@ func (wk *W) Data(data fnpb.BeamFnData_DataServer) error {
 			for _, d := range resp.GetData() {
 				cr, ok := wk.activeInstructions[d.GetInstructionId()]
 				if !ok {
-					slog.Info("data.Recv data for unknown bundle", "response", resp)
+					slog.Debug("data.Recv data for unknown bundle", "response", resp)
 					continue
 				}
 				// Received data is always for an active ProcessBundle instruction
@@ -405,7 +405,7 @@ func (wk *W) Data(data fnpb.BeamFnData_DataServer) error {
 			for _, t := range resp.GetTimers() {
 				cr, ok := wk.activeInstructions[t.GetInstructionId()]
 				if !ok {
-					slog.Info("data.Recv timers for unknown bundle", "response", resp)
+					slog.Debug("data.Recv timers for unknown bundle", "response", resp)
 					continue
 				}
 				// Received data is always for an active ProcessBundle instruction