[Feature] [ML] Deployment Handler (#1500)

chart/kube-arangodb/templates/ml/role.yaml

@@ -26,5 +26,13 @@ rules:
         - "arangomlstorages/status"
       verbs:
         - "*"
+    - apiGroups:
+        - "database.arangodb.com"
+      resources:
+        - "arangodeployments"
+      verbs:
+        - "get"
+        - "list"
+        - "watch"
 {{- end }}
 {{- end }}

cmd/cmd.go

@@ -160,6 +160,10 @@ var (
 		backupArangoD       time.Duration
 		backupUploadArangoD time.Duration
 	}
+	operatorReconciliationRetry struct {
+		delay time.Duration
+		count int
+	}
 	chaosOptions struct {
 		allowed bool
 	}
@@ -222,6 +226,8 @@ func init() {
 	f.DurationVar(&operatorTimeouts.backupUploadArangoD, "timeout.backup-upload", globals.BackupUploadArangoClientTimeout, "The request timeout to the ArangoDB during uploading files")
 	f.DurationVar(&shutdownOptions.delay, "shutdown.delay", defaultShutdownDelay, "The delay before running shutdown handlers")
 	f.DurationVar(&shutdownOptions.timeout, "shutdown.timeout", defaultShutdownTimeout, "Timeout for shutdown handlers")
+	f.DurationVar(&operatorReconciliationRetry.delay, "operator.reconciliation.retry.delay", globals.DefaultOperatorUpdateRetryDelay, "Delay between Object Update operations in the Reconciliation loop")
+	f.IntVar(&operatorReconciliationRetry.count, "operator.reconciliation.retry.count", globals.DefaultOperatorUpdateRetryCount, "Count of retries during Object Update operations in the Reconciliation loop")
 	f.BoolVar(&operatorOptions.scalingIntegrationEnabled, "internal.scaling-integration", false, "Enable Scaling Integration")
 	f.DurationVar(&operatorOptions.reconciliationDelay, "reconciliation.delay", 0, "Delay between reconciliation loops (<= 0 -> Disabled)")
 	f.Int64Var(&operatorKubernetesOptions.maxBatchSize, "kubernetes.max-batch-size", globals.DefaultKubernetesRequestBatchSize, "Size of batch during objects read")
@@ -281,6 +287,9 @@ func executeMain(cmd *cobra.Command, args []string) {
 	globals.GetGlobalTimeouts().BackupArangoClientTimeout().Set(operatorTimeouts.backupArangoD)
 	globals.GetGlobalTimeouts().BackupArangoClientUploadTimeout().Set(operatorTimeouts.backupUploadArangoD)
 
+	globals.GetGlobals().Retry().OperatorUpdateRetryDelay().Set(operatorReconciliationRetry.delay)
+	globals.GetGlobals().Retry().OperatorUpdateRetryCount().Set(operatorReconciliationRetry.count)
+
 	globals.GetGlobals().Kubernetes().RequestBatchSize().Set(operatorKubernetesOptions.maxBatchSize)
 	globals.GetGlobals().Backup().ConcurrentUploads().Set(operatorBackup.concurrentUploads)
 

docs/api/ArangoMLExtension.V1Alpha1.md

@@ -4,3 +4,9 @@
 
 ## Status
 
+### .status.conditions
+
+Type: `api.Conditions` <sup>[\[ref\]](https://github.com/arangodb/kube-arangodb/blob/1.2.35/pkg/apis/ml/v1alpha1/extension_status.go#L28)</sup>
+
+Conditions specific to the entire extension
+

pkg/apis/backup/v1/backup.go

@@ -61,3 +61,11 @@ type ArangoBackup struct {
 	Spec   ArangoBackupSpec   `json:"spec"`
 	Status ArangoBackupStatus `json:"status"`
 }
+
+func (a *ArangoBackup) GetStatus() ArangoBackupStatus {
+	return a.Status
+}
+
+func (a *ArangoBackup) SetStatus(status ArangoBackupStatus) {
+	a.Status = status
+}

pkg/apis/ml/v1alpha1/extension.go

@@ -45,3 +45,11 @@ type ArangoMLExtension struct {
 	Spec   ArangoMLExtensionSpec   `json:"spec"`
 	Status ArangoMLExtensionStatus `json:"status"`
 }
+
+func (a *ArangoMLExtension) GetStatus() ArangoMLExtensionStatus {
+	return a.Status
+}
+
+func (a *ArangoMLExtension) SetStatus(status ArangoMLExtensionStatus) {
+	a.Status = status
+}

pkg/apis/ml/v1alpha1/extension_conditions.go

@@ -0,0 +1,27 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package v1alpha1
+
+import api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+
+const (
+	ExtensionDeploymentFoundCondition api.ConditionType = "DeploymentFound"
+)

pkg/apis/ml/v1alpha1/extension_status.go

@@ -20,5 +20,10 @@
 
 package v1alpha1
 
+import api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+
 type ArangoMLExtensionStatus struct {
+	// Conditions specific to the entire extension
+	// +doc/type: api.Conditions
+	Conditions api.ConditionList `json:"conditions,omitempty"`
 }

pkg/debug_package/generator.go

@@ -42,6 +42,7 @@ var rootFactories = []shared.Factory{
 	kubernetes.Services(),
 	kubernetes.Deployments(),
 	kubernetes.AgencyDump(),
+	kubernetes.ML(),
 }
 
 func InitCommand(cmd *cobra.Command) {

pkg/debug_package/generators/kubernetes/arango_deployments.go

@@ -64,21 +64,15 @@ func deployments(logger zerolog.Logger, files chan<- shared.File) error {
 		return err
 	}
 
-	errDeployments := make([]error, len(deploymentList))
-
-	for id := range deploymentList {
-		errDeployments[id] = deployment(k, deploymentList[id], files)
-	}
-
-	if err := errors.Errors(errDeployments...); err != nil {
+	if err := errors.ExecuteWithErrorArrayP2(deployment, k, files, deploymentList...); err != nil {
 		logger.Err(err).Msgf("Error while collecting arango deployments")
 		return err
 	}
 
 	return nil
 }
 
-func deployment(client kclient.Client, depl *api.ArangoDeployment, files chan<- shared.File) error {
+func deployment(client kclient.Client, files chan<- shared.File, depl *api.ArangoDeployment) error {
 	files <- shared.NewYAMLFile(fmt.Sprintf("kubernetes/arango/deployments/%s.yaml", depl.GetName()), func() ([]interface{}, error) {
 		return []interface{}{depl}, nil
 	})

pkg/debug_package/generators/kubernetes/arango_ml.go

@@ -0,0 +1,62 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package kubernetes
+
+import (
+	"github.com/rs/zerolog"
+
+	"github.com/arangodb/kube-arangodb/pkg/debug_package/shared"
+	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
+)
+
+func ML() shared.Factory {
+	return shared.NewFactory("ml", true, ml)
+}
+
+func ml(logger zerolog.Logger, files chan<- shared.File) error {
+	k, ok := kclient.GetDefaultFactory().Client()
+	if !ok {
+		return errors.Newf("Client is not initialised")
+	}
+
+	if err := mlExtensions(logger, files, k); err != nil {
+		logger.Err(err).Msgf("Error while collecting arango ml extension")
+		return err
+	}
+
+	if err := mlStorages(logger, files, k); err != nil {
+		logger.Err(err).Msgf("Error while collecting arango ml storage")
+		return err
+	}
+
+	if err := mlBatchJobs(logger, files, k); err != nil {
+		logger.Err(err).Msgf("Error while collecting arango ml batch jobs")
+		return err
+	}
+
+	if err := mlCronJobs(logger, files, k); err != nil {
+		logger.Err(err).Msgf("Error while collecting arango ml cron jobs")
+		return err
+	}
+
+	return nil
+}

pkg/debug_package/generators/kubernetes/arango_ml_batch_job.go

@@ -0,0 +1,73 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package kubernetes
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/rs/zerolog"
+
+	mlApi "github.com/arangodb/kube-arangodb/pkg/apis/ml/v1alpha1"
+	"github.com/arangodb/kube-arangodb/pkg/debug_package/cli"
+	"github.com/arangodb/kube-arangodb/pkg/debug_package/shared"
+	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/kerrors"
+	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
+)
+
+func mlBatchJobs(logger zerolog.Logger, files chan<- shared.File, client kclient.Client) error {
+	batchjobs, err := listMLBatchJobs(client)
+	if err != nil {
+		if kerrors.IsForbiddenOrNotFound(err) {
+			return nil
+		}
+
+		return err
+	}
+
+	if err := errors.ExecuteWithErrorArrayP2(mlBatchJob, client, files, batchjobs...); err != nil {
+		logger.Err(err).Msgf("Error while collecting arango ml batchjobs")
+		return err
+	}
+
+	return nil
+}
+
+func mlBatchJob(client kclient.Client, files chan<- shared.File, ext *mlApi.ArangoMLBatchJob) error {
+	files <- shared.NewYAMLFile(fmt.Sprintf("kubernetes/arango/ml/batchjobs/%s.yaml", ext.GetName()), func() ([]interface{}, error) {
+		return []interface{}{ext}, nil
+	})
+
+	return nil
+}
+
+func listMLBatchJobs(client kclient.Client) ([]*mlApi.ArangoMLBatchJob, error) {
+	return ListObjects[*mlApi.ArangoMLBatchJobList, *mlApi.ArangoMLBatchJob](context.Background(), client.Arango().MlV1alpha1().ArangoMLBatchJobs(cli.GetInput().Namespace), func(result *mlApi.ArangoMLBatchJobList) []*mlApi.ArangoMLBatchJob {
+		q := make([]*mlApi.ArangoMLBatchJob, len(result.Items))
+
+		for id, e := range result.Items {
+			q[id] = e.DeepCopy()
+		}
+
+		return q
+	})
+}

pkg/debug_package/generators/kubernetes/arango_ml_cron_job.go

@@ -0,0 +1,73 @@
+//
+// DISCLAIMER
+//
+// Copyright 2023 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package kubernetes
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/rs/zerolog"
+
+	mlApi "github.com/arangodb/kube-arangodb/pkg/apis/ml/v1alpha1"
+	"github.com/arangodb/kube-arangodb/pkg/debug_package/cli"
+	"github.com/arangodb/kube-arangodb/pkg/debug_package/shared"
+	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/kerrors"
+	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
+)
+
+func mlCronJobs(logger zerolog.Logger, files chan<- shared.File, client kclient.Client) error {
+	cronjobs, err := listMLCronJobs(client)
+	if err != nil {
+		if kerrors.IsForbiddenOrNotFound(err) {
+			return nil
+		}
+
+		return err
+	}
+
+	if err := errors.ExecuteWithErrorArrayP2(mlCronJob, client, files, cronjobs...); err != nil {
+		logger.Err(err).Msgf("Error while collecting arango ml cronjobs")
+		return err
+	}
+
+	return nil
+}
+
+func mlCronJob(client kclient.Client, files chan<- shared.File, ext *mlApi.ArangoMLCronJob) error {
+	files <- shared.NewYAMLFile(fmt.Sprintf("kubernetes/arango/ml/cronjobs/%s.yaml", ext.GetName()), func() ([]interface{}, error) {
+		return []interface{}{ext}, nil
+	})
+
+	return nil
+}
+
+func listMLCronJobs(client kclient.Client) ([]*mlApi.ArangoMLCronJob, error) {
+	return ListObjects[*mlApi.ArangoMLCronJobList, *mlApi.ArangoMLCronJob](context.Background(), client.Arango().MlV1alpha1().ArangoMLCronJobs(cli.GetInput().Namespace), func(result *mlApi.ArangoMLCronJobList) []*mlApi.ArangoMLCronJob {
+		q := make([]*mlApi.ArangoMLCronJob, len(result.Items))
+
+		for id, e := range result.Items {
+			q[id] = e.DeepCopy()
+		}
+
+		return q
+	})
+}