feat: track event delivery stats

enterprise/reporting/event_stats.go

@@ -0,0 +1,56 @@
+package reporting
+
+import (
+ "database/sql"
+ "fmt"
+ "strconv"
+
+ "github.com/rudderlabs/rudder-go-kit/stats"
+ "github.com/rudderlabs/rudder-server/utils/types"
+)
+
+const (
+ EventsDeliveredMetricName = "events_delivered_total"
+ EventsAbortedMetricName = "events_aborted_total"
+)
+
+type EventStatsReporter struct {
+ stats stats.Stats
+ configSubscriber *configSubscriber
+}
+
+func NewEventStatsReporter(configSubscriber *configSubscriber, stats stats.Stats) *EventStatsReporter {
+ return &EventStatsReporter{
+ stats: stats,
+ configSubscriber: configSubscriber,
+ }
+}
+
+func (es *EventStatsReporter) Report(metrics []*types.PUReportedMetric, _ *sql.Tx) error {
+ // tracking delivery event stats which has state - succeeded, aborted
+ for _, metric := range metrics {
+ metric := metric
+ if !(metric.StatusDetail.Status == "aborted" || metric.StatusDetail.Status == "succeeded") {
+ continue
+ }
+ tags := stats.Tags{
+ "workspaceId": es.configSubscriber.WorkspaceIDFromSource(metric.ConnectionDetails.SourceID),
+ "sourceId": metric.ConnectionDetails.SourceID,
+ "destinationId": metric.ConnectionDetails.DestinationID,
+ "reportedBy": metric.PUDetails.PU,
+ "sourceCategory": metric.ConnectionDetails.SourceCategory,
+ "terminal": strconv.FormatBool(metric.PUDetails.TerminalPU),
+ }
+ fmt.Println(tags)
+ metricName := EventsDeliveredMetricName
+ if metric.StatusDetail.Status == "aborted" {
+ metricName = EventsAbortedMetricName
+ }
+ es.stats.NewTaggedStat(metricName, stats.CountType, tags).Count(int(metric.StatusDetail.Count))
+ }
+ return nil
+}
+
+func (es *EventStatsReporter) DatabaseSyncer(types.SyncerConfig) types.ReportingSyncer {
+ return func() {}
+}

enterprise/reporting/event_stats_test.go

@@ -0,0 +1,142 @@
+package reporting
+
+import (
+ "context"
+ "testing"
+
+ "github.com/golang/mock/gomock"
+ "github.com/stretchr/testify/require"
+
+ "github.com/rudderlabs/rudder-go-kit/logger"
+ "github.com/rudderlabs/rudder-go-kit/stats/memstats"
+ backendconfig "github.com/rudderlabs/rudder-server/backend-config"
+ mocksBackendConfig "github.com/rudderlabs/rudder-server/mocks/backend-config"
+ "github.com/rudderlabs/rudder-server/utils/pubsub"
+ "github.com/rudderlabs/rudder-server/utils/types"
+)
+
+func TestEventStatsReporter(t *testing.T) {
+ workspaceID := "test-workspace-id"
+ sourceID := "test-source-id"
+ destinationID := "test-destination-id"
+ reportedBy := "test-reported-by"
+ sourceCategory := "test-source-category"
+ ctrl := gomock.NewController(t)
+ mockBackendConfig := mocksBackendConfig.NewMockBackendConfig(ctrl)
+ mockBackendConfig.EXPECT().Subscribe(gomock.Any(), backendconfig.TopicBackendConfig).DoAndReturn(func(ctx context.Context, topic backendconfig.Topic) pubsub.DataChannel {
+ ch := make(chan pubsub.DataEvent, 1)
+ ch <- pubsub.DataEvent{
+ Data: map[string]backendconfig.ConfigT{
+ workspaceID: {
+ WorkspaceID: workspaceID,
+ Sources: []backendconfig.SourceT{
+ {
+ ID: sourceID,
+ Enabled: true,
+ SourceDefinition: backendconfig.SourceDefinitionT{
+ Category: sourceCategory,
+ },
+ Destinations: []backendconfig.DestinationT{
+ {
+ ID: destinationID,
+ Enabled: true,
+ },
+ },
+ },
+ },
+ },
+ },
+ Topic: string(backendconfig.TopicBackendConfig),
+ }
+ close(ch)
+ return ch
+ }).AnyTimes()
+
+ statsStore := memstats.New()
+ ctx := context.Background()
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+ cs := newConfigSubscriber(logger.NOP)
+
+ subscribeDone := make(chan struct{})
+ go func() {
+ defer close(subscribeDone)
+
+ cs.Subscribe(ctx, mockBackendConfig)
+ }()
+
+ testReports := []*types.PUReportedMetric{
+ {
+ ConnectionDetails: types.ConnectionDetails{
+ SourceID: sourceID,
+ DestinationID: destinationID,
+ SourceCategory: sourceCategory,
+ },
+ PUDetails: types.PUDetails{
+ PU: reportedBy,
+ TerminalPU: true,
+ },
+ StatusDetail: &types.StatusDetail{
+ Count: 10,
+ Status: "succeeded",
+ },
+ },
+ {
+ ConnectionDetails: types.ConnectionDetails{
+ SourceID: sourceID,
+ DestinationID: destinationID,
+ SourceCategory: sourceCategory,
+ },
+ PUDetails: types.PUDetails{
+ PU: reportedBy,
+ TerminalPU: false,
+ },
+ StatusDetail: &types.StatusDetail{
+ Count: 10,
+ Status: "aborted",
+ },
+ },
+ // {
+ // ConnectionDetails: types.ConnectionDetails{
+ // SourceID: "momentum",
+ // DestinationID: destinationID,
+ // SourceCategory: sourceCategory,
+ // },
+ // PUDetails: types.PUDetails{
+ // PU: reportedBy,
+ // TerminalPU: false,
+ // },
+ // StatusDetail: &types.StatusDetail{
+ // Count: 10,
+ // Status: "momentum",
+ // },
+ // },
+ }
+ esr := NewEventStatsReporter(cs, statsStore)
+ err := esr.Report(testReports, nil)
+ require.NoError(t, err)
+ require.Equal(t, statsStore.Get(EventsDeliveredMetricName, map[string]string{
+ "workspaceId": workspaceID,
+ "sourceId": sourceID,
+ "destinationId": destinationID,
+ "reportedBy": reportedBy,
+ "sourceCategory": sourceCategory,
+ "terminal": "true",
+ }).LastValue(), float64(10))
+ require.Equal(t, statsStore.Get(EventsAbortedMetricName, map[string]string{
+ "workspaceId": workspaceID,
+ "sourceId": sourceID,
+ "destinationId": destinationID,
+ "reportedBy": reportedBy,
+ "sourceCategory": sourceCategory,
+ "terminal": "false",
+ }).LastValue(), float64(10))
+ // require.Equal(t, statsStore.Get(EventsAbortedMetricName, map[string]string{
+ // "workspaceId": workspaceID,
+ // "sourceId": "momentum",
+ // "destinationId": destinationID,
+ // "reportedBy": reportedBy,
+ // "sourceCategory": sourceCategory,
+ // "terminal": "false",
+ // }), nil)
+}

enterprise/reporting/mediator.go

@@ -4,6 +4,8 @@ import (
  "context"
  "database/sql"
 
+ "github.com/rudderlabs/rudder-go-kit/stats"
+
  "golang.org/x/sync/errgroup"
 
  "github.com/rudderlabs/rudder-go-kit/config"
@@ -18,11 +20,13 @@ type Mediator struct {
  g *errgroup.Group
  ctx context.Context
  reporters []types.Reporting
+ stats stats.Stats
 }
 
 func NewReportingMediator(ctx context.Context, log logger.Logger, enterpriseToken string, backendConfig backendconfig.BackendConfig) *Mediator {
  rm := &Mediator{
- log: log,
+ log: log,
+ stats: stats.Default,
  }
  rm.g, rm.ctx = errgroup.WithContext(ctx)
 
@@ -52,6 +56,7 @@ func NewReportingMediator(ctx context.Context, log logger.Logger, enterpriseToke
  errorIndexReporter := NewErrorIndexReporter(rm.ctx, config.Default, rm.log, configSubscriber)
  rm.reporters = append(rm.reporters, errorIndexReporter)
  }
+ rm.reporters = append(rm.reporters, NewEventStatsReporter(configSubscriber, rm.stats))
 
  return rm
 }

enterprise/reporting/setup_test.go

@@ -79,21 +79,21 @@ func TestSetupForDelegates(t *testing.T) {
  errorReportingEnabled: false,
  enterpriseTokenExists: true,
  errorIndexReportingEnabled: false,
- expectedDelegates: 1,
+ expectedDelegates: 2,
  },
  {
  reportingEnabled: true,
  errorReportingEnabled: true,
  enterpriseTokenExists: true,
  errorIndexReportingEnabled: false,
- expectedDelegates: 2,
+ expectedDelegates: 3,
  },
  {
  reportingEnabled: true,
  errorReportingEnabled: true,
  enterpriseTokenExists: true,
  errorIndexReportingEnabled: true,
- expectedDelegates: 3,
+ expectedDelegates: 4,
  },
  {
  reportingEnabled: true,