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

feat(go): expose gateway async sink metrics

Browse Source
This commit is contained in:
lingniu
2026-07-03 18:28:07 +08:00
parent f8fbeb8936
commit fb03c80a95
6 changed files with 169 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
go/vehicle-gateway/cmd/gateway/main.go
View File

@@ -30,7 +30,8 @@ func main() {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
sink, err := buildSink(ctx, logger)
registry := metrics.NewRegistry()
sink, err := buildSink(ctx, logger, registry)
if err != nil {
logger.Error("build sink failed", "error", err)
os.Exit(1)
@@ -42,7 +43,6 @@ func main() {
os.Exit(1)
}
defer closeResolver()
registry := metrics.NewRegistry()
health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-gateway", nil, registry))
publishUnified := envBool("PUBLISH_UNIFIED_ENABLED", false)
@@ -161,7 +161,7 @@ func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.R
return resolver, func() { _ = db.Close() }, nil
}
func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) {
func buildSink(ctx context.Context, logger *slog.Logger, registry *metrics.Registry) (eventbus.Sink, error) {
if strings.TrimSpace(os.Getenv("NATS_URL")) != "" {
sink, err := eventbus.NewNATSSink(natsSinkConfigFromEnv())
if err != nil {
@@ -180,6 +180,8 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error)
QueueSize: queueSize,
Workers: workers,
OperationTimeout: timeout,
Metrics: registry,
Name: "nats",
OnError: func(err error) {
logger.Warn("nats async publish failed", "error", err)
},
@@ -235,6 +237,8 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error)
QueueSize: queueSize,
Workers: workers,
OperationTimeout: timeout,
Metrics: registry,
Name: "kafka",
OnError: func(err error) {
logger.Warn("kafka async publish failed", "error", err)
},

17
go/vehicle-gateway/cmd/gateway/main_test.go
View File

@@ -56,6 +56,23 @@ func TestGatewayDefaultsTo100KConnectionCeiling(t *testing.T) {
}
}
func TestGatewayPassesMetricsRegistryToAsyncSink(t *testing.T) {
source, err := os.ReadFile("main.go")
if err != nil {
t.Fatalf("read main.go: %v", err)
}
for _, want := range []string{
"buildSink(ctx, logger, registry)",
"Metrics: registry",
`Name: "nats"`,
`Name: "kafka"`,
} {
if !strings.Contains(string(source), want) {
t.Fatalf("gateway async sink metrics wiring missing %s", want)
}
}
}
func TestNATSSinkConfigFromEnvDefaultsToGoSubjects(t *testing.T) {
t.Setenv("NATS_URL", "nats://172.17.111.56:4222")

57
go/vehicle-gateway/internal/eventbus/async_sink.go
View File

@@ -7,6 +7,7 @@ import (
"time"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
)
type AsyncConfig struct {
@@ -14,6 +15,8 @@ type AsyncConfig struct {
Workers int
OperationTimeout time.Duration
OnError func(error)
Metrics *metrics.Registry
Name string
}
type AsyncSink struct {
@@ -21,6 +24,8 @@ type AsyncSink struct {
jobs chan asyncJob
timeout time.Duration
onError func(error)
metrics *metrics.Registry
name string
closeOnce sync.Once
closed chan struct{}
@@ -48,11 +53,16 @@ func NewAsyncSink(delegate Sink, cfg AsyncConfig) *AsyncSink {
if cfg.OperationTimeout <= 0 {
cfg.OperationTimeout = 30 * time.Second
}
if cfg.Name == "" {
cfg.Name = "async"
}
s := &AsyncSink{
delegate: delegate,
jobs: make(chan asyncJob, cfg.QueueSize),
timeout: cfg.OperationTimeout,
onError: cfg.OnError,
metrics: cfg.Metrics,
name: cfg.Name,
closed: make(chan struct{}),
done: make(chan struct{}),
}
@@ -91,15 +101,21 @@ func (s *AsyncSink) Close() error {
func (s *AsyncSink) enqueue(ctx context.Context, job asyncJob) error {
select {
case <-s.closed:
s.recordEnqueue(job.kind, "closed")
return ErrAsyncSinkClosed
default:
}
select {
case s.jobs <- job:
s.recordEnqueue(job.kind, "queued")
s.recordQueueDepth()
return nil
case <-s.closed:
s.recordEnqueue(job.kind, "closed")
return ErrAsyncSinkClosed
case <-ctx.Done():
s.recordEnqueue(job.kind, "timeout")
s.recordQueueDepth()
return ctx.Err()
}
}
@@ -107,7 +123,9 @@ func (s *AsyncSink) enqueue(ctx context.Context, job asyncJob) error {
func (s *AsyncSink) worker() {
defer s.wg.Done()
for job := range s.jobs {
s.recordQueueDepth()
ctx, cancel := context.WithTimeout(context.Background(), s.timeout)
started := time.Now()
var err error
switch job.kind {
case "raw":
@@ -118,8 +136,47 @@ func (s *AsyncSink) worker() {
err = s.delegate.PublishFields(ctx, job.env)
}
cancel()
status := "ok"
if err != nil {
status = "error"
}
s.recordPublish(job.kind, status, time.Since(started))
if err != nil && s.onError != nil {
s.onError(err)
}
s.recordQueueDepth()
}
}
func (s *AsyncSink) recordEnqueue(kind string, status string) {
if s.metrics == nil {
return
}
s.metrics.IncCounter("vehicle_async_sink_enqueue_total", metrics.Labels{
"sink": s.name,
"kind": kind,
"status": status,
})
}
func (s *AsyncSink) recordPublish(kind string, status string, elapsed time.Duration) {
if s.metrics == nil {
return
}
labels := metrics.Labels{
"sink": s.name,
"kind": kind,
"status": status,
}
s.metrics.IncCounter("vehicle_async_sink_publish_total", labels)
s.metrics.SetGauge("vehicle_async_sink_publish_duration_ms", labels, float64(elapsed.Milliseconds()))
}
func (s *AsyncSink) recordQueueDepth() {
if s.metrics == nil {
return
}
s.metrics.SetGauge("vehicle_async_sink_queue_depth", metrics.Labels{
"sink": s.name,
}, float64(len(s.jobs)))
}

81
go/vehicle-gateway/internal/eventbus/async_sink_test.go
View File

@@ -2,10 +2,12 @@ package eventbus
import (
"context"
"strings"
"testing"
"time"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
)
func TestAsyncSinkPublishRawReturnsAfterQueueing(t *testing.T) {
@@ -51,6 +53,85 @@ func TestAsyncSinkPublishesInFIFOOrderWithSingleWorker(t *testing.T) {
}
}
func TestAsyncSinkRecordsQueueMetrics(t *testing.T) {
registry := metrics.NewRegistry()
delegate := newBlockingSink()
sink := NewAsyncSink(delegate, AsyncConfig{
QueueSize: 2,
Workers: 1,
OperationTimeout: time.Second,
Metrics: registry,
Name: "nats",
})
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001"}
if err := sink.PublishRaw(context.Background(), env); err != nil {
t.Fatalf("PublishRaw() error = %v", err)
}
if err := sink.PublishFields(context.Background(), env); err != nil {
t.Fatalf("PublishFields() error = %v", err)
}
delegate.release()
if err := sink.Close(); err != nil {
t.Fatalf("Close() error = %v", err)
}
text := registry.Render()
for _, want := range []string{
`vehicle_async_sink_enqueue_total{kind="raw",sink="nats",status="queued"} 1`,
`vehicle_async_sink_enqueue_total{kind="fields",sink="nats",status="queued"} 1`,
`vehicle_async_sink_queue_depth{sink="nats"}`,
`vehicle_async_sink_publish_total{kind="raw",sink="nats",status="ok"} 1`,
`vehicle_async_sink_publish_total{kind="fields",sink="nats",status="ok"} 1`,
} {
if !strings.Contains(text, want) {
t.Fatalf("async sink metric missing %s:\n%s", want, text)
}
}
}
func TestAsyncSinkRecordsEnqueueTimeoutWhenQueueIsFull(t *testing.T) {
registry := metrics.NewRegistry()
delegate := newBlockingSink()
sink := NewAsyncSink(delegate, AsyncConfig{
QueueSize: 1,
Workers: 1,
OperationTimeout: time.Second,
Metrics: registry,
Name: "nats",
})
defer sink.Close()
env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425"}
if err := sink.PublishRaw(context.Background(), env); err != nil {
t.Fatalf("first PublishRaw() error = %v", err)
}
select {
case <-delegate.rawStarted:
case <-time.After(time.Second):
t.Fatal("delegate raw publish was not started")
}
if err := sink.PublishFields(context.Background(), env); err != nil {
t.Fatalf("second PublishFields() error = %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
defer cancel()
if err := sink.PublishUnified(ctx, env); err == nil {
t.Fatal("third PublishUnified() error = nil, want context deadline exceeded")
}
text := registry.Render()
for _, want := range []string{
`vehicle_async_sink_enqueue_total{kind="unified",sink="nats",status="timeout"} 1`,
`vehicle_async_sink_queue_depth{sink="nats"} 1`,
} {
if !strings.Contains(text, want) {
t.Fatalf("async sink timeout metric missing %s:\n%s", want, text)
}
}
delegate.release()
}
type blockingSink struct {
rawStarted chan struct{}
releaseRaw chan struct{}