From d5570244f09a387a4282799945e1c013c2cb37b2 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 15 Jul 2026 05:11:20 +0200 Subject: [PATCH 1/3] =?UTF-8?q?fix:=20=E6=81=A2=E5=A4=8D=20Workbench=20Kaf?= =?UTF-8?q?ka=20refresh=20replay=20=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/hwlab-node-lanes.yaml | 32 ++-- .../hwlab/runtime-gitops-postprocess.mjs | 53 ++++-- scripts/src/hwlab-node-lanes.ts | 116 +++++------- scripts/src/hwlab-node/render.ts | 67 ++++--- scripts/src/hwlab-node/web-probe.ts | 170 ++++++++---------- 5 files changed, 215 insertions(+), 223 deletions(-) diff --git a/config/hwlab-node-lanes.yaml b/config/hwlab-node-lanes.yaml index bb3faaa0..e6b72af8 100644 --- a/config/hwlab-node-lanes.yaml +++ b/config/hwlab-node-lanes.yaml @@ -211,30 +211,26 @@ lanes: codexStdioSupervisor: repo-owned kafkaEventBridge: enabled: true - transactionalProjector: - heartbeatIntervalMs: 2000 - projectionOutboxRelay: - intervalMs: 250 - batchSize: 100 - leaseMs: 30000 - sendTimeoutMs: 10000 - retryBackoffMs: 1000 - projectionRealtime: - outboxTailBatchSize: 100 - sseHeartbeatMs: 15000 - sseDrainTimeoutMs: 2500 - healthThresholds: - consumerLagWarning: 0 - outboxBacklogWarning: 0 - retryCountWarning: 0 - failedInboxWarning: 0 - dlqWarning: 0 + 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 agentRunEventTopic: agentrun.event.v1 hwlabEventTopic: hwlab.event.v1 clientId: hwlab-v03-cloud-api + directPublishConsumerGroupId: hwlab-v03-agentrun-event-direct-publish transactionalProjectorConsumerGroupId: hwlab-v03-agentrun-event-projector hwlabEventConsumerGroupId: hwlab-v03-workbench-live-sse publicExposure: diff --git a/scripts/native/hwlab/runtime-gitops-postprocess.mjs b/scripts/native/hwlab/runtime-gitops-postprocess.mjs index 604035d7..6388989f 100644 --- a/scripts/native/hwlab/runtime-gitops-postprocess.mjs +++ b/scripts/native/hwlab/runtime-gitops-postprocess.mjs @@ -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; diff --git a/scripts/src/hwlab-node-lanes.ts b/scripts/src/hwlab-node-lanes.ts index 3d3d9be3..4856c92a 100644 --- a/scripts/src/hwlab-node-lanes.ts +++ b/scripts/src/hwlab-node-lanes.ts @@ -559,27 +559,20 @@ export interface HwlabRuntimeKafkaShadowProducerSpec { export interface HwlabRuntimeKafkaEventBridgeSpec { readonly enabled: boolean; - readonly transactionalProjector?: { - readonly heartbeatIntervalMs: number; + readonly features: { + readonly directPublish: boolean; + readonly liveKafkaSse: boolean; + readonly kafkaRefreshReplay: boolean; + readonly transactionalProjector: boolean; + readonly projectionOutboxRelay: boolean; + readonly projectionRealtime: boolean; }; - readonly projectionOutboxRelay?: { - readonly intervalMs: number; - readonly batchSize: number; - readonly leaseMs: number; - readonly sendTimeoutMs: number; - readonly retryBackoffMs: number; - }; - readonly projectionRealtime?: { - readonly outboxTailBatchSize: number; - readonly sseHeartbeatMs: number; - readonly sseDrainTimeoutMs: number; - }; - readonly healthThresholds: { - readonly consumerLagWarning: number; - readonly outboxBacklogWarning: number; - readonly retryCountWarning: number; - readonly failedInboxWarning: number; - readonly dlqWarning: number; + readonly refreshReplay?: { + readonly groupIdPrefix: string; + readonly timeoutMs: number; + readonly scanLimit: number; + readonly matchedEventLimit: number; + readonly liveBufferLimit: number; }; readonly configRef: string; readonly bootstrapServers: string; @@ -587,6 +580,7 @@ export interface HwlabRuntimeKafkaEventBridgeSpec { readonly agentRunEventTopic: string; readonly hwlabEventTopic: string; readonly clientId: string; + readonly directPublishConsumerGroupId: string; readonly transactionalProjectorConsumerGroupId: string; readonly hwlabEventConsumerGroupId: string; } @@ -1245,75 +1239,55 @@ function codeAgentKafkaShadowProducerConfig(value: unknown, path: string): Hwlab function codeAgentKafkaEventBridgeConfig(value: unknown, path: string): HwlabRuntimeKafkaEventBridgeSpec { const raw = asRecord(value, path); - if (raw.refreshReplay !== undefined) throw new Error(`${path}.refreshReplay was removed with the fixed projection realtime architecture`); - const enabled = booleanField(raw, "enabled", path); - const transactionalProjectorConfig = raw.transactionalProjector === undefined + 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 - : codeAgentKafkaTransactionalProjectorConfig(raw.transactionalProjector, `${path}.transactionalProjector`); - const projectionOutboxRelayConfig = raw.projectionOutboxRelay === undefined - ? undefined - : codeAgentKafkaProjectionOutboxRelayConfig(raw.projectionOutboxRelay, `${path}.projectionOutboxRelay`); - const projectionRealtimeConfig = raw.projectionRealtime === undefined - ? undefined - : codeAgentKafkaProjectionRealtimeConfig(raw.projectionRealtime, `${path}.projectionRealtime`); - if (transactionalProjectorConfig === undefined) throw new Error(`${path}.transactionalProjector is required by the fixed transactional projection architecture`); - if (projectionOutboxRelayConfig === undefined) throw new Error(`${path}.projectionOutboxRelay is required by the fixed transactional projection architecture`); - if (projectionRealtimeConfig === undefined) throw new Error(`${path}.projectionRealtime is required by the fixed transactional projection architecture`); + : 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); - if (transactionalProjectorConsumerGroupId === hwlabEventConsumerGroupId) { - throw new Error(`${path} projector and realtime consumer group ids must be distinct`); + if (new Set([directPublishConsumerGroupId, transactionalProjectorConsumerGroupId, hwlabEventConsumerGroupId]).size !== 3) { + throw new Error(`${path} consumer group ids must be distinct so enabled capabilities receive independent event streams`); } return { - enabled, - transactionalProjector: transactionalProjectorConfig, - projectionOutboxRelay: projectionOutboxRelayConfig, - projectionRealtime: projectionRealtimeConfig, - healthThresholds: codeAgentKafkaHealthThresholdsConfig(raw.healthThresholds, `${path}.healthThresholds`), + enabled: booleanField(raw, "enabled", path), + features: { + directPublish: booleanField(features, "directPublish", `${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), agentRunEventTopic: stringField(raw, "agentRunEventTopic", path), hwlabEventTopic: stringField(raw, "hwlabEventTopic", path), clientId: stringField(raw, "clientId", path), + directPublishConsumerGroupId, transactionalProjectorConsumerGroupId, hwlabEventConsumerGroupId, }; } -function codeAgentKafkaTransactionalProjectorConfig(value: unknown, path: string): NonNullable { - const raw = asRecord(value, path); - return { heartbeatIntervalMs: numberField(raw, "heartbeatIntervalMs", path) }; -} - -function codeAgentKafkaProjectionOutboxRelayConfig(value: unknown, path: string): NonNullable { +function codeAgentKafkaRefreshReplayConfig(value: unknown, path: string): NonNullable { const raw = asRecord(value, path); return { - intervalMs: numberField(raw, "intervalMs", path), - batchSize: numberField(raw, "batchSize", path), - leaseMs: numberField(raw, "leaseMs", path), - sendTimeoutMs: numberField(raw, "sendTimeoutMs", path), - retryBackoffMs: numberField(raw, "retryBackoffMs", path), - }; -} - -function codeAgentKafkaProjectionRealtimeConfig(value: unknown, path: string): NonNullable { - const raw = asRecord(value, path); - return { - outboxTailBatchSize: numberField(raw, "outboxTailBatchSize", path), - sseHeartbeatMs: numberField(raw, "sseHeartbeatMs", path), - sseDrainTimeoutMs: numberField(raw, "sseDrainTimeoutMs", path), - }; -} - -function codeAgentKafkaHealthThresholdsConfig(value: unknown, path: string): HwlabRuntimeKafkaEventBridgeSpec["healthThresholds"] { - const raw = asRecord(value, path); - return { - consumerLagWarning: nonNegativeIntegerField(raw, "consumerLagWarning", path), - outboxBacklogWarning: nonNegativeIntegerField(raw, "outboxBacklogWarning", path), - retryCountWarning: nonNegativeIntegerField(raw, "retryCountWarning", path), - failedInboxWarning: nonNegativeIntegerField(raw, "failedInboxWarning", path), - dlqWarning: nonNegativeIntegerField(raw, "dlqWarning", path), + 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), }; } diff --git a/scripts/src/hwlab-node/render.ts b/scripts/src/hwlab-node/render.ts index bd3eb10a..588f53e0 100644 --- a/scripts/src/hwlab-node/render.ts +++ b/scripts/src/hwlab-node/render.ts @@ -1859,6 +1859,12 @@ export function nodeRuntimePipelinePostprocessScript(featureConfigSchemaStage: s "function cloudWebRuntimeEnvEntries() {", " const exposure = overlay.publicExposure;", " const entries = !exposure || !Array.isArray(exposure.extraProxies) ? [] : exposure.extraProxies.filter((proxy) => proxy && proxy.cloudWebEnvName && proxy.publicBaseUrl).map((proxy) => ({ name: String(proxy.cloudWebEnvName), value: String(proxy.publicBaseUrl) }));", + " 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;", "}", "function startupProbeFrom(probe) {", @@ -1956,35 +1962,31 @@ export function nodeRuntimePipelinePostprocessScript(featureConfigSchemaStage: s " if (codeAgentRuntime.kafkaEventBridge) {", " const kafka = codeAgentRuntime.kafkaEventBridge;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_ENABLED', String(kafka.enabled)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled)) || codeAgentRuntimeChanged;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled && (kafka.features.directPublish || kafka.features.transactionalProjector))) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_BOOTSTRAP_SERVERS', String(kafka.bootstrapServers)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_STDIO_TOPIC', String(kafka.stdioTopic)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC', String(kafka.agentRunEventTopic)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_EVENT_TOPIC', String(kafka.hwlabEventTopic)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_CLIENT_ID', String(kafka.clientId)) || codeAgentRuntimeChanged;", + " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID', String(kafka.directPublishConsumerGroupId)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_PROJECTOR_GROUP_ID', String(kafka.transactionalProjectorConsumerGroupId)) || codeAgentRuntimeChanged;", " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID', String(kafka.hwlabEventConsumerGroupId)) || codeAgentRuntimeChanged;", - " if (kafka.enabled) {", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS', String(kafka.transactionalProjector.heartbeatIntervalMs)) || 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_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS']) || codeAgentRuntimeChanged;", - " }", - " if (kafka.enabled) {", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS', String(kafka.projectionOutboxRelay.intervalMs)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE', String(kafka.projectionOutboxRelay.batchSize)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS', String(kafka.projectionOutboxRelay.leaseMs)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS', String(kafka.projectionOutboxRelay.sendTimeoutMs)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS', String(kafka.projectionOutboxRelay.retryBackoffMs)) || codeAgentRuntimeChanged;", - " } else {", - " codeAgentRuntimeChanged = removeEnvValues(container, ['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']) || codeAgentRuntimeChanged;", - " }", - " if (kafka.enabled) {", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE', String(kafka.projectionRealtime.outboxTailBatchSize)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_SSE_HEARTBEAT_MS', String(kafka.projectionRealtime.sseHeartbeatMs)) || codeAgentRuntimeChanged;", - " codeAgentRuntimeChanged = setEnvValue(container, 'HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS', String(kafka.projectionRealtime.sseDrainTimeoutMs)) || codeAgentRuntimeChanged;", - " } else {", - " codeAgentRuntimeChanged = removeEnvValues(container, ['HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE', 'HWLAB_WORKBENCH_SSE_HEARTBEAT_MS', 'HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS']) || codeAgentRuntimeChanged;", + " 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;", " }", " }", " }", @@ -2331,6 +2333,12 @@ export function nodeRuntimePipelinePostprocessScript(featureConfigSchemaStage: s "function cloudWebRuntimeEnvEntries() {", " const exposure = overlay.publicExposure;", " const entries = !exposure || !Array.isArray(exposure.extraProxies) ? [] : exposure.extraProxies.filter((proxy) => proxy && proxy.cloudWebEnvName && proxy.publicBaseUrl).map((proxy) => ({ name: String(proxy.cloudWebEnvName), value: String(proxy.publicBaseUrl) }));", + " 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;", "}", "function workloadRef(item, file, container) { return { file, kind: item && item.kind, name: item && item.metadata && item.metadata.name, container: container && container.name }; }", @@ -2404,14 +2412,31 @@ export function nodeRuntimePipelinePostprocessScript(featureConfigSchemaStage: s " if (codeAgentRuntime.kafkaEventBridge) {", " const kafka = codeAgentRuntime.kafkaEventBridge;", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_ENABLED', String(kafka.enabled));", - " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled));", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED', String(kafka.enabled && (kafka.features.directPublish || kafka.features.transactionalProjector)));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_BOOTSTRAP_SERVERS', String(kafka.bootstrapServers));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_STDIO_TOPIC', String(kafka.stdioTopic));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC', String(kafka.agentRunEventTopic));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_EVENT_TOPIC', String(kafka.hwlabEventTopic));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_CLIENT_ID', String(kafka.clientId));", + " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID', String(kafka.directPublishConsumerGroupId));", " checkCodeAgentRuntimeValue(item, file, container, 'HWLAB_KAFKA_PROJECTOR_GROUP_ID', String(kafka.transactionalProjectorConsumerGroupId));", " 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));", " }", " }", " }", diff --git a/scripts/src/hwlab-node/web-probe.ts b/scripts/src/hwlab-node/web-probe.ts index 19f2b921..fb9a7346 100644 --- a/scripts/src/hwlab-node/web-probe.ts +++ b/scripts/src/hwlab-node/web-probe.ts @@ -1493,7 +1493,6 @@ export function nodeRuntimeCodeAgentRuntimeStatus(spec: HwlabRuntimeLaneSpec, na const apiKey = secretKeyStatus(spec, runtime.apiKeySecretName, runtime.apiKeySecretKey); const cloudApiEnv = nodeRuntimeCodeAgentCloudApiEnvStatus(spec, runtime); const cloudWebEnv = nodeRuntimeCodeAgentCloudWebEnvStatus(spec, runtime); - const kafkaAuthority = nodeRuntimeKafkaAuthorityStatus(spec, runtime); const ready = runnerNamespace.exitCode === 0 && secretNamespace.exitCode === 0 && managerNamespace.exitCode === 0 @@ -1550,73 +1549,12 @@ export function nodeRuntimeCodeAgentRuntimeStatus(spec: HwlabRuntimeLaneSpec, na apiKey, cloudApiEnv, cloudWebEnv, - kafkaAuthority, repoUrlFrom: runtime.repoUrlFrom, providerIdFrom: runtime.providerIdFrom, valuesPrinted: false, }; } -function nodeRuntimeKafkaAuthorityStatus(spec: HwlabRuntimeLaneSpec, runtime: NonNullable): Record { - const kafka = runtime.kafkaEventBridge; - if (kafka === undefined || kafka.enabled === false) return { required: false, authority: kafka === undefined ? null : "transactional-projection", valuesPrinted: false }; - const healthUrl = `${spec.publicApiUrl.replace(/\/+$/u, "")}/health/ready`; - const result = runCommand(["curl", "-k", "-sS", "--connect-timeout", "5", "--max-time", "12", healthUrl], repoRoot, { timeoutMs: 15_000 }); - let health: Record = {}; - try { - health = result.exitCode === 0 ? record(JSON.parse(result.stdout)) : {}; - } catch { - health = {}; - } - const projector = record(health.kafkaProjector); - const components = Array.isArray(projector.components) ? projector.components.map(record) : []; - const observations = [projector, ...components]; - const consumerLag = observations.map((item) => record(item.consumerLag).total).find((value) => typeof value === "number") ?? null; - const backlogCount = observations.map((item) => item.backlogCount).find((value) => typeof value === "number") ?? null; - const retryCount = observations.map((item) => item.retryCount).find((value) => typeof value === "number") ?? null; - const failedInboxCount = observations.map((item) => item.failedInboxCount).find((value) => typeof value === "number") ?? null; - const dlqCount = observations.map((item) => item.dlqCount).find((value) => typeof value === "number") ?? null; - const thresholds = kafka.healthThresholds; - const warningReasons = [ - typeof consumerLag === "number" && consumerLag > thresholds.consumerLagWarning ? "consumer-lag" : null, - typeof backlogCount === "number" && backlogCount > thresholds.outboxBacklogWarning ? "outbox-backlog" : null, - typeof retryCount === "number" && retryCount > thresholds.retryCountWarning ? "outbox-retry" : null, - typeof failedInboxCount === "number" && failedInboxCount > thresholds.failedInboxWarning ? "failed-inbox" : null, - typeof dlqCount === "number" && dlqCount > thresholds.dlqWarning ? "dlq" : null, - ].filter((value): value is string => value !== null); - return { - required: true, - authority: "transactional-projection", - fixedGroups: { - transactionalProjector: kafka.transactionalProjectorConsumerGroupId, - liveSse: kafka.hwlabEventConsumerGroupId, - }, - capabilities: { - transactionalProjector: true, - projectionOutboxRelay: true, - projectionRealtime: true, - }, - liveReplay: { - live: true, - replay: true, - cursor: "projection-outbox-seq", - }, - healthObserved: Object.keys(projector).length > 0, - healthStatus: projector.status ?? null, - consumerLag, - backlogCount, - retryCount, - failedInboxCount, - dlqCount, - thresholds, - warning: warningReasons.length > 0, - warningReasons, - blocking: false, - result: compactRuntimeCommand(result), - valuesPrinted: false, - }; -} - function nodeRuntimeCodeAgentManagerTarget(managerUrl: string, fallbackNamespace: string): { namespace: string; serviceName: string } { let parsed: URL | null = null; try { @@ -1705,55 +1643,47 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti const kafkaEventBridge = runtime.kafkaEventBridge; if (kafkaEventBridge !== undefined) { expectValue("HWLAB_KAFKA_ENABLED", String(kafkaEventBridge.enabled)); - expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafkaEventBridge.enabled)); + expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_CONSUME_ENABLED", String(kafkaEventBridge.enabled && (kafkaEventBridge.features.directPublish || kafkaEventBridge.features.transactionalProjector))); expectValue("HWLAB_KAFKA_BOOTSTRAP_SERVERS", kafkaEventBridge.bootstrapServers); expectValue("HWLAB_KAFKA_STDIO_TOPIC", kafkaEventBridge.stdioTopic); expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC", kafkaEventBridge.agentRunEventTopic); expectValue("HWLAB_KAFKA_EVENT_TOPIC", kafkaEventBridge.hwlabEventTopic); expectValue("HWLAB_KAFKA_CLIENT_ID", kafkaEventBridge.clientId); + expectValue("HWLAB_KAFKA_AGENTRUN_EVENT_GROUP_ID", kafkaEventBridge.directPublishConsumerGroupId); expectValue("HWLAB_KAFKA_PROJECTOR_GROUP_ID", kafkaEventBridge.transactionalProjectorConsumerGroupId); expectValue("HWLAB_KAFKA_HWLAB_EVENT_GROUP_ID", kafkaEventBridge.hwlabEventConsumerGroupId); - if (kafkaEventBridge.enabled) { - if (kafkaEventBridge.transactionalProjector === undefined) throw new Error("codeAgentRuntime.kafkaEventBridge.transactionalProjector is required when transactional projector is enabled"); - expectValue("HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS", String(kafkaEventBridge.transactionalProjector.heartbeatIntervalMs)); - } else { - expectAbsent("HWLAB_KAFKA_PROJECTOR_HEARTBEAT_INTERVAL_MS"); - } - const relayEnvNames = [ - "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", + 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 (kafkaEventBridge.enabled) { - const relay = kafkaEventBridge.projectionOutboxRelay; - if (relay === undefined) throw new Error("codeAgentRuntime.kafkaEventBridge.projectionOutboxRelay is required when projection outbox relay is enabled"); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_INTERVAL_MS", String(relay.intervalMs)); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_BATCH_SIZE", String(relay.batchSize)); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_LEASE_MS", String(relay.leaseMs)); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_SEND_TIMEOUT_MS", String(relay.sendTimeoutMs)); - expectValue("HWLAB_KAFKA_OUTBOX_RELAY_RETRY_BACKOFF_MS", String(relay.retryBackoffMs)); + 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 relayEnvNames) expectAbsent(name); - } - const projectionRealtimeEnvNames = ["HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE", "HWLAB_WORKBENCH_SSE_HEARTBEAT_MS", "HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS"]; - if (kafkaEventBridge.enabled) { - const realtime = kafkaEventBridge.projectionRealtime; - if (realtime === undefined) throw new Error("codeAgentRuntime.kafkaEventBridge.projectionRealtime is required when projection realtime is enabled"); - expectValue("HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE", String(realtime.outboxTailBatchSize)); - expectValue("HWLAB_WORKBENCH_SSE_HEARTBEAT_MS", String(realtime.sseHeartbeatMs)); - expectValue("HWLAB_WORKBENCH_SSE_DRAIN_TIMEOUT_MS", String(realtime.sseDrainTimeoutMs)); - } else { - for (const name of projectionRealtimeEnvNames) expectAbsent(name); + 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)); } return { ready: mismatches.length === 0, deploymentExists: true, deploymentName: "hwlab-cloud-api", containerName: "hwlab-cloud-api", - checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 22), + checkedEnvCount: 9 + (kafkaShadowProducer === undefined ? 0 : 5) + (kafkaEventBridge === undefined ? 0 : 21), envCount: envByName.size, mismatches, result: compactCodeAgentDeploymentProbeResult(envProbe), @@ -1762,9 +1692,53 @@ function nodeRuntimeCodeAgentCloudApiEnvStatus(spec: HwlabRuntimeLaneSpec, runti } function nodeRuntimeCodeAgentCloudWebEnvStatus(spec: HwlabRuntimeLaneSpec, runtime: NonNullable): Record { - void spec; - void runtime; - return { required: false, ready: true, checkedEnvCount: 0, valuesPrinted: false }; + const kafkaEventBridge = runtime.kafkaEventBridge; + if (kafkaEventBridge === undefined) return { required: false, ready: true, checkedEnvCount: 0, valuesPrinted: false }; + const envProbe = runNodeK3sArgs(spec, [ + "kubectl", + "-n", + spec.runtimeNamespace, + "get", + "deployment", + "hwlab-cloud-web", + "-o", + 'jsonpath={range .spec.template.spec.containers[?(@.name=="hwlab-cloud-web")].env[*]}{.name}{"\\t"}{.value}{"\\t"}{.valueFrom.secretKeyRef.name}{"\\t"}{.valueFrom.secretKeyRef.key}{"\\n"}{end}', + ], 60); + if (envProbe.exitCode !== 0) { + return { + required: true, + ready: false, + deploymentExists: false, + mismatches: [{ name: "hwlab-cloud-web", reason: "deployment-missing" }], + result: compactCodeAgentDeploymentProbeResult(envProbe), + valuesPrinted: false, + }; + } + const envByName = parseCodeAgentEnvProbe(envProbe.stdout); + const mismatches: Array> = []; + const expectValue = (name: string, expected: string): void => { + const item = envByName.get(name); + if (item === undefined) { + mismatches.push({ name, expected, present: false, kind: "value" }); + return; + } + 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, + ready: mismatches.length === 0, + deploymentExists: true, + deploymentName: "hwlab-cloud-web", + containerName: "hwlab-cloud-web", + checkedEnvCount: 3, + envCount: envByName.size, + mismatches, + result: compactCodeAgentDeploymentProbeResult(envProbe), + valuesPrinted: false, + }; } type CodeAgentEnvProbeRow = { From f44dca58a29d0864a5e490cc57e403ac6143e311 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 15 Jul 2026 05:16:47 +0200 Subject: [PATCH 2/3] docs(pikaoa): track HTTP web login fix --- docs/MDTODO/pikaoa-enterprise-platform.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs/MDTODO/pikaoa-enterprise-platform.md b/docs/MDTODO/pikaoa-enterprise-platform.md index 0ab97215..7b234812 100644 --- a/docs/MDTODO/pikaoa-enterprise-platform.md +++ b/docs/MDTODO/pikaoa-enterprise-platform.md @@ -126,6 +126,9 @@ ##### R1.5.9.11 [in_progress] 修复合同迁移中 PL/pgSQL trigger function 的 Goose StatementBegin/StatementEnd 边界并完成 00001-00007 真实迁移验收（pikainc/pikaoa#20），完成任务后将详细报告写入[任务报告](./details/pikaoa-enterprise-platform/R1.5.9.11_Task_Report.md)。 +##### R1.5.9.12 [in_progress] + +修复 PikaOA HTTP hostIP 测试入口因 crypto.randomUUID 不可用导致 Web 登录失败的问题，并完成桌面与移动端真实入口验收（pikainc/pikaoa#22），完成任务后将详细报告写入[任务报告](./details/pikaoa-enterprise-platform/R1.5.9.12_Task_Report.md)。 ### R1.6 审核并合并双仓 PR，执行首次受控 PaC bootstrap，由正常 source PR merge 自动发布，依赖 R1.3-R1.5，完成任务后将详细报告写入[任务报告](./details/pikaoa-enterprise-platform/R1.6_Task_Report.md)。 From 684553f9a0e034c6c451a094d05183c228e91f8a Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 15 Jul 2026 05:17:40 +0200 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20=E4=BF=AE=E6=AD=A3=20PikaOA=20?= =?UTF-8?q?=E5=8F=AF=E8=A7=82=E6=B5=8B=E6=80=A7=E5=8F=91=E7=8E=B0=E5=A3=B0?= =?UTF-8?q?=E6=98=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../pikaoa-observability-invalid-label.yaml | 5 + config/fixtures/pikaoa-test-target.yaml | 3 +- config/pikaoa.yaml | 5 +- .../R1.5.9.10_Task_Report.md | 41 +++++++ scripts/src/pikaoa-test-target-async.test.ts | 65 +++++++++- scripts/src/pikaoa-test-target.ts | 112 ++++++++++++++++-- 6 files changed, 217 insertions(+), 14 deletions(-) create mode 100644 config/fixtures/pikaoa-observability-invalid-label.yaml create mode 100644 docs/MDTODO/details/pikaoa-enterprise-platform/R1.5.9.10_Task_Report.md diff --git a/config/fixtures/pikaoa-observability-invalid-label.yaml b/config/fixtures/pikaoa-observability-invalid-label.yaml new file mode 100644 index 00000000..3da4cab1 --- /dev/null +++ b/config/fixtures/pikaoa-observability-invalid-label.yaml @@ -0,0 +1,5 @@ +metricsBackend: + discovery: + labelSelector: + key: invalid label key + value: "true" diff --git a/config/fixtures/pikaoa-test-target.yaml b/config/fixtures/pikaoa-test-target.yaml index 9b17b945..37b81d12 100644 --- a/config/fixtures/pikaoa-test-target.yaml +++ b/config/fixtures/pikaoa-test-target.yaml @@ -97,8 +97,9 @@ testRuntime: hostIP: 192.0.2.10 port: 32080 observability: - otlpEndpoint: http://otel-collector.observability.svc.cluster.local:4318 + otlpEndpoint: otel-collector.observability.svc.cluster.local:4317 prometheus: + configRef: config/platform-infra/observability.yaml#metricsBackend.discovery.labelSelector scrape: true apiPath: /metrics workerPath: /metrics diff --git a/config/pikaoa.yaml b/config/pikaoa.yaml index e3a4a12a..78698329 100644 --- a/config/pikaoa.yaml +++ b/config/pikaoa.yaml @@ -117,8 +117,9 @@ testRuntime: port: 32080 observability: configRef: config/platform-infra/observability.yaml - otlpEndpoint: http://otel-collector.platform-infra.svc.cluster.local:4318 + otlpEndpoint: otel-collector.platform-infra.svc.cluster.local:4317 prometheus: + configRef: config/platform-infra/observability.yaml#metricsBackend.discovery.labelSelector scrape: true apiPath: /metrics workerPath: /metrics @@ -228,7 +229,7 @@ delivery: configRef: config/platform-db/postgres-pk01.yaml#exports.connectionStrings.pikaoa-database-url observability: configRef: config/platform-infra/observability.yaml - otlpHttpEndpoint: http://otel-collector.platform-infra.svc.cluster.local:4318 + otlpEndpoint: otel-collector.platform-infra.svc.cluster.local:4317 prometheusScrape: true propagation: - tracecontext diff --git a/docs/MDTODO/details/pikaoa-enterprise-platform/R1.5.9.10_Task_Report.md b/docs/MDTODO/details/pikaoa-enterprise-platform/R1.5.9.10_Task_Report.md new file mode 100644 index 00000000..a0ada990 --- /dev/null +++ b/docs/MDTODO/details/pikaoa-enterprise-platform/R1.5.9.10_Task_Report.md @@ -0,0 +1,41 @@ +# R1.5.9.10 任务报告 + +## 状态 + +- 当前状态：`in_progress`。 +- 关联 issue：`pikasTech/unidesk#2092`。 +- 实现分支：`fix/pikaoa-observability`。 +- 本阶段仅完成源码、owning YAML、fixture 和定向验证；未执行任何运行面 mutation，未启动 Prometheus、Collector、Tempo 或 PikaOA Pod。 +- 后续仍需主代理通过受控 observability apply 完成 NC01 trace/metrics 原入口验收后再决定完成状态。 + +## 实现 + +- `config/pikaoa.yaml`： + - 测试与生产 `otlpEndpoint` 统一为无 scheme 的 OTLP gRPC endpoint `otel-collector.platform-infra.svc.cluster.local:4317`； + - 删除生产旧字段 `otlpHttpEndpoint`，不保留兼容入口； + - 测试 Prometheus 新增 `configRef`，指向 `config/platform-infra/observability.yaml#metricsBackend.discovery.labelSelector`。 +- `scripts/src/pikaoa-test-target.ts`： + - 复用共享 `scripts/src/ops/config-refs.ts` 的 `resolveConfigRef`； + - 校验 configRef、OTLP gRPC endpoint 及 Kubernetes label key/value； + - 将 selector label 加入 API/Worker Pod template，Web 不添加； + - 在 plan/status 投影 selector 的 `sourceRef`、`presence`、`fingerprint` 和 `valuesPrinted=false`； + - selector 漂移与 OTel exporter 失败均保持 `blocking=false` warning，不阻塞业务 MVP。 +- fixture 与测试： + - 更新 endpoint 与 selector configRef； + - 增加非法 label fixture； + - 覆盖 endpoint、API/Worker selector label、Web 无 selector、configRef 缺失、引用路径错误、非法 Kubernetes label、旧生产字段删除及 `valuesPrinted=false`。 + +## 验证 + +- `bun --check scripts/src/pikaoa-test-target.ts`：通过。 +- `bun test scripts/src/pikaoa-test-target-async.test.ts`：`2 pass, 0 fail`。 +- `bun scripts/cli.ts check --syntax-only`：`11 passed, 0 failed`。 +- `bun scripts/cli.ts pikaoa test-target plan ...` compact text：通过，`mutation=false`、`valuesPrinted=false`；clean Artificer workspace 缺 fixture Secret 时仅显示既有非变更 blocker，不读取或打印 Secret，也不作为本任务运行面门禁。 +- `git diff --check`：通过。 +- `scripts/src/pikaoa-test-target.ts`：1910 行，低于 2000 行。 + +## 边界与后续 + +- 用户确认开发初期旧数据可丢；本任务不实现旧数据迁移、兼容或回滚，现有 migration 命名仅表示全新数据库初始化。 +- Prometheus 当前未 apply/ready 不阻塞业务 MVP；后续由主代理按受控入口完成 apply、Tempo trace 和 Prometheus metrics 验收。 +- 本报告回链 MDTODO `R1.5.9.10`，任务保持进行中。 diff --git a/scripts/src/pikaoa-test-target-async.test.ts b/scripts/src/pikaoa-test-target-async.test.ts index e2e4e34c..61ad054c 100644 --- a/scripts/src/pikaoa-test-target-async.test.ts +++ b/scripts/src/pikaoa-test-target-async.test.ts @@ -162,13 +162,15 @@ exit 64 return JSON.parse(result.renderedText) as Record; }; const waitFor = async (instance: string, expectedCode: string, step = "all"): Promise> => { + let lastStatus: Record | null = null; for (let attempt = 0; attempt < 80; attempt += 1) { const value = await projection("status", instance, false, step); const status = value.status as Record; if (status.code === expectedCode) return value; + lastStatus = status; await Bun.sleep(25); } - assert.fail(`status did not reach ${expectedCode}`); + assert.fail(`status did not reach ${expectedCode}: ${JSON.stringify(lastStatus)}`); }; try { @@ -178,9 +180,21 @@ exit 64 const sources = secret.sources as Array>; assert.equal(sources.find((source) => source.purpose === "database.connection")?.presence, true); const objects = plan.objects as Array>; + const observability = plan.observability as Record; + const prometheus = observability.prometheus as Record; + const selector = prometheus.selector as Record; assert.equal(objects.some((object) => object.kind === "StatefulSet" || object.name === "pikaoa-postgres"), false); assert.equal(objects.some((object) => object.kind === "Deployment" && object.name === "pikaoa-web"), true); assert.equal(objects.find((object) => object.kind === "Deployment" && object.name === "pikaoa-api")?.podFsGroup, 65532); + assert.equal(observability.otlpEndpoint, "otel-collector.observability.svc.cluster.local:4317"); + assert.equal(selector.sourceRef, "config/platform-infra/observability.yaml#metricsBackend.discovery.labelSelector"); + assert.equal(selector.presence, true); + assert.match(String(selector.fingerprint), /^sha256:[0-9a-f]{64}$/u); + assert.equal(selector.valuesPrinted, false); + for (const component of ["pikaoa-api", "pikaoa-worker"]) { + assert.equal(objects.find((object) => object.kind === "Deployment" && object.name === component)?.prometheusSelectorLabel, selector.value); + } + assert.equal(objects.find((object) => object.kind === "Deployment" && object.name === "pikaoa-web")?.prometheusSelectorLabel, undefined); assert.deepEqual(plan.exposure, { serviceType: "NodePort", serviceName: "pikaoa-web", hostIP: "192.0.2.10", port: 32080 }); assert.deepEqual(plan.webApiUpstream, { envName: "PIKAOA_API_UPSTREAM", value: "http://pikaoa-api:8080", scope: "same-namespace" }); @@ -387,6 +401,7 @@ test("validation-only fixture never invokes remote capture", async () => { valuesPrinted: false, }); assert.equal(planPayload.valuesPrinted, false); + assert.equal(((planPayload.observability as Record).prometheus as Record).selector !== undefined, true); assert.equal(JSON.stringify(planPayload).includes("fixture-db-password"), false); assert.equal(JSON.stringify(planPayload).includes("postgres://"), false); @@ -407,6 +422,54 @@ test("validation-only fixture never invokes remote capture", async () => { rmSync(missingSchemaRoot, { recursive: true, force: true }); } + const missingSelectorRoot = mkdtempSync(join(tmpdir(), "pikaoa-test-target-missing-selector-")); + const missingSelectorPath = join(missingSelectorRoot, "pikaoa.yaml"); + const fixture = readFileSync(rootPath("config", "fixtures", "pikaoa-test-target.yaml"), "utf8"); + writeFileSync(missingSelectorPath, fixture.replace(" configRef: config/platform-infra/observability.yaml#metricsBackend.discovery.labelSelector\n", "")); + try { + await assert.rejects( + runPikaoaCommand({} as UniDeskConfig, ["test-target", "status", "--config", missingSelectorPath, "--target", "TEST01", "--instance", "missing-selector", "--output", "json"], forbiddenCapture), + /observability\.prometheus\.configRef 必须是非空字符串/u, + ); + } finally { + rmSync(missingSelectorRoot, { recursive: true, force: true }); + } + + const invalidSelectorRoot = mkdtempSync(join(tmpdir(), "pikaoa-test-target-invalid-selector-")); + const invalidSelectorPath = join(invalidSelectorRoot, "pikaoa.yaml"); + writeFileSync(invalidSelectorPath, fixture.replace( + "config/platform-infra/observability.yaml#metricsBackend.discovery.labelSelector", + "config/fixtures/pikaoa-observability-invalid-label.yaml#metricsBackend.discovery.labelSelector", + )); + try { + await assert.rejects( + runPikaoaCommand({} as UniDeskConfig, ["test-target", "status", "--config", invalidSelectorPath, "--target", "TEST01", "--instance", "invalid-selector", "--output", "json"], forbiddenCapture), + /observability\.prometheus\.configRef\.key 必须是合法 Kubernetes label key/u, + ); + } finally { + rmSync(invalidSelectorRoot, { recursive: true, force: true }); + } + + const missingSelectorPathRoot = mkdtempSync(join(tmpdir(), "pikaoa-test-target-missing-selector-path-")); + const missingSelectorConfigPath = join(missingSelectorPathRoot, "pikaoa.yaml"); + writeFileSync(missingSelectorConfigPath, fixture.replace("metricsBackend.discovery.labelSelector", "metricsBackend.discovery.missingSelector")); + try { + await assert.rejects( + runPikaoaCommand({} as UniDeskConfig, ["test-target", "status", "--config", missingSelectorConfigPath, "--target", "TEST01", "--instance", "missing-selector-path", "--output", "json"], forbiddenCapture), + /observability\.prometheus\.configRef 解析失败.*missingSelector/u, + ); + } finally { + rmSync(missingSelectorPathRoot, { recursive: true, force: true }); + } + + const owningConfig = Bun.YAML.parse(readFileSync(rootPath("config", "pikaoa.yaml"), "utf8")) as Record; + const testTarget = (((owningConfig.testRuntime as Record).targets as Record).NC01 as Record); + assert.equal((testTarget.observability as Record).otlpEndpoint, "otel-collector.platform-infra.svc.cluster.local:4317"); + const production = (((owningConfig.delivery as Record).targets as Record).NC01 as Record); + const productionObservability = (production.runtime as Record).observability as Record; + assert.equal(productionObservability.otlpEndpoint, "otel-collector.platform-infra.svc.cluster.local:4317"); + assert.equal("otlpHttpEndpoint" in productionObservability, false); + const unconfiguredRoot = mkdtempSync(join(tmpdir(), "pikaoa-test-target-unconfigured-")); const unconfiguredPath = join(unconfiguredRoot, "pikaoa.yaml"); writeFileSync(unconfiguredPath, "version: 1\nkind: pikaoa-platform-delivery\ntestRuntime:\n defaultTargetId: null\n targets: {}\n"); diff --git a/scripts/src/pikaoa-test-target.ts b/scripts/src/pikaoa-test-target.ts index 37a33861..1ea5c84d 100644 --- a/scripts/src/pikaoa-test-target.ts +++ b/scripts/src/pikaoa-test-target.ts @@ -9,6 +9,7 @@ import { renderMachine } from "./cicd-render"; import type { RenderedCliResult } from "./output"; import { CliInputError } from "./output"; import { capture, compactCapture, parseJsonOutput, readYamlRecord, sha256Fingerprint, shQuote } from "./platform-infra-ops-library"; +import { resolveConfigRef } from "./ops/config-refs"; type TestTargetAction = "plan" | "status" | "start" | "stop"; type TestTargetStep = "all" | "foundation" | "migration" | "api" | "worker" | "web"; @@ -31,6 +32,15 @@ interface SecretSourceSpec { targetKey: string; } +interface PrometheusSelectorSpec { + sourceRef: string; + key: string; + value: string; + presence: true; + fingerprint: string; + valuesPrinted: false; +} + interface TestTargetSpec { id: string; enabled: boolean; @@ -107,6 +117,7 @@ interface TestTargetSpec { observability: { otlpEndpoint: string; prometheusScrape: boolean; + prometheusSelector: PrometheusSelectorSpec; apiMetricsPath: string; workerMetricsPath: string; exporterWarning: boolean; @@ -387,8 +398,9 @@ function parseTarget(id: string, root: Record, configLabel: str port: positiveInteger(exposure.port, `${path}.exposure.port`, 65_535), }, observability: { - otlpEndpoint: nonEmpty(observability.otlpEndpoint, `${path}.observability.otlpEndpoint`), + otlpEndpoint: otlpGrpcEndpoint(observability.otlpEndpoint, `${path}.observability.otlpEndpoint`), prometheusScrape: boolean(prometheus.scrape, `${path}.observability.prometheus.scrape`), + prometheusSelector: prometheusSelector(prometheus.configRef, `${path}.observability.prometheus`), apiMetricsPath: httpPath(prometheus.apiPath, `${path}.observability.prometheus.apiPath`), workerMetricsPath: httpPath(prometheus.workerPath, `${path}.observability.prometheus.workerPath`), exporterWarning: boolean(exporterFailure.warning, `${path}.observability.exporterFailure.warning`), @@ -439,6 +451,7 @@ function planPayload(options: TestTargetOptions, selection: Selection): Record objectSummary(object, target.observability.prometheusSelector.key)), selectors, secret: { name: target.runtime.secretName, @@ -504,8 +517,7 @@ function renderedPlan(selection: Selection, context: RenderContext): Record { + return { + otlpEndpoint: target.observability.otlpEndpoint, + prometheus: { + scrape: target.observability.prometheusScrape, + selector: target.observability.prometheusSelector, + }, + valuesPrinted: false, + }; +} + +function prometheusRuntimeStatus(items: Record[], selector: PrometheusSelectorSpec, runtimePresent: boolean, step: TestTargetStep): Record { + const components = step === "all" ? ["api", "worker"] : step === "api" || step === "worker" ? [step] : []; + const workloads = components.map((component) => { + const deployment = items.find((item) => item.kind === "Deployment" && optionalRecord(item.metadata, "resource.metadata")?.name === `pikaoa-${component}`); + const spec = deployment === undefined ? null : optionalRecord(deployment.spec, "resource.spec"); + const template = spec === null ? null : optionalRecord(spec.template, "resource.spec.template"); + const metadata = template === null ? null : optionalRecord(template.metadata, "resource.spec.template.metadata"); + const labels = metadata === null ? null : optionalRecord(metadata.labels, "resource.spec.template.metadata.labels"); + const actual = labels?.[selector.key]; + return { component, present: deployment !== undefined, actual: typeof actual === "string" ? actual : null, matches: actual === selector.value }; + }); + const checked = runtimePresent && components.length > 0; + const drift = checked && workloads.some((workload) => !workload.matches); + return { + selector, + checked, + workloads, + drift, + warning: drift ? warning("prometheus-selector-label-drift", "API/Worker Pod template 的 Prometheus selector label 与 owning configRef 不一致；该漂移不阻塞业务运行。") : null, + blocking: false, + valuesPrinted: false, + }; +} + function buildManifest(target: TestTargetSpec, context: RenderContext, secrets: SecretMaterial): Record[] { const labels = ownershipLabels(target, context); const secretData = runtimeSecretData(target, context, secrets); @@ -677,6 +724,9 @@ function buildManifest(target: TestTargetSpec, context: RenderContext, secrets: const workerSelector = componentSelector("worker", context.instanceId); const webSelector = componentSelector("web", context.instanceId); const migrationSelector = componentSelector("migration", context.instanceId); + const prometheusPodLabels = target.observability.prometheusScrape ? { + [target.observability.prometheusSelector.key]: target.observability.prometheusSelector.value, + } : {}; const commonPodAnnotations = (path: string, port: number): Record => target.observability.prometheusScrape ? { "prometheus.io/scrape": "true", "prometheus.io/path": path, @@ -728,7 +778,7 @@ function buildManifest(target: TestTargetSpec, context: RenderContext, secrets: }, }, }, - deployment(target, context, "api", context.apiImage, apiSelector, target.runtime.apiPort, commonPodAnnotations(target.observability.apiMetricsPath, target.runtime.apiPort), [ + deployment(target, context, "api", context.apiImage, apiSelector, target.runtime.apiPort, prometheusPodLabels, commonPodAnnotations(target.observability.apiMetricsPath, target.runtime.apiPort), [ { name: "PIKAOA_CONFIG", value: "/etc/pikaoa/pikaoa.yaml" }, { name: "PIKAOA_SESSION_SECRET", valueFrom: { secretKeyRef: { name: target.runtime.secretName, key: target.runtime.sessionSecret.targetKey } } }, { name: "PIKAOA_BOOTSTRAP_ADMIN_USERNAME", value: target.runtime.administrator.username }, @@ -736,10 +786,10 @@ function buildManifest(target: TestTargetSpec, context: RenderContext, secrets: { name: "PIKAOA_BOOTSTRAP_EMPLOYEE_USERNAME", value: target.runtime.employee.username }, { name: "PIKAOA_BOOTSTRAP_EMPLOYEE_PASSWORD", valueFrom: { secretKeyRef: { name: target.runtime.secretName, key: target.runtime.employee.password.targetKey } } }, ], target.probes.apiLivenessPath, target.probes.apiReadinessPath), - deployment(target, context, "worker", context.workerImage, workerSelector, target.runtime.workerMetricsPort, commonPodAnnotations(target.observability.workerMetricsPath, target.runtime.workerMetricsPort), [ + deployment(target, context, "worker", context.workerImage, workerSelector, target.runtime.workerMetricsPort, prometheusPodLabels, commonPodAnnotations(target.observability.workerMetricsPath, target.runtime.workerMetricsPort), [ { name: "PIKAOA_CONFIG", value: "/etc/pikaoa/pikaoa.yaml" }, ], target.probes.workerLivenessPath, target.probes.workerReadinessPath), - deployment(target, context, "web", context.webImage, webSelector, target.runtime.webPort, {}, [ + deployment(target, context, "web", context.webImage, webSelector, target.runtime.webPort, {}, {}, [ { name: target.runtime.webApiUpstreamEnv, value: `http://${target.runtime.apiServiceName}:${target.runtime.apiPort}` }, ], target.probes.webLivenessPath, target.probes.webReadinessPath), ]; @@ -765,6 +815,7 @@ function deployment( image: string, selector: Record, port: number, + podLabels: Record, annotations: Record, env: Array>, livenessPath: string, @@ -776,7 +827,7 @@ function deployment( spec: { replicas: 1, selector: { matchLabels: selector }, template: { - metadata: { labels: { ...labels, ...selector }, annotations }, + metadata: { labels: { ...labels, ...selector, ...podLabels }, annotations }, spec: { ...(component === "api" ? { securityContext: { fsGroup: target.runtime.attachment.fsGroup } } : {}), containers: [{ @@ -1419,6 +1470,7 @@ function summarizeRemoteStatus(parsed: Record, target: TestTarg const labels = optionalRecord(metadata.labels, "status.namespace.metadata.labels") ?? {}; const resources = optionalRecord(runtime.resources, "status.runtime.resources") ?? {}; const items = Array.isArray(resources.items) ? resources.items.filter(isRecord) : []; + const prometheus = prometheusRuntimeStatus(items, target.observability.prometheusSelector, runtime.exists === true, context.step); return { ok: parsed.ok === true, code: parsed.code ?? "start-status-unknown", @@ -1453,7 +1505,7 @@ function summarizeRemoteStatus(parsed: Record, target: TestTarg }), database: { external: true, configRef: target.database.configRef, ...secretSourceSummary(configPath, target.database.connection) }, exposure: { hostIP: target.exposure.hostIP, port: target.exposure.port, serviceName: target.exposure.serviceName }, - observability: { otlpEndpoint: target.observability.otlpEndpoint, prometheusScrape: target.observability.prometheusScrape, blocking: false }, + observability: { ...observabilitySummary(target), prometheus, blocking: false }, }, valuesPrinted: false, }; @@ -1566,17 +1618,20 @@ function namespaceAnnotations(target: TestTargetSpec, context: RenderContext): R }; } -function objectSummary(object: Record): Record { +function objectSummary(object: Record, prometheusSelectorKey: string): Record { const metadata = optionalRecord(object.metadata, "metadata") ?? {}; const labels = optionalRecord(metadata.labels, "metadata.labels") ?? {}; const spec = optionalRecord(object.spec, "spec") ?? {}; const template = optionalRecord(spec.template, "spec.template") ?? {}; + const templateMetadata = optionalRecord(template.metadata, "spec.template.metadata") ?? {}; + const podLabels = optionalRecord(templateMetadata.labels, "spec.template.metadata.labels") ?? {}; const podSpec = optionalRecord(template.spec, "spec.template.spec") ?? {}; const securityContext = optionalRecord(podSpec.securityContext, "spec.template.spec.securityContext") ?? {}; return { kind: object.kind ?? null, name: metadata.name ?? null, owned: labels["app.kubernetes.io/managed-by"] === MANAGED_BY && labels[TEST_RUNTIME_LABEL] === "true" && typeof labels[TARGET_LABEL] === "string" && typeof labels[INSTANCE_LABEL] === "string", + prometheusSelectorLabel: object.kind === "Deployment" && typeof podLabels[prometheusSelectorKey] === "string" ? podLabels[prometheusSelectorKey] : undefined, podFsGroup: securityContext.fsGroup, secretValuesRedacted: object.kind === "Secret" ? true : undefined, }; @@ -1705,6 +1760,43 @@ function imageReference(value: unknown, path: string): string { return image; } +function otlpGrpcEndpoint(value: unknown, path: string): string { + const endpoint = nonEmpty(value, path); + if (endpoint.includes("://") || !/^[A-Za-z0-9.-]+:[1-9][0-9]{0,4}$/u.test(endpoint)) throw new Error(`${path} 必须是无 scheme 的 OTLP gRPC host:port`); + const port = Number(endpoint.slice(endpoint.lastIndexOf(":") + 1)); + if (port > 65_535) throw new Error(`${path} 端口必须是 1-65535`); + return endpoint; +} + +function prometheusSelector(value: unknown, path: string): PrometheusSelectorSpec { + const sourceRef = nonEmpty(value, `${path}.configRef`); + let resolved: ReturnType; + try { + resolved = resolveConfigRef(sourceRef, `${path}.configRef`); + } catch (error) { + throw new Error(`${path}.configRef 解析失败：${error instanceof Error ? error.message : String(error)}`); + } + const selector = record(resolved.value, `${path}.configRef (${sourceRef})`); + const key = kubernetesLabelKey(nonEmpty(selector.key, `${path}.configRef.key`), `${path}.configRef.key`); + const selectorValue = kubernetesLabelValue(nonEmpty(selector.value, `${path}.configRef.value`), `${path}.configRef.value`); + return { sourceRef, key, value: selectorValue, presence: true, fingerprint: `sha256:${resolved.sha256}`, valuesPrinted: false }; +} + +function kubernetesLabelKey(value: string, path: string): string { + const parts = value.split("/"); + const name = parts.pop() ?? ""; + const prefix = parts.length === 1 ? parts[0] : null; + const labelNamePattern = /^[A-Za-z0-9](?:[A-Za-z0-9_.-]{0,61}[A-Za-z0-9])?$/u; + const dnsPrefixPattern = /^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)*$/u; + if (parts.length > 1 || !labelNamePattern.test(name) || (prefix !== null && !dnsPrefixPattern.test(prefix))) throw new Error(`${path} 必须是合法 Kubernetes label key`); + return value; +} + +function kubernetesLabelValue(value: string, path: string): string { + if (!/^[A-Za-z0-9](?:[A-Za-z0-9_.-]{0,61}[A-Za-z0-9])?$/u.test(value)) throw new Error(`${path} 必须是合法 Kubernetes label value`); + return value; +} + function postgresIdentifier(value: unknown, path: string): string { const name = nonEmpty(value, path); if (!/^[a-z_][a-z0-9_]{0,62}$/u.test(name)) throw new Error(`${path} 必须是简单 PostgreSQL identifier`);