Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Created new Broadcaser which start an informer to watch Nodes for Res…
…ourceOffer generation/update Add logic reacting after node add, update and delete Add ReadResources() function to read safely the cluster resources Added clusterconfig watcher to get scaling percentage. Refactor of resource writing to manage all kind of resources. Added new tests and fixed linting Added Node Phase check. Write Update and Delete Node tests.
- Loading branch information
Showing
6 changed files
with
337 additions
and
34 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,184 @@ | ||
| package resourcerequestoperator | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "sync" | ||
|
|
||
| crdclient "github.com/liqotech/liqo/pkg/crdClient" | ||
|
|
||
| "k8s.io/client-go/kubernetes/scheme" | ||
| "k8s.io/client-go/rest" | ||
|
|
||
| configv1alpha1 "github.com/liqotech/liqo/apis/config/v1alpha1" | ||
| "github.com/liqotech/liqo/pkg/utils" | ||
|
|
||
| corev1 "k8s.io/api/core/v1" | ||
| "k8s.io/apimachinery/pkg/api/resource" | ||
| "k8s.io/client-go/informers" | ||
| "k8s.io/client-go/kubernetes" | ||
| "k8s.io/client-go/tools/cache" | ||
| "k8s.io/klog" | ||
| ) | ||
|
|
||
| // Broadcaster is an object which is used to get resources of the cluster. | ||
| type Broadcaster struct { | ||
| allocatable corev1.ResourceList | ||
| clusterConfig configv1alpha1.ClusterConfig | ||
| offerMutex sync.RWMutex | ||
| configMutex sync.RWMutex | ||
| } | ||
|
|
||
| // SetupBroadcaster create the informer e run it to signal node changes updating Offers. | ||
| func (b *Broadcaster) SetupBroadcaster(clientset *kubernetes.Clientset, config *rest.Config, stopper chan struct{}) error { | ||
| b.allocatable = corev1.ResourceList{} | ||
| factory := informers.NewSharedInformerFactory(clientset, 0) | ||
| informer := factory.Core().V1().Nodes().Informer() | ||
| informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ | ||
| AddFunc: b.onAdd, | ||
| UpdateFunc: b.onUpdate, | ||
| DeleteFunc: b.onDelete, | ||
| }) | ||
|
|
||
| config.ContentConfig.GroupVersion = &configv1alpha1.GroupVersion | ||
| config.NegotiatedSerializer = scheme.Codecs.WithoutConversion() | ||
| config.UserAgent = rest.DefaultKubernetesUserAgent() | ||
| CRDclient, err := crdclient.NewFromConfig(config) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| go utils.WatchConfiguration(b.setConfig, CRDclient, "") | ||
| go informer.Run(stopper) | ||
| return nil | ||
| } | ||
|
|
||
| func (b *Broadcaster) setConfig(configuration *configv1alpha1.ClusterConfig) { | ||
| b.configMutex.Lock() | ||
| defer b.configMutex.Unlock() | ||
| b.clusterConfig = *configuration | ||
| } | ||
|
|
||
| // react to a Node Creation/First informer run. | ||
| func (b *Broadcaster) onAdd(obj interface{}) { | ||
| node := obj.(*corev1.Node) | ||
| if node.Status.Phase == corev1.NodeRunning { | ||
| toAdd := &node.Status.Allocatable | ||
| currentResources := b.allocatable.DeepCopy() | ||
| addResources(¤tResources, toAdd) | ||
|
|
||
| if err := b.writeResources(currentResources); err != nil { | ||
| klog.Errorf("OnAdd error: unable to write allocatable of Node %s: %s", node.Name, err) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // react to a Node Update. | ||
| func (b *Broadcaster) onUpdate(oldObj, newObj interface{}) { | ||
| oldNode := oldObj.(*corev1.Node) | ||
| newNode := newObj.(*corev1.Node) | ||
| oldNodeResources := oldNode.Status.Allocatable | ||
| newNodeResources := newNode.Status.Allocatable | ||
| currentResources := b.allocatable.DeepCopy() | ||
| if newNode.Status.Phase == corev1.NodeRunning { | ||
| // node was already Running, update with possible new resources. | ||
| if oldNode.Status.Phase == corev1.NodeRunning { | ||
| updateResources(¤tResources, &oldNodeResources, &newNodeResources) | ||
| // node is starting, add all its resources. | ||
| } else if oldNode.Status.Phase == corev1.NodePending || oldNode.Status.Phase == corev1.NodeTerminated { | ||
| addResources(¤tResources, &newNodeResources) | ||
| } | ||
| // node is terminating or stopping, delete all its resources. | ||
| } else if oldNode.Status.Phase == corev1.NodeRunning && | ||
| (newNode.Status.Phase == corev1.NodeTerminated || newNode.Status.Phase == corev1.NodePending) { | ||
| subResources(¤tResources, &oldNodeResources) | ||
| } | ||
| if err := b.writeResources(currentResources); err != nil { | ||
| klog.Errorf("OnUpdate error: unable to write allocatable of Node %s: %s", newNode.Name, err) | ||
| } | ||
| } | ||
|
|
||
| // react to a Node Delete. | ||
| func (b *Broadcaster) onDelete(obj interface{}) { | ||
| node := obj.(*corev1.Node) | ||
| toDelete := &node.Status.Allocatable | ||
| currentResources := b.allocatable.DeepCopy() | ||
| if node.Status.Phase == corev1.NodeRunning { | ||
| subResources(¤tResources, toDelete) | ||
| if err := b.writeResources(currentResources); err != nil { | ||
| klog.Errorf("OnAdd error: unable to write allocatable of Node %s: %s", node.Name, err) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // write cluster resources in thread safe mode. | ||
| func (b *Broadcaster) writeResources(newResources corev1.ResourceList) error { | ||
| b.offerMutex.Lock() | ||
| defer b.offerMutex.Unlock() | ||
| if newResources != nil { | ||
| b.allocatable = newResources.DeepCopy() | ||
| return nil | ||
| } | ||
|
|
||
| return fmt.Errorf("some error occurred during cluster resources read. Attempting writing nil resources") | ||
| } | ||
|
|
||
| // ReadResources return in thread safe mode a scaled value of the resources. | ||
| func (b *Broadcaster) ReadResources() (corev1.ResourceList, error) { | ||
| b.offerMutex.RLock() | ||
| defer b.offerMutex.RUnlock() | ||
| if b.allocatable == nil { | ||
| return nil, fmt.Errorf("error getting cluster resources") | ||
| } | ||
| toRead := b.allocatable.DeepCopy() | ||
| for resourceName, quantity := range toRead { | ||
| scaled := quantity | ||
| b.scaleResources(&scaled) | ||
| toRead[resourceName] = scaled | ||
| } | ||
| return toRead, nil | ||
| } | ||
|
|
||
| func (b *Broadcaster) scaleResources(quantity *resource.Quantity) { | ||
| b.configMutex.RLock() | ||
| percentage := int64(b.clusterConfig.Spec.AdvertisementConfig.OutgoingConfig.ResourceSharingPercentage) | ||
| b.configMutex.RUnlock() | ||
| if percentage == 0 { | ||
| return | ||
| } | ||
|
|
||
| quantity.Set(quantity.Value() * percentage / 100) | ||
| } | ||
|
|
||
| // addResources is an utility function to add resources. | ||
| func addResources(currentResources, toAdd *corev1.ResourceList) { | ||
| for resourceName, quantity := range *toAdd { | ||
| value := (*currentResources)[resourceName] | ||
| if !quantity.IsZero() { | ||
| value.Add(quantity) | ||
| } | ||
| (*currentResources)[resourceName] = value | ||
| } | ||
| } | ||
|
|
||
| // subResources is an utility function to subtract resources. | ||
| func subResources(currentResources, toSub *corev1.ResourceList) { | ||
| for resourceName, quantity := range *toSub { | ||
| value := (*currentResources)[resourceName] | ||
| if !quantity.IsZero() { | ||
| value.Sub(quantity) | ||
| } | ||
| (*currentResources)[resourceName] = value | ||
| } | ||
| } | ||
|
|
||
| // updateResources is an utility function to update resources. | ||
| func updateResources(currentResources, oldResources, newResources *corev1.ResourceList) { | ||
| for resourceName, quantity := range *newResources { | ||
| oldQuantity := (*oldResources)[resourceName] | ||
| value := (*currentResources)[resourceName] | ||
| if quantityToUpdate := resource.NewQuantity(quantity.Value()-oldQuantity.Value(), | ||
| quantity.Format); !quantityToUpdate.IsZero() { | ||
| value.Add(*quantityToUpdate) | ||
| } | ||
| (*currentResources)[resourceName] = value | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.