[1fanwang]

When either Edge log route referenced a missing task instance, the API returned HTTP 500 instead of treating it as a missing resource. Both routes use the same path lookup, so the failure affected log-path reads and log uploads.

The shared lookup now returns HTTP 404. Both route contracts document that response, and authenticated route tests cover the GET and POST paths.

Testing Done

Scenario	Result
Missing task instance on the log-path route	HTTP 500 before, HTTP 404 after
Missing task instance on the log-upload route	HTTP 500 before, HTTP 404 after

Against current main before this change:

breeze run python dev/pr72908_log_routes_repro.py --expected-status 500

GET /logs/logfile_path/missing_dag/missing_task/manual__missing/1/-1 -> 500
Internal Server Error
POST /logs/push/missing_dag/missing_task/manual__missing/1/-1 -> 500
Internal Server Error

With this branch:

breeze run python dev/pr72908_log_routes_repro.py --expected-status 404

GET /logs/logfile_path/missing_dag/missing_task/manual__missing/1/-1 -> 404
{"detail":"TaskInstance not found for dag_id=missing_dag, task_id=missing_task, run_id=manual__missing, map_index=-1"}
POST /logs/push/missing_dag/missing_task/manual__missing/1/-1 -> 404
{"detail":"TaskInstance not found for dag_id=missing_dag, task_id=missing_task, run_id=manual__missing, map_index=-1"}

Reproducer source: dev/pr72908_log_routes_repro.py

# 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.

from __future__ import annotations

import argparse
import json
import socket
import threading
import time
from datetime import UTC, datetime
from urllib.error import HTTPError
from urllib.request import Request, urlopen

import uvicorn
from fastapi import FastAPI

from airflow.api_fastapi.auth.tokens import JWTGenerator
from airflow.providers.edge3.worker_api.auth import jwt_validator
from airflow.providers.edge3.worker_api.routes.logs import logs_router
from airflow.utils import db

from tests_common.test_utils.config import conf_vars

SECRET = "edge-log-e2e-secret-0000000000000000000000000000000000000000000000000000000000000000"
DAG_ID = "missing_dag"
TASK_ID = "missing_task"
RUN_ID = "manual__missing"


def request(method: str, url: str, token: str, body: dict[str, str] | None = None) -> tuple[int, str]:
    data = json.dumps(body).encode() if body else None
    headers = {"Authorization": token}
    if data:
        headers["Content-Type"] = "application/json"
    try:
        with urlopen(Request(url, data=data, headers=headers, method=method), timeout=10) as response:
            return response.status, response.read().decode()
    except HTTPError as error:
        return error.code, error.read().decode()


def run_requests(expected_status: int) -> None:
    db.resetdb()

    app = FastAPI()
    app.include_router(logs_router, prefix="/edge_worker/v1")
    server_socket = socket.socket()
    server_socket.bind(("127.0.0.1", 0))
    port = server_socket.getsockname()[1]
    server = uvicorn.Server(uvicorn.Config(app, log_level="critical"))
    thread = threading.Thread(target=server.run, kwargs={"sockets": [server_socket]}, daemon=True)
    thread.start()
    for _ in range(100):
        if server.started:
            break
        time.sleep(0.05)
    else:
        raise RuntimeError("Uvicorn did not start")

    generator = JWTGenerator(secret_key=SECRET, valid_for=60, audience="api")
    base_url = f"http://127.0.0.1:{port}/edge_worker/v1"
    path = f"{DAG_ID}/{TASK_ID}/{RUN_ID}/1/-1"
    requests = [
        ("GET", f"logs/logfile_path/{path}", None),
        (
            "POST",
            f"logs/push/{path}",
            {"log_chunk_data": "missing task log", "log_chunk_time": datetime.now(UTC).isoformat()},
        ),
    ]
    statuses: list[int] = []
    try:
        for method, route, body in requests:
            status_code, response_body = request(
                method=method,
                url=f"{base_url}/{route}",
                token=generator.generate(extras={"method": route}),
                body=body,
            )
            statuses.append(status_code)
            print(f"{method} /{route} -> {status_code}")
            print(response_body)
    finally:
        server.should_exit = True
        thread.join(timeout=10)
        jwt_validator.cache_clear()

    expected_statuses = [expected_status, expected_status]
    if statuses != expected_statuses:
        raise AssertionError(f"Expected {expected_statuses}, got {statuses}")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--expected-status", type=int, required=True)
    args = parser.parse_args()
    with conf_vars({("api_auth", "jwt_secret"): SECRET}):
        jwt_validator.cache_clear()
        try:
            run_requests(expected_status=args.expected_status)
        finally:
            jwt_validator.cache_clear()


if __name__ == "__main__":
    main()

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: GitHub Copilot CLI (GPT-6 Astra and GPT-5.6 Sol) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[dheerajturaga]

Thanks for the fix. Could you please address these before merging?

• Rerun or investigate the failing documentation CI job so the PR is green.
• _logfile_path() is also used by push_logs(), so this change makes missing task instances return 404 from that endpoint too. Please document 404 for the push route, regenerate the OpenAPI spec, and add a regression test—or scope the 404 handling to the GET route only.

---

Drafted-by: Codex (GPT-5); reviewed by @dheerajturaga before posting

[1fanwang]

Please document 404 for the push route, regenerate the OpenAPI spec, and add a regression test

@dheerajturaga done in e6633fb. Both log routes document 404, the generated OpenAPI spec includes both responses, and authenticated regression tests cover GET and POST. The current documentation build, spellcheck and publication checks all pass.