diff --git a/config/platform-infra/kafka.yaml b/config/platform-infra/kafka.yaml index c76e8a31..d07aca09 100644 --- a/config/platform-infra/kafka.yaml +++ b/config/platform-infra/kafka.yaml @@ -8,15 +8,16 @@ metadata: relatedIssues: - 2243 - 2449 + - 2462 defaults: targetId: NC01 switch: enabled: true - mode: shadow-produce-only + mode: durable-event-stream appIntegrationEnabled: true - shadowProduceEnabled: true - shadowConsumeEnabled: false + produceEnabled: true + consumeEnabled: true operator: implementation: strimzi @@ -178,5 +179,5 @@ management: pattern: ^hwlab-v03-cloud-api-sse-([0-9]{13})-[0-9a-f]{8}$ captureGroup: 1 unit: unix-ms - defaultShadowTopic: hwlab.event.v1 - shadowProducerId: unidesk-platform-infra-kafka-cli + defaultTopic: hwlab.event.v1 + diagnosticProducerId: unidesk-platform-infra-kafka-cli diff --git a/scripts/src/platform-infra-kafka.ts b/scripts/src/platform-infra-kafka.ts index 7369dcf2..bbdc5711 100644 --- a/scripts/src/platform-infra-kafka.ts +++ b/scripts/src/platform-infra-kafka.ts @@ -74,10 +74,10 @@ interface PlatformKafkaConfig { targetId: string; switch: { enabled: boolean; - mode: string; + mode: "durable-event-stream"; appIntegrationEnabled: boolean; - shadowProduceEnabled: boolean; - shadowConsumeEnabled: boolean; + produceEnabled: boolean; + consumeEnabled: boolean; }; }; operator: { @@ -127,8 +127,8 @@ interface PlatformKafkaConfig { deleteBatchSize: number; policies: KafkaGroupCleanupPolicy[]; }; - defaultShadowTopic: string; - shadowProducerId: string; + defaultTopic: string; + diagnosticProducerId: string; }; } @@ -230,7 +230,7 @@ export function kafkaHelp(): Record { "bun scripts/cli.ts platform-infra kafka tail --node NC01 --topic hwlab.event.v1 --limit 5", "bun scripts/cli.ts platform-infra kafka produce --node NC01 --topic hwlab.event.v1 --key ", ], - boundary: "Kafka runtime belongs to platform-infra. P2 supports shadow produce/query only; Kafka consumer cutover stays disabled.", + boundary: "Kafka runtime belongs to platform-infra. Durable producers and fixed-group consumers are enabled; browser delivery remains a database-projected SSE concern.", }; } @@ -262,10 +262,10 @@ function readKafkaConfig(): PlatformKafkaConfig { targetId: y.stringField(defaults, "targetId", "defaults"), switch: { enabled: y.booleanField(switchRecord, "enabled", "defaults.switch"), - mode: y.stringField(switchRecord, "mode", "defaults.switch"), + mode: y.enumField(switchRecord, "mode", "defaults.switch", ["durable-event-stream"] as const), appIntegrationEnabled: y.booleanField(switchRecord, "appIntegrationEnabled", "defaults.switch"), - shadowProduceEnabled: y.booleanField(switchRecord, "shadowProduceEnabled", "defaults.switch"), - shadowConsumeEnabled: y.booleanField(switchRecord, "shadowConsumeEnabled", "defaults.switch"), + produceEnabled: y.booleanField(switchRecord, "produceEnabled", "defaults.switch"), + consumeEnabled: y.booleanField(switchRecord, "consumeEnabled", "defaults.switch"), }, }, operator: { @@ -363,8 +363,8 @@ function parseManagement(root: Record): PlatformKafkaConfig["ma deleteBatchSize: positiveInteger(cleanup, "deleteBatchSize", "management.consumerGroupCleanup"), policies: y.arrayOfRecords(cleanup.policies, "management.consumerGroupCleanup.policies").map(parseGroupCleanupPolicy), }, - defaultShadowTopic: topicNameField(management, "defaultShadowTopic", "management"), - shadowProducerId: y.stringField(management, "shadowProducerId", "management"), + defaultTopic: topicNameField(management, "defaultTopic", "management"), + diagnosticProducerId: y.stringField(management, "diagnosticProducerId", "management"), }; } @@ -406,7 +406,7 @@ function validateConfig(kafka: PlatformKafkaConfig): void { if (kafka.cluster.replicas !== 1) throw new Error(`${configLabel}.cluster.replicas must be 1 for the target-scoped Kafka POC`); if (!/^[0-9]+Gi$/u.test(kafka.cluster.storage.size)) throw new Error(`${configLabel}.cluster.storage.size must be a Gi quantity`); if (!kafka.topics.some((topic) => topic.name === kafka.validation.smokeTopic)) throw new Error(`${configLabel}.validation.smokeTopic must reference one of topics[].name`); - if (!kafka.topics.some((topic) => topic.name === kafka.management.defaultShadowTopic)) throw new Error(`${configLabel}.management.defaultShadowTopic must reference one of topics[].name`); + if (!kafka.topics.some((topic) => topic.name === kafka.management.defaultTopic)) throw new Error(`${configLabel}.management.defaultTopic must reference one of topics[].name`); if (kafka.management.defaultTailLimit > kafka.management.maxTailLimit) throw new Error(`${configLabel}.management.defaultTailLimit must be <= management.maxTailLimit`); if (kafka.management.defaultGroupListLimit > kafka.management.maxGroupListLimit) throw new Error(`${configLabel}.management.defaultGroupListLimit must be <= management.maxGroupListLimit`); if (kafka.management.consumerGroupCleanup.defaultCandidateLimit > kafka.management.consumerGroupCleanup.maxCandidateLimit) throw new Error(`${configLabel}.management.consumerGroupCleanup.defaultCandidateLimit must be <= maxCandidateLimit`); @@ -418,7 +418,9 @@ function validateConfig(kafka: PlatformKafkaConfig): void { cleanupPolicyIds.add(policy.id); cleanupPrefixes.add(policy.prefix); } - if (kafka.defaults.switch.shadowConsumeEnabled) throw new Error(`${configLabel}.defaults.switch.shadowConsumeEnabled must stay false for shadow-produce-only stage`); + if (!kafka.defaults.switch.enabled || !kafka.defaults.switch.appIntegrationEnabled || !kafka.defaults.switch.produceEnabled || !kafka.defaults.switch.consumeEnabled) { + throw new Error(`${configLabel}.defaults.switch must enable durable event production and consumption`); + } const topicNames = new Set(); for (const topic of kafka.topics) { if (topicNames.has(topic.name)) throw new Error(`${configLabel}.topics contains duplicate topic ${topic.name}`); @@ -644,7 +646,7 @@ async function offsets(config: UniDeskConfig, options: KafkaInspectOptions): Pro async function tail(config: UniDeskConfig, options: KafkaInspectOptions): Promise> { const kafka = readKafkaConfig(); const target = resolveTarget(kafka, options.targetId); - const topic = resolveTopic(kafka, options.topic, kafka.management.defaultShadowTopic); + const topic = resolveTopic(kafka, options.topic, kafka.management.defaultTopic); const limit = boundedLimit(kafka, options.limit); const result = await capture(config, target.route, ["sh"], tailScript(kafka, target, topic, limit)); const parsed = parseJsonOutput(result.stdout); @@ -662,15 +664,15 @@ async function tail(config: UniDeskConfig, options: KafkaInspectOptions): Promis async function produce(config: UniDeskConfig, options: KafkaProduceOptions): Promise> { const kafka = readKafkaConfig(); const target = resolveTarget(kafka, options.targetId); - const topic = resolveTopic(kafka, options.topic, kafka.management.defaultShadowTopic); - const payload = buildShadowPayload(kafka, target, options, topic); + const topic = resolveTopic(kafka, options.topic, kafka.management.defaultTopic); + const payload = buildDiagnosticPayload(kafka, target, options, topic); const result = await capture(config, target.route, ["sh"], produceScript(kafka, target, topic, options.key, payload)); const parsed = parseJsonOutput(result.stdout); return { ok: result.exitCode === 0 && parsed?.ok === true, action: "platform-infra-kafka-produce", mutation: true, - mode: "shadow-produce-only", + mode: "diagnostic-produce", target: targetSummary(target), config: compactConfigSummary(kafka, target), production: parsed ?? null, @@ -763,6 +765,9 @@ metadata: annotations: unidesk.ai/spec: ${kafka.metadata.spec} unidesk.ai/app-integration-enabled: "${kafka.defaults.switch.appIntegrationEnabled}" + unidesk.ai/event-stream-mode: "${kafka.defaults.switch.mode}" + unidesk.ai/event-produce-enabled: "${kafka.defaults.switch.produceEnabled}" + unidesk.ai/event-consume-enabled: "${kafka.defaults.switch.consumeEnabled}" spec: kafka: version: ${kafka.cluster.kafkaVersion} @@ -1453,7 +1458,7 @@ PY } function produceScript(kafka: PlatformKafkaConfig, target: KafkaTarget, topic: string, key: string | null, payload: string): string { - const keyValue = key ?? `${kafka.management.shadowProducerId}-${Date.now()}`; + const keyValue = key ?? `${kafka.management.diagnosticProducerId}-${Date.now()}`; const separator = "|"; if (keyValue.includes(separator) || keyValue.includes("\n")) throw new Error("kafka produce --key must not contain | or newline"); return ` @@ -1496,9 +1501,9 @@ payload_obj = { "bytes": len(payload.encode()), "valuesPrinted": False, }, - "producer": "${kafka.management.shadowProducerId}", - "mode": "shadow-produce-only", - "consume": {"enabled": False, "performed": False}, + "producer": "${kafka.management.diagnosticProducerId}", + "mode": "diagnostic-produce", + "consume": {"enabled": True, "performed": False}, "steps": {"produce": {"exitCode": produce_rc, "stdoutTail": text("produce.out"), "stderrTail": text("produce.err")}}, "valuesPrinted": False, } @@ -1543,13 +1548,13 @@ function compactConfigSummary(kafka: PlatformKafkaConfig, target: KafkaTarget): topicCount: kafka.topics.length, clientCount: kafka.clients.length, management: { - defaultShadowTopic: kafka.management.defaultShadowTopic, + defaultTopic: kafka.management.defaultTopic, defaultTailLimit: kafka.management.defaultTailLimit, maxTailLimit: kafka.management.maxTailLimit, defaultGroupListLimit: kafka.management.defaultGroupListLimit, maxGroupListLimit: kafka.management.maxGroupListLimit, consumerGroupCleanup: kafka.management.consumerGroupCleanup, - shadowProducerId: kafka.management.shadowProducerId, + diagnosticProducerId: kafka.management.diagnosticProducerId, }, valuesPrinted: false, }; @@ -1622,9 +1627,9 @@ function policyChecks(kafka: PlatformKafkaConfig, target: KafkaTarget, manifest: { name: "no-public-exposure", ok: !/^\s*type:\s*(NodePort|LoadBalancer)\s*$/mu.test(manifest) && !/^\s*kind:\s*Ingress\s*$/mu.test(manifest), detail: "Kafka POC is ClusterIP/internal only." }, { name: "single-broker-poc", ok: kafka.cluster.replicas === 1, detail: "The target-scoped POC uses one KRaft broker/controller; production HA remains out of scope." }, { name: "allow-all-network-policy", ok: manifest.includes("kind: NetworkPolicy") && manifest.includes("name: allow-all") && manifest.includes(`namespace: ${target.namespace}`), detail: `NetworkPolicy/allow-all is rendered in ${target.namespace}.` }, - { name: "shadow-produce-enabled", ok: kafka.defaults.switch.shadowProduceEnabled === true, detail: "Apps may publish shadow events to Kafka." }, - { name: "app-integration-shadow-only", ok: kafka.defaults.switch.appIntegrationEnabled === true && kafka.defaults.switch.mode === "shadow-produce-only", detail: "App integration is limited to shadow production in this stage." }, - { name: "shadow-consume-disabled", ok: kafka.defaults.switch.shadowConsumeEnabled === false, detail: "Kafka consumer cutover remains disabled." }, + { name: "durable-produce-enabled", ok: kafka.defaults.switch.produceEnabled === true, detail: "Transactional outbox relays may publish durable application events." }, + { name: "durable-event-stream-mode", ok: kafka.defaults.switch.appIntegrationEnabled === true && kafka.defaults.switch.mode === "durable-event-stream", detail: "Application integration uses the durable event stream contract." }, + { name: "fixed-group-consume-enabled", ok: kafka.defaults.switch.consumeEnabled === true, detail: "Durable projector consumers are enabled." }, ]; } @@ -1686,7 +1691,7 @@ function renderPlan(result: Record): RenderedCliResult { ` status: ${stringValue(next.status)}`, ` validate: ${stringValue(next.validate)}`, "", - "Boundary: Kafka runtime is platform-infra only; P2 allows shadow produce/query and keeps consumer cutover disabled.", + "Boundary: Kafka runtime is platform-infra only; durable event production and fixed-group projector consumption are enabled.", "Disclosure: Secret values are not printed; client entries show only object/key metadata.", ]); } @@ -1960,7 +1965,7 @@ async function parseProduceOptions(args: string[]): Promise let topic: string | null = null; let key: string | null = null; let source = "manual-cli"; - let eventType = "platform-infra.kafka.shadow-produce.v1"; + let eventType = "platform-infra.kafka.diagnostic-produce.v1"; let payload: string | null = null; for (let index = 0; index < args.length; index += 1) { const arg = args[index]; @@ -2026,7 +2031,7 @@ function boundedGroupCleanupLimit(kafka: PlatformKafkaConfig, limit: number | nu return value; } -function buildShadowPayload(kafka: PlatformKafkaConfig, target: KafkaTarget, options: KafkaProduceOptions, topic: string): string { +function buildDiagnosticPayload(kafka: PlatformKafkaConfig, target: KafkaTarget, options: KafkaProduceOptions, topic: string): string { let userPayload: unknown = null; if (options.payload !== null) { const trimmed = options.payload.trim(); @@ -2042,20 +2047,19 @@ function buildShadowPayload(kafka: PlatformKafkaConfig, target: KafkaTarget, opt eventId: `evt_${target.id}_${Date.now()}_${Math.random().toString(16).slice(2)}`, schemaVersion: 1, eventType: options.eventType, - producer: kafka.management.shadowProducerId, + producer: kafka.management.diagnosticProducerId, source: options.source, node: target.id, namespace: target.namespace, topic, - mode: "shadow-produce-only", - consumeEnabled: false, + mode: kafka.defaults.switch.mode, appIntegrationEnabled: kafka.defaults.switch.appIntegrationEnabled, - shadowProduceEnabled: kafka.defaults.switch.shadowProduceEnabled, - shadowConsumeEnabled: kafka.defaults.switch.shadowConsumeEnabled, + produceEnabled: kafka.defaults.switch.produceEnabled, + consumeEnabled: kafka.defaults.switch.consumeEnabled, partitionKey: options.key ?? null, occurredAt: new Date().toISOString(), payload: userPayload ?? { - kind: "manual-shadow-produce", + kind: "manual-diagnostic-produce", spec: kafka.metadata.spec, valuesPrinted: false, },