feat(build_stream): complete ER-BSM-003 log observability implementation

src/build_stream/app/api/jobs/routes.py

@@ -18,6 +18,7 @@
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException, Response, status
+from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import JSONResponse
 
 from core.cleanup.exceptions import (
@@ -43,10 +44,12 @@
 )
 from orchestrator.jobs.commands import CreateJobCommand
 from orchestrator.jobs.use_cases import CreateJobUseCase
+from orchestrator.logs.use_cases import StageLogMetadataService
 
 from api.cleanup.dependencies import get_cleanup_job_use_case
 from api.dependencies import verify_token
 from api.logging_utils import create_job_log_file, log_secure_info, remove_job_logger
+from api.logs.dependencies import get_stage_log_metadata_service
 from api.jobs.dependencies import (
     get_audit_repo,
     get_correlation_id,
@@ -238,6 +241,7 @@ async def get_job(
     job_repo = Depends(get_job_repo),
     stage_repo = Depends(get_stage_repo),
     audit_repo = Depends(get_audit_repo),
+    stage_log_metadata: StageLogMetadataService = Depends(get_stage_log_metadata_service),
 ) -> GetJobResponse:
     """Return a job if it exists for the requesting client."""
 
@@ -278,7 +282,10 @@ async def get_job(
         # Get stage breakdown — include all stages (architecture filtering
         # is handled by image_build_manager domain playbook directly)
         stages_entities = stage_repo.find_all_by_job(validated_job_id)  # pylint: disable=no-member
-
+        log_metadata = await run_in_threadpool(
+            lambda: [stage_log_metadata.describe(str(validated_job_id), s) for s in stages_entities]
+        )
+
         stages = [
             GetStageResponse(
                 stage_name=str(s.stage_name),
@@ -290,8 +297,10 @@ async def get_job(
                 attempt=s.attempt,
                 log_file_path=s.log_file_path,
                 result_detail=s.result_detail,
+                has_log=meta.has_log,
+                log_lines=meta.log_lines,
             )
-            for s in stages_entities
+            for s, meta in zip(stages_entities, log_metadata)
         ]
 
         # Get audit events for state change timestamps

src/build_stream/app/api/jobs/schemas.py

@@ -86,6 +86,13 @@ class GetStageResponse(BaseModel):
     attempt: int = Field(default=1, description="Stage attempt number (1-indexed, +1 per retry)")
     log_file_path: Optional[str] = Field(default=None, description="Ansible log file path on OIM host (NFS share)")
     result_detail: Optional[Dict[str, Any]] = Field(default=None, description="Detailed stage results (JSONB) including log_path, test_summary, artifact_dir")
+    has_log: bool = Field(
+        default=False,
+        description="True when the stage log can be fetched from GET /jobs/{job_id}/logs",
+    )
+    log_lines: int = Field(
+        default=0, description="Number of lines in the stage log (0 when has_log is false)"
+    )
 
 
 class CreateJobResponse(BaseModel):

src/build_stream/app/api/logs/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2026 Dell Inc. or its subsidiaries. All Rights Reserved.
+#
+# Licensed 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.
+
+"""Pipeline log query API (ER-BSM-003)."""

src/build_stream/app/api/logs/dependencies.py

@@ -0,0 +1,157 @@
+# Copyright 2026 Dell Inc. or its subsidiaries. All Rights Reserved.
+#
+# Licensed 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.
+
+# ****
+# Secure code generated utilizing Code Guardian.
+# Applicable rules used: CG-AUTH-001.6, CG-AUTH-001.9, CG-SECRETS-001.1
+# ****
+
+"""FastAPI dependency providers for the pipeline log query API."""
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Annotated, Iterator
+
+import yaml
+from fastapi import Depends, HTTPException, status
+
+from api.dependencies import get_job_repo, get_stage_repo, verify_token
+from api.logging_utils import log_secure_info
+from api.logs.limiter import (
+    ConcurrencyLimiter,
+    ConcurrencyLimitExceededError,
+    RateLimitExceededError,
+    SlidingWindowRateLimiter,
+)
+from common.config import LogQueryConfig, load_config, load_log_query_inputs
+from core.logs.redaction import RedactionService
+from infra.logs.filesystem_adapter import FileSystemLogAdapter
+from orchestrator.cleanup.retention_config import build_stream_config_path
+from orchestrator.logs.use_cases import (
+    LogQuerySettings,
+    LogQueryUseCase,
+    StageLogMetadataService,
+)
+
+
+@lru_cache(maxsize=1)
+def get_log_query_config() -> LogQueryConfig:
+    """Resolve the log query settings.
+
+    Precedence: ``build_stream_config.yml`` operator inputs, then ``[log_query]``
+    in build_stream.ini, then built-in defaults. Invalid operator inputs are
+    rejected as a whole and the ini/default values stay in effect.
+    """
+    base = LogQueryConfig()
+    try:
+        base = load_config().log_query
+    except FileNotFoundError:
+        log_secure_info("warning", "build_stream.ini not found; using log query defaults")
+    except ValueError:
+        log_secure_info(
+            "error", "Invalid log query configuration; using safe defaults", exc_info=True
+        )
+    try:
+        return load_log_query_inputs(base, build_stream_config_path())
+    except (OSError, ValueError, yaml.YAMLError):
+        log_secure_info(
+            "error",
+            "Invalid log observability inputs in build_stream_config.yml; "
+            "using ini/default log query settings",
+            exc_info=True,
+        )
+    return base
+
+
+@lru_cache(maxsize=1)
+def get_log_file_service() -> FileSystemLogAdapter:
+    """Provide the shared filesystem log adapter (keeps the line-count cache)."""
+    return FileSystemLogAdapter(Path(get_log_query_config().log_root))
+
+
+@lru_cache(maxsize=1)
+def get_redaction_service() -> RedactionService:
+    """Provide the shared redaction service."""
+    return RedactionService()
+
+
+@lru_cache(maxsize=1)
+def get_rate_limiter() -> SlidingWindowRateLimiter:
+    """Provide the per-client log query rate limiter."""
+    return SlidingWindowRateLimiter(get_log_query_config().rate_limit_per_minute)
+
+
+@lru_cache(maxsize=1)
+def get_concurrency_limiter() -> ConcurrencyLimiter:
+    """Provide the global log query concurrency limiter."""
+    return ConcurrencyLimiter(get_log_query_config().max_concurrent)
+
+
+def get_log_query_use_case(
+    job_repo=Depends(get_job_repo),
+    stage_repo=Depends(get_stage_repo),
+    log_files: FileSystemLogAdapter = Depends(get_log_file_service),
+    redaction: RedactionService = Depends(get_redaction_service),
+    config: LogQueryConfig = Depends(get_log_query_config),
+) -> LogQueryUseCase:
+    """Provide the log query use case bound to request-scoped repositories."""
+    return LogQueryUseCase(
+        job_repo=job_repo,
+        stage_repo=stage_repo,
+        log_files=log_files,
+        redaction=redaction,
+        settings=LogQuerySettings(
+            default_tail_lines=config.default_tail_lines,
+            max_tail_lines=config.max_tail_lines,
+            max_response_bytes=config.max_response_bytes,
+        ),
+    )
+
+
+def get_stage_log_metadata_service(
+    log_files: FileSystemLogAdapter = Depends(get_log_file_service),
+) -> StageLogMetadataService:
+    """Provide the stage log metadata service for GET /jobs/{job_id}."""
+    return StageLogMetadataService(log_files)
+
+
+def enforce_log_query_limits(
+    token_data: Annotated[dict, Depends(verify_token)],
+    rate_limiter: SlidingWindowRateLimiter = Depends(get_rate_limiter),
+    concurrency: ConcurrencyLimiter = Depends(get_concurrency_limiter),
+) -> Iterator[None]:
+    """Apply the per-client rate limit and hold a concurrency slot for the request."""
+    client_id = token_data["client_id"]
+    try:
+        rate_limiter.check(client_id)
+    except RateLimitExceededError as exc:
+        log_secure_info("warning", "Log query rate limit exceeded, status=429", client_id)
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail={"error": "RATE_LIMIT_EXCEEDED", "message": "Rate limit exceeded"},
+            headers={"Retry-After": str(exc.retry_after)},
+        ) from exc
+    try:
+        with concurrency.slot():
+            yield
+    except ConcurrencyLimitExceededError as exc:
+        log_secure_info("warning", "Log query concurrency limit reached, status=503", client_id)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail={
+                "error": "LOG_QUERY_BUSY",
+                "message": "Log query capacity exhausted; retry later",
+            },
+            headers={"Retry-After": str(exc.retry_after)},
+        ) from exc

src/build_stream/app/api/logs/limiter.py

@@ -0,0 +1,116 @@
+# Copyright 2026 Dell Inc. or its subsidiaries. All Rights Reserved.
+#
+# Licensed 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.
+
+# ****
+# Secure code generated utilizing Code Guardian.
+# Applicable rules used: CG-AUTH-001.6
+# ****
+
+"""In-process rate and concurrency limits for the log query API (ER-BSM-003 HLD 6.2).
+
+BSM runs a single uvicorn process, so in-memory counters are authoritative.
+Counters reset when the service restarts.
+"""
+
+import math
+import threading
+import time
+from collections import deque
+from contextlib import contextmanager
+from typing import Callable, Deque, Dict, Iterator
+
+
+class RateLimitExceededError(Exception):
+    """The client exceeded its request budget for the current window."""
+
+    def __init__(self, retry_after: int) -> None:
+        """Store the number of seconds after which a retry may succeed."""
+        super().__init__("Rate limit exceeded")
+        self.retry_after = retry_after
+
+
+class ConcurrencyLimitExceededError(Exception):
+    """All concurrent log query slots are in use."""
+
+    def __init__(self, retry_after: int = 1) -> None:
+        """Store the suggested retry delay in seconds."""
+        super().__init__("Log query capacity exhausted")
+        self.retry_after = retry_after
+
+
+class SlidingWindowRateLimiter:  # pylint: disable=too-few-public-methods
+    """Allow at most ``limit`` requests per key within ``window_seconds``."""
+
+    def __init__(
+        self,
+        limit: int,
+        window_seconds: float = 60.0,
+        clock: Callable[[], float] = time.monotonic,
+        max_keys: int = 1024,
+    ) -> None:
+        """Create the limiter; ``clock`` is injectable for tests."""
+        self._limit = limit
+        self._window = window_seconds
+        self._clock = clock
+        self._max_keys = max_keys
+        self._events: Dict[str, Deque[float]] = {}
+        self._lock = threading.Lock()
+
+    def check(self, key: str) -> None:
+        """Record one request for ``key``.
+
+        Raises:
+            RateLimitExceededError: The key already used its budget in this window.
+        """
+        now = self._clock()
+        with self._lock:
+            events = self._events.get(key)
+            if events is None:
+                if len(self._events) >= self._max_keys:
+                    self._prune(now)
+                events = self._events.setdefault(key, deque())
+            while events and now - events[0] >= self._window:
+                events.popleft()
+            if len(events) >= self._limit:
+                retry_after = max(1, math.ceil(self._window - (now - events[0])))
+                raise RateLimitExceededError(retry_after)
+            events.append(now)
+
+    def _prune(self, now: float) -> None:
+        """Drop keys with no requests inside the current window."""
+        stale = [k for k, ev in self._events.items() if not ev or now - ev[-1] >= self._window]
+        for key in stale:
+            del self._events[key]
+
+
+class ConcurrencyLimiter:  # pylint: disable=too-few-public-methods
+    """Bound the number of log queries served at the same time."""
+
+    def __init__(self, max_concurrent: int) -> None:
+        """Create the limiter with ``max_concurrent`` slots."""
+        self._semaphore = threading.BoundedSemaphore(max_concurrent)
+
+    @contextmanager
+    def slot(self) -> Iterator[None]:
+        """Hold one slot for the duration of the block.
+
+        Raises:
+            ConcurrencyLimitExceededError: No slot is free.
+        """
+        if not self._semaphore.acquire(blocking=False):  # pylint: disable=consider-using-with
+            raise ConcurrencyLimitExceededError()
+        try:
+            yield
+        finally:
+            self._semaphore.release()