diff --git a/enterprise/activationrecords/noop.go b/enterprise/activationrecords/noop.go index cb30353816..dca0c40fda 100644 --- a/enterprise/activationrecords/noop.go +++ b/enterprise/activationrecords/noop.go @@ -17,7 +17,7 @@ func NewNoopActivationRecordsReporter() *NoopActivationRecordsReporter { return &NoopActivationRecordsReporter{} } -func (n *NoopActivationRecordsReporter) GenerateReportsFromJobs([]*jobsdb.JobT, map[string]string) []*ActivationRecord { +func (n *NoopActivationRecordsReporter) GenerateReportsFromJobs([]*jobsdb.JobT, map[string]SourceMetadata) []*ActivationRecord { return nil } diff --git a/enterprise/activationrecords/records_reporter.go b/enterprise/activationrecords/records_reporter.go index 8d692513d3..4991c17e70 100644 --- a/enterprise/activationrecords/records_reporter.go +++ b/enterprise/activationrecords/records_reporter.go @@ -34,12 +34,29 @@ const ( activationRecordsTable = "activation_records_reports" // retlSourceCategory is the SourceDefinition.Category of reverse-ETL ("warehouse - // actions") sources. MAR meters activation records from these sources only; the - // classification is resolved from the source's config category (via the source_id -> - // category map passed to GenerateReportsFromJobs), not from the job's source_category param. + // actions") sources. MAR meters activation records from these sources only when + // the source definition name is also allow-listed; classification is resolved + // from backend-config source metadata, not from the job's source_category param. retlSourceCategory = "warehouse" ) +var defaultAllowedSourceDefinitionNames = []string{ + "postgres", + "redshift", + "snowflake", + "bigquery", + "mysql", + "databricks", + "trino", +} + +// SourceMetadata is the source-definition metadata needed to classify whether a +// job belongs to a real reverse-ETL warehouse source for MAR metering. +type SourceMetadata struct { + Category string + Name string +} + // recordKey is the aggregation grain for activation records: one HLL sketch per // (workspace, source, destination). origin is intentionally NOT part of the key — // it is constant per source, so it is carried on the report as a plain column @@ -67,18 +84,19 @@ type ActivationRecord struct { // ActivationRecordsReporter is the interface to report monthly active records (MAR). type ActivationRecordsReporter interface { - GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceCategoriesBySourceID map[string]string) []*ActivationRecord + GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceMetadataBySourceID map[string]SourceMetadata) []*ActivationRecord ReportActivationRecords(ctx context.Context, reports []*ActivationRecord, tx *txn.Tx) error MigrateDatabase(dbConn string, conf *config.Config) error } // UniqueActivationRecordsReporter implements ActivationRecordsReporter using HLL sketches. type UniqueActivationRecordsReporter struct { - log logger.Logger - hllSettings *hll.Settings - instanceID string - now func() time.Time - stats stats.Stats + log logger.Logger + hllSettings *hll.Settings + instanceID string + now func() time.Time + stats stats.Stats + allowedSourceDefinitionNames config.ValueLoader[[]string] } // NewUniqueActivationRecordsReporter constructs a UniqueActivationRecordsReporter. @@ -100,6 +118,10 @@ func NewUniqueActivationRecordsReporter(log logger.Logger, conf *config.Config, hllSettings: hllSettings, instanceID: conf.GetStringVar("1", "INSTANCE_ID"), stats: stats, + allowedSourceDefinitionNames: conf.GetReloadableStringSliceVar( + defaultAllowedSourceDefinitionNames, + "ActivationRecords.allowedSourceDefinitionNames", + ), now: func() time.Time { return timeutil.Now() }, @@ -131,33 +153,34 @@ func (u *UniqueActivationRecordsReporter) MigrateDatabase(dbConn string, conf *c // GenerateReportsFromJobs aggregates activation records from a batch of jobs. // It is FAIL-CLOSED: jobs missing fingerprint or origin are skipped (counted via stats). -// sourceCategoriesBySourceID maps a source_id to its SourceDefinition.Category (from the -// backend config); reverse-ETL classification is done via this map rather than the job's -// source_category param, which is not populated on every ingestion path. -func (u *UniqueActivationRecordsReporter) GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceCategoriesBySourceID map[string]string) []*ActivationRecord { +// sourceMetadataBySourceID maps a source_id to source-definition metadata (from +// backend config); reverse-ETL classification is done via this map rather than +// the job's source_category param, which is not populated on every ingestion path. +func (u *UniqueActivationRecordsReporter) GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceMetadataBySourceID map[string]SourceMetadata) []*ActivationRecord { if len(jobs) == 0 { return nil } accumulators := make(map[recordKey]*recordAccumulator) + allowedSourceDefinitionNames := normalizedSourceDefinitionNames(u.allowedSourceDefinitionNames.Load()) for _, job := range jobs { - if job.WorkspaceId == "" { - u.log.Warnn("workspace_id not found in job", logger.NewIntField("jobId", job.JobID)) - u.recordSkip("missing_workspace") - continue - } - sourceID := jsonparser.GetStringOrEmpty(job.Parameters, "source_id") - // MAR meters reverse-ETL (warehouse) sources only. Classify by the source's - // SourceDefinition.Category from the backend config (looked up by source_id), NOT + // MAR meters real reverse-ETL SQL warehouse sources only. Classify by + // SourceDefinition metadata from backend config (looked up by source_id), NOT // the job's source_category param: the internal-batch ingestion path leaves that - // param empty, so trusting it would under-count. This also prevents a client from - // being metered by stamping context.activation.fingerprint on a non-rETL source. - // A missing or unknown source_id resolves to "" here and is skipped. Non-rETL is - // the expected majority of traffic, so skip it silently — no per-job skip stat. - if !strings.EqualFold(sourceCategoriesBySourceID[sourceID], retlSourceCategory) { + // param empty, so trusting it would under-count. Requiring both category=warehouse + // and an allow-listed source-definition name avoids metering cloud-storage sources + // such as S3 that also use category=warehouse. Missing, unknown, non-warehouse, and + // warehouse-but-not-allow-listed sources are skipped silently — no per-job skip stat. + if !u.isAllowedWarehouseSource(sourceMetadataBySourceID[sourceID], allowedSourceDefinitionNames) { + continue + } + + if job.WorkspaceId == "" { + u.log.Warnn("workspace_id not found in job", logger.NewIntField("jobId", job.JobID)) + u.recordSkip("missing_workspace") continue } @@ -246,6 +269,29 @@ func (u *UniqueActivationRecordsReporter) GenerateReportsFromJobs(jobs []*jobsdb return reports } +func (u *UniqueActivationRecordsReporter) isAllowedWarehouseSource(source SourceMetadata, allowedSourceDefinitionNames map[string]struct{}) bool { + if !strings.EqualFold(source.Category, retlSourceCategory) { + return false + } + _, ok := allowedSourceDefinitionNames[normalizeSourceDefinitionName(source.Name)] + return ok +} + +func normalizedSourceDefinitionNames(names []string) map[string]struct{} { + allowed := make(map[string]struct{}, len(names)) + for _, name := range names { + normalized := normalizeSourceDefinitionName(name) + if normalized != "" { + allowed[normalized] = struct{}{} + } + } + return allowed +} + +func normalizeSourceDefinitionName(name string) string { + return strings.ToLower(strings.TrimSpace(name)) +} + // ReportActivationRecords writes activation records to the database via a COPY statement. func (u *UniqueActivationRecordsReporter) ReportActivationRecords(ctx context.Context, reports []*ActivationRecord, tx *txn.Tx) error { if len(reports) == 0 { diff --git a/enterprise/activationrecords/records_reporter_test.go b/enterprise/activationrecords/records_reporter_test.go index 3498ad81b5..c9a084e9cf 100644 --- a/enterprise/activationrecords/records_reporter_test.go +++ b/enterprise/activationrecords/records_reporter_test.go @@ -11,6 +11,7 @@ import ( "github.com/rudderlabs/rudder-go-kit/config" "github.com/rudderlabs/rudder-go-kit/logger" "github.com/rudderlabs/rudder-go-kit/stats" + "github.com/rudderlabs/rudder-go-kit/stats/memstats" "github.com/rudderlabs/rudder-server/jobsdb" ) @@ -18,7 +19,7 @@ import ( func TestUniqueActivationRecordsReporter(t *testing.T) { // prepareJob builds a job with the standard activation payload shape. The Parameters // intentionally omit source_category: on the internal-batch ingestion path it is empty, - // and the reporter classifies reverse-ETL sources from the config map (categoriesByID + // and the reporter classifies reverse-ETL sources from the config map (metadataByID // below), not the param. prepareJob := func(sourceID, destinationID, fingerprint, origin, workspaceID string) *jobsdb.JobT { return &jobsdb.JobT{ @@ -43,13 +44,14 @@ func TestUniqueActivationRecordsReporter(t *testing.T) { } } - // categoriesByID maps source_id -> SourceDefinition.Category, mirroring the map the - // processor passes from the backend config. "src1" is a reverse-ETL (warehouse) - // source; "src-eventstream" is not. Unknown source_ids resolve to "" and are treated - // as non-rETL. - categoriesByID := map[string]string{ - "src1": "warehouse", - "src-eventstream": "eventStream", + // metadataByID maps source_id -> SourceDefinition metadata, mirroring the map the + // processor passes from the backend config. "src1" is a reverse-ETL SQL warehouse + // source; "src-eventstream" is not. Unknown source_ids resolve to zero-value metadata + // and are treated as non-rETL. + metadataByID := map[string]SourceMetadata{ + "src1": {Category: "warehouse", Name: "Snowflake"}, + "src-eventstream": {Category: "eventStream", Name: "javascript"}, + "src-s3": {Category: "warehouse", Name: "s3"}, } t.Run("constructor validates HLL settings", func(t *testing.T) { @@ -92,7 +94,7 @@ func TestUniqueActivationRecordsReporter(t *testing.T) { }, }, { - name: "missing fingerprint - skip", + name: "allow-listed warehouse missing fingerprint - skip", jobs: []*jobsdb.JobT{ prepareJob("src1", "dst1", "", "org1", "ws1"), }, @@ -137,6 +139,15 @@ func TestUniqueActivationRecordsReporter(t *testing.T) { require.Empty(t, reports) }, }, + { + name: "warehouse source not in allow-list - skip silently even with fingerprint", + jobs: []*jobsdb.JobT{ + prepareJob("src-s3", "dst1", "fp-cloud-storage", "org1", "ws1"), + }, + verify: func(t *testing.T, reports []*ActivationRecord) { + require.Empty(t, reports) + }, + }, { name: "config category is authoritative over a spoofed source_category param - skip", jobs: []*jobsdb.JobT{ @@ -204,12 +215,64 @@ func TestUniqueActivationRecordsReporter(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - reports := reporter.GenerateReportsFromJobs(tc.jobs, categoriesByID) + reports := reporter.GenerateReportsFromJobs(tc.jobs, metadataByID) tc.verify(t, reports) }) } }) + t.Run("GenerateReportsFromJobs_SourceDefinitionGateStats", func(t *testing.T) { + t.Run("warehouse source not in allow-list skips silently without missing fingerprint stat", func(t *testing.T) { + statsStore, err := memstats.New() + require.NoError(t, err) + reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.New(), statsStore) + require.NoError(t, err) + + reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{ + prepareJob("src-s3", "dst1", "", "org1", "ws1"), + }, metadataByID) + + require.Empty(t, reports) + require.Empty(t, statsStore.GetByName("activation_records_skipped")) + }) + + t.Run("allow-listed warehouse missing fingerprint increments missing_fingerprint", func(t *testing.T) { + statsStore, err := memstats.New() + require.NoError(t, err) + reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.New(), statsStore) + require.NoError(t, err) + + reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{ + prepareJob("src1", "dst1", "", "org1", "ws1"), + }, metadataByID) + + require.Empty(t, reports) + metric := statsStore.Get("activation_records_skipped", stats.Tags{"reason": "missing_fingerprint"}) + require.NotNil(t, metric) + require.Equal(t, float64(1), metric.LastValue()) + }) + + t.Run("allow-list config changes are observed without reconstructing reporter", func(t *testing.T) { + conf := config.New() + conf.Set("ActivationRecords.allowedSourceDefinitionNames", []string{"snowflake"}) + reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, conf, stats.NOP) + require.NoError(t, err) + + customMetadataByID := map[string]SourceMetadata{ + "src-custom": {Category: "warehouse", Name: "custom_sql"}, + } + job := prepareJob("src-custom", "dst1", "fp1", "org1", "ws1") + + require.Empty(t, reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, customMetadataByID)) + + conf.Set("ActivationRecords.allowedSourceDefinitionNames", []string{"custom_sql"}) + + reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, customMetadataByID) + require.Len(t, reports, 1) + require.Equal(t, uint64(1), reports[0].FingerprintHll.Cardinality()) + }) + }) + t.Run("GenerateReportsFromJobs_MultiEventBatch", func(t *testing.T) { reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.Default, stats.NOP) require.NoError(t, err) @@ -231,7 +294,7 @@ func TestUniqueActivationRecordsReporter(t *testing.T) { t.Run("two distinct fingerprints in same batch => cardinality 2", func(t *testing.T) { job := prepareTwoEventJob("src1", "dst1", "fp-1", "fp-2", "org1", "ws1") - reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, categoriesByID) + reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID) require.Len(t, reports, 1) require.Equal(t, "org1", reports[0].Origin) require.NotNil(t, reports[0].FingerprintHll) @@ -248,7 +311,7 @@ func TestUniqueActivationRecordsReporter(t *testing.T) { CustomVal: "GW", WorkspaceId: "ws1", } - reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, categoriesByID) + reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID) require.Len(t, reports, 1) require.Equal(t, uint64(1), reports[0].FingerprintHll.Cardinality()) }) @@ -271,11 +334,11 @@ func TestUniqueActivationRecordsReporter(t *testing.T) { job := prepareJob("src1", "dst1", "fp1", "org1", "ws1") - first := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, categoriesByID) + first := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID) require.Len(t, first, 1) require.Equal(t, uint64(1), first[0].FingerprintHll.Cardinality()) - second := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, categoriesByID) + second := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID) require.Len(t, second, 1) require.Equal(t, uint64(1), second[0].FingerprintHll.Cardinality()) diff --git a/enterprise/activationrecords/wire_compat_test.go b/enterprise/activationrecords/wire_compat_test.go index c64caea2be..43f8d33ba9 100644 --- a/enterprise/activationrecords/wire_compat_test.go +++ b/enterprise/activationrecords/wire_compat_test.go @@ -92,7 +92,9 @@ func TestWireCompat(t *testing.T) { }), } - reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, map[string]string{"src-1": "warehouse"}) + reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, map[string]SourceMetadata{ + "src-1": {Category: "warehouse", Name: "snowflake"}, + }) require.Len(t, reports, 1) // The reporter must resolve the full grain from the gateway's actual on-the-wire diff --git a/integration_test/activationrecordsreporting/activation_records_reporting_test.go b/integration_test/activationrecordsreporting/activation_records_reporting_test.go index 8efe17c0ec..75c90db5f6 100644 --- a/integration_test/activationrecordsreporting/activation_records_reporting_test.go +++ b/integration_test/activationrecordsreporting/activation_records_reporting_test.go @@ -315,8 +315,9 @@ func setup(t testing.TB) testConfig { // destination into source.Destinations so the gateway's authDestIDForSource can // resolve the X-Rudder-Destination-Id header into the job's destination_id // parameter (which MAR metering requires). The source carries the "warehouse" - // category (reverse-ETL): MAR meters warehouse sources only, so the gateway stamps - // source_category=warehouse into the job params for these records to be metered. + // category and an allow-listed source definition name (reverse-ETL): MAR meters + // only real warehouse sources, so the gateway stamps source_category=warehouse + // into the job params and processor source metadata keeps these records in scope. // ConfigBuilder.WithConnection additionally registers the top-level connection the // router requires before it will deliver warehouse-source jobs (router/worker.go). bcServer := backendconfigtest.NewBuilder(). @@ -327,6 +328,7 @@ func setup(t testing.TB) testConfig { WithWorkspaceID(workspaceID). WithID(sourceID). WithSourceCategory("warehouse"). + WithSourceType("snowflake"). WithConnection( backendconfigtest.NewDestinationBuilder("WEBHOOK"). WithID(destinationID). diff --git a/processor/processor.go b/processor/processor.go index 2beb773e20..85bbfcc023 100644 --- a/processor/processor.go +++ b/processor/processor.go @@ -107,7 +107,7 @@ type trackedUsersReporter interface { } type activationRecordsReporter interface { - GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceCategoriesBySourceID map[string]string) []*activationrecords.ActivationRecord + GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceMetadataBySourceID map[string]activationrecords.SourceMetadata) []*activationrecords.ActivationRecord ReportActivationRecords(ctx context.Context, reports []*activationrecords.ActivationRecord, tx *Tx) error } @@ -194,7 +194,7 @@ type Handle struct { eventAuditEnabled map[string]bool credentialsMap map[string][]types.Credential nonEventStreamSources map[string]bool - sourceIdCategoryMap map[string]string + sourceIDMetadataMap map[string]activationrecords.SourceMetadata enableConcurrentStore config.ValueLoader[bool] userTransformationMirroringSanitySampling config.ValueLoader[float64] userTransformationMirroringFireAndForget config.ValueLoader[bool] @@ -870,7 +870,7 @@ func (proc *Handle) backendConfigSubscriber(ctx context.Context) { eventAuditEnabled = make(map[string]bool) credentialsMap = make(map[string][]types.Credential) nonEventStreamSources = make(map[string]bool) - sourceIdCategoryMap = make(map[string]string) + sourceIDMetadataMap = make(map[string]activationrecords.SourceMetadata) connectionConfigMap = make(map[connection]backendconfig.Connection) ) for workspaceID, wConfig := range config { @@ -880,7 +880,10 @@ func (proc *Handle) backendConfigSubscriber(ctx context.Context) { for i := range wConfig.Sources { source := &wConfig.Sources[i] sourceIdSourceMap[source.ID] = *source - sourceIdCategoryMap[source.ID] = source.SourceDefinition.Category + sourceIDMetadataMap[source.ID] = activationrecords.SourceMetadata{ + Category: source.SourceDefinition.Category, + Name: source.SourceDefinition.Name, + } if source.Enabled { sourceIdDestinationMap[source.ID] = source.Destinations genericConsentManagementMap[SourceID(source.ID)] = make(DestConsentMap) @@ -922,7 +925,7 @@ func (proc *Handle) backendConfigSubscriber(ctx context.Context) { proc.config.eventAuditEnabled = eventAuditEnabled proc.config.credentialsMap = credentialsMap proc.config.nonEventStreamSources = nonEventStreamSources - proc.config.sourceIdCategoryMap = sourceIdCategoryMap + proc.config.sourceIDMetadataMap = sourceIDMetadataMap proc.config.configSubscriberLock.Unlock() if !initDone { initDone = true @@ -961,15 +964,15 @@ func (proc *Handle) getNonEventStreamSources() map[string]bool { return proc.config.nonEventStreamSources } -// getSourceCategoriesBySourceID returns the shared source_id -> SourceDefinition.Category +// getSourceMetadataBySourceID returns the shared source_id -> source-definition metadata // map, which the config subscriber rebuilds and swaps on each backend-config change (hence -// the read lock). Activation-records (MAR) metering uses it to classify reverse-ETL sources -// without depending on the source_category job param (which is not populated on every -// ingestion path). The returned map is shared and must not be mutated. -func (proc *Handle) getSourceCategoriesBySourceID() map[string]string { +// the read lock). Activation-records (MAR) metering uses it to classify real reverse-ETL +// warehouse sources without depending on the source_category job param (which is not populated +// on every ingestion path). The returned map is shared and must not be mutated. +func (proc *Handle) getSourceMetadataBySourceID() map[string]activationrecords.SourceMetadata { proc.config.configSubscriberLock.RLock() defer proc.config.configSubscriberLock.RUnlock() - return proc.config.sourceIdCategoryMap + return proc.config.sourceIDMetadataMap } func (proc *Handle) getEnabledDestinations(sourceId, destinationName string) []backendconfig.DestinationT { @@ -2457,11 +2460,12 @@ func (proc *Handle) pretransformStage(partition string, preTrans *preTransformat // GenerateReportsFromJobs reads context.activation from the raw job payloads // (left intact here so metering still works); the destination-bound copies are // already stripped per-event in preprocessStage. Fetch the config subscriber's - // current source_id -> category map once here (single lock; the map is swapped + // current source_id -> source-definition metadata map once here (single lock; + // the map is swapped // atomically on config change), then pass it to the reporter, which indexes it - // per job to classify reverse-ETL sources. - sourceCategoriesBySourceID := proc.getSourceCategoriesBySourceID() - activationRecordsReports := proc.activationRecordsReporter.GenerateReportsFromJobs(preTrans.jobList, sourceCategoriesBySourceID) + // per job to classify real reverse-ETL warehouse sources. + sourceMetadataBySourceID := proc.getSourceMetadataBySourceID() + activationRecordsReports := proc.activationRecordsReporter.GenerateReportsFromJobs(preTrans.jobList, sourceMetadataBySourceID) return &transformationMessage{ ctx: preTrans.subJobs.ctx, diff --git a/processor/processor_test.go b/processor/processor_test.go index 4bcbb7770b..1b74de3a91 100644 --- a/processor/processor_test.go +++ b/processor/processor_test.go @@ -1405,6 +1405,27 @@ var _ = Describe("Processor with event schemas v2", Ordered, func() { }) }) +func TestSourceMetadataBySourceID(t *testing.T) { + initProcessor() + + var c testContext + c.Setup(t) + defer c.Finish() + + c.mockGatewayJobsDB.EXPECT().DeleteExecuting() + + processor := NewHandle(config.New(), nil) + Setup(processor, &c, false, false, t) + defer processor.Shutdown() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + require.NoError(t, processor.config.asyncInit.WaitContext(ctx)) + + metadataBySourceID := processor.getSourceMetadataBySourceID() + require.Equal(t, activationrecords.SourceMetadata{Category: "webhook", Name: "fbla"}, metadataBySourceID[fblaSourceId]) +} + func TestArchival(t *testing.T) { initProcessor()