feat: CloudQuery v2

.gitignore

@@ -6,3 +6,4 @@ config.hcl
 
 .vscode
 vendor
+cover.out

Makefile

@@ -6,6 +6,6 @@ test:
 lint:
 	golangci-lint run
 
-.PHONY: generate-protobuf
-generate-protobuf:
+.PHONY: gen-proto
+gen-proto:
 	protoc --proto_path=. --go_out . --go_opt=paths=source_relative --go-grpc_out=. --go-grpc_opt=paths=source_relative internal/pb/base.proto internal/pb/source.proto internal/pb/destination.proto

clients/destination.go

@@ -2,15 +2,14 @@ package clients
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
 	"github.com/cloudquery/plugin-sdk/internal/pb"
 	"github.com/cloudquery/plugin-sdk/plugins"
 	"github.com/cloudquery/plugin-sdk/schema"
 	"github.com/cloudquery/plugin-sdk/specs"
-	"github.com/vmihailenco/msgpack/v5"
 	"google.golang.org/grpc"
-	"gopkg.in/yaml.v3"
 )
 
 type DestinationClient struct {
@@ -31,23 +30,9 @@ func NewLocalDestinationClient(p plugins.DestinationPlugin) *DestinationClient {
 	}
 }
 
-func (c *DestinationClient) Configure(ctx context.Context, s specs.DestinationSpec) error {
-	if c.localClient != nil {
-		return c.localClient.Configure(ctx, s)
-	}
-	b, err := yaml.Marshal(s)
-	if err != nil {
-		return fmt.Errorf("failed to marshal spec: %w", err)
-	}
-	if _, err := c.pbClient.Configure(ctx, &pb.Configure_Request{Config: b}); err != nil {
-		return err
-	}
-	return nil
-}
-
 func (c *DestinationClient) GetExampleConfig(ctx context.Context) (string, error) {
 	if c.localClient != nil {
-		return c.localClient.GetExampleConfig(ctx), nil
+		return c.localClient.ExampleConfig(), nil
 	}
 	res, err := c.pbClient.GetExampleConfig(ctx, &pb.GetExampleConfig_Request{})
 	if err != nil {
@@ -56,42 +41,52 @@ func (c *DestinationClient) GetExampleConfig(ctx context.Context) (string, error
 	return res.Config, nil
 }
 
-func (c *DestinationClient) Save(ctx context.Context, msg *FetchResultMessage) error {
-	var saveClient pb.Destination_SaveClient
-	var err error
-	if c.pbClient != nil {
-		saveClient, err = c.pbClient.Save(ctx)
-		if err != nil {
-			return fmt.Errorf("failed to create save client: %w", err)
-		}
-	}
+func (c *DestinationClient) Initialize(ctx context.Context, spec specs.Destination) error {
 	if c.localClient != nil {
-		var resource schema.Resource
-		if err := msgpack.Unmarshal(msg.Resource, &resource); err != nil {
-			return fmt.Errorf("failed to unmarshal resources: %w", err)
-		}
-		if err := c.localClient.Save(ctx, []*schema.Resource{&resource}); err != nil {
-			return fmt.Errorf("failed to save resources: %w", err)
-		}
-	} else {
-		if err := saveClient.Send(&pb.Save_Request{Resources: msg.Resource}); err != nil {
-			return err
-		}
+		return c.localClient.Initialize(ctx, spec)
+	}
+	b, err := json.Marshal(spec)
+	if err != nil {
+		return fmt.Errorf("destination configure: failed to marshal spec: %w", err)
+	}
+	_, err = c.pbClient.Configure(ctx, &pb.Configure_Request{
+		Config: b,
+	})
+	if err != nil {
+		return fmt.Errorf("destination configure: failed to configure: %w", err)
 	}
-
 	return nil
 }
 
-func (c *DestinationClient) CreateTables(ctx context.Context, tables []*schema.Table) error {
+func (c *DestinationClient) Migrate(ctx context.Context, tables []*schema.Table) error {
 	if c.localClient != nil {
-		return c.localClient.CreateTables(ctx, tables)
+		return c.localClient.Migrate(ctx, tables)
+	}
+	b, err := json.Marshal(tables)
+	if err != nil {
+		return fmt.Errorf("destination migrate: failed to marshal plugin: %w", err)
 	}
-	b, err := yaml.Marshal(tables)
+	_, err = c.pbClient.Migrate(ctx, &pb.Migrate_Request{Tables: b})
 	if err != nil {
-		return fmt.Errorf("failed to marshal tables: %w", err)
+		return fmt.Errorf("destination migrate: failed to migrate: %w", err)
 	}
-	if _, err := c.pbClient.CreateTables(ctx, &pb.CreateTables_Request{Tables: b}); err != nil {
-		return err
+	return nil
+}
+
+func (c *DestinationClient) Write(ctx context.Context, table string, data map[string]interface{}) error {
+	// var saveClient pb.Destination_SaveClient
+	// var err error
+	// if c.pbClient != nil {
+	// 	saveClient, err = c.pbClient.Write(ctx)
+	// 	if err != nil {
+	// 		return fmt.Errorf("failed to create save client: %w", err)
+	// 	}
+	// }
+	if c.localClient != nil {
+		if err := c.localClient.Write(ctx, table, data); err != nil {
+			return fmt.Errorf("failed to save resources: %w", err)
+		}
 	}
+
 	return nil
 }

clients/source.go

@@ -3,20 +3,15 @@
 package clients
 
 import (
-	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
-	"text/template"
 
 	"github.com/cloudquery/plugin-sdk/internal/pb"
 	"github.com/cloudquery/plugin-sdk/schema"
 	"github.com/cloudquery/plugin-sdk/specs"
-	"github.com/pkg/errors"
-	"github.com/vmihailenco/msgpack/v5"
-	"github.com/xeipuuv/gojsonschema"
 	"google.golang.org/grpc"
-	"gopkg.in/yaml.v3"
 )
 
 type SourceClient struct {
@@ -27,14 +22,6 @@ type FetchResultMessage struct {
 	Resource []byte
 }
 
-const sourcePluginExampleConfigTemplate = `kind: source
-spec:
-  name: {{.Name}}
-  version: {{.Version}}
-  configuration:
-  {{.PluginExampleConfig | indent 4}}
-`
-
 func NewSourceClient(cc grpc.ClientConnInterface) *SourceClient {
 	return &SourceClient{
 		pbClient: pb.NewSourceClient(cc),
@@ -47,52 +34,30 @@ func (c *SourceClient) GetTables(ctx context.Context) ([]*schema.Table, error) {
 		return nil, err
 	}
 	var tables []*schema.Table
-	if err := msgpack.Unmarshal(res.Tables, &tables); err != nil {
+	if err := json.Unmarshal(res.Tables, &tables); err != nil {
 		return nil, err
 	}
 	return tables, nil
 }
 
-func (c *SourceClient) Configure(ctx context.Context, spec specs.SourceSpec) (*gojsonschema.Result, error) {
-	b, err := yaml.Marshal(spec)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to marshal source spec")
-	}
-	res, err := c.pbClient.Configure(ctx, &pb.Configure_Request{Config: b})
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to configure source")
-	}
-	var validationResult gojsonschema.Result
-	if err := msgpack.Unmarshal(res.JsonschemaResult, &validationResult); err != nil {
-		return nil, errors.Wrap(err, "failed to unmarshal validation result")
-	}
-	return &validationResult, nil
-}
-
-func (c *SourceClient) GetExampleConfig(ctx context.Context) (string, error) {
+func (c *SourceClient) ExampleConfig(ctx context.Context) (string, error) {
 	res, err := c.pbClient.GetExampleConfig(ctx, &pb.GetExampleConfig_Request{})
 	if err != nil {
 		return "", fmt.Errorf("failed to get example config: %w", err)
 	}
-	t, err := template.New("source_plugin").Funcs(templateFuncMap()).Parse(sourcePluginExampleConfigTemplate)
-	if err != nil {
-		return "", fmt.Errorf("failed to parse template: %w", err)
-	}
-	var tpl bytes.Buffer
-	if err := t.Execute(&tpl, map[string]interface{}{
-		"Name":                res.Name,
-		"Version":             res.Version,
-		"PluginExampleConfig": res.Config,
-	}); err != nil {
-		return "", fmt.Errorf("failed to generate example config: %w", err)
-	}
-	return tpl.String(), nil
+	return res.Config, nil
 }
 
-func (c *SourceClient) Fetch(ctx context.Context, spec specs.SourceSpec, res chan<- *FetchResultMessage) error {
-	stream, err := c.pbClient.Fetch(ctx, &pb.Fetch_Request{})
+func (c *SourceClient) Sync(ctx context.Context, spec specs.Source, res chan<- *schema.Resource) error {
+	b, err := json.Marshal(spec)
 	if err != nil {
-		return fmt.Errorf("failed to fetch resources: %w", err)
+		return fmt.Errorf("failed to marshal source spec: %w", err)
+	}
+	stream, err := c.pbClient.Sync(ctx, &pb.Sync_Request{
+		Spec: b,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to sync resources: %w", err)
 	}
 	for {
 		r, err := stream.Recv()
@@ -102,8 +67,12 @@ func (c *SourceClient) Fetch(ctx context.Context, spec specs.SourceSpec, res cha
 			}
 			return fmt.Errorf("failed to fetch resources from stream: %w", err)
 		}
-		res <- &FetchResultMessage{
-			Resource: r.Resource,
+		var resource schema.Resource
+		err = json.Unmarshal(r.Resource, &resource)
+		if err != nil {
+			return fmt.Errorf("failed to unmarshal resource: %w", err)
 		}
+
+		res <- &resource
 	}
 }

codegen/doc.go

@@ -0,0 +1,2 @@
+//codgen helps autogenerate cloudquery plugins configured by definition
+package codegen