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

Merge pull request #1738 from pikasTech/fix/2486-kafka-refresh-yaml
Pipelines as Code CI / hwlab-web-probe-sentinel-nc01- Success
Details
Pipelines as Code CI / unidesk-host- Success
Details

Browse Source 
feat: 为 HWLAB 声明 Kafka 刷新重放配置
This commit is contained in:
Lyon
2026-07-11 11:00:36 +08:00
committed by GitHub
parent ce5c20f627 499e0d8aa7
commit 27b1bb8344
5 changed files with 103 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/hwlab-node-lanes.yaml
+7
View File
@@ -203,9 +203,16 @@ lanes:
features:
directPublish: true
liveKafkaSse: true
kafkaRefreshReplay: true
transactionalProjector: false
projectionOutboxRelay: false
projectionRealtime: false
refreshReplay:
groupIdPrefix: hwlab-v03-cloud-api-sse
timeoutMs: 30000
scanLimit: 1000000
matchedEventLimit: 2000
liveBufferLimit: 2000
configRef: config/platform-infra/kafka.yaml#clients.hwlab-v03-cloud-api
bootstrapServers: platform-infra-kafka-kafka-bootstrap.platform-infra.svc.cluster.local:9092
stdioTopic: codex-stdio.raw.v1
scripts/src/hwlab-node-lanes.ts
+33 -1
View File
@@ -474,10 +474,18 @@ export interface HwlabRuntimeKafkaEventBridgeSpec {
readonly features: {
readonly directPublish: boolean;
readonly liveKafkaSse: boolean;
readonly kafkaRefreshReplay: boolean;
readonly transactionalProjector: boolean;
readonly projectionOutboxRelay: boolean;
readonly projectionRealtime: boolean;
};
readonly refreshReplay?: {
readonly groupIdPrefix: string;
readonly timeoutMs: number;
readonly scanLimit: number;
readonly matchedEventLimit: number;
readonly liveBufferLimit: number;
};
readonly configRef: string;
readonly bootstrapServers: string;
readonly stdioTopic: string;
@@ -1087,6 +1095,17 @@ function codeAgentKafkaShadowProducerConfig(value: unknown, path: string): Hwlab
function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRuntimeKafkaEventBridgeSpec {
const raw = asRecord(value, path);
const features = asRecord(raw.features, `${path}.features`);
const liveKafkaSse = booleanField(features, "liveKafkaSse", `${path}.features`);
const kafkaRefreshReplay = booleanField(features, "kafkaRefreshReplay", `${path}.features`);
const refreshReplay = raw.refreshReplay === undefined
? undefined
: codeAgentKafkaRefreshReplayConfig(raw.refreshReplay, `${path}.refreshReplay`);
if (kafkaRefreshReplay && refreshReplay === undefined) {
throw new Error(`${path}.refreshReplay is required when ${path}.features.kafkaRefreshReplay is true`);
}
if (kafkaRefreshReplay && !liveKafkaSse) {
throw new Error(`${path}.features.liveKafkaSse must be true when ${path}.features.kafkaRefreshReplay is true`);
}
const directPublishConsumerGroupId = stringField(raw, "directPublishConsumerGroupId", path);
const transactionalProjectorConsumerGroupId = stringField(raw, "transactionalProjectorConsumerGroupId", path);
const hwlabEventConsumerGroupId = stringField(raw, "hwlabEventConsumerGroupId", path);
@@ -1097,11 +1116,13 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
enabled: booleanField(raw, "enabled", path),
features: {
directPublish: booleanField(features, "directPublish", `${path}.features`),
liveKafkaSse: booleanField(features, "liveKafkaSse", `${path}.features`),
liveKafkaSse,
kafkaRefreshReplay,
transactionalProjector: booleanField(features, "transactionalProjector", `${path}.features`),
projectionOutboxRelay: booleanField(features, "projectionOutboxRelay", `${path}.features`),
projectionRealtime: booleanField(features, "projectionRealtime", `${path}.features`),
},
...(refreshReplay === undefined ? {} : { refreshReplay }),
configRef: stringField(raw, "configRef", path),
bootstrapServers: stringField(raw, "bootstrapServers", path),
stdioTopic: stringField(raw, "stdioTopic", path),
@@ -1114,6 +1135,17 @@ function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRun
};
}
function codeAgentKafkaRefreshReplayConfig(value: unknown, path: string): NonNullable<HwlabRuntimeKafkaEventBridgeSpec["refreshReplay"]> {
const raw = asRecord(value, path);
return {
groupIdPrefix: stringField(raw, "groupIdPrefix", path),
timeoutMs: numberField(raw, "timeoutMs", path),
scanLimit: numberField(raw, "scanLimit", path),
matchedEventLimit: numberField(raw, "matchedEventLimit", path),
liveBufferLimit: numberField(raw, "liveBufferLimit", path),
};
}
function commandNameField(value: string, path: string): string {
if (!/^[A-Za-z0-9._+-]+$/u.test(value)) throw new Error(`${path} must be a simple command name`);
return value;
scripts/src/hwlab-node/render.ts
+35
View File
@@ -1810,6 +1810,13 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
" delete item.valueFrom;",
" return changed;",
"}",
"function removeEnvValues(container, names) {",
" if (!isObject(container) || !Array.isArray(container.env)) return false;",
" const next = container.env.filter((env) => !(env && names.includes(env.name)));",
" if (next.length === container.env.length) return false;",
" container.env = next;",
" return true;",
"}",
"function isEnvReuseContainer(container) { return envValue(container, 'HWLAB_RUNTIME_MODE') === 'env-reuse-git-mirror-checkout' || envValue(container, 'HWLAB_BOOT_SH') !== undefined || envValue(container, 'HWLAB_BOOT_COMMIT') !== undefined; }",
"function workloadName(item) { return item && item.metadata && item.metadata.labels && item.metadata.labels['app.kubernetes.io/name'] ? String(item.metadata.labels['app.kubernetes.io/name']) : String(item && item.metadata && item.metadata.name || ''); }",
"function expectedPublicEndpoint(item) { return workloadName(item) === 'hwlab-cloud-web' ? overlay.publicWebUrl : overlay.publicApiUrl; }",
@@ -1819,6 +1826,7 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
" const kafka = overlay.codeAgentRuntime && overlay.codeAgentRuntime.kafkaEventBridge;",
" if (kafka) {",
" entries.push({ name: 'HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED', value: String(kafka.enabled && kafka.features.liveKafkaSse) });",
" entries.push({ name: 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED', value: String(kafka.enabled && kafka.features.kafkaRefreshReplay) });",
" entries.push({ name: 'HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED', value: String(kafka.enabled && kafka.features.projectionRealtime) });",
" }",
" return entries;",
@@ -1929,6 +1937,17 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID', String(kafka.hwlabEventConsumerGroupId)) || codeAgentRuntimeChanged;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED', String(kafka.enabled && kafka.features.directPublish)) || codeAgentRuntimeChanged;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED', String(kafka.enabled && kafka.features.liveKafkaSse)) || codeAgentRuntimeChanged;",
" const refreshReplayEnabled = kafka.enabled && kafka.features.kafkaRefreshReplay;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED', String(refreshReplayEnabled)) || codeAgentRuntimeChanged;",
" if (refreshReplayEnabled) {",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX', String(kafka.refreshReplay.groupIdPrefix)) || codeAgentRuntimeChanged;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS', String(kafka.refreshReplay.timeoutMs)) || codeAgentRuntimeChanged;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT', String(kafka.refreshReplay.scanLimit)) || codeAgentRuntimeChanged;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT', String(kafka.refreshReplay.matchedEventLimit)) || codeAgentRuntimeChanged;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT', String(kafka.refreshReplay.liveBufferLimit)) || codeAgentRuntimeChanged;",
" } else {",
" codeAgentRuntimeChanged = removeEnvValues(container, ['HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT']) || codeAgentRuntimeChanged;",
" }",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED', String(kafka.enabled && kafka.features.transactionalProjector)) || codeAgentRuntimeChanged;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED', String(kafka.enabled && kafka.features.projectionOutboxRelay)) || codeAgentRuntimeChanged;",
" codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED', String(kafka.enabled && kafka.features.projectionRealtime)) || codeAgentRuntimeChanged;",
@@ -2259,6 +2278,7 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
" const kafka = overlay.codeAgentRuntime && overlay.codeAgentRuntime.kafkaEventBridge;",
" if (kafka) {",
" entries.push({ name: 'HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED', value: String(kafka.enabled && kafka.features.liveKafkaSse) });",
" entries.push({ name: 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED', value: String(kafka.enabled && kafka.features.kafkaRefreshReplay) });",
" entries.push({ name: 'HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED', value: String(kafka.enabled && kafka.features.projectionRealtime) });",
" }",
" return entries;",
@@ -2285,6 +2305,10 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
" const value = envValue(container, envName);",
" if (value !== expected) wrongCodeAgentRuntimeEnvs.push({ ...workloadRef(item, file, container), envName, kind: 'value', expected, value: value ?? null });",
" }",
" function checkCodeAgentRuntimeAbsent(item, file, container, envName) {",
" const value = envValue(container, envName);",
" if (value !== undefined) wrongCodeAgentRuntimeEnvs.push({ ...workloadRef(item, file, container), envName, kind: 'unexpected-value', expected: null, value });",
" }",
" function checkCodeAgentRuntimeSecret(item, file, container, envName, expectedSecret, expectedKey) {",
" const secretRef = envSecretRef(container, envName);",
" if (secretRef.name !== expectedSecret || secretRef.key !== expectedKey) wrongCodeAgentRuntimeEnvs.push({ ...workloadRef(item, file, container), envName, kind: 'secretKeyRef', expectedSecret, expectedKey, secret: secretRef.name ?? null, key: secretRef.key ?? null });",
@@ -2341,6 +2365,17 @@ export function nodeRuntimePipelinePostprocessScript(): string[] {
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID', String(kafka.hwlabEventConsumerGroupId));",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED', String(kafka.enabled && kafka.features.directPublish));",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED', String(kafka.enabled && kafka.features.liveKafkaSse));",
" const refreshReplayEnabled = kafka.enabled && kafka.features.kafkaRefreshReplay;",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED', String(refreshReplayEnabled));",
" if (refreshReplayEnabled) {",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX', String(kafka.refreshReplay.groupIdPrefix));",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS', String(kafka.refreshReplay.timeoutMs));",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT', String(kafka.refreshReplay.scanLimit));",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT', String(kafka.refreshReplay.matchedEventLimit));",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT', String(kafka.refreshReplay.liveBufferLimit));",
" } else {",
" for (const envName of ['HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT', 'HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT']) checkCodeAgentRuntimeAbsent(item, file, container, envName);",
" }",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED', String(kafka.enabled && kafka.features.transactionalProjector));",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED', String(kafka.enabled && kafka.features.projectionOutboxRelay));",
" checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED', String(kafka.enabled && kafka.features.projectionRealtime));",
scripts/src/hwlab-node/web-probe-observe-actions.test.ts
+1
View File
@@ -23,6 +23,7 @@ test("realtime fanout runner contract derives topics, groups, and independent ca
assert.deepEqual(expectedKafka.capabilities, {
directPublish: true,
liveKafkaSse: true,
kafkaRefreshReplay: true,
transactionalProjector: false,
projectionOutboxRelay: false,
projectionRealtime: false,
scripts/src/hwlab-node/web-probe.ts
+27 -2
View File
@@ -1596,6 +1596,10 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti
}
if (item.value !== expected) mismatches.push({ name, expected, value: item.value, present: true, kind: "value" });
};
const expectAbsent = (name: string): void => {
const item = envByName.get(name);
if (item !== undefined) mismatches.push({ name, expected: null, value: item.value, present: true, kind: "unexpected-value" });
};
const expectSecret = (name: string, expectedSecret: string, expectedKey: string): void => {
const item = envByName.get(name);
if (item === undefined) {
@@ -1637,6 +1641,26 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti
expectValue("HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID", kafkaEventBridge.hwlabEventConsumerGroupId);
expectValue("HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.directPublish));
expectValue("HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.liveKafkaSse));
const refreshReplayEnabled = kafkaEventBridge.enabled && kafkaEventBridge.features.kafkaRefreshReplay;
expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(refreshReplayEnabled));
const refreshReplayEnvNames = [
"HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX",
"HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS",
"HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT",
"HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT",
"HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT",
];
if (refreshReplayEnabled) {
const refreshReplay = kafkaEventBridge.refreshReplay;
if (refreshReplay === undefined) throw new Error("codeAgentRuntime.kafkaEventBridge.refreshReplay is required when Kafka refresh replay is enabled");
expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_GROUP_PREFIX", refreshReplay.groupIdPrefix);
expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_TIMEOUT_MS", String(refreshReplay.timeoutMs));
expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_SCAN_LIMIT", String(refreshReplay.scanLimit));
expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_MATCHED_EVENT_LIMIT", String(refreshReplay.matchedEventLimit));
expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_LIVE_BUFFER_LIMIT", String(refreshReplay.liveBufferLimit));
} else {
for (const name of refreshReplayEnvNames) expectAbsent(name);
}
expectValue("HWLAB_KAFKA_TRANSACTIONAL_PROJECTOR_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.transactionalProjector));
expectValue("HWLAB_KAFKA_PROJECTION_OUTBOX_RELAY_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.projectionOutboxRelay));
expectValue("HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.projectionRealtime));
@@ -1646,7 +1670,7 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti
deploymentExists: true,
deploymentName: "hwlab-cloud-api",
containerName: "hwlab-cloud-api",
checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 14),
checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 21),
envCount: envByName.size,
mismatches,
result: compactCodeAgentDeploymentProbeResult(envProbe),
@@ -1688,6 +1712,7 @@ function nodeRuntimeCodeAgentCloudWebEnvStatus(spec: HwlabRuntimeLaneSpec, runti
if (item.value !== expected) mismatches.push({ name, expected, value: item.value, present: true, kind: "value" });
};
expectValue("HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.liveKafkaSse));
expectValue("HWLAB_WORKBENCH_KAFKA_REFRESH_REPLAY_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.kafkaRefreshReplay));
expectValue("HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED", String(kafkaEventBridge.enabled && kafkaEventBridge.features.projectionRealtime));
return {
required: true,
@@ -1695,7 +1720,7 @@ function nodeRuntimeCodeAgentCloudWebEnvStatus(spec: HwlabRuntimeLaneSpec, runti
deploymentExists: true,
deploymentName: "hwlab-cloud-web",
containerName: "hwlab-cloud-web",
checkedEnvCount: 2,
checkedEnvCount: 3,
envCount: envByName.size,
mismatches,
result: compactCodeAgentDeploymentProbeResult(envProbe),