feat: add subscription buffer size usage metric (#22020)

CHANGELOG.md

@@ -6,6 +6,7 @@
 -	[#21752](https://github.com/influxdata/influxdb/pull/21752): feat: add total-buffer-bytes config parameter to subscriptions
 -	[#21820](https://github.com/influxdata/influxdb/pull/21820): chore: update flux to v0.120.1
 -	[#21983](https://github.com/influxdata/influxdb/pull/21983): feat: SHOW TAG VALUES should produce results from one specific RP
+-	[#22020](https://github.com/influxdata/influxdb/pull/22020): feat: add subscription buffer size usage metric
 
 ### Bugfixes
 

query/control/controller_test.go

@@ -208,7 +208,7 @@ func TestController_QueryRuntimeError(t *testing.T) {
 			q, err := ctrl.Query(context.Background(), &mock.Compiler{
 				CompileFn: func(ctx context.Context) (flux.Program, error) {
 					// ensure we have non-zero compile time
-					time.Sleep(1*time.Millisecond)
+					time.Sleep(1 * time.Millisecond)
 					return &mock.Program{
 						ExecuteFn: func(ctx context.Context, q *mock.Query, alloc *memory.Allocator) {
 							q.SetErr(errors.New("runtime error"))

services/snapshotter/service_test.go

@@ -492,4 +492,4 @@ func (m *MetaClient) Data() meta.Data {
 func (m *MetaClient) SetData(data *meta.Data) error {
 	m.data = *data.Clone()
 	return nil
-}
+}

services/subscriber/service.go

@@ -25,6 +25,7 @@ const (
 	statCreateFailures = "createFailures"
 	statPointsWritten  = "pointsWritten"
 	statWriteFailures  = "writeFailures"
+	statMemUsage       = "memUsage"
 )
 
 // WriteRequest is a parsed write request.
@@ -240,9 +241,12 @@ func (s *Service) Statistics(tags map[string]string) []models.Statistic {
 
 	s.mu.Lock()
 	defer s.mu.Unlock()
+	totalSize := int64(0)
 	for _, cw := range s.subs {
 		statistics = append(statistics, cw.Statistics(tags)...)
+		totalSize += atomic.LoadInt64(&cw.queueSize)
 	}
+	statistics[0].Values[statMemUsage] = totalSize
 
 	return statistics
 }