neoLsH commented on issue #12716: URL: https://github.com/apache/gravitino/issues/12716#issuecomment-5525513122
> Hi [@neoLsH](https://github.com/neoLsH) thanks a lot. I would like to know more about how you design this from end to end, like SDK/REST API design, internal logic design, so that I can better evaluate your proposal. @jerryshao thanks. Before I write any code let me lay out how I read this, so you can correct the direction early if I've got it wrong. **What I understand the task to be.** `JobExecutor` today exposes only `submitJob` / `getJobStatus` / `cancelJob`, so neither callers nor the REST API can see a job's stdout or stderr, which makes a FAILED job hard to debug. The ask is to add output-log retrieval to the SPI as a backward-compatible default method, implement it for the built-in local runner using output it already captures, and surface it through `JobHandle` in Java and Python plus the REST `Job` payload in `docs/open-api/jobs.yaml` — the same cross-layer pattern as `62c5a2f2b`. **My direction: this is a read-through, not a new persisted field.** That came out of reading the local runner. `JobManager.runJob` already creates a staging directory per run at `{stagingDir}/{metalake}/{jobTemplateName}/job-{id}` and materializes the resolved executable into it; `LocalProcessBuilder` takes the executable's parent as its working directory (`LocalProcessBuilder.java:37`), which is that staging directory; and `ShellProcessBuilder.java:54-57` / `SparkProcessBuilder.java:121-125` already redirect stdout and stderr to `output.log` and `error.log` there. I specifically checked whether two runs of one template could share a file, since the directory is derived from the executable path rather than the job id — they can't, because the executable itself is per job. So the files exist and are already uniquely keyed per job. That's the meaningful difference from `62c5a2f2b`: queuedAt and startedAt had to be stored, which is why that commit touched `JobPO`, `JobEntity`, the SQL providers and the h2/mysql/postgresql schemas plus upgrade scripts. Output logs need none of that — no storage change, no migration, no schema diff. **Preliminary approach:** 1. SPI and value type. `JobExecutor` gains `retrieveJobOutputLogs(String jobExecutionId)` as a default method throwing `UnsupportedOperationException`, matching the `queuedAt`/`startedAt`/`finishedAt` shape so existing implementers keep compiling. `JobOutputLogs` (stdout, stderr, truncated) goes in the api module next to `JobHandle`, since `JobHandle` returns it and both core and clients/client-java depend on api. On scope: `LocalJobExecutor` is the only implementation in the repo — Airflow and Livy are listed under Future Work in `docs/manage-jobs-in-gravitino.md`, and Airflow otherwise appears just as a `com.example.MyAirflowJobExecutor` placeholder in `docs/development/custom-job-executor.md`. So the default method is there for out-of-tree executors, and I'd implement it only for the local runner. 2. Local executor. This is the only place needing real plumbing, because `runJob` doesn't retain the working directory and `runningProcesses` is cleared on completion, so nothing survives to look up later. I'd keep a `Map<String, File> jobWorkingDirs` under the existing lock, populated in `submitJob` where the resolved `JobTemplate` is still in hand, deriving the directory through a small helper on `LocalProcessBuilder` so the logic isn't duplicated, and evicted alongside `jobStatus` on `jobStatusKeepTimeInMs`. It has to be retained rather than recomputed because the executor is keyed by `jobExecutionId` (`local-job-{uuid}`), not the `job-{id}` in the staging path. A missing log file reads as an empty string, not an error. 3. Server side. `JobManager` implements `JobOperationDispatcher`, so both get a method that resolves the entity to its `jobExecutionId` and delegates, catching `UnsupportedOperationException` / `NoSuchJobException` / `IOException` and returning null. That catch matters more than it looks: `getJob` is a pure `entityStore` read today (`JobManager.java:405`) and the executor's in-memory maps don't survive a restart, so a propagated failure would start turning valid metadata GETs into 404s after a restart. Log retrieval should never fail the job GET. `JobOperations.getJob` then composes the two dispatcher calls into a new `toDTO(entity, logs)` overload, leaving the existing `toDTO` and `toJobDTOs` untouched, and `JobDTO` grows from 8 constructor args to 9 — the same mechanical step as before. 4. SDKs and spec. `JobHandle.outputLogs()` is another default-throwing method; `GenericJobHandle` returns it straight from the DTO. Note `GenericJobHandle` holds only a `JobDTO` and no REST client, so it can't fetch logs on demand — the value has to arrive in the payload. Python mirrors all of it (`JobOutputLogs` under `gravitino/api/job/`, `output_logs()` raising `NotImplementedError` like its siblings, `_output_logs` defaulting to None so old servers and new clients interoperate). The `Job` schema gains one nullable `outputLogs` object. **Two calls I made that I'd like you to check**, since they're the ones that change the shape of the API: - Populate only on `GET runs/{jobId}`, not on the list. The list can return many jobs, so filling logs there means two file reads per job on an endpoint that is a pure storage query today, and it would push an executor dependency into `toJobDTOs`. The field is nullable and simply absent in list responses, so nothing breaks. The alternative is a dedicated `GET runs/{jobId}/logs`, which separates concerns better and leaves room for range parameters later, but the issue asks for this on `JobHandle` and the `Job` payload. I went with the payload and can switch if you prefer the sub-resource. - Return a bounded tail rather than the whole file, since a runaway job can produce gigabytes. New `LocalJobExecutorConfigs.OUTPUT_LOG_MAX_BYTES`, default 256 KiB per stream, so the user-facing key is `gravitino.jobExecutor.local.outputLogMaxBytes` and it's read in `LocalJobExecutor.initialize()` exactly like `waitingQueueSize` and `jobStatusKeepTimeInMs`. I put it there rather than in `Configs` because the cap is purely a local-runner concern — it's the only executor reading files off disk — and the `gravitino.jobExecutor.<name>.*` prefix is the mechanism the factory already strips and hands to `initialize()`. I'd add a row to the config table in `docs/manage-jobs-in-gravitino.md` to match. Reading is via `RandomAccessFile.seek` cut on a UTF-8 character boundary, with `truncated: true` marking partial content. Tail rather than head because for a FAILED job the stack trace is at the end. Related: I'd keep null and empty distinct — null means this response carries no logs, `s tdout: ""` means the job genuinely produced nothing, which is normal while QUEUED. Collapsing them makes an empty successful job look like an unsupported executor. One thing that makes this safe without an expiry state on the wire: `cleanUpStagingDirs()` deletes the `JobEntity` and the staging directory in the same pass, gated on terminal status plus `finishedAt + jobStagingDirKeepTimeInMs` (`gravitino.job.stagingDirKeepTimeInMs`, default 7 days), so the record and its logs disappear together and there's no window where `getJob` succeeds but the logs are gone. Does this match what you had in mind? If the direction is right I'll write it up with the full file list and test plan, and I'm happy to split it into two PRs — SPI plus local executor first, then the SDK and REST surface. -- 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]
