mattcasters commented on code in PR #8822: URL: https://github.com/apache/hop/pull/8822#discussion_r4229712937
########## plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageStallWatchdog.java: ########## @@ -0,0 +1,473 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.hop.vfs.gs; + +import java.io.IOException; +import java.time.Duration; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; +import org.apache.hop.core.Const; +import org.apache.hop.core.logging.LogChannel; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.vfs.gs.config.GoogleCloudConfig; + +/** + * Reports a {@code gs://} read or write that stopped making progress, and ends a read that stays + * stuck. + * + * <p>The storage client can keep a single read call busy indefinitely: it re-opens a stalled + * download over and over inside one call, ignoring the configured attempts and total timeout, so + * the code calling it learns nothing. Retries are logged by {@link LoggingStorageRetryStrategy}, + * but a transfer that does not move is only visible from outside the call. Each read or write marks + * itself busy while it is inside the client, and a single background thread looks at those every so + * often: a call that has been busy for longer than the warning period is logged, again after every + * further period, and once more if it gets going again. + * + * <p>A read that is still stuck after the time the retry settings allow is interrupted. The client + * does not retry an interrupted read, so the call ends and the stream fails with a clear error. + * Writes are never interrupted: the client uploads a whole 16 MB chunk inside one call, so a slow + * upload cannot be told apart from a stuck one, and is only reported. + */ +final class GoogleStorageStallWatchdog { + + private static final Class<?> PKG = GoogleStorageStallWatchdog.class; + + private static final GoogleStorageStallWatchdog INSTANCE = + new GoogleStorageStallWatchdog( + message -> LogChannel.GENERAL.logBasic("Google Cloud Storage: " + message), + System::nanoTime, + Duration.ofSeconds(1), + daemonInterrupter()); + + /** Where a stall is reported. */ + @FunctionalInterface + interface StallLog { + void report(String message); + } + + /** Whether a transfer reads or writes, which is all the messages need to know. */ + enum Direction { + READING("Reading"), + WRITING("Writing"); + + private final String key; + + Direction(String key) { + this.key = key; + } + } + + /** + * How long a call may go without progress before it is reported or, for a read, ended. + * + * @param readWarning a read busy for this long is reported as stalled + * @param readAbort a read busy for this long is ended; zero for never + * @param writeWarning a write busy for this long is reported as slow or stalled + */ + record Limits(Duration readWarning, Duration readAbort, Duration writeWarning) { + + /** Never report a stall sooner than this, whatever the read timeout. */ + private static final long MINIMUM_WARNING_SECONDS = 60; + + /** The client's upload chunk, sent whole inside one write call. Hop leaves it as it is. */ + private static final long UPLOAD_CHUNK_BYTES = 16L * 1024 * 1024; + + /** Below this an upload is too slow to tell from a stuck one. */ + private static final long SLOWEST_UPLOAD_BYTES_PER_SECOND = 100L * 1024; + + /** + * Derive the limits from the Google Cloud configuration. + * + * <ul> + * <li>A read is reported after three read timeouts without progress, at least a minute: a + * single timed-out attempt is already reported as a retry. + * <li>A read is ended after the time the retry settings allow for one call - every attempt + * running into both the connection and the read timeout, plus the delays between them - + * capped at the total timeout, and never before it was reported. Until then the client + * may still give up by itself, with its own error. + * <li>A write is reported once one upload chunk has taken longer than it would at the slowest + * rate we still call working, and not before a read would be. + * </ul> + * + * @param config the Google Cloud configuration + * @return the limits + */ + static Limits from(GoogleCloudConfig config) { + long readTimeout = timeoutSeconds(config.getReadTimeout()); + long connectTimeout = timeoutSeconds(config.getConnectionTimeout()); + Duration readWarning = Duration.ofSeconds(Math.max(MINIMUM_WARNING_SECONDS, 3 * readTimeout)); + + Duration readAbort = retryBudget(config, connectTimeout + readTimeout); + long totalTimeoutMinutes = Const.toLong(config.getTotalTimeout(), 50); + if (totalTimeoutMinutes > 0) { + Duration totalTimeout = Duration.ofMinutes(totalTimeoutMinutes); + if (readAbort.isZero() || readAbort.compareTo(totalTimeout) > 0) { + readAbort = totalTimeout; + } + } + if (!readAbort.isZero() && readAbort.compareTo(readWarning) < 0) { + readAbort = readWarning; Review Comment: **[bug]** The total-timeout cap is undone here. After `readAbort` is reduced to the configured total timeout, it is raised back to `readWarning` whenever that cap is shorter than three read timeouts (and at least 60 seconds). A stalled read then keeps going past the total timeout the user set. With a 120s read timeout and a 2 minute total timeout the give-up time becomes 6 minutes, not 2. The manual says this time is at most the total timeout, and `theTotalTimeoutCapsTheAbort` only covers a cap that is already above the warning. **Suggestion:** If the total timeout is shorter than the warning interval, lower the warning as well and abort at the total timeout (or abort at the total timeout even when no warning has been logged yet). Add a case where the total timeout is below `3 * readTimeout`. ########## plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageStallWatchdog.java: ########## @@ -0,0 +1,473 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.hop.vfs.gs; + +import java.io.IOException; +import java.time.Duration; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; +import org.apache.hop.core.Const; +import org.apache.hop.core.logging.LogChannel; +import org.apache.hop.i18n.BaseMessages; +import org.apache.hop.vfs.gs.config.GoogleCloudConfig; + +/** + * Reports a {@code gs://} read or write that stopped making progress, and ends a read that stays + * stuck. + * + * <p>The storage client can keep a single read call busy indefinitely: it re-opens a stalled + * download over and over inside one call, ignoring the configured attempts and total timeout, so + * the code calling it learns nothing. Retries are logged by {@link LoggingStorageRetryStrategy}, + * but a transfer that does not move is only visible from outside the call. Each read or write marks + * itself busy while it is inside the client, and a single background thread looks at those every so + * often: a call that has been busy for longer than the warning period is logged, again after every + * further period, and once more if it gets going again. + * + * <p>A read that is still stuck after the time the retry settings allow is interrupted. The client + * does not retry an interrupted read, so the call ends and the stream fails with a clear error. + * Writes are never interrupted: the client uploads a whole 16 MB chunk inside one call, so a slow + * upload cannot be told apart from a stuck one, and is only reported. + */ +final class GoogleStorageStallWatchdog { + + private static final Class<?> PKG = GoogleStorageStallWatchdog.class; + + private static final GoogleStorageStallWatchdog INSTANCE = + new GoogleStorageStallWatchdog( + message -> LogChannel.GENERAL.logBasic("Google Cloud Storage: " + message), + System::nanoTime, + Duration.ofSeconds(1), + daemonInterrupter()); + + /** Where a stall is reported. */ + @FunctionalInterface + interface StallLog { + void report(String message); + } + + /** Whether a transfer reads or writes, which is all the messages need to know. */ + enum Direction { + READING("Reading"), + WRITING("Writing"); + + private final String key; + + Direction(String key) { + this.key = key; + } + } + + /** + * How long a call may go without progress before it is reported or, for a read, ended. + * + * @param readWarning a read busy for this long is reported as stalled + * @param readAbort a read busy for this long is ended; zero for never + * @param writeWarning a write busy for this long is reported as slow or stalled + */ + record Limits(Duration readWarning, Duration readAbort, Duration writeWarning) { + + /** Never report a stall sooner than this, whatever the read timeout. */ + private static final long MINIMUM_WARNING_SECONDS = 60; + + /** The client's upload chunk, sent whole inside one write call. Hop leaves it as it is. */ + private static final long UPLOAD_CHUNK_BYTES = 16L * 1024 * 1024; + + /** Below this an upload is too slow to tell from a stuck one. */ + private static final long SLOWEST_UPLOAD_BYTES_PER_SECOND = 100L * 1024; + + /** + * Derive the limits from the Google Cloud configuration. + * + * <ul> + * <li>A read is reported after three read timeouts without progress, at least a minute: a + * single timed-out attempt is already reported as a retry. + * <li>A read is ended after the time the retry settings allow for one call - every attempt + * running into both the connection and the read timeout, plus the delays between them - + * capped at the total timeout, and never before it was reported. Until then the client + * may still give up by itself, with its own error. + * <li>A write is reported once one upload chunk has taken longer than it would at the slowest + * rate we still call working, and not before a read would be. + * </ul> + * + * @param config the Google Cloud configuration + * @return the limits + */ + static Limits from(GoogleCloudConfig config) { + long readTimeout = timeoutSeconds(config.getReadTimeout()); + long connectTimeout = timeoutSeconds(config.getConnectionTimeout()); + Duration readWarning = Duration.ofSeconds(Math.max(MINIMUM_WARNING_SECONDS, 3 * readTimeout)); + + Duration readAbort = retryBudget(config, connectTimeout + readTimeout); + long totalTimeoutMinutes = Const.toLong(config.getTotalTimeout(), 50); + if (totalTimeoutMinutes > 0) { + Duration totalTimeout = Duration.ofMinutes(totalTimeoutMinutes); + if (readAbort.isZero() || readAbort.compareTo(totalTimeout) > 0) { + readAbort = totalTimeout; + } + } + if (!readAbort.isZero() && readAbort.compareTo(readWarning) < 0) { + readAbort = readWarning; + } + + long chunkSeconds = + (UPLOAD_CHUNK_BYTES + SLOWEST_UPLOAD_BYTES_PER_SECOND - 1) + / SLOWEST_UPLOAD_BYTES_PER_SECOND; + Duration writeWarning = Duration.ofSeconds(Math.max(readWarning.toSeconds(), chunkSeconds)); + + return new Limits(readWarning, readAbort, writeWarning); + } + + /** A configured timeout in seconds; none at all counts as the default, to size the limits. */ + private static long timeoutSeconds(String configured) { + long seconds = Const.toInt(configured, 20); + return seconds > 0 ? seconds : 20; Review Comment: **[suggestion]** A configured timeout of `0` is replaced with 20 seconds here, but `GoogleStorageFileSystem.buildTransportOptions` passes `0` milliseconds through, and `HttpTransportOptions` treats 0 as an infinite socket timeout (a negative value is the 20s default). A read timeout of `0` therefore never fires in the client, while the stall budget is sized as a 20s timeout, so the watchdog gives up much sooner than the socket timeout that was configured. A blank value is already the 20s default via `Const.toInt`. **Suggestion:** Leave `0` as "no socket timeout" and, in that case, bound the abort with the total timeout alone instead of inventing a 20s attempt length. ########## plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java: ########## @@ -118,60 +149,82 @@ private void setServiceAccountCredentials( credentials = tryApplicationDefaultCredentials(); break; } + if (credentials == null) { + builder.setCredentialsProblem( + newFileSystemOptions, + "Google Storage connection '" + + scheme + + "' uses Application Default Credentials, which are not available (" + + applicationDefaultProblem + + ")"); + return; + } } - if (credentials != null) { - GoogleStorageFileSystemConfigBuilder.getInstance() - .setGoogleCredentials(newFileSystemOptions, credentials); - } - GoogleStorageFileSystemConfigBuilder.getInstance().setSchema(newFileSystemOptions, scheme); + builder.setGoogleCredentials(newFileSystemOptions, credentials); } catch (Exception e) { - // Do not log error for the default GS account - if (googleStorageMetadataType != null) { - LogChannel.GENERAL.logError( - "Unable to set service account credentials for vfs name: " - + googleStorageMetadataType.getName(), - e); - } + // Only a named connection gets here: it is configured on purpose, so say so right away. + LogChannel.GENERAL.logError( + "Google Cloud Storage: Unable to set service account credentials for vfs name: " + scheme, + e); + builder.setCredentialsProblem( + newFileSystemOptions, + "the credentials of Google Storage connection '" + + scheme + + "' could not be loaded (" + + describe(e) + + ")"); } catch (LinkageError e) { // NoClassDefFoundError (e.g. io.grpc.Context) must not prevent HopEnvironment.init - if (googleStorageMetadataType != null) { + if (defaultScheme) { + LogChannel.GENERAL.logDetailed( + "Google Cloud Storage: Application Default Credentials unavailable (" + + describe(e) + + "). gs:// will work after configuring a service account key."); + builder.setCredentialsProblem( Review Comment: **[bug]** A `LinkageError` while reading the configured service-account key file is not handled by the inner `catch (Exception)` around `fromStream` (line 111). `tryApplicationDefaultCredentials()` already swallows `LinkageError`, so for the default scheme this branch is what a key-file linkage failure hits, but the message says Application Default Credentials are missing and tells the user to set a key file they already set. The first `gs://` use then fails with the wrong reason — the failure mode this change is meant to make clear. **Suggestion:** Catch `Exception | LinkageError` in the key-file `try` and keep the key-file wording. Reserve this branch for a real Application Default Credentials failure. ########## plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/WriteChannelOutputStream.java: ########## @@ -100,7 +121,14 @@ public void close() throws IOException { WriteChannel ch = channel; if (ch != null) { channel = null; - ch.close(); + // The last part of the upload happens here. + transfer.begin(); Review Comment: **[bug]** `close()` calls `begin()`/`end()` on the same `Transfer` that `write()` may already be inside. `close()` deliberately does not take `writeLock`, so the two overlap on a pipeline stop and also in the normal case where `close()` waits for the in-flight `write()` and then sends the last chunk. `Transfer` has no in-flight count: the second `begin()` resets `busySince`, `thread`, and `stalled`, and whichever `end()` runs first removes the transfer from the busy set. The call still inside the client — often `close()`, which is where the last upload chunk is sent — is no longer watched, so a stalled upload goes silent. **Suggestion:** Reference-count calls that are inside the client and stay busy until the last `end()`. Do not reset `busySince` while a call is already in progress, and do not let `write()`'s `end()` unregister a `close()` that has already begun. ########## plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ReadChannelInputStream.java: ########## @@ -68,16 +90,32 @@ public int read(byte[] buf, int off, int len) throws IOException { bytes.limit(len); } int res; - try { + transfer.begin(); Review Comment: **[suggestion]** `transfer.begin()` is not paired with `end()` in a `finally`. An `Error`, or any other throwable that is neither `IOException` nor `RuntimeException`, leaves the transfer in the busy set with `thread` still set, so a later check can interrupt that thread after it has moved on. On the paths that do call `end()`, a hard failure after a stall warning is also logged as "Reading … resumed", because `end()` treats every non-abort return as the transfer getting going again. **Suggestion:** Always `end()` from a `finally`, and only log the resumed message when the call actually returned data. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
