mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 恢复 Workbench Kafka refresh replay 配置

Browse Source
This commit is contained in:
Codex
2026-07-15 05:11:20 +02:00
parent 407ca55c70
commit d5570244f0
5 changed files with 215 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/native/hwlab/runtime-gitops-postprocess.mjs
+38 -15
View File
@@ -62,6 +62,11 @@ function patchCodeAgentRuntimeWorkloads() {
if (workloadName === "hwlab-cloud-api" && container.name === "hwlab-cloud-api") {
fileChanged = patchCloudApiKafkaEnv(container, kafka) || fileChanged;
}
if (workloadName === "hwlab-cloud-web" && container.name === "hwlab-cloud-web") {
fileChanged = setEnvValue(container, "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", String(kafka.enabled && kafka.features.liveKafkaSse)) || fileChanged;
fileChanged = setEnvValue(container, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(kafka.enabled && kafka.features.kafkaRefreshReplay)) || fileChanged;
fileChanged = setEnvValue(container, "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED", String(kafka.enabled && kafka.features.projectionRealtime)) || fileChanged;
}
}
}
}
@@ -76,24 +81,35 @@ function patchCodeAgentRuntimeWorkloads() {
function patchCloudApiKafkaEnv(container, kafka) {
let changed = false;
changed = setEnvValue(container, "HWLAB_KAFKA_ENABLED", String(kafka.enabled)) || changed;
changed = setEnvValue(container, "HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafka.enabled)) || changed;
changed = setEnvValue(container, "HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafka.enabled && (kafka.features.directPublish || kafka.features.transactionalProjector))) || changed;
changed = setEnvValue(container, "HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", String(kafka.enabled && kafka.features.directPublish)) || changed;
changed = setEnvValue(container, "HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", String(kafka.enabled && kafka.features.liveKafkaSse)) || changed;
const refreshReplayEnabled = kafka.enabled && kafka.features.kafkaRefreshReplay;
changed = setEnvValue(container, "HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(refreshReplayEnabled)) || changed;
changed = setEnvValue(container, "HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", String(kafka.enabled && kafka.features.transactionalProjector)) || changed;
changed = setEnvValue(container, "HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", String(kafka.enabled && kafka.features.projectionOutboxRelay)) || changed;
changed = setEnvValue(container, "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED", String(kafka.enabled && kafka.features.projectionRealtime)) || changed;
changed = setEnvValue(container, "HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID", String(kafka.directPublishConsumerGroupId)) || changed;
changed = setEnvValue(container, "HWLAB_KAFKA_PROJECTOR_GROUP_ID", String(kafka.transactionalProjectorConsumerGroupId)) || changed;
changed = setEnvValue(container, "HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID", String(kafka.hwlabEventConsumerGroupId)) || changed;
changed = patchOptionalEnvGroup(container, kafka.enabled, {
HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS: kafka.transactionalProjector?.heartbeatIntervalMs,
}) || changed;
changed = patchOptionalEnvGroup(container, kafka.enabled, {
HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS: kafka.projectionOutboxRelay?.intervalMs,
HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE: kafka.projectionOutboxRelay?.batchSize,
HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS: kafka.projectionOutboxRelay?.leaseMs,
HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS: kafka.projectionOutboxRelay?.sendTimeoutMs,
HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS: kafka.projectionOutboxRelay?.retryBackoffMs,
}) || changed;
changed = patchOptionalEnvGroup(container, kafka.enabled, {
HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE: kafka.projectionRealtime?.outboxTailBatchSize,
HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: kafka.projectionRealtime?.sseHeartbeatMs,
HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS: kafka.projectionRealtime?.sseDrainTimeoutMs,
changed = patchOptionalEnvGroup(container, refreshReplayEnabled, {
HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX: kafka.refreshReplay?.groupIdPrefix,
HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS: kafka.refreshReplay?.timeoutMs,
HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT: kafka.refreshReplay?.scanLimit,
HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT: kafka.refreshReplay?.matchedEventLimit,
HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT: kafka.refreshReplay?.liveBufferLimit,
}) || changed;
changed = removeEnvValues(container, [
"HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS",
"HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS",
"HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE",
"HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS",
"HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS",
"HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS",
"HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE",
"HWLAB_WORKBENCH_SSE_HEARTBEAT_MS",
"HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS",
]) || changed;
return changed;
}
@@ -105,6 +121,13 @@ function patchOptionalEnvGroup(container, enabled, values) {
return changed;
}
function removeEnvValues(container, names) {
const next = container.env.filter((item) => !names.includes(item?.name));
if (next.length === container.env.length) return false;
container.env = next;
return true;
}
function setEnvValue(container, name, value) {
const existing = container.env.find((item) => item?.name === name);
if (existing?.value === value && existing.valueFrom === undefined) return false;