Scheme *runtime.Scheme

client.Client

ClusterID liqov1beta1.ClusterID

// RegisteredResources is a list of GVRs of resources to be replicated.

RegisteredResources []resources.Resource

// ReflectionManager is the object managing the reflection towards remote clusters.

ReflectionManager *reflection.Manager

// Reflectors is a map containing the reflectors towards each remote cluster.

Reflectors map[liqov1beta1.ClusterID]*reflection.Reflector

// IdentityReader is an interface to manage remote identities, and to get the rest config.

IdentityReader identitymanager.IdentityReader

var prefix string

tracer := trace.New("Reconcile", trace.Field{Key: "Secret", Value: req.Name})

defer tracer.LogIfLong(traceutils.LongThreshold())

var secret corev1.Secret

if err := c.Get(ctx, req.NamespacedName, &secret); err != nil {

if errors.IsNotFound(err) {

klog.V(4).Infof("secret %q not found", req.NamespacedName)

return ctrl.Result{}, nil

}

klog.Errorf("unable to get secret %q: %v", req.NamespacedName, err)

return ctrl.Result{}, err

}

klog.Infof("Processing Secret %q", req.NamespacedName)

// Validate the secret to ensure it is a control plane identity and it has all the required labels and annotations.

if err := c.validateSecret(&secret); err != nil {

klog.Errorf("secret %q is not valid: %v", req.NamespacedName, err)

return ctrl.Result{}, nil

}

// Extract remote cluster informations from the secret

remoteClusterID := liqov1beta1.ClusterID(secret.Labels[consts.RemoteClusterID])

localTenantNamespace := secret.Namespace

remoteTenantNamespace := secret.Annotations[consts.RemoteTenantNamespaceAnnotKey]

// examine DeletionTimestamp to determine if object is under deletion

if !secret.ObjectMeta.DeletionTimestamp.IsZero() {

// the object is being deleted

if controllerutil.ContainsFinalizer(&secret, finalizer) {

// close remote watcher for remote cluster

if err := c.stopReflector(remoteClusterID, false); err != nil {

klog.Errorf("%sFailed to stop reflection: %v", prefix, err)

return ctrl.Result{}, err

}

// remove the finalizer from the list and update it.

if err := c.ensureFinalizer(ctx, &secret, controllerutil.RemoveFinalizer); err != nil {

klog.Errorf("An error occurred while removing the finalizer to %q: %v", req.NamespacedName, err)

return ctrl.Result{}, err

}

return ctrl.Result{}, nil

}

}

// Defer the function to start/stop the reflection of the different resources.

defer func() {

if err == nil {

err = c.enforceReflectionStatus(ctx, remoteClusterID, !secret.GetDeletionTimestamp().IsZero())

}

}()

// Add the finalizer to ensure the reflection is correctly stopped

if err := c.ensureFinalizer(ctx, &secret, controllerutil.AddFinalizer); err != nil {

klog.Errorf("An error occurred while adding the finalizer to %q: %v", req.NamespacedName, err)

return ctrl.Result{}, err

}

// Check if reflection towards the remote cluster has already been started.

if reflector, found := c.Reflectors[remoteClusterID]; found {

// We ignore the case in which the secret lacks of the kubeconfig, as in that case we still want to delete the reflector

// and manage the error.

secretContent := secret.Data[consts.KubeconfigSecretField]

secretHash := c.hashSecretConfig(secretContent)

// If there are no changes on the secret or on the remote namespace where the reflector operats, skip reconciliation.

if reflector.GetSecretHash() == secretHash && reflector.GetRemoteTenantNamespace() == remoteTenantNamespace {

return ctrl.Result{}, nil

}

// If there have been a change on the secret, delete the secret to allow the creation of a new reflector.

klog.Infof("%sChanges detected on the control plane secret %q for clusterID %q: recreating reflector",

prefix, req.NamespacedName, remoteClusterID)

// Stop the reflection to update the reflector

if err := c.stopReflector(remoteClusterID, true); err != nil {

klog.Errorf("%sFailed to stop reflection: %v", prefix, err)

return ctrl.Result{}, err

}

}

// We need to get the secret to make sure that there are not multiple secrets pointing to the same cluster

currSecret, err := getters.GetControlPlaneKubeconfigSecretByClusterID(ctx, c.Client, remoteClusterID)

if err != nil {

klog.Errorf("%sUnable to process secret for clusterID %q: %v", prefix, remoteClusterID, err)

return ctrl.Result{}, nil

}

config, err := c.IdentityReader.GetConfigFromSecret(remoteClusterID, currSecret)

if err != nil {

klog.Errorf("%sUnable to retrieve config for clusterID %q: %v", prefix, remoteClusterID, err)

return ctrl.Result{}, nil

}

return ctrl.Result{}, c.setupReflectionToPeeringCluster(ctx, currSecret, config, remoteClusterID, localTenantNamespace, remoteTenantNamespace)

resourceToBeProccesedPredicate := predicate.Funcs{

DeleteFunc: func(_ event.DeleteEvent) bool {

return false

},

}

secretsFilter, err := predicate.LabelSelectorPredicate(c.controlPlaneIdentitySecretLabelSelector())

if err != nil {

return err

}

return ctrl.NewControllerManagedBy(mgr).Named(consts.CtrlSecretCRDReplicator).

For(&corev1.Secret{}, builder.WithPredicates(secretsFilter)).

WithEventFilter(resourceToBeProccesedPredicate).

Complete(c)

return metav1.LabelSelector{

MatchExpressions: []metav1.LabelSelectorRequirement{

{

Key: consts.RemoteClusterID,

Operator: metav1.LabelSelectorOpExists,

},

{

Key: consts.IdentityTypeLabelKey,

Operator: metav1.LabelSelectorOpIn,

Values: []string{string(authv1beta1.ControlPlaneIdentityType)},

},

},

}

nsName := client.ObjectKeyFromObject(secret)

if secret.Labels == nil || secret.Annotations == nil {

return fmt.Errorf("secret %q does not have the required labels and annotations", nsName)

}

if idType, ok := secret.Labels[consts.IdentityTypeLabelKey]; ok && idType != string(authv1beta1.ControlPlaneIdentityType) {

return fmt.Errorf("secret %q does not contain a control plane identity", nsName)

}

if _, ok := secret.Labels[consts.RemoteClusterID]; !ok {

return fmt.Errorf("secret %q does not have the remote cluster ID label %s", nsName, consts.RemoteClusterID)

}

if _, ok := secret.Annotations[consts.RemoteTenantNamespaceAnnotKey]; !ok {

return fmt.Errorf("secret %q does not have the remote tenant namespace annotation key %s", nsName, consts.RemoteTenantNamespaceAnnotKey)

}

return nil

updater func(client.Object, string) bool) error {

// Do not perform any action if the finalizer is already as expected

if !updater(secret, finalizer) {

return nil

}

return c.Client.Update(ctx, secret)

remoteClusterID liqov1beta1.ClusterID, localNamespace, remoteNamespace string) error {

dynamicClient, err := dynamic.NewForConfig(config)

if err != nil {

klog.Errorf("%sUnable to create dynamic client for remote cluster: %v", remoteClusterID, err)

return err

}

secretHash := c.hashSecretConfig(secret.Data[consts.KubeconfigSecretField])

reflector := c.ReflectionManager.NewForRemote(dynamicClient, remoteClusterID, localNamespace, remoteNamespace, secretHash)

reflector.Start(ctx)

c.Reflectors[remoteClusterID] = reflector

return nil

reflector, ok := c.Reflectors[remoteClusterID]

if ok {

stopFn := reflector.Stop

// Use the StopForce function if we want to skip the checks

if skipChecks {

stopFn = reflector.StopForce

}

if err := stopFn(); err != nil {

return err

}

delete(c.Reflectors, remoteClusterID)

}

return nil

reflector, found := c.Reflectors[remoteClusterID]

if !found {

// The reflector object has not yet been setup

return nil

}

for i := range c.RegisteredResources {

res := &c.RegisteredResources[i]

if !deleting && !reflector.ResourceStarted(res) {

reflector.StartForResource(ctx, res)

} else if deleting && reflector.ResourceStarted(res) {

if err := reflector.StopForResource(res); err != nil {

return err

}

}

}

return nil