chore: make reporting 413 handling unconditional

enterprise/reporting/error_reporting.go

@@ -101,8 +101,7 @@ type ErrorDetailReporter struct {
 	maxOpenConnections    int
 	vacuumFull            config.ValueLoader[bool]
 
-	requestEntityTooLargeHandling config.ValueLoader[bool]
-	maxSampleEventSizeBytes       config.ValueLoader[int64]
+	maxSampleEventSizeBytes config.ValueLoader[int64]
 
 	errorDetailExtractor *ExtractorHandle
 	errorNormalizer      ErrorNormalizer
@@ -138,7 +137,6 @@ func NewErrorDetailReporter(
 	eventSamplingDuration := conf.GetReloadableDurationVar(60, time.Minute, "Reporting.eventSampling.durationInMinutes")
 	eventSamplerType := conf.GetReloadableStringVar("badger", "Reporting.eventSampling.type")
 	eventSamplingCardinality := conf.GetReloadableIntVar(100000, 1, "Reporting.eventSampling.cardinality")
-	requestEntityTooLargeHandling := conf.GetReloadableBoolVar(false, "Reporting.errorReporting.requestEntityTooLargeHandling", "Reporting.requestEntityTooLargeHandling")
 	maxSampleEventSizeBytes := conf.GetReloadableInt64Var(50*bytesize.MB, 1, "Reporting.errorReporting.maxSampleEventSizeBytes", "Reporting.maxSampleEventSizeBytes")
 
 	log := logger.NewLogger().Child("enterprise").Child("error-detail-reporting")
@@ -160,17 +158,15 @@ func NewErrorDetailReporter(
 	}
 
 	return &ErrorDetailReporter{
-		ctx:                   ctx,
-		cancel:                cancel,
-		g:                     g,
-		log:                   log,
-		sleepInterval:         sleepInterval,
-		mainLoopSleepInterval: mainLoopSleepInterval,
-		maxConcurrentRequests: maxConcurrentRequests,
-		vacuumFull:            conf.GetReloadableBoolVar(true, "Reporting.errorReporting.vacuumFull", "Reporting.vacuumFull"),
-
-		requestEntityTooLargeHandling: requestEntityTooLargeHandling,
-		maxSampleEventSizeBytes:       maxSampleEventSizeBytes,
+		ctx:                     ctx,
+		cancel:                  cancel,
+		g:                       g,
+		log:                     log,
+		sleepInterval:           sleepInterval,
+		mainLoopSleepInterval:   mainLoopSleepInterval,
+		maxConcurrentRequests:   maxConcurrentRequests,
+		vacuumFull:              conf.GetReloadableBoolVar(true, "Reporting.errorReporting.vacuumFull", "Reporting.vacuumFull"),
+		maxSampleEventSizeBytes: maxSampleEventSizeBytes,
 
 		eventSamplingEnabled:  eventSamplingEnabled,
 		eventSamplingDuration: eventSamplingDuration,
@@ -838,10 +834,6 @@ func (edr *ErrorDetailReporter) Stop() {
 }
 
 func (edr *ErrorDetailReporter) sendEDMetric(ctx context.Context, metric *types.EDMetric) error {
-	if !edr.requestEntityTooLargeHandling.Load() {
-		return edr.commonClient.SendWithoutFailFast(ctx, metric)
-	}
-
 	err := edr.commonClient.Send(ctx, metric)
 	if !errors.Is(err, client.ErrPayloadTooLarge) {
 		return err

enterprise/reporting/flusher/flusher.go

@@ -47,8 +47,6 @@ type Flusher struct {
 	vacuumFull                          config.ValueLoader[bool]
 	vacuumInterval                      config.ValueLoader[time.Duration]
 
-	requestEntityTooLargeHandling config.ValueLoader[bool]
-
 	minConcurrentRequests config.ValueLoader[int]
 	maxConcurrentRequests config.ValueLoader[int]
 	batchSizeToReporting  config.ValueLoader[int]
@@ -79,24 +77,22 @@ func NewFlusher(db *sql.DB, log logger.Logger, stats stats.Stats, conf *config.C
 	vacuumThresholdDeletedRows := conf.GetReloadableIntVar(100000, 1, "Reporting.flusher.vacuumThresholdDeletedRows")
 	vacuumInterval := conf.GetReloadableDurationVar(15, time.Minute, "Reporting.flusher.vacuumInterval", "Reporting.vacuumInterval")
 	vacuumThresholdBytes := conf.GetReloadableInt64Var(10*bytesize.GB, 1, "Reporting.flusher.vacuumThresholdBytes", "Reporting.vacuumThresholdBytes")
-	requestEntityTooLargeHandling := conf.GetReloadableBoolVar(false, "Reporting.flusher.requestEntityTooLargeHandling", "Reporting.requestEntityTooLargeHandling")
 
 	f := Flusher{
-		db:                            db,
-		log:                           log,
-		instanceId:                    conf.GetStringVar("1", "INSTANCE_ID"),
-		sleepInterval:                 sleepInterval,
-		flushWindow:                   flushWindow,
-		recentExclusionWindow:         recentExclusionWindow,
-		minConcurrentRequests:         minConcReqs,
-		maxConcurrentRequests:         maxConcReqs,
-		stats:                         stats,
-		batchSizeFromDB:               batchSizeFromDB,
-		vacuumThresholdDeletedRows:    vacuumThresholdDeletedRows,
-		vacuumFull:                    conf.GetReloadableBoolVar(false, "Reporting.flusher.vacuumFull", "Reporting.vacuumFull"),
-		vacuumInterval:                vacuumInterval,
-		vacuumThresholdBytes:          vacuumThresholdBytes,
-		requestEntityTooLargeHandling: requestEntityTooLargeHandling,
+		db:                         db,
+		log:                        log,
+		instanceId:                 conf.GetStringVar("1", "INSTANCE_ID"),
+		sleepInterval:              sleepInterval,
+		flushWindow:                flushWindow,
+		recentExclusionWindow:      recentExclusionWindow,
+		minConcurrentRequests:      minConcReqs,
+		maxConcurrentRequests:      maxConcReqs,
+		stats:                      stats,
+		batchSizeFromDB:            batchSizeFromDB,
+		vacuumThresholdDeletedRows: vacuumThresholdDeletedRows,
+		vacuumFull:                 conf.GetReloadableBoolVar(false, "Reporting.flusher.vacuumFull", "Reporting.vacuumFull"),
+		vacuumInterval:             vacuumInterval,
+		vacuumThresholdBytes:       vacuumThresholdBytes,
 
 		table:                               table,
 		aggregator:                          aggregator,
@@ -306,10 +302,6 @@ func (f *Flusher) send(ctx context.Context, aggReports []json.RawMessage) error
 }
 
 func (f *Flusher) sendBatch(ctx context.Context, batch []json.RawMessage) error {
-	if !f.requestEntityTooLargeHandling.Load() {
-		return f.commonClient.SendWithoutFailFast(ctx, batch)
-	}
-
 	err := f.commonClient.Send(ctx, batch)
 	if !errors.Is(err, client.ErrPayloadTooLarge) {
 		return err

enterprise/reporting/flusher/flusher_test.go

@@ -209,7 +209,6 @@ func newLargePayloadTestFlusher(t *testing.T, serverURL string, maxRetries, batc
 	t.Helper()
 	conf := config.New()
 	conf.Set("REPORTING_URL", serverURL)
-	conf.Set("Reporting.flusher.requestEntityTooLargeHandling", true)
 	conf.Set("Reporting.httpClient.backoff.maxRetries", maxRetries)
 	conf.Set("Reporting.flusher.batchSizeToReporting", batchSize)
 	conf.Set("Reporting.flusher.minConcurrentRequests", 1)

enterprise/reporting/reporting.go

@@ -74,8 +74,7 @@ type DefaultReporter struct {
 	eventSamplingDuration config.ValueLoader[time.Duration]
 	eventSampler          event_sampler.EventSampler
 
-	requestEntityTooLargeHandling config.ValueLoader[bool]
-	maxSampleEventSizeBytes       config.ValueLoader[int64]
+	maxSampleEventSizeBytes config.ValueLoader[int64]
 
 	eventNamePrefixLength config.ValueLoader[int]
 	eventNameSuffixLength config.ValueLoader[int]
@@ -97,7 +96,6 @@ func NewDefaultReporter(ctx context.Context, conf *config.Config, log logger.Log
 	maxOpenConnections := config.GetIntVar(32, 1, "Reporting.maxOpenConnections")
 	dbQueryTimeout = config.GetReloadableDurationVar(60, time.Second, "Reporting.dbQueryTimeout")
 	maxReportsCountInARequest := conf.GetReloadableIntVar(10, 1, "Reporting.maxReportsCountInARequest")
-	requestEntityTooLargeHandling := conf.GetReloadableBoolVar(false, "Reporting.requestEntityTooLargeHandling")
 	maxSampleEventSizeBytes := conf.GetReloadableInt64Var(80*bytesize.MB, 1, "Reporting.maxSampleEventSizeBytes")
 	eventSamplingEnabled := conf.GetReloadableBoolVar(false, "Reporting.eventSampling.enabled")
 	eventSamplingDuration := conf.GetReloadableDurationVar(60, time.Minute, "Reporting.eventSampling.durationInMinutes")
@@ -142,7 +140,6 @@ func NewDefaultReporter(ctx context.Context, conf *config.Config, log logger.Log
 		eventSamplingEnabled:                 eventSamplingEnabled,
 		eventSamplingDuration:                eventSamplingDuration,
 		eventSampler:                         eventSampler,
-		requestEntityTooLargeHandling:        requestEntityTooLargeHandling,
 		maxSampleEventSizeBytes:              maxSampleEventSizeBytes,
 		eventNamePrefixLength:                eventNamePrefixLength,
 		eventNameSuffixLength:                eventNameSuffixLength,
@@ -764,10 +761,6 @@ func (r *DefaultReporter) Stop() {
 }
 
 func (r *DefaultReporter) sendMetric(ctx context.Context, metric *types.Metric) error {
-	if !r.requestEntityTooLargeHandling.Load() {
-		return r.commonClient.SendWithoutFailFast(ctx, metric)
-	}
-
 	err := r.commonClient.Send(ctx, metric)
 	if !errors.Is(err, client.ErrPayloadTooLarge) {
 		return err

enterprise/reporting/utils_test.go

@@ -160,11 +160,10 @@ func newLargePayloadTestReporter(t *testing.T, commonClient *client.Client) (*De
 	statsStore, err := memstats.New()
 	require.NoError(t, err)
 	return &DefaultReporter{
-		commonClient:                  commonClient,
-		stats:                         statsStore,
-		log:                           logger.NOP,
-		requestEntityTooLargeHandling: config.SingleValueLoader(true),
-		maxSampleEventSizeBytes:       config.SingleValueLoader(maxSampleEventSizeBytesForTest),
+		commonClient:            commonClient,
+		stats:                   statsStore,
+		log:                     logger.NOP,
+		maxSampleEventSizeBytes: config.SingleValueLoader(maxSampleEventSizeBytesForTest),
 	}, statsStore
 }
 
@@ -174,12 +173,11 @@ func newLargePayloadTestEDReporter(t *testing.T, commonClient *client.Client) (*
 	statsStore, err := memstats.New()
 	require.NoError(t, err)
 	return &ErrorDetailReporter{
-		commonClient:                  commonClient,
-		stats:                         statsStore,
-		log:                           logger.NOP,
-		statsManager:                  NewErrorReportingStats(statsStore),
-		requestEntityTooLargeHandling: config.SingleValueLoader(true),
-		maxSampleEventSizeBytes:       config.SingleValueLoader(maxSampleEventSizeBytesForTest),
+		commonClient:            commonClient,
+		stats:                   statsStore,
+		log:                     logger.NOP,
+		statsManager:            NewErrorReportingStats(statsStore),
+		maxSampleEventSizeBytes: config.SingleValueLoader(maxSampleEventSizeBytesForTest),
 	}, statsStore
 }
 

warehouse/integrations/bigquery/middleware/middleware.go

@@ -22,6 +22,8 @@ type Client struct {
 	logger             loggerMW
 	keysAndValues      []any
 	slowQueryThreshold time.Duration
+	runQuery           func(context.Context, *bigquery.Query) (*bigquery.Job, error)
+	readQuery          func(context.Context, *bigquery.Query) (*bigquery.RowIterator, error)
 }
 
 func WithLogger(logger loggerMW) Opt {
@@ -53,6 +55,12 @@ func New(client *bigquery.Client, opts ...Opt) *Client {
 		Client:             client,
 		since:              time.Since,
 		slowQueryThreshold: 300 * time.Second,
+		runQuery: func(ctx context.Context, query *bigquery.Query) (*bigquery.Job, error) {
+			return query.Run(ctx)
+		},
+		readQuery: func(ctx context.Context, query *bigquery.Query) (*bigquery.RowIterator, error) {
+			return query.Read(ctx)
+		},
 	}
 	for _, opt := range opts {
 		opt(s)
@@ -62,14 +70,14 @@ func New(client *bigquery.Client, opts ...Opt) *Client {
 
 func (client *Client) Run(ctx context.Context, query *bigquery.Query) (*bigquery.Job, error) {
 	startedAt := time.Now()
-	job, err := query.Run(ctx)
+	job, err := client.runQuery(ctx, query)
 	client.logQuery(query, client.since(startedAt))
 	return job, err
 }
 
 func (client *Client) Read(ctx context.Context, query *bigquery.Query) (it *bigquery.RowIterator, err error) {
 	startedAt := time.Now()
-	it, err = query.Read(ctx)
+	it, err = client.readQuery(ctx, query)
 	client.logQuery(query, client.since(startedAt))
 	return it, err
 }

warehouse/integrations/bigquery/middleware/middleware_test.go

@@ -1,8 +1,7 @@
-package middleware_test
+package middleware
 
 import (
 	"context"
-	"os"
 	"testing"
 	"time"
 
@@ -13,29 +12,12 @@ import (
 
 	"github.com/rudderlabs/rudder-go-kit/logger/mock_logger"
 
-	"github.com/rudderlabs/rudder-server/warehouse/integrations/bigquery/middleware"
-	bqHelper "github.com/rudderlabs/rudder-server/warehouse/integrations/bigquery/testhelper"
 	"github.com/rudderlabs/rudder-server/warehouse/logfield"
 )
 
 func TestQueryWrapper(t *testing.T) {
-	if _, exists := os.LookupEnv(bqHelper.TestKey); !exists {
-		if os.Getenv("FORCE_RUN_INTEGRATION_TESTS") == "true" {
-			t.Fatalf("%s environment variable not set", bqHelper.TestKey)
-		}
-		t.Skipf("Skipping %s as %s is not set", t.Name(), bqHelper.TestKey)
-	}
-
-	bqTestCredentials, err := bqHelper.GetBQTestCredentials()
-	require.NoError(t, err)
-
 	ctx := context.Background()
-
-	db, err := bigquery.NewClient(
-		ctx,
-		bqTestCredentials.ProjectID,
-		option.WithAuthCredentialsJSON(option.ServiceAccount, []byte(bqTestCredentials.Credentials)),
-	)
+	db, err := bigquery.NewClient(ctx, "test-project", option.WithoutAuthentication())
 	require.NoError(t, err)
 
 	testCases := []struct {
@@ -67,14 +49,22 @@ func TestQueryWrapper(t *testing.T) {
 
 			mockLogger := mock_logger.NewMockLogger(mockCtrl)
 
-			qw := middleware.New(
-				db,
-				middleware.WithSlowQueryThreshold(queryThreshold),
-				middleware.WithLogger(mockLogger),
-				middleware.WithKeyAndValues(keysAndValues...),
-				middleware.WithSince(func(time.Time) time.Duration {
+			qw := New(
+				nil,
+				WithSlowQueryThreshold(queryThreshold),
+				WithLogger(mockLogger),
+				WithKeyAndValues(keysAndValues...),
+				WithSince(func(time.Time) time.Duration {
 					return tc.executionTime
 				}),
+				func(client *Client) {
+					client.runQuery = func(context.Context, *bigquery.Query) (*bigquery.Job, error) {
+						return &bigquery.Job{}, nil
+					}
+					client.readQuery = func(context.Context, *bigquery.Query) (*bigquery.RowIterator, error) {
+						return &bigquery.RowIterator{}, nil
+					}
+				},
 			)
 
 			queryStatement := "SELECT 1;"