fix data race & add test

pkg/server/metrics.go

@@ -77,7 +77,25 @@ func (m *peerMetricsCollector) Describe(out chan<- *prometheus.Desc) {
 }
 
 func (m *peerMetricsCollector) Collect(out chan<- prometheus.Metric) {
-	msg := m.collectMessageCounters()
+	var msg config.Messages
+	type familyCnt struct {
+		received, accepted int
+	}
+	routeCnt := make(map[bgp.RouteFamily]familyCnt)
+
+	m.server.mgmtOperation(func() error {
+		msg = m.collectMessageCounters()
+
+		for _, family := range m.peer.configuredRFlist() {
+			flist := []bgp.RouteFamily{family}
+			routeCnt[family] = familyCnt{
+				received: m.peer.adjRibIn.Count(flist),
+				accepted: m.peer.adjRibIn.Accepted(flist),
+			}
+		}
+		return nil
+	}, false)
+
 	send := func(desc *prometheus.Desc, cnt uint64) {
 		out <- prometheus.MustNewConstMetric(desc, prometheus.CounterValue, float64(cnt))
 	}
@@ -104,21 +122,17 @@ func (m *peerMetricsCollector) Collect(out chan<- prometheus.Metric) {
 	send(bgpSentDiscardedTotalDesc, msg.Sent.Discarded)
 	send(bgpSentMessageTotalDesc, msg.Sent.Total)
 
-	for _, family := range m.peer.configuredRFlist() {
-		flist := []bgp.RouteFamily{family}
-		received := m.peer.adjRibIn.Count(flist)
-		accepted := m.peer.adjRibIn.Accepted(flist)
-
+	for family, cnt := range routeCnt {
 		out <- prometheus.MustNewConstMetric(
 			bgpRoutesReceivedDesc,
 			prometheus.GaugeValue,
-			float64(received),
+			float64(cnt.received),
 			family.String(),
 		)
 		out <- prometheus.MustNewConstMetric(
 			bgpRoutesAcceptedDesc,
 			prometheus.GaugeValue,
-			float64(accepted),
+			float64(cnt.accepted),
 			family.String(),
 		)
 	}

pkg/server/metrics_test.go

@@ -0,0 +1,134 @@
+package server
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/assert"
+	apb "google.golang.org/protobuf/types/known/anypb"
+
+	api "github.com/osrg/gobgp/v3/api"
+)
+
+func TestMetrics(test *testing.T) {
+	registry := prometheus.NewRegistry()
+
+	assert := assert.New(test)
+	s := NewBgpServer(Metrics(registry))
+	go s.Serve()
+	err := s.StartBgp(context.Background(), &api.StartBgpRequest{
+		Global: &api.Global{
+			Asn:        1,
+			RouterId:   "1.1.1.1",
+			ListenPort: 10179,
+		},
+	})
+	assert.Nil(err)
+	defer s.StopBgp(context.Background(), &api.StopBgpRequest{})
+
+	p1 := &api.Peer{
+		Conf: &api.PeerConf{
+			NeighborAddress: "127.0.0.1",
+			PeerAsn:         2,
+		},
+		Transport: &api.Transport{
+			PassiveMode: true,
+		},
+	}
+	err = s.AddPeer(context.Background(), &api.AddPeerRequest{Peer: p1})
+	assert.Nil(err)
+
+	t := NewBgpServer()
+	go t.Serve()
+	err = t.StartBgp(context.Background(), &api.StartBgpRequest{
+		Global: &api.Global{
+			Asn:        2,
+			RouterId:   "2.2.2.2",
+			ListenPort: -1,
+		},
+	})
+	assert.Nil(err)
+	defer t.StopBgp(context.Background(), &api.StopBgpRequest{})
+
+	p2 := &api.Peer{
+		Conf: &api.PeerConf{
+			NeighborAddress: "127.0.0.1",
+			PeerAsn:         1,
+		},
+		Transport: &api.Transport{
+			RemotePort: 10179,
+		},
+		Timers: &api.Timers{
+			Config: &api.TimersConfig{
+				ConnectRetry:           1,
+				IdleHoldTimeAfterReset: 1,
+			},
+		},
+	}
+
+	ch := make(chan struct{})
+	waitEstablished(s, ch)
+
+	err = t.AddPeer(context.Background(), &api.AddPeerRequest{Peer: p2})
+	assert.Nil(err)
+	<-ch
+
+	family := &api.Family{
+		Afi:  api.Family_AFI_IP,
+		Safi: api.Family_SAFI_UNICAST,
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	ch = make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				ch <- struct{}{}
+				return
+			default:
+				nlri1, _ := apb.New(&api.IPAddressPrefix{
+					Prefix:    "10.1.0.0",
+					PrefixLen: 24,
+				})
+
+				a1, _ := apb.New(&api.OriginAttribute{
+					Origin: 0,
+				})
+				a2, _ := apb.New(&api.NextHopAttribute{
+					NextHop: "10.0.0.1",
+				})
+				attrs := []*apb.Any{a1, a2}
+
+				t.AddPath(context.Background(), &api.AddPathRequest{
+					TableType: api.TableType_GLOBAL,
+					Path: &api.Path{
+						Family: family,
+						Nlri:   nlri1,
+						Pattrs: attrs,
+					},
+				})
+				t.DeletePath(context.Background(), &api.DeletePathRequest{
+					TableType: api.TableType_GLOBAL,
+					Path: &api.Path{
+						Family: family,
+						Nlri:   nlri1,
+						Pattrs: attrs,
+					},
+				})
+			}
+		}
+	}()
+
+	for i := 0; i < 100; i++ {
+		metrics, err := registry.Gather()
+		assert.Nil(err)
+		assert.NotEmpty(metrics)
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	cancel()
+	<-ch
+}

pkg/server/server.go

@@ -30,7 +30,7 @@ import (
 
 	"github.com/eapache/channels"
 	"github.com/google/uuid"
-	"github.com/grpc-ecosystem/go-grpc-prometheus"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	"github.com/prometheus/client_golang/prometheus"
 	"google.golang.org/grpc"