det101 commented on code in PR #18463:
URL:
https://github.com/apache/dolphinscheduler/pull/18463#discussion_r3733184331
##########
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:
If remote returns null/empty (archive missing), this still flush()es and the
download ends as HTTP 200 with only the log header. Please throw when bytes are
absent so a missing remote log is not reported as a successful download.
--
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]
