(base_config: dict[str, Any], stage: StageSpec, request_index: int)
| 154 | |
| 155 | |
| 156 | def _build_request_payload(base_config: dict[str, Any], stage: StageSpec, request_index: int) -> dict[str, Any]: |
| 157 | payload = copy.deepcopy(base_config) |
| 158 | variant = stage.config_variants[request_index % len(stage.config_variants)] if stage.config_variants else {} |
| 159 | payload = _deep_merge(payload, variant) |
| 160 | |
| 161 | payload.setdefault("request_metrics", {}) |
| 162 | payload["request_metrics"]["request_id"] = request_index |
| 163 | payload["request_metrics"]["client_send_ts"] = time.time() |
| 164 | payload["request_metrics"]["stage_name"] = stage.name |
| 165 | payload["request_metrics"]["load_stage"] = stage.name |
| 166 | |
| 167 | if "data_bootstrap_room" not in payload: |
| 168 | payload["data_bootstrap_room"] = request_index |
| 169 | |
| 170 | save_path_prefix = os.getenv("DISAGG_WORKLOAD_SAVE_PREFIX") |
| 171 | if save_path_prefix: |
| 172 | save_root = Path(save_path_prefix) |
| 173 | save_root.parent.mkdir(parents=True, exist_ok=True) |
| 174 | payload["save_path"] = str(save_root.with_name(f"{save_root.stem}_{stage.name}_{request_index}{save_root.suffix}")) |
| 175 | |
| 176 | return payload |
| 177 | |
| 178 | |
| 179 | def _get_controller_target() -> tuple[str, int]: |
no test coverage detected