fix: intial fixes

internal/parser/parser.go

@@ -2,8 +2,6 @@ package parser
 
 import (
 	"archive/tar"
-	"bufio"
-	"bytes"
 	"context"
 	"io"
 	"os"
@@ -17,6 +15,7 @@ import (
 	"github.com/google/go-containerregistry/pkg/name"
 	"github.com/google/go-containerregistry/pkg/v1/remote"
 	"github.com/google/go-containerregistry/pkg/v1/tarball"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/cli-runtime/pkg/printers"
@@ -56,19 +55,19 @@ func (p *Parser) Run(ctx context.Context, in io.Reader) error {
 		Workers: 1,
 	})
 
-	manifests := make(chan []byte, p.Workers)
+	objects := make(chan runtime.Object, p.Workers)
 
 	outWriter.Push(worker.Task(func(ctx context.Context) error {
 		for {
 			select {
 			case <-ctx.Done():
 				return nil
-			case manifest, ok := <-manifests:
+			case obj, ok := <-objects:
 				if !ok {
 					return nil
 				}
 
-				_, err := p.Output.Write(append([]byte("---\n"), manifest...))
+				err := p.Printer.PrintObj(obj, p.Output)
 				if err != nil {
 					p.Logger.Error(err, "failed to write manifests to output")
 					return abort(err)
@@ -83,26 +82,27 @@ func (p *Parser) Run(ctx context.Context, in io.Reader) error {
 	}
 
 	for _, resourceYAML := range strings.Split(string(manifest), "---") {
+		r := resourceYAML
 		pool.Push(worker.Task(func(ctx context.Context) error {
 			if len(resourceYAML) == 0 {
 				return nil
 			}
 
 			obj, gvk, err := p.Decoder.Decode(
-				[]byte(resourceYAML),
+				[]byte(r),
 				nil,
 				nil)
 			if err != nil {
 				return nil
 			}
 
-			return p.handleResource(obj, gvk, manifests)
+			return p.handleResource(obj, gvk, objects)
 		}))
 
 	}
 
 	p.exit(pool)
-	close(manifests)
+	close(objects)
 	p.exit(outWriter)
 	return nil
 }
@@ -117,10 +117,9 @@ func (p *Parser) exit(waiters ...worker.Waiter) {
 	}
 }
 
-func (p *Parser) handleResource(obj runtime.Object, gvk *schema.GroupVersionKind, out chan []byte) error {
+func (p *Parser) handleResource(obj runtime.Object, gvk *schema.GroupVersionKind, out chan runtime.Object) error {
 	if gvk.Group == "pkg.crossplane.io" && gvk.Kind == "Provider" {
 		provider := obj.(*crossplanev1.Provider)
-
 		p.Logger.Info("unpacking provider", "name", provider.Name, "url", provider.Spec.Package)
 
 		manifest, err := p.unpack(provider)
@@ -139,37 +138,33 @@ func (p *Parser) handleResource(obj runtime.Object, gvk *schema.GroupVersionKind
 
 		crd.Kind = "CustomResourceDefinition"
 		crd.APIVersion = "apiextensions.k8s.io/v1"
-
-		var b bytes.Buffer
-		err = p.Printer.PrintObj(crd, bufio.NewWriter(&b))
-		if err != nil {
-			return err
-		}
-
-		out <- b.Bytes()
+		out <- crd
 	}
 
 	return nil
 }
 
-func (p *Parser) parseManifest(manifest []byte, out chan []byte) error {
+func (p *Parser) parseManifest(manifest []byte, out chan runtime.Object) error {
 	for _, resourceYAML := range strings.Split(string(manifest), "---") {
 		if len(resourceYAML) == 0 {
 			continue
 		}
 
+		obj := unstructured.Unstructured{}
 		_, gvk, err := p.Decoder.Decode(
 			[]byte(resourceYAML),
 			nil,
-			nil)
+			&obj)
 
-		if err != nil {
-			out <- []byte(resourceYAML)
-		} else {
-			// exclude meta resources
-			if gvk.Group != "meta.pkg.crossplane.io" {
-				out <- []byte(resourceYAML)
-			}
+		if err != nil && !runtime.IsMissingKind(err) {
+			return err
+		} else if runtime.IsMissingKind(err) {
+			continue
+		}
+
+		// exclude meta resources
+		if gvk.Group != "meta.pkg.crossplane.io" {
+			out <- &obj
 		}
 	}
 

main.go

@@ -22,11 +22,11 @@ import (
 
 type Config struct {
 	Log struct {
-		Level    string `env:"LOG_LEVEL"`
-		Encoding string `env:"LOG_ENCODING"`
+		Level    string `env:"LOG_LEVEL, default=info"`
+		Encoding string `env:"LOG_ENCODING, default=json"`
 	}
-	File         string `env:"FILE"`
-	Output       string `env:"OUTPUT"`
+	File         string `env:"FILE, default=/dev/stdin"`
+	Output       string `env:"OUTPUT, default=/dev/stdout"`
 	FailFast     bool   `env:"FAIL_FAST"`
 	AllowFailure bool   `env:"ALLOW_FAILURE"`
 	Workers      int    `env:"WORKERS"`
@@ -37,23 +37,27 @@ var (
 )
 
 func init() {
-	flag.StringVarP(&config.Log.Level, "log-level", "l", "info", "Define the log level (default is warning) [debug,info,warn,error]")
-	flag.StringVarP(&config.Log.Encoding, "log-encoding", "e", "json", "Define the log format (default is json) [json,console]")
-	flag.StringVarP(&config.File, "file", "f", "/dev/stdin", "Path to input")
-	flag.StringVarP(&config.Output, "output", "o", "/dev/stdout", "Path to output")
+	flag.StringVarP(&config.Log.Level, "log-level", "l", "", "Define the log level (default is warning) [debug,info,warn,error]")
+	flag.StringVarP(&config.Log.Encoding, "log-encoding", "e", "", "Define the log format (default is json) [json,console]")
+	flag.StringVarP(&config.File, "file", "f", "", "Path to input")
+	flag.StringVarP(&config.Output, "output", "o", "", "Path to output")
 	flag.BoolVar(&config.AllowFailure, "allow-failure", false, "Do not exit > 0 if an error occured")
 	flag.BoolVar(&config.FailFast, "fail-fast", false, "Exit early if an error occured")
-	flag.IntVar(&config.Workers, "workers", runtime.NumCPU(), "Workers used to parse manifests")
+	flag.IntVar(&config.Workers, "workers", 0, "Workers used to parse manifests")
 }
 
 func main() {
-	ctx := context.Background()
+	ctx := context.TODO()
 	if err := envconfig.Process(ctx, config); err != nil {
 		log.Fatal(err)
 	}
 
 	flag.Parse()
 
+	if config.Workers == 0 {
+		config.Workers = runtime.NumCPU()
+	}
+
 	logger, err := buildLogger()
 	must(err)