refactor(log): replace panics with log fatal or log panic functions (#1723)

Signed-off-by: Laurentiu Niculae <niculae.laurentiu1@gmail.com>

pkg/storage/cache.go

@@ -1,16 +1,16 @@
 package storage
 
 import (
-	"zotregistry.io/zot/errors"
+	zerr "zotregistry.io/zot/errors"
 	"zotregistry.io/zot/pkg/api/config"
 	zlog "zotregistry.io/zot/pkg/log"
 	"zotregistry.io/zot/pkg/storage/cache"
 	"zotregistry.io/zot/pkg/storage/constants"
 )
 
-func CreateCacheDatabaseDriver(storageConfig config.StorageConfig, log zlog.Logger) cache.Cache {
+func CreateCacheDatabaseDriver(storageConfig config.StorageConfig, log zlog.Logger) (cache.Cache, error) {
 	if !storageConfig.Dedupe && storageConfig.StorageDriver == nil {
-		return nil
+		return nil, nil
 	}
 
 	// local cache
@@ -20,9 +20,7 @@ func CreateCacheDatabaseDriver(storageConfig config.StorageConfig, log zlog.Logg
 		params.Name = constants.BoltdbName
 		params.UseRelPaths = getUseRelPaths(&storageConfig)
 
-		driver, _ := Create("boltdb", params, log)
-
-		return driver
+		return Create("boltdb", params, log)
 	}
 
 	// remote cache
@@ -31,13 +29,13 @@ func CreateCacheDatabaseDriver(storageConfig config.StorageConfig, log zlog.Logg
 		if !ok {
 			log.Warn().Msg("remote cache driver name missing!")
 
-			return nil
+			return nil, nil
 		}
 
 		if name != constants.DynamoDBDriverName {
 			log.Warn().Str("driver", name).Msg("remote cache driver unsupported!")
 
-			return nil
+			return nil, nil
 		}
 
 		// dynamodb
@@ -46,27 +44,25 @@ func CreateCacheDatabaseDriver(storageConfig config.StorageConfig, log zlog.Logg
 		dynamoParams.Region, _ = storageConfig.CacheDriver["region"].(string)
 		dynamoParams.TableName, _ = storageConfig.CacheDriver["cachetablename"].(string)
 
-		driver, _ := Create("dynamodb", dynamoParams, log)
-
-		return driver
+		return Create("dynamodb", dynamoParams, log)
 	}
 
-	return nil
+	return nil, nil
 }
 
 func Create(dbtype string, parameters interface{}, log zlog.Logger) (cache.Cache, error) {
 	switch dbtype {
 	case "boltdb":
 		{
-			return cache.NewBoltDBCache(parameters, log), nil
+			return cache.NewBoltDBCache(parameters, log)
 		}
 	case "dynamodb":
 		{
-			return cache.NewDynamoDBCache(parameters, log), nil
+			return cache.NewDynamoDBCache(parameters, log)
 		}
 	default:
 		{
-			return nil, errors.ErrBadConfig
+			return nil, zerr.ErrBadConfig
 		}
 	}
 }

pkg/storage/cache/boltdb.go

@@ -1,6 +1,7 @@
 package cache
 
 import (
+	"fmt"
 	"os"
 	"path"
 	"path/filepath"
@@ -9,7 +10,7 @@ import (
 	godigest "github.com/opencontainers/go-digest"
 	"go.etcd.io/bbolt"
 
-	"zotregistry.io/zot/errors"
+	zerr "zotregistry.io/zot/errors"
 	zlog "zotregistry.io/zot/pkg/log"
 	"zotregistry.io/zot/pkg/storage/constants"
 )
@@ -27,17 +28,20 @@ type BoltDBDriverParameters struct {
 	UseRelPaths bool
 }
 
-func NewBoltDBCache(parameters interface{}, log zlog.Logger) Cache {
+func NewBoltDBCache(parameters interface{}, log zlog.Logger) (*BoltDBDriver, error) {
 	properParameters, ok := parameters.(BoltDBDriverParameters)
 	if !ok {
-		panic("Failed type assertion")
+		log.Error().Err(zerr.ErrTypeAssertionFailed).Msgf("expected type '%T' but got '%T'",
+			BoltDBDriverParameters{}, parameters)
+
+		return nil, zerr.ErrTypeAssertionFailed
 	}
 
 	err := os.MkdirAll(properParameters.RootDir, constants.DefaultDirPerms)
 	if err != nil {
 		log.Error().Err(err).Str("directory", properParameters.RootDir).Msg("unable to create directory for cache db")
 
-		return nil
+		return nil, err
 	}
 
 	dbPath := path.Join(properParameters.RootDir, properParameters.Name+constants.DBExtensionName)
@@ -48,9 +52,17 @@ func NewBoltDBCache(parameters interface{}, log zlog.Logger) Cache {
 
 	cacheDB, err := bbolt.Open(dbPath, 0o600, dbOpts) //nolint:gomnd
 	if err != nil {
+		if strings.Contains(err.Error(), "timeout") {
+			err := fmt.Errorf("%w: %w, path '%s'", zerr.ErrTimeout, zerr.ErrDatabaseFileAlreadyInUse, dbPath)
+
+			log.Error().Err(err).Str("dbPath", dbPath).Msg("unable to create cache db")
+
+			return nil, err
+		}
+
 		log.Error().Err(err).Str("dbPath", dbPath).Msg("unable to create cache db")
 
-		return nil
+		return nil, err
 	}
 
 	if err := cacheDB.Update(func(tx *bbolt.Tx) error {
@@ -66,7 +78,7 @@ func NewBoltDBCache(parameters interface{}, log zlog.Logger) Cache {
 		// something went wrong
 		log.Error().Err(err).Msg("unable to create a cache")
 
-		return nil
+		return nil, err
 	}
 
 	return &BoltDBDriver{
@@ -74,7 +86,7 @@ func NewBoltDBCache(parameters interface{}, log zlog.Logger) Cache {
 		db:          cacheDB,
 		useRelPaths: properParameters.UseRelPaths,
 		log:         log,
-	}
+	}, nil
 }
 
 func (d *BoltDBDriver) UsesRelativePaths() bool {
@@ -87,9 +99,9 @@ func (d *BoltDBDriver) Name() string {
 
 func (d *BoltDBDriver) PutBlob(digest godigest.Digest, path string) error {
 	if path == "" {
-		d.log.Error().Err(errors.ErrEmptyValue).Str("digest", digest.String()).Msg("empty path provided")
+		d.log.Error().Err(zerr.ErrEmptyValue).Str("digest", digest.String()).Msg("empty path provided")
 
-		return errors.ErrEmptyValue
+		return zerr.ErrEmptyValue
 	}
 
 	// use only relative (to rootDir) paths on blobs
@@ -105,7 +117,7 @@ func (d *BoltDBDriver) PutBlob(digest godigest.Digest, path string) error {
 		root := tx.Bucket([]byte(constants.BlobsCache))
 		if root == nil {
 			// this is a serious failure
-			err := errors.ErrCacheRootBucket
+			err := zerr.ErrCacheRootBucket
 			d.log.Error().Err(err).Msg("unable to access root bucket")
 
 			return err
@@ -168,7 +180,7 @@ func (d *BoltDBDriver) GetBlob(digest godigest.Digest) (string, error) {
 		root := tx.Bucket([]byte(constants.BlobsCache))
 		if root == nil {
 			// this is a serious failure
-			err := errors.ErrCacheRootBucket
+			err := zerr.ErrCacheRootBucket
 			d.log.Error().Err(err).Msg("unable to access root bucket")
 
 			return err
@@ -182,7 +194,7 @@ func (d *BoltDBDriver) GetBlob(digest godigest.Digest) (string, error) {
 			return nil
 		}
 
-		return errors.ErrCacheMiss
+		return zerr.ErrCacheMiss
 	}); err != nil {
 		return "", err
 	}
@@ -195,7 +207,7 @@ func (d *BoltDBDriver) HasBlob(digest godigest.Digest, blob string) bool {
 		root := tx.Bucket([]byte(constants.BlobsCache))
 		if root == nil {
 			// this is a serious failure
-			err := errors.ErrCacheRootBucket
+			err := zerr.ErrCacheRootBucket
 			d.log.Error().Err(err).Msg("unable to access root bucket")
 
 			return err
@@ -203,22 +215,22 @@ func (d *BoltDBDriver) HasBlob(digest godigest.Digest, blob string) bool {
 
 		bucket := root.Bucket([]byte(digest.String()))
 		if bucket == nil {
-			return errors.ErrCacheMiss
+			return zerr.ErrCacheMiss
 		}
 
 		origin := bucket.Bucket([]byte(constants.OriginalBucket))
 		if origin == nil {
-			return errors.ErrCacheMiss
+			return zerr.ErrCacheMiss
 		}
 
 		deduped := bucket.Bucket([]byte(constants.DuplicatesBucket))
 		if deduped == nil {
-			return errors.ErrCacheMiss
+			return zerr.ErrCacheMiss
 		}
 
 		if origin.Get([]byte(blob)) == nil {
 			if deduped.Get([]byte(blob)) == nil {
-				return errors.ErrCacheMiss
+				return zerr.ErrCacheMiss
 			}
 		}
 
@@ -255,7 +267,7 @@ func (d *BoltDBDriver) DeleteBlob(digest godigest.Digest, path string) error {
 		root := tx.Bucket([]byte(constants.BlobsCache))
 		if root == nil {
 			// this is a serious failure
-			err := errors.ErrCacheRootBucket
+			err := zerr.ErrCacheRootBucket
 			d.log.Error().Err(err).Msg("unable to access root bucket")
 
 			return err
@@ -263,12 +275,12 @@ func (d *BoltDBDriver) DeleteBlob(digest godigest.Digest, path string) error {
 
 		bucket := root.Bucket([]byte(digest.String()))
 		if bucket == nil {
-			return errors.ErrCacheMiss
+			return zerr.ErrCacheMiss
 		}
 
 		deduped := bucket.Bucket([]byte(constants.DuplicatesBucket))
 		if deduped == nil {
-			return errors.ErrCacheMiss
+			return zerr.ErrCacheMiss
 		}
 
 		if err := deduped.Delete([]byte(path)); err != nil {

pkg/storage/cache/boltdb_internal_test.go

@@ -0,0 +1,63 @@
+package cache
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/opencontainers/go-digest"
+	. "github.com/smartystreets/goconvey/convey"
+	"go.etcd.io/bbolt"
+
+	"zotregistry.io/zot/pkg/storage/constants"
+)
+
+func TestBoltDriverErrors(t *testing.T) {
+	Convey("Make a new cache", t, func() {
+		tmpDir := t.TempDir()
+
+		boltDB, err := bbolt.Open(filepath.Join(tmpDir, "bolt.db"), 0o644, bbolt.DefaultOptions)
+		So(err, ShouldBeNil)
+
+		driver := BoltDBDriver{
+			db: boltDB,
+		}
+
+		Convey("Empty boltdb", func() {
+			// bucket not found
+			err = driver.PutBlob(digest.FromString("s"), "path")
+			So(err, ShouldNotBeNil)
+
+			_, err = driver.GetBlob(digest.FromString("s"))
+			So(err, ShouldNotBeNil)
+
+			has := driver.HasBlob(digest.FromString("s"), "blob")
+			So(has, ShouldBeFalse)
+
+			err = driver.DeleteBlob(digest.FromString("s"), "blob")
+			So(err, ShouldNotBeNil)
+		})
+
+		Convey("cache miss", func() {
+			goodDigest := digest.FromString("s")
+
+			err := driver.db.Update(func(tx *bbolt.Tx) error {
+				buck, err := tx.CreateBucketIfNotExists([]byte(constants.BlobsCache))
+				So(err, ShouldBeNil)
+
+				_, err = buck.CreateBucket([]byte(goodDigest))
+				So(err, ShouldBeNil)
+
+				return nil
+			})
+			So(err, ShouldBeNil)
+
+			// digest bucket not found
+			err = driver.DeleteBlob(digest.FromString("bad-digest"), "path")
+			So(err, ShouldNotBeNil)
+
+			// duplicate bucket not exist
+			err = driver.DeleteBlob(goodDigest, "path")
+			So(err, ShouldNotBeNil)
+		})
+	})
+}

pkg/storage/cache/boltdb_test.go

@@ -19,7 +19,8 @@ func TestBoltDBCache(t *testing.T) {
 		log := log.NewLogger("debug", "")
 		So(log, ShouldNotBeNil)
 
-		So(func() { _, _ = storage.Create("boltdb", "failTypeAssertion", log) }, ShouldPanic)
+		_, err := storage.Create("boltdb", "failTypeAssertion", log)
+		So(err, ShouldNotBeNil)
 
 		cacheDriver, _ := storage.Create("boltdb", cache.BoltDBDriverParameters{"/deadBEEF", "cache_test", true}, log)
 		So(cacheDriver, ShouldBeNil)

pkg/storage/cache/dynamodb.go

@@ -61,10 +61,13 @@ func (d *DynamoDBDriver) NewTable(tableName string) error {
 	return nil
 }
 
-func NewDynamoDBCache(parameters interface{}, log zlog.Logger) Cache {
+func NewDynamoDBCache(parameters interface{}, log zlog.Logger) (*DynamoDBDriver, error) {
 	properParameters, ok := parameters.(DynamoDBDriverParameters)
 	if !ok {
-		panic("Failed type assertion!")
+		log.Error().Err(zerr.ErrTypeAssertionFailed).Msgf("expected type '%T' but got '%T'",
+			BoltDBDriverParameters{}, parameters)
+
+		return nil, zerr.ErrTypeAssertionFailed
 	}
 
 	// custom endpoint resolver to point to localhost
@@ -85,7 +88,7 @@ func NewDynamoDBCache(parameters interface{}, log zlog.Logger) Cache {
 	if err != nil {
 		log.Error().Err(err).Msg("unable to load AWS SDK config for dynamodb")
 
-		return nil
+		return nil, err
 	}
 
 	driver := &DynamoDBDriver{client: dynamodb.NewFromConfig(cfg), tableName: properParameters.TableName, log: log}
@@ -93,10 +96,16 @@ func NewDynamoDBCache(parameters interface{}, log zlog.Logger) Cache {
 	err = driver.NewTable(driver.tableName)
 	if err != nil {
 		log.Error().Err(err).Str("tableName", driver.tableName).Msg("unable to create table for cache")
+
+		return nil, err
 	}
 
 	// Using the Config value, create the DynamoDB client
-	return driver
+	return driver, nil
+}
+
+func (d *DynamoDBDriver) SetTableName(table string) {
+	d.tableName = table
 }
 
 func (d *DynamoDBDriver) UsesRelativePaths() bool {
@@ -212,7 +221,7 @@ func (d *DynamoDBDriver) DeleteBlob(digest godigest.Digest, path string) error {
 	// if original blob is the one deleted
 	if originBlob == path {
 		// move duplicate blob to original, storage will move content here
-		originBlob, _ = d.getDuplicateBlob(digest)
+		originBlob, _ = d.GetDuplicateBlob(digest)
 		if originBlob != "" {
 			if err := d.putOriginBlob(digest, originBlob); err != nil {
 				return err
@@ -232,7 +241,7 @@ func (d *DynamoDBDriver) DeleteBlob(digest godigest.Digest, path string) error {
 	return nil
 }
 
-func (d *DynamoDBDriver) getDuplicateBlob(digest godigest.Digest) (string, error) {
+func (d *DynamoDBDriver) GetDuplicateBlob(digest godigest.Digest) (string, error) {
 	resp, err := d.client.GetItem(context.TODO(), &dynamodb.GetItemInput{
 		TableName: aws.String(d.tableName),
 		Key: map[string]types.AttributeValue{

pkg/storage/cache/dynamodb_test.go

@@ -22,9 +22,8 @@ func TestDynamoDB(t *testing.T) {
 
 		// bad params
 
-		So(func() {
-			_ = cache.NewDynamoDBCache("bad params", log)
-		}, ShouldPanic)
+		_, err := cache.NewDynamoDBCache("bad params", log)
+		So(err, ShouldNotBeNil)
 
 		keyDigest := godigest.FromString("key")
 
@@ -33,20 +32,7 @@ func TestDynamoDB(t *testing.T) {
 			TableName: "BlobTable",
 			Region:    "us-east-2",
 		}, log)
-		So(cacheDriver, ShouldNotBeNil)
-		So(err, ShouldBeNil)
-
-		val, err := cacheDriver.GetBlob(keyDigest)
-		So(err, ShouldNotBeNil)
-		So(val, ShouldBeEmpty)
-
-		err = cacheDriver.PutBlob(keyDigest, path.Join(dir, "value"))
-		So(err, ShouldNotBeNil)
-
-		exists := cacheDriver.HasBlob(keyDigest, path.Join(dir, "value"))
-		So(exists, ShouldBeFalse)
-
-		err = cacheDriver.DeleteBlob(keyDigest, path.Join(dir, "value"))
+		So(cacheDriver, ShouldBeNil)
 		So(err, ShouldNotBeNil)
 
 		cacheDriver, err = storage.Create("dynamodb", cache.DynamoDBDriverParameters{
@@ -60,7 +46,7 @@ func TestDynamoDB(t *testing.T) {
 		returnedName := cacheDriver.Name()
 		So(returnedName, ShouldEqual, "dynamodb")
 
-		val, err = cacheDriver.GetBlob(keyDigest)
+		val, err := cacheDriver.GetBlob(keyDigest)
 		So(err, ShouldNotBeNil)
 		So(val, ShouldBeEmpty)
 
@@ -74,7 +60,7 @@ func TestDynamoDB(t *testing.T) {
 		So(err, ShouldBeNil)
 		So(val, ShouldNotBeEmpty)
 
-		exists = cacheDriver.HasBlob(keyDigest, path.Join(dir, "value"))
+		exists := cacheDriver.HasBlob(keyDigest, path.Join(dir, "value"))
 		So(exists, ShouldBeTrue)
 
 		err = cacheDriver.DeleteBlob(keyDigest, path.Join(dir, "value"))
@@ -159,3 +145,32 @@ func TestDynamoDB(t *testing.T) {
 		So(val, ShouldBeEmpty)
 	})
 }
+
+func TestDynamoDBError(t *testing.T) {
+	tskip.SkipDynamo(t)
+	Convey("Test dynamoDB", t, func(c C) {
+		log := log.NewLogger("debug", "")
+
+		cacheDriver, err := cache.NewDynamoDBCache(cache.DynamoDBDriverParameters{
+			Endpoint:  os.Getenv("DYNAMODBMOCK_ENDPOINT"),
+			TableName: "BlobTable",
+			Region:    "us-east-2",
+		}, log)
+		So(cacheDriver, ShouldNotBeNil)
+		So(err, ShouldBeNil)
+
+		returnedName := cacheDriver.Name()
+		So(returnedName, ShouldEqual, "dynamodb")
+
+		cacheDriver.SetTableName("bad-table")
+
+		_, err = cacheDriver.GetBlob(godigest.FromString("str"))
+		So(err, ShouldNotBeNil)
+		found := cacheDriver.HasBlob(godigest.FromString("str"), "path")
+		So(found, ShouldBeFalse)
+		_, err = cacheDriver.GetDuplicateBlob(godigest.FromString("str"))
+		So(err, ShouldNotBeNil)
+		err = cacheDriver.DeleteBlob(godigest.FromString("str"), "path")
+		So(err, ShouldNotBeNil)
+	})
+}

pkg/storage/cache_test.go

@@ -19,7 +19,8 @@ func TestCache(t *testing.T) {
 		log := log.NewLogger("debug", "")
 		So(log, ShouldNotBeNil)
 
-		So(func() { _, _ = storage.Create("boltdb", "failTypeAssertion", log) }, ShouldPanic)
+		_, err := storage.Create("boltdb", "failTypeAssertion", log)
+		So(err, ShouldNotBeNil)
 
 		cacheDriver, _ := storage.Create("boltdb", cache.BoltDBDriverParameters{
 			RootDir:     "/deadBEEF",

pkg/storage/storage.go

@@ -9,7 +9,7 @@ import (
 	godigest "github.com/opencontainers/go-digest"
 	ispec "github.com/opencontainers/image-spec/specs-go/v1"
 
-	"zotregistry.io/zot/errors"
+	zerr "zotregistry.io/zot/errors"
 	"zotregistry.io/zot/pkg/api/config"
 	zcommon "zotregistry.io/zot/pkg/common"
 	"zotregistry.io/zot/pkg/extensions/monitoring"
@@ -28,9 +28,9 @@ func New(config *config.Config, linter common.Lint, metrics monitoring.MetricSer
 
 	if config.Storage.RootDirectory == "" {
 		// we can't proceed without global storage
-		log.Error().Err(errors.ErrImgStoreNotFound).Msg("controller: no storage config provided")
+		log.Error().Err(zerr.ErrImgStoreNotFound).Msg("controller: no storage config provided")
 
-		return storeController, errors.ErrImgStoreNotFound
+		return storeController, zerr.ErrImgStoreNotFound
 	}
 
 	// no need to validate hard links work on s3
@@ -47,18 +47,24 @@ func New(config *config.Config, linter common.Lint, metrics monitoring.MetricSer
 	var defaultStore storageTypes.ImageStore
 
 	if config.Storage.StorageDriver == nil {
+		cacheDriver, err := CreateCacheDatabaseDriver(config.Storage.StorageConfig, log)
+		if err != nil {
+			return storeController, err
+		}
+
 		// false positive lint - linter does not implement Lint method
 		//nolint:typecheck,contextcheck
 		rootDir := config.Storage.RootDirectory
 		defaultStore = local.NewImageStore(rootDir,
-			config.Storage.Dedupe, config.Storage.Commit, log, metrics, linter,
-			CreateCacheDatabaseDriver(config.Storage.StorageConfig, log),
+			config.Storage.Dedupe, config.Storage.Commit, log, metrics, linter, cacheDriver,
 		)
 	} else {
 		storeName := fmt.Sprintf("%v", config.Storage.StorageDriver["name"])
 		if storeName != constants.S3StorageDriverName {
-			log.Fatal().Err(errors.ErrBadConfig).Str("storageDriver", storeName).
+			log.Error().Err(zerr.ErrBadConfig).Str("storageDriver", storeName).
 				Msg("unsupported storage driver")
+
+			return storeController, fmt.Errorf("storageDriver '%s' unsupported storage driver: %w", storeName, zerr.ErrBadConfig)
 		}
 		// Init a Storager from connection string.
 		store, err := factory.Create(storeName, config.Storage.StorageDriver)
@@ -75,11 +81,15 @@ func New(config *config.Config, linter common.Lint, metrics monitoring.MetricSer
 			rootDir = fmt.Sprintf("%v", config.Storage.StorageDriver["rootdirectory"])
 		}
 
+		cacheDriver, err := CreateCacheDatabaseDriver(config.Storage.StorageConfig, log)
+		if err != nil {
+			return storeController, err
+		}
+
 		// false positive lint - linter does not implement Lint method
 		//nolint: typecheck,contextcheck
 		defaultStore = s3.NewImageStore(rootDir, config.Storage.RootDirectory,
-			config.Storage.Dedupe, config.Storage.Commit, log, metrics, linter, store,
-			CreateCacheDatabaseDriver(config.Storage.StorageConfig, log))
+			config.Storage.Dedupe, config.Storage.Commit, log, metrics, linter, store, cacheDriver)
 	}
 
 	storeController.DefaultStore = defaultStore
@@ -128,9 +138,9 @@ func getSubStore(cfg *config.Config, subPaths map[string]config.StorageConfig,
 			isSame, _ := config.SameFile(cfg.Storage.RootDirectory, storageConfig.RootDirectory)
 
 			if isSame {
-				log.Error().Err(errors.ErrBadConfig).Msg("sub path storage directory is same as root directory")
+				log.Error().Err(zerr.ErrBadConfig).Msg("sub path storage directory is same as root directory")
 
-				return nil, errors.ErrBadConfig
+				return nil, zerr.ErrBadConfig
 			}
 
 			isUnique := true
@@ -149,10 +159,14 @@ func getSubStore(cfg *config.Config, subPaths map[string]config.StorageConfig,
 			// add it to uniqueSubFiles
 			// Create a new image store and assign it to imgStoreMap
 			if isUnique {
+				cacheDriver, err := CreateCacheDatabaseDriver(storageConfig, log)
+				if err != nil {
+					return nil, err
+				}
+
 				rootDir := storageConfig.RootDirectory
 				imgStoreMap[storageConfig.RootDirectory] = local.NewImageStore(rootDir,
-					storageConfig.Dedupe, storageConfig.Commit, log, metrics, linter,
-					CreateCacheDatabaseDriver(storageConfig, log),
+					storageConfig.Dedupe, storageConfig.Commit, log, metrics, linter, cacheDriver,
 				)
 
 				subImageStore[route] = imgStoreMap[storageConfig.RootDirectory]
@@ -160,8 +174,10 @@ func getSubStore(cfg *config.Config, subPaths map[string]config.StorageConfig,
 		} else {
 			storeName := fmt.Sprintf("%v", storageConfig.StorageDriver["name"])
 			if storeName != constants.S3StorageDriverName {
-				log.Fatal().Err(errors.ErrBadConfig).Str("storageDriver", storeName).
+				log.Error().Err(zerr.ErrBadConfig).Str("storageDriver", storeName).
 					Msg("unsupported storage driver")
+
+				return nil, fmt.Errorf("storageDriver '%s' unsupported storage driver: %w", storeName, zerr.ErrBadConfig)
 			}
 
 			// Init a Storager from connection string.
@@ -179,11 +195,18 @@ func getSubStore(cfg *config.Config, subPaths map[string]config.StorageConfig,
 				rootDir = fmt.Sprintf("%v", cfg.Storage.StorageDriver["rootdirectory"])
 			}
 
+			cacheDriver, err := CreateCacheDatabaseDriver(storageConfig, log)
+			if err != nil {
+				log.Error().Err(err).Any("config", storageConfig).
+					Msg("failed creating storage driver")
+
+				return nil, err
+			}
+
 			// false positive lint - linter does not implement Lint method
 			//nolint: typecheck
 			subImageStore[route] = s3.NewImageStore(rootDir, storageConfig.RootDirectory,
-				storageConfig.Dedupe, storageConfig.Commit, log, metrics, linter, store,
-				CreateCacheDatabaseDriver(storageConfig, log),
+				storageConfig.Dedupe, storageConfig.Commit, log, metrics, linter, store, cacheDriver,
 			)
 		}
 	}

pkg/storage/storage_test.go

@@ -121,6 +121,18 @@ var testCases = []struct {
 	},
 }
 
+func TestStorageNew(t *testing.T) {
+	Convey("New fail", t, func() {
+		// store name is wrong
+		conf := config.New()
+		conf.Storage.RootDirectory = "dir"
+		conf.Storage.StorageDriver = map[string]interface{}{}
+
+		_, err := storage.New(conf, nil, nil, zlog.NewLogger("debug", ""))
+		So(err, ShouldNotBeNil)
+	})
+}
+
 func TestStorageAPIs(t *testing.T) {
 	for _, testcase := range testCases {
 		testcase := testcase
@@ -871,6 +883,73 @@ func TestMandatoryAnnotations(t *testing.T) {
 	}
 }
 
+func TestStorageSubpaths(t *testing.T) {
+	Convey("Reach for coverage", t, func() {
+		tmpDirSubpath := t.TempDir()
+		config := &config.Config{
+			Storage: config.GlobalStorageConfig{
+				StorageConfig: config.StorageConfig{RootDirectory: t.TempDir()},
+				SubPaths: map[string]config.StorageConfig{
+					"a/":          {RootDirectory: tmpDirSubpath},
+					tmpDirSubpath: {RootDirectory: tmpDirSubpath},
+				},
+			},
+		}
+
+		_, err := storage.New(config, nil, nil, zlog.NewLogger("debug", ""))
+		So(err, ShouldBeNil)
+	})
+
+	Convey("Create unique subpath, error for cache driver", t, func() {
+		tmpDirSubpath := t.TempDir()
+		config := &config.Config{
+			Storage: config.GlobalStorageConfig{
+				StorageConfig: config.StorageConfig{RootDirectory: t.TempDir()},
+				SubPaths: map[string]config.StorageConfig{
+					"a/": {
+						RootDirectory: tmpDirSubpath,
+						RemoteCache:   true,
+						StorageDriver: map[string]interface{}{},
+						Dedupe:        true,
+					},
+				},
+			},
+		}
+
+		// create boltdb file and make it un-openable
+		dbPath := path.Join(tmpDirSubpath, storageConstants.BoltdbName+storageConstants.DBExtensionName)
+		err := os.WriteFile(dbPath, []byte(""), 0o000)
+		So(err, ShouldBeNil)
+
+		_, err = storage.New(config, nil, nil, zlog.NewLogger("debug", ""))
+		So(err, ShouldNotBeNil)
+
+		err = os.Chmod(dbPath, 0o600)
+		So(err, ShouldBeNil)
+	})
+
+	Convey("storeName != constants.S3StorageDriverName", t, func() {
+		tmpDirSubpath := t.TempDir()
+		config := &config.Config{
+			Storage: config.GlobalStorageConfig{
+				StorageConfig: config.StorageConfig{RootDirectory: t.TempDir()},
+				SubPaths: map[string]config.StorageConfig{
+					"a/": {
+						RootDirectory: tmpDirSubpath,
+						RemoteCache:   true,
+						StorageDriver: map[string]interface{}{
+							"name": "bad-name",
+						},
+					},
+				},
+			},
+		}
+
+		_, err := storage.New(config, nil, nil, zlog.NewLogger("debug", ""))
+		So(err, ShouldNotBeNil)
+	})
+}
+
 func TestDeleteBlobsInUse(t *testing.T) {
 	for _, testcase := range testCases {
 		testcase := testcase