shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(capacity): accept dedicated fields projector

Browse Source
This commit is contained in:
lingniu
2026-09-01 11:12:08 +08:00
parent 6fceb0f5b4
commit 4201878c35
2 changed files with 23 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/vehicle-gateway/internal/capacity/check.go
+4 -3
View File
@@ -2786,7 +2786,7 @@ func findingsForTopicFlow(parsedByService map[string]parsedMetricSet, thresholds
var findings []string var findings []string
findings = append(findings, findingsForGatewayFieldFlow(parsedByService["gateway"], thresholds)...) findings = append(findings, findingsForGatewayFieldFlow(parsedByService["gateway"], thresholds)...)
findings = append(findings, findingsForGatewayFieldCounts(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, findingsForGatewayToFastWriterFlow(parsedByService["gateway"], parsedByService["fast-writer"], thresholds)...)
findings = append(findings, findingsForBridgeRawFanout(parsedByService["bridge"], parsedByService["history"], parsedByService["realtime"], thresholds)...) findings = append(findings, findingsForBridgeRawFanout(parsedByService["bridge"], parsedByService["history"], parsedByService["realtime"], thresholds)...)
findings = append(findings, findingsForBridgeToStatFieldFlow(parsedByService["bridge"], parsedByService["stat"], thresholds)...) findings = append(findings, findingsForBridgeToStatFieldFlow(parsedByService["bridge"], parsedByService["stat"], thresholds)...)
@@ -3045,12 +3045,13 @@ type gatewayPublishGroup struct {
count float64 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 { if thresholds.GatewayBridgeMinPublish <= 0 {
return nil return nil
} }
gatewayPublishes := gatewayPublishGroups(gateway.samples) gatewayPublishes := gatewayPublishGroups(gateway.samples)
bridgeWrites := metricValuesByTopic(bridge.samples, "vehicle_bridge_kafka_writes_total", "ok", "vehicle.") 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 var findings []string
for _, group := range gatewayPublishes { for _, group := range gatewayPublishes {
if group.count < thresholds.GatewayBridgeMinPublish { if group.count < thresholds.GatewayBridgeMinPublish {
@@ -3066,7 +3067,7 @@ func findingsForGatewayToBridgeFlow(gateway parsedMetricSet, bridge parsedMetric
)) ))
continue continue
} }
if bridgeWrites[topic] <= 0 { if bridgeWrites[topic] <= 0 && (group.kind != "fields" || projectorWrites[topic] <= 0) {
findings = append(findings, fmt.Sprintf( findings = append(findings, fmt.Sprintf(
"bridge writes missing for gateway %s publish protocol %s: gateway ok %.0f, expected topic %s, bridge writes ok %.0f", "bridge writes missing for gateway %s publish protocol %s: gateway ok %.0f, expected topic %s, bridge writes ok %.0f",
group.kind, group.kind,
go/vehicle-gateway/internal/capacity/check_test.go
+19
View File
@@ -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) { func TestEvaluateSkipsGatewayBridgeFlowWhenSamplesAreLow(t *testing.T) {
report := EvaluateWithThresholds(map[string]string{ report := EvaluateWithThresholds(map[string]string{
"gateway": `vehicle_service_info{service="vehicle-gateway"} 1 "gateway": `vehicle_service_info{service="vehicle-gateway"} 1