xmg333 commented on code in PR #18463:
URL:
https://github.com/apache/dolphinscheduler/pull/18463#discussion_r3737094147
##########
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/logging/LogClientDelegate.java:
##########
@@ -109,4 +114,90 @@ private boolean checkNodeExists(TaskInstance taskInstance)
{
return exists;
}
+ /**
+ * Stream the entire task instance log to {@code outputStream} using
chunked RPC.
+ *
+ * <p>Strategy:
+ * <ul>
+ * <li>If the worker node is gone, read straight from remote log storage
(archive).</li>
+ * <li>Otherwise try the chunk RPC. If it fails before any byte is
written (e.g. an old worker
+ * that does not implement {@code getTaskInstanceLogFileChunk}),
fall back to the legacy
+ * whole-file worker RPC; if that also fails, fall back to remote
storage.</li>
+ * <li>If the failure happens mid-stream (bytes already written), throw
IOException to avoid
+ * corrupting the download.</li>
+ * </ul>
+ */
+ public void streamWholeLog(final TaskInstance taskInstance, final
OutputStream outputStream) throws IOException {
+ checkArgs(taskInstance);
+ if (!checkNodeExists(taskInstance)) {
+ writeRemoteLegacy(taskInstance, outputStream);
+ return;
+ }
+ long offset = 0;
+ try {
+ while (true) {
+ final TaskInstanceLogFileDownloadResponse chunk =
+ localLogClient.getLogChunk(taskInstance, offset,
LOG_CHUNK_SIZE);
+ if (chunk.getCode() != LogResponseStatus.SUCCESS) {
+ if (offset == 0) {
+ log.warn("First chunk failed for task instance {}: {},
falling back to legacy whole-file RPC",
+ taskInstance.getId(), chunk.getMessage());
+ writeLocalLegacy(taskInstance, outputStream);
+ return;
+ }
+ throw new IOException("Worker chunk failed at offset " +
offset + ": " + chunk.getMessage());
+ }
+ final byte[] data = chunk.getLogBytes();
+ if (data != null && data.length > 0) {
+ outputStream.write(data);
+ offset += data.length;
+ }
+ if (chunk.isEof() || (data == null || data.length == 0)) {
+ return;
+ }
+ }
+ } catch (Exception e) {
+ if (offset == 0) {
+ log.warn("Chunked streaming failed before any byte written for
task instance {}, "
+ + "falling back to legacy whole-file RPC",
taskInstance.getId(), e);
+ writeLocalLegacy(taskInstance, outputStream);
+ return;
+ }
+ throw new IOException("Log streaming failed at offset " + offset,
e);
+ }
+ }
+
+ /**
+ * Fall back to the legacy whole-file worker RPC ({@code
getTaskInstanceWholeLogFileBytes}),
+ * which every worker — including old ones built before the chunk RPC
existed — implements. If
+ * that returns nothing or fails, fall through to remote log storage as a
last resort.
+ */
+ private void writeLocalLegacy(final TaskInstance taskInstance,
+ final OutputStream outputStream) throws
IOException {
+ try {
+ final TaskInstanceLogFileDownloadResponse response =
localLogClient.getWholeLog(taskInstance);
+ if (response != null && response.getCode() ==
LogResponseStatus.SUCCESS) {
+ final byte[] bytes = response.getLogBytes();
+ if (bytes != null && bytes.length > 0) {
+ outputStream.write(bytes);
+ outputStream.flush();
+ return;
+ }
+ }
+ } catch (Exception e) {
+ log.warn("Legacy whole-file RPC failed for task instance {},
falling back to remote storage",
+ taskInstance.getId(), e);
+ }
+ writeRemoteLegacy(taskInstance, outputStream);
+ }
+
+ private void writeRemoteLegacy(final TaskInstance taskInstance,
+ final OutputStream outputStream) throws
IOException {
+ final byte[] bytes = remoteLogClient.getWholeLog(taskInstance);
+ if (bytes != null && bytes.length > 0) {
+ outputStream.write(bytes);
+ }
+ outputStream.flush();
Review Comment:
Thanks for catching this. Fixed in LogClientDelegate.writeRemoteLegacy: when
remoteLogClient.getWholeLog(...) returns null/empty (remote archive missing),
it now throws IOException instead of flushing an empty body. The exception
propagates through streamWholeLog → StreamingResponseBody and aborts the
response, so a missing log is no longer reported as a successful HTTP 200
download.
While reviewing the fix I also found and fixed a related gap:
LoggerServiceImpl.checkDownloadLogAuth validated host but not logPath. It's now
fixed and would return a clear error before streaming starts.
--
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]
