Adds anonymous statistics reporting (#356)

* Adds anonymous statistics reporting

* Update go mod

* Review feedback

Co-authored-by: Christian Simon <simon@swine.de>

Author: cyriltovena

docs/sources/operators-guide/configure/reference-configuration-parameters/index.md

@@ -211,6 +211,11 @@ storage:
 # CLI flag: -auth.multitenancy-enabled
 [multitenancy_enabled: <boolean> | default = false]
 
+analytics:
+  # Enable anonymous usage reporting.
+  # CLI flag: -reporting.enabled
+  [reporting_enabled: <boolean> | default = true]
+
 # yaml file to load
 # CLI flag: -config.file
 [configfile: <string> | default = ""]

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/grafana/regexp v0.0.0-20220304095617-2e8d9baf4ac2
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.11.2
+	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/compress v1.15.9
 	github.com/minio/minio-go/v7 v7.0.23
 	github.com/mitchellh/go-wordwrap v1.0.0
@@ -163,7 +164,6 @@ require (
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/jpillora/backoff v1.0.0 // indirect
-	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/julienschmidt/httprouter v1.3.0 // indirect
 	github.com/klauspost/cpuid v1.3.1 // indirect
 	github.com/kolo/xmlrpc v0.0.0-20201022064351-38db28db192b // indirect

pkg/distributor/distributor.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"flag"
+	"fmt"
 	"hash/fnv"
 	"strconv"
 	"time"
@@ -27,18 +28,25 @@ import (
 	phlaremodel "github.com/grafana/phlare/pkg/model"
 	"github.com/grafana/phlare/pkg/pprof"
 	"github.com/grafana/phlare/pkg/tenant"
+	"github.com/grafana/phlare/pkg/usagestats"
 )
 
 type PushClient interface {
 	Push(context.Context, *connect.Request[pushv1.PushRequest]) (*connect.Response[pushv1.PushResponse], error)
 }
 
 // todo: move to non global metrics.
-var clients = promauto.NewGauge(prometheus.GaugeOpts{
-	Namespace: "phlare",
-	Name:      "distributor_ingester_clients",
-	Help:      "The current number of ingester clients.",
-})
+var (
+	clients = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: "phlare",
+		Name:      "distributor_ingester_clients",
+		Help:      "The current number of ingester clients.",
+	})
+	rfStats                 = usagestats.NewInt("distributor_replication_factor")
+	bytesReceivedStats      = usagestats.NewStatistics("distributor_bytes_received")
+	bytesReceivedTotalStats = usagestats.NewCounter("distributor_bytes_received_total")
+	profileReceivedStats    = usagestats.NewCounter("distributor_profiles_received")
+)
 
 // Config for a Distributor.
 type Config struct {
@@ -83,6 +91,7 @@ func New(cfg Config, ingestersRing ring.ReadRing, factory ring_client.PoolFactor
 	d.subservicesWatcher = services.NewFailureWatcher()
 	d.subservicesWatcher.WatchManager(d.subservices)
 	d.Service = services.NewBasicService(d.starting, d.running, d.stopping)
+	rfStats.Set(int64(ingestersRing.ReplicationFactor()))
 	return d, nil
 }
 
@@ -117,6 +126,10 @@ func (d *Distributor) Push(ctx context.Context, req *connect.Request[pushv1.Push
 		keys = append(keys, TokenFor(tenantID, labelsString(series.Labels)))
 		profName := phlaremodel.Labels(series.Labels).Get(scrape.ProfileName)
 		for _, raw := range series.Samples {
+			usagestats.NewCounter(fmt.Sprintf("distributor_profile_type_%s_received", profName)).Inc(1)
+			profileReceivedStats.Inc(1)
+			bytesReceivedTotalStats.Inc(int64(len(raw.RawProfile)))
+			bytesReceivedStats.Record(float64(len(raw.RawProfile)))
 			d.metrics.receivedCompressedBytes.WithLabelValues(profName).Observe(float64(len(raw.RawProfile)))
 			p, err := pprof.RawFromBytes(raw.RawProfile)
 			if err != nil {

pkg/gen/google/v1/profile.pb.go

@@ -22,9 +22,12 @@ import (
 	"github.com/grafana/phlare/pkg/phlaredb"
 	"github.com/grafana/phlare/pkg/pprof"
 	"github.com/grafana/phlare/pkg/tenant"
+	"github.com/grafana/phlare/pkg/usagestats"
 	"github.com/grafana/phlare/pkg/util"
 )
 
+var activeTenantsStats = usagestats.NewInt("ingester_active_tenants")
+
 type Config struct {
 	LifecyclerConfig ring.LifecyclerConfig `yaml:"lifecycler,omitempty"`
 }
@@ -137,6 +140,7 @@ func (i *Ingester) GetOrCreateInstance(tenantID string) (*instance, error) { //n
 			return nil, err
 		}
 		i.instances[tenantID] = inst
+		activeTenantsStats.Set(int64(len(i.instances)))
 	}
 	return inst, nil
 }

pkg/ingester/ingester.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"os"
 
 	grpchealth "github.com/bufbuild/connect-grpchealth-go"
 	"github.com/go-kit/log"
@@ -35,10 +36,13 @@ import (
 	"github.com/grafana/phlare/pkg/gen/push/v1/pushv1connect"
 	"github.com/grafana/phlare/pkg/gen/querier/v1/querierv1connect"
 	"github.com/grafana/phlare/pkg/ingester"
+	"github.com/grafana/phlare/pkg/objstore"
 	objstoreclient "github.com/grafana/phlare/pkg/objstore/client"
+	"github.com/grafana/phlare/pkg/objstore/providers/filesystem"
 	"github.com/grafana/phlare/pkg/openapiv2"
 	phlarecontext "github.com/grafana/phlare/pkg/phlare/context"
 	"github.com/grafana/phlare/pkg/querier"
+	"github.com/grafana/phlare/pkg/usagestats"
 	"github.com/grafana/phlare/pkg/util"
 	"github.com/grafana/phlare/pkg/util/build"
 )
@@ -55,6 +59,7 @@ const (
 	Querier      string = "querier"
 	GRPCGateway  string = "grpc-gateway"
 	Storage      string = "storage"
+	UsageReport  string = "usage-stats"
 
 	// RuntimeConfig            string = "runtime-config"
 	// Overrides                string = "overrides"
@@ -70,9 +75,10 @@ const (
 	// IndexGateway             string = "index-gateway"
 	// IndexGatewayRing         string = "index-gateway-ring"
 	// QueryScheduler           string = "query-scheduler"
-	// UsageReport              string = "usage-report"
 )
 
+var objectStoreTypeStats = usagestats.NewString("store_object_type")
+
 func (f *Phlare) initQuerier() (services.Service, error) {
 	q, err := querier.New(f.Cfg.Querier, f.ring, nil, f.logger, f.auth)
 	if err != nil {
@@ -138,6 +144,7 @@ func (f *Phlare) initMemberlistKV() (services.Service, error) {
 	f.Cfg.MemberlistKV.MetricsRegisterer = f.reg
 	f.Cfg.MemberlistKV.Codecs = []codec.Codec{
 		ring.GetCodec(),
+		usagestats.JSONCodec,
 	}
 
 	dnsProviderReg := prometheus.WrapRegistererWithPrefix(
@@ -166,6 +173,7 @@ func (f *Phlare) initRing() (_ services.Service, err error) {
 }
 
 func (f *Phlare) initStorage() (_ services.Service, err error) {
+	objectStoreTypeStats.Set(f.Cfg.Storage.Bucket.Backend)
 	if cfg := f.Cfg.Storage.Bucket; cfg.Backend != "filesystem" {
 		b, err := objstoreclient.NewBucket(
 			f.context(),
@@ -178,6 +186,10 @@ func (f *Phlare) initStorage() (_ services.Service, err error) {
 		f.storageBucket = b
 	}
 
+	if f.Cfg.Target.String() != All && f.storageBucket == nil {
+		return nil, errors.New("storage bucket configuration is required when running in microservices mode")
+	}
+
 	return nil, nil
 }
 
@@ -267,6 +279,39 @@ func (f *Phlare) initServer() (services.Service, error) {
 	return s, nil
 }
 
+func (f *Phlare) initUsageReport() (services.Service, error) {
+	if !f.Cfg.Analytics.Enabled {
+		return nil, nil
+	}
+	f.Cfg.Analytics.Leader = false
+	// ingester is the only component that can be a leader
+	if f.isModuleActive(Ingester) {
+		f.Cfg.Analytics.Leader = true
+	}
+
+	usagestats.Target(f.Cfg.Target.String())
+
+	var b objstore.Bucket
+	if f.storageBucket == nil {
+		if err := os.MkdirAll(f.Cfg.PhlareDB.DataPath, 0o777); err != nil {
+			return nil, fmt.Errorf("mkdir %s: %w", f.Cfg.PhlareDB.DataPath, err)
+		}
+		fs, err := filesystem.NewBucket(f.Cfg.PhlareDB.DataPath)
+		if err != nil {
+			return nil, err
+		}
+		b = fs
+	}
+
+	ur, err := usagestats.NewReporter(f.Cfg.Analytics, f.Cfg.Ingester.LifecyclerConfig.RingConfig.KVStore, b, f.logger, f.reg)
+	if err != nil {
+		level.Info(f.logger).Log("msg", "failed to initialize usage report", "err", err)
+		return nil, nil
+	}
+	f.usageReport = ur
+	return ur, nil
+}
+
 type statusService struct {
 	commonv1.UnimplementedStatusServiceServer
 	configYaml    string
@@ -360,6 +405,32 @@ func (f *Phlare) statusService() commonv1.StatusServiceServer {
 	}
 }
 
+func (f *Phlare) isModuleActive(m string) bool {
+	for _, target := range f.Cfg.Target {
+		if target == m {
+			return true
+		}
+		if f.recursiveIsModuleActive(target, m) {
+			return true
+		}
+	}
+	return false
+}
+
+func (f *Phlare) recursiveIsModuleActive(target, m string) bool {
+	if targetDeps, ok := f.deps[target]; ok {
+		for _, dep := range targetDeps {
+			if dep == m {
+				return true
+			}
+			if f.recursiveIsModuleActive(dep, m) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // NewServerService constructs service from Server component.
 // servicesToWaitFor is called when server is stopping, and should return all
 // services that need to terminate before server actually stops.

pkg/phlare/modules.go

@@ -42,6 +42,7 @@ import (
 	"github.com/grafana/phlare/pkg/querier"
 	"github.com/grafana/phlare/pkg/tenant"
 	"github.com/grafana/phlare/pkg/tracing"
+	"github.com/grafana/phlare/pkg/usagestats"
 	"github.com/grafana/phlare/pkg/util"
 )
 
@@ -58,8 +59,10 @@ type Config struct {
 
 	Storage StorageConfig `yaml:"storage"`
 
-	MultitenancyEnabled bool `yaml:"multitenancy_enabled,omitempty"`
-	ConfigFile          string
+	MultitenancyEnabled bool              `yaml:"multitenancy_enabled,omitempty"`
+	Analytics           usagestats.Config `yaml:"analytics"`
+
+	ConfigFile string `yaml:"-"`
 }
 
 func newDefaultConfig() *Config {
@@ -98,6 +101,7 @@ func (c *Config) RegisterFlagsWithContext(ctx context.Context, f *flag.FlagSet)
 	c.PhlareDB.RegisterFlags(f)
 	c.Tracing.RegisterFlags(f)
 	c.Storage.RegisterFlagsWithContext(ctx, f)
+	c.Analytics.RegisterFlags(f)
 }
 
 // registerServerFlagsWithChangedDefaultValues registers *Config.Server flags, but overrides some defaults set by the weaveworks package.
@@ -175,6 +179,7 @@ type Phlare struct {
 	ring               *ring.Ring
 	agent              *agent.Agent
 	pusherClient       pushv1connect.PusherServiceClient
+	usageReport        *usagestats.Reporter
 
 	storageBucket objstore.Bucket
 
@@ -185,6 +190,7 @@ type Phlare struct {
 
 func New(cfg Config) (*Phlare, error) {
 	logger := initLogger(&cfg.Server)
+	usagestats.Edition("oss")
 
 	phlare := &Phlare{
 		Cfg:    cfg,
@@ -234,15 +240,17 @@ func (f *Phlare) setupModuleManager() error {
 	mm.RegisterModule(Distributor, f.initDistributor)
 	mm.RegisterModule(Querier, f.initQuerier)
 	mm.RegisterModule(Agent, f.initAgent)
+	mm.RegisterModule(UsageReport, f.initUsageReport)
 	mm.RegisterModule(All, nil)
 
 	// Add dependencies
 	deps := map[string][]string{
 		All:          {Agent, Ingester, Distributor, Querier},
-		Distributor:  {Ring, Server},
-		Querier:      {Ring, Server},
+		UsageReport:  {Storage, MemberlistKV},
+		Distributor:  {Ring, Server, UsageReport},
+		Querier:      {Ring, Server, UsageReport},
 		Agent:        {Server},
-		Ingester:     {Server, MemberlistKV, Storage},
+		Ingester:     {Server, MemberlistKV, Storage, UsageReport},
 		Ring:         {Server, MemberlistKV},
 		MemberlistKV: {Server},
 		Server:       {GRPCGateway},