Add resources sync configuration (#431)

* Add resources sync configuration

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* update docs

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* wsl

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* wsl

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* refactor cluster sync

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* wsl

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* more fixes

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* more fixes

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* simplify the syncerContext

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* simplify the syncerContext

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* drop the ClusterClient struct

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* Fix updates to syncer

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* fix

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* lint

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* refactor secrets/configmaps sync

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* refactor secrets/configmaps sync

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* docs

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* Add imagepullsecret translation

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* fix test

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* add exception for deleted resources

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* linting fixes

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* remove the option to disable imagepullsecret translation

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* fix

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

---------

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>
This commit is contained in:
Hussein Galal
2025-09-01 14:34:29 +03:00
committed by GitHub
parent b3f7a8ab7f
commit caf785f23b
28 changed files with 2651 additions and 919 deletions
+102
View File
@@ -574,6 +574,108 @@ spec:
x-kubernetes-validations:
- message: serviceCIDR is immutable
rule: self == oldSelf
sync:
default: {}
description: Sync specifies the resources types that will be synced
from virtual cluster to host cluster.
properties:
configmaps:
default:
enabled: true
description: ConfigMaps resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
ingresses:
default:
enabled: false
description: Ingresses resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
persistentVolumeClaims:
default:
enabled: false
description: PersistentVolumeClaims resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
priorityClasses:
default:
enabled: false
description: PriorityClasses resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
secrets:
default:
enabled: true
description: Secrets resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
services:
default:
enabled: true
description: Services resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
type: object
tlsSANs:
description: TLSSANs specifies subject alternative names for the K3s
server certificate.
@@ -225,6 +225,107 @@ spec:
type: array
x-kubernetes-list-type: atomic
type: object
sync:
description: Sync specifies the synchronization configuration for
resources that can be applied to clusters
properties:
configmaps:
default:
enabled: true
description: ConfigMaps resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
ingresses:
default:
enabled: false
description: Ingresses resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
persistentVolumeClaims:
default:
enabled: false
description: PersistentVolumeClaims resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
priorityClasses:
default:
enabled: false
description: PriorityClasses resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
secrets:
default:
enabled: true
description: Secrets resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
services:
default:
enabled: true
description: Services resources sync configuration.
properties:
enabled:
description: Enabled is an on/off switch for syncing resources.
type: boolean
selector:
additionalProperties:
type: string
description: |-
Selector specifies set of labels of the resources that will be synced, if empty
then all resources of the given type will be synced.
type: object
type: object
type: object
type: object
status:
description: Status reflects the observed state of the VirtualClusterPolicy.
+11
View File
@@ -195,6 +195,17 @@ func newCluster(name, namespace string, config *CreateConfig) *v1alpha1.Cluster
StorageClassName: ptr.To(config.storageClassName),
StorageRequestSize: config.storageRequestSize,
},
Sync: v1alpha1.SyncConfig{
ConfigMaps: v1alpha1.ConfigMapSyncConfig{
Enabled: true,
},
Secrets: v1alpha1.SecretSyncConfig{
Enabled: true,
},
Services: v1alpha1.ServiceSyncConfig{
Enabled: true,
},
},
MirrorHostNodes: config.mirrorHostNodes,
},
}
+126
View File
@@ -134,10 +134,28 @@ _Appears in:_
| `workerLimit` _[ResourceList](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.31/#resourcelist-v1-core)_ | WorkerLimit specifies resource limits for agent nodes. | | |
| `mirrorHostNodes` _boolean_ | MirrorHostNodes controls whether node objects from the host cluster<br />are mirrored into the virtual cluster. | | |
| `customCAs` _[CustomCAs](#customcas)_ | CustomCAs specifies the cert/key pairs for custom CA certificates. | | |
| `sync` _[SyncConfig](#syncconfig)_ | Sync specifies the resources types that will be synced from virtual cluster to host cluster. | \{ \} | |
#### ConfigMapSyncConfig
ConfigMapSyncConfig specifies the sync options for services.
_Appears in:_
- [SyncConfig](#syncconfig)
| Field | Description | Default | Validation |
| --- | --- | --- | --- |
| `enabled` _boolean_ | Enabled is an on/off switch for syncing resources. | | |
| `selector` _object (keys:string, values:string)_ | Selector specifies set of labels of the resources that will be synced, if empty<br />then all resources of the given type will be synced. | | |
#### CredentialSource
@@ -229,6 +247,23 @@ _Appears in:_
| `ingressClassName` _string_ | IngressClassName specifies the IngressClass to use for the Ingress. | | |
#### IngressSyncConfig
IngressSyncConfig specifies the sync options for services.
_Appears in:_
- [SyncConfig](#syncconfig)
| Field | Description | Default | Validation |
| --- | --- | --- | --- |
| `enabled` _boolean_ | Enabled is an on/off switch for syncing resources. | | |
| `selector` _object (keys:string, values:string)_ | Selector specifies set of labels of the resources that will be synced, if empty<br />then all resources of the given type will be synced. | | |
#### LoadBalancerConfig
@@ -294,6 +329,23 @@ _Appears in:_
#### PersistentVolumeClaimSyncConfig
PersistentVolumeClaimSyncConfig specifies the sync options for services.
_Appears in:_
- [SyncConfig](#syncconfig)
| Field | Description | Default | Validation |
| --- | --- | --- | --- |
| `enabled` _boolean_ | Enabled is an on/off switch for syncing resources. | | |
| `selector` _object (keys:string, values:string)_ | Selector specifies set of labels of the resources that will be synced, if empty<br />then all resources of the given type will be synced. | | |
#### PodSecurityAdmissionLevel
_Underlying type:_ _string_
@@ -308,6 +360,79 @@ _Appears in:_
#### PriorityClassSyncConfig
PriorityClassSyncConfig specifies the sync options for services.
_Appears in:_
- [SyncConfig](#syncconfig)
| Field | Description | Default | Validation |
| --- | --- | --- | --- |
| `enabled` _boolean_ | Enabled is an on/off switch for syncing resources. | | |
| `selector` _object (keys:string, values:string)_ | Selector specifies set of labels of the resources that will be synced, if empty<br />then all resources of the given type will be synced. | | |
#### SecretSyncConfig
SecretSyncConfig specifies the sync options for services.
_Appears in:_
- [SyncConfig](#syncconfig)
| Field | Description | Default | Validation |
| --- | --- | --- | --- |
| `enabled` _boolean_ | Enabled is an on/off switch for syncing resources. | | |
| `selector` _object (keys:string, values:string)_ | Selector specifies set of labels of the resources that will be synced, if empty<br />then all resources of the given type will be synced. | | |
#### ServiceSyncConfig
ServiceSyncConfig specifies the sync options for services.
_Appears in:_
- [SyncConfig](#syncconfig)
| Field | Description | Default | Validation |
| --- | --- | --- | --- |
| `enabled` _boolean_ | Enabled is an on/off switch for syncing resources. | | |
| `selector` _object (keys:string, values:string)_ | Selector specifies set of labels of the resources that will be synced, if empty<br />then all resources of the given type will be synced. | | |
#### SyncConfig
SyncConfig will contain the resources that should be synced from virtual cluster to host cluster.
_Appears in:_
- [ClusterSpec](#clusterspec)
- [VirtualClusterPolicySpec](#virtualclusterpolicyspec)
| Field | Description | Default | Validation |
| --- | --- | --- | --- |
| `services` _[ServiceSyncConfig](#servicesyncconfig)_ | Services resources sync configuration. | \{ enabled:true \} | |
| `configmaps` _[ConfigMapSyncConfig](#configmapsyncconfig)_ | ConfigMaps resources sync configuration. | \{ enabled:true \} | |
| `secrets` _[SecretSyncConfig](#secretsyncconfig)_ | Secrets resources sync configuration. | \{ enabled:true \} | |
| `ingresses` _[IngressSyncConfig](#ingresssyncconfig)_ | Ingresses resources sync configuration. | \{ enabled:false \} | |
| `persistentVolumeClaims` _[PersistentVolumeClaimSyncConfig](#persistentvolumeclaimsyncconfig)_ | PersistentVolumeClaims resources sync configuration. | \{ enabled:false \} | |
| `priorityClasses` _[PriorityClassSyncConfig](#priorityclasssyncconfig)_ | PriorityClasses resources sync configuration. | \{ enabled:false \} | |
#### VirtualClusterPolicy
@@ -366,6 +491,7 @@ _Appears in:_
| `allowedMode` _[ClusterMode](#clustermode)_ | AllowedMode specifies the allowed cluster provisioning mode. Defaults to "shared". | shared | Enum: [shared virtual] <br /> |
| `disableNetworkPolicy` _boolean_ | DisableNetworkPolicy indicates whether to disable the creation of a default network policy for cluster isolation. | | |
| `podSecurityAdmissionLevel` _[PodSecurityAdmissionLevel](#podsecurityadmissionlevel)_ | PodSecurityAdmissionLevel specifies the pod security admission level applied to the pods in the namespace. | | Enum: [privileged baseline restricted] <br /> |
| `sync` _[SyncConfig](#syncconfig)_ | Sync specifies the synchronization configuration for resources that can be applied to clusters | | |
-198
View File
@@ -1,198 +0,0 @@
package controller
import (
"context"
"fmt"
"sync"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/client-go/util/retry"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"github.com/rancher/k3k/pkg/controller"
k3klog "github.com/rancher/k3k/pkg/log"
)
const ConfigMapSyncerName = "configmap-syncer"
type ConfigMapSyncer struct {
mutex sync.RWMutex
// VirtualClient is the client for the virtual cluster
VirtualClient client.Client
// CoreClient is the client for the host cluster
HostClient client.Client
// TranslateFunc is the function that translates a given resource from it's virtual representation to the host
// representation
TranslateFunc func(*corev1.ConfigMap) (*corev1.ConfigMap, error)
// Logger is the logger that the controller will use
Logger *k3klog.Logger
// objs are the objects that the syncer should watch/syncronize. Should only be manipulated
// through add/remove
objs sets.Set[types.NamespacedName]
}
func (c *ConfigMapSyncer) Name() string {
return ConfigMapSyncerName
}
// Reconcile implements reconcile.Reconciler and synchronizes the objects in objs to the host cluster
func (c *ConfigMapSyncer) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
if !c.isWatching(req.NamespacedName) {
// return immediately without re-enqueueing. We aren't watching this resource
return reconcile.Result{}, nil
}
var virtual corev1.ConfigMap
if err := c.VirtualClient.Get(ctx, req.NamespacedName, &virtual); err != nil {
return reconcile.Result{
Requeue: true,
}, fmt.Errorf("unable to get configmap %s/%s from virtual cluster: %w", req.Namespace, req.Name, err)
}
translated, err := c.TranslateFunc(&virtual)
if err != nil {
return reconcile.Result{
Requeue: true,
}, fmt.Errorf("unable to translate configmap %s/%s from virtual cluster: %w", req.Namespace, req.Name, err)
}
translatedKey := types.NamespacedName{
Namespace: translated.Namespace,
Name: translated.Name,
}
var host corev1.ConfigMap
if err = c.HostClient.Get(ctx, translatedKey, &host); err != nil {
if apierrors.IsNotFound(err) {
err = c.HostClient.Create(ctx, translated)
// for simplicity's sake, we don't check for conflict errors. The existing object will get
// picked up on in the next re-enqueue
return reconcile.Result{
Requeue: true,
}, fmt.Errorf("unable to create host configmap %s/%s for virtual configmap %s/%s: %w",
translated.Namespace, translated.Name, req.Namespace, req.Name, err)
}
return reconcile.Result{Requeue: true}, fmt.Errorf("unable to get host configmap %s/%s: %w", translated.Namespace, translated.Name, err)
}
// we are going to use the host in order to avoid conflicts on update
host.Data = translated.Data
if host.Labels == nil {
host.Labels = make(map[string]string, len(translated.Labels))
}
// we don't want to override labels made on the host cluster by other applications
// but we do need to make sure the labels that the kubelet uses to track host cluster values
// are being tracked appropriately
for key, value := range translated.Labels {
host.Labels[key] = value
}
if err = c.HostClient.Update(ctx, &host); err != nil {
return reconcile.Result{
Requeue: true,
}, fmt.Errorf("unable to update host configmap %s/%s for virtual configmap %s/%s: %w",
translated.Namespace, translated.Name, req.Namespace, req.Name, err)
}
return reconcile.Result{}, nil
}
// isWatching is a utility method to determine if a key is in objs without the caller needing
// to handle mutex lock/unlock.
func (c *ConfigMapSyncer) isWatching(key types.NamespacedName) bool {
c.mutex.RLock()
defer c.mutex.RUnlock()
return c.objs.Has(key)
}
// AddResource adds a given resource to the list of resources that will be synced. Safe to call multiple times for the
// same resource.
func (c *ConfigMapSyncer) AddResource(ctx context.Context, namespace, name string) error {
objKey := types.NamespacedName{
Namespace: namespace,
Name: name,
}
// if we already sync this object, no need to writelock/add it
if c.isWatching(objKey) {
return nil
}
// lock in write mode since we are now adding the key
c.mutex.Lock()
if c.objs == nil {
c.objs = sets.Set[types.NamespacedName]{}
}
c.objs = c.objs.Insert(objKey)
c.mutex.Unlock()
_, err := c.Reconcile(ctx, reconcile.Request{
NamespacedName: objKey,
})
if err != nil {
return fmt.Errorf("unable to reconcile new object %s/%s: %w", objKey.Namespace, objKey.Name, err)
}
return nil
}
// RemoveResource removes a given resource from the list of resources that will be synced. Safe to call for an already
// removed resource.
func (c *ConfigMapSyncer) RemoveResource(ctx context.Context, namespace, name string) error {
objKey := types.NamespacedName{
Namespace: namespace,
Name: name,
}
// if we don't sync this object, no need to writelock/add it
if !c.isWatching(objKey) {
return nil
}
if err := retry.OnError(controller.Backoff, func(err error) bool {
return err != nil
}, func() error {
return c.removeHostConfigMap(ctx, namespace, name)
}); err != nil {
return fmt.Errorf("unable to remove configmap: %w", err)
}
c.mutex.Lock()
if c.objs == nil {
c.objs = sets.Set[types.NamespacedName]{}
}
c.objs = c.objs.Delete(objKey)
c.mutex.Unlock()
return nil
}
func (c *ConfigMapSyncer) removeHostConfigMap(ctx context.Context, virtualNamespace, virtualName string) error {
var vConfigMap corev1.ConfigMap
key := types.NamespacedName{
Namespace: virtualNamespace,
Name: virtualName,
}
if err := c.VirtualClient.Get(ctx, key, &vConfigMap); err != nil {
return fmt.Errorf("unable to get virtual configmap %s/%s: %w", virtualNamespace, virtualName, err)
}
translated, err := c.TranslateFunc(&vConfigMap)
if err != nil {
return fmt.Errorf("unable to translate virtual secret: %s/%s: %w", virtualNamespace, virtualName, err)
}
return c.HostClient.Delete(ctx, translated)
}
-135
View File
@@ -1,135 +0,0 @@
package controller
import (
"context"
"fmt"
"sync"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
v1 "k8s.io/api/core/v1"
ctrl "sigs.k8s.io/controller-runtime"
"github.com/rancher/k3k/k3k-kubelet/translate"
k3klog "github.com/rancher/k3k/pkg/log"
)
type ControllerHandler struct {
sync.RWMutex
// Mgr is the manager used to run new controllers - from the virtual cluster
Mgr manager.Manager
// Scheme is the scheme used to run new controllers - from the virtual cluster
Scheme runtime.Scheme
// HostClient is the client used to communicate with the host cluster
HostClient client.Client
// VirtualClient is the client used to communicate with the virtual cluster
VirtualClient client.Client
// Translator is the translator that will be used to adjust objects before they
// are made on the host cluster
Translator translate.ToHostTranslator
// Logger is the logger that the controller will use to log errors
Logger *k3klog.Logger
// controllers are the controllers which are currently running
controllers map[schema.GroupVersionKind]updateableReconciler
}
// updateableReconciler is a reconciler that only syncs specific resources (by name/namespace). This list can
// be altered through the Add and Remove methods
type updateableReconciler interface {
reconcile.Reconciler
Name() string
AddResource(ctx context.Context, namespace string, name string) error
RemoveResource(ctx context.Context, namespace string, name string) error
}
func (c *ControllerHandler) AddResource(ctx context.Context, obj client.Object) error {
c.RLock()
controllers := c.controllers
if controllers != nil {
if r, ok := c.controllers[obj.GetObjectKind().GroupVersionKind()]; ok {
err := r.AddResource(ctx, obj.GetNamespace(), obj.GetName())
c.RUnlock()
return err
}
}
// we need to manually lock/unlock since we intned on write locking to add a new controller
c.RUnlock()
var r updateableReconciler
switch obj.(type) {
case *v1.Secret:
r = &SecretSyncer{
HostClient: c.HostClient,
VirtualClient: c.VirtualClient,
// TODO: Need actual function
TranslateFunc: func(s *v1.Secret) (*v1.Secret, error) {
// note that this doesn't do any type safety - fix this
// when generics work
c.Translator.TranslateTo(s)
// Remove service-account-token types when synced to the host
if s.Type == v1.SecretTypeServiceAccountToken {
s.Type = v1.SecretTypeOpaque
}
return s, nil
},
Logger: c.Logger,
}
case *v1.ConfigMap:
r = &ConfigMapSyncer{
HostClient: c.HostClient,
VirtualClient: c.VirtualClient,
// TODO: Need actual function
TranslateFunc: func(s *v1.ConfigMap) (*v1.ConfigMap, error) {
c.Translator.TranslateTo(s)
return s, nil
},
Logger: c.Logger,
}
default:
// TODO: Technically, the configmap/secret syncers are relatively generic, and this
// logic could be used for other types.
return fmt.Errorf("unrecognized type: %T", obj)
}
err := ctrl.NewControllerManagedBy(c.Mgr).
Named(r.Name()).
For(&v1.ConfigMap{}).
Complete(r)
if err != nil {
return fmt.Errorf("unable to start configmap controller: %w", err)
}
c.Lock()
if c.controllers == nil {
c.controllers = map[schema.GroupVersionKind]updateableReconciler{}
}
c.controllers[obj.GetObjectKind().GroupVersionKind()] = r
c.Unlock()
return r.AddResource(ctx, obj.GetNamespace(), obj.GetName())
}
func (c *ControllerHandler) RemoveResource(ctx context.Context, obj client.Object) error {
// since we aren't adding a new controller, we don't need to lock
c.RLock()
ctrl, ok := c.controllers[obj.GetObjectKind().GroupVersionKind()]
c.RUnlock()
if !ok {
return fmt.Errorf("no controller found for gvk %s", obj.GetObjectKind().GroupVersionKind())
}
return ctrl.RemoveResource(ctx, obj.GetNamespace(), obj.GetName())
}
-197
View File
@@ -1,197 +0,0 @@
package controller
import (
"context"
"fmt"
"sync"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/client-go/util/retry"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"github.com/rancher/k3k/pkg/controller"
k3klog "github.com/rancher/k3k/pkg/log"
)
const SecretSyncerName = "secret-syncer"
type SecretSyncer struct {
mutex sync.RWMutex
// VirtualClient is the client for the virtual cluster
VirtualClient client.Client
// CoreClient is the client for the host cluster
HostClient client.Client
// TranslateFunc is the function that translates a given resource from it's virtual representation to the host
// representation
TranslateFunc func(*corev1.Secret) (*corev1.Secret, error)
// Logger is the logger that the controller will use
Logger *k3klog.Logger
// objs are the objects that the syncer should watch/syncronize. Should only be manipulated
// through add/remove
objs sets.Set[types.NamespacedName]
}
func (s *SecretSyncer) Name() string {
return SecretSyncerName
}
// Reconcile implements reconcile.Reconciler and synchronizes the objects in objs to the host cluster
func (s *SecretSyncer) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
if !s.isWatching(req.NamespacedName) {
// return immediately without re-enqueueing. We aren't watching this resource
return reconcile.Result{}, nil
}
var virtual corev1.Secret
if err := s.VirtualClient.Get(ctx, req.NamespacedName, &virtual); err != nil {
return reconcile.Result{
Requeue: true,
}, fmt.Errorf("unable to get secret %s/%s from virtual cluster: %w", req.Namespace, req.Name, err)
}
translated, err := s.TranslateFunc(&virtual)
if err != nil {
return reconcile.Result{
Requeue: true,
}, fmt.Errorf("unable to translate secret %s/%s from virtual cluster: %w", req.Namespace, req.Name, err)
}
translatedKey := types.NamespacedName{
Namespace: translated.Namespace,
Name: translated.Name,
}
var host corev1.Secret
if err = s.HostClient.Get(ctx, translatedKey, &host); err != nil {
if apierrors.IsNotFound(err) {
err = s.HostClient.Create(ctx, translated)
// for simplicity's sake, we don't check for conflict errors. The existing object will get
// picked up on in the next re-enqueue
return reconcile.Result{
Requeue: true,
}, fmt.Errorf("unable to create host secret %s/%s for virtual secret %s/%s: %w",
translated.Namespace, translated.Name, req.Namespace, req.Name, err)
}
return reconcile.Result{Requeue: true}, fmt.Errorf("unable to get host secret %s/%s: %w", translated.Namespace, translated.Name, err)
}
// we are going to use the host in order to avoid conflicts on update
host.Data = translated.Data
if host.Labels == nil {
host.Labels = make(map[string]string, len(translated.Labels))
}
// we don't want to override labels made on the host cluster by other applications
// but we do need to make sure the labels that the kubelet uses to track host cluster values
// are being tracked appropriately
for key, value := range translated.Labels {
host.Labels[key] = value
}
if err = s.HostClient.Update(ctx, &host); err != nil {
return reconcile.Result{
Requeue: true,
}, fmt.Errorf("unable to update host secret %s/%s for virtual secret %s/%s: %w",
translated.Namespace, translated.Name, req.Namespace, req.Name, err)
}
return reconcile.Result{}, nil
}
// isWatching is a utility method to determine if a key is in objs without the caller needing
// to handle mutex lock/unlock.
func (s *SecretSyncer) isWatching(key types.NamespacedName) bool {
s.mutex.RLock()
defer s.mutex.RUnlock()
return s.objs.Has(key)
}
// AddResource adds a given resource to the list of resources that will be synced. Safe to call multiple times for the
// same resource.
func (s *SecretSyncer) AddResource(ctx context.Context, namespace, name string) error {
objKey := types.NamespacedName{
Namespace: namespace,
Name: name,
}
// if we already sync this object, no need to writelock/add it
if s.isWatching(objKey) {
return nil
}
// lock in write mode since we are now adding the key
s.mutex.Lock()
if s.objs == nil {
s.objs = sets.Set[types.NamespacedName]{}
}
s.objs = s.objs.Insert(objKey)
s.mutex.Unlock()
_, err := s.Reconcile(ctx, reconcile.Request{
NamespacedName: objKey,
})
if err != nil {
return fmt.Errorf("unable to reconcile new object %s/%s: %w", objKey.Namespace, objKey.Name, err)
}
return nil
}
// RemoveResource removes a given resource from the list of resources that will be synced. Safe to call for an already
// removed resource.
func (s *SecretSyncer) RemoveResource(ctx context.Context, namespace, name string) error {
objKey := types.NamespacedName{
Namespace: namespace,
Name: name,
}
// if we don't sync this object, no need to writelock/add it
if !s.isWatching(objKey) {
return nil
}
// lock in write mode since we are now adding the key
if err := retry.OnError(controller.Backoff, func(err error) bool {
return err != nil
}, func() error {
return s.removeHostSecret(ctx, namespace, name)
}); err != nil {
return fmt.Errorf("unable to remove secret: %w", err)
}
s.mutex.Lock()
if s.objs == nil {
s.objs = sets.Set[types.NamespacedName]{}
}
s.objs = s.objs.Delete(objKey)
s.mutex.Unlock()
return nil
}
func (s *SecretSyncer) removeHostSecret(ctx context.Context, virtualNamespace, virtualName string) error {
var vSecret corev1.Secret
err := s.VirtualClient.Get(ctx, types.NamespacedName{
Namespace: virtualNamespace,
Name: virtualName,
}, &vSecret)
if err != nil {
return fmt.Errorf("unable to get virtual secret %s/%s: %w", virtualNamespace, virtualName, err)
}
translated, err := s.TranslateFunc(&vSecret)
if err != nil {
return fmt.Errorf("unable to translate virtual secret: %s/%s: %w", virtualNamespace, virtualName, err)
}
return s.HostClient.Delete(ctx, translated)
}
+150
View File
@@ -0,0 +1,150 @@
package syncer
import (
"context"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
ctrl "sigs.k8s.io/controller-runtime"
"github.com/rancher/k3k/k3k-kubelet/translate"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
)
const (
configMapControllerName = "configmap-syncer"
configMapFinalizerName = "configmap.k3k.io/finalizer"
)
type ConfigMapSyncer struct {
// SyncerContext contains all client information for host and virtual cluster
*SyncerContext
}
func (c *ConfigMapSyncer) Name() string {
return configMapControllerName
}
// AddConfigMapSyncer adds configmap syncer controller to the manager of the virtual cluster
func AddConfigMapSyncer(ctx context.Context, virtMgr, hostMgr manager.Manager, clusterName, clusterNamespace string) error {
reconciler := ConfigMapSyncer{
SyncerContext: &SyncerContext{
VirtualClient: virtMgr.GetClient(),
HostClient: hostMgr.GetClient(),
Translator: translate.ToHostTranslator{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
},
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
},
}
name := reconciler.Translator.TranslateName(clusterNamespace, configMapControllerName)
return ctrl.NewControllerManagedBy(virtMgr).
Named(name).
For(&corev1.ConfigMap{}).WithEventFilter(predicate.NewPredicateFuncs(reconciler.filterResources)).
Complete(&reconciler)
}
func (c *ConfigMapSyncer) filterResources(object client.Object) bool {
var cluster v1alpha1.Cluster
ctx := context.Background()
if err := c.HostClient.Get(ctx, types.NamespacedName{Name: c.ClusterName, Namespace: c.ClusterNamespace}, &cluster); err != nil {
return false
}
// check for configMap Sync Config
syncConfig := cluster.Spec.Sync.ConfigMaps
// If syncing is disabled, only process deletions to allow for cleanup.
if !syncConfig.Enabled {
return object.GetDeletionTimestamp() != nil
}
labelSelector := labels.SelectorFromSet(syncConfig.Selector)
if labelSelector.Empty() {
return true
}
return labelSelector.Matches(labels.Set(object.GetLabels()))
}
// Reconcile implements reconcile.Reconciler and synchronizes the objects in objs to the host cluster
func (c *ConfigMapSyncer) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx).WithValues("cluster", c.ClusterName, "clusterNamespace", c.ClusterName)
ctx = ctrl.LoggerInto(ctx, log)
var cluster v1alpha1.Cluster
if err := c.HostClient.Get(ctx, types.NamespacedName{Name: c.ClusterName, Namespace: c.ClusterNamespace}, &cluster); err != nil {
return reconcile.Result{}, err
}
var virtualConfigMap corev1.ConfigMap
if err := c.VirtualClient.Get(ctx, req.NamespacedName, &virtualConfigMap); err != nil {
return reconcile.Result{}, client.IgnoreNotFound(err)
}
syncedConfigMap := c.translateConfigMap(&virtualConfigMap)
// handle deletion
if !virtualConfigMap.DeletionTimestamp.IsZero() {
// deleting the synced configMap if exist
if err := c.HostClient.Delete(ctx, syncedConfigMap); err != nil && !apierrors.IsNotFound(err) {
return reconcile.Result{}, err
}
// remove the finalizer after cleaning up the synced configMap
if controllerutil.RemoveFinalizer(&virtualConfigMap, configMapFinalizerName) {
if err := c.VirtualClient.Update(ctx, &virtualConfigMap); err != nil {
return reconcile.Result{}, err
}
}
return reconcile.Result{}, nil
}
// Add finalizer if it does not exist
if controllerutil.AddFinalizer(&virtualConfigMap, configMapFinalizerName) {
if err := c.VirtualClient.Update(ctx, &virtualConfigMap); err != nil {
return reconcile.Result{}, err
}
}
var hostConfigMap corev1.ConfigMap
if err := c.HostClient.Get(ctx, types.NamespacedName{Name: syncedConfigMap.Name, Namespace: syncedConfigMap.Namespace}, &hostConfigMap); err != nil {
if apierrors.IsNotFound(err) {
log.Info("creating the ConfigMap for the first time on the host cluster")
return reconcile.Result{}, c.HostClient.Create(ctx, syncedConfigMap)
}
return reconcile.Result{}, err
}
// TODO: Add option to keep labels/annotation set by the host cluster
log.Info("updating ConfigMap on the host cluster")
return reconcile.Result{}, c.HostClient.Update(ctx, syncedConfigMap)
}
// translateConfigMap will translate a given configMap created in the virtual cluster and
// translates it to host cluster object
func (c *ConfigMapSyncer) translateConfigMap(configMap *corev1.ConfigMap) *corev1.ConfigMap {
hostConfigMap := configMap.DeepCopy()
c.Translator.TranslateTo(hostConfigMap)
return hostConfigMap
}
@@ -0,0 +1,243 @@
package syncer_test
import (
"context"
"fmt"
"time"
"sigs.k8s.io/controller-runtime/pkg/client"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/rancher/k3k/k3k-kubelet/controller/syncer"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
var ConfigMapTests = func() {
var (
namespace string
cluster v1alpha1.Cluster
)
BeforeEach(func() {
ctx := context.Background()
ns := v1.Namespace{
ObjectMeta: metav1.ObjectMeta{GenerateName: "ns-"},
}
err := hostTestEnv.k8sClient.Create(ctx, &ns)
Expect(err).NotTo(HaveOccurred())
namespace = ns.Name
cluster = v1alpha1.Cluster{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cluster-",
Namespace: namespace,
},
Spec: v1alpha1.ClusterSpec{
Sync: v1alpha1.SyncConfig{
ConfigMaps: v1alpha1.ConfigMapSyncConfig{
Enabled: true,
},
},
},
}
err = hostTestEnv.k8sClient.Create(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
err = syncer.AddConfigMapSyncer(ctx, virtManager, hostManager, cluster.Name, cluster.Namespace)
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
ns := v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
err := hostTestEnv.k8sClient.Delete(context.Background(), &ns)
Expect(err).NotTo(HaveOccurred())
})
It("creates a ConfigMap on the host cluster", func() {
ctx := context.Background()
configMap := &v1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cm-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Data: map[string]string{
"foo": "bar",
},
}
err := virtTestEnv.k8sClient.Create(ctx, configMap)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created configmap %s in virtual cluster", configMap.Name))
var hostConfigMap v1.ConfigMap
hostConfigMapName := translateName(cluster, configMap.Namespace, configMap.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostConfigMapName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostConfigMap)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created Configmap %s in host cluster", hostConfigMapName))
Expect(hostConfigMap.Data).To(Equal(configMap.Data))
Expect(hostConfigMap.Labels).To(ContainElement("bar"))
GinkgoWriter.Printf("labels: %v\n", hostConfigMap.Labels)
})
It("updates a ConfigMap on the host cluster", func() {
ctx := context.Background()
configMap := &v1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cm-",
Namespace: "default",
},
Data: map[string]string{
"foo": "bar",
},
}
err := virtTestEnv.k8sClient.Create(ctx, configMap)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created configmap %s in virtual cluster", configMap.Name))
var hostConfigMap v1.ConfigMap
hostConfigMapName := translateName(cluster, configMap.Namespace, configMap.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostConfigMapName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostConfigMap)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created configmap %s in host cluster", hostConfigMapName))
Expect(hostConfigMap.Data).To(Equal(configMap.Data))
Expect(hostConfigMap.Labels).NotTo(ContainElement("bar"))
key := client.ObjectKeyFromObject(configMap)
err = virtTestEnv.k8sClient.Get(ctx, key, configMap)
Expect(err).NotTo(HaveOccurred())
configMap.Labels = map[string]string{"foo": "bar"}
// update virtual configmap
err = virtTestEnv.k8sClient.Update(ctx, configMap)
Expect(err).NotTo(HaveOccurred())
Expect(configMap.Labels).To(ContainElement("bar"))
err = virtTestEnv.k8sClient.Get(ctx, key, configMap)
// check hostConfigMap
Eventually(func() map[string]string {
key := client.ObjectKey{Name: hostConfigMapName, Namespace: namespace}
err = hostTestEnv.k8sClient.Get(ctx, key, &hostConfigMap)
Expect(err).NotTo(HaveOccurred())
return hostConfigMap.Labels
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(ContainElement("bar"))
})
It("deletes a configMap on the host cluster", func() {
ctx := context.Background()
configMap := &v1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cm-",
Namespace: "default",
},
Data: map[string]string{
"foo": "bar",
},
}
err := virtTestEnv.k8sClient.Create(ctx, configMap)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created configmap %s in virtual cluster", configMap.Name))
var hostConfigMap v1.ConfigMap
hostConfigMapName := translateName(cluster, configMap.Namespace, configMap.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostConfigMapName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostConfigMap)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created configmap %s in host cluster", hostConfigMapName))
Expect(hostConfigMap.Data).To(Equal(hostConfigMap.Data))
err = virtTestEnv.k8sClient.Delete(ctx, configMap)
Expect(err).NotTo(HaveOccurred())
Eventually(func() bool {
key := client.ObjectKey{Name: hostConfigMapName, Namespace: namespace}
err := hostTestEnv.k8sClient.Get(ctx, key, &hostConfigMap)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
It("will not sync a configMap if disabled", func() {
ctx := context.Background()
cluster.Spec.Sync.ConfigMaps.Enabled = false
err := hostTestEnv.k8sClient.Update(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
configMap := &v1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cm-",
Namespace: "default",
},
Data: map[string]string{
"foo": "bar",
},
}
err = virtTestEnv.k8sClient.Create(ctx, configMap)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created configmap %s in virtual cluster", configMap.Name))
var hostConfigMap v1.ConfigMap
hostConfigMapName := translateName(cluster, configMap.Namespace, configMap.Name)
Eventually(func() bool {
key := client.ObjectKey{Name: hostConfigMapName, Namespace: namespace}
err := hostTestEnv.k8sClient.Get(ctx, key, &hostConfigMap)
GinkgoWriter.Printf("error: %v", err)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
}
+161
View File
@@ -0,0 +1,161 @@
package syncer
import (
"context"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
networkingv1 "k8s.io/api/networking/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
ctrl "sigs.k8s.io/controller-runtime"
ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
"github.com/rancher/k3k/k3k-kubelet/translate"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
)
const (
ingressControllerName = "ingress-syncer-controller"
ingressFinalizerName = "ingress.k3k.io/finalizer"
)
type IngressReconciler struct {
*SyncerContext
}
// AddIngressSyncer adds ingress syncer controller to the manager of the virtual cluster
func AddIngressSyncer(ctx context.Context, virtMgr, hostMgr manager.Manager, clusterName, clusterNamespace string) error {
reconciler := IngressReconciler{
SyncerContext: &SyncerContext{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
VirtualClient: virtMgr.GetClient(),
HostClient: hostMgr.GetClient(),
Translator: translate.ToHostTranslator{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
},
},
}
name := reconciler.Translator.TranslateName(clusterNamespace, ingressControllerName)
return ctrl.NewControllerManagedBy(virtMgr).
Named(name).
For(&networkingv1.Ingress{}).
WithEventFilter(predicate.NewPredicateFuncs(reconciler.filterResources)).
Complete(&reconciler)
}
func (r *IngressReconciler) filterResources(object ctrlruntimeclient.Object) bool {
var cluster v1alpha1.Cluster
ctx := context.Background()
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return false
}
// check for ingressConfig
syncConfig := cluster.Spec.Sync.Ingresses
// If syncing is disabled, only process deletions to allow for cleanup.
if !syncConfig.Enabled {
return object.GetDeletionTimestamp() != nil
}
labelSelector := labels.SelectorFromSet(syncConfig.Selector)
if labelSelector.Empty() {
return true
}
return labelSelector.Matches(labels.Set(object.GetLabels()))
}
func (r *IngressReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.ClusterName, "clusterNamespace", r.ClusterNamespace)
ctx = ctrl.LoggerInto(ctx, log)
log.Info("reconciling ingress object")
var (
virtIngress networkingv1.Ingress
cluster v1alpha1.Cluster
)
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return reconcile.Result{}, err
}
if err := r.VirtualClient.Get(ctx, req.NamespacedName, &virtIngress); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
syncedIngress := r.ingress(&virtIngress)
if err := controllerutil.SetControllerReference(&cluster, syncedIngress, r.HostClient.Scheme()); err != nil {
return reconcile.Result{}, err
}
// handle deletion
if !virtIngress.DeletionTimestamp.IsZero() {
// deleting the synced service if exists
if err := r.HostClient.Delete(ctx, syncedIngress); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
// remove the finalizer after cleaning up the synced service
if controllerutil.RemoveFinalizer(&virtIngress, ingressFinalizerName) {
if err := r.VirtualClient.Update(ctx, &virtIngress); err != nil {
return reconcile.Result{}, err
}
}
return reconcile.Result{}, nil
}
// Add finalizer if it does not exist
if controllerutil.AddFinalizer(&virtIngress, ingressFinalizerName) {
if err := r.VirtualClient.Update(ctx, &virtIngress); err != nil {
return reconcile.Result{}, err
}
}
// create or update the ingress on host
var hostIngress networkingv1.Ingress
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: syncedIngress.Name, Namespace: r.ClusterNamespace}, &hostIngress); err != nil {
if apierrors.IsNotFound(err) {
log.Info("creating the ingress for the first time on the host cluster")
return reconcile.Result{}, r.HostClient.Create(ctx, syncedIngress)
}
return reconcile.Result{}, err
}
log.Info("updating ingress on the host cluster")
return reconcile.Result{}, r.HostClient.Update(ctx, syncedIngress)
}
func (s *IngressReconciler) ingress(obj *networkingv1.Ingress) *networkingv1.Ingress {
hostIngress := obj.DeepCopy()
s.Translator.TranslateTo(hostIngress)
for _, rule := range hostIngress.Spec.Rules {
// modify services in rules to point to the synced services
if rule.HTTP != nil {
for _, path := range rule.HTTP.Paths {
if path.Backend.Service != nil {
path.Backend.Service.Name = s.Translator.TranslateName(obj.GetNamespace(), path.Backend.Service.Name)
}
}
}
}
// don't sync finalizers to the host
return hostIngress
}
@@ -0,0 +1,349 @@
package syncer_test
import (
"context"
"fmt"
"time"
"k8s.io/utils/ptr"
"sigs.k8s.io/controller-runtime/pkg/client"
v1 "k8s.io/api/core/v1"
networkingv1 "k8s.io/api/networking/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/rancher/k3k/k3k-kubelet/controller/syncer"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
var IngressTests = func() {
var (
namespace string
cluster v1alpha1.Cluster
)
BeforeEach(func() {
ctx := context.Background()
ns := v1.Namespace{
ObjectMeta: metav1.ObjectMeta{GenerateName: "ns-"},
}
err := hostTestEnv.k8sClient.Create(ctx, &ns)
Expect(err).NotTo(HaveOccurred())
namespace = ns.Name
cluster = v1alpha1.Cluster{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cluster-",
Namespace: namespace,
},
Spec: v1alpha1.ClusterSpec{
Sync: v1alpha1.SyncConfig{
Ingresses: v1alpha1.IngressSyncConfig{
Enabled: true,
},
},
},
}
err = hostTestEnv.k8sClient.Create(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
err = syncer.AddIngressSyncer(ctx, virtManager, hostManager, cluster.Name, cluster.Namespace)
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
ns := v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
err := hostTestEnv.k8sClient.Delete(context.Background(), &ns)
Expect(err).NotTo(HaveOccurred())
})
It("creates a Ingress on the host cluster", func() {
ctx := context.Background()
ingress := &networkingv1.Ingress{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "ingress-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Spec: networkingv1.IngressSpec{
Rules: []networkingv1.IngressRule{
{
Host: "test.com",
IngressRuleValue: networkingv1.IngressRuleValue{
HTTP: &networkingv1.HTTPIngressRuleValue{
Paths: []networkingv1.HTTPIngressPath{
{
Path: "/",
PathType: ptr.To(networkingv1.PathTypePrefix),
Backend: networkingv1.IngressBackend{
Service: &networkingv1.IngressServiceBackend{
Name: "test-service",
Port: networkingv1.ServiceBackendPort{
Name: "test-port",
},
},
},
},
},
},
},
},
},
},
}
err := virtTestEnv.k8sClient.Create(ctx, ingress)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created Ingress %s in virtual cluster", ingress.Name))
var hostIngress networkingv1.Ingress
hostIngressName := translateName(cluster, ingress.Namespace, ingress.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostIngressName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostIngress)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created Ingress %s in host cluster", hostIngressName))
Expect(len(hostIngress.Spec.Rules)).To(Equal(1))
Expect(hostIngress.Spec.Rules[0].Host).To(Equal("test.com"))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Path).To(Equal("/"))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name).To(Equal(translateName(cluster, ingress.Namespace, "test-service")))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Port.Name).To(Equal("test-port"))
GinkgoWriter.Printf("labels: %v\n", hostIngress.Labels)
})
It("updates a Ingress on the host cluster", func() {
ctx := context.Background()
ingress := &networkingv1.Ingress{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "ingress-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Spec: networkingv1.IngressSpec{
Rules: []networkingv1.IngressRule{
{
Host: "test.com",
IngressRuleValue: networkingv1.IngressRuleValue{
HTTP: &networkingv1.HTTPIngressRuleValue{
Paths: []networkingv1.HTTPIngressPath{
{
Path: "/",
PathType: ptr.To(networkingv1.PathTypePrefix),
Backend: networkingv1.IngressBackend{
Service: &networkingv1.IngressServiceBackend{
Name: "test-service",
Port: networkingv1.ServiceBackendPort{
Name: "test-port",
},
},
},
},
},
},
},
},
},
},
}
err := virtTestEnv.k8sClient.Create(ctx, ingress)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created Ingress %s in virtual cluster", ingress.Name))
var hostIngress networkingv1.Ingress
hostIngressName := translateName(cluster, ingress.Namespace, ingress.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostIngressName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostIngress)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created Ingress %s in host cluster", hostIngressName))
Expect(len(hostIngress.Spec.Rules)).To(Equal(1))
Expect(hostIngress.Spec.Rules[0].Host).To(Equal("test.com"))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Path).To(Equal("/"))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name).To(Equal(translateName(cluster, ingress.Namespace, "test-service")))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Port.Name).To(Equal("test-port"))
key := client.ObjectKeyFromObject(ingress)
err = virtTestEnv.k8sClient.Get(ctx, key, ingress)
Expect(err).NotTo(HaveOccurred())
ingress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name = "test-service-updated"
// update virtual ingress
err = virtTestEnv.k8sClient.Update(ctx, ingress)
Expect(err).NotTo(HaveOccurred())
// check hostIngress
Eventually(func() string {
key := client.ObjectKey{Name: hostIngressName, Namespace: namespace}
err = hostTestEnv.k8sClient.Get(ctx, key, &hostIngress)
Expect(err).NotTo(HaveOccurred())
return hostIngress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(Equal(translateName(cluster, ingress.Namespace, "test-service-updated")))
})
It("deletes a Ingress on the host cluster", func() {
ctx := context.Background()
ingress := &networkingv1.Ingress{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "ingress-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Spec: networkingv1.IngressSpec{
Rules: []networkingv1.IngressRule{
{
Host: "test.com",
IngressRuleValue: networkingv1.IngressRuleValue{
HTTP: &networkingv1.HTTPIngressRuleValue{
Paths: []networkingv1.HTTPIngressPath{
{
Path: "/",
PathType: ptr.To(networkingv1.PathTypePrefix),
Backend: networkingv1.IngressBackend{
Service: &networkingv1.IngressServiceBackend{
Name: "test-service",
Port: networkingv1.ServiceBackendPort{
Name: "test-port",
},
},
},
},
},
},
},
},
},
},
}
err := virtTestEnv.k8sClient.Create(ctx, ingress)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created Ingress %s in virtual cluster", ingress.Name))
var hostIngress networkingv1.Ingress
hostIngressName := translateName(cluster, ingress.Namespace, ingress.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostIngressName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostIngress)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created Ingress %s in host cluster", hostIngressName))
Expect(len(hostIngress.Spec.Rules)).To(Equal(1))
Expect(hostIngress.Spec.Rules[0].Host).To(Equal("test.com"))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Path).To(Equal("/"))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Name).To(Equal(translateName(cluster, ingress.Namespace, "test-service")))
Expect(hostIngress.Spec.Rules[0].HTTP.Paths[0].Backend.Service.Port.Name).To(Equal("test-port"))
err = virtTestEnv.k8sClient.Delete(ctx, ingress)
Expect(err).NotTo(HaveOccurred())
Eventually(func() bool {
key := client.ObjectKey{Name: hostIngressName, Namespace: namespace}
err := hostTestEnv.k8sClient.Get(ctx, key, &hostIngress)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
It("will not sync an Ingress if disabled", func() {
ctx := context.Background()
cluster.Spec.Sync.Ingresses.Enabled = false
err := hostTestEnv.k8sClient.Update(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
ingress := &networkingv1.Ingress{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "ingress-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Spec: networkingv1.IngressSpec{
Rules: []networkingv1.IngressRule{
{
Host: "test.com",
IngressRuleValue: networkingv1.IngressRuleValue{
HTTP: &networkingv1.HTTPIngressRuleValue{
Paths: []networkingv1.HTTPIngressPath{
{
Path: "/",
PathType: ptr.To(networkingv1.PathTypePrefix),
Backend: networkingv1.IngressBackend{
Service: &networkingv1.IngressServiceBackend{
Name: "test-service",
Port: networkingv1.ServiceBackendPort{
Name: "test-port",
},
},
},
},
},
},
},
},
},
},
}
err = virtTestEnv.k8sClient.Create(ctx, ingress)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created Ingress %s in virtual cluster", ingress.Name))
var hostIngress networkingv1.Ingress
hostIngressName := translateName(cluster, ingress.Namespace, ingress.Name)
Eventually(func() bool {
key := client.ObjectKey{Name: hostIngressName, Namespace: namespace}
err := hostTestEnv.k8sClient.Get(ctx, key, &hostIngress)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
}
@@ -1,12 +1,13 @@
package controller
package syncer
import (
"context"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
v1 "k8s.io/api/core/v1"
@@ -19,48 +20,65 @@ import (
)
const (
pvcController = "pvc-syncer-controller"
pvcFinalizerName = "pvc.k3k.io/finalizer"
pvcControllerName = "pvc-syncer-controller"
pvcFinalizerName = "pvc.k3k.io/finalizer"
)
type PVCReconciler struct {
clusterName string
clusterNamespace string
virtualClient ctrlruntimeclient.Client
hostClient ctrlruntimeclient.Client
Scheme *runtime.Scheme
HostScheme *runtime.Scheme
Translator translate.ToHostTranslator
*SyncerContext
}
// AddPVCSyncer adds persistentvolumeclaims syncer controller to k3k-kubelet
func AddPVCSyncer(ctx context.Context, virtMgr, hostMgr manager.Manager, clusterName, clusterNamespace string) error {
translator := translate.ToHostTranslator{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
}
// initialize a new Reconciler
reconciler := PVCReconciler{
clusterName: clusterName,
clusterNamespace: clusterNamespace,
virtualClient: virtMgr.GetClient(),
hostClient: hostMgr.GetClient(),
Scheme: virtMgr.GetScheme(),
HostScheme: hostMgr.GetScheme(),
Translator: translator,
SyncerContext: &SyncerContext{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
VirtualClient: virtMgr.GetClient(),
HostClient: hostMgr.GetClient(),
Translator: translate.ToHostTranslator{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
},
},
}
name := reconciler.Translator.TranslateName(clusterNamespace, pvcControllerName)
return ctrl.NewControllerManagedBy(virtMgr).
Named(pvcController).
Named(name).
For(&v1.PersistentVolumeClaim{}).
WithEventFilter(predicate.NewPredicateFuncs(reconciler.filterResources)).
Complete(&reconciler)
}
func (r *PVCReconciler) filterResources(object ctrlruntimeclient.Object) bool {
var cluster v1alpha1.Cluster
ctx := context.Background()
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return false
}
// check for pvc config
syncConfig := cluster.Spec.Sync.PersistentVolumeClaims
// If syncing is disabled, only process deletions to allow for cleanup.
if !syncConfig.Enabled {
return object.GetDeletionTimestamp() != nil
}
labelSelector := labels.SelectorFromSet(syncConfig.Selector)
if labelSelector.Empty() {
return true
}
return labelSelector.Matches(labels.Set(object.GetLabels()))
}
func (r *PVCReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.clusterName, "clusterNamespace", r.clusterNamespace)
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.ClusterName, "clusterNamespace", r.ClusterNamespace)
ctx = ctrl.LoggerInto(ctx, log)
var (
@@ -68,28 +86,28 @@ func (r *PVCReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
cluster v1alpha1.Cluster
)
if err := r.hostClient.Get(ctx, types.NamespacedName{Name: r.clusterName, Namespace: r.clusterNamespace}, &cluster); err != nil {
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return reconcile.Result{}, err
}
if err := r.virtualClient.Get(ctx, req.NamespacedName, &virtPVC); err != nil {
if err := r.VirtualClient.Get(ctx, req.NamespacedName, &virtPVC); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
syncedPVC := r.pvc(&virtPVC)
if err := controllerutil.SetControllerReference(&cluster, syncedPVC, r.HostScheme); err != nil {
if err := controllerutil.SetControllerReference(&cluster, syncedPVC, r.HostClient.Scheme()); err != nil {
return reconcile.Result{}, err
}
// handle deletion
if !virtPVC.DeletionTimestamp.IsZero() {
// deleting the synced service if exists
if err := r.hostClient.Delete(ctx, syncedPVC); !apierrors.IsNotFound(err) {
// deleting the synced pvc if exists
if err := r.HostClient.Delete(ctx, syncedPVC); !apierrors.IsNotFound(err) {
return reconcile.Result{}, err
}
// remove the finalizer after cleaning up the synced service
// remove the finalizer after cleaning up the synced pvc
if controllerutil.RemoveFinalizer(&virtPVC, pvcFinalizerName) {
if err := r.virtualClient.Update(ctx, &virtPVC); err != nil {
if err := r.VirtualClient.Update(ctx, &virtPVC); err != nil {
return reconcile.Result{}, err
}
}
@@ -99,17 +117,17 @@ func (r *PVCReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
// Add finalizer if it does not exist
if controllerutil.AddFinalizer(&virtPVC, pvcFinalizerName) {
if err := r.virtualClient.Update(ctx, &virtPVC); err != nil {
if err := r.VirtualClient.Update(ctx, &virtPVC); err != nil {
return reconcile.Result{}, err
}
}
// create the pvc on host
log.Info("creating the persistent volume for the first time on the host cluster")
log.Info("creating the persistent volume claim for the first time on the host cluster")
// note that we dont need to update the PVC on the host cluster, only syncing the PVC to allow being
// handled by the host cluster.
return reconcile.Result{}, ctrlruntimeclient.IgnoreAlreadyExists(r.hostClient.Create(ctx, syncedPVC))
return reconcile.Result{}, ctrlruntimeclient.IgnoreAlreadyExists(r.HostClient.Create(ctx, syncedPVC))
}
func (r *PVCReconciler) pvc(obj *v1.PersistentVolumeClaim) *v1.PersistentVolumeClaim {
@@ -0,0 +1,111 @@
package syncer_test
import (
"context"
"fmt"
"time"
"k8s.io/apimachinery/pkg/api/resource"
"k8s.io/utils/ptr"
"sigs.k8s.io/controller-runtime/pkg/client"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/rancher/k3k/k3k-kubelet/controller/syncer"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
var PVCTests = func() {
var (
namespace string
cluster v1alpha1.Cluster
)
BeforeEach(func() {
ctx := context.Background()
ns := v1.Namespace{
ObjectMeta: metav1.ObjectMeta{GenerateName: "ns-"},
}
err := hostTestEnv.k8sClient.Create(ctx, &ns)
Expect(err).NotTo(HaveOccurred())
namespace = ns.Name
cluster = v1alpha1.Cluster{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cluster-",
Namespace: namespace,
},
Spec: v1alpha1.ClusterSpec{
Sync: v1alpha1.SyncConfig{
PersistentVolumeClaims: v1alpha1.PersistentVolumeClaimSyncConfig{
Enabled: true,
},
},
},
}
err = hostTestEnv.k8sClient.Create(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
err = syncer.AddPVCSyncer(ctx, virtManager, hostManager, cluster.Name, cluster.Namespace)
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
ns := v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
err := hostTestEnv.k8sClient.Delete(context.Background(), &ns)
Expect(err).NotTo(HaveOccurred())
})
It("creates a pvc on the host cluster", func() {
ctx := context.Background()
pvc := &v1.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "pvc-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Spec: v1.PersistentVolumeClaimSpec{
StorageClassName: ptr.To("test-sc"),
AccessModes: []v1.PersistentVolumeAccessMode{
v1.ReadOnlyMany,
},
Resources: v1.VolumeResourceRequirements{
Requests: v1.ResourceList{
"storage": resource.MustParse("1G"),
},
},
},
}
err := virtTestEnv.k8sClient.Create(ctx, pvc)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created PVC %s in virtual cluster", pvc.Name))
var hostPVC v1.PersistentVolumeClaim
hostPVCName := translateName(cluster, pvc.Namespace, pvc.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostPVCName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostPVC)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created PVC %s in host cluster", hostPVCName))
Expect(*hostPVC.Spec.StorageClassName).To(Equal("test-sc"))
GinkgoWriter.Printf("labels: %v\n", hostPVC.Labels)
})
}
@@ -1,12 +1,12 @@
package controller
package syncer
import (
"context"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/component-helpers/storage/volume"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
v1 "k8s.io/api/core/v1"
@@ -19,48 +19,54 @@ import (
)
const (
podController = "pod-pvc-controller"
pseudoPVLabel = "pod.k3k.io/pseudoPV"
podControllerName = "pod-pvc-controller"
pseudoPVLabel = "pod.k3k.io/pseudoPV"
)
type PodReconciler struct {
clusterName string
clusterNamespace string
virtualClient ctrlruntimeclient.Client
hostClient ctrlruntimeclient.Client
Scheme *runtime.Scheme
HostScheme *runtime.Scheme
Translator translate.ToHostTranslator
*SyncerContext
}
// AddPodPVCController adds pod controller to k3k-kubelet
func AddPodPVCController(ctx context.Context, virtMgr, hostMgr manager.Manager, clusterName, clusterNamespace string) error {
translator := translate.ToHostTranslator{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
}
// initialize a new Reconciler
reconciler := PodReconciler{
clusterName: clusterName,
clusterNamespace: clusterNamespace,
virtualClient: virtMgr.GetClient(),
hostClient: hostMgr.GetClient(),
Scheme: virtMgr.GetScheme(),
HostScheme: hostMgr.GetScheme(),
Translator: translator,
SyncerContext: &SyncerContext{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
VirtualClient: virtMgr.GetClient(),
HostClient: hostMgr.GetClient(),
Translator: translate.ToHostTranslator{},
},
}
name := reconciler.Translator.TranslateName(clusterNamespace, podControllerName)
return ctrl.NewControllerManagedBy(virtMgr).
Named(podController).
Named(name).
For(&v1.Pod{}).
WithEventFilter(predicate.NewPredicateFuncs(reconciler.filterResources)).
Complete(&reconciler)
}
func (r *PodReconciler) filterResources(object ctrlruntimeclient.Object) bool {
var cluster v1alpha1.Cluster
ctx := context.Background()
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return false
}
// check for pvc config
syncConfig := cluster.Spec.Sync.PersistentVolumeClaims
// If PVC syncing is disabled, only process deletions to allow for cleanup.
return syncConfig.Enabled || object.GetDeletionTimestamp() != nil
}
func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.clusterName, "clusterNamespace", r.clusterNamespace)
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.ClusterName, "clusterNamespace", r.ClusterName)
ctx = ctrl.LoggerInto(ctx, log)
var (
@@ -68,11 +74,11 @@ func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (r
cluster v1alpha1.Cluster
)
if err := r.hostClient.Get(ctx, types.NamespacedName{Name: r.clusterName, Namespace: r.clusterNamespace}, &cluster); err != nil {
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterName}, &cluster); err != nil {
return reconcile.Result{}, err
}
if err := r.virtualClient.Get(ctx, req.NamespacedName, &virtPod); err != nil {
if err := r.VirtualClient.Get(ctx, req.NamespacedName, &virtPod); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
@@ -103,14 +109,14 @@ func (r *PodReconciler) reconcilePodWithPVC(ctx context.Context, pod *v1.Pod, pv
Namespace: pod.Namespace,
}
if err := r.virtualClient.Get(ctx, key, &pvc); err != nil {
if err := r.VirtualClient.Get(ctx, key, &pvc); err != nil {
return ctrlruntimeclient.IgnoreNotFound(err)
}
log.Info("Creating pseudo Persistent Volume")
pv := r.pseudoPV(&pvc)
if err := r.virtualClient.Create(ctx, pv); err != nil {
if err := r.VirtualClient.Create(ctx, pv); err != nil {
return ctrlruntimeclient.IgnoreAlreadyExists(err)
}
@@ -119,7 +125,7 @@ func (r *PodReconciler) reconcilePodWithPVC(ctx context.Context, pod *v1.Pod, pv
Phase: v1.VolumeBound,
}
if err := r.virtualClient.Status().Patch(ctx, pv, ctrlruntimeclient.MergeFrom(orig)); err != nil {
if err := r.VirtualClient.Status().Patch(ctx, pv, ctrlruntimeclient.MergeFrom(orig)); err != nil {
return err
}
@@ -135,7 +141,7 @@ func (r *PodReconciler) reconcilePodWithPVC(ctx context.Context, pod *v1.Pod, pv
pvcPatch.Status.Phase = v1.ClaimBound
pvcPatch.Status.AccessModes = pvcPatch.Spec.AccessModes
return r.virtualClient.Status().Update(ctx, pvcPatch)
return r.VirtualClient.Status().Update(ctx, pvcPatch)
}
func (r *PodReconciler) pseudoPV(obj *v1.PersistentVolumeClaim) *v1.PersistentVolume {
@@ -1,4 +1,4 @@
package controller_test
package syncer_test
import (
"context"
@@ -12,8 +12,7 @@ import (
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/rancher/k3k/k3k-kubelet/controller"
"github.com/rancher/k3k/k3k-kubelet/translate"
"github.com/rancher/k3k/k3k-kubelet/controller/syncer"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
. "github.com/onsi/ginkgo/v2"
@@ -42,11 +41,18 @@ var PriorityClassTests = func() {
GenerateName: "cluster-",
Namespace: namespace,
},
Spec: v1alpha1.ClusterSpec{
Sync: v1alpha1.SyncConfig{
PriorityClasses: v1alpha1.PriorityClassSyncConfig{
Enabled: true,
},
},
},
}
err = hostTestEnv.k8sClient.Create(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
err = controller.AddPriorityClassReconciler(ctx, virtManager, hostManager, cluster.Name, cluster.Namespace)
err = syncer.AddPriorityClassSyncer(ctx, virtManager, hostManager, cluster.Name, cluster.Namespace)
Expect(err).NotTo(HaveOccurred())
})
@@ -215,15 +221,36 @@ var PriorityClassTests = func() {
Expect(hostPriorityClass.Value).To(Equal(priorityClass.Value))
Expect(hostPriorityClass.GlobalDefault).To(BeFalse())
Expect(hostPriorityClass.Annotations[controller.PriorityClassGlobalDefaultAnnotation]).To(Equal("true"))
Expect(hostPriorityClass.Annotations[syncer.PriorityClassGlobalDefaultAnnotation]).To(Equal("true"))
})
It("will not create a priorityClass on the host cluster if disabled", func() {
ctx := context.Background()
cluster.Spec.Sync.PriorityClasses.Enabled = false
err := hostTestEnv.k8sClient.Update(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
priorityClass := &schedulingv1.PriorityClass{
ObjectMeta: metav1.ObjectMeta{GenerateName: "pc-"},
Value: 1001,
}
err = virtTestEnv.k8sClient.Create(ctx, priorityClass)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created priorityClass %s in virtual cluster", priorityClass.Name))
var hostPriorityClass schedulingv1.PriorityClass
hostPriorityClassName := translateName(cluster, priorityClass.Namespace, priorityClass.Name)
Eventually(func() bool {
key := client.ObjectKey{Name: hostPriorityClassName}
err = hostTestEnv.k8sClient.Get(ctx, key, &hostPriorityClass)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
}
func translateName(cluster v1alpha1.Cluster, namespace, name string) string {
translator := translate.ToHostTranslator{
ClusterName: cluster.Name,
ClusterNamespace: cluster.Namespace,
}
return translator.TranslateName(namespace, name)
}
@@ -1,10 +1,10 @@
package controller
package syncer
import (
"context"
"strings"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/event"
@@ -28,42 +28,32 @@ const (
priorityClassFinalizerName = "priorityclass.k3k.io/finalizer"
)
type PriorityClassReconciler struct {
clusterName string
clusterNamespace string
virtualClient ctrlruntimeclient.Client
hostClient ctrlruntimeclient.Client
Scheme *runtime.Scheme
HostScheme *runtime.Scheme
Translator translate.ToHostTranslator
type PriorityClassSyncer struct {
*SyncerContext
}
// AddPriorityClassReconciler adds a PriorityClass reconciler to k3k-kubelet
func AddPriorityClassReconciler(ctx context.Context, virtMgr, hostMgr manager.Manager, clusterName, clusterNamespace string) error {
translator := translate.ToHostTranslator{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
}
// AddPriorityClassSyncer adds a PriorityClass reconciler to k3k-kubelet
func AddPriorityClassSyncer(ctx context.Context, virtMgr, hostMgr manager.Manager, clusterName, clusterNamespace string) error {
// initialize a new Reconciler
reconciler := PriorityClassReconciler{
clusterName: clusterName,
clusterNamespace: clusterNamespace,
virtualClient: virtMgr.GetClient(),
hostClient: hostMgr.GetClient(),
Scheme: virtMgr.GetScheme(),
HostScheme: hostMgr.GetScheme(),
Translator: translator,
reconciler := PriorityClassSyncer{
SyncerContext: &SyncerContext{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
VirtualClient: virtMgr.GetClient(),
HostClient: hostMgr.GetClient(),
Translator: translate.ToHostTranslator{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
},
},
}
name := translator.TranslateName("", priorityClassControllerName)
name := reconciler.Translator.TranslateName(clusterNamespace, priorityClassControllerName)
return ctrl.NewControllerManagedBy(virtMgr).
Named(name).
For(&schedulingv1.PriorityClass{}).
WithEventFilter(ignoreSystemPrefixPredicate).
For(&schedulingv1.PriorityClass{}).WithEventFilter(ignoreSystemPrefixPredicate).
WithEventFilter(predicate.NewPredicateFuncs(reconciler.filterResources)).
Complete(&reconciler)
}
@@ -83,8 +73,33 @@ var ignoreSystemPrefixPredicate = predicate.Funcs{
},
}
func (r *PriorityClassReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.clusterName, "clusterNamespace", r.clusterNamespace)
func (r *PriorityClassSyncer) filterResources(object ctrlruntimeclient.Object) bool {
var cluster v1alpha1.Cluster
ctx := context.Background()
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return false
}
// check for priorityClassConfig
syncConfig := cluster.Spec.Sync.PriorityClasses
// If syncing is disabled, only process deletions to allow for cleanup.
if !syncConfig.Enabled {
return object.GetDeletionTimestamp() != nil
}
labelSelector := labels.SelectorFromSet(syncConfig.Selector)
if labelSelector.Empty() {
return true
}
return labelSelector.Matches(labels.Set(object.GetLabels()))
}
func (r *PriorityClassSyncer) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.ClusterName, "clusterNamespace", r.ClusterNamespace)
ctx = ctrl.LoggerInto(ctx, log)
var (
@@ -92,11 +107,11 @@ func (r *PriorityClassReconciler) Reconcile(ctx context.Context, req reconcile.R
cluster v1alpha1.Cluster
)
if err := r.hostClient.Get(ctx, types.NamespacedName{Name: r.clusterName, Namespace: r.clusterNamespace}, &cluster); err != nil {
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return reconcile.Result{}, err
}
if err := r.virtualClient.Get(ctx, req.NamespacedName, &priorityClass); err != nil {
if err := r.VirtualClient.Get(ctx, req.NamespacedName, &priorityClass); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
@@ -106,13 +121,13 @@ func (r *PriorityClassReconciler) Reconcile(ctx context.Context, req reconcile.R
if !priorityClass.DeletionTimestamp.IsZero() {
// deleting the synced service if exists
// TODO add test for previous implementation without err != nil check, and also check the other controllers
if err := r.hostClient.Delete(ctx, hostPriorityClass); err != nil && !apierrors.IsNotFound(err) {
if err := r.HostClient.Delete(ctx, hostPriorityClass); err != nil && !apierrors.IsNotFound(err) {
return reconcile.Result{}, err
}
// remove the finalizer after cleaning up the synced service
if controllerutil.RemoveFinalizer(&priorityClass, priorityClassFinalizerName) {
if err := r.virtualClient.Update(ctx, &priorityClass); err != nil {
if err := r.VirtualClient.Update(ctx, &priorityClass); err != nil {
return reconcile.Result{}, err
}
}
@@ -122,7 +137,7 @@ func (r *PriorityClassReconciler) Reconcile(ctx context.Context, req reconcile.R
// Add finalizer if it does not exist
if controllerutil.AddFinalizer(&priorityClass, priorityClassFinalizerName) {
if err := r.virtualClient.Update(ctx, &priorityClass); err != nil {
if err := r.VirtualClient.Update(ctx, &priorityClass); err != nil {
return reconcile.Result{}, err
}
}
@@ -130,19 +145,19 @@ func (r *PriorityClassReconciler) Reconcile(ctx context.Context, req reconcile.R
// create the priorityClass on the host
log.Info("creating the priorityClass for the first time on the host cluster")
err := r.hostClient.Create(ctx, hostPriorityClass)
err := r.HostClient.Create(ctx, hostPriorityClass)
if err != nil {
if !apierrors.IsAlreadyExists(err) {
return reconcile.Result{}, err
}
return reconcile.Result{}, r.hostClient.Update(ctx, hostPriorityClass)
return reconcile.Result{}, r.HostClient.Update(ctx, hostPriorityClass)
}
return reconcile.Result{}, nil
}
func (r *PriorityClassReconciler) translatePriorityClass(priorityClass schedulingv1.PriorityClass) *schedulingv1.PriorityClass {
func (r *PriorityClassSyncer) translatePriorityClass(priorityClass schedulingv1.PriorityClass) *schedulingv1.PriorityClass {
hostPriorityClass := priorityClass.DeepCopy()
r.Translator.TranslateTo(hostPriorityClass)
+155
View File
@@ -0,0 +1,155 @@
package syncer
import (
"context"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
ctrl "sigs.k8s.io/controller-runtime"
"github.com/rancher/k3k/k3k-kubelet/translate"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
)
const (
secretControllerName = "secret-syncer"
secretFinalizerName = "secret.k3k.io/finalizer"
)
type SecretSyncer struct {
// SyncerContext contains all client information for host and virtual cluster
*SyncerContext
}
func (s *SecretSyncer) Name() string {
return secretControllerName
}
// AddSecretSyncer adds secret syncer controller to the manager of the virtual cluster
func AddSecretSyncer(ctx context.Context, virtMgr, hostMgr manager.Manager, clusterName, clusterNamespace string) error {
reconciler := SecretSyncer{
SyncerContext: &SyncerContext{
VirtualClient: virtMgr.GetClient(),
HostClient: hostMgr.GetClient(),
Translator: translate.ToHostTranslator{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
},
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
},
}
name := reconciler.Translator.TranslateName(clusterNamespace, secretControllerName)
return ctrl.NewControllerManagedBy(virtMgr).
Named(name).
For(&v1.Secret{}).WithEventFilter(predicate.NewPredicateFuncs(reconciler.filterResources)).
Complete(&reconciler)
}
func (r *SecretSyncer) filterResources(object client.Object) bool {
var cluster v1alpha1.Cluster
ctx := context.Background()
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return false
}
// check for Secrets Sync Config
syncConfig := cluster.Spec.Sync.Secrets
// If syncing is disabled, only process deletions to allow for cleanup.
if !syncConfig.Enabled {
return object.GetDeletionTimestamp() != nil
}
labelSelector := labels.SelectorFromSet(syncConfig.Selector)
if labelSelector.Empty() {
return true
}
return labelSelector.Matches(labels.Set(object.GetLabels()))
}
// Reconcile implements reconcile.Reconciler and synchronizes the objects in objs to the host cluster
func (s *SecretSyncer) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx).WithValues("cluster", s.ClusterName, "clusterNamespace", s.ClusterName)
ctx = ctrl.LoggerInto(ctx, log)
var cluster v1alpha1.Cluster
if err := s.HostClient.Get(ctx, types.NamespacedName{Name: s.ClusterName, Namespace: s.ClusterNamespace}, &cluster); err != nil {
return reconcile.Result{}, err
}
var virtualSecret v1.Secret
if err := s.VirtualClient.Get(ctx, req.NamespacedName, &virtualSecret); err != nil {
return reconcile.Result{}, client.IgnoreNotFound(err)
}
syncedSecret := s.translateSecret(&virtualSecret)
// handle deletion
if !virtualSecret.DeletionTimestamp.IsZero() {
// deleting the synced secret if exist
if err := s.HostClient.Delete(ctx, syncedSecret); err != nil && !apierrors.IsNotFound(err) {
return reconcile.Result{}, err
}
// remove the finalizer after cleaning up the synced secret
if controllerutil.RemoveFinalizer(&virtualSecret, secretFinalizerName) {
if err := s.VirtualClient.Update(ctx, &virtualSecret); err != nil {
return reconcile.Result{}, err
}
}
return reconcile.Result{}, nil
}
// Add finalizer if it does not exist
if controllerutil.AddFinalizer(&virtualSecret, secretFinalizerName) {
if err := s.VirtualClient.Update(ctx, &virtualSecret); err != nil {
return reconcile.Result{}, err
}
}
var hostSecret v1.Secret
if err := s.HostClient.Get(ctx, types.NamespacedName{Name: syncedSecret.Name, Namespace: syncedSecret.Namespace}, &hostSecret); err != nil {
if apierrors.IsNotFound(err) {
log.Info("creating the Secret for the first time on the host cluster")
return reconcile.Result{}, s.HostClient.Create(ctx, syncedSecret)
}
return reconcile.Result{}, err
}
// TODO: Add option to keep labels/annotation set by the host cluster
log.Info("updating Secret on the host cluster")
return reconcile.Result{}, s.HostClient.Update(ctx, syncedSecret)
}
// translateSecret will translate a given secret created in the virtual cluster and
// translates it to host cluster object
func (s *SecretSyncer) translateSecret(secret *v1.Secret) *v1.Secret {
hostSecret := secret.DeepCopy()
if hostSecret.Type == v1.SecretTypeServiceAccountToken {
hostSecret.Type = v1.SecretTypeOpaque
}
s.Translator.TranslateTo(hostSecret)
return hostSecret
}
@@ -0,0 +1,240 @@
package syncer_test
import (
"context"
"fmt"
"time"
"sigs.k8s.io/controller-runtime/pkg/client"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/rancher/k3k/k3k-kubelet/controller/syncer"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
var SecretTests = func() {
var (
namespace string
cluster v1alpha1.Cluster
)
BeforeEach(func() {
ctx := context.Background()
ns := v1.Namespace{
ObjectMeta: metav1.ObjectMeta{GenerateName: "ns-"},
}
err := hostTestEnv.k8sClient.Create(ctx, &ns)
Expect(err).NotTo(HaveOccurred())
namespace = ns.Name
cluster = v1alpha1.Cluster{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cluster-",
Namespace: namespace,
},
Spec: v1alpha1.ClusterSpec{
Sync: v1alpha1.SyncConfig{
Secrets: v1alpha1.SecretSyncConfig{
Enabled: true,
},
},
},
}
err = hostTestEnv.k8sClient.Create(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
err = syncer.AddSecretSyncer(ctx, virtManager, hostManager, cluster.Name, cluster.Namespace)
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
ns := v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
err := hostTestEnv.k8sClient.Delete(context.Background(), &ns)
Expect(err).NotTo(HaveOccurred())
})
It("creates a Secret on the host cluster", func() {
ctx := context.Background()
secret := &v1.Secret{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "secret-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Data: map[string][]byte{
"foo": []byte("bar"),
},
}
err := virtTestEnv.k8sClient.Create(ctx, secret)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created Secret %s in virtual cluster", secret.Name))
var hostSecret v1.Secret
hostSecretName := translateName(cluster, secret.Namespace, secret.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostSecretName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostSecret)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created Secret %s in host cluster", hostSecretName))
Expect(hostSecret.Data).To(Equal(secret.Data))
Expect(hostSecret.Labels).To(ContainElement("bar"))
GinkgoWriter.Printf("labels: %v\n", hostSecret.Labels)
})
It("updates a Secret on the host cluster", func() {
ctx := context.Background()
secret := &v1.Secret{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "secret-",
Namespace: "default",
},
Data: map[string][]byte{
"foo": []byte("bar"),
},
}
err := virtTestEnv.k8sClient.Create(ctx, secret)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created secret %s in virtual cluster", secret.Name))
var hostSecret v1.Secret
hostSecretName := translateName(cluster, secret.Namespace, secret.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostSecretName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostSecret)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created secret %s in host cluster", hostSecretName))
Expect(hostSecret.Data).To(Equal(secret.Data))
Expect(hostSecret.Labels).NotTo(ContainElement("bar"))
key := client.ObjectKeyFromObject(secret)
err = virtTestEnv.k8sClient.Get(ctx, key, secret)
Expect(err).NotTo(HaveOccurred())
secret.Labels = map[string]string{"foo": "bar"}
// update virtual secret
err = virtTestEnv.k8sClient.Update(ctx, secret)
Expect(err).NotTo(HaveOccurred())
Expect(secret.Labels).To(ContainElement("bar"))
// check hostSecret
Eventually(func() map[string]string {
key := client.ObjectKey{Name: hostSecretName, Namespace: namespace}
err = hostTestEnv.k8sClient.Get(ctx, key, &hostSecret)
Expect(err).NotTo(HaveOccurred())
return hostSecret.Labels
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(ContainElement("bar"))
})
It("deletes a secret on the host cluster", func() {
ctx := context.Background()
secret := &v1.Secret{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "secret-",
Namespace: "default",
},
Data: map[string][]byte{
"foo": []byte("bar"),
},
}
err := virtTestEnv.k8sClient.Create(ctx, secret)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created secret %s in virtual cluster", secret.Name))
var hostSecret v1.Secret
hostSecretName := translateName(cluster, secret.Namespace, secret.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostSecretName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostSecret)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created secret %s in host cluster", hostSecretName))
Expect(hostSecret.Data).To(Equal(hostSecret.Data))
err = virtTestEnv.k8sClient.Delete(ctx, secret)
Expect(err).NotTo(HaveOccurred())
Eventually(func() bool {
key := client.ObjectKey{Name: hostSecretName, Namespace: namespace}
err := hostTestEnv.k8sClient.Get(ctx, key, &hostSecret)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
It("will not create a secret on the host cluster if disabled", func() {
ctx := context.Background()
cluster.Spec.Sync.Secrets.Enabled = false
err := hostTestEnv.k8sClient.Update(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
secret := &v1.Secret{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "secret-",
Namespace: "default",
},
Data: map[string][]byte{
"foo": []byte("bar"),
},
}
err = virtTestEnv.k8sClient.Create(ctx, secret)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created secret %s in virtual cluster", secret.Name))
var hostSecret v1.Secret
hostSecretName := translateName(cluster, secret.Namespace, secret.Name)
Eventually(func() bool {
key := client.ObjectKey{Name: hostSecretName, Namespace: namespace}
err = hostTestEnv.k8sClient.Get(ctx, key, &hostSecret)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
}
@@ -1,12 +1,13 @@
package controller
package syncer
import (
"context"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
v1 "k8s.io/api/core/v1"
@@ -19,19 +20,12 @@ import (
)
const (
serviceSyncerController = "service-syncer-controller"
serviceFinalizerName = "service.k3k.io/finalizer"
serviceControllerName = "service-syncer-controller"
serviceFinalizerName = "service.k3k.io/finalizer"
)
type ServiceReconciler struct {
clusterName string
clusterNamespace string
virtualClient ctrlruntimeclient.Client
hostClient ctrlruntimeclient.Client
Scheme *runtime.Scheme
HostScheme *runtime.Scheme
Translator translate.ToHostTranslator
*SyncerContext
}
// AddServiceSyncer adds service syncer controller to the manager of the virtual cluster
@@ -42,24 +36,25 @@ func AddServiceSyncer(ctx context.Context, virtMgr, hostMgr manager.Manager, clu
}
reconciler := ServiceReconciler{
clusterName: clusterName,
clusterNamespace: clusterNamespace,
virtualClient: virtMgr.GetClient(),
hostClient: hostMgr.GetClient(),
Scheme: virtMgr.GetScheme(),
HostScheme: hostMgr.GetScheme(),
Translator: translator,
SyncerContext: &SyncerContext{
ClusterName: clusterName,
ClusterNamespace: clusterNamespace,
VirtualClient: virtMgr.GetClient(),
HostClient: hostMgr.GetClient(),
Translator: translator,
},
}
name := reconciler.Translator.TranslateName(clusterNamespace, serviceControllerName)
return ctrl.NewControllerManagedBy(virtMgr).
Named(serviceSyncerController).
For(&v1.Service{}).
Named(name).
For(&v1.Service{}).WithEventFilter(predicate.NewPredicateFuncs(reconciler.filterResources)).
Complete(&reconciler)
}
func (r *ServiceReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.clusterName, "clusterNamespace", r.clusterNamespace)
log := ctrl.LoggerFrom(ctx).WithValues("cluster", r.ClusterName, "clusterNamespace", r.ClusterNamespace)
ctx = ctrl.LoggerInto(ctx, log)
if req.Name == "kubernetes" || req.Name == "kube-dns" {
@@ -71,31 +66,29 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req reconcile.Request
cluster v1alpha1.Cluster
)
if err := r.hostClient.Get(ctx, types.NamespacedName{Name: r.clusterName, Namespace: r.clusterNamespace}, &cluster); err != nil {
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return reconcile.Result{}, err
}
if err := r.virtualClient.Get(ctx, req.NamespacedName, &virtService); err != nil {
if err := r.VirtualClient.Get(ctx, req.NamespacedName, &virtService); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
syncedService := r.service(&virtService)
if err := controllerutil.SetControllerReference(&cluster, syncedService, r.HostScheme); err != nil {
if err := controllerutil.SetControllerReference(&cluster, syncedService, r.HostClient.Scheme()); err != nil {
return reconcile.Result{}, err
}
// handle deletion
if !virtService.DeletionTimestamp.IsZero() {
// deleting the synced service if exists
if err := r.hostClient.Delete(ctx, syncedService); err != nil {
if err := r.HostClient.Delete(ctx, syncedService); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
// remove the finalizer after cleaning up the synced service
if controllerutil.ContainsFinalizer(&virtService, serviceFinalizerName) {
controllerutil.RemoveFinalizer(&virtService, serviceFinalizerName)
if err := r.virtualClient.Update(ctx, &virtService); err != nil {
if controllerutil.RemoveFinalizer(&virtService, serviceFinalizerName) {
if err := r.VirtualClient.Update(ctx, &virtService); err != nil {
return reconcile.Result{}, err
}
}
@@ -104,20 +97,18 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req reconcile.Request
}
// Add finalizer if it does not exist
if !controllerutil.ContainsFinalizer(&virtService, serviceFinalizerName) {
controllerutil.AddFinalizer(&virtService, serviceFinalizerName)
if err := r.virtualClient.Update(ctx, &virtService); err != nil {
if controllerutil.AddFinalizer(&virtService, serviceFinalizerName) {
if err := r.VirtualClient.Update(ctx, &virtService); err != nil {
return reconcile.Result{}, err
}
}
// create or update the service on host
var hostService v1.Service
if err := r.hostClient.Get(ctx, types.NamespacedName{Name: syncedService.Name, Namespace: r.clusterNamespace}, &hostService); err != nil {
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: syncedService.Name, Namespace: r.ClusterNamespace}, &hostService); err != nil {
if apierrors.IsNotFound(err) {
log.Info("creating the service for the first time on the host cluster")
return reconcile.Result{}, r.hostClient.Create(ctx, syncedService)
return reconcile.Result{}, r.HostClient.Create(ctx, syncedService)
}
return reconcile.Result{}, err
@@ -125,7 +116,32 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req reconcile.Request
log.Info("updating service on the host cluster")
return reconcile.Result{}, r.hostClient.Update(ctx, syncedService)
return reconcile.Result{}, r.HostClient.Update(ctx, syncedService)
}
func (r *ServiceReconciler) filterResources(object ctrlruntimeclient.Object) bool {
var cluster v1alpha1.Cluster
ctx := context.Background()
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: r.ClusterName, Namespace: r.ClusterNamespace}, &cluster); err != nil {
return false
}
// check for serviceSyncConfig
syncConfig := cluster.Spec.Sync.Services
// If syncing is disabled, only process deletions to allow for cleanup.
if !syncConfig.Enabled {
return object.GetDeletionTimestamp() != nil
}
labelSelector := labels.SelectorFromSet(syncConfig.Selector)
if labelSelector.Empty() {
return true
}
return labelSelector.Matches(labels.Set(object.GetLabels()))
}
func (s *ServiceReconciler) service(obj *v1.Service) *v1.Service {
@@ -0,0 +1,276 @@
package syncer_test
import (
"context"
"fmt"
"time"
"k8s.io/apimachinery/pkg/util/intstr"
"sigs.k8s.io/controller-runtime/pkg/client"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/rancher/k3k/k3k-kubelet/controller/syncer"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
var ServiceTests = func() {
var (
namespace string
cluster v1alpha1.Cluster
)
BeforeEach(func() {
ctx := context.Background()
ns := v1.Namespace{
ObjectMeta: metav1.ObjectMeta{GenerateName: "ns-"},
}
err := hostTestEnv.k8sClient.Create(ctx, &ns)
Expect(err).NotTo(HaveOccurred())
namespace = ns.Name
cluster = v1alpha1.Cluster{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "cluster-",
Namespace: namespace,
},
Spec: v1alpha1.ClusterSpec{
Sync: v1alpha1.SyncConfig{
Services: v1alpha1.ServiceSyncConfig{
Enabled: true,
},
},
},
}
err = hostTestEnv.k8sClient.Create(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
err = syncer.AddServiceSyncer(ctx, virtManager, hostManager, cluster.Name, cluster.Namespace)
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
ns := v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
err := hostTestEnv.k8sClient.Delete(context.Background(), &ns)
Expect(err).NotTo(HaveOccurred())
})
It("creates a service on the host cluster", func() {
ctx := context.Background()
service := &v1.Service{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "service-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Spec: v1.ServiceSpec{
Type: v1.ServiceTypeNodePort,
Ports: []v1.ServicePort{
{
Name: "test-port",
Port: 8888,
TargetPort: intstr.FromInt32(8888),
},
},
},
}
err := virtTestEnv.k8sClient.Create(ctx, service)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created service %s in virtual cluster", service.Name))
var hostService v1.Service
hostServiceName := translateName(cluster, service.Namespace, service.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostServiceName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostService)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created Service %s in host cluster", hostServiceName))
Expect(hostService.Spec.Type).To(Equal(v1.ServiceTypeNodePort))
Expect(hostService.Spec.Ports[0].Name).To(Equal("test-port"))
Expect(hostService.Spec.Ports[0].Port).To(Equal(int32(8888)))
GinkgoWriter.Printf("labels: %v\n", hostService.Labels)
})
It("updates a service on the host cluster", func() {
ctx := context.Background()
service := &v1.Service{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "service-",
Namespace: "default",
Labels: map[string]string{
"foo": "bar",
},
},
Spec: v1.ServiceSpec{
Type: v1.ServiceTypeNodePort,
Ports: []v1.ServicePort{
{
Name: "test-port",
Port: 8888,
TargetPort: intstr.FromInt32(8888),
},
},
},
}
err := virtTestEnv.k8sClient.Create(ctx, service)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created service %s in virtual cluster", service.Name))
var hostService v1.Service
hostServiceName := translateName(cluster, service.Namespace, service.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostServiceName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostService)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created Service %s in host cluster", hostServiceName))
Expect(hostService.Spec.Type).To(Equal(v1.ServiceTypeNodePort))
Expect(hostService.Spec.Ports[0].Name).To(Equal("test-port"))
Expect(hostService.Spec.Ports[0].Port).To(Equal(int32(8888)))
key := client.ObjectKeyFromObject(service)
err = virtTestEnv.k8sClient.Get(ctx, key, service)
Expect(err).NotTo(HaveOccurred())
service.Spec.Ports[0].Name = "test-port-updated"
// update virtual service
err = virtTestEnv.k8sClient.Update(ctx, service)
Expect(err).NotTo(HaveOccurred())
// check hostService
Eventually(func() string {
key := client.ObjectKey{Name: hostServiceName, Namespace: namespace}
err = hostTestEnv.k8sClient.Get(ctx, key, &hostService)
Expect(err).NotTo(HaveOccurred())
return hostService.Spec.Ports[0].Name
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(Equal("test-port-updated"))
})
It("deletes a service on the host cluster", func() {
ctx := context.Background()
service := &v1.Service{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "service-",
Namespace: "default",
},
Spec: v1.ServiceSpec{
Type: v1.ServiceTypeNodePort,
Ports: []v1.ServicePort{
{
Name: "test-port",
Port: 8888,
TargetPort: intstr.FromInt32(8888),
},
},
},
}
err := virtTestEnv.k8sClient.Create(ctx, service)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created service %s in virtual cluster", service.Name))
var hostService v1.Service
hostServiceName := translateName(cluster, service.Namespace, service.Name)
Eventually(func() error {
key := client.ObjectKey{Name: hostServiceName, Namespace: namespace}
return hostTestEnv.k8sClient.Get(ctx, key, &hostService)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeNil())
By(fmt.Sprintf("Created service %s in host cluster", hostServiceName))
Expect(hostService.Spec.Type).To(Equal(v1.ServiceTypeNodePort))
Expect(hostService.Spec.Ports[0].Name).To(Equal("test-port"))
Expect(hostService.Spec.Ports[0].Port).To(Equal(int32(8888)))
err = virtTestEnv.k8sClient.Delete(ctx, service)
Expect(err).NotTo(HaveOccurred())
Eventually(func() bool {
key := client.ObjectKey{Name: hostServiceName, Namespace: namespace}
err := hostTestEnv.k8sClient.Get(ctx, key, &hostService)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
It("will not create a service on the host cluster if disabled", func() {
ctx := context.Background()
cluster.Spec.Sync.Services.Enabled = false
err := hostTestEnv.k8sClient.Update(ctx, &cluster)
Expect(err).NotTo(HaveOccurred())
service := &v1.Service{
ObjectMeta: metav1.ObjectMeta{
GenerateName: "service-",
Namespace: "default",
},
Spec: v1.ServiceSpec{
Type: v1.ServiceTypeNodePort,
Ports: []v1.ServicePort{
{
Name: "test-port",
Port: 8888,
TargetPort: intstr.FromInt32(8888),
},
},
},
}
err = virtTestEnv.k8sClient.Create(ctx, service)
Expect(err).NotTo(HaveOccurred())
By(fmt.Sprintf("Created service %s in virtual cluster", service.Name))
var hostService v1.Service
hostServiceName := translateName(cluster, service.Namespace, service.Name)
Eventually(func() bool {
key := client.ObjectKey{Name: hostServiceName, Namespace: namespace}
err = hostTestEnv.k8sClient.Get(ctx, key, &hostService)
return apierrors.IsNotFound(err)
}).
WithPolling(time.Millisecond * 300).
WithTimeout(time.Second * 10).
Should(BeTrue())
})
}
+15
View File
@@ -0,0 +1,15 @@
package syncer
import (
"sigs.k8s.io/controller-runtime/pkg/client"
"github.com/rancher/k3k/k3k-kubelet/translate"
)
type SyncerContext struct {
ClusterName string
ClusterNamespace string
VirtualClient client.Client
HostClient client.Client
Translator translate.ToHostTranslator
}
@@ -1,4 +1,4 @@
package controller_test
package syncer_test
import (
"context"
@@ -19,6 +19,7 @@ import (
ctrl "sigs.k8s.io/controller-runtime"
metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
"github.com/rancher/k3k/k3k-kubelet/translate"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
. "github.com/onsi/ginkgo/v2"
@@ -91,7 +92,7 @@ func NewTestEnv() *TestEnv {
By("bootstrapping test environment")
testEnv := &envtest.Environment{
CRDDirectoryPaths: []string{filepath.Join("..", "..", "charts", "k3k", "crds")},
CRDDirectoryPaths: []string{filepath.Join("..", "..", "..", "charts", "k3k", "crds")},
ErrorIfCRDPathMissing: true,
BinaryAssetsDirectory: tempDir,
Scheme: buildScheme(),
@@ -165,5 +166,19 @@ var _ = Describe("Kubelet Controller", func() {
cancel()
})
Describe("PriorityClass", PriorityClassTests)
Describe("PriorityClass Syncer", PriorityClassTests)
Describe("ConfigMap Syncer", ConfigMapTests)
Describe("Secret Syncer", SecretTests)
Describe("Service Syncer", ServiceTests)
Describe("Ingress Syncer", IngressTests)
Describe("PersistentVolumeClaim Syncer", PVCTests)
})
func translateName(cluster v1alpha1.Cluster, namespace, name string) string {
translator := translate.ToHostTranslator{
ClusterName: cluster.Name,
ClusterNamespace: cluster.Namespace,
}
return translator.TranslateName(namespace, name)
}
+56 -23
View File
@@ -35,7 +35,7 @@ import (
ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
ctrlserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
k3kkubeletcontroller "github.com/rancher/k3k/k3k-kubelet/controller"
"github.com/rancher/k3k/k3k-kubelet/controller/syncer"
k3kwebhook "github.com/rancher/k3k/k3k-kubelet/controller/webhook"
"github.com/rancher/k3k/k3k-kubelet/provider"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
@@ -156,28 +156,8 @@ func newKubelet(ctx context.Context, c *config, logger *k3klog.Logger) (*kubelet
return nil, errors.New("unable to add pod mutator webhook for virtual cluster: " + err.Error())
}
logger.Info("adding service syncer controller")
if err := k3kkubeletcontroller.AddServiceSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return nil, errors.New("failed to add service syncer controller: " + err.Error())
}
logger.Info("adding pvc syncer controller")
if err := k3kkubeletcontroller.AddPVCSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return nil, errors.New("failed to add pvc syncer controller: " + err.Error())
}
logger.Info("adding pod pvc controller")
if err := k3kkubeletcontroller.AddPodPVCController(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return nil, errors.New("failed to add pod pvc controller: " + err.Error())
}
logger.Info("adding priorityclass controller")
if err := k3kkubeletcontroller.AddPriorityClassReconciler(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return nil, errors.New("failed to add priorityclass controller: " + err.Error())
if err := addControllers(ctx, hostMgr, virtualMgr, c, hostClient); err != nil {
return nil, errors.New("failed to add controller: " + err.Error())
}
clusterIP, err := clusterIP(ctx, c.ServiceName, c.ClusterNamespace, hostClient)
@@ -447,3 +427,56 @@ func loadTLSConfig(ctx context.Context, hostClient ctrlruntimeclient.Client, clu
Certificates: []tls.Certificate{clientCert},
}, nil
}
func addControllers(ctx context.Context, hostMgr, virtualMgr manager.Manager, c *config, hostClient ctrlruntimeclient.Client) error {
var cluster v1alpha1.Cluster
objKey := types.NamespacedName{
Namespace: c.ClusterNamespace,
Name: c.ClusterName,
}
if err := hostClient.Get(ctx, objKey, &cluster); err != nil {
return err
}
if err := syncer.AddConfigMapSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return errors.New("failed to add configmap global syncer: " + err.Error())
}
if err := syncer.AddSecretSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return errors.New("failed to add secret global syncer: " + err.Error())
}
logger.Info("adding service syncer controller")
if err := syncer.AddServiceSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return errors.New("failed to add service syncer controller: " + err.Error())
}
logger.Info("adding ingress syncer controller")
if err := syncer.AddIngressSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return errors.New("failed to add ingress syncer controller: " + err.Error())
}
logger.Info("adding pvc syncer controller")
if err := syncer.AddPVCSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return errors.New("failed to add pvc syncer controller: " + err.Error())
}
logger.Info("adding pod pvc controller")
if err := syncer.AddPodPVCController(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return errors.New("failed to add pod pvc controller: " + err.Error())
}
logger.Info("adding priorityclass controller")
if err := syncer.AddPriorityClassSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
return errors.New("failed to add priorityclass controller: " + err.Error())
}
return nil
}
+10 -200
View File
@@ -18,7 +18,6 @@ import (
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/selection"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/rest"
@@ -32,13 +31,11 @@ import (
dto "github.com/prometheus/client_model/go"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
cv1 "k8s.io/client-go/kubernetes/typed/core/v1"
compbasemetrics "k8s.io/component-base/metrics"
stats "k8s.io/kubelet/pkg/apis/stats/v1alpha1"
"github.com/rancher/k3k/k3k-kubelet/controller"
"github.com/rancher/k3k/k3k-kubelet/controller/webhook"
"github.com/rancher/k3k/k3k-kubelet/provider/collectors"
"github.com/rancher/k3k/k3k-kubelet/translate"
@@ -53,10 +50,10 @@ var _ nodeutil.Provider = (*Provider)(nil)
// Provider implements nodetuil.Provider from virtual Kubelet.
// TODO: Implement NotifyPods and the required usage so that this can be an async provider
type Provider struct {
Handler controller.ControllerHandler
Translator translate.ToHostTranslator
HostClient client.Client
VirtualClient client.Client
VirtualManager manager.Manager
ClientConfig rest.Config
CoreClient cv1.CoreV1Interface
ClusterNamespace string
@@ -80,16 +77,9 @@ func New(hostConfig rest.Config, hostMgr, virtualMgr manager.Manager, logger *k3
}
p := Provider{
Handler: controller.ControllerHandler{
Mgr: virtualMgr,
Scheme: *virtualMgr.GetScheme(),
HostClient: hostMgr.GetClient(),
VirtualClient: virtualMgr.GetClient(),
Translator: translator,
Logger: logger,
},
HostClient: hostMgr.GetClient(),
VirtualClient: virtualMgr.GetClient(),
VirtualManager: virtualMgr,
Translator: translator,
ClientConfig: hostConfig,
CoreClient: coreClient,
@@ -397,7 +387,7 @@ func (p *Provider) createPod(ctx context.Context, pod *corev1.Pod) error {
}
// volumes will often refer to resources in the virtual cluster, but instead need to refer to the sync'd
// host cluster version
if err := p.transformVolumes(ctx, pod.Namespace, tPod.Spec.Volumes); err != nil {
if err := p.transformVolumes(pod.Namespace, tPod.Spec.Volumes); err != nil {
return fmt.Errorf("unable to sync volumes for pod %s/%s: %w", pod.Namespace, pod.Name, err)
}
// sync serviceaccount token to a the host cluster
@@ -405,6 +395,10 @@ func (p *Provider) createPod(ctx context.Context, pod *corev1.Pod) error {
return fmt.Errorf("unable to transform tokens for pod %s/%s: %w", pod.Namespace, pod.Name, err)
}
for i, imagePullSecret := range tPod.Spec.ImagePullSecrets {
tPod.Spec.ImagePullSecrets[i].Name = p.Translator.TranslateName(pod.Namespace, imagePullSecret.Name)
}
// inject networking information to the pod including the virtual cluster controlplane endpoint
configureNetworking(tPod, pod.Name, pod.Namespace, p.serverIP, p.dnsIP)
@@ -450,58 +444,22 @@ func (p *Provider) withRetry(ctx context.Context, f func(context.Context, *corev
// transformVolumes changes the volumes to the representation in the host cluster. Will return an error
// if one/more volumes couldn't be transformed
func (p *Provider) transformVolumes(ctx context.Context, podNamespace string, volumes []corev1.Volume) error {
func (p *Provider) transformVolumes(podNamespace string, volumes []corev1.Volume) error {
for _, volume := range volumes {
var optional bool
if strings.HasPrefix(volume.Name, kubeAPIAccessPrefix) {
continue
}
// note: this needs to handle downward api volumes as well, but more thought is needed on how to do that
if volume.ConfigMap != nil {
if volume.ConfigMap.Optional != nil {
optional = *volume.ConfigMap.Optional
}
if err := p.syncConfigmap(ctx, podNamespace, volume.ConfigMap.Name, optional); err != nil {
return fmt.Errorf("unable to sync configmap volume %s: %w", volume.Name, err)
}
volume.ConfigMap.Name = p.Translator.TranslateName(podNamespace, volume.ConfigMap.Name)
} else if volume.Secret != nil {
if volume.Secret.Optional != nil {
optional = *volume.Secret.Optional
}
if err := p.syncSecret(ctx, podNamespace, volume.Secret.SecretName, optional); err != nil {
return fmt.Errorf("unable to sync secret volume %s: %w", volume.Name, err)
}
volume.Secret.SecretName = p.Translator.TranslateName(podNamespace, volume.Secret.SecretName)
} else if volume.Projected != nil {
for _, source := range volume.Projected.Sources {
if source.ConfigMap != nil {
if source.ConfigMap.Optional != nil {
optional = *source.ConfigMap.Optional
}
configMapName := source.ConfigMap.Name
if err := p.syncConfigmap(ctx, podNamespace, configMapName, optional); err != nil {
return fmt.Errorf("unable to sync projected configmap %s: %w", configMapName, err)
}
source.ConfigMap.Name = p.Translator.TranslateName(podNamespace, configMapName)
source.ConfigMap.Name = p.Translator.TranslateName(podNamespace, source.ConfigMap.Name)
} else if source.Secret != nil {
if source.Secret.Optional != nil {
optional = *source.Secret.Optional
}
secretName := source.Secret.Name
if err := p.syncSecret(ctx, podNamespace, secretName, optional); err != nil {
return fmt.Errorf("unable to sync projected secret %s: %w", secretName, err)
}
source.Secret.Name = p.Translator.TranslateName(podNamespace, secretName)
source.Secret.Name = p.Translator.TranslateName(podNamespace, source.Secret.Name)
}
}
} else if volume.PersistentVolumeClaim != nil {
@@ -524,57 +482,6 @@ func (p *Provider) transformVolumes(ctx context.Context, podNamespace string, vo
return nil
}
// syncConfigmap will add the configmap object to the queue of the syncer controller to be synced to the host cluster
func (p *Provider) syncConfigmap(ctx context.Context, podNamespace string, configMapName string, optional bool) error {
var configMap corev1.ConfigMap
nsName := types.NamespacedName{
Namespace: podNamespace,
Name: configMapName,
}
if err := p.VirtualClient.Get(ctx, nsName, &configMap); err != nil {
// check if its optional configmap
if apierrors.IsNotFound(err) && optional {
return nil
}
return fmt.Errorf("unable to get configmap to sync %s/%s: %w", nsName.Namespace, nsName.Name, err)
}
if err := p.Handler.AddResource(ctx, &configMap); err != nil {
return fmt.Errorf("unable to add configmap to sync %s/%s: %w", nsName.Namespace, nsName.Name, err)
}
return nil
}
// syncSecret will add the secret object to the queue of the syncer controller to be synced to the host cluster
func (p *Provider) syncSecret(ctx context.Context, podNamespace string, secretName string, optional bool) error {
p.logger.Infow("Syncing secret", "Name", secretName, "Namespace", podNamespace, "optional", optional)
var secret corev1.Secret
nsName := types.NamespacedName{
Namespace: podNamespace,
Name: secretName,
}
if err := p.VirtualClient.Get(ctx, nsName, &secret); err != nil {
if apierrors.IsNotFound(err) && optional {
return nil
}
return fmt.Errorf("unable to get secret to sync %s/%s: %w", nsName.Namespace, nsName.Name, err)
}
if err := p.Handler.AddResource(ctx, &secret); err != nil {
return fmt.Errorf("unable to add secret to sync %s/%s: %w", nsName.Namespace, nsName.Name, err)
}
return nil
}
// UpdatePod executes updatePod with retry
func (p *Provider) UpdatePod(ctx context.Context, pod *corev1.Pod) error {
return p.withRetry(ctx, p.updatePod, pod)
@@ -690,81 +597,11 @@ func (p *Provider) deletePod(ctx context.Context, pod *corev1.Pod) error {
return fmt.Errorf("unable to delete pod %s/%s: %w", pod.Namespace, pod.Name, err)
}
if err = p.pruneUnusedVolumes(ctx, pod); err != nil {
// note that we don't return an error here. The pod was successfully deleted, another process
// should clean this without affecting the user
p.logger.Errorf("failed to prune leftover volumes for %s/%s: %w, resources may be left", pod.Namespace, pod.Name, err)
}
p.logger.Infof("Deleted pod %s", pod.Name)
return nil
}
// pruneUnusedVolumes removes volumes in use by pod that aren't used by any other pods
func (p *Provider) pruneUnusedVolumes(ctx context.Context, pod *corev1.Pod) error {
rawSecrets, rawConfigMaps := getSecretsAndConfigmaps(pod)
// since this pod was removed, originally mark all of the secrets/configmaps it uses as eligible
// for pruning
pruneSecrets := sets.Set[string]{}.Insert(rawSecrets...)
pruneConfigMap := sets.Set[string]{}.Insert(rawConfigMaps...)
var pods corev1.PodList
// only pods in the same namespace could be using secrets/configmaps that this pod is using
err := p.VirtualClient.List(ctx, &pods, &client.ListOptions{
Namespace: pod.Namespace,
})
if err != nil {
return fmt.Errorf("unable to list pods: %w", err)
}
for _, vPod := range pods.Items {
if vPod.Name == pod.Name {
continue
}
secrets, configMaps := getSecretsAndConfigmaps(&vPod)
pruneSecrets.Delete(secrets...)
pruneConfigMap.Delete(configMaps...)
}
for _, secretName := range pruneSecrets.UnsortedList() {
var secret corev1.Secret
key := types.NamespacedName{
Name: secretName,
Namespace: pod.Namespace,
}
if err := p.VirtualClient.Get(ctx, key, &secret); err != nil {
return fmt.Errorf("unable to get secret %s/%s for pod volume: %w", pod.Namespace, secretName, err)
}
if err = p.Handler.RemoveResource(ctx, &secret); err != nil {
return fmt.Errorf("unable to remove secret %s/%s for pod volume: %w", pod.Namespace, secretName, err)
}
}
for _, configMapName := range pruneConfigMap.UnsortedList() {
var configMap corev1.ConfigMap
key := types.NamespacedName{
Name: configMapName,
Namespace: pod.Namespace,
}
if err := p.VirtualClient.Get(ctx, key, &configMap); err != nil {
return fmt.Errorf("unable to get configMap %s/%s for pod volume: %w", pod.Namespace, configMapName, err)
}
if err = p.Handler.RemoveResource(ctx, &configMap); err != nil {
return fmt.Errorf("unable to remove configMap %s/%s for pod volume: %w", pod.Namespace, configMapName, err)
}
}
return nil
}
// GetPod retrieves a pod by name from the provider (can be cached).
// The Pod returned is expected to be immutable, and may be accessed
// concurrently outside of the calling goroutine. Therefore it is recommended
@@ -923,33 +760,6 @@ func mergeEnvVars(orig, updated []corev1.EnvVar) []corev1.EnvVar {
return orig
}
// getSecretsAndConfigmaps retrieves a list of all secrets/configmaps that are in use by a given pod. Useful
// for removing/seeing which virtual cluster resources need to be in the host cluster.
func getSecretsAndConfigmaps(pod *corev1.Pod) ([]string, []string) {
var (
secrets []string
configMaps []string
)
for _, volume := range pod.Spec.Volumes {
if volume.Secret != nil {
secrets = append(secrets, volume.Secret.SecretName)
} else if volume.ConfigMap != nil {
configMaps = append(configMaps, volume.ConfigMap.Name)
} else if volume.Projected != nil {
for _, source := range volume.Projected.Sources {
if source.ConfigMap != nil {
configMaps = append(configMaps, source.ConfigMap.Name)
} else if source.Secret != nil {
secrets = append(secrets, source.Secret.Name)
}
}
}
}
return secrets, configMaps
}
// configureFieldPathEnv will retrieve all annotations created by the pod mutator webhook
// to assign env fieldpaths to pods, it will also make sure to change the metadata.name and metadata.namespace to the
// assigned annotations
+117
View File
@@ -176,6 +176,118 @@ type ClusterSpec struct {
//
// +optional
CustomCAs CustomCAs `json:"customCAs,omitempty"`
// Sync specifies the resources types that will be synced from virtual cluster to host cluster.
//
// +kubebuilder:default={}
// +optional
Sync SyncConfig `json:"sync,omitempty"`
}
// SyncConfig will contain the resources that should be synced from virtual cluster to host cluster.
type SyncConfig struct {
// Services resources sync configuration.
//
// +kubebuilder:default={"enabled": true}
Services ServiceSyncConfig `json:"services,omitempty"`
// ConfigMaps resources sync configuration.
//
// +kubebuilder:default={"enabled": true}
ConfigMaps ConfigMapSyncConfig `json:"configmaps,omitempty"`
// Secrets resources sync configuration.
//
// +kubebuilder:default={"enabled": true}
Secrets SecretSyncConfig `json:"secrets,omitempty"`
// Ingresses resources sync configuration.
//
// +kubebuilder:default={"enabled": false}
Ingresses IngressSyncConfig `json:"ingresses,omitempty"`
// PersistentVolumeClaims resources sync configuration.
//
// +kubebuilder:default={"enabled": false}
PersistentVolumeClaims PersistentVolumeClaimSyncConfig `json:"persistentVolumeClaims,omitempty"`
// PriorityClasses resources sync configuration.
//
// +kubebuilder:default={"enabled": false}
PriorityClasses PriorityClassSyncConfig `json:"priorityClasses,omitempty"`
}
// SecretSyncConfig specifies the sync options for services.
type SecretSyncConfig struct {
// Enabled is an on/off switch for syncing resources.
// +optional
Enabled bool `json:"enabled,omitempty"`
// Selector specifies set of labels of the resources that will be synced, if empty
// then all resources of the given type will be synced.
//
// +optional
Selector map[string]string `json:"selector,omitempty"`
}
// ServiceSyncConfig specifies the sync options for services.
type ServiceSyncConfig struct {
// Enabled is an on/off switch for syncing resources.
// +optional
Enabled bool `json:"enabled,omitempty"`
// Selector specifies set of labels of the resources that will be synced, if empty
// then all resources of the given type will be synced.
//
// +optional
Selector map[string]string `json:"selector,omitempty"`
}
// ConfigMapSyncConfig specifies the sync options for services.
type ConfigMapSyncConfig struct {
// Enabled is an on/off switch for syncing resources.
// +optional
Enabled bool `json:"enabled,omitempty"`
// Selector specifies set of labels of the resources that will be synced, if empty
// then all resources of the given type will be synced.
//
// +optional
Selector map[string]string `json:"selector,omitempty"`
}
// IngressSyncConfig specifies the sync options for services.
type IngressSyncConfig struct {
// Enabled is an on/off switch for syncing resources.
// +optional
Enabled bool `json:"enabled,omitempty"`
// Selector specifies set of labels of the resources that will be synced, if empty
// then all resources of the given type will be synced.
//
// +optional
Selector map[string]string `json:"selector,omitempty"`
}
// PersistentVolumeClaimSyncConfig specifies the sync options for services.
type PersistentVolumeClaimSyncConfig struct {
// Enabled is an on/off switch for syncing resources.
// +optional
Enabled bool `json:"enabled,omitempty"`
// Selector specifies set of labels of the resources that will be synced, if empty
// then all resources of the given type will be synced.
//
// +optional
Selector map[string]string `json:"selector,omitempty"`
}
// PriorityClassSyncConfig specifies the sync options for services.
type PriorityClassSyncConfig struct {
// Enabled is an on/off switch for syncing resources.
// +optional
Enabled bool `json:"enabled,omitempty"`
// Selector specifies set of labels of the resources that will be synced, if empty
// then all resources of the given type will be synced.
//
// +optional
Selector map[string]string `json:"selector,omitempty"`
}
// ClusterMode is the possible provisioning mode of a Cluster.
@@ -484,6 +596,11 @@ type VirtualClusterPolicySpec struct {
//
// +optional
PodSecurityAdmissionLevel *PodSecurityAdmissionLevel `json:"podSecurityAdmissionLevel,omitempty"`
// Sync specifies the synchronization configuration for resources that can be applied to clusters
//
// +optional
Sync SyncConfig `json:"sync,omitempty"`
}
// PodSecurityAdmissionLevel is the policy level applied to the pods in the namespace.
@@ -164,6 +164,7 @@ func (in *ClusterSpec) DeepCopyInto(out *ClusterSpec) {
}
}
out.CustomCAs = in.CustomCAs
in.Sync.DeepCopyInto(&out.Sync)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ClusterSpec.
@@ -203,6 +204,28 @@ func (in *ClusterStatus) DeepCopy() *ClusterStatus {
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ConfigMapSyncConfig) DeepCopyInto(out *ConfigMapSyncConfig) {
*out = *in
if in.Selector != nil {
in, out := &in.Selector, &out.Selector
*out = make(map[string]string, len(*in))
for key, val := range *in {
(*out)[key] = val
}
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ConfigMapSyncConfig.
func (in *ConfigMapSyncConfig) DeepCopy() *ConfigMapSyncConfig {
if in == nil {
return nil
}
out := new(ConfigMapSyncConfig)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *CredentialSource) DeepCopyInto(out *CredentialSource) {
*out = *in
@@ -307,6 +330,28 @@ func (in *IngressConfig) DeepCopy() *IngressConfig {
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *IngressSyncConfig) DeepCopyInto(out *IngressSyncConfig) {
*out = *in
if in.Selector != nil {
in, out := &in.Selector, &out.Selector
*out = make(map[string]string, len(*in))
for key, val := range *in {
(*out)[key] = val
}
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new IngressSyncConfig.
func (in *IngressSyncConfig) DeepCopy() *IngressSyncConfig {
if in == nil {
return nil
}
out := new(IngressSyncConfig)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *LoadBalancerConfig) DeepCopyInto(out *LoadBalancerConfig) {
*out = *in
@@ -377,6 +422,115 @@ func (in *PersistenceConfig) DeepCopy() *PersistenceConfig {
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *PersistentVolumeClaimSyncConfig) DeepCopyInto(out *PersistentVolumeClaimSyncConfig) {
*out = *in
if in.Selector != nil {
in, out := &in.Selector, &out.Selector
*out = make(map[string]string, len(*in))
for key, val := range *in {
(*out)[key] = val
}
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PersistentVolumeClaimSyncConfig.
func (in *PersistentVolumeClaimSyncConfig) DeepCopy() *PersistentVolumeClaimSyncConfig {
if in == nil {
return nil
}
out := new(PersistentVolumeClaimSyncConfig)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *PriorityClassSyncConfig) DeepCopyInto(out *PriorityClassSyncConfig) {
*out = *in
if in.Selector != nil {
in, out := &in.Selector, &out.Selector
*out = make(map[string]string, len(*in))
for key, val := range *in {
(*out)[key] = val
}
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PriorityClassSyncConfig.
func (in *PriorityClassSyncConfig) DeepCopy() *PriorityClassSyncConfig {
if in == nil {
return nil
}
out := new(PriorityClassSyncConfig)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *SecretSyncConfig) DeepCopyInto(out *SecretSyncConfig) {
*out = *in
if in.Selector != nil {
in, out := &in.Selector, &out.Selector
*out = make(map[string]string, len(*in))
for key, val := range *in {
(*out)[key] = val
}
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SecretSyncConfig.
func (in *SecretSyncConfig) DeepCopy() *SecretSyncConfig {
if in == nil {
return nil
}
out := new(SecretSyncConfig)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *ServiceSyncConfig) DeepCopyInto(out *ServiceSyncConfig) {
*out = *in
if in.Selector != nil {
in, out := &in.Selector, &out.Selector
*out = make(map[string]string, len(*in))
for key, val := range *in {
(*out)[key] = val
}
}
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ServiceSyncConfig.
func (in *ServiceSyncConfig) DeepCopy() *ServiceSyncConfig {
if in == nil {
return nil
}
out := new(ServiceSyncConfig)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *SyncConfig) DeepCopyInto(out *SyncConfig) {
*out = *in
in.Services.DeepCopyInto(&out.Services)
in.ConfigMaps.DeepCopyInto(&out.ConfigMaps)
in.Secrets.DeepCopyInto(&out.Secrets)
in.Ingresses.DeepCopyInto(&out.Ingresses)
in.PersistentVolumeClaims.DeepCopyInto(&out.PersistentVolumeClaims)
in.PriorityClasses.DeepCopyInto(&out.PriorityClasses)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SyncConfig.
func (in *SyncConfig) DeepCopy() *SyncConfig {
if in == nil {
return nil
}
out := new(SyncConfig)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *VirtualClusterPolicy) DeepCopyInto(out *VirtualClusterPolicy) {
*out = *in
@@ -461,6 +615,7 @@ func (in *VirtualClusterPolicySpec) DeepCopyInto(out *VirtualClusterPolicySpec)
*out = new(PodSecurityAdmissionLevel)
**out = **in
}
in.Sync.DeepCopyInto(&out.Sync)
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new VirtualClusterPolicySpec.
+5
View File
@@ -367,6 +367,11 @@ func (s *SharedAgent) role(ctx context.Context) error {
Resources: []string{"persistentvolumeclaims", "pods", "pods/log", "pods/attach", "pods/exec", "pods/ephemeralcontainers", "secrets", "configmaps", "services"},
Verbs: []string{"*"},
},
{
APIGroups: []string{"networking.k8s.io"},
Resources: []string{"ingresses"},
Verbs: []string{"*"},
},
{
APIGroups: []string{"k3k.io"},
Resources: []string{"clusters"},
+5
View File
@@ -718,6 +718,11 @@ func (c *ClusterReconciler) validate(cluster *v1alpha1.Cluster, policy v1alpha1.
}
}
// validate sync policy
if !equality.Semantic.DeepEqual(cluster.Spec.Sync, policy.Spec.Sync) {
return fmt.Errorf("sync configuration %v is not allowed by the policy %q", cluster.Spec.Sync, policy.Name)
}
return nil
}