fix: Use logger instead of stdout (#20007)

erezrokah · web-flow · commit 72ac737d6ac2 · 2024-12-19T17:51:07.000Z
#### Summary We should be using the logger instead of `fmt.Println`. Also don't think an array should be an error so made it an `Info`. On AWS if you run with `tables: ["*"]` you get quite a lot of arrays: ![image](https://github.com/user-attachments/assets/9648ad09-f7ae-4017-ae42-c324b37a3a7e)
diff --git a/plugins/transformer/jsonflattener/client/client.go b/plugins/transformer/jsonflattener/client/client.go
@@ -37,7 +37,7 @@ func New(_ context.Context, logger zerolog.Logger, s []byte, opts plugin.NewClie
 		return nil, err
 	}
 
-	tf, err := transformers.NewFromSpec(c.spec)
+	tf, err := transformers.NewFromSpec(logger, c.spec)
 	if err != nil {
 		return nil, err
 	}
diff --git a/plugins/transformer/jsonflattener/client/recordupdater/record_updater.go b/plugins/transformer/jsonflattener/client/recordupdater/record_updater.go
@@ -10,20 +10,23 @@ import (
 	"github.com/cloudquery/cloudquery/plugins/transformer/jsonflattener/client/util"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/cloudquery/plugin-sdk/v4/types"
+	"github.com/rs/zerolog"
 )
 
 // RecordUpdater takes an `arrow.Record` and knows how to make simple subsequent changes to it.
 // It doesn't know which table it belongs to or if the changes make sense.
 type RecordUpdater struct {
+	logger        zerolog.Logger
 	record        arrow.Record
 	schemaUpdater *schemaupdater.SchemaUpdater
 	tableName     string
 }
 
-func New(record arrow.Record) *RecordUpdater {
+func New(logger zerolog.Logger, record arrow.Record) *RecordUpdater {
 	tableName, _ := record.Schema().Metadata().GetValue(schema.MetadataTableName)
 
 	return &RecordUpdater{
+		logger:        logger,
 		record:        record,
 		schemaUpdater: schemaupdater.New(record.Schema()),
 		tableName:     tableName,
@@ -38,14 +41,18 @@ func (r *RecordUpdater) FlattenJSONFields() (arrow.Record, error) {
 		if !ok || rawTypeSchema == "" {
 			continue
 		}
-		var unprocessedTypeSchema map[string]any
+		var unprocessedTypeSchema any
 		if err := json.Unmarshal([]byte(rawTypeSchema), &unprocessedTypeSchema); err != nil {
-			// In this case it can be an array
-			fmt.Println("failed to unmarshal type schema", rawTypeSchema)
+			r.logger.Error().Err(err).Msg("failed to unmarshal type schema")
 			continue
 		}
-		typeSchema := preprocessTypeSchema(unprocessedTypeSchema)
-		fieldTypeSchemas[field.Name] = typeSchema
+		switch s := unprocessedTypeSchema.(type) {
+		case map[string]any:
+			typeSchema := preprocessTypeSchema(s)
+			fieldTypeSchemas[field.Name] = typeSchema
+		default:
+			r.logger.Info().Msgf("skipping unsupported type schema: %T", unprocessedTypeSchema)
+		}
 	}
 
 	if len(fieldTypeSchemas) == 0 {
diff --git a/plugins/transformer/jsonflattener/client/recordupdater/record_updater_test.go b/plugins/transformer/jsonflattener/client/recordupdater/record_updater_test.go
@@ -8,6 +8,7 @@ import (
 	"github.com/apache/arrow-go/v18/arrow/array"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/cloudquery/plugin-sdk/v4/types"
+	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/require"
 )
 
@@ -17,7 +18,7 @@ func TestFlattenJSONFields(t *testing.T) {
 		map[string]string{"col1": `{"key_a": "utf8", "key_b": "int64", "key_c": "bool"}`},
 		[]arrow.Array{buildJSONColumn([]*any{toP(`{"key_a": "value", "key_b": 2, "key_c": true}`)})},
 	)
-	updater := New(record)
+	updater := New(zerolog.Nop(), record)
 
 	updatedRecord, err := updater.FlattenJSONFields()
 	require.NoError(t, err)
@@ -39,7 +40,7 @@ func TestFlattenJSONFieldsWithTimestamp(t *testing.T) {
 		map[string]string{"col1": `{"key_a": "timestamp[us, tz=UTC]"}`},
 		[]arrow.Array{buildJSONColumn([]*any{toP(`{"key_a": "2024-01-02T03:04:05.006Z"}`)})},
 	)
-	updater := New(record)
+	updater := New(zerolog.Nop(), record)
 
 	updatedRecord, err := updater.FlattenJSONFields()
 	require.NoError(t, err)
@@ -57,7 +58,7 @@ func TestFlattenJSONFieldsDoesntFlattenFieldsKeyedUTF8(t *testing.T) {
 		map[string]string{"col1": `{"key_a": "utf8", "key_b": "int64", "utf8": "any"}`},
 		[]arrow.Array{buildJSONColumn([]*any{toP(`{"key_a": "value", "key_b": 2, "utf8": "any"}`)})},
 	)
-	updater := New(record)
+	updater := New(zerolog.Nop(), record)
 
 	updatedRecord, err := updater.FlattenJSONFields()
 	require.NoError(t, err)
@@ -77,7 +78,7 @@ func TestNestedJSONFlattenedToFirstLevel(t *testing.T) {
 		map[string]string{"col1": `{"nested": {"key_a": "utf8", "key_b": "int64", "key_c": "bool"}}`},
 		[]arrow.Array{buildJSONColumn([]*any{toP(`{"nested": {"key_a": "value", "key_b": 2, "key_c": true}}`)})},
 	)
-	updater := New(record)
+	updater := New(zerolog.Nop(), record)
 
 	updatedRecord, err := updater.FlattenJSONFields()
 	require.NoError(t, err)
@@ -97,7 +98,7 @@ func TestDifferentCasingWorks(t *testing.T) {
 		map[string]string{"col": `{"subcolumn_one": "utf8"}`}, // Note the different casing
 		[]arrow.Array{buildJSONColumn([]*any{toP(`{"subcolumnOne": "value", "unknownColumn": 2}`)})},
 	)
-	updater := New(record)
+	updater := New(zerolog.Nop(), record)
 
 	updatedRecord, err := updater.FlattenJSONFields()
 	require.NoError(t, err)
@@ -118,7 +119,7 @@ func TestDifferentCasingWorksEvenWhenFirstRowIsNull(t *testing.T) {
 		// Note first 3 rows are nil
 		[]arrow.Array{buildJSONColumn([]*any{nil, nil, nil, toP(`{"subcolumnOne": "value", "unknownColumn": 2}`)})},
 	)
-	updater := New(record)
+	updater := New(zerolog.Nop(), record)
 
 	updatedRecord, err := updater.FlattenJSONFields()
 	require.NoError(t, err)
diff --git a/plugins/transformer/jsonflattener/client/transformers/transformers.go b/plugins/transformer/jsonflattener/client/transformers/transformers.go
@@ -7,6 +7,7 @@ import (
 	"github.com/cloudquery/cloudquery/plugins/transformer/jsonflattener/client/recordupdater"
 	"github.com/cloudquery/cloudquery/plugins/transformer/jsonflattener/client/spec"
 	"github.com/cloudquery/cloudquery/plugins/transformer/jsonflattener/client/tablematcher"
+	"github.com/rs/zerolog"
 )
 
 type TransformationFn = func(arrow.Record) (arrow.Record, error)
@@ -18,10 +19,10 @@ type Transformer struct {
 	schemaFn SchemaTransformationFn
 }
 
-func NewFromSpec(sp spec.Spec) (*Transformer, error) {
+func NewFromSpec(logger zerolog.Logger, sp spec.Spec) (*Transformer, error) {
 	tr := &Transformer{matcher: tablematcher.New(sp.Tables)}
 
-	tr.fn = FlattenJSONFields()
+	tr.fn = FlattenJSONFields(logger)
 	tr.schemaFn = transformSchema(tr.fn)
 
 	return tr, nil
@@ -58,9 +59,9 @@ func (tr *Transformer) TransformSchema(schema *arrow.Schema) (*arrow.Schema, err
 	return tr.schemaFn(schema)
 }
 
-func FlattenJSONFields() TransformationFn {
+func FlattenJSONFields(logger zerolog.Logger) TransformationFn {
 	return func(record arrow.Record) (arrow.Record, error) {
-		record, err := recordupdater.New(record).FlattenJSONFields()
+		record, err := recordupdater.New(logger, record).FlattenJSONFields()
 		if err != nil {
 			return nil, err
 		}
diff --git a/plugins/transformer/jsonflattener/client/transformers/transformers_test.go b/plugins/transformer/jsonflattener/client/transformers/transformers_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/cloudquery/plugin-sdk/v4/types"
 	"github.com/goccy/go-json"
+	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/require"
 )
 
@@ -43,7 +44,7 @@ func TestTransform(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			transformer, err := NewFromSpec(tt.spec)
+			transformer, err := NewFromSpec(zerolog.Nop(), tt.spec)
 			require.NoError(t, err, "NewFromSpec() should not return an error")
 
 			transformedRecord, err := transformer.Transform(tt.record)

Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ func New(_ context.Context, logger zerolog.Logger, s []byte, opts plugin.NewClie`
`37`	`37`	`return nil, err`
`38`	`38`	`}`
`39`	`39`
`40`		`- tf, err := transformers.NewFromSpec(c.spec)`
	`40`	`+ tf, err := transformers.NewFromSpec(logger, c.spec)`
`41`	`41`	`if err != nil {`
`42`	`42`	`return nil, err`
`43`	`43`	`}`