Commit
- Loading branch information
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| /* | ||
| Copyright The Kubernetes Authors. | ||
| Licensed under the Apache License, Version 2.0 (the "License"); | ||
| you may not use this file except in compliance with the License. | ||
| You may obtain a copy of the License at | ||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
| Unless required by applicable law or agreed to in writing, software | ||
| distributed under the License is distributed on an "AS IS" BASIS, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|
|
||
| package terminator | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| set "github.com/deckarep/golang-set" | ||
| v1 "k8s.io/api/core/v1" | ||
| apierrors "k8s.io/apimachinery/pkg/api/errors" | ||
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| "k8s.io/apimachinery/pkg/types" | ||
| corev1 "k8s.io/client-go/kubernetes/typed/core/v1" | ||
| "k8s.io/client-go/util/workqueue" | ||
| "knative.dev/pkg/logging" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" | ||
| "sigs.k8s.io/controller-runtime/pkg/manager" | ||
| "sigs.k8s.io/controller-runtime/pkg/reconcile" | ||
|
|
||
| terminatorevents "sigs.k8s.io/karpenter/pkg/controllers/node/termination/terminator/events" | ||
| "sigs.k8s.io/karpenter/pkg/operator/controller" | ||
|
|
||
| "sigs.k8s.io/karpenter/pkg/events" | ||
| ) | ||
|
|
||
| const ( | ||
| deletionQueueBaseDelay = 100 * time.Millisecond | ||
|
Check failure on line 43 in pkg/controllers/node/termination/terminator/deletion.go
|
||
| deletionQueueMaxDelay = 10 * time.Second | ||
|
Check failure on line 44 in pkg/controllers/node/termination/terminator/deletion.go
|
||
| ) | ||
|
|
||
| type DeletionQueue struct { | ||
| workqueue.RateLimitingInterface | ||
| set.Set | ||
|
|
||
| coreV1Client corev1.CoreV1Interface | ||
| recorder events.Recorder | ||
| } | ||
|
|
||
| func NewDeletionQueue(coreV1Client corev1.CoreV1Interface, recorder events.Recorder) *DeletionQueue { | ||
| queue := &DeletionQueue{ | ||
| RateLimitingInterface: workqueue.NewRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(evictionQueueBaseDelay, evictionQueueMaxDelay)), | ||
| Set: set.NewSet(), | ||
| coreV1Client: coreV1Client, | ||
| recorder: recorder, | ||
| } | ||
| return queue | ||
| } | ||
|
|
||
| func (q *DeletionQueue) Name() string { | ||
| return "deletion-queue" | ||
| } | ||
|
|
||
| func (q *DeletionQueue) Builder(_ context.Context, m manager.Manager) controller.Builder { | ||
| return controller.NewSingletonManagedBy(m) | ||
| } | ||
|
|
||
| // Add adds pods to the Queue | ||
| func (q *DeletionQueue) Add(pods ...*v1.Pod) { | ||
| for _, pod := range pods { | ||
| if nn := client.ObjectKeyFromObject(pod); !q.Set.Contains(nn) { | ||
| q.Set.Add(nn) | ||
| q.RateLimitingInterface.Add(nn) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (q *DeletionQueue) Reconcile(ctx context.Context, _ reconcile.Request) (reconcile.Result, error) { | ||
| // Check if the queue is empty. client-go recommends not using this function to gate the subsequent | ||
| // get call, but since we're popping items off the queue synchronously, there should be no synchonization | ||
| // issues. | ||
| if q.Len() == 0 { | ||
| return reconcile.Result{RequeueAfter: 1 * time.Second}, nil | ||
| } | ||
| // Get pod from queue. This waits until queue is non-empty. | ||
| item, shutdown := q.RateLimitingInterface.Get() | ||
| if shutdown { | ||
| return reconcile.Result{}, fmt.Errorf("EvictionQueue is broken and has shutdown") | ||
| } | ||
| nn := item.(types.NamespacedName) | ||
| defer q.RateLimitingInterface.Done(nn) | ||
|
|
||
| if q.Delete(ctx, nn) { | ||
| q.RateLimitingInterface.Forget(nn) | ||
| q.Set.Remove(nn) | ||
| return reconcile.Result{RequeueAfter: controller.Immediately}, nil | ||
| } | ||
| // Requeue pod if delete failed | ||
| q.RateLimitingInterface.AddRateLimited(nn) | ||
| return reconcile.Result{RequeueAfter: controller.Immediately}, nil | ||
| } | ||
|
|
||
| // Delete returns true if successful delete call, and false if there was an error | ||
| func (q *DeletionQueue) Delete(ctx context.Context, nn types.NamespacedName) bool { | ||
| ctx = logging.WithLogger(ctx, logging.FromContext(ctx).With("pod", nn)) | ||
| if err := q.coreV1Client.Pods(nn.Namespace).Delete(ctx, nn.Name, metav1.DeleteOptions{}); err != nil { | ||
| if apierrors.IsNotFound(err) { // 404 | ||
| return true | ||
| } | ||
| logging.FromContext(ctx).Errorf("deleting pod, %s", err) | ||
| return false | ||
| } | ||
| q.recorder.Publish(terminatorevents.DeletePod(&v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: nn.Name, Namespace: nn.Namespace}})) | ||
| return true | ||
| } | ||
|
|
||
| func (q *DeletionQueue) Reset() { | ||
| q.RateLimitingInterface = workqueue.NewRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(evictionQueueBaseDelay, evictionQueueMaxDelay)) | ||
| q.Set = set.NewSet() | ||
| } | ||
