chore: stop collecting errors in proc_error jobsdb (#6232)

# Description

We are no longer be collecting errors in `proc_error` jobsdb as we are
planning to remove error collection for debugging purposes in the near
future.

What was happening until now:

- rudder-server was collecting failed jobs from webhook, processor
steps, router, and batchrouter, and was storing these failed jobs in the
`proc_error` JobsDB.
- periodically, the `stash` processor component was creating backup
files from the `proc_error` JobsDB and uploading them to
`rudder-proc-err-logs` in object storage.

In the past, we relied on `proc_error` jobs for replays, but that’s no
longer the case. At this point, it feels like we’re spending unnecessary
resources collecting and storing all this information which no-one
actually needs, using as an excuse that it is useful for debugging
purposes…

As a first step we are stopping collection and storage of failed jobs to
`proc_error`. In a follow up step we will be removing all relevant
`proc_error` code altogether.

## Linear Ticket

resolves PIPE-2295

## Security

- [x] The code changed/added as part of this pull request won't create
any security issues with how the software is being used.

Author: atzoum

app/apphandlers/embeddedAppHandler.go

@@ -200,27 +200,27 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 	defer batchRouterDB.Close()
 
 	// We need two errorDBs, one in read & one in write mode to support separate gateway to store failures
-	errDBForRead := jobsdb.NewForRead(
+	errorDBForRead := jobsdb.NewForRead(
 		"proc_error",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithDSLimit(a.config.procErrorDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", false)),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(dbPool),
 	)
-	defer errDBForRead.Close()
-	errDBForWrite := jobsdb.NewForWrite(
+	defer errorDBForRead.Close()
+	errorDBForWrite := jobsdb.NewForWrite(
 		"proc_error",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", true)),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(dbPool),
 	)
-	defer errDBForWrite.Close()
-	if err = errDBForWrite.Start(); err != nil {
-		return fmt.Errorf("could not start errDBForWrite: %w", err)
+	defer errorDBForWrite.Close()
+	if err = errorDBForWrite.Start(); err != nil {
+		return fmt.Errorf("could not start errorDBForWrite: %w", err)
 	}
-	defer errDBForWrite.Stop()
+	defer errorDBForWrite.Stop()
 
 	schemaDB := jobsdb.NewForReadWrite(
 		"esch",
@@ -281,8 +281,8 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		gwDBForProcessor,
 		routerDB,
 		batchRouterDB,
-		errDBForRead,
-		errDBForWrite,
+		errorDBForRead,
+		errorDBForWrite,
 		schemaDB,
 		archivalDB,
 		reporting,
@@ -310,7 +310,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 			config.GetReloadableDurationVar(1, time.Second, "JobsDB.rt.parameterValuesCacheTtl", "JobsDB.parameterValuesCacheTtl"),
 			routerDB,
 		),
-		ProcErrorDB:                errDBForWrite,
+		ProcErrorDB:                errorDBForWrite,
 		TransientSources:           transientSources,
 		RsourcesService:            rsourcesService,
 		TransformerFeaturesService: transformerFeaturesService,
@@ -326,7 +326,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 			config.GetReloadableDurationVar(1, time.Second, "JobsDB.rt.parameterValuesCacheTtl", "JobsDB.parameterValuesCacheTtl"),
 			batchRouterDB,
 		),
-		ProcErrorDB:           errDBForWrite,
+		ProcErrorDB:           errorDBForWrite,
 		TransientSources:      transientSources,
 		RsourcesService:       rsourcesService,
 		Debugger:              destinationHandle,
@@ -340,7 +340,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		GatewayDB:       gwDBForProcessor,
 		RouterDB:        routerDB,
 		BatchRouterDB:   batchRouterDB,
-		ErrorDB:         errDBForRead,
+		ErrorDB:         errorDBForRead,
 		EventSchemaDB:   schemaDB,
 		ArchivalDB:      archivalDB,
 		Processor:       proc,
@@ -373,7 +373,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 	streamMsgValidator := stream.NewMessageValidator()
 	gw := gateway.Handle{}
 	err = gw.Setup(ctx, config, logger.NewLogger().Child("gateway"), statsFactory, a.app, backendconfig.DefaultBackendConfig,
-		gatewayDB, errDBForWrite, rateLimiter, a.versionHandler, rsourcesService, transformerFeaturesService, sourceHandle,
+		gatewayDB, errorDBForWrite, rateLimiter, a.versionHandler, rsourcesService, transformerFeaturesService, sourceHandle,
 		streamMsgValidator, gateway.WithInternalHttpHandlers(
 			map[string]http.Handler{
 				"/drain": drainConfigManager.DrainConfigHttpHandler(),

app/apphandlers/gatewayAppHandler.go

@@ -90,19 +90,19 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, options *app.Options)
 	}
 	defer gatewayDB.Stop()
 
-	errDB := jobsdb.NewForWrite(
+	errorDB := jobsdb.NewForWrite(
 		"proc_error",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Gateway.jobsDB.skipMaintenanceError", true)),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(dbPool),
 	)
-	defer errDB.Close()
+	defer errorDB.Close()
 
-	if err := errDB.Start(); err != nil {
-		return fmt.Errorf("could not start errDB: %w", err)
+	if err := errorDB.Start(); err != nil {
+		return fmt.Errorf("could not start errorDB: %w", err)
 	}
-	defer errDB.Stop()
+	defer errorDB.Stop()
 
 	g, ctx := errgroup.WithContext(ctx)
 
@@ -145,7 +145,7 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, options *app.Options)
 	}
 	streamMsgValidator := stream.NewMessageValidator()
 	err = gw.Setup(ctx, config, logger.NewLogger().Child("gateway"), statsFactory, a.app, backendconfig.DefaultBackendConfig,
-		gatewayDB, errDB, rateLimiter, a.versionHandler, rsourcesService, transformerFeaturesService, sourceHandle,
+		gatewayDB, errorDB, rateLimiter, a.versionHandler, rsourcesService, transformerFeaturesService, sourceHandle,
 		streamMsgValidator, gateway.WithInternalHttpHandlers(
 			map[string]http.Handler{
 				"/drain": drainConfigHttpHandler,

app/apphandlers/processorAppHandler.go

@@ -186,27 +186,27 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		jobsdb.WithDBHandle(dbPool),
 	)
 	defer batchRouterDB.Close()
-	errDBForRead := jobsdb.NewForRead(
+	errorDBForRead := jobsdb.NewForRead(
 		"proc_error",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithDSLimit(a.config.procErrorDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", false)),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(dbPool),
 	)
-	defer errDBForRead.Close()
-	errDBForWrite := jobsdb.NewForWrite(
+	defer errorDBForRead.Close()
+	errorDBForWrite := jobsdb.NewForWrite(
 		"proc_error",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", true)),
 		jobsdb.WithStats(statsFactory),
 		jobsdb.WithDBHandle(dbPool),
 	)
-	errDBForWrite.Close()
-	if err = errDBForWrite.Start(); err != nil {
-		return fmt.Errorf("could not start errDBForWrite: %w", err)
+	errorDBForWrite.Close()
+	if err = errorDBForWrite.Start(); err != nil {
+		return fmt.Errorf("could not start errorDBForWrite: %w", err)
 	}
-	defer errDBForWrite.Stop()
+	defer errorDBForWrite.Stop()
 	schemaDB := jobsdb.NewForReadWrite(
 		"esch",
 		jobsdb.WithClearDB(options.ClearDB),
@@ -277,8 +277,8 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		gwDBForProcessor,
 		routerDB,
 		batchRouterDB,
-		errDBForRead,
-		errDBForWrite,
+		errorDBForRead,
+		errorDBForWrite,
 		schemaDB,
 		archivalDB,
 		reporting,
@@ -306,7 +306,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 			config.GetReloadableDurationVar(1, time.Second, "JobsDB.rt.parameterValuesCacheTtl", "JobsDB.parameterValuesCacheTtl"),
 			routerDB,
 		),
-		ProcErrorDB:                errDBForWrite,
+		ProcErrorDB:                errorDBForWrite,
 		TransientSources:           transientSources,
 		RsourcesService:            rsourcesService,
 		TransformerFeaturesService: transformerFeaturesService,
@@ -322,7 +322,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 			config.GetReloadableDurationVar(1, time.Second, "JobsDB.rt.parameterValuesCacheTtl", "JobsDB.parameterValuesCacheTtl"),
 			batchRouterDB,
 		),
-		ProcErrorDB:           errDBForWrite,
+		ProcErrorDB:           errorDBForWrite,
 		TransientSources:      transientSources,
 		RsourcesService:       rsourcesService,
 		Debugger:              destinationHandle,
@@ -337,7 +337,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		GatewayDB:        gwDBForProcessor,
 		RouterDB:         routerDB,
 		BatchRouterDB:    batchRouterDB,
-		ErrorDB:          errDBForRead,
+		ErrorDB:          errorDBForRead,
 		SchemaForwarder:  schemaForwarder,
 		EventSchemaDB:    schemaDB,
 		ArchivalDB:       archivalDB,

app/cluster/integration_test.go

@@ -114,11 +114,11 @@ func TestDynamicClusterManager(t *testing.T) {
 
 	archDB := jobsdb.NewForReadWrite("archival", jobsdb.WithStats(stats.NOP))
 	defer archDB.TearDown()
-	readErrDB := jobsdb.NewForRead("proc_error", jobsdb.WithStats(stats.NOP))
-	defer readErrDB.TearDown()
-	writeErrDB := jobsdb.NewForWrite("proc_error", jobsdb.WithStats(stats.NOP))
-	require.NoError(t, writeErrDB.Start())
-	defer writeErrDB.TearDown()
+	readErrorDB := jobsdb.NewForRead("proc_error", jobsdb.WithStats(stats.NOP))
+	defer readErrorDB.TearDown()
+	writeErrorDB := jobsdb.NewForWrite("proc_error", jobsdb.WithStats(stats.NOP))
+	require.NoError(t, writeErrorDB.Start())
+	defer writeErrorDB.TearDown()
 
 	clearDb := false
 	ctx := context.Background()
@@ -131,8 +131,8 @@ func TestDynamicClusterManager(t *testing.T) {
 		gwDB,
 		rtDB,
 		brtDB,
-		readErrDB,
-		writeErrDB,
+		readErrorDB,
+		writeErrorDB,
 		eschDB,
 		archDB,
 		&reporting.NOOP{},
@@ -155,7 +155,7 @@ func TestDynamicClusterManager(t *testing.T) {
 		Reporting:                  &reporting.NOOP{},
 		BackendConfig:              mockBackendConfig,
 		RouterDB:                   rtDB,
-		ProcErrorDB:                readErrDB,
+		ProcErrorDB:                readErrorDB,
 		TransientSources:           transientsource.NewEmptyService(),
 		RsourcesService:            mockRsourcesService,
 		TransformerFeaturesService: transformer.NewNoOpService(),
@@ -165,7 +165,7 @@ func TestDynamicClusterManager(t *testing.T) {
 		Reporting:        &reporting.NOOP{},
 		BackendConfig:    mockBackendConfig,
 		RouterDB:         brtDB,
-		ProcErrorDB:      readErrDB,
+		ProcErrorDB:      readErrorDB,
 		TransientSources: transientsource.NewEmptyService(),
 		RsourcesService:  mockRsourcesService,
 	}
@@ -192,7 +192,7 @@ func TestDynamicClusterManager(t *testing.T) {
 		GatewayDB:       gwDB,
 		RouterDB:        rtDB,
 		BatchRouterDB:   brtDB,
-		ErrorDB:         readErrDB,
+		ErrorDB:         readErrorDB,
 		EventSchemaDB:   eschDB,
 		ArchivalDB:      archDB,
 		SchemaForwarder: schemaForwarder,

gateway/gateway_test.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/base64"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -49,7 +48,6 @@ import (
 	"github.com/rudderlabs/rudder-server/enterprise/suppress-user/model"
 	gwstats "github.com/rudderlabs/rudder-server/gateway/internal/stats"
 	"github.com/rudderlabs/rudder-server/gateway/response"
-	webhookModel "github.com/rudderlabs/rudder-server/gateway/webhook/model"
 	"github.com/rudderlabs/rudder-server/jobsdb"
 	mocksApp "github.com/rudderlabs/rudder-server/mocks/app"
 	mocksBackendConfig "github.com/rudderlabs/rudder-server/mocks/backend-config"
@@ -1663,59 +1661,6 @@ var _ = Describe("Gateway", func() {
 			err := gateway.Shutdown()
 			Expect(err).To(BeNil())
 		})
-
-		It("should save failures to error db", func() {
-			c.mockErrJobsDB.
-				EXPECT().Store(
-				gomock.Any(),
-				gomock.Any(),
-			).
-				DoAndReturn(
-					func(
-						ctx context.Context,
-						jobs []*jobsdb.JobT,
-					) error {
-						for idx, job := range jobs {
-							Expect(misc.IsValidUUID(job.UUID.String())).To(BeTrue())
-							Expect(job.CustomVal).To(Equal("WEBHOOK"))
-
-							var paramsMap, expectedParamsMap map[string]interface{}
-							var expectedStr []byte
-
-							switch idx {
-							case 0:
-								Expect(job.EventPayload).To(Equal(json.RawMessage(`{"a1": "b1"}`)))
-								expectedStr = []byte(fmt.Sprintf(`{"source_id": "%v", "stage": "webhook", "source_type": "cio", "reason": "err1"}`, SourceIDEnabled))
-							case 1:
-								Expect(job.EventPayload).To(Equal(json.RawMessage(`{"a2": "b2"}`)))
-								expectedStr = []byte(fmt.Sprintf(`{"source_id": "%v", "stage": "webhook", "source_type": "af", "reason": "err2"}`, SourceIDEnabled))
-							}
-
-							_ = jsonrs.Unmarshal(job.Parameters, &paramsMap)
-							_ = jsonrs.Unmarshal(expectedStr, &expectedParamsMap)
-							equals := reflect.DeepEqual(paramsMap, expectedParamsMap)
-							Expect(equals).To(BeTrue())
-						}
-						return nil
-					}).
-				Times(1)
-
-			reqs := make([]*webhookModel.FailedWebhookPayload, 2)
-			reqs[0] = &webhookModel.FailedWebhookPayload{
-				RequestContext: rCtxEnabled,
-				Payload:        []byte(`{"a1": "b1"}`),
-				SourceType:     "cio",
-				Reason:         "err1",
-			}
-			reqs[1] = &webhookModel.FailedWebhookPayload{
-				RequestContext: rCtxEnabled,
-				Payload:        []byte(`{"a2": "b2"}`),
-				SourceType:     "af",
-				Reason:         "err2",
-			}
-			err := gateway.SaveWebhookFailures(reqs)
-			Expect(err).To(BeNil())
-		})
 	})
 
 	Context("Internal Batch", func() {

gateway/handle.go

@@ -62,7 +62,7 @@ type Handle struct {
 	application     app.App
 	backendConfig   backendconfig.BackendConfig
 	jobsDB          jobsdb.JobsDB
-	errDB           jobsdb.JobsDB
+	errorDB         jobsdb.JobsDB
 	rateLimiter     throttler.Throttler
 	versionHandler  func(w http.ResponseWriter, r *http.Request)
 	rsourcesService rsources.JobService
@@ -132,6 +132,7 @@ type Handle struct {
 		enableInternalBatchEnrichment        config.ValueLoader[bool]
 		enableEventBlocking                  config.ValueLoader[bool]
 		webhookV2HandlerEnabled              bool
+		errorDBEnabled                       config.ValueLoader[bool]
 	}
 
 	// additional internal http handlers

gateway/handle_lifecycle.go

@@ -70,7 +70,7 @@ This function will block until backend config is initially received.
 func (gw *Handle) Setup(
 	ctx context.Context,
 	config *config.Config, logger logger.Logger, stat stats.Stats,
-	application app.App, backendConfig backendconfig.BackendConfig, jobsDB, errDB jobsdb.JobsDB,
+	application app.App, backendConfig backendconfig.BackendConfig, jobsDB, errorDB jobsdb.JobsDB,
 	rateLimiter throttler.Throttler, versionHandler func(w http.ResponseWriter, r *http.Request),
 	rsourcesService rsources.JobService, transformerFeaturesService transformer.FeaturesService,
 	sourcehandle sourcedebugger.SourceDebugger, streamMsgValidator func(message *stream.Message) error,
@@ -83,7 +83,7 @@ func (gw *Handle) Setup(
 	gw.application = application
 	gw.backendConfig = backendConfig
 	gw.jobsDB = jobsDB
-	gw.errDB = errDB
+	gw.errorDB = errorDB
 	gw.rateLimiter = rateLimiter
 	gw.versionHandler = versionHandler
 	gw.rsourcesService = rsourcesService
@@ -129,6 +129,7 @@ func (gw *Handle) Setup(
 	gw.conf.enableInternalBatchEnrichment = config.GetReloadableBoolVar(true, "gateway.enableBatchEnrichment")
 	// enable webhook v2 handler. disabled by default
 	gw.conf.webhookV2HandlerEnabled = config.GetBoolVar(false, "Gateway.webhookV2HandlerEnabled")
+	gw.conf.errorDBEnabled = config.GetReloadableBoolVar(false, "ErrorDB.enabled")
 	// enable event blocking. false by default
 	gw.conf.enableEventBlocking = config.GetReloadableBoolVar(false, "enableEventBlocking")
 	// Registering stats

gateway/handle_test.go

@@ -92,6 +92,7 @@ func createTestGateway(t *testing.T, enableEventBlocking bool, eventBlockingSett
 			enableInternalBatchEnrichment                                                     config.ValueLoader[bool]
 			enableEventBlocking                                                               config.ValueLoader[bool]
 			webhookV2HandlerEnabled                                                           bool
+			errorDBEnabled                                                                    config.ValueLoader[bool]
 		}{
 			enableEventBlocking:           config.SingleValueLoader(enableEventBlocking),
 			enableInternalBatchValidator:  config.SingleValueLoader(false),

gateway/handle_webhook.go

@@ -30,6 +30,10 @@ func (gw *Handle) ProcessTransformedWebhookRequest(w *http.ResponseWriter, r *ht
 
 // SaveWebhookFailures saves errors to the error db
 func (gw *Handle) SaveWebhookFailures(reqs []*model.FailedWebhookPayload) error {
+	if !gw.conf.errorDBEnabled.Load() {
+		return nil
+	}
+
 	jobs := make([]*jobsdb.JobT, 0, len(reqs))
 	for _, req := range reqs {
 		params := map[string]any{
@@ -63,5 +67,5 @@ func (gw *Handle) SaveWebhookFailures(reqs []*model.FailedWebhookPayload) error
 
 	ctx, cancel := context.WithTimeout(context.Background(), gw.conf.WriteTimeout)
 	defer cancel()
-	return gw.errDB.Store(ctx, jobs)
+	return gw.errorDB.Store(ctx, jobs)
 }