Storage registry: fail at init if config is missing any providers (cs…

…3org#4370)

changelog/unreleased/storage-registry-initfail.md

@@ -0,0 +1,8 @@
+Enhancement: Storage registry: fail at init if config is missing any providers
+
+This change makes the dynamic storage registry fail at startup if there are
+missing rules in the config file. That is, any `mount_id` in the routing table
+must have a corresponding `storage_id`/`address` pair in the config, otherwise
+the registry will fail to start.
+
+https://github.com/cs3org/reva/pull/4370

pkg/ocm/share/repository/sql/sql_test.go

@@ -47,7 +47,7 @@ import (
 var (
 	dbName               = "reva_tests"
 	address              = "localhost"
-	port                 = 3306
+	port                 = 33059
 	m                    sync.Mutex // for increasing the port
 	ocmShareTable        = "ocm_shares"
 	ocmAccessMethodTable = "ocm_shares_access_methods"

pkg/storage/registry/dynamic/dynamic.go

@@ -23,6 +23,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"strings"
 
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	registrypb "github.com/cs3org/go-cs3apis/cs3/storage/registry/v1beta1"
@@ -74,7 +75,7 @@ func New(ctx context.Context, m map[string]interface{}) (storage.Registry, error
 	log := appctx.GetLogger(ctx)
 	annotatedLog := log.With().Str("storageregistry", "dynamic").Logger()
 
-	rt, err := initRoutingTree(c.DBUsername, c.DBPassword, c.DBHost, c.DBPort, c.DBName)
+	rt, err := initRoutingTree(c.DBUsername, c.DBPassword, c.DBHost, c.DBPort, c.DBName, c.Rules)
 	if err != nil {
 		return nil, errors.Wrap(err, "error initializing routing tree")
 	}
@@ -95,7 +96,7 @@ func New(ctx context.Context, m map[string]interface{}) (storage.Registry, error
 	return d, nil
 }
 
-func initRoutingTree(dbUsername, dbPassword, dbHost string, dbPort int, dbName string) (*routingtree.RoutingTree, error) {
+func initRoutingTree(dbUsername, dbPassword, dbHost string, dbPort int, dbName string, rules map[string]string) (*routingtree.RoutingTree, error) {
 	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", dbUsername, dbPassword, dbHost, dbPort, dbName))
 	if err != nil {
 		return nil, errors.Wrap(err, "error opening sql connection")
@@ -108,15 +109,24 @@ func initRoutingTree(dbUsername, dbPassword, dbHost string, dbPort int, dbName s
 
 	rs := make(map[string]string)
 
+	missingRules := []string{}
+
 	for results.Next() {
 		var p, m string
 		err = results.Scan(&p, &m)
 		if err != nil {
 			return nil, errors.Wrap(err, "error scanning rows from db")
 		}
+		if _, ok := rules[m]; !ok {
+			missingRules = append(missingRules, m)
+		}
 		rs[p] = m
 	}
 
+	if len(missingRules) != 0 {
+		return nil, errors.New("config: missing routes for: " + strings.Join(missingRules, ", "))
+	}
+
 	return routingtree.New(rs), nil
 }
 
@@ -185,7 +195,9 @@ func (d *dynamic) FindProviders(ctx context.Context, ref *provider.Reference) ([
 			}}, nil
 		}
 
-		return nil, errtypes.NotFound("storage provider not found for ref " + ref.String())
+		err := errtypes.InternalError("storage provider address not found for " + storageID)
+		l.Error().Err(err).Send()
+		return nil, err
 	}
 
 	providerAlias := d.ur.GetAlias(ctx, ref.Path)
@@ -203,7 +215,7 @@ func (d *dynamic) FindProviders(ctx context.Context, ref *provider.Reference) ([
 			})
 		} else {
 			err := errtypes.InternalError("storage provider address not configured for mountID " + p.ProviderId)
-			l.Error().Err(err).Msgf("error finding providers")
+			l.Error().Err(err).Send()
 			return nil, err
 		}
 	}

pkg/storage/registry/dynamic/dynamic_test.go

@@ -63,19 +63,13 @@ var _ = Describe("Dynamic storage provider", func() {
 				OpaqueId: "bob",
 			},
 		})
-		ctxCharlie = appctx.ContextSetUser(context.Background(), &userpb.User{
-			Id: &userpb.UserId{
-				OpaqueId: "charlie",
-			},
-		})
 
 		dbHost = "localhost"
-		dbPort = 3305
+		dbPort = 33059
 		dbName = "reva_tests"
 		routes = map[string]string{
 			"/home-a":                   "eoshome-i01",
 			"/home-b":                   "eoshome-i02",
-			"/home-c":                   "eoshome-i03",
 			"/eos/user/a":               "eosuser-i01",
 			"/eos/user/b":               "eosuser-i02",
 			"/eos/project/a":            "eosproject-i00",
@@ -93,6 +87,13 @@ var _ = Describe("Dynamic storage provider", func() {
 			"cephfs":         "cephfs:1234",
 			"vaultssda01":    "vaultssda01:1234",
 		}
+		badRules = map[string]string{
+			"eoshome-i01":    "eoshome-i01:1234",
+			"eosuser-i01":    "eosuser-i01:1234",
+			"eosuser-i02":    "eosuser-i02:1234",
+			"eosproject-i02": "eosproject-i02:1234",
+			"cephfs":         "cephfs:1234",
+		}
 		rewrites = map[string]string{
 			"/home":   "/home-{{substr 0 1 .Id.OpaqueId}}",
 			"/Shares": "/{{substr 0 1 .Id.OpaqueId}}",
@@ -221,14 +222,28 @@ var _ = Describe("Dynamic storage provider", func() {
 					"db_name":     dbName,
 				})
 
-				prv, _ := d.FindProviders(context.Background(), &provider.Reference{Path: "/eos/"})
-				fmt.Printf("\n\n\n%+v\n\n\n", prv)
-
 				Expect(d).ToNot(BeNil())
 				Expect(err).ToNot(HaveOccurred())
 			})
 		})
 
+		When("passed a config missing some rules", func() {
+			It("should return an error", func() {
+				_, err = New(context.Background(), map[string]interface{}{
+					"rules":       badRules,
+					"rewrites":    rewrites,
+					"home_path":   homePath,
+					"db_username": "test",
+					"db_password": "test",
+					"db_host":     dbHost,
+					"db_port":     dbPort,
+					"db_name":     dbName,
+				})
+
+				Expect(err).To(HaveOccurred())
+			})
+		})
+
 		When("passed a bad db host in the config", func() {
 			It("should return a en error", func() {
 				d, err = New(context.Background(), map[string]interface{}{
@@ -242,8 +257,6 @@ var _ = Describe("Dynamic storage provider", func() {
 					"db_name":     dbName,
 				})
 
-				fmt.Printf("\n\n\n%+v\n\n\n", err)
-
 				Expect(d).To(BeNil())
 				Expect(err).To(HaveOccurred())
 			})
@@ -304,14 +317,6 @@ var _ = Describe("Dynamic storage provider", func() {
 			})
 		})
 
-		When("passed context for user charlie who has home provider with a defined route but no rule in config", func() {
-			It("should return a provider not found error", func() {
-				h, err = d.GetHome(ctxCharlie)
-				Expect(err).To(HaveOccurred())
-				Expect(err).To(MatchError(errtypes.InternalError("storage provider address not configured for mountID eoshome-i03")))
-			})
-		})
-
 		When("passed context without user", func() {
 			It("should return an error", func() {
 				h, err = d.GetHome(context.Background())
@@ -359,8 +364,7 @@ var _ = Describe("Dynamic storage provider", func() {
 				}
 
 				_, err := d.FindProviders(ctxAlice, ref)
-				Expect(err).To(HaveOccurred())
-				Expect(err).To(MatchError(errtypes.NotFound("storage provider not found for ref resource_id:<storage_id:\"nope\" > ")))
+				Expect(err).To(MatchError(errtypes.InternalError("storage provider address not found for nope")))
 			})
 		})
 
@@ -393,17 +397,16 @@ var _ = Describe("Dynamic storage provider", func() {
 			})
 		}
 
-		When("passed a home path for user charlie who has home provider with a defined route but no rule in config", func() {
-			It("should return a provider not found error", func() {
-				_, err := d.FindProviders(ctxCharlie, testHomeRefs["/home"])
+		When("passed a path which storage id has no entry in the configuration", func() {
+			It("should return an internal error", func() {
+				_, err := d.FindProviders(context.Background(), &provider.Reference{Path: "/cephfs/project/n/notconf"})
 				Expect(err).To(HaveOccurred())
-				Expect(err).To(MatchError(errtypes.InternalError("storage provider address not configured for mountID eoshome-i03")))
 			})
 		})
 
 		When("passed a non-existing path", func() {
 			It("should return a provider not found error", func() {
-				_, err := d.FindProviders(ctxCharlie, &provider.Reference{
+				_, err := d.FindProviders(ctxAlice, &provider.Reference{
 					Path: "/non/existent",
 				})
 				Expect(err).To(HaveOccurred())