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

feat(go): expose nats bridge pending metrics

Browse Source
This commit is contained in:
lingniu
2026-07-02 19:27:36 +08:00
parent 1f4a23ff69
commit df96ec8879
3 changed files with 54 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
go/vehicle-gateway/cmd/nats-kafka-bridge/main.go
View File

@@ -80,7 +80,7 @@ func main() {
"durable", cfg.NATSDurable,
"filter", cfg.NATSFilter,
"kafka_brokers", strings.Join(cfg.KafkaBrokers, ","))
runBridge(ctx, logger, registry, sub, writer, cfg)
runBridge(ctx, logger, registry, js, sub, writer, cfg)
}
type config struct {
@@ -127,6 +127,10 @@ type natsPullSubscription interface {
Fetch(int, ...nats.PullOpt) ([]*nats.Msg, error)
}
type natsConsumerInfoReader interface {
ConsumerInfo(stream string, name string, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
}
type kafkaBatchWriter interface {
WriteMessages(context.Context, ...kafka.Message) error
}
@@ -137,11 +141,23 @@ type bridgeMessage struct {
ack func() error
}
func runBridge(ctx context.Context, logger *slog.Logger, registry *metrics.Registry, sub natsPullSubscription, writer kafkaBatchWriter, cfg config) {
func runBridge(ctx context.Context, logger *slog.Logger, registry *metrics.Registry, infoReader natsConsumerInfoReader, sub natsPullSubscription, writer kafkaBatchWriter, cfg config) {
var lastConsumerInfoAt time.Time
for {
if ctx.Err() != nil {
return
}
if time.Since(lastConsumerInfoAt) >= time.Duration(envInt("NATS_CONSUMER_METRICS_INTERVAL_SECONDS", 10))*time.Second {
lastConsumerInfoAt = time.Now()
if infoReader != nil {
info, err := infoReader.ConsumerInfo(cfg.NATSStream, cfg.NATSDurable, nats.Context(ctx))
if err != nil {
logger.Warn("nats consumer info failed", "stream", cfg.NATSStream, "durable", cfg.NATSDurable, "error", err)
} else {
recordNATSConsumerInfoMetrics(registry, cfg, info)
}
}
}
msgs, err := sub.Fetch(cfg.BatchSize, nats.MaxWait(cfg.FetchWait))
if err != nil {
if errors.Is(err, nats.ErrTimeout) {
@@ -246,6 +262,16 @@ func addBridgeTopicMetric(registry *metrics.Registry, name string, topic string,
registry.IncCounter(name, metrics.Labels{"topic": topic, "status": status})
}
func recordNATSConsumerInfoMetrics(registry *metrics.Registry, cfg config, info *nats.ConsumerInfo) {
if registry == nil || info == nil {
return
}
labels := metrics.Labels{"stream": cfg.NATSStream, "consumer": cfg.NATSDurable}
registry.SetGauge("vehicle_bridge_nats_consumer_pending", labels, float64(info.NumPending))
registry.SetGauge("vehicle_bridge_nats_consumer_ack_pending", labels, float64(info.NumAckPending))
registry.SetGauge("vehicle_bridge_nats_consumer_waiting", labels, float64(info.NumWaiting))
}
func kafkaMessage(topic string, data []byte) kafka.Message {
var env envelope.FrameEnvelope
message := kafka.Message{Topic: topic, Value: data}