[Refactor] Configurable throttle info (#937)

Author: ajanikow

CHANGELOG.md

@@ -11,6 +11,7 @@
 - (Feature) Allow to configure action timeouts
 - (Feature) (AT) Add ArangoTask API
 - (Bugfix) Fix NPE in State fetcher
+- (Refactor) Configurable throttle inspector
 
 ## [1.2.8](https://github.com/arangodb/kube-arangodb/tree/1.2.8) (2022-02-24)
 - Do not check License V2 on Community images

cmd/admin.go

@@ -46,7 +46,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/constants"
 	"github.com/arangodb/kube-arangodb/pkg/util/errors"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
-	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/secret"
+	secretv1 "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/secret/v1"
 	"github.com/arangodb/kube-arangodb/pkg/util/kclient"
 )
 
@@ -291,7 +291,7 @@ func createClient(endpoints []string, certCA *x509.CertPool, auth connection.Aut
 }
 
 // getJWTTokenFromSecrets returns token from the secret.
-func getJWTTokenFromSecrets(ctx context.Context, secrets secret.ReadInterface, name string) (connection.Authentication, error) {
+func getJWTTokenFromSecrets(ctx context.Context, secrets secretv1.ReadInterface, name string) (connection.Authentication, error) {
 	ctxChild, cancel := globals.GetGlobalTimeouts().Kubernetes().WithTimeout(ctx)
 	defer cancel()
 
@@ -309,7 +309,7 @@ func getJWTTokenFromSecrets(ctx context.Context, secrets secret.ReadInterface, n
 }
 
 // getCACertificate returns CA certificate from the secret.
-func getCACertificate(ctx context.Context, secrets secret.ReadInterface, name string) (*x509.CertPool, error) {
+func getCACertificate(ctx context.Context, secrets secretv1.ReadInterface, name string) (*x509.CertPool, error) {
 	ctxChild, cancel := globals.GetGlobalTimeouts().Kubernetes().WithTimeout(ctx)
 	defer cancel()
 

cmd/main.go

@@ -165,7 +165,7 @@ func init() {
 	f.BoolVar(&operatorOptions.enableStorage, "operator.storage", false, "Enable to run the ArangoLocalStorage operator")
 	f.BoolVar(&operatorOptions.enableBackup, "operator.backup", false, "Enable to run the ArangoBackup operator")
 	f.BoolVar(&operatorOptions.enableApps, "operator.apps", false, "Enable to run the ArangoApps operator")
-	f.BoolVar(&operatorOptions.enableK2KClusterSync, "operator.k2k-cluster-sync", false, "Enable to run the ArangoClusterSynchronizations operator")
+	f.BoolVar(&operatorOptions.enableK2KClusterSync, "operator.k2k-cluster-sync", false, "Enable to run the ListSimple operator")
 	f.BoolVar(&operatorOptions.versionOnly, "operator.version", false, "Enable only version endpoint in Operator")
 	f.StringVar(&operatorOptions.alpineImage, "operator.alpine-image", UBIImageEnv.GetOrDefault(defaultAlpineImage), "Docker image used for alpine containers")
 	f.MarkDeprecated("operator.alpine-image", "Value is not used anymore")
@@ -226,10 +226,12 @@ func executeMain(cmd *cobra.Command, args []string) {
 
 	// Prepare log service
 	var err error
-	logService, err = logging.NewService(defaultLogLevel, logLevels)
-	if err != nil {
+	if err := logging.InitGlobalLogger(defaultLogLevel, logLevels); err != nil {
 		cliLog.Fatal().Err(err).Msg("Failed to initialize log service")
 	}
+
+	logService = logging.GlobalLogger()
+
 	logService.ConfigureRootLogger(func(log zerolog.Logger) zerolog.Logger {
 		podNameParts := strings.Split(name, "-")
 		operatorID := podNameParts[len(podNameParts)-1]

cmd/storage.go

@@ -62,13 +62,13 @@ func init() {
 
 // Run the provisioner
 func cmdStorageProvisionerRun(cmd *cobra.Command, args []string) {
-	//goflag.CommandLine.Parse([]string{"-logtostderr"})
 	var err error
-	logService, err = logging.NewService(defaultLogLevel, logLevels)
-	if err != nil {
+	if err := logging.InitGlobalLogger(defaultLogLevel, logLevels); err != nil {
 		cliLog.Fatal().Err(err).Msg("Failed to initialize log service")
 	}
 
+	logService = logging.GlobalLogger()
+
 	// Log version
 
 	cliLog.Info().Msgf("Starting arangodb local storage provisioner (%s), version %s build %s", version.GetVersionV1().Edition.Title(), version.GetVersionV1().Version, version.GetVersionV1().Build)

go.mod

@@ -28,6 +28,7 @@ require (
 	github.com/arangodb/go-driver v1.2.1
 	github.com/arangodb/go-driver/v2 v2.0.0-20211021031401-d92dcd5a4c83
 	github.com/arangodb/go-upgrade-rules v0.0.0-20180809110947-031b4774ff21
+	github.com/arangodb/rebalancer v0.1.1
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/dchest/uniuri v0.0.0-20160212164326-8902c56451e9
 	github.com/ghodss/yaml v1.0.0

pkg/apis/deployment/definitions.go

@@ -33,6 +33,10 @@ const (
 	ArangoClusterSynchronizationResourceKind   = "ArangoClusterSynchronization"
 	ArangoClusterSynchronizationResourcePlural = "arangoclustersynchronizations"
 
+	ArangoTaskCRDName        = ArangoTaskResourcePlural + "." + ArangoDeploymentGroupName
+	ArangoTaskResourceKind   = "ArangoTask"
+	ArangoTaskResourcePlural = "arangotasks"
+
 	ArangoDeploymentGroupName = "database.arangodb.com"
 )
 

pkg/deployment/access_package.go

@@ -49,7 +49,6 @@ const (
 func (d *Deployment) createAccessPackages(ctx context.Context) error {
 	log := d.deps.Log
 	spec := d.apiObject.Spec
-	secrets := d.deps.Client.Kubernetes().CoreV1().Secrets(d.GetNamespace())
 
 	if !spec.Sync.IsEnabled() {
 		// We're only relevant when sync is enabled
@@ -66,21 +65,15 @@ func (d *Deployment) createAccessPackages(ctx context.Context) error {
 	}
 
 	// Remove all access packages that we did build, but are no longer needed
-	ctxChild, cancel := globals.GetGlobalTimeouts().Kubernetes().WithTimeout(ctx)
-	defer cancel()
-	secretList, err := secrets.List(ctxChild, metav1.ListOptions{})
-	if err != nil {
-		log.Debug().Err(err).Msg("Failed to list secrets")
-		return errors.WithStack(err)
-	}
-	for _, secret := range secretList.Items {
-		if d.isOwnerOf(&secret) {
+	secretList := d.currentState.Secret().V1().ListSimple()
+	for _, secret := range secretList {
+		if d.isOwnerOf(secret) {
 			if _, found := secret.Data[constants.SecretAccessPackageYaml]; found {
 				// Secret is an access package
 				if _, wanted := apNameMap[secret.GetName()]; !wanted {
 					// We found an obsolete access package secret. Remove it.
-					err = globals.GetGlobalTimeouts().Kubernetes().RunWithTimeout(ctx, func(ctxChild context.Context) error {
-						return secrets.Delete(ctxChild, secret.GetName(), metav1.DeleteOptions{
+					err := globals.GetGlobalTimeouts().Kubernetes().RunWithTimeout(ctx, func(ctxChild context.Context) error {
+						return d.SecretsModInterface().Delete(ctxChild, secret.GetName(), metav1.DeleteOptions{
 							Preconditions: &metav1.Preconditions{UID: &secret.UID},
 						})
 					})
@@ -105,14 +98,9 @@ func (d *Deployment) createAccessPackages(ctx context.Context) error {
 // it is does not already exist.
 func (d *Deployment) ensureAccessPackage(ctx context.Context, apSecretName string) error {
 	log := d.deps.Log
-	ns := d.GetNamespace()
-	secrets := d.deps.Client.Kubernetes().CoreV1().Secrets(ns)
 	spec := d.apiObject.Spec
 
-	err := globals.GetGlobalTimeouts().Kubernetes().RunWithTimeout(ctx, func(ctxChild context.Context) error {
-		_, err := secrets.Get(ctxChild, apSecretName, metav1.GetOptions{})
-		return err
-	})
+	_, err := d.currentState.Secret().V1().Read().Get(ctx, apSecretName, metav1.GetOptions{})
 	if err == nil {
 		// Secret already exists
 		return nil
@@ -123,17 +111,15 @@ func (d *Deployment) ensureAccessPackage(ctx context.Context, apSecretName strin
 
 	// Fetch client authentication CA
 	clientAuthSecretName := spec.Sync.Authentication.GetClientCASecretName()
-	ctxChild, cancel := globals.GetGlobalTimeouts().Kubernetes().WithTimeout(ctx)
-	defer cancel()
-	clientAuthCert, clientAuthKey, _, err := k8sutil.GetCASecret(ctxChild, secrets, clientAuthSecretName, nil)
+	clientAuthCert, clientAuthKey, _, err := k8sutil.GetCASecret(ctx, d.currentState.Secret().V1().Read(), clientAuthSecretName, nil)
 	if err != nil {
 		log.Debug().Err(err).Msg("Failed to get client-auth CA secret")
 		return errors.WithStack(err)
 	}
 
 	// Fetch TLS CA public key
 	tlsCASecretName := spec.Sync.TLS.GetCASecretName()
-	tlsCACert, err := k8sutil.GetCACertficateSecret(ctx, secrets, tlsCASecretName)
+	tlsCACert, err := k8sutil.GetCACertficateSecret(ctx, d.currentState.Secret().V1().Read(), tlsCASecretName)
 	if err != nil {
 		log.Debug().Err(err).Msg("Failed to get TLS CA secret")
 		return errors.WithStack(err)
@@ -220,7 +206,7 @@ func (d *Deployment) ensureAccessPackage(ctx context.Context, apSecretName strin
 	// Attach secret to owner
 	secret.SetOwnerReferences(append(secret.GetOwnerReferences(), d.apiObject.AsOwner()))
 	err = globals.GetGlobalTimeouts().Kubernetes().RunWithTimeout(ctx, func(ctxChild context.Context) error {
-		_, err := secrets.Create(ctxChild, secret, metav1.CreateOptions{})
+		_, err := d.SecretsModInterface().Create(ctxChild, secret, metav1.CreateOptions{})
 		return err
 	})
 	if err != nil {

pkg/deployment/cleanup.go

@@ -28,11 +28,12 @@ import (
 	core "k8s.io/api/core/v1"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	"github.com/arangodb/kube-arangodb/pkg/deployment/resources/inspector"
 	"github.com/arangodb/kube-arangodb/pkg/util"
 	"github.com/arangodb/kube-arangodb/pkg/util/constants"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
 	inspectorInterface "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector"
+	pvcv1 "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/persistentvolumeclaim/v1"
+	podv1 "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/pod/v1"
 )
 
 // removePodFinalizers removes all finalizers from all pods owned by us.
@@ -41,7 +42,7 @@ func (d *Deployment) removePodFinalizers(ctx context.Context, cachedStatus inspe
 
 	found := false
 
-	if err := cachedStatus.IteratePods(func(pod *core.Pod) error {
+	if err := cachedStatus.Pod().V1().Iterate(func(pod *core.Pod) error {
 		log.Info().Str("pod", pod.GetName()).Msgf("Removing Pod Finalizer")
 		if count, err := k8sutil.RemovePodFinalizers(ctx, cachedStatus, log, d.PodsModInterface(), pod, constants.ManagedFinalizers(), true); err != nil {
 			log.Warn().Err(err).Msg("Failed to remove pod finalizers")
@@ -62,7 +63,7 @@ func (d *Deployment) removePodFinalizers(ctx context.Context, cachedStatus inspe
 			}
 		}
 		return nil
-	}, inspector.FilterPodsByLabels(k8sutil.LabelsForDeployment(d.GetName(), ""))); err != nil {
+	}, podv1.FilterPodsByLabels(k8sutil.LabelsForDeployment(d.GetName(), ""))); err != nil {
 		return false, err
 	}
 
@@ -75,7 +76,7 @@ func (d *Deployment) removePVCFinalizers(ctx context.Context, cachedStatus inspe
 
 	found := false
 
-	if err := cachedStatus.IteratePersistentVolumeClaims(func(pvc *core.PersistentVolumeClaim) error {
+	if err := cachedStatus.PersistentVolumeClaim().V1().Iterate(func(pvc *core.PersistentVolumeClaim) error {
 		log.Info().Str("pvc", pvc.GetName()).Msgf("Removing PVC Finalizer")
 		if count, err := k8sutil.RemovePVCFinalizers(ctx, cachedStatus, log, d.PersistentVolumeClaimsModInterface(), pvc, constants.ManagedFinalizers(), true); err != nil {
 			log.Warn().Err(err).Msg("Failed to remove PVC finalizers")
@@ -84,7 +85,7 @@ func (d *Deployment) removePVCFinalizers(ctx context.Context, cachedStatus inspe
 			found = true
 		}
 		return nil
-	}, inspector.FilterPersistentVolumeClaimsByLabels(k8sutil.LabelsForDeployment(d.GetName(), ""))); err != nil {
+	}, pvcv1.FilterPersistentVolumeClaimsByLabels(k8sutil.LabelsForDeployment(d.GetName(), ""))); err != nil {
 		return false, err
 	}