Add StorageID to GetRegion (#8521)

* Add StorageID to Repository entity

* Bug fixes

* Remove API changes

* Add StorageID to Repo endpoints

* Almost empty commit

* Revert almso empty commit

* Remove from API

* Add param to Creation

* Fix PR comments

* Fix tests

* Add basic unit-tests

* Fix param

* Update tests

* Allow only empty StorageID

* Revert "Allow only empty StorageID"

This reverts commit 298bc0d.

* Revert "Revert "Allow only empty StorageID""

This reverts commit 299c292.

* Change validation

* Remove StorageID in a test

* Update tests

* Add StorageID to adapter.GetRegion

* Fix tests

pkg/api/controller.go

@@ -1923,6 +1923,9 @@ func (c *Controller) ListRepositories(w http.ResponseWriter, r *http.Request, pa
 }
 
 func (c *Controller) CreateRepository(w http.ResponseWriter, r *http.Request, body apigen.CreateRepositoryJSONRequestBody, params apigen.CreateRepositoryParams) {
+	storageID := swag.StringValue(body.StorageId)
+	storageNamespace := body.StorageNamespace
+
 	if !c.authorize(w, r, permissions.Node{
 		Type: permissions.NodeTypeAnd,
 		Nodes: []permissions.Node{
@@ -1935,7 +1938,7 @@ func (c *Controller) CreateRepository(w http.ResponseWriter, r *http.Request, bo
 			{
 				Permission: permissions.Permission{
 					Action:   permissions.AttachStorageNamespaceAction,
-					Resource: permissions.StorageNamespace(body.StorageNamespace),
+					Resource: permissions.StorageNamespace(storageNamespace),
 				},
 			},
 		},
@@ -1960,13 +1963,13 @@ func (c *Controller) CreateRepository(w http.ResponseWriter, r *http.Request, bo
 		c.LogAction(ctx, "repo_sample_data", r, body.Name, "", "")
 	}
 
-	if err := c.validateStorageNamespace(body.StorageNamespace); err != nil {
+	if err := c.validateStorageNamespace(storageNamespace); err != nil {
 		writeError(w, r, http.StatusBadRequest, err)
 		return
 	}
 
 	if !c.Config.Installation.AllowInterRegionStorage {
-		if err := block.ValidateInterRegionStorage(r.Context(), c.BlockAdapter, body.StorageNamespace); err != nil {
+		if err := block.ValidateInterRegionStorage(r.Context(), c.BlockAdapter, storageID, storageNamespace); err != nil {
 			writeError(w, r, http.StatusBadRequest, err)
 			return
 		}
@@ -1977,10 +1980,8 @@ func (c *Controller) CreateRepository(w http.ResponseWriter, r *http.Request, bo
 		defaultBranch = "main"
 	}
 
-	storageID := swag.StringValue(body.StorageId)
-
 	if !swag.BoolValue(body.ReadOnly) {
-		if err := c.ensureStorageNamespace(ctx, body.StorageNamespace); err != nil {
+		if err := c.ensureStorageNamespace(ctx, storageNamespace); err != nil {
 			var (
 				reason string
 				retErr error
@@ -2002,7 +2003,7 @@ func (c *Controller) CreateRepository(w http.ResponseWriter, r *http.Request, bo
 			}
 			c.Logger.
 				WithError(err).
-				WithField("storage_namespace", body.StorageNamespace).
+				WithField("storage_namespace", storageNamespace).
 				WithField("reason", reason).
 				Warn("Could not access storage namespace")
 			writeError(w, r, http.StatusBadRequest, fmt.Errorf("failed to create repository: %w", retErr))
@@ -2013,7 +2014,7 @@ func (c *Controller) CreateRepository(w http.ResponseWriter, r *http.Request, bo
 	if swag.BoolValue(params.Bare) {
 		// create a bare repository. This is useful in conjunction with refs-restore to create a copy
 		// of another repository by e.g. copying the _lakefs/ directory and restoring its refs
-		repo, err := c.Catalog.CreateBareRepository(ctx, body.Name, storageID, body.StorageNamespace, defaultBranch, swag.BoolValue(body.ReadOnly))
+		repo, err := c.Catalog.CreateBareRepository(ctx, body.Name, storageID, storageNamespace, defaultBranch, swag.BoolValue(body.ReadOnly))
 		if c.handleAPIError(ctx, w, r, err) {
 			return
 		}
@@ -2028,7 +2029,7 @@ func (c *Controller) CreateRepository(w http.ResponseWriter, r *http.Request, bo
 		return
 	}
 
-	newRepo, err := c.Catalog.CreateRepository(ctx, body.Name, storageID, body.StorageNamespace, defaultBranch, swag.BoolValue(body.ReadOnly))
+	newRepo, err := c.Catalog.CreateRepository(ctx, body.Name, storageID, storageNamespace, defaultBranch, swag.BoolValue(body.ReadOnly))
 	if err != nil {
 		c.handleAPIError(ctx, w, r, fmt.Errorf("error creating repository: %w", err))
 		return

pkg/block/adapter.go

@@ -198,7 +198,8 @@ type Adapter interface {
 	GetStorageNamespaceInfo() StorageNamespaceInfo
 	ResolveNamespace(storageNamespace, key string, identifierType IdentifierType) (QualifiedKey, error)
 
-	GetRegion(ctx context.Context, storageNamespace string) (string, error)
+	// GetRegion storageID is not actively used, and it's here mainly for completeness
+	GetRegion(ctx context.Context, storageID, storageNamespace string) (string, error)
 
 	RuntimeStats() map[string]string
 }

pkg/block/azure/adapter.go

@@ -625,7 +625,7 @@ func (a *Adapter) ResolveNamespace(storageNamespace, key string, identifierType
 	return block.DefaultResolveNamespace(storageNamespace, key, identifierType)
 }
 
-func (a *Adapter) GetRegion(_ context.Context, _ string) (string, error) {
+func (a *Adapter) GetRegion(_ context.Context, _, _ string) (string, error) {
 	return "", block.ErrOperationNotSupported
 }
 

pkg/block/gs/adapter.go

@@ -693,7 +693,7 @@ func (a *Adapter) ResolveNamespace(storageNamespace, key string, identifierType
 	return qualifiedKey, nil
 }
 
-func (a *Adapter) GetRegion(_ context.Context, _ string) (string, error) {
+func (a *Adapter) GetRegion(_ context.Context, _, _ string) (string, error) {
 	return "", block.ErrOperationNotSupported
 }
 

pkg/block/local/adapter.go

@@ -559,7 +559,7 @@ func (l *Adapter) ResolveNamespace(storageNamespace, key string, identifierType
 	}, nil
 }
 
-func (l *Adapter) GetRegion(_ context.Context, _ string) (string, error) {
+func (l *Adapter) GetRegion(_ context.Context, _, _ string) (string, error) {
 	return "", block.ErrOperationNotSupported
 }
 

pkg/block/mem/adapter.go

@@ -354,7 +354,7 @@ func (a *Adapter) ResolveNamespace(storageNamespace, key string, identifierType
 	return block.DefaultResolveNamespace(storageNamespace, key, identifierType)
 }
 
-func (a *Adapter) GetRegion(_ context.Context, _ string) (string, error) {
+func (a *Adapter) GetRegion(_ context.Context, _, _ string) (string, error) {
 	return "", block.ErrOperationNotSupported
 }
 

pkg/block/metrics.go

@@ -123,9 +123,9 @@ func (m *MetricsAdapter) ResolveNamespace(storageNamespace, key string, identifi
 	return m.adapter.ResolveNamespace(storageNamespace, key, identifierType)
 }
 
-func (m *MetricsAdapter) GetRegion(ctx context.Context, storageNamespace string) (string, error) {
+func (m *MetricsAdapter) GetRegion(ctx context.Context, storageID, storageNamespace string) (string, error) {
 	ctx = httputil.SetClientTrace(ctx, m.adapter.BlockstoreType())
-	return m.adapter.GetRegion(ctx, storageNamespace)
+	return m.adapter.GetRegion(ctx, storageID, storageNamespace)
 }
 
 func (m *MetricsAdapter) RuntimeStats() map[string]string {

pkg/block/s3/adapter.go

@@ -890,7 +890,7 @@ func (a *Adapter) ResolveNamespace(storageNamespace, key string, identifierType
 	return block.DefaultResolveNamespace(storageNamespace, key, identifierType)
 }
 
-func (a *Adapter) GetRegion(ctx context.Context, storageNamespace string) (string, error) {
+func (a *Adapter) GetRegion(ctx context.Context, _, storageNamespace string) (string, error) {
 	namespaceURL, err := url.Parse(storageNamespace)
 	if err != nil {
 		return "", fmt.Errorf(`%s isn't a valid url': %w`, storageNamespace, block.ErrInvalidNamespace)

pkg/block/transient/adapter.go

@@ -163,7 +163,7 @@ func (a *Adapter) ResolveNamespace(storageNamespace, key string, identifierType
 	return block.DefaultResolveNamespace(storageNamespace, key, identifierType)
 }
 
-func (a *Adapter) GetRegion(_ context.Context, _ string) (string, error) {
+func (a *Adapter) GetRegion(_ context.Context, _, _ string) (string, error) {
 	return "", block.ErrOperationNotSupported
 }
 

pkg/block/validations.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 )
 
-func ValidateInterRegionStorage(ctx context.Context, adapter Adapter, storageNamespace string) error {
+func ValidateInterRegionStorage(ctx context.Context, adapter Adapter, storageID, storageNamespace string) error {
 	blockstoreMetadata, err := adapter.BlockstoreMetadata(ctx)
 	if errors.Is(err, ErrOperationNotSupported) {
 		// region detection not supported for the server's blockstore, skip validation
@@ -15,7 +15,7 @@ func ValidateInterRegionStorage(ctx context.Context, adapter Adapter, storageNam
 		return fmt.Errorf("failed to get blockstore region: %w", err)
 	}
 
-	bucketRegion, err := adapter.GetRegion(ctx, storageNamespace)
+	bucketRegion, err := adapter.GetRegion(ctx, storageID, storageNamespace)
 	if err != nil {
 		return fmt.Errorf("failed to get region of storage namespace %s: %w", storageNamespace, ErrInvalidNamespace)
 	}

pkg/block/validations_test.go

@@ -20,7 +20,7 @@ func TestController_ValidateInterRegionStorage(t *testing.T) {
 		}
 		adapter := testutil.NewMockAdapter(opts...)
 
-		err := block.ValidateInterRegionStorage(ctx, adapter, "s3://us-west-2/bucket-1")
+		err := block.ValidateInterRegionStorage(ctx, adapter, "", "s3://us-west-2/bucket-1")
 		require.NoError(t, err)
 	})
 
@@ -31,7 +31,7 @@ func TestController_ValidateInterRegionStorage(t *testing.T) {
 		}
 		adapter := testutil.NewMockAdapter(opts...)
 
-		err := block.ValidateInterRegionStorage(ctx, adapter, "s3://us-west-2/bucket-1")
+		err := block.ValidateInterRegionStorage(ctx, adapter, "", "s3://us-west-2/bucket-1")
 		require.ErrorIs(t, err, block.ErrInvalidNamespace)
 	})
 }

pkg/testutil/adapter.go

@@ -152,7 +152,7 @@ func (a *MockAdapter) ResolveNamespace(storageNamespace, key string, identifierT
 	return block.DefaultResolveNamespace(storageNamespace, key, identifierType)
 }
 
-func (a *MockAdapter) GetRegion(_ context.Context, _ string) (string, error) {
+func (a *MockAdapter) GetRegion(_ context.Context, _, _ string) (string, error) {
 	if a.namespaceRegion != nil {
 		return *a.namespaceRegion, nil
 	} else {