feat(generator): ускорен учебный профиль daily-wave
- Зачем: - суточная волна должна быть видна на занятии за минуты, а не за сутки работы стенда. - Что: - профиль daily-wave переведён на speed 60 при тике 1 секунда. - приглушены подробные live-логи успешного тика без изменения сохранения state. - обновлены тесты, спека и инструкции запуска быстрого профиля. - Проверка: - make generator-test. - git diff --cached --check.
This commit is contained in:
@@ -47,7 +47,9 @@ make generated-history-analytics
|
|||||||
```
|
```
|
||||||
|
|
||||||
По умолчанию это быстрый профиль `ci`: 6 часов модельного времени. Историю на
|
По умолчанию это быстрый профиль `ci`: 6 часов модельного времени. Историю на
|
||||||
2 суток с суточной волной можно прогнать одной командой:
|
2 суток с суточной волной можно прогнать одной командой. В live-продолжении
|
||||||
|
`daily-wave` идёт с ×60: модельные сутки проходят примерно за 24 настенные
|
||||||
|
минуты.
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
PROFILE=daily-wave make generated-history-analytics
|
PROFILE=daily-wave make generated-history-analytics
|
||||||
|
|||||||
+4
-2
@@ -178,7 +178,9 @@ make generated-history-check
|
|||||||
```
|
```
|
||||||
|
|
||||||
По умолчанию команда использует быстрый профиль `ci`: 6 часов модельного
|
По умолчанию команда использует быстрый профиль `ci`: 6 часов модельного
|
||||||
времени. Историю на 2 суток с суточной волной можно получить одной командой:
|
времени. Историю на 2 суток с суточной волной можно получить одной командой.
|
||||||
|
В live-продолжении `daily-wave` идёт с ×60 и тикает раз в секунду, поэтому
|
||||||
|
модельные сутки проходят примерно за 24 настенные минуты:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
PROFILE=daily-wave make generated-history-analytics
|
PROFILE=daily-wave make generated-history-analytics
|
||||||
@@ -232,7 +234,7 @@ Live-продолжение стартует с этого state. Использ
|
|||||||
```bash
|
```bash
|
||||||
PROFILE=daily-wave make generator-continue
|
PROFILE=daily-wave make generator-continue
|
||||||
|
|
||||||
sleep 130
|
sleep 5
|
||||||
make generator-down
|
make generator-down
|
||||||
sleep 10
|
sleep 10
|
||||||
bash scripts/run_batch.sh
|
bash scripts/run_batch.sh
|
||||||
|
|||||||
@@ -26,10 +26,10 @@
|
|||||||
|
|
||||||
Основной способ — глагол плюс профиль:
|
Основной способ — глагол плюс профиль:
|
||||||
|
|
||||||
| Профиль | Для чего | Длительность |
|
| Профиль | Для чего | Длительность | Live-ход |
|
||||||
|---------|----------|--------------|
|
|---------|----------|--------------|----------|
|
||||||
| `ci` | Быстрая проверка и CI | `6h` |
|
| `ci` | Быстрая проверка и CI | `6h` | ×1, тик 60 с |
|
||||||
| `daily-wave` | История с видимой суточной волной | `2d` |
|
| `daily-wave` | История с видимой суточной волной | `2d` | ×60, тик 1 с |
|
||||||
|
|
||||||
Длительность можно переопределить через `GEN_HISTORY_DURATION`, например `2d`.
|
Длительность можно переопределить через `GEN_HISTORY_DURATION`, например `2d`.
|
||||||
Команда сама считает `GEN_MODEL_T_END` от `GEN_MODEL_T0`.
|
Команда сама считает `GEN_MODEL_T_END` от `GEN_MODEL_T0`.
|
||||||
@@ -84,7 +84,8 @@ make startup-history-export
|
|||||||
|
|
||||||
Файл по умолчанию: `/tmp/clickstream-startup-history.json`.
|
Файл по умолчанию: `/tmp/clickstream-startup-history.json`.
|
||||||
|
|
||||||
История на 2 суток с суточной волной:
|
История на 2 суток с суточной волной. В live-продолжении этот профиль проживает
|
||||||
|
модельные сутки примерно за 24 настенные минуты:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
ARTIFACT=/tmp/clickstream-startup-history-2d.json \
|
ARTIFACT=/tmp/clickstream-startup-history-2d.json \
|
||||||
@@ -133,10 +134,11 @@ ClickHouse уже успел прочитать частичные сообще
|
|||||||
|
|
||||||
## Live-продолжение
|
## Live-продолжение
|
||||||
|
|
||||||
После импорта запускайте live с тем же профилем, что был в артефакте:
|
После импорта запускайте live с тем же профилем, что был в артефакте. Для
|
||||||
|
артефакта `daily-wave`:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
PROFILE=ci make generator-continue
|
PROFILE=daily-wave make generator-continue
|
||||||
```
|
```
|
||||||
|
|
||||||
Если читаемый state есть, но настройки не совпадают, генератор падает с перечнем
|
Если читаемый state есть, но настройки не совпадают, генератор падает с перечнем
|
||||||
@@ -146,6 +148,9 @@ PROFILE=ci make generator-continue
|
|||||||
После нестандартного мира `make generator-continue` нужно запускать с теми же
|
После нестандартного мира `make generator-continue` нужно запускать с теми же
|
||||||
настройками, что были у backfill/import. При расхождении генератор громко
|
настройками, что были у backfill/import. При расхождении генератор громко
|
||||||
покажет поля, которые не совпали.
|
покажет поля, которые не совпали.
|
||||||
|
Старые артефакты `daily-wave`, созданные до перехода на ×60 и тик 1 с, с новым
|
||||||
|
профилем несовместимы. Это ожидаемо: защита от смешения миров должна остановить
|
||||||
|
такой запуск.
|
||||||
|
|
||||||
Старые переменные `GEN_RUN_MODE`, `GEN_STATE_RESET` и `GEN_MODEL_T_END` остаются
|
Старые переменные `GEN_RUN_MODE`, `GEN_STATE_RESET` и `GEN_MODEL_T_END` остаются
|
||||||
низкоуровневым способом для отладки и прямого `docker compose run`.
|
низкоуровневым способом для отладки и прямого `docker compose run`.
|
||||||
|
|||||||
@@ -104,10 +104,23 @@ ADR-0005 решил отвязать время генератора от реа
|
|||||||
- `continue` — подставляет `GEN_RUN_MODE=live` и `GEN_STATE_RESET=false`;
|
- `continue` — подставляет `GEN_RUN_MODE=live` и `GEN_STATE_RESET=false`;
|
||||||
- `reset` — подставляет `GEN_RUN_MODE=live` и `GEN_STATE_RESET=true`.
|
- `reset` — подставляет `GEN_RUN_MODE=live` и `GEN_STATE_RESET=true`.
|
||||||
|
|
||||||
Профиль `ci` даёт быстрый 6-часовой прогон. Профиль `daily-wave` даёт 2 суток,
|
Профиль `ci` даёт быстрый 6-часовой прогон и остаётся на ×1 с тиком 60 с.
|
||||||
чтобы была видна суточная волна. `GEN_HISTORY_DURATION` задаёт длительность
|
Профиль `daily-wave` даёт 2 суток, чтобы была видна суточная волна, а в live
|
||||||
вида `6h` или `2d`; `GEN_MODEL_T_END` считается от `GEN_MODEL_T0` внутри слоя
|
идёт с `GEN_MODEL_TIME_SPEED=60` и `GEN_TICK_SECONDS=1`: модельные сутки
|
||||||
запуска. Старые переменные остаются низкоуровневым механизмом.
|
проходят примерно за 24 настенные минуты. Пара «тик 1 с, скорость ×60» выбрана,
|
||||||
|
чтобы модельный шаг тика остался 60 секунд. Так событийный бюджет и форма волны
|
||||||
|
не меняются относительно старого минутного тика, а пиковый тик не упирается в
|
||||||
|
`GEN_MAX_EVENTS_PER_TICK`. Вариант «тик 60 с, скорость ×60» не подходит: один
|
||||||
|
тик стал бы модельным часом и слипал бы события в грубые часовые пачки.
|
||||||
|
`GEN_HISTORY_DURATION` задаёт длительность вида `6h` или `2d`; `GEN_MODEL_T_END`
|
||||||
|
считается от `GEN_MODEL_T0` внутри слоя запуска. Старые переменные остаются
|
||||||
|
низкоуровневым механизмом.
|
||||||
|
|
||||||
|
Цена быстрого live-хода `daily-wave`: при тике 1 с сервис пишет примерно 86 400
|
||||||
|
state-записей и столько же записей истории пачек за настенные сутки. Частота
|
||||||
|
state-записей остаётся прежним контрактом возобновления после сбоя: успешный тик
|
||||||
|
сразу сохраняет новую модельную точку. Чтобы быстрый профиль не заливал журналы,
|
||||||
|
подробные строки успешного тика пишутся на DEBUG, а не на INFO.
|
||||||
|
|
||||||
#### Ход часов
|
#### Ход часов
|
||||||
|
|
||||||
|
|||||||
+5
-3
@@ -127,7 +127,9 @@ make generated-history-analytics
|
|||||||
они пока используются только как фактура для генератора.
|
они пока используются только как фактура для генератора.
|
||||||
|
|
||||||
По умолчанию используется быстрый профиль `ci`: 6 часов модельного времени.
|
По умолчанию используется быстрый профиль `ci`: 6 часов модельного времени.
|
||||||
Профиль `daily-wave` даёт 2 суток, чтобы была видна суточная волна:
|
Профиль `daily-wave` даёт 2 суток, чтобы была видна суточная волна. В live он
|
||||||
|
идёт с `GEN_MODEL_TIME_SPEED=60` и `GEN_TICK_SECONDS=1`: модельные сутки
|
||||||
|
проходят примерно за 24 настенные минуты.
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
PROFILE=daily-wave make generated-history-analytics
|
PROFILE=daily-wave make generated-history-analytics
|
||||||
@@ -139,9 +141,9 @@ PROFILE=daily-wave make generated-history-analytics
|
|||||||
GEN_HISTORY_DURATION=2d make generated-history-analytics
|
GEN_HISTORY_DURATION=2d make generated-history-analytics
|
||||||
```
|
```
|
||||||
|
|
||||||
### Режим "раз в минуту" (для демо)
|
### Режим "раз в минуту" (для ручных экспериментов)
|
||||||
|
|
||||||
Для контролируемых демо можно установить:
|
Для ручных экспериментов можно установить:
|
||||||
```bash
|
```bash
|
||||||
GEN_TICK_SECONDS=60
|
GEN_TICK_SECONDS=60
|
||||||
GEN_MIN_EVENTS_PER_TICK=50
|
GEN_MIN_EVENTS_PER_TICK=50
|
||||||
|
|||||||
@@ -52,8 +52,8 @@ PROFILES = {
|
|||||||
"GEN_SEED": "4242",
|
"GEN_SEED": "4242",
|
||||||
"GEN_MODEL_T0": "2026-01-01T00:00:00+00:00",
|
"GEN_MODEL_T0": "2026-01-01T00:00:00+00:00",
|
||||||
"GEN_MODEL_TIMEZONE": "UTC",
|
"GEN_MODEL_TIMEZONE": "UTC",
|
||||||
"GEN_MODEL_TIME_SPEED": "1",
|
"GEN_MODEL_TIME_SPEED": "60",
|
||||||
"GEN_TICK_SECONDS": "60",
|
"GEN_TICK_SECONDS": "1",
|
||||||
"GEN_LAMBDA_BASE_PER_MIN": "60",
|
"GEN_LAMBDA_BASE_PER_MIN": "60",
|
||||||
"GEN_JITTER_PCT": "0",
|
"GEN_JITTER_PCT": "0",
|
||||||
"GEN_MIN_EVENTS_PER_TICK": "1",
|
"GEN_MIN_EVENTS_PER_TICK": "1",
|
||||||
|
|||||||
@@ -529,13 +529,13 @@ class GeneratorService:
|
|||||||
model_time = self._model_time
|
model_time = self._model_time
|
||||||
|
|
||||||
with METRICS_TICK_DURATION.time():
|
with METRICS_TICK_DURATION.time():
|
||||||
logger.info(f"=== Tick {self._tick} (batch_id={batch_id}) ===")
|
logger.debug(f"=== Tick {self._tick} (batch_id={batch_id}) ===")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
events_count = self.generator._calculate_events_count(
|
events_count = self.generator._calculate_events_count(
|
||||||
now=model_time,
|
now=model_time,
|
||||||
)
|
)
|
||||||
logger.info(f"Generating with event budget ~{events_count}")
|
logger.debug(f"Generating with event budget ~{events_count}")
|
||||||
|
|
||||||
gen_start = time.time()
|
gen_start = time.time()
|
||||||
batch = self.stream.generate_tick(
|
batch = self.stream.generate_tick(
|
||||||
@@ -591,7 +591,7 @@ class GeneratorService:
|
|||||||
METRICS_ERRORS_TOTAL.labels(topic="history").inc()
|
METRICS_ERRORS_TOTAL.labels(topic="history").inc()
|
||||||
|
|
||||||
tick_duration = time.time() - tick_start
|
tick_duration = time.time() - tick_start
|
||||||
logger.info(
|
logger.debug(
|
||||||
f"Batch {batch_id} completed: "
|
f"Batch {batch_id} completed: "
|
||||||
f"sent={total_sent}, errors={total_errors}, "
|
f"sent={total_sent}, errors={total_errors}, "
|
||||||
f"gen_time={gen_duration:.3f}s, pub_time={pub_duration:.3f}s, "
|
f"gen_time={gen_duration:.3f}s, pub_time={pub_duration:.3f}s, "
|
||||||
@@ -600,7 +600,7 @@ class GeneratorService:
|
|||||||
|
|
||||||
for topic, counts in sent_counts.items():
|
for topic, counts in sent_counts.items():
|
||||||
if counts["sent"] > 0:
|
if counts["sent"] > 0:
|
||||||
logger.info(f" {topic}: {counts['sent']} sent")
|
logger.debug(f" {topic}: {counts['sent']} sent")
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.exception(f"Error in tick {self._tick}: {e}")
|
logger.exception(f"Error in tick {self._tick}: {e}")
|
||||||
|
|||||||
@@ -24,6 +24,8 @@ def test_daily_wave_profile_uses_two_days_by_default():
|
|||||||
assert env["GEN_LAUNCH_PROFILE"] == "daily-wave"
|
assert env["GEN_LAUNCH_PROFILE"] == "daily-wave"
|
||||||
assert env["GEN_HISTORY_DURATION"] == "2d"
|
assert env["GEN_HISTORY_DURATION"] == "2d"
|
||||||
assert env["GEN_MODEL_T_END"] == "2026-01-03T00:00:00+00:00"
|
assert env["GEN_MODEL_T_END"] == "2026-01-03T00:00:00+00:00"
|
||||||
|
assert env["GEN_MODEL_TIME_SPEED"] == "60"
|
||||||
|
assert env["GEN_TICK_SECONDS"] == "1"
|
||||||
|
|
||||||
|
|
||||||
def test_backfill_verb_maps_to_existing_low_level_flags():
|
def test_backfill_verb_maps_to_existing_low_level_flags():
|
||||||
@@ -61,6 +63,7 @@ def test_ci_profile_keeps_previous_default_backfill_window():
|
|||||||
t_end = datetime.fromisoformat(env["GEN_MODEL_T_END"])
|
t_end = datetime.fromisoformat(env["GEN_MODEL_T_END"])
|
||||||
assert t0 == datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc)
|
assert t0 == datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc)
|
||||||
assert (t_end - t0).total_seconds() == 6 * 60 * 60
|
assert (t_end - t0).total_seconds() == 6 * 60 * 60
|
||||||
|
assert env["GEN_MODEL_TIME_SPEED"] == "1"
|
||||||
assert env["GEN_TICK_SECONDS"] == "60"
|
assert env["GEN_TICK_SECONDS"] == "60"
|
||||||
assert env["GEN_LAMBDA_BASE_PER_MIN"] == "60"
|
assert env["GEN_LAMBDA_BASE_PER_MIN"] == "60"
|
||||||
assert env["GEN_JITTER_PCT"] == "0"
|
assert env["GEN_JITTER_PCT"] == "0"
|
||||||
|
|||||||
@@ -184,6 +184,42 @@ class TestGeneratorServiceSteadyStream:
|
|||||||
"2026-01-01 10:10:00.000000",
|
"2026-01-01 10:10:00.000000",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def test_successful_live_tick_does_not_log_per_tick_info(self, base_config, caplog):
|
||||||
|
"""Успешный live-тик не пишет подробный пер-тиковый журнал на INFO."""
|
||||||
|
config = replace(
|
||||||
|
base_config,
|
||||||
|
tick_seconds=1,
|
||||||
|
lambda_base_per_min=60,
|
||||||
|
jitter_pct=0,
|
||||||
|
min_events_per_tick=1,
|
||||||
|
max_events_per_tick=1000,
|
||||||
|
max_session_events=1,
|
||||||
|
max_active_sessions=10,
|
||||||
|
population_max=11,
|
||||||
|
)
|
||||||
|
service = GeneratorService(config)
|
||||||
|
service.publisher = MagicMock()
|
||||||
|
service.publisher.publish.side_effect = (
|
||||||
|
lambda topic, events: (len(events), 0)
|
||||||
|
)
|
||||||
|
service.history = MagicMock()
|
||||||
|
service._running = True
|
||||||
|
|
||||||
|
def stop_after_first_tick(_sleep_seconds):
|
||||||
|
service._running = False
|
||||||
|
|
||||||
|
with caplog.at_level(logging.INFO, logger="generator"), \
|
||||||
|
patch(
|
||||||
|
"clickstream_generator.service.time.sleep",
|
||||||
|
side_effect=stop_after_first_tick,
|
||||||
|
):
|
||||||
|
service._main_loop()
|
||||||
|
|
||||||
|
assert "=== Tick" not in caplog.text
|
||||||
|
assert "Generating with event budget" not in caplog.text
|
||||||
|
assert "completed:" not in caplog.text
|
||||||
|
assert "sent" not in caplog.text
|
||||||
|
|
||||||
def _run_service_ticks(self, config, wall_now: datetime, ticks_count: int):
|
def _run_service_ticks(self, config, wall_now: datetime, ticks_count: int):
|
||||||
service = GeneratorService(config)
|
service = GeneratorService(config)
|
||||||
service.publisher = MagicMock()
|
service.publisher = MagicMock()
|
||||||
|
|||||||
Reference in New Issue
Block a user