feature; Dynamic container loading support & Error collection (flyteo…

…rg#64)

* feature; Dynamic container loading support

* Fix linter

* Error migration done too

* Update storage/stowstore.go

* rename file

* Cleaned up the storage package

* lint fix

* updated config, fixes config

boilerplate/lyft/golang_test_targets/Makefile

@@ -39,9 +39,15 @@ test_benchmark:
 
 .PHONY: test_unit_cover
 test_unit_cover:
-	go test ./... -coverprofile /tmp/cover.out -covermode=count; go tool cover -func /tmp/cover.out
+	go test ./... -coverprofile /tmp/cover.out -covermode=count
+	go tool cover -func /tmp/cover.out
 
 .PHONY: test_unit_visual
 test_unit_visual:
-	go test ./... -coverprofile /tmp/cover.out -covermode=count; go tool cover -html=/tmp/cover.out 
-
+	go test ./... -coverprofile /tmp/cover.out -covermode=count
+	go tool cover -html=/tmp/cover.out
+
+.PHONY: test_unit_codecov
+test_unit_codecov:
+	go test ./... -race -coverprofile=coverage.txt -covermode=atomic
+	curl -s https://codecov.io/bash > codecov_bash.sh && bash codecov_bash.sh

config/errors.go

@@ -6,32 +6,3 @@ var (
 	ErrStrictModeValidation       = fmt.Errorf("failed strict mode check")
 	ErrChildConfigOverridesConfig = fmt.Errorf("child config attempts to override an existing native config property")
 )
-
-// A helper object that collects errors.
-type ErrorCollection []error
-
-func (e ErrorCollection) Error() string {
-	res := ""
-	for _, err := range e {
-		res = fmt.Sprintf("%v\n%v", res, err.Error())
-	}
-
-	return res
-}
-
-func (e ErrorCollection) ErrorOrDefault() error {
-	if len(e) == 0 {
-		return nil
-	}
-
-	return e
-}
-
-func (e *ErrorCollection) Append(err error) bool {
-	if err != nil {
-		*e = append(*e, err)
-		return true
-	}
-
-	return false
-}

config/utils.go

@@ -6,6 +6,8 @@ import (
 	"reflect"
 
 	"github.com/pkg/errors"
+
+	stdLibErrs "github.com/lyft/flytestdlib/errors"
 )
 
 // Uses Json marshal/unmarshal to make a deep copy of a config object.
@@ -42,7 +44,7 @@ func toInterface(config Config) (interface{}, error) {
 
 // Builds a generic map out of the root section config and its sub-sections configs.
 func AllConfigsAsMap(root Section) (m map[string]interface{}, err error) {
-	errs := ErrorCollection{}
+	errs := stdLibErrs.ErrorCollection{}
 	allConfigs := make(map[string]interface{}, len(root.GetSections()))
 	if root.GetConfig() != nil {
 		rootConfig, err := toInterface(root.GetConfig())

config/viper/collection.go

@@ -7,7 +7,7 @@ import (
 	"os"
 	"strings"
 
-	"github.com/lyft/flytestdlib/config"
+	"github.com/lyft/flytestdlib/errors"
 
 	"github.com/lyft/flytestdlib/logger"
 
@@ -38,7 +38,7 @@ type CollectionProxy struct {
 }
 
 func (c *CollectionProxy) BindPFlags(flags *pflag.FlagSet) error {
-	err := config.ErrorCollection{}
+	err := errors.ErrorCollection{}
 	for _, v := range c.underlying {
 		err.Append(v.BindPFlags(flags))
 	}
@@ -49,7 +49,7 @@ func (c *CollectionProxy) BindPFlags(flags *pflag.FlagSet) error {
 }
 
 func (c *CollectionProxy) BindEnv(input ...string) error {
-	err := config.ErrorCollection{}
+	err := errors.ErrorCollection{}
 	for _, v := range c.underlying {
 		err.Append(v.BindEnv(input...))
 	}
@@ -72,7 +72,7 @@ func (c *CollectionProxy) AutomaticEnv() {
 }
 
 func (c CollectionProxy) ReadInConfig() error {
-	err := config.ErrorCollection{}
+	err := errors.ErrorCollection{}
 	for _, v := range c.underlying {
 		err.Append(v.ReadInConfig())
 	}

config/viper/viper.go

@@ -11,10 +11,13 @@ import (
 
 	"github.com/pkg/errors"
 
+	stdLibErrs "github.com/lyft/flytestdlib/errors"
+
+	"github.com/spf13/cobra"
+
 	"github.com/lyft/flytestdlib/config"
 	"github.com/lyft/flytestdlib/config/files"
 	"github.com/lyft/flytestdlib/logger"
-	"github.com/spf13/cobra"
 
 	"github.com/fsnotify/fsnotify"
 	"github.com/mitchellh/mapstructure"
@@ -94,7 +97,7 @@ func (v viperAccessor) bindViperConfigsFromEnv(root config.Section) (err error)
 }
 
 func (v viperAccessor) bindViperConfigsEnvDepth(m map[string]interface{}, prefix string) error {
-	errs := config.ErrorCollection{}
+	errs := stdLibErrs.ErrorCollection{}
 	for key, val := range m {
 		subKey := prefix + key
 		if asMap, ok := val.(map[string]interface{}); ok {
@@ -120,7 +123,7 @@ func (v viperAccessor) updateConfig(ctx context.Context, r config.Section) error
 	// If a config file is found, read it in.
 	if err = v.viper.ReadInConfig(); err == nil {
 		logger.Printf(ctx, "Using config file: %+v", v.viper.ConfigFilesUsed())
-	} else if asErrorCollection, ok := err.(config.ErrorCollection); ok {
+	} else if asErrorCollection, ok := err.(stdLibErrs.ErrorCollection); ok {
 		shouldWatchChanges = false
 		for i, e := range asErrorCollection {
 			if _, isNotFound := errors.Cause(e).(viperLib.ConfigFileNotFoundError); isNotFound {
@@ -196,7 +199,7 @@ func (v viperAccessor) parseViperConfig(root config.Section) error {
 }
 
 func (v viperAccessor) parseViperConfigRecursive(root config.Section, settings interface{}) error {
-	errs := config.ErrorCollection{}
+	errs := stdLibErrs.ErrorCollection{}
 	var mine interface{}
 	myKeysCount := 0
 	if asMap, casted := settings.(map[string]interface{}); casted {

errors/errors.go

@@ -0,0 +1,32 @@
+package errors
+
+import "fmt"
+
+// A helper object that collects errors.
+type ErrorCollection []error
+
+func (e ErrorCollection) Error() string {
+	res := ""
+	for _, err := range e {
+		res = fmt.Sprintf("%v\n%v", res, err.Error())
+	}
+
+	return res
+}
+
+func (e ErrorCollection) ErrorOrDefault() error {
+	if len(e) == 0 {
+		return nil
+	}
+
+	return e
+}
+
+func (e *ErrorCollection) Append(err error) bool {
+	if err != nil {
+		*e = append(*e, err)
+		return true
+	}
+
+	return false
+}

config/errors_test.go → errors/errors_test.go

@@ -1,4 +1,4 @@
-package config
+package errors
 
 import (
 	"fmt"

storage/config.go

@@ -39,15 +39,20 @@ var (
 			Region:   "us-east-1",
 			AuthType: "iam",
 		},
+		MultiContainerEnabled: false,
 	}
 )
 
 // A common storage config.
 type Config struct {
-	Type          Type             `json:"type" pflag:",Sets the type of storage to configure [s3/minio/local/mem/stow]."`
-	Connection    ConnectionConfig `json:"connection"`
-	Stow          *StowConfig      `json:"stow,omitempty"`
-	InitContainer string           `json:"container" pflag:",Initial container to create -if it doesn't exist-.'"`
+	Type       Type             `json:"type" pflag:",Sets the type of storage to configure [s3/minio/local/mem/stow]."`
+	Connection ConnectionConfig `json:"connection"`
+	Stow       *StowConfig      `json:"stow,omitempty"`
+	// Container here is misleading, it refers to a Bucket (AWS S3) like blobstore entity. In some terms it could be a table
+	InitContainer string `json:"container" pflag:",Initial container (in s3 a bucket) to create -if it doesn't exist-.'"`
+	// By default if this is not enabled, multiple containers are not supported by the storage layer. Only the configured `container` InitContainer will be allowed to requests data from. But, if enabled then data will be loaded to written to any
+	// container specified in the DataReference.
+	MultiContainerEnabled bool `json:"enable-multicontainer" pflag:",If this is true, then the container argument is overlooked and redundant. This config will automatically open new connections to new containers/buckets as they are encountered"`
 	// Caching is recommended to improve the performance of underlying systems. It caches the metadata and resolving
 	// inputs is accelerated. The size of the cache is large so understand how to configure the cache.
 	// TODO provide some default config choices