fybrik · shlomitk1 · Jan 19, 2023 · Jan 18, 2023 · Jan 18, 2023 · Jan 18, 2023
diff --git a/manager/controllers/app/fybrikapplication_controller_unit_test.go b/manager/controllers/app/fybrikapplication_controller_unit_test.go
@@ -1535,7 +1535,7 @@ func TestWriteUnregisteredAsset(t *testing.T) {
 	err = cl.Get(context.Background(), plotterObjectKey, plotter)
 	g.Expect(err).NotTo(gomega.HaveOccurred())
 	g.Expect(plotter.Spec.Assets).To(gomega.HaveLen(1))
-	g.Expect(plotter.Spec.Assets["s3-not-exists/new-dataset"].DataStore.Connection.Name).To(gomega.Equal(utils.GetDefaultConnectionType()))
+	g.Expect(plotter.Spec.Assets["s3-not-exists/new-dataset"].DataStore.Connection.Name).To(gomega.Equal(account.Spec.Type))
 	g.Expect(plotter.Spec.Assets["s3-not-exists/new-dataset"].DataStore.Format).ToNot(gomega.BeEmpty())
 	g.Expect(plotter.Spec.Templates).To(gomega.HaveLen(1))
 }

diff --git a/manager/controllers/app/select_modules.go b/manager/controllers/app/select_modules.go
@@ -11,7 +11,6 @@ import (
 	"github.com/rs/zerolog"
 
 	fapp "fybrik.io/fybrik/manager/apis/app/v1beta1"
-	"fybrik.io/fybrik/manager/controllers/utils"
 	"fybrik.io/fybrik/pkg/adminconfig"
 	"fybrik.io/fybrik/pkg/datapath"
 	"fybrik.io/fybrik/pkg/environment"
@@ -102,7 +101,17 @@ func (p *PathBuilder) validateStorageRequirements(element *datapath.ResolvedEdge
 		// validate restrictions
 		moduleCapability := element.Module.Spec.Capabilities[element.CapabilityIndex]
 		account := p.Env.StorageAccounts[accountInd]
-		if !p.validateRestrictions(
+		matchStorageType := false
+		for _, inter := range moduleCapability.SupportedInterfaces {
+			if inter.Sink == nil {
+				continue
+			}
+			if inter.Sink.Protocol == account.Spec.Type {
+				matchStorageType = true
+			}
+		}
+
+		if !matchStorageType || !p.validateRestrictions(
 			p.Asset.Configuration.ConfigDecisions[moduleCapability.Capability].DeploymentRestrictions.StorageAccounts,
 			&account.Spec, account.Name) {
 			p.Log.Debug().Str(logging.DATASETID, p.Asset.Context.DataSetID).Msgf("storage account %s does not match the requirements",
@@ -320,7 +329,9 @@ func match(source, sink *taxonomy.Interface) bool {
 	if source == nil || sink == nil {
 		return false
 	}
-	if source.Protocol != sink.Protocol {
+	// an empty Protocol value is not checked
+	// either a module supports any protocol, or any protocol can be selected (no requirements)
+	if string(source.Protocol) != "" && string(sink.Protocol) != "" && source.Protocol != sink.Protocol {
 		return false
 	}
 	// an empty DataFormat value is not checked
@@ -393,10 +404,7 @@ func supportsSinkInterface(edge *datapath.Edge, sinkNode *datapath.Node) bool {
 func (p *PathBuilder) getAssetConnectionNode() *datapath.Node {
 	var protocol taxonomy.ConnectionType
 	var dataFormat taxonomy.DataFormat
-	// If the connection name is empty, the default protocol is s3.
-	if p.Asset.DataDetails == nil || p.Asset.DataDetails.Details.Connection.Name == "" {
-		protocol = utils.GetDefaultConnectionType()
-	} else {
+	if p.Asset.DataDetails != nil && p.Asset.DataDetails.Details.Connection.Name != "" {
 		protocol = p.Asset.DataDetails.Details.Connection.Name
 		dataFormat = p.Asset.DataDetails.Details.DataFormat
 	}

diff --git a/manager/controllers/app/solve_test.go b/manager/controllers/app/solve_test.go
@@ -510,6 +510,76 @@ func TestWriteNewAsset(t *testing.T) {
 	g.Expect(solution.DataPath[0].Module.Name).To(gomega.Equal(writeModule.Name))
 }
 
+// This test checks the write scenario for a new asset.
+// Storage account is for MySQL, module supports S3 only.
+// Result: failure to create the data path
+func TestStorageAndModuleMismatch(t *testing.T) {
+	t.Parallel()
+	if environment.UseCSP() {
+		t.Skip()
+	}
+	g := gomega.NewGomegaWithT(t)
+	env := newEnvironment()
+	writeModule := &fapp.FybrikModule{}
+	g.Expect(readObjectFromFile("../../testdata/unittests/module-read-write.yaml", writeModule)).NotTo(gomega.HaveOccurred())
+	addModule(env, writeModule)
+	account := &saApi.FybrikStorageAccount{}
+	g.Expect(readStorageAccountData("../../testdata/unittests/account-theshire.yaml", account)).NotTo(gomega.HaveOccurred())
+	account.Spec.Type = "mysql"
+	addStorageAccount(env, account)
+	addCluster(env, multicluster.Cluster{Metadata: multicluster.ClusterMetadata{Region: string(account.Spec.Geography)}})
+	asset := createWriteNewAssetRequest()
+	asset.StorageRequirements[account.Spec.Geography] = []taxonomy.Action{}
+	asset.Configuration.ConfigDecisions["write"] = adminconfig.Decision{
+		Deploy: adminconfig.StatusTrue,
+		DeploymentRestrictions: adminconfig.Restrictions{
+			StorageAccounts: []adminconfig.Restriction{{Property: "geography", Values: adminconfig.StringList{string(account.Spec.Geography)}}}},
+	}
+	_, err := solveSingleDataset(env, asset, &testLog)
+	g.Expect(err).To(gomega.HaveOccurred())
+}
+
+// This test checks the write scenario for a new asset.
+// Storage accounts are for MySQL and S3, modules support MySQL and S3.
+// Policy to select MySQL storage
+// Result: the correct module is chosen.
+func TestStorageTypeRestriction(t *testing.T) {
+	t.Parallel()
+	if environment.UseCSP() {
+		t.Skip()
+	}
+	g := gomega.NewGomegaWithT(t)
+	env := newEnvironment()
+	writeS3 := &fapp.FybrikModule{}
+	g.Expect(readObjectFromFile("../../testdata/unittests/module-read-write.yaml", writeS3)).NotTo(gomega.HaveOccurred())
+	addModule(env, writeS3)
+	writeMySQL := &fapp.FybrikModule{}
+	g.Expect(readObjectFromFile("../../testdata/unittests/module-write-mysql.yaml", writeMySQL)).NotTo(gomega.HaveOccurred())
+	addModule(env, writeMySQL)
+	accountS3 := &saApi.FybrikStorageAccount{}
+	g.Expect(readStorageAccountData("../../testdata/unittests/account-theshire.yaml", accountS3)).NotTo(gomega.HaveOccurred())
+	addStorageAccount(env, accountS3)
+	accountMySQL := &saApi.FybrikStorageAccount{}
+	g.Expect(readStorageAccountData("../../testdata/unittests/account-neverland.yaml", accountMySQL)).NotTo(gomega.HaveOccurred())
+	accountMySQL.Spec.Type = "mysql"
+	addStorageAccount(env, accountMySQL)
+	addCluster(env, multicluster.Cluster{Metadata: multicluster.ClusterMetadata{Region: string(accountS3.Spec.Geography)}})
+	asset := createWriteNewAssetRequest()
+	asset.StorageRequirements[accountS3.Spec.Geography] = []taxonomy.Action{}
+	asset.StorageRequirements[accountMySQL.Spec.Geography] = []taxonomy.Action{}
+	asset.Configuration.ConfigDecisions["write"] = adminconfig.Decision{
+		Deploy: adminconfig.StatusTrue,
+		DeploymentRestrictions: adminconfig.Restrictions{
+			StorageAccounts: []adminconfig.Restriction{{Property: "type", Values: adminconfig.StringList{string(accountMySQL.Spec.Type)}}}},
+	}
+	solution, err := solveSingleDataset(env, asset, &testLog)
+	g.Expect(err).ToNot(gomega.HaveOccurred())
+	g.Expect(solution.DataPath).To(gomega.HaveLen(1))
+	// write
+	g.Expect(solution.DataPath[0].StorageAccount.Type).To(gomega.Equal(accountMySQL.Spec.Type))
+	g.Expect(solution.DataPath[0].Module.Name).To(gomega.Equal(writeMySQL.Name))
+}
+
 // This test checks the write scenario
 // Asset exists, no storage is required
 func TestWriteExistingAsset(t *testing.T) {

diff --git a/manager/controllers/utils/utils.go b/manager/controllers/utils/utils.go
@@ -72,7 +72,3 @@ func UpdateStatus(ctx context.Context, cl client.Client, obj client.Object, prev
 		return cl.Status().Update(ctx, res)
 	})
 }
-
-func GetDefaultConnectionType() taxonomy.ConnectionType {
-	return utils.S3
-}
diff --git a/manager/testdata/unittests/module-write-mysql.yaml b/manager/testdata/unittests/module-write-mysql.yaml
@@ -0,0 +1,25 @@
+# Copyright 2020 IBM Corp.
+# SPDX-License-Identifier: Apache-2.0
+
+apiVersion: app.fybrik.io/v1beta1
+kind: FybrikModule
+metadata:
+  name: write-mysql
+  namespace: fybrik-system
+spec:
+  chart:
+    name:  ghcr.io/fybrik/fybrik-template:0.1.0
+  type: service
+  capabilities:
+    - capability: write
+      scope: workload
+      api:
+        connection:
+          name: fybrik-arrow-flight
+          fybrik-arrow-flight:
+            hostname: read-write-module
+            port: 80
+            scheme: grpc
+      supportedInterfaces:
+      - sink:
+          protocol: mysql
diff --git a/pkg/optimizer/datapath_csp.go b/pkg/optimizer/datapath_csp.go
@@ -11,7 +11,6 @@ import (
 
 	fappv1 "fybrik.io/fybrik/manager/apis/app/v1beta1"
 	fappv2 "fybrik.io/fybrik/manager/apis/app/v1beta2"
-	"fybrik.io/fybrik/manager/controllers/utils"
 	"fybrik.io/fybrik/pkg/adminconfig"
 	"fybrik.io/fybrik/pkg/datapath"
 	"fybrik.io/fybrik/pkg/model/datacatalog"
@@ -999,7 +998,7 @@ func arrayOfSameInt(num, arrayLen int) []string {
 
 func getAssetInterface(connection *datacatalog.GetAssetResponse) taxonomy.Interface {
 	if connection == nil || connection.Details.Connection.Name == "" {
-		return taxonomy.Interface{Protocol: utils.GetDefaultConnectionType(), DataFormat: ""}
+		return taxonomy.Interface{Protocol: "s3", DataFormat: ""}
 	}
 	return taxonomy.Interface{Protocol: connection.Details.Connection.Name, DataFormat: connection.Details.DataFormat}
 }

diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go
@@ -10,16 +10,13 @@ import (
 	"fmt"
 	"os"
 	"runtime"
-
-	"fybrik.io/fybrik/pkg/model/taxonomy"
 )
 
 const (
-	StepNameHashLength                               = 10
-	hashPostfixLength                                = 5
-	k8sMaxConformNameLength                          = 63
-	helmMaxConformNameLength                         = 53
-	S3                       taxonomy.ConnectionType = "s3"
+	StepNameHashLength       = 10
+	hashPostfixLength        = 5
+	k8sMaxConformNameLength  = 63
+	helmMaxConformNameLength = 53
 )
 
 // Intersection finds a common subset of two given sets of strings