Add ForwardOutput worker plugin to forward stdout and stderr to client.

[hendrikmakait]

This PR adds a ForwardOutput worker plugin that forwards all output from stdout and stderr on all workers to all clients using the internals implemented for distributed.worker.print. This plugin is not intended for chatty workloads and may put significant load on the scheduler.

Note that this plugin is related to #7276 but takes a slightly different route as it tries to avoid adding additional client methods that start to bloat the client interface. This comes at the cost of forwarding to all clients. This is the same behavior that we already have with distributed.print(). Unless somebody is particularly interested in using this plugin in a multi-client environment, I would leave this as is until we have added client plugins that allow adding functionality on the client without the need for dedicated methods.

• Closes Plugin to pipe stdout/stderr to a log event #7202

• Tests added / passed
• Passes pre-commit run --all-files

[github-actions]

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

       15 files  ±0         15 suites  ±0   6h 27m 34s ⏱️ - 4m 41s
  3 217 tests +1    3 131 ✔️ ±0    84 💤 ±0  2 ❌ +1 
23 789 runs  +8  22 878 ✔️ +7  909 💤 ±0  2 ❌ +1 

For more details on these failures, see this check.

Results for commit 3c84bbd. ± Comparison against base commit ffc418e.

♻️ This comment has been updated with latest results.

[hendrikmakait]

I just realized that this does not capture any logging.StreamHandler that has been initialized before the plugin was installed.

[hendrikmakait]

I just realized that this does not capture any logging.StreamHandler that has been initialized before the plugin was installed.

Fixed by monkey-patching instead of replacing the stream objects.

[fjetter]

FWIW I don't really care about the logging stream handlers. We can forward logging on different channels. If it works like this, I'm OK with it though

[hendrikmakait]

The issue I encountered with logging.StreamHandlers generalized to any object that stores a reference to objects sys.std{out|err} reference before we replace them, so this might occur in a few other places that follow the same pattern. It wasn't much effort to tweak since all the heavy-lifting of forwarding the logs had been implemented, it merely became a small exercise in monkey-patching instance methods in the right spot.