@@ -1566,6 +1566,185 @@ func TestBatchClient_Metrics(t *testing.T) {
15661566 })
15671567}
15681568
1569+ func TestBatchClient_ClientNameMetricAttribute (t * testing.T ) {
1570+ const clientName = "test_client"
1571+
1572+ batchMetricNames := []string {
1573+ "chip_ingress.batch.send_requests_total" ,
1574+ "chip_ingress.batch.request_size_messages" ,
1575+ "chip_ingress.batch.request_size_bytes" ,
1576+ "chip_ingress.batch.request_latency_ms" ,
1577+ "chip_ingress.batch.config.info" ,
1578+ "chip_ingress.batch.batch_splits_total" ,
1579+ "chip_ingress.batch.results_mismatch_total" ,
1580+ }
1581+
1582+ t .Run ("with WithClientName sets client_name on all batch metrics" , func (t * testing.T ) {
1583+ reader , restore := useTestMeterProvider (t )
1584+ defer restore ()
1585+
1586+ events := []* chipingress.CloudEventPb {
1587+ largeTestEvent ("chip-client-1" ),
1588+ largeTestEvent ("chip-client-2" ),
1589+ largeTestEvent ("chip-client-3" ),
1590+ }
1591+ msgs2 := []* messageWithCallback {{event : events [0 ]}, {event : events [1 ]}}
1592+ _ , maxRequestSize := newBatchRequest (msgs2 , false )
1593+
1594+ mockClient := mocks .NewClient (t )
1595+ done := make (chan struct {})
1596+ var mu sync.Mutex
1597+ var publishCount int
1598+ mockClient .
1599+ On ("PublishBatch" , mock .Anything , mock .Anything ).
1600+ Return (& chipingress.PublishResponse {}, nil ).
1601+ Run (func (_ mock.Arguments ) {
1602+ mu .Lock ()
1603+ publishCount ++
1604+ if publishCount == 2 {
1605+ close (done )
1606+ }
1607+ mu .Unlock ()
1608+ })
1609+
1610+ client , err := NewBatchClient (
1611+ mockClient ,
1612+ WithClientName (clientName ),
1613+ WithBatchSize (1 ),
1614+ WithBatchInterval (time .Second ),
1615+ WithMessageBuffer (10 ),
1616+ WithMaxGRPCRequestSize (minMaxGRPCRequestSize ),
1617+ )
1618+ require .NoError (t , err )
1619+ client .maxGRPCRequestSize = maxRequestSize
1620+ client .effectiveMaxRequestSize = maxRequestSize
1621+ client .metrics .recordConfig (t .Context (), client )
1622+
1623+ messages := make ([]* messageWithCallback , 0 , len (events ))
1624+ for _ , event := range events {
1625+ messages = append (messages , & messageWithCallback {event : event })
1626+ }
1627+ client .sendBatch (t .Context (), messages )
1628+
1629+ select {
1630+ case <- done :
1631+ case <- time .After (time .Second ):
1632+ t .Fatal ("timeout waiting for split batches" )
1633+ }
1634+
1635+ client .completeBatchCallbacksFromResults (
1636+ []* messageWithCallback {
1637+ {event : & chipingress.CloudEventPb {Id : "m1" , Source : "s" , Type : "t" }},
1638+ {event : & chipingress.CloudEventPb {Id : "m2" , Source : "s" , Type : "t" }},
1639+ },
1640+ []* chipingress.PublishResult {{EventId : "m1" }},
1641+ )
1642+ client .completeBatchCallbacksFromResults (
1643+ []* messageWithCallback {
1644+ {event : & chipingress.CloudEventPb {Id : "m1" , Source : "s" , Type : "t" }},
1645+ },
1646+ []* chipingress.PublishResult {{EventId : "wrong-id" }},
1647+ )
1648+
1649+ rm := collectResourceMetrics (t , reader )
1650+ for _ , name := range batchMetricNames {
1651+ metric := mustMetric (t , rm , name )
1652+ assertMetricHasClientName (t , metric , clientName )
1653+ }
1654+ })
1655+
1656+ t .Run ("without WithClientName omits client_name" , func (t * testing.T ) {
1657+ reader , restore := useTestMeterProvider (t )
1658+ defer restore ()
1659+
1660+ mockClient := mocks .NewClient (t )
1661+ mockClient .EXPECT ().Close ().Return (nil ).Maybe ()
1662+ done := make (chan struct {})
1663+ mockClient .
1664+ On ("PublishBatch" , mock .Anything , mock .Anything ).
1665+ Return (& chipingress.PublishResponse {}, nil ).
1666+ Run (func (_ mock.Arguments ) { close (done ) }).
1667+ Once ()
1668+
1669+ client , err := NewBatchClient (
1670+ mockClient ,
1671+ WithBatchSize (1 ),
1672+ WithBatchInterval (time .Second ),
1673+ WithMessageBuffer (10 ),
1674+ )
1675+ require .NoError (t , err )
1676+ client .Start (t .Context ())
1677+
1678+ require .NoError (t , client .QueueMessage (& chipingress.CloudEventPb {
1679+ Id : "no-chip-client" , Source : "platform" , Type : "Test" ,
1680+ }, nil ))
1681+
1682+ select {
1683+ case <- done :
1684+ case <- time .After (time .Second ):
1685+ t .Fatal ("timeout waiting for PublishBatch" )
1686+ }
1687+ client .Stop ()
1688+
1689+ rm := collectResourceMetrics (t , reader )
1690+ for _ , sm := range rm .ScopeMetrics {
1691+ for _ , metric := range sm .Metrics {
1692+ if ! strings .HasPrefix (metric .Name , "chip_ingress.batch." ) {
1693+ continue
1694+ }
1695+ assertMetricOmitsClientName (t , metric )
1696+ }
1697+ }
1698+ })
1699+ }
1700+
1701+ func assertMetricHasClientName (t * testing.T , metric metricdata.Metrics , clientName string ) {
1702+ t .Helper ()
1703+ forEachMetricAttrSet (t , metric , func (attrs attribute.Set ) {
1704+ assert .True (t , hasStringAttr (attrs , "client_name" , clientName ),
1705+ "metric %s missing client_name=%q" , metric .Name , clientName )
1706+ })
1707+ }
1708+
1709+ func assertMetricOmitsClientName (t * testing.T , metric metricdata.Metrics ) {
1710+ t .Helper ()
1711+ forEachMetricAttrSet (t , metric , func (attrs attribute.Set ) {
1712+ for _ , kv := range attrs .ToSlice () {
1713+ assert .NotEqual (t , "client_name" , string (kv .Key ), "metric %s should not have client_name" , metric .Name )
1714+ }
1715+ })
1716+ }
1717+
1718+ func forEachMetricAttrSet (t * testing.T , metric metricdata.Metrics , fn func (attribute.Set )) {
1719+ t .Helper ()
1720+ var count int
1721+ record := func (attrs attribute.Set ) {
1722+ count ++
1723+ fn (attrs )
1724+ }
1725+ switch data := metric .Data .(type ) {
1726+ case metricdata.Sum [int64 ]:
1727+ for _ , dp := range data .DataPoints {
1728+ record (dp .Attributes )
1729+ }
1730+ case metricdata.Histogram [int64 ]:
1731+ for _ , dp := range data .DataPoints {
1732+ record (dp .Attributes )
1733+ }
1734+ case metricdata.Histogram [float64 ]:
1735+ for _ , dp := range data .DataPoints {
1736+ record (dp .Attributes )
1737+ }
1738+ case metricdata.Gauge [int64 ]:
1739+ for _ , dp := range data .DataPoints {
1740+ record (dp .Attributes )
1741+ }
1742+ default :
1743+ t .Fatalf ("metric %s has unsupported type %T" , metric .Name , metric .Data )
1744+ }
1745+ require .NotZero (t , count , "metric %s has no datapoints" , metric .Name )
1746+ }
1747+
15691748func TestSplitMessagesByRequestSize (t * testing.T ) {
15701749 t .Run ("empty messages returns nil" , func (t * testing.T ) {
15711750 result := splitMessagesByRequestSize (nil , 1024 , false )
0 commit comments