Merge branch 'send_consumer_usage'

internal/usage/usage.go

@@ -10,12 +10,14 @@ import (
 	"math/rand"
 	"net/http"
 	"runtime"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/centrifugal/centrifugo/v5/internal/build"
+	"github.com/centrifugal/centrifugo/v5/internal/consuming"
 	"github.com/centrifugal/centrifugo/v5/internal/rule"
 
 	"github.com/centrifugal/centrifuge"
@@ -89,6 +91,8 @@ type Features struct {
 	SubRefreshProxy      bool
 	SubscribeStreamProxy bool
 
+	EnabledConsumers []string
+
 	// Uses GRPC server API.
 	GrpcAPI bool
 	// Admin interface enabled.
@@ -393,6 +397,12 @@ func (s *Sender) prepareMetrics() ([]*metric, error) {
 	if s.features.SubscribeStreamProxy {
 		metrics = append(metrics, createPoint("proxies_enabled.subscribe_stream"))
 	}
+	if len(s.features.EnabledConsumers) > 0 {
+		metrics = append(metrics, createPoint("features_enabled.consumers"))
+	}
+	for _, consumerType := range s.features.EnabledConsumers {
+		metrics = append(metrics, createPoint("consumers_enabled."+consumerType))
+	}
 	if s.features.GrpcAPI {
 		metrics = append(metrics, createPoint("features_enabled.grpc_api"))
 	}
@@ -602,3 +612,13 @@ func (s *Sender) sendUsageStats(metrics []*metric, statsEndpoint, statsToken str
 
 	return nil
 }
+
+func GetEnabledConsumers(consumers []consuming.ConsumerConfig) []string {
+	var enabledConsumers []string
+	for _, c := range consumers {
+		if !c.Disabled && !slices.Contains(enabledConsumers, string(c.Type)) {
+			enabledConsumers = append(enabledConsumers, string(c.Type))
+		}
+	}
+	return enabledConsumers
+}

internal/usage/usage_test.go

@@ -1,6 +1,13 @@
 package usage
 
-import "testing"
+import (
+	"testing"
+
+	"github.com/centrifugal/centrifugo/v5/internal/rule"
+
+	"github.com/centrifugal/centrifuge"
+	"github.com/stretchr/testify/require"
+)
 
 func Test_getHistogramMetric(t *testing.T) {
 	type args struct {
@@ -48,3 +55,28 @@ func Test_getHistogramMetric(t *testing.T) {
 		})
 	}
 }
+
+func nodeWithMemoryEngine(t *testing.T) *centrifuge.Node {
+	n, err := centrifuge.New(centrifuge.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = n.Run()
+	if err != nil {
+		t.Fatal(err)
+	}
+	return n
+}
+
+func TestPrepareMetrics(t *testing.T) {
+	node := nodeWithMemoryEngine(t)
+	ruleConfig := rule.DefaultConfig
+	ruleContainer, err := rule.NewContainer(ruleConfig)
+	require.NoError(t, err)
+	sender := NewSender(node, ruleContainer, Features{})
+	err = sender.updateMaxValues()
+	require.NoError(t, err)
+	metrics, err := sender.prepareMetrics()
+	require.NoError(t, err)
+	require.True(t, len(metrics) > 0)
+}

main.go

@@ -849,6 +849,8 @@ func main() {
 					UniSSE:        viper.GetBool("uni_sse"),
 					UniGRPC:       viper.GetBool("uni_grpc"),
 
+					EnabledConsumers: usage.GetEnabledConsumers(consumers),
+
 					GrpcAPI:             viper.GetBool("grpc_api"),
 					SubscribeToPersonal: viper.GetBool("user_subscribe_to_personal"),
 					Admin:               viper.GetBool("admin"),