Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added OfferUpdater component which implement all ResourceOffer update.
Added BroadcasterInterface and UpdaterInterface to avoid cyclid dependencies. Added lastReadResources parameter to track last read resources. Implemented update method. added checkThreshold method. Some other fix. fix deadlock. added logic to remove a clusterId from both broadcaster pod map and updater queue. fix bug in tests.
- Loading branch information
Showing
10 changed files
with
375 additions
and
31 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
13 changes: 13 additions & 0 deletions
13
internal/resource-request-operator/interfaces/broadcasterInterface.go
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,13 @@ | ||
| package interfaces | ||
|
|
||
| import corev1 "k8s.io/api/core/v1" | ||
|
|
||
| // ClusterResourceInterface represents a generic subset of Broadcaster exported methods to be used instead of a direct access to | ||
| // the Broadcaster instance and get/update some cluster resources information. | ||
| type ClusterResourceInterface interface { | ||
| // ReadResources returns all free cluster resources calculated for a given clusterID scaled by a percentage value. | ||
| // If clusterID is not valid all scaled cluster resources will be returned. | ||
| ReadResources(clusterID string) corev1.ResourceList | ||
| // RemoveClusterID removes given clusterID from all internal structures and it will be no more valid. | ||
| RemoveClusterID(clusterID string) | ||
| } |
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,2 @@ | ||
| // Package interfaces contains all the ResourceRequestOperator interfaces representing some of its components. | ||
| package interfaces |
17 changes: 17 additions & 0 deletions
17
internal/resource-request-operator/interfaces/updaterInterface.go
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,17 @@ | ||
| package interfaces | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| ) | ||
|
|
||
| // UpdaterInterface represents a generic subset of Updater exported methods to be used instead of a direct access to | ||
| // a particular Updater instance. | ||
| type UpdaterInterface interface { | ||
| // Start run an instance of an updater which will be stopped when ctx.Done() is called. | ||
| Start(ctx context.Context, group *sync.WaitGroup) | ||
| // Push add the clusterID to the internal queue to be processed as soon as possible. | ||
| Push(clusterID string) | ||
| // Remove removes the clusterID from the internal queue and will be no longer processed. | ||
| Remove(clusterID string) | ||
| } |
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,146 @@ | ||
| package resourcerequestoperator | ||
|
|
||
| import ( | ||
| "context" | ||
| "crypto/rand" | ||
| "fmt" | ||
| "math/big" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "k8s.io/apimachinery/pkg/util/wait" | ||
| "k8s.io/client-go/util/workqueue" | ||
| "k8s.io/klog/v2" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
|
|
||
| discoveryv1alpha1 "github.com/liqotech/liqo/apis/discovery/v1alpha1" | ||
| sharingv1alpha1 "github.com/liqotech/liqo/apis/sharing/v1alpha1" | ||
| "github.com/liqotech/liqo/internal/resource-request-operator/interfaces" | ||
| "github.com/liqotech/liqo/pkg/discovery" | ||
| ) | ||
|
|
||
| const requeueTimeout = 5 * time.Minute | ||
|
|
||
| // OfferUpdater is a component which wraps all ResourceOffer update logic. | ||
| type OfferUpdater struct { | ||
| queue workqueue.RateLimitingInterface | ||
| client.Client | ||
| broadcasterInt interfaces.ClusterResourceInterface | ||
| } | ||
|
|
||
| // Setup initialize all parameters of the OfferUpdater component. | ||
| func (u *OfferUpdater) Setup(broadcaster interfaces.ClusterResourceInterface, k8Client client.Client) { | ||
| u.broadcasterInt = broadcaster | ||
| u.Client = k8Client | ||
| u.queue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Offer update queue") | ||
| } | ||
|
|
||
| // Start run the OfferUpdate worker. | ||
| func (u *OfferUpdater) Start(ctx context.Context, group *sync.WaitGroup) { | ||
| defer u.queue.ShutDown() | ||
| group.Add(1) | ||
| defer group.Done() | ||
| go u.startRunner(ctx) | ||
| <-ctx.Done() | ||
| } | ||
|
|
||
| func (u *OfferUpdater) startRunner(ctx context.Context) { | ||
| wait.Until(u.run, 2*time.Second, ctx.Done()) | ||
| } | ||
|
|
||
| func (u *OfferUpdater) run() { | ||
| for u.processNextItem() { | ||
| } | ||
| } | ||
|
|
||
| func (u *OfferUpdater) processNextItem() bool { | ||
| obj, shutdown := u.queue.Get() | ||
| if shutdown { | ||
| return false | ||
| } | ||
| err := func(obj interface{}) error { | ||
| defer u.queue.Done(obj) | ||
| var clusterID string | ||
| var ok bool | ||
| if clusterID, ok = obj.(string); !ok { | ||
| // As the item in the workqueue is actually invalid, we call | ||
| // Forget here else we'd go into a loop of attempting to | ||
| // process a work item that is invalid. | ||
| u.queue.Forget(obj) | ||
| return fmt.Errorf("error getting object %v from OfferUpater queue. It is not a string", obj) | ||
| } | ||
| // check if the clusterID is still valid and can be processed | ||
| if !u.isValidClusterID(clusterID) { | ||
| // if clusterID is not valid means that the peering has been dropped or something went wrong. | ||
| // so we call Forget and RemoveClusterID in order to invalidate it. | ||
| u.broadcasterInt.RemoveClusterID(clusterID) | ||
| return fmt.Errorf("clusterID %s is no more valid and will be deleted", clusterID) | ||
| } | ||
| if err := u.updateOffer(clusterID); err != nil { | ||
| // Put the item back on the workqueue to handle any transient errors. | ||
| u.queue.AddRateLimited(clusterID) | ||
| return fmt.Errorf("error during updating ResourceOffer for cluster %s: %w", clusterID, err) | ||
| } | ||
| return nil | ||
| }(obj) | ||
| if err != nil { | ||
| klog.Errorf("Error occurred during ResourceOffer update %s", err) | ||
| return true | ||
| } | ||
| klog.Infof("Update cluster %s processed", obj.(string)) | ||
|
|
||
| // requeue after timeout seconds | ||
| u.queue.AddAfter(obj, requeueTimeout+time.Duration(getRandomDelta())*time.Second) | ||
| return true | ||
| } | ||
|
|
||
| func (u *OfferUpdater) updateOffer(clusterID string) error { | ||
| offerList := &sharingv1alpha1.ResourceOfferList{} | ||
| if err := u.Client.List(context.Background(), offerList, client.MatchingLabels{ | ||
| discovery.ClusterIDLabel: clusterID, | ||
| }); err != nil { | ||
| return err | ||
| } | ||
| if len(offerList.Items) > 1 { | ||
| return fmt.Errorf("too many ResourceOffers for cluster %s", clusterID) | ||
| } else if len(offerList.Items) == 0 { | ||
| return fmt.Errorf("no ResourceOffer for cluster %s", clusterID) | ||
| } | ||
| offer := &offerList.Items[0] | ||
| resources := u.broadcasterInt.ReadResources(clusterID) | ||
| offer.Spec.ResourceQuota.Hard = resources.DeepCopy() | ||
| if err := u.Client.Update(context.Background(), offer); err != nil { | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Push add new clusterID to update queue which will be processes as soon as possible. | ||
| func (u *OfferUpdater) Push(clusterID string) { | ||
| u.queue.Add(clusterID) | ||
| } | ||
|
|
||
| // Remove remove a specified clusterID from the update queue and it will be no more processed. | ||
| func (u *OfferUpdater) Remove(clusterID string) { | ||
| u.queue.Forget(clusterID) | ||
| klog.Infof("Removed cluster %s from update queue", clusterID) | ||
| } | ||
|
|
||
| func (u *OfferUpdater) isValidClusterID(clusterID string) bool { | ||
| resourceRequestList := &discoveryv1alpha1.ResourceRequestList{} | ||
| err := u.Client.List(context.Background(), resourceRequestList) | ||
| if err != nil || len(resourceRequestList.Items) != 1 { | ||
| return false | ||
| } | ||
| return true | ||
| } | ||
|
|
||
| func getRandomDelta() int64 { | ||
| max := new(big.Int) | ||
| max.SetInt64(int64(60)) | ||
| n, err := rand.Int(rand.Reader, max) | ||
| if err != nil { | ||
| return int64(0) | ||
| } | ||
| return n.Int64() | ||
| } |
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.