feat: Remove IgnoreError and send sentry only on panics (#60)

We will be logging everything as log and moving log analysis
to external tools such as jq.

Also, for sentry we will be only looking for panics and not
for incorrect classification as we dont classify anything anymore

Author: yevgenypats

plugins/source.go

@@ -21,8 +21,6 @@ type SourcePlugin struct {
 	name string
 	// Version of the plugin
 	version string
-	// Classify error and return it's severity and type
-	ignoreError schema.IgnoreErrorFunc
 	// Called upon configure call to validate and init configuration
 	newExecutionClient SourceNewExecutionClientFunc
 	// Tables is all tables supported by this source plugin
@@ -49,12 +47,6 @@ func WithSourceLogger(logger zerolog.Logger) SourceOption {
 	}
 }
 
-func WithClassifyError(ignoreError schema.IgnoreErrorFunc) SourceOption {
-	return func(p *SourcePlugin) {
-		p.ignoreError = ignoreError
-	}
-}
-
 // Add internal columns
 func addInternalColumns(tables []*schema.Table) {
 	for _, table := range tables {
@@ -81,12 +73,6 @@ func NewSourcePlugin(name string, version string, tables []*schema.Table, newExe
 	if err := p.validate(); err != nil {
 		panic(err)
 	}
-	// if ignore error is not set on a table level set it with the plugin
-	for _, table := range p.tables {
-		if table.IgnoreError == nil && p.ignoreError != nil {
-			table.IgnoreError = p.ignoreError
-		}
-	}
 	return &p
 }
 
@@ -162,10 +148,6 @@ func (p *SourcePlugin) Sync(ctx context.Context, spec specs.Source, res chan<- *
 		if table.Multiplex != nil {
 			clients = table.Multiplex(c)
 		}
-		// because table can't import sourceplugin we need to set classifyError if it is not set by table
-		if table.IgnoreError == nil {
-			table.IgnoreError = p.ignoreError
-		}
 		// we call this here because we dont know when the following goroutine will be called and we do want an order
 		// of table by table
 		// totalClients := len(clients)

schema/table.go

@@ -24,9 +24,6 @@ type TableResolver func(ctx context.Context, meta ClientMeta, parent *Resource,
 
 type RowResolver func(ctx context.Context, meta ClientMeta, resource *Resource) error
 
-// Classify error and return it's severity and type
-type IgnoreErrorFunc func(err error) (bool, string)
-
 type Tables []*Table
 
 type Table struct {
@@ -40,8 +37,6 @@ type Table struct {
 	Relations Tables `json:"relations"`
 	// Resolver is the main entry point to fetching table data and
 	Resolver TableResolver `json:"-"`
-	// IgnoreError is a function that classifies error and returns it's severity and type
-	IgnoreError IgnoreErrorFunc `json:"-"`
 	// Multiplex returns re-purposed meta clients. The sdk will execute the table with each of them
 	Multiplex func(meta ClientMeta) []ClientMeta `json:"-"`
 	// PostResourceResolver is called after all columns have been resolved, but before the Resource is sent to be inserted. The ordering of resolvers is:
@@ -162,34 +157,23 @@ func (t Table) Resolve(ctx context.Context, meta ClientMeta, syncTime time.Time,
 	go func() {
 		defer func() {
 			if err := recover(); err != nil {
+				sentry.WithScope(func(scope *sentry.Scope) {
+					scope.SetTag("table", t.Name)
+					sentry.CurrentHub().Recover(err)
+				})
 				stack := string(debug.Stack())
 				meta.Logger().Error().Interface("error", err).Str("table_name", t.Name).TimeDiff("duration", time.Now(), startTime).Str("stack", stack).Msg("table resolver finished with panic")
 			}
 			close(res)
 		}()
 		meta.Logger().Debug().Str("table_name", t.Name).Msg("table resolver started")
 		if err := t.Resolver(ctx, meta, parent, res); err != nil {
-			if t.IgnoreError != nil {
-				if ignore, errType := t.IgnoreError(err); ignore {
-					meta.Logger().Debug().Stack().Str("table_name", t.Name).TimeDiff("duration", time.Now(), startTime).Str("error_type", errType).Err(err).Msg("table resolver finished with error")
-					return
-				}
-			}
-			sentry.WithScope(func(scope *sentry.Scope) {
-				scope.SetTag("table", t.Name)
-				scope.SetLevel(sentry.LevelError)
-				sentry.CaptureMessage(err.Error())
-			})
 			meta.Logger().Error().Str("table_name", t.Name).TimeDiff("duration", time.Now(), startTime).Err(err).Msg("table resolver finished with error")
 			return
 		}
 		meta.Logger().Debug().Str("table_name", t.Name).TimeDiff("duration", time.Now(), startTime).Msg("table resolver finished successfully")
 	}()
 	totalResources := 0
-	// we want to check for data integrity
-	// in the future we can do that as an optinoal feature via a flag
-	// pks := map[string]bool{}
-	// each result is an array of interface{}
 	for elem := range res {
 		objects := helpers.InterfaceSlice(elem)
 		if len(objects) == 0 {
@@ -214,11 +198,6 @@ func (t Table) Resolve(ctx context.Context, meta ClientMeta, syncTime time.Time,
 					meta.Logger().Trace().Str("table_name", t.Name).Msg("post resource resolver finished successfully")
 				}
 			}
-			// if pks[resource.PrimaryKeyValue()] {
-			// 	meta.Logger().Error().Str("table_name", t.Name).Str("primary_key", resource.PrimaryKeyValue()).Msg("duplicate primary key found")
-			// } else {
-			// 	pks[resource.PrimaryKeyValue()] = true
-			// }
 			resolvedResources <- resource
 			for _, rel := range t.Relations {
 				totalResources += rel.Resolve(ctx, meta, syncTime, resource, resolvedResources)

serve/serve.go

@@ -96,8 +96,10 @@ func newCmdServe(opts Options) *cobra.Command {
 
 			if !noSentry && version != "development" {
 				err = sentry.Init(sentry.ClientOptions{
-					Dsn:     opts.SentryDsn,
-					Release: opts.SourcePlugin.Version(),
+					Dsn:              opts.SentryDsn,
+					Debug:            false,
+					AttachStacktrace: true,
+					Release:          opts.SourcePlugin.Version(),
 					// https://docs.sentry.io/platforms/go/configuration/options/#removing-default-integrations
 					Integrations: func(integrations []sentry.Integration) []sentry.Integration {
 						var filteredIntegrations []sentry.Integration