UsageStats: Add connected users and client (#38811)

pkg/infra/usagestats/service.go

@@ -10,6 +10,7 @@ import (
 	"github.com/grafana/grafana/pkg/login/social"
 	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/services/alerting"
+	"github.com/grafana/grafana/pkg/services/live"
 	"github.com/grafana/grafana/pkg/services/sqlstore"
 	"github.com/grafana/grafana/pkg/setting"
 )
@@ -31,17 +32,29 @@ type UsageStatsService struct {
 	AlertingUsageStats alerting.UsageStatsQuerier
 	PluginManager      plugins.Manager
 	SocialService      social.Service
+	grafanaLive        *live.GrafanaLive
 
 	log log.Logger
 
 	oauthProviders           map[string]bool
 	externalMetrics          []MetricsFunc
 	concurrentUserStatsCache memoConcurrentUserStats
+	liveStats                liveUsageStats
+}
+
+type liveUsageStats struct {
+	numClientsMax int
+	numClientsMin int
+	numClientsSum int
+	numUsersMax   int
+	numUsersMin   int
+	numUsersSum   int
+	sampleCount   int
 }
 
 func ProvideService(cfg *setting.Cfg, bus bus.Bus, sqlStore *sqlstore.SQLStore,
 	alertingStats alerting.UsageStatsQuerier, pluginManager plugins.Manager,
-	socialService social.Service) *UsageStatsService {
+	socialService social.Service, grafanaLive *live.GrafanaLive) *UsageStatsService {
 	s := &UsageStatsService{
 		Cfg:                cfg,
 		Bus:                bus,
@@ -49,6 +62,7 @@ func ProvideService(cfg *setting.Cfg, bus bus.Bus, sqlStore *sqlstore.SQLStore,
 		AlertingUsageStats: alertingStats,
 		oauthProviders:     socialService.GetOAuthProviders(),
 		PluginManager:      pluginManager,
+		grafanaLive:        grafanaLive,
 		log:                log.New("infra.usagestats"),
 	}
 	return s
@@ -59,6 +73,7 @@ func (uss *UsageStatsService) Run(ctx context.Context) error {
 
 	sendReportTicker := time.NewTicker(time.Hour * 24)
 	updateStatsTicker := time.NewTicker(time.Minute * 30)
+
 	defer sendReportTicker.Stop()
 	defer updateStatsTicker.Stop()
 
@@ -68,8 +83,11 @@ func (uss *UsageStatsService) Run(ctx context.Context) error {
 			if err := uss.sendUsageStats(ctx); err != nil {
 				metricsLogger.Warn("Failed to send usage stats", "err", err)
 			}
+			// always reset live stats every report tick
+			uss.resetLiveStats()
 		case <-updateStatsTicker.C:
 			uss.updateTotalStats()
+			uss.sampleLiveStats()
 		case <-ctx.Done():
 			return ctx.Err()
 		}

pkg/infra/usagestats/usage_stats.go

@@ -78,6 +78,20 @@ func (uss *UsageStatsService) GetUsageReport(ctx context.Context) (UsageReport,
 	metrics["stats.folders_viewers_can_edit.count"] = statsQuery.Result.FoldersViewersCanEdit
 	metrics["stats.folders_viewers_can_admin.count"] = statsQuery.Result.FoldersViewersCanAdmin
 
+	liveUsersAvg := 0
+	liveClientsAvg := 0
+	if uss.liveStats.sampleCount > 0 {
+		liveUsersAvg = uss.liveStats.numUsersSum / uss.liveStats.sampleCount
+		liveClientsAvg = uss.liveStats.numClientsSum / uss.liveStats.sampleCount
+	}
+	metrics["stats.live_samples.count"] = uss.liveStats.sampleCount
+	metrics["stats.live_users_max.count"] = uss.liveStats.numUsersMax
+	metrics["stats.live_users_min.count"] = uss.liveStats.numUsersMin
+	metrics["stats.live_users_avg.count"] = liveUsersAvg
+	metrics["stats.live_clients_max.count"] = uss.liveStats.numClientsMax
+	metrics["stats.live_clients_min.count"] = uss.liveStats.numClientsMin
+	metrics["stats.live_clients_avg.count"] = liveClientsAvg
+
 	ossEditionCount := 1
 	enterpriseEditionCount := 0
 	if uss.Cfg.IsEnterprise {
@@ -279,9 +293,9 @@ func (uss *UsageStatsService) sendUsageStats(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+
 	data := bytes.NewBuffer(out)
 	sendUsageStats(data)
-
 	return nil
 }
 
@@ -302,6 +316,34 @@ var sendUsageStats = func(data *bytes.Buffer) {
 	}()
 }
 
+func (uss *UsageStatsService) sampleLiveStats() {
+	current := uss.grafanaLive.UsageStats()
+
+	uss.liveStats.sampleCount++
+	uss.liveStats.numClientsSum += current.NumClients
+	uss.liveStats.numUsersSum += current.NumUsers
+
+	if current.NumClients > uss.liveStats.numClientsMax {
+		uss.liveStats.numClientsMax = current.NumClients
+	}
+
+	if current.NumClients < uss.liveStats.numClientsMin {
+		uss.liveStats.numClientsMin = current.NumClients
+	}
+
+	if current.NumUsers > uss.liveStats.numUsersMax {
+		uss.liveStats.numUsersMax = current.NumUsers
+	}
+
+	if current.NumUsers < uss.liveStats.numUsersMin {
+		uss.liveStats.numUsersMin = current.NumUsers
+	}
+}
+
+func (uss *UsageStatsService) resetLiveStats() {
+	uss.liveStats = liveUsageStats{}
+}
+
 func (uss *UsageStatsService) updateTotalStats() {
 	if !uss.Cfg.MetricsEndpointEnabled || uss.Cfg.MetricsEndpointDisableTotalStats {
 		return

pkg/infra/usagestats/usage_stats_test.go

@@ -11,12 +11,14 @@ import (
 	"testing"
 	"time"
 
+	"github.com/grafana/grafana/pkg/api/routing"
 	"github.com/grafana/grafana/pkg/bus"
 	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/plugins/manager"
 	"github.com/grafana/grafana/pkg/services/alerting"
+	"github.com/grafana/grafana/pkg/services/live"
 	"github.com/grafana/grafana/pkg/services/sqlstore"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/stretchr/testify/assert"
@@ -322,6 +324,8 @@ func TestMetrics(t *testing.T) {
 			assert.Equal(t, 18, metrics.Get("stats.alert_rules.count").MustInt())
 			assert.Equal(t, 19, metrics.Get("stats.library_panels.count").MustInt())
 			assert.Equal(t, 20, metrics.Get("stats.library_variables.count").MustInt())
+			assert.Equal(t, 0, metrics.Get("stats.live_users.count").MustInt())
+			assert.Equal(t, 0, metrics.Get("stats.live_clients.count").MustInt())
 
 			assert.Equal(t, 9, metrics.Get("stats.ds."+models.DS_ES+".count").MustInt())
 			assert.Equal(t, 10, metrics.Get("stats.ds."+models.DS_PROMETHEUS+".count").MustInt())
@@ -608,5 +612,13 @@ func createService(t *testing.T, cfg setting.Cfg) *UsageStatsService {
 		AlertingUsageStats: &alertingUsageMock{},
 		externalMetrics:    make([]MetricsFunc, 0),
 		PluginManager:      &fakePluginManager{},
+		grafanaLive:        newTestLive(t),
 	}
 }
+
+func newTestLive(t *testing.T) *live.GrafanaLive {
+	cfg := &setting.Cfg{AppURL: "http://localhost:3000/"}
+	gLive, err := live.ProvideService(nil, cfg, routing.NewRouteRegister(), nil, nil, nil, nil, sqlstore.InitTestDB(t))
+	require.NoError(t, err)
+	return gLive
+}

pkg/services/live/live.go

@@ -47,15 +47,6 @@ var (
 	loggerCF = log.New("live.centrifuge")
 )
 
-func NewGrafanaLive() *GrafanaLive {
-	return &GrafanaLive{
-		channels: make(map[string]models.ChannelHandler),
-		GrafanaScope: CoreGrafanaScope{
-			Features: make(map[string]models.ChannelHandlerFactory),
-		},
-	}
-}
-
 // CoreGrafanaScope list of core features
 type CoreGrafanaScope struct {
 	Features map[string]models.ChannelHandlerFactory
@@ -341,6 +332,11 @@ type GrafanaLive struct {
 	storage          *database.Storage
 }
 
+type UsageStats struct {
+	NumClients int
+	NumUsers   int
+}
+
 func (g *GrafanaLive) getStreamPlugin(pluginID string) (backend.StreamHandler, error) {
 	plugin, ok := g.PluginManager.BackendPluginManager.Get(pluginID)
 	if !ok {
@@ -722,6 +718,13 @@ func (g *GrafanaLive) ClientCount(orgID int64, channel string) (int, error) {
 	return len(p.Presence), nil
 }
 
+func (g *GrafanaLive) UsageStats() UsageStats {
+	clients := g.node.Hub().NumClients()
+	users := g.node.Hub().NumUsers()
+
+	return UsageStats{NumClients: clients, NumUsers: users}
+}
+
 func (g *GrafanaLive) HandleHTTPPublish(ctx *models.ReqContext, cmd dtos.LivePublishCmd) response.Response {
 	addr, err := live.ParseChannel(cmd.Channel)
 	if err != nil {