Skip to content

spark_on_k8s.history.settings

Implementation reference. These symbols describe how the project works; their presence here does not make them supported application imports. Use the documented entry points and support policy.

HistorySettings dataclass

Source code in spark_on_k8s/history/settings.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@dataclass(frozen=True)
class HistorySettings:
    enabled: bool = False
    path: Path | None = None
    namespaces: tuple[str, ...] = ("default",)
    interval: int = 30
    collection_timeout: int = 15
    request_timeout: int = 5
    page_size: int = 200
    max_pages: int = 20
    retention_days: int = 30
    max_rows: int = 10000
    metrics_enabled: bool = False
    readiness: str = "degraded"

    @property
    def lease_seconds(self):
        return self.interval + self.collection_timeout + 10

    @property
    def freshness_seconds(self):
        return self.interval * 2 + self.collection_timeout + 10

    def is_fresh(self, last_success, now) -> bool:
        """Whether a namespace's last successful collection is within the readiness window."""
        return last_success is not None and 0 <= now - last_success <= self.freshness_seconds

    @classmethod
    def from_env(cls):
        enabled = boolean("SPARK_ON_K8S_RUN_HISTORY_ENABLED")
        metrics = boolean("SPARK_ON_K8S_METRICS_ENABLED")
        if not enabled:
            return cls(metrics_enabled=metrics)
        raw_path = os.getenv("SPARK_ON_K8S_RUN_HISTORY_PATH", "")
        path = Path(raw_path).expanduser()
        if not raw_path or not path.is_absolute() or path.name in {"", ":memory:"}:
            raise ValueError("SPARK_ON_K8S_RUN_HISTORY_PATH must be an absolute SQLite file path")
        readiness = os.getenv("SPARK_ON_K8S_RUN_HISTORY_READINESS", "degraded").lower()
        if readiness not in {"degraded", "strict"}:
            raise ValueError("SPARK_ON_K8S_RUN_HISTORY_READINESS must be degraded or strict")
        return cls(
            enabled=True,
            path=path,
            namespaces=namespaces(os.getenv("SPARK_ON_K8S_RUN_HISTORY_NAMESPACES", "default")),
            interval=integer("SPARK_ON_K8S_RUN_HISTORY_INTERVAL", 30, 5, 3600),
            collection_timeout=integer("SPARK_ON_K8S_RUN_HISTORY_TIMEOUT", 15, 1, 300),
            request_timeout=integer("SPARK_ON_K8S_RUN_HISTORY_REQUEST_TIMEOUT", 5, 1, 30),
            page_size=integer("SPARK_ON_K8S_RUN_HISTORY_PAGE_SIZE", 200, 1, 500),
            max_pages=integer("SPARK_ON_K8S_RUN_HISTORY_MAX_PAGES", 20, 1, 100),
            retention_days=integer("SPARK_ON_K8S_RUN_HISTORY_RETENTION_DAYS", 30, 1, 3650),
            max_rows=integer("SPARK_ON_K8S_RUN_HISTORY_MAX_ROWS", 10000, 1, 1000000),
            metrics_enabled=metrics,
            readiness=readiness,
        )

is_fresh(last_success, now)

Whether a namespace's last successful collection is within the readiness window.

Source code in spark_on_k8s/history/settings.py 
60
61
62
def is_fresh(self, last_success, now) -> bool:
    """Whether a namespace's last successful collection is within the readiness window."""
    return last_success is not None and 0 <= now - last_success <= self.freshness_seconds