feat: Improve gRPC status codes and remove .cq file suffix (#30)

Author: yevgenypats

internal/pb/base.pb.go

@@ -31,7 +31,5 @@ message Write {
     bytes resource = 1;
   }
   message Response {
-    // error
-    string error = 1;
   }
 }

internal/pb/destination.pb.go

@@ -10,7 +10,8 @@ import (
 	"github.com/cloudquery/plugin-sdk/plugins"
 	"github.com/cloudquery/plugin-sdk/schema"
 	"github.com/cloudquery/plugin-sdk/specs"
-	"github.com/pkg/errors"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 type SourceServer struct {
@@ -21,7 +22,7 @@ type SourceServer struct {
 func (s *SourceServer) GetTables(context.Context, *pb.GetTables_Request) (*pb.GetTables_Response, error) {
 	b, err := json.Marshal(s.Plugin.Tables())
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to marshal tables")
+		return nil, fmt.Errorf("failed to marshal tables: %w", err)
 	}
 	return &pb.GetTables_Response{
 		Tables: b,
@@ -41,36 +42,36 @@ func (s *SourceServer) GetExampleConfig(context.Context, *pb.GetExampleConfig_Re
 
 func (s *SourceServer) Sync(req *pb.Sync_Request, stream pb.Source_SyncServer) error {
 	resources := make(chan *schema.Resource)
-	var fetchErr error
+	var syncErr error
 
 	var spec specs.Source
 	dec := json.NewDecoder(bytes.NewReader(req.Spec))
 	dec.UseNumber()
 	dec.DisallowUnknownFields()
 	if err := dec.Decode(&spec); err != nil {
-		return fmt.Errorf("failed to decode source spec: %w", err)
+		return status.Errorf(codes.InvalidArgument, "failed to decode spec: %v", err)
 	}
 
 	go func() {
 		defer close(resources)
 		if err := s.Plugin.Sync(stream.Context(), spec, resources); err != nil {
-			fetchErr = errors.Wrap(err, "failed to fetch resources")
+			syncErr = fmt.Errorf("failed to sync resources: %w", err)
 		}
 	}()
 
 	for resource := range resources {
 		b, err := json.Marshal(resource)
 		if err != nil {
-			return errors.Wrap(err, "failed to marshal resource")
+			return status.Errorf(codes.Internal, "failed to marshal resource: %v", err)
 		}
 		if err := stream.Send(&pb.Sync_Response{
 			Resource: b,
 		}); err != nil {
-			return errors.Wrap(err, "failed to send resource")
+			return status.Errorf(codes.Internal, "failed to send resource: %v", err)
 		}
 	}
-	if fetchErr != nil {
-		return fetchErr
+	if syncErr != nil {
+		return syncErr
 	}
 
 	return nil

internal/pb/destination.proto

@@ -12,7 +12,7 @@ import (
 	"github.com/thoas/go-funk"
 )
 
-// TableResolver is the main entry point when a table fetch is called.
+// TableResolver is the main entry point when a table is sync is called.
 //
 // Table resolver has 3 main arguments:
 // - meta(ClientMeta): is the client returned by the plugin.Provider Configure call
@@ -163,7 +163,7 @@ func (t Table) TableNames() []string {
 }
 
 // Call the table resolver with with all of it's relation for every reolved resource
-func (t Table) Resolve(ctx context.Context, meta ClientMeta, fetchTime time.Time, parent *Resource, resolvedResources chan<- *Resource) int {
+func (t Table) Resolve(ctx context.Context, meta ClientMeta, syncTime time.Time, parent *Resource, resolvedResources chan<- *Resource) int {
 	res := make(chan interface{})
 	startTime := time.Now()
 	go func() {
@@ -199,7 +199,7 @@ func (t Table) Resolve(ctx context.Context, meta ClientMeta, fetchTime time.Time
 		}
 		totalResources += len(objects)
 		for i := range objects {
-			resource := NewResourceData(&t, parent, fetchTime, objects[i])
+			resource := NewResourceData(&t, parent, syncTime, objects[i])
 			t.resolveColumns(ctx, meta, resource)
 			if t.PostResourceResolver != nil {
 				meta.Logger().Trace().Str("table_name", t.Name).Msg("post resource resolver started")
@@ -216,7 +216,7 @@ func (t Table) Resolve(ctx context.Context, meta ClientMeta, fetchTime time.Time
 			}
 			resolvedResources <- resource
 			for _, rel := range t.Relations {
-				totalResources += rel.Resolve(ctx, meta, fetchTime, resource, resolvedResources)
+				totalResources += rel.Resolve(ctx, meta, syncTime, resource, resolvedResources)
 			}
 		}
 	}

internal/pb/destination_grpc.pb.go

@@ -155,7 +155,7 @@ spec:
 		}
 	}
 	if err := wg.Wait(); err != nil {
-		t.Fatalf("Failed to fetch resources: %v", err)
+		t.Fatalf("Failed to sync resources: %v", err)
 	}
 
 	exampleConfig, err := c.ExampleConfig(ctx)

internal/pb/source.pb.go

@@ -24,7 +24,7 @@ func NewSpecReader(directory string) (*SpecReader, error) {
 	}
 
 	for _, file := range files {
-		if !file.IsDir() && strings.HasSuffix(file.Name(), ".cq.yml") {
+		if !file.IsDir() && strings.HasSuffix(file.Name(), ".yml") {
 			data, err := os.ReadFile(filepath.Join(directory, file.Name()))
 			if err != nil {
 				return nil, fmt.Errorf("failed to read file %s: %w", file.Name(), err)