diff --git a/go/vehicle-gateway/internal/capacity/check.go b/go/vehicle-gateway/internal/capacity/check.go index cc001fde..c5e540e2 100644 --- a/go/vehicle-gateway/internal/capacity/check.go +++ b/go/vehicle-gateway/internal/capacity/check.go @@ -2786,7 +2786,7 @@ func findingsForTopicFlow(parsedByService map[string]parsedMetricSet, thresholds var findings []string findings = append(findings, findingsForGatewayFieldFlow(parsedByService["gateway"], thresholds)...) findings = append(findings, findingsForGatewayFieldCounts(parsedByService["gateway"], thresholds)...) - findings = append(findings, findingsForGatewayToBridgeFlow(parsedByService["gateway"], parsedByService["bridge"], thresholds)...) + findings = append(findings, findingsForGatewayToBridgeFlow(parsedByService["gateway"], parsedByService["bridge"], parsedByService["fields-projector"], thresholds)...) findings = append(findings, findingsForGatewayToFastWriterFlow(parsedByService["gateway"], parsedByService["fast-writer"], thresholds)...) findings = append(findings, findingsForBridgeRawFanout(parsedByService["bridge"], parsedByService["history"], parsedByService["realtime"], thresholds)...) findings = append(findings, findingsForBridgeToStatFieldFlow(parsedByService["bridge"], parsedByService["stat"], thresholds)...) @@ -3045,12 +3045,13 @@ type gatewayPublishGroup struct { count float64 } -func findingsForGatewayToBridgeFlow(gateway parsedMetricSet, bridge parsedMetricSet, thresholds Thresholds) []string { +func findingsForGatewayToBridgeFlow(gateway parsedMetricSet, bridge parsedMetricSet, fieldsProjector parsedMetricSet, thresholds Thresholds) []string { if thresholds.GatewayBridgeMinPublish <= 0 { return nil } gatewayPublishes := gatewayPublishGroups(gateway.samples) bridgeWrites := metricValuesByTopic(bridge.samples, "vehicle_bridge_kafka_writes_total", "ok", "vehicle.") + projectorWrites := metricValuesByTopic(fieldsProjector.samples, "vehicle_fields_projector_kafka_writes_total", "ok", "vehicle.fields.go.") var findings []string for _, group := range gatewayPublishes { if group.count < thresholds.GatewayBridgeMinPublish { @@ -3066,7 +3067,7 @@ func findingsForGatewayToBridgeFlow(gateway parsedMetricSet, bridge parsedMetric )) continue } - if bridgeWrites[topic] <= 0 { + if bridgeWrites[topic] <= 0 && (group.kind != "fields" || projectorWrites[topic] <= 0) { findings = append(findings, fmt.Sprintf( "bridge writes missing for gateway %s publish protocol %s: gateway ok %.0f, expected topic %s, bridge writes ok %.0f", group.kind, diff --git a/go/vehicle-gateway/internal/capacity/check_test.go b/go/vehicle-gateway/internal/capacity/check_test.go index 8fb136b1..afd659a6 100644 --- a/go/vehicle-gateway/internal/capacity/check_test.go +++ b/go/vehicle-gateway/internal/capacity/check_test.go @@ -2158,6 +2158,25 @@ vehicle_bridge_kafka_writes_total{status="ok",topic="vehicle.fields.go.yutong-mq } } +func TestEvaluateAcceptsFieldsProjectorForGatewayFieldsPublish(t *testing.T) { + report := EvaluateWithThresholds(map[string]string{ + "gateway": `vehicle_service_info{service="vehicle-gateway"} 1 +vehicle_gateway_publish_total{kind="fields",protocol="JT808",status="ok"} 120`, + "bridge": `vehicle_service_info{service="nats-kafka-bridge"} 1 +vehicle_bridge_config{setting="derive_fields_from_raw_enabled"} 0`, + "fields-projector": `vehicle_service_info{service="fields-projector"} 1 +vehicle_kafka_consumer_info{service="fields-projector",group="vehicle-fields-projector-v1-jt808",topic="vehicle.raw.go.jt808.v1"} 1 +vehicle_fields_projector_kafka_writes_total{status="ok",topic="vehicle.fields.go.jt808.v1"} 120`, + }, Thresholds{GatewayBridgeMinPublish: 100}) + + if report.Status != StatusOK { + t.Fatalf("status = %s, want ok: %#v", report.Status, report) + } + if len(report.Findings) != 0 { + t.Fatalf("findings = %#v, want none", report.Findings) + } +} + func TestEvaluateSkipsGatewayBridgeFlowWhenSamplesAreLow(t *testing.T) { report := EvaluateWithThresholds(map[string]string{ "gateway": `vehicle_service_info{service="vehicle-gateway"} 1