[Prism] Fix two kinds of errors when using json log format.

[shunping]

A follow-up PR of #36008

Fixed two types of errors can be seen when using json log format.

INFO:apache_beam.utils.subprocess_server:{"time":"2025-08-29T21:58:03.309126-04:00","level":"DEBUG",
  "source":{"function":"github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal.executePipeline",
  "file":"sdks/go/pkg/beam/runners/prism/internal/execute.go","line":363},
  "msg":"pipeline done!","job":"job-001[job]","error":null,
  "topo":"!ERROR:json: unsupported type: func(io.Reader) []uint8"}

INFO:apache_beam.utils.subprocess_server:{"time":"2025-08-29T22:05:24.038412-04:00","level":"DEBUG",
  "source":{"function":"github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal.(*stage).Execute",
  "file":"sdks/go/pkg/beam/runners/prism/internal/stage.go","line":287},
  "msg":"Execute: committing data",
  "bundle":{"ID":"inst002","stage":"stage-005","watermark":"!ERROR:time.Time year outside of range [0,9999]""294247-01-10T04:00:54.775Z"},"outputsWithData":[],"outputs":[]}

After the fix, it prints out the correct fields in topo ([]*stage) and uses epoch millisecond for watermarks.

There are also some minor improvements on the logging logic. For example, logs from expansion service are now recorded with logger name "ExpansionService".

[shunping]

/gemini review

[gemini-code-assist]

Code Review

This pull request aims to fix two logging-related errors when using the JSON log format. The changes correctly address the original issues by modifying how watermarks are logged and by improving the logging logic for container logs and expansion services.

However, the new slog.LogValuer implementations for PColInfo and stage introduce new logging issues. Specifically, PColInfo.LogValue attempts to log function fields, which will cause a JSON marshaling error. Additionally, both stage.LogValue and the topology logging in executePipeline incorrectly use slog.Any to log a slice of attributes, which should be handled with slog.Group. I've provided specific comments and suggestions to fix these new issues.

The Python-side changes to support this are well-implemented and improve the robustness and structure of the logging system.

[codecov]

Codecov Report

❌ Patch coverage is 12.82051% with 34 lines in your changes missing coverage. Please review.
✅ Project coverage is 56.76%. Comparing base (e72d9c7) to head (b544c7e).
⚠️ Report is 20 commits behind head on master.

Files with missing lines	Patch %	Lines
...am/runners/prism/internal/engine/elementmanager.go	0.00%	17 Missing ⚠️
sdks/go/pkg/beam/runners/prism/internal/stage.go	0.00%	11 Missing ⚠️
...go/pkg/beam/runners/prism/internal/environments.go	0.00%	6 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #36020      +/-   ##
============================================
- Coverage     56.77%   56.76%   -0.01%     
  Complexity     3383     3383              
============================================
  Files          1220     1220              
  Lines        184968   184998      +30     
  Branches       3508     3508              
============================================
+ Hits         105015   105016       +1     
- Misses        76625    76658      +33     
+ Partials       3328     3324       -4     

Flag	Coverage Δ
go	28.21% <12.82%> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Assigning reviewers:

R: @jrmccluskey for label python.
R: @lostluck for label go.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).