You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "scwhittle (via GitHub)" <gi...@apache.org> on 2023/01/26 11:25:18 UTC
[GitHub] [beam] scwhittle opened a new pull request, #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
scwhittle opened a new pull request, #25186:
URL: https://github.com/apache/beam/pull/25186
Otherwise it can cause deadlocks if the log buffer fills up but is no longer being drained.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/get-started-contributing/#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1404873295
R: @lukecwik
Thoughts? This seems better than getting stuck. We could try to reconnect the stream but that is a larger change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1582312812
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/25186?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#25186](https://app.codecov.io/gh/apache/beam/pull/25186?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (19348af) into [master](https://app.codecov.io/gh/apache/beam/commit/74bad1add9cb6954c5f62aff81371c327546cc89?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (74bad1a) will **increase** coverage by `9.48%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 19348af differs from pull request most recent head 8fa642e. Consider uploading reports for the commit 8fa642e to get more accurate results
```diff
@@ Coverage Diff @@
## master #25186 +/- ##
==========================================
+ Coverage 71.53% 81.02% +9.48%
==========================================
Files 854 559 -295
Lines 104093 68122 -35971
==========================================
- Hits 74459 55193 -19266
+ Misses 28087 12929 -15158
+ Partials 1547 0 -1547
```
| Flag | Coverage Δ | |
|---|---|---|
| go | `?` | |
| python | `81.02% <ø> (-0.02%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
[see 301 files with indirect coverage changes](https://app.codecov.io/gh/apache/beam/pull/25186/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1098681684
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -287,9 +284,24 @@ public void run() {
outboundObserver.onError(
Status.INTERNAL.withDescription(getStackTraceAsString(thrown)).asException());
throw new IllegalStateException(thrown);
- } else {
- outboundObserver.onCompleted();
}
+ if (inboundObserverCompletion.isDone()) {
+ try {
+ inboundObserverCompletion.get();
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly with success before it was closed by the client.");
+ } catch (ExecutionException e) {
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly before it was closed by the client with error: "
+ + e.getCause());
+ } catch (InterruptedException e) {
+ // Should never happen because of the isDone check.
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ outboundObserver.onCompleted();
Review Comment:
Can you clarify? The only onError is followed by throwing an exception, is there something else I'm missing?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1582310919
R: @Abacn
Yi woudl you be able to take over this review from Luke? I lost track of it but I believe it still fixes a possible stuckness issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1098683719
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -122,7 +124,9 @@ public BeamFnLoggingClient(
logRecordHandler.setLevel(Level.ALL);
logRecordHandler.setFormatter(DEFAULT_FORMATTER);
logRecordHandler.executeOn(options.as(ExecutorOptions.class).getScheduledExecutorService());
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1087723142
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -102,7 +104,7 @@ public BeamFnLoggingClient(
Function<Endpoints.ApiServiceDescriptor, ManagedChannel> channelFactory) {
this.apiServiceDescriptor = apiServiceDescriptor;
this.inboundObserverCompletion = new CompletableFuture<>();
- this.phaser = new Phaser(1);
Review Comment:
Since this was a normal Phaser, I think that it didn't block for flow control previously after the first OnReady false->True transition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "lukecwik (via GitHub)" <gi...@apache.org>.
lukecwik commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1095077081
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -287,9 +284,24 @@ public void run() {
outboundObserver.onError(
Status.INTERNAL.withDescription(getStackTraceAsString(thrown)).asException());
throw new IllegalStateException(thrown);
- } else {
- outboundObserver.onCompleted();
}
+ if (inboundObserverCompletion.isDone()) {
+ try {
+ inboundObserverCompletion.get();
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly with success before it was closed by the client.");
+ } catch (ExecutionException e) {
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly before it was closed by the client with error: "
+ + e.getCause());
+ } catch (InterruptedException e) {
+ // Should never happen because of the isDone check.
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ outboundObserver.onCompleted();
Review Comment:
We can't call outboundObserver.onCompleted() if we already called outboundObserver.onError()
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -102,7 +104,7 @@ public BeamFnLoggingClient(
Function<Endpoints.ApiServiceDescriptor, ManagedChannel> channelFactory) {
this.apiServiceDescriptor = apiServiceDescriptor;
this.inboundObserverCompletion = new CompletableFuture<>();
- this.phaser = new Phaser(1);
Review Comment:
makes sense
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -122,7 +124,9 @@ public BeamFnLoggingClient(
logRecordHandler.setLevel(Level.ALL);
logRecordHandler.setFormatter(DEFAULT_FORMATTER);
logRecordHandler.executeOn(options.as(ExecutorOptions.class).getScheduledExecutorService());
Review Comment:
we have a race here, `executeOn` will start the logging thread before `outboundObserver` has been initialized.
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -122,7 +124,9 @@ public BeamFnLoggingClient(
logRecordHandler.setLevel(Level.ALL);
logRecordHandler.setFormatter(DEFAULT_FORMATTER);
logRecordHandler.executeOn(options.as(ExecutorOptions.class).getScheduledExecutorService());
- outboundObserver = (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver);
+ outboundObserver =
+ new DirectStreamObserver<BeamFnApi.LogEntry.List>(
+ phaser, (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver));
Review Comment:
There was no point in using the `DirectStreamObserver` since we guarantee that all writing to the stream happens from the single thread in `LogRecordHandler#run()` which will never be the inbound gRPC thread so we don't have to worry about waiting for the phaser and re-checking outbound observer readiness.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn merged pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #25186:
URL: https://github.com/apache/beam/pull/25186
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1580738522
This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1422698654
Sorry I'm used to other review systems where comments are buffered. I am working on improving the testing (figured out how with clientcall interceptor) and also refactoring things to log to default logger if this disconnects and to ensure buffer remains draining until closed to prevent deadlocks.
Do you think I should make it reconnect to the worker if stream breaks and otherwise dump to previous default logmanager?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1613217236
Hi, it appears this change breaks Java Unit Test GItHub Action on MacOS and Windows: #27292
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1235534965
##########
sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/stream/DirectStreamObserver.java:
##########
@@ -76,12 +77,13 @@ public void onNext(T value) {
// Record the initial phase in case we are in the inbound gRPC thread where the phase won't
// advance.
int initialPhase = phase;
- while (!outboundObserver.isReady()) {
+ // A negative phase indicates that the phaser is terminated.
+ while (phase >= 0 && !outboundObserver.isReady()) {
try {
phase = phaser.awaitAdvanceInterruptibly(phase, waitTime, TimeUnit.SECONDS);
} catch (TimeoutException e) {
totalTimeWaited += waitTime;
- waitTime = waitTime * 2;
+ waitTime = Math.min(waitTime * 2, 60);
Review Comment:
Done
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -87,28 +93,93 @@ public class BeamFnLoggingClient implements AutoCloseable {
private static final Object COMPLETED = new Object();
+ private final Endpoints.ApiServiceDescriptor apiServiceDescriptor;
+
+ private final StreamWriter streamWriter;
+
+ private final LogRecordHandler logRecordHandler;
+
/* We need to store a reference to the configured loggers so that they are not
* garbage collected. java.util.logging only has weak references to the loggers
* so if they are garbage collected, our hierarchical configuration will be lost. */
- private final Collection<Logger> configuredLoggers;
- private final Endpoints.ApiServiceDescriptor apiServiceDescriptor;
- private final ManagedChannel channel;
- private final CallStreamObserver<BeamFnApi.LogEntry.List> outboundObserver;
- private final LogControlObserver inboundObserver;
- private final LogRecordHandler logRecordHandler;
- private final CompletableFuture<Object> inboundObserverCompletion;
- private final Phaser phaser;
+ private final Collection<Logger> configuredLoggers = new ArrayList<>();
+
private @Nullable ProcessBundleHandler processBundleHandler;
- public BeamFnLoggingClient(
+ private final BlockingQueue<LogEntry> bufferedLogEntries =
+ new ArrayBlockingQueue<>(MAX_BUFFERED_LOG_ENTRY_COUNT);
+
+ //
Review Comment:
added comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "lukecwik (via GitHub)" <gi...@apache.org>.
lukecwik commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1098928903
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -287,9 +284,24 @@ public void run() {
outboundObserver.onError(
Status.INTERNAL.withDescription(getStackTraceAsString(thrown)).asException());
throw new IllegalStateException(thrown);
- } else {
- outboundObserver.onCompleted();
}
+ if (inboundObserverCompletion.isDone()) {
+ try {
+ inboundObserverCompletion.get();
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly with success before it was closed by the client.");
+ } catch (ExecutionException e) {
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly before it was closed by the client with error: "
+ + e.getCause());
+ } catch (InterruptedException e) {
+ // Should never happen because of the isDone check.
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ outboundObserver.onCompleted();
Review Comment:
your not missing anything, I missed that the exception was being thrown
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1438399226
I changed to a background thread that consumes the queue until the stream breaks or the client is closed.
It restores the previous loggers and flushes remaining logs to them if unclean closed.
It doesn't try to reconnect the logging stream, it surfaces the error of the logging client the same spot we check the control client is not completed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1594751875
Thanks, just back from travel, taking a look now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle closed pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle closed pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
URL: https://github.com/apache/beam/pull/25186
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1235538475
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -117,75 +188,216 @@ public BeamFnLoggingClient(
rootLogger.removeHandler(handler);
}
// configure loggers from default sdk harness log level and log level overrides
- this.configuredLoggers =
- SdkHarnessOptions.getConfiguredLoggerFromOptions(options.as(SdkHarnessOptions.class));
+ this.configuredLoggers.addAll(SdkHarnessOptions.getConfiguredLoggerFromOptions(options));
- BeamFnLoggingGrpc.BeamFnLoggingStub stub = BeamFnLoggingGrpc.newStub(channel);
- inboundObserver = new LogControlObserver();
- logRecordHandler = new LogRecordHandler();
- logRecordHandler.setLevel(Level.ALL);
- logRecordHandler.setFormatter(DEFAULT_FORMATTER);
- logRecordHandler.executeOn(options.as(ExecutorOptions.class).getScheduledExecutorService());
- boolean logMdc = options.as(SdkHarnessOptions.class).getLogMdc();
- logRecordHandler.setLogMdc(logMdc);
- QuotaEvent.setEnabled(logMdc);
- outboundObserver = (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver);
+ // Install a handler that queues to the buffer read by the background thread.
rootLogger.addHandler(logRecordHandler);
}
- public void setProcessBundleHandler(ProcessBundleHandler processBundleHandler) {
- this.processBundleHandler = processBundleHandler;
+ private static class StreamWriter {
+ private final ManagedChannel channel;
+ private final StreamObserver<BeamFnApi.LogEntry.List> outboundObserver;
+ private final LogControlObserver inboundObserver;
+
+ private final CompletableFuture<Object> inboundObserverCompletion;
+ private final AdvancingPhaser streamPhaser;
+
+ // Used to note we are attempting to close the logging client and to gracefully drain the
+ // current logs to the stream.
+ private final CompletableFuture<Object> softClosing = new CompletableFuture<>();
+
+ public StreamWriter(ManagedChannel channel) {
+ this.inboundObserverCompletion = new CompletableFuture<>();
+ this.streamPhaser = new AdvancingPhaser(1);
+ this.channel = channel;
+
+ BeamFnLoggingGrpc.BeamFnLoggingStub stub = BeamFnLoggingGrpc.newStub(channel);
+ this.inboundObserver = new LogControlObserver();
+ this.outboundObserver =
+ new DirectStreamObserver<BeamFnApi.LogEntry.List>(
+ this.streamPhaser,
+ (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver));
+ }
+
+ public void drainQueueToStream(BlockingQueue<BeamFnApi.LogEntry> bufferedLogEntries) {
+ Throwable thrown = null;
+ try {
+ List<BeamFnApi.LogEntry> additionalLogEntries =
+ new ArrayList<>(MAX_BUFFERED_LOG_ENTRY_COUNT);
+ // As long as we haven't yet terminated the stream, then attempt to send on it.
+ while (!streamPhaser.isTerminated()) {
+ // We wait for a limited period so that we can evaluate if the stream closed or if
+ // we are gracefully closing the client.
+ BeamFnApi.LogEntry logEntry = bufferedLogEntries.poll(1, SECONDS);
+ if (logEntry == null) {
+ if (softClosing.isDone()) {
+ break;
+ }
+ continue;
+ }
+
+ // Batch together as many log messages as possible that are held within the buffer
+ BeamFnApi.LogEntry.List.Builder builder =
+ BeamFnApi.LogEntry.List.newBuilder().addLogEntries(logEntry);
+ bufferedLogEntries.drainTo(additionalLogEntries);
+ builder.addAllLogEntries(additionalLogEntries);
+ outboundObserver.onNext(builder.build());
+ additionalLogEntries.clear();
+ }
+ if (inboundObserverCompletion.isDone()) {
+ try {
+ // If the inbound observer failed with an exception, get() will throw an
+ // ExecutionException.
+ inboundObserverCompletion.get();
+ // Otherwise it is an error for the server to close the stream before we closed our end.
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly with success before it was closed by the client.");
+ } catch (ExecutionException e) {
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly before it was closed by the client with error: "
+ + e.getCause());
+ } catch (InterruptedException e) {
+ // Should never happen because of the isDone check.
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+ } catch (Throwable t) {
+ thrown = t;
+ throw new RuntimeException(t);
+ } finally {
+ if (thrown == null) {
+ outboundObserver.onCompleted();
+ } else {
+ outboundObserver.onError(thrown);
+ }
+ channel.shutdown();
+ boolean shutdownFinished = false;
+ try {
+ shutdownFinished = channel.awaitTermination(10, SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ if (!shutdownFinished) {
+ channel.shutdownNow();
+ }
+ }
+ }
+ }
+
+ public void softClose() {
+ softClosing.complete(COMPLETED);
+ }
+
+ public void hardClose() {
+ streamPhaser.forceTermination();
+ }
+
+ private class LogControlObserver
+ implements ClientResponseObserver<BeamFnApi.LogEntry, BeamFnApi.LogControl> {
+
+ @Override
+ public void beforeStart(ClientCallStreamObserver<BeamFnApi.LogEntry> requestStream) {
+ requestStream.setOnReadyHandler(streamPhaser::arrive);
+ }
+
+ @Override
+ public void onNext(BeamFnApi.LogControl value) {}
+
+ @Override
+ public void onError(Throwable t) {
+ inboundObserverCompletion.completeExceptionally(t);
+ hardClose();
+ }
+
+ @Override
+ public void onCompleted() {
+ inboundObserverCompletion.complete(COMPLETED);
+ hardClose();
+ }
+ }
}
@Override
+ @SuppressWarnings("nullness")
Review Comment:
For some reason with the Preconditions.checkNotNull I still got errors. Interestingly I tried casting explicitly to @NonNull afterwards and it said both that it wasn't safe and that it was unnecessary :P
I tried going to CompletableFuture<Object> instead of CompletableFuture<?> in case that was messing it up but it didn't help.
I was able to get it to work by removing the Nullable by initializing it in the constructor instead of via the helper start method. To do so I had to add some additional underinitialization annotations.
PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "lukecwik (via GitHub)" <gi...@apache.org>.
lukecwik commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1424963988
>
Reconnect no, just swap log managers.
The question is whether we should continue to do work or not in this unhealthy state. I would say no and would rather exit having the SDK restart and reconnect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1098682773
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -122,7 +124,9 @@ public BeamFnLoggingClient(
logRecordHandler.setLevel(Level.ALL);
logRecordHandler.setFormatter(DEFAULT_FORMATTER);
logRecordHandler.executeOn(options.as(ExecutorOptions.class).getScheduledExecutorService());
- outboundObserver = (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver);
+ outboundObserver =
+ new DirectStreamObserver<BeamFnApi.LogEntry.List>(
+ phaser, (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver));
Review Comment:
I was thinking the logging for slow write would be nice to show that it is falling behind due to runner not consuming.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "lukecwik (via GitHub)" <gi...@apache.org>.
lukecwik commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1099015350
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -122,7 +124,9 @@ public BeamFnLoggingClient(
logRecordHandler.setLevel(Level.ALL);
logRecordHandler.setFormatter(DEFAULT_FORMATTER);
logRecordHandler.executeOn(options.as(ExecutorOptions.class).getScheduledExecutorService());
- outboundObserver = (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver);
+ outboundObserver =
+ new DirectStreamObserver<BeamFnApi.LogEntry.List>(
+ phaser, (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver));
Review Comment:
Checked performance impact, seems like we get a net win since we do the readiness checking less often
before:
```
Benchmark Mode Cnt Score Error Units
BeamFnLoggingClientBenchmark.testLogging thrpt 15 415642.810 ± 17448.195 ops/s
BeamFnLoggingClientBenchmark.testLoggingWithAllOptionalParameters thrpt 15 380592.277 ± 8459.947 ops/s
```
after:
```
Benchmark Mode Cnt Score Error Units
BeamFnLoggingClientBenchmark.testLogging thrpt 15 418828.259 ± 15726.823 ops/s
BeamFnLoggingClientBenchmark.testLoggingWithAllOptionalParameters thrpt 15 402892.740 ± 10908.775 ops/s
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Abacn commented on a diff in pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #25186:
URL: https://github.com/apache/beam/pull/25186#discussion_r1235381896
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -87,28 +93,93 @@ public class BeamFnLoggingClient implements AutoCloseable {
private static final Object COMPLETED = new Object();
+ private final Endpoints.ApiServiceDescriptor apiServiceDescriptor;
+
+ private final StreamWriter streamWriter;
+
+ private final LogRecordHandler logRecordHandler;
+
/* We need to store a reference to the configured loggers so that they are not
* garbage collected. java.util.logging only has weak references to the loggers
* so if they are garbage collected, our hierarchical configuration will be lost. */
- private final Collection<Logger> configuredLoggers;
- private final Endpoints.ApiServiceDescriptor apiServiceDescriptor;
- private final ManagedChannel channel;
- private final CallStreamObserver<BeamFnApi.LogEntry.List> outboundObserver;
- private final LogControlObserver inboundObserver;
- private final LogRecordHandler logRecordHandler;
- private final CompletableFuture<Object> inboundObserverCompletion;
- private final Phaser phaser;
+ private final Collection<Logger> configuredLoggers = new ArrayList<>();
+
private @Nullable ProcessBundleHandler processBundleHandler;
- public BeamFnLoggingClient(
+ private final BlockingQueue<LogEntry> bufferedLogEntries =
+ new ArrayBlockingQueue<>(MAX_BUFFERED_LOG_ENTRY_COUNT);
+
+ //
Review Comment:
possible leftover - empty comment line
##########
sdks/java/fn-execution/src/main/java/org/apache/beam/sdk/fn/stream/DirectStreamObserver.java:
##########
@@ -76,12 +77,13 @@ public void onNext(T value) {
// Record the initial phase in case we are in the inbound gRPC thread where the phase won't
// advance.
int initialPhase = phase;
- while (!outboundObserver.isReady()) {
+ // A negative phase indicates that the phaser is terminated.
+ while (phase >= 0 && !outboundObserver.isReady()) {
try {
phase = phaser.awaitAdvanceInterruptibly(phase, waitTime, TimeUnit.SECONDS);
} catch (TimeoutException e) {
totalTimeWaited += waitTime;
- waitTime = waitTime * 2;
+ waitTime = Math.min(waitTime * 2, 60);
Review Comment:
there is a hard coded constant here (60 s) possibly add some comments for its purpose?
##########
sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java:
##########
@@ -117,75 +188,216 @@ public BeamFnLoggingClient(
rootLogger.removeHandler(handler);
}
// configure loggers from default sdk harness log level and log level overrides
- this.configuredLoggers =
- SdkHarnessOptions.getConfiguredLoggerFromOptions(options.as(SdkHarnessOptions.class));
+ this.configuredLoggers.addAll(SdkHarnessOptions.getConfiguredLoggerFromOptions(options));
- BeamFnLoggingGrpc.BeamFnLoggingStub stub = BeamFnLoggingGrpc.newStub(channel);
- inboundObserver = new LogControlObserver();
- logRecordHandler = new LogRecordHandler();
- logRecordHandler.setLevel(Level.ALL);
- logRecordHandler.setFormatter(DEFAULT_FORMATTER);
- logRecordHandler.executeOn(options.as(ExecutorOptions.class).getScheduledExecutorService());
- boolean logMdc = options.as(SdkHarnessOptions.class).getLogMdc();
- logRecordHandler.setLogMdc(logMdc);
- QuotaEvent.setEnabled(logMdc);
- outboundObserver = (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver);
+ // Install a handler that queues to the buffer read by the background thread.
rootLogger.addHandler(logRecordHandler);
}
- public void setProcessBundleHandler(ProcessBundleHandler processBundleHandler) {
- this.processBundleHandler = processBundleHandler;
+ private static class StreamWriter {
+ private final ManagedChannel channel;
+ private final StreamObserver<BeamFnApi.LogEntry.List> outboundObserver;
+ private final LogControlObserver inboundObserver;
+
+ private final CompletableFuture<Object> inboundObserverCompletion;
+ private final AdvancingPhaser streamPhaser;
+
+ // Used to note we are attempting to close the logging client and to gracefully drain the
+ // current logs to the stream.
+ private final CompletableFuture<Object> softClosing = new CompletableFuture<>();
+
+ public StreamWriter(ManagedChannel channel) {
+ this.inboundObserverCompletion = new CompletableFuture<>();
+ this.streamPhaser = new AdvancingPhaser(1);
+ this.channel = channel;
+
+ BeamFnLoggingGrpc.BeamFnLoggingStub stub = BeamFnLoggingGrpc.newStub(channel);
+ this.inboundObserver = new LogControlObserver();
+ this.outboundObserver =
+ new DirectStreamObserver<BeamFnApi.LogEntry.List>(
+ this.streamPhaser,
+ (CallStreamObserver<BeamFnApi.LogEntry.List>) stub.logging(inboundObserver));
+ }
+
+ public void drainQueueToStream(BlockingQueue<BeamFnApi.LogEntry> bufferedLogEntries) {
+ Throwable thrown = null;
+ try {
+ List<BeamFnApi.LogEntry> additionalLogEntries =
+ new ArrayList<>(MAX_BUFFERED_LOG_ENTRY_COUNT);
+ // As long as we haven't yet terminated the stream, then attempt to send on it.
+ while (!streamPhaser.isTerminated()) {
+ // We wait for a limited period so that we can evaluate if the stream closed or if
+ // we are gracefully closing the client.
+ BeamFnApi.LogEntry logEntry = bufferedLogEntries.poll(1, SECONDS);
+ if (logEntry == null) {
+ if (softClosing.isDone()) {
+ break;
+ }
+ continue;
+ }
+
+ // Batch together as many log messages as possible that are held within the buffer
+ BeamFnApi.LogEntry.List.Builder builder =
+ BeamFnApi.LogEntry.List.newBuilder().addLogEntries(logEntry);
+ bufferedLogEntries.drainTo(additionalLogEntries);
+ builder.addAllLogEntries(additionalLogEntries);
+ outboundObserver.onNext(builder.build());
+ additionalLogEntries.clear();
+ }
+ if (inboundObserverCompletion.isDone()) {
+ try {
+ // If the inbound observer failed with an exception, get() will throw an
+ // ExecutionException.
+ inboundObserverCompletion.get();
+ // Otherwise it is an error for the server to close the stream before we closed our end.
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly with success before it was closed by the client.");
+ } catch (ExecutionException e) {
+ throw new IllegalStateException(
+ "Logging stream terminated unexpectedly before it was closed by the client with error: "
+ + e.getCause());
+ } catch (InterruptedException e) {
+ // Should never happen because of the isDone check.
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+ } catch (Throwable t) {
+ thrown = t;
+ throw new RuntimeException(t);
+ } finally {
+ if (thrown == null) {
+ outboundObserver.onCompleted();
+ } else {
+ outboundObserver.onError(thrown);
+ }
+ channel.shutdown();
+ boolean shutdownFinished = false;
+ try {
+ shutdownFinished = channel.awaitTermination(10, SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ if (!shutdownFinished) {
+ channel.shutdownNow();
+ }
+ }
+ }
+ }
+
+ public void softClose() {
+ softClosing.complete(COMPLETED);
+ }
+
+ public void hardClose() {
+ streamPhaser.forceTermination();
+ }
+
+ private class LogControlObserver
+ implements ClientResponseObserver<BeamFnApi.LogEntry, BeamFnApi.LogControl> {
+
+ @Override
+ public void beforeStart(ClientCallStreamObserver<BeamFnApi.LogEntry> requestStream) {
+ requestStream.setOnReadyHandler(streamPhaser::arrive);
+ }
+
+ @Override
+ public void onNext(BeamFnApi.LogControl value) {}
+
+ @Override
+ public void onError(Throwable t) {
+ inboundObserverCompletion.completeExceptionally(t);
+ hardClose();
+ }
+
+ @Override
+ public void onCompleted() {
+ inboundObserverCompletion.complete(COMPLETED);
+ hardClose();
+ }
+ }
}
@Override
+ @SuppressWarnings("nullness")
Review Comment:
there is `Preconditions.checkNotNull` below, is this SuppressWarnings still needed (same as `terminationFuture` below)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] scwhittle commented on pull request #25186: Ensure that the BeamFnLoggingClient terminates process if stream breaks
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on PR #25186:
URL: https://github.com/apache/beam/pull/25186#issuecomment-1594574004
Rebased to fix merge conflicts
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org