feat(worker-rejoin-resync): wire config fields into xceadmin build (task 1.5)
Add workerResyncWindowMs/workerResyncTimeoutMs to xce.schema.json (minimum:0, matching the optional-scalar-field precedent -- these aren't `required` with a floor like heartbeatInterval, so minimum:1 would be wrong) and to mapping.py's YAML->capnp field table. Regenerate both golden config-build fixtures (task 1.1 grew ClusterConfig's wire layout by 2 fields; full/expected.bin and minimal/expected.bin each grew by exactly 8 bytes). full's fixture YAML now sets real values (900000ms/15000ms); minimal deliberately left untouched (no xce.yaml) per its own locked-down "hosts+queues only" boundary test -- its expected.bin still regenerates to reflect the new default-0 fields. Two new tests assert specific decoded values, not just "build succeeds". Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
This commit is contained in:
parent
9f3dfe7903
commit
ce3d391e5e
8 changed files with 81 additions and 9 deletions
|
|
@ -29,6 +29,16 @@
|
|||
"type": "integer",
|
||||
"minimum": 0,
|
||||
"description": "network I/O timeout in seconds"
|
||||
},
|
||||
"worker_resync_window_ms": {
|
||||
"type": "integer",
|
||||
"minimum": 0,
|
||||
"description": "worker job-state reconciliation: how far back (in milliseconds) a recently-finished job stays eligible for extra reporting on rejoin"
|
||||
},
|
||||
"worker_resync_timeout_ms": {
|
||||
"type": "integer",
|
||||
"minimum": 0,
|
||||
"description": "worker job-state reconciliation: how long (in milliseconds) the controller waits for a rejoining worker's job-status report before resolving the reconciliation as timed out"
|
||||
}
|
||||
},
|
||||
"required": ["cluster_name", "heartbeat_interval", "inactive_heartbeat_count"]
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ entries -- ``hosts.yaml`` is the one file that feeds two separate capnp
|
|||
fields, ``hgroups`` and ``hostConfigs``; see the Data Models table in
|
||||
design.md). Two mapper shapes cover every section:
|
||||
|
||||
- :func:`map_scalar_section` -- the single flat ``xce.yaml`` section, 5
|
||||
- :func:`map_scalar_section` -- the single flat ``xce.yaml`` section, 7
|
||||
top-level ``ClusterConfig`` scalar fields.
|
||||
- :func:`map_list_section` -- the 7 remaining sections, each a YAML list
|
||||
mapped to a capnp ``List(Struct)`` field, preserving YAML source order
|
||||
|
|
@ -180,6 +180,12 @@ _XCE_SCALAR_FIELDS: tuple[tuple[str, str], ...] = (
|
|||
("inactive_heartbeat_count", "inactiveHeartbeatCount"),
|
||||
("fetch_job_interval", "fetchJobInterval"),
|
||||
("netio_timeout", "netioTimeout"),
|
||||
# worker-rejoin-resync (Requirements 3.7, 5.3): shared, cluster-wide
|
||||
# via operational-uniformity override (design.md Configuration
|
||||
# Model) -- same domain as heartbeat_interval/inactive_heartbeat_count
|
||||
# above, wired the same way (task 1.5).
|
||||
("worker_resync_window_ms", "workerResyncWindowMs"),
|
||||
("worker_resync_timeout_ms", "workerResyncTimeoutMs"),
|
||||
)
|
||||
|
||||
# resources.yaml ``type``/``order`` -> Resource.valtype/ordtype, mirroring
|
||||
|
|
@ -500,7 +506,7 @@ def map_list_section(
|
|||
def map_scalar_section(
|
||||
spec: SectionSpec, values: dict | None, target: "capnp._DynamicStructBuilder" # noqa: ARG001
|
||||
) -> None:
|
||||
"""Set each of the 5 ``xce.yaml``-sourced scalar ``ClusterConfig``
|
||||
"""Set each of the 7 ``xce.yaml``-sourced scalar ``ClusterConfig``
|
||||
fields directly from ``values`` (:data:`_XCE_SCALAR_FIELDS`).
|
||||
|
||||
``spec`` is accepted for interface symmetry with :func:`map_list_section`
|
||||
|
|
|
|||
|
|
@ -4,3 +4,5 @@ heartbeat_interval: 30
|
|||
inactive_heartbeat_count: 3
|
||||
fetch_job_interval: 10
|
||||
netio_timeout: 30
|
||||
worker_resync_window_ms: 900000
|
||||
worker_resync_timeout_ms: 15000
|
||||
|
|
|
|||
BIN
src/xceadmin/tests/fixtures/full/expected.bin
vendored
BIN
src/xceadmin/tests/fixtures/full/expected.bin
vendored
Binary file not shown.
BIN
src/xceadmin/tests/fixtures/minimal/expected.bin
vendored
BIN
src/xceadmin/tests/fixtures/minimal/expected.bin
vendored
Binary file not shown.
|
|
@ -83,6 +83,50 @@ def test_build_same_input_twice_is_byte_identical(tmp_path):
|
|||
assert first == second
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# worker-rejoin-resync (Requirements 3.7, 5.3): workerResyncWindowMs/
|
||||
# workerResyncTimeoutMs are operator-settable through xce.yaml's scalar
|
||||
# section, following the heartbeat_interval/inactive_heartbeat_count
|
||||
# pattern exactly (task 1.5). These prove the VALUE round-trips
|
||||
# (YAML -> binary -> decode) with a specific, checkable pair of numbers --
|
||||
# not just that some golden blob happens to match.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_build_maps_worker_resync_window_and_timeout_fields(tmp_path):
|
||||
_write_yaml(
|
||||
tmp_path / "xce.yaml",
|
||||
{
|
||||
"cluster_name": "eda1",
|
||||
"heartbeat_interval": 30,
|
||||
"inactive_heartbeat_count": 3,
|
||||
"worker_resync_window_ms": 900000,
|
||||
"worker_resync_timeout_ms": 15000,
|
||||
},
|
||||
)
|
||||
|
||||
data = build.build_cluster_config(str(tmp_path))
|
||||
|
||||
mod = load_schema()
|
||||
with mod.ClusterConfig.from_bytes(data) as reader:
|
||||
assert reader.workerResyncWindowMs == 900000
|
||||
assert reader.workerResyncTimeoutMs == 15000
|
||||
|
||||
|
||||
def test_build_leaves_worker_resync_fields_at_default_when_absent(tmp_path):
|
||||
_write_yaml(
|
||||
tmp_path / "xce.yaml",
|
||||
{"cluster_name": "eda1", "heartbeat_interval": 30, "inactive_heartbeat_count": 3},
|
||||
)
|
||||
|
||||
data = build.build_cluster_config(str(tmp_path))
|
||||
|
||||
mod = load_schema()
|
||||
with mod.ClusterConfig.from_bytes(data) as reader:
|
||||
assert reader.workerResyncWindowMs == 0
|
||||
assert reader.workerResyncTimeoutMs == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Hard failures propagate (schema violation, reserved name, duplicates)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -60,11 +60,12 @@ def test_section_map_hosts_yaml_feeds_two_distinct_capnp_fields():
|
|||
assert yaml_keys == {"hostgroups", "hosts"}
|
||||
|
||||
|
||||
def test_section_map_covers_all_13_clusterconfig_fields_or_documents_exclusion():
|
||||
# 5 scalar (xce.yaml) + 7 list capnp_fields covers 12 of the 13
|
||||
# ClusterConfig fields; configGeneration (@12) is the one deliberately
|
||||
# excluded field (Requirement 1.9), documented in the module docstring,
|
||||
# not present in SECTION_MAP at all.
|
||||
def test_section_map_covers_all_14_clusterconfig_fields_or_documents_exclusion():
|
||||
# 7 scalar (xce.yaml, including worker-rejoin-resync's workerResyncWindowMs/
|
||||
# workerResyncTimeoutMs, task 1.5) + 7 list capnp_fields covers 14 of the
|
||||
# 15 ClusterConfig fields (@0-@14); configGeneration (@12) is the one
|
||||
# deliberately excluded field (Requirement 1.9), documented in the
|
||||
# module docstring, not present in SECTION_MAP at all.
|
||||
list_fields = {s["capnp_field"] for s in mapping.SECTION_MAP if s["kind"] == "list"}
|
||||
assert list_fields == {
|
||||
"resources",
|
||||
|
|
@ -83,6 +84,8 @@ def test_section_map_covers_all_13_clusterconfig_fields_or_documents_exclusion()
|
|||
"inactiveHeartbeatCount",
|
||||
"fetchJobInterval",
|
||||
"netioTimeout",
|
||||
"workerResyncWindowMs",
|
||||
"workerResyncTimeoutMs",
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -91,7 +94,7 @@ def test_section_map_covers_all_13_clusterconfig_fields_or_documents_exclusion()
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_map_scalar_section_sets_all_5_fields(cluster_config):
|
||||
def test_map_scalar_section_sets_all_7_fields(cluster_config):
|
||||
spec = mapping.SECTION_MAP[0]
|
||||
values = {
|
||||
"cluster_name": "eda1",
|
||||
|
|
@ -99,6 +102,8 @@ def test_map_scalar_section_sets_all_5_fields(cluster_config):
|
|||
"inactive_heartbeat_count": 3,
|
||||
"fetch_job_interval": 10,
|
||||
"netio_timeout": 30,
|
||||
"worker_resync_window_ms": 900000,
|
||||
"worker_resync_timeout_ms": 15000,
|
||||
}
|
||||
mapping.map_scalar_section(spec, values, cluster_config)
|
||||
|
||||
|
|
@ -107,6 +112,8 @@ def test_map_scalar_section_sets_all_5_fields(cluster_config):
|
|||
assert cluster_config.inactiveHeartbeatCount == 3
|
||||
assert cluster_config.fetchJobInterval == 10
|
||||
assert cluster_config.netioTimeout == 30
|
||||
assert cluster_config.workerResyncWindowMs == 900000
|
||||
assert cluster_config.workerResyncTimeoutMs == 15000
|
||||
|
||||
|
||||
def test_map_scalar_section_leaves_missing_optional_keys_at_default(cluster_config):
|
||||
|
|
@ -117,6 +124,8 @@ def test_map_scalar_section_leaves_missing_optional_keys_at_default(cluster_conf
|
|||
assert cluster_config.clusterName == "eda1"
|
||||
assert cluster_config.fetchJobInterval == 0
|
||||
assert cluster_config.netioTimeout == 0
|
||||
assert cluster_config.workerResyncWindowMs == 0
|
||||
assert cluster_config.workerResyncTimeoutMs == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("empty", [None, {}])
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue