feat: Add sentry for serve.Serve function (#54)

* feat: Add sentry for serve.Serve function

plugin developers will be able to direct their plugin errors
to their own projects on sentry as we dont want to monitor any
community errors anyway.

Also each plugin should go into it's own project on sentry
because we are using server side grouping which are mostly
project based.

Author: yevgenypats

go.mod

@@ -27,6 +27,7 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/getsentry/sentry-go v0.13.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect

go.sum

@@ -30,6 +30,8 @@ github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/frankban/quicktest v1.14.3 h1:FJKSZTDHjyhriyC81FLQ0LY93eSai0ZyR/ZIkd3ZUKE=
 github.com/frankban/quicktest v1.14.3/go.mod h1:mgiwOwqx65TmIk1wJ6Q7wvnVMocbUorkibMOrVTHZps=
+github.com/getsentry/sentry-go v0.13.0 h1:20dgTiUSfxRB/EhMPtxcL9ZEbM1ZdR+W/7f7NWD+xWo=
+github.com/getsentry/sentry-go v0.13.0/go.mod h1:EOsfu5ZdvKPfeHYV6pTVQnsjfp30+XA7//UooKNumH0=
 github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=

plugins/source.go

@@ -81,6 +81,12 @@ 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
 }
 

schema/table.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/cloudquery/plugin-sdk/helpers"
+	"github.com/getsentry/sentry-go"
 	"github.com/iancoleman/strcase"
 	"github.com/thoas/go-funk"
 )
@@ -178,10 +179,15 @@ func (t Table) Resolve(ctx context.Context, meta ClientMeta, syncTime time.Time,
 		if err := t.Resolver(ctx, meta, parent, res); err != nil {
 			if t.IgnoreError != nil {
 				if ignore, errType := t.IgnoreError(err); ignore {
-					meta.Logger().Debug().Str("table_name", t.Name).TimeDiff("duration", time.Now(), startTime).Str("error_type", errType).Msg("table resolver finished with error")
+					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
 		}
@@ -190,7 +196,7 @@ func (t Table) Resolve(ctx context.Context, meta ClientMeta, syncTime time.Time,
 	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{}
+	// pks := map[string]bool{}
 	// each result is an array of interface{}
 	for elem := range res {
 		objects := helpers.InterfaceSlice(elem)
@@ -211,16 +217,16 @@ func (t Table) Resolve(ctx context.Context, meta ClientMeta, syncTime time.Time,
 			if t.PostResourceResolver != nil {
 				meta.Logger().Trace().Str("table_name", t.Name).Msg("post resource resolver started")
 				if err := t.PostResourceResolver(ctx, meta, resource); err != nil {
-					meta.Logger().Error().Str("table_name", t.Name).Err(err).Msg("post resource resolver finished with error")
+					meta.Logger().Error().Str("table_name", t.Name).Stack().Err(err).Msg("post resource resolver finished with error")
 				} else {
 					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
-			}
+			// 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

@@ -5,10 +5,12 @@ import (
 	"net"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/cloudquery/plugin-sdk/internal/pb"
 	"github.com/cloudquery/plugin-sdk/internal/servers"
 	"github.com/cloudquery/plugin-sdk/plugins"
+	"github.com/getsentry/sentry-go"
 	grpczerolog "github.com/grpc-ecosystem/go-grpc-middleware/providers/zerolog/v2"
 	middleware "github.com/grpc-ecosystem/go-grpc-middleware/v2"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/logging"
@@ -23,13 +25,14 @@ type Options struct {
 	// Required: Source or destination plugin to serve.
 	SourcePlugin      *plugins.SourcePlugin
 	DestinationPlugin plugins.DestinationPlugin
+	SentryDsn         string
 }
 
-// bufSize used for unit testing grpc server and client
-const testBufSize = 1024 * 1024
-
 const (
 	serveShort = `Start plugin server`
+	// bufSize used for unit testing grpc server and client
+	testBufSize  = 1024 * 1024
+	flushTimeout = 5 * time.Second
 )
 
 // lis used for unit testing grpc server and client
@@ -38,6 +41,7 @@ var testListener *bufconn.Listener
 func newCmdServe(opts Options) *cobra.Command {
 	var address string
 	var network string
+	var noSentry bool
 	logLevel := newEnum([]string{"trace", "debug", "info", "warn", "error"}, "info")
 	logFormat := newEnum([]string{"text", "json"}, "text")
 	cmd := &cobra.Command{
@@ -56,6 +60,7 @@ func newCmdServe(opts Options) *cobra.Command {
 			} else {
 				logger = log.Output(zerolog.ConsoleWriter{Out: os.Stdout}).Level(zerologLevel)
 			}
+
 			// opts.Plugin.Logger = logger
 			var listener net.Listener
 			if network == "test" {
@@ -75,17 +80,39 @@ func newCmdServe(opts Options) *cobra.Command {
 				middleware.WithStreamServerChain(
 					logging.StreamServerInterceptor(grpczerolog.InterceptorLogger(logger)),
 				),
-				// grpc.ChainStreamInterceptor(grpc_zero),
-				// grpc.ChainUnaryInterceptor(),
 			)
 
+			version := "development"
 			if opts.SourcePlugin != nil {
 				opts.SourcePlugin.SetLogger(logger)
 				pb.RegisterSourceServer(s, &servers.SourceServer{Plugin: opts.SourcePlugin})
+				version = opts.SourcePlugin.Version()
 			}
 			if opts.DestinationPlugin != nil {
 				// opts.DestinationPlugin.Logger = logger
 				pb.RegisterDestinationServer(s, &servers.DestinationServer{Plugin: opts.DestinationPlugin})
+				version = opts.DestinationPlugin.Version()
+			}
+
+			if !noSentry && version != "development" {
+				err = sentry.Init(sentry.ClientOptions{
+					Dsn:     opts.SentryDsn,
+					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
+						for _, integration := range integrations {
+							if integration.Name() == "Modules" {
+								continue
+							}
+							filteredIntegrations = append(filteredIntegrations, integration)
+						}
+						return filteredIntegrations
+					},
+				})
+				if err != nil {
+					log.Error().Err(err).Msg("Error initializing sentry")
+				}
 			}
 
 			logger.Info().Str("address", listener.Addr().String()).Msg("server listening")
@@ -99,6 +126,8 @@ func newCmdServe(opts Options) *cobra.Command {
 	cmd.Flags().StringVar(&network, "network", "tcp", `the network must be "tcp", "tcp4", "tcp6", "unix" or "unixpacket"`)
 	cmd.Flags().Var(logLevel, "log-level", fmt.Sprintf("log level. one of: %s", strings.Join(logLevel.Allowed, ",")))
 	cmd.Flags().Var(logFormat, "log-format", fmt.Sprintf("log format. one of: %s", strings.Join(logFormat.Allowed, ",")))
+	cmd.Flags().BoolVar(&noSentry, "no-sentry", false, "disable sentry")
+
 	return cmd
 }
 
@@ -113,7 +142,10 @@ func newCmdRoot(opts Options) *cobra.Command {
 
 func Serve(opts Options) {
 	if err := newCmdRoot(opts).Execute(); err != nil {
+		sentry.CaptureMessage(err.Error())
+		sentry.Flush(flushTimeout)
 		fmt.Println(err)
 		os.Exit(1)
 	}
+	sentry.Flush(flushTimeout)
 }