Add temporary prometheus metrics for snapshot size

state/accumulator.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/matrix-org/sliding-sync/internal"
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/getsentry/sentry-go"
 
@@ -22,12 +23,13 @@ import (
 // Accumulate function for timeline events. v2 sync must be called with a large enough timeline.limit
 // for this to work!
 type Accumulator struct {
-	db            *sqlx.DB
-	roomsTable    *RoomsTable
-	eventsTable   *EventTable
-	snapshotTable *SnapshotTable
-	spacesTable   *SpacesTable
-	entityName    string
+	db              *sqlx.DB
+	roomsTable      *RoomsTable
+	eventsTable     *EventTable
+	snapshotTable   *SnapshotTable
+	spacesTable     *SpacesTable
+	entityName      string
+	snapshotSizeVec *prometheus.HistogramVec // TODO: Remove, this is temporary to get a feeling how often a new snapshot is created
 }
 
 func NewAccumulator(db *sqlx.DB) *Accumulator {
@@ -280,6 +282,10 @@ func (a *Accumulator) Initialise(roomID string, state []json.RawMessage) (Initia
 		if err != nil {
 			return fmt.Errorf("failed to insert snapshot: %w", err)
 		}
+		if a.snapshotSizeVec != nil {
+			logger.Trace().Str("room_id", roomID).Int("members", len(memberNIDs)).Msg("Inserted new snapshot")
+			a.snapshotSizeVec.WithLabelValues(roomID).Observe(float64(len(memberNIDs)))
+		}
 		res.AddedEvents = true
 		latestNID := int64(0)
 		for _, nid := range otherNIDs {
@@ -481,6 +487,10 @@ func (a *Accumulator) Accumulate(txn *sqlx.Tx, userID, roomID string, prevBatch
 			if err = a.snapshotTable.Insert(txn, newSnapshot); err != nil {
 				return 0, nil, fmt.Errorf("failed to insert new snapshot: %w", err)
 			}
+			if a.snapshotSizeVec != nil {
+				logger.Trace().Str("room_id", roomID).Int("members", len(memNIDs)).Msg("Inserted new snapshot")
+				a.snapshotSizeVec.WithLabelValues(roomID).Observe(float64(len(memNIDs)))
+			}
 			snapID = newSnapshot.SnapshotID
 		}
 		if err := a.eventsTable.UpdateBeforeSnapshotID(txn, ev.NID, beforeSnapID, replacesNID); err != nil {

state/storage.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/getsentry/sentry-go"
 	"github.com/lib/pq"
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/jmoiron/sqlx"
 	"github.com/matrix-org/sliding-sync/internal"
@@ -73,10 +74,10 @@ func NewStorage(postgresURI string) *Storage {
 		// TODO: if we panic(), will sentry have a chance to flush the event?
 		logger.Panic().Err(err).Str("uri", postgresURI).Msg("failed to open SQL DB")
 	}
-	return NewStorageWithDB(db)
+	return NewStorageWithDB(db, false)
 }
 
-func NewStorageWithDB(db *sqlx.DB) *Storage {
+func NewStorageWithDB(db *sqlx.DB, addPrometheusMetrics bool) *Storage {
 	acc := &Accumulator{
 		db:            db,
 		roomsTable:    NewRoomsTable(db),
@@ -85,6 +86,18 @@ func NewStorageWithDB(db *sqlx.DB) *Storage {
 		spacesTable:   NewSpacesTable(db),
 		entityName:    "server",
 	}
+
+	if addPrometheusMetrics {
+		acc.snapshotSizeVec = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Namespace: "sliding_sync",
+			Subsystem: "poller",
+			Name:      "snapshot_size",
+			Help:      "Number of membership events in a snapshot",
+			Buckets:   []float64{100.0, 500.0, 1000.0, 5000.0, 10000.0, 20000.0, 50000.0, 100000.0, 150000.0},
+		}, []string{"room_id"})
+		prometheus.MustRegister(acc.snapshotSizeVec)
+	}
+
 	return &Storage{
 		Accumulator:       acc,
 		ToDeviceTable:     NewToDeviceTable(db),

v3.go

@@ -104,7 +104,7 @@ func Setup(destHomeserver, postgresURI, secret string, opts Opts) (*handler2.Han
 	if opts.DBConnMaxIdleTime > 0 {
 		db.SetConnMaxIdleTime(opts.DBConnMaxIdleTime)
 	}
-	store := state.NewStorageWithDB(db)
+	store := state.NewStorageWithDB(db, opts.AddPrometheusMetrics)
 	storev2 := sync2.NewStoreWithDB(db, secret)
 
 	// Automatically execute migrations