[Feature] [ML] Extension Storage Condition (#1518)

CHANGELOG.md

@@ -26,6 +26,7 @@
 - (Feature) (ML) Storage S3 sidecar implementation
 - (Feature) TLS CA Secret Key
 - (Refactoring) Extract Resource Helpers
+- (Feature) (ML) Extension Storage Condition
 
 ## [1.2.35](https://github.com/arangodb/kube-arangodb/tree/1.2.35) (2023-11-06)
 - (Maintenance) Update go-driver to v1.6.0, update IsNotFound() checks

docs/api/ArangoMLExtension.V1Alpha1.md

@@ -20,6 +20,14 @@ ArangoPipeDatabase define Database name to be used as MetadataService Backend in
 
 Default Value: `arangopipe`
 
+***
+
+### .spec.storage
+
+Type: `string` <sup>[\[ref\]](https://github.com/arangodb/kube-arangodb/blob/1.2.35/pkg/apis/ml/v1alpha1/extension_spec.go#L31)</sup>
+
+Storage specify the ArangoMLStorage used within Extension
+
 ## Status
 
 ### .status.conditions

pkg/apis/ml/v1alpha1/extension_conditions.go

@@ -23,6 +23,7 @@ package v1alpha1
 import api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
 
 const (
+	ExtensionStorageFoundCondition         api.ConditionType = "StorageFound"
 	ExtensionDeploymentFoundCondition      api.ConditionType = "DeploymentFound"
 	ExtensionMetadataServiceValidCondition api.ConditionType = "MetadataServiceValid"
 	LicenseValidCondition                  api.ConditionType = "LicenseValid"

pkg/apis/ml/v1alpha1/extension_spec.go

@@ -26,6 +26,9 @@ type ArangoMLExtensionSpec struct {
 	// MetadataService keeps the MetadataService configuration
 	// +doc/immutable: This setting cannot be changed after the MetadataService has been created.
 	MetadataService *ArangoMLExtensionSpecMetadataService `json:"metadataService,omitempty"`
+
+	// Storage specify the ArangoMLStorage used within Extension
+	Storage *string `json:"storage,omitempty"`
 }
 
 func (a *ArangoMLExtensionSpec) GetMetadataService() *ArangoMLExtensionSpecMetadataService {
@@ -36,8 +39,17 @@ func (a *ArangoMLExtensionSpec) GetMetadataService() *ArangoMLExtensionSpecMetad
 	return a.MetadataService
 }
 
+func (a *ArangoMLExtensionSpec) GetStorage() *string {
+	if a == nil || a.Storage == nil {
+		return nil
+	}
+
+	return a.Storage
+}
+
 func (a *ArangoMLExtensionSpec) Validate() error {
 	return shared.WithErrors(shared.PrefixResourceErrors("spec",
 		shared.PrefixResourceErrors("metadataService", a.GetMetadataService().Validate()),
+		shared.PrefixResourceErrors("storage", shared.ValidateRequired(a.GetStorage(), shared.ValidateResourceName)),
 	))
 }

pkg/apis/ml/v1alpha1/zz_generated.deepcopy.go

@@ -301,6 +301,11 @@ func (in *ArangoMLExtensionSpec) DeepCopyInto(out *ArangoMLExtensionSpec) {
 		*out = new(ArangoMLExtensionSpecMetadataService)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Storage != nil {
+		in, out := &in.Storage, &out.Storage
+		*out = new(string)
+		**out = **in
+	}
 	return
 }
 

pkg/apis/shared/validate.go

@@ -73,3 +73,12 @@ func ValidateUID(uid types.UID) error {
 
 	return nil
 }
+
+// ValidateRequired validates if required resource is provided
+func ValidateRequired[T any](in *T, validator func(T) error) error {
+	if in == nil {
+		return errors.Newf("resource should be not nil")
+	}
+
+	return validator(*in)
+}

pkg/crd/crds/ml-extension.schema.generated.yaml

@@ -17,6 +17,9 @@ v1alpha1:
                     type: string
                 type: object
             type: object
+          storage:
+            description: Storage specify the ArangoMLStorage used within Extension
+            type: string
         type: object
     type: object
     x-kubernetes-preserve-unknown-fields: true

pkg/operatorV2/handle.go

@@ -57,6 +57,8 @@ type HandleP3Func[P1, P2, P3 interface{}] func(ctx context.Context, p1 P1, p2 P2
 
 type HandleP4Func[P1, P2, P3, P4 interface{}] func(ctx context.Context, p1 P1, p2 P2, p3 P3, p4 P4) (bool, error)
 
+type HandleP5Func[P1, P2, P3, P4, P5 interface{}] func(ctx context.Context, p1 P1, p2 P2, p3 P3, p4 P4, p5 P5) (bool, error)
+
 type HandleP9Func[P1, P2, P3, P4, P5, P6, P7, P8, P9 interface{}] func(ctx context.Context, p1 P1, p2 P2, p3 P3, p4 P4, p5 P5, p6 P6, p7 P7, p8 P8, p9 P9) (bool, error)
 
 func HandleP0(ctx context.Context, handler ...HandleP0Func) (bool, error) {
@@ -185,6 +187,36 @@ func HandleP4WithCondition[P1, P2, P3, P4 interface{}](ctx context.Context, cond
 	return WithCondition(conditions, condition, changed, err)
 }
 
+func HandleP5[P1, P2, P3, P4, P5 interface{}](ctx context.Context, p1 P1, p2 P2, p3 P3, p4 P4, p5 P5, handler ...HandleP5Func[P1, P2, P3, P4, P5]) (bool, error) {
+	isChanged := false
+	for _, h := range handler {
+		changed, err := h(ctx, p1, p2, p3, p4, p5)
+		if changed {
+			isChanged = true
+		}
+
+		if err != nil {
+			return isChanged, err
+		}
+	}
+
+	return isChanged, nil
+}
+
+func HandleP5WithStop[P1, P2, P3, P4, P5 interface{}](ctx context.Context, p1 P1, p2 P2, p3 P3, p4 P4, p5 P5, handler ...HandleP5Func[P1, P2, P3, P4, P5]) (bool, error) {
+	changed, err := HandleP5[P1, P2, P3, P4, P5](ctx, p1, p2, p3, p4, p5, handler...)
+	if IsStop(err) {
+		return changed, nil
+	}
+
+	return changed, err
+}
+
+func HandleP5WithCondition[P1, P2, P3, P4, P5 interface{}](ctx context.Context, conditions *api.ConditionList, condition api.ConditionType, p1 P1, p2 P2, p3 P3, p4 P4, p5 P5, handler ...HandleP5Func[P1, P2, P3, P4, P5]) (bool, error) {
+	changed, err := HandleP5[P1, P2, P3, P4, P5](ctx, p1, p2, p3, p4, p5, handler...)
+	return WithCondition(conditions, condition, changed, err)
+}
+
 func HandleP9[P1, P2, P3, P4, P5, P6, P7, P8, P9 interface{}](ctx context.Context, p1 P1, p2 P2, p3 P3, p4 P4, p5 P5, p6 P6, p7 P7, p8 P8, p9 P9, handler ...HandleP9Func[P1, P2, P3, P4, P5, P6, P7, P8, P9]) (bool, error) {
 	isChanged := false
 	for _, h := range handler {
@@ -209,3 +241,8 @@ func HandleP9WithStop[P1, P2, P3, P4, P5, P6, P7, P8, P9 interface{}](ctx contex
 
 	return changed, err
 }
+
+func HandleP9WithCondition[P1, P2, P3, P4, P5, P6, P7, P8, P9 interface{}](ctx context.Context, conditions *api.ConditionList, condition api.ConditionType, p1 P1, p2 P2, p3 P3, p4 P4, p5 P5, p6 P6, p7 P7, p8 P8, p9 P9, handler ...HandleP9Func[P1, P2, P3, P4, P5, P6, P7, P8, P9]) (bool, error) {
+	changed, err := HandleP9[P1, P2, P3, P4, P5, P6, P7, P8, P9](ctx, p1, p2, p3, p4, p5, p6, p7, p8, p9, handler...)
+	return WithCondition(conditions, condition, changed, err)
+}

pkg/util/tests/kubernetes.go

@@ -27,6 +27,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	batch "k8s.io/api/batch/v1"
 	core "k8s.io/api/core/v1"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/uuid"
@@ -42,6 +43,7 @@ import (
 	operator "github.com/arangodb/kube-arangodb/pkg/operatorV2"
 	"github.com/arangodb/kube-arangodb/pkg/operatorV2/operation"
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/kerrors"
 )
 
 func Handle(handler operator.Handler, item operation.Item) error {
@@ -72,6 +74,12 @@ type KubernetesObject interface {
 func CreateObjects(t *testing.T, k8s kubernetes.Interface, arango arangoClientSet.Interface, objects ...interface{}) func(t *testing.T) {
 	for _, object := range objects {
 		switch v := object.(type) {
+		case **batch.Job:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := k8s.BatchV1().Jobs(vl.GetNamespace()).Create(context.Background(), vl, meta.CreateOptions{})
+			require.NoError(t, err)
 		case **core.Secret:
 			require.NotNil(t, v)
 
@@ -102,6 +110,67 @@ func CreateObjects(t *testing.T, k8s kubernetes.Interface, arango arangoClientSe
 			vl := *v
 			_, err := arango.MlV1alpha1().ArangoMLExtensions(vl.GetNamespace()).Create(context.Background(), vl, meta.CreateOptions{})
 			require.NoError(t, err)
+		case **mlApi.ArangoMLStorage:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := arango.MlV1alpha1().ArangoMLStorages(vl.GetNamespace()).Create(context.Background(), vl, meta.CreateOptions{})
+			require.NoError(t, err)
+		default:
+			require.Fail(t, fmt.Sprintf("Unable to create object: %s", reflect.TypeOf(v).String()))
+		}
+	}
+
+	return func(t *testing.T) {
+		RefreshObjects(t, k8s, arango, objects...)
+	}
+}
+
+func UpdateObjects(t *testing.T, k8s kubernetes.Interface, arango arangoClientSet.Interface, objects ...interface{}) func(t *testing.T) {
+	for _, object := range objects {
+		switch v := object.(type) {
+		case **batch.Job:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := k8s.BatchV1().Jobs(vl.GetNamespace()).Update(context.Background(), vl, meta.UpdateOptions{})
+			require.NoError(t, err)
+		case **core.Secret:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := k8s.CoreV1().Secrets(vl.GetNamespace()).Update(context.Background(), vl, meta.UpdateOptions{})
+			require.NoError(t, err)
+		case **api.ArangoDeployment:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := arango.DatabaseV1().ArangoDeployments(vl.GetNamespace()).Update(context.Background(), vl, meta.UpdateOptions{})
+			require.NoError(t, err)
+		case **api.ArangoClusterSynchronization:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := arango.DatabaseV1().ArangoClusterSynchronizations(vl.GetNamespace()).Update(context.Background(), vl, meta.UpdateOptions{})
+			require.NoError(t, err)
+		case **backupApi.ArangoBackup:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := arango.BackupV1().ArangoBackups(vl.GetNamespace()).Update(context.Background(), vl, meta.UpdateOptions{})
+			require.NoError(t, err)
+		case **mlApi.ArangoMLExtension:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := arango.MlV1alpha1().ArangoMLExtensions(vl.GetNamespace()).Update(context.Background(), vl, meta.UpdateOptions{})
+			require.NoError(t, err)
+		case **mlApi.ArangoMLStorage:
+			require.NotNil(t, v)
+
+			vl := *v
+			_, err := arango.MlV1alpha1().ArangoMLStorages(vl.GetNamespace()).Update(context.Background(), vl, meta.UpdateOptions{})
+			require.NoError(t, err)
 		default:
 			require.Fail(t, fmt.Sprintf("Unable to create object: %s", reflect.TypeOf(v).String()))
 		}
@@ -115,51 +184,111 @@ func CreateObjects(t *testing.T, k8s kubernetes.Interface, arango arangoClientSe
 func RefreshObjects(t *testing.T, k8s kubernetes.Interface, arango arangoClientSet.Interface, objects ...interface{}) {
 	for _, object := range objects {
 		switch v := object.(type) {
+		case **batch.Job:
+			require.NotNil(t, v)
+
+			vl := *v
+
+			vn, err := k8s.BatchV1().Jobs(vl.GetNamespace()).Get(context.Background(), vl.GetName(), meta.GetOptions{})
+			if err != nil {
+				if kerrors.IsNotFound(err) {
+					*v = nil
+				} else {
+					require.NoError(t, err)
+				}
+			} else {
+				*v = vn
+			}
 		case **core.Secret:
 			require.NotNil(t, v)
 
 			vl := *v
 
 			vn, err := k8s.CoreV1().Secrets(vl.GetNamespace()).Get(context.Background(), vl.GetName(), meta.GetOptions{})
-			require.NoError(t, err)
+			if err != nil {
-
+				if kerrors.IsNotFound(err) {
-			*v = vn
+					*v = nil
+				} else {
+					require.NoError(t, err)
+				}
+			} else {
+				*v = vn
+			}
 		case **api.ArangoDeployment:
 			require.NotNil(t, v)
 
 			vl := *v
 
 			vn, err := arango.DatabaseV1().ArangoDeployments(vl.GetNamespace()).Get(context.Background(), vl.GetName(), meta.GetOptions{})
-			require.NoError(t, err)
+			if err != nil {
-
+				if kerrors.IsNotFound(err) {
-			*v = vn
+					*v = nil
+				} else {
+					require.NoError(t, err)
+				}
+			} else {
+				*v = vn
+			}
 		case **api.ArangoClusterSynchronization:
 			require.NotNil(t, v)
 
 			vl := *v
 
 			vn, err := arango.DatabaseV1().ArangoClusterSynchronizations(vl.GetNamespace()).Get(context.Background(), vl.GetName(), meta.GetOptions{})
-			require.NoError(t, err)
+			if err != nil {
-
+				if kerrors.IsNotFound(err) {
-			*v = vn
+					*v = nil
+				} else {
+					require.NoError(t, err)
+				}
+			} else {
+				*v = vn
+			}
 		case **backupApi.ArangoBackup:
 			require.NotNil(t, v)
 
 			vl := *v
 
 			vn, err := arango.BackupV1().ArangoBackups(vl.GetNamespace()).Get(context.Background(), vl.GetName(), meta.GetOptions{})
-			require.NoError(t, err)
+			if err != nil {
-
+				if kerrors.IsNotFound(err) {
-			*v = vn
+					*v = nil
+				} else {
+					require.NoError(t, err)
+				}
+			} else {
+				*v = vn
+			}
 		case **mlApi.ArangoMLExtension:
 			require.NotNil(t, v)
 
 			vl := *v
 
 			vn, err := arango.MlV1alpha1().ArangoMLExtensions(vl.GetNamespace()).Get(context.Background(), vl.GetName(), meta.GetOptions{})
-			require.NoError(t, err)
+			if err != nil {
+				if kerrors.IsNotFound(err) {
+					*v = nil
+				} else {
+					require.NoError(t, err)
+				}
+			} else {
+				*v = vn
+			}
+		case **mlApi.ArangoMLStorage:
+			require.NotNil(t, v)
 
-			*v = vn
+			vl := *v
+
+			vn, err := arango.MlV1alpha1().ArangoMLStorages(vl.GetNamespace()).Get(context.Background(), vl.GetName(), meta.GetOptions{})
+			if err != nil {
+				if kerrors.IsNotFound(err) {
+					*v = nil
+				} else {
+					require.NoError(t, err)
+				}
+			} else {
+				*v = vn
+			}
 		default:
 			require.Fail(t, fmt.Sprintf("Unable to create object: %s", reflect.TypeOf(v).String()))
 		}
@@ -170,6 +299,12 @@ type MetaObjectMod[T meta.Object] func(t *testing.T, obj T)
 
 func SetMetaBasedOnType(t *testing.T, object meta.Object) {
 	switch v := object.(type) {
+	case *batch.Job:
+		v.Kind = "Job"
+		v.APIVersion = "batch/v1"
+		v.SetSelfLink(fmt.Sprintf("/api/batch/v1/jobs/%s/%s",
+			object.GetNamespace(),
+			object.GetName()))
 	case *core.Secret:
 		v.Kind = "Secret"
 		v.APIVersion = "v1"
@@ -208,6 +343,14 @@ func SetMetaBasedOnType(t *testing.T, object meta.Object) {
 			ml.ArangoMLExtensionResourcePlural,
 			object.GetNamespace(),
 			object.GetName()))
+	case *mlApi.ArangoMLStorage:
+		v.Kind = ml.ArangoMLStorageResourceKind
+		v.APIVersion = mlApi.SchemeGroupVersion.String()
+		v.SetSelfLink(fmt.Sprintf("/api/%s/%s/%s/%s",
+			mlApi.SchemeGroupVersion.String(),
+			ml.ArangoMLStorageResourcePlural,
+			object.GetNamespace(),
+			object.GetName()))
 	default:
 		require.Fail(t, fmt.Sprintf("Unable to create object: %s", reflect.TypeOf(v).String()))
 	}
@@ -244,6 +387,10 @@ func NewItem(t *testing.T, o operation.Operation, object meta.Object) operation.
 	}
 
 	switch v := object.(type) {
+	case *batch.Job:
+		item.Group = "batch"
+		item.Version = "v1"
+		item.Kind = "Job"
 	case *core.Secret:
 		item.Group = ""
 		item.Version = "v1"
@@ -264,6 +411,10 @@ func NewItem(t *testing.T, o operation.Operation, object meta.Object) operation.
 		item.Group = ml.ArangoMLGroupName
 		item.Version = mlApi.ArangoMLVersion
 		item.Kind = ml.ArangoMLExtensionResourceKind
+	case *mlApi.ArangoMLStorage:
+		item.Group = ml.ArangoMLGroupName
+		item.Version = mlApi.ArangoMLVersion
+		item.Kind = ml.ArangoMLStorageResourceKind
 	default:
 		require.Fail(t, fmt.Sprintf("Unable to create object: %s", reflect.TypeOf(v).String()))
 	}

pkg/util/tests/kubernetes_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	batch "k8s.io/api/batch/v1"
 	core "k8s.io/api/core/v1"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -51,14 +52,18 @@ func NewMetaObjectRun[T meta.Object](t *testing.T) {
 			refresh := CreateObjects(t, c.Kubernetes(), c.Arango(), &obj)
 
 			refresh(t)
+
+			UpdateObjects(t, c.Kubernetes(), c.Arango(), &obj)
 		})
 	})
 }
 
 func Test_NewMetaObject(t *testing.T) {
+	NewMetaObjectRun[*batch.Job](t)
 	NewMetaObjectRun[*core.Secret](t)
 	NewMetaObjectRun[*api.ArangoDeployment](t)
 	NewMetaObjectRun[*api.ArangoClusterSynchronization](t)
 	NewMetaObjectRun[*backupApi.ArangoBackup](t)
 	NewMetaObjectRun[*mlApi.ArangoMLExtension](t)
+	NewMetaObjectRun[*mlApi.ArangoMLStorage](t)
 }