Add Evict operator

.gitignore

@@ -1,2 +1,3 @@
 /k-rail
+/evicter
 vendor

Dockerfile

@@ -3,13 +3,14 @@ ARG GO_VERSION=1.13
 FROM golang:${GO_VERSION}-buster AS builder
 WORKDIR /build
 COPY ./ /build/
-RUN make test
+RUN make clean test
 RUN make build
 
 # Production image build stage
 FROM scratch
 EXPOSE 8443/tcp
 COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
 COPY --from=builder /build/k-rail /k-rail
+COPY --from=builder /build/evicter /evicter
 USER 65534
 ENTRYPOINT ["/k-rail", "-config", "/config/config.yml"]

Makefile

@@ -9,13 +9,27 @@
 # 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.
+.PHONY: all build test image clean
 
-ensure:
-		dep ensure
+LDFLAGS = -extldflags=-static -s -w
+BUILD_FLAGS = -mod=readonly -ldflags '$(LDFLAGS)' -trimpath
+BUILD_VERSION ?= manual
+IMAGE_NAME = "cruise/k-rail:${BUILD_VERSION}"
+
+all: dist
+
+dist: image
+
+clean:
+	rm -f evicter k-rail
+	go mod verify
 
 build:
-		GO111MODULE=on CGO_ENABLED=0 go build -o k-rail cmd/main.go
+	GO111MODULE=on CGO_ENABLED=0 go build ${BUILD_FLAGS} -o k-rail cmd/k-rail/main.go
+	GO111MODULE=on CGO_ENABLED=0 go build ${BUILD_FLAGS} -o evicter cmd/evicter/*.go
 
 test:
-		GO111MODULE=on CGO_ENABLED=1 go test -race -cover $(shell go list ./... | grep -v /vendor/)
-
+	GO111MODULE=on CGO_ENABLED=1 go test -mod=readonly -race  ./...
+
+image: build
+	docker build --pull -t $(IMAGE_NAME) .

cmd/evicter/README.md

@@ -0,0 +1,6 @@
+# Evict tainted pods after period
+Operator that runs within k8s to find and evict `tainted` pods.
+
+* `k-rail/tainted-timestamp` store the unix timestamp when the root event happend
+* `k-rail/tainted-prevent-eviction` is a break-glass annotation to prevent automated eviction
+* `k-rail/reason` intended for humans

cmd/evicter/controller.go

@@ -0,0 +1,188 @@
+package main
+
+import (
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/pkg/errors"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog"
+)
+
+type podProvisioner interface {
+	Evict(pod *v1.Pod, reason, msg string) error
+}
+
+type Controller struct {
+	podStore                cache.Store
+	queue                   workqueue.RateLimitingInterface
+	informer                cache.Controller
+	podProvisioner          podProvisioner
+	incubationPeriodSeconds time.Duration
+	started                 time.Time
+}
+
+func NewController(queue workqueue.RateLimitingInterface, indexer cache.Store, informer cache.Controller, podProvisioner podProvisioner, incubationPeriodSeconds int64) *Controller {
+	return &Controller{
+		informer:                informer,
+		podStore:                indexer,
+		queue:                   queue,
+		podProvisioner:          podProvisioner,
+		incubationPeriodSeconds: time.Duration(incubationPeriodSeconds) * time.Second,
+	}
+}
+
+func (c *Controller) processNextItem() bool {
+	key, quit := c.queue.Get()
+	if quit {
+		return false
+	}
+	defer c.queue.Done(key)
+
+	err := c.evictPod(key.(string))
+	c.handleErr(err, key)
+	return true
+}
+
+const (
+	// annotationPreventEviction is a break-glass annotation to prevent automated eviction
+	annotationPreventEviction = "k-rail/tainted-prevent-eviction"
+	// annotationTimestamp stores the unix timestamp when the root event happened
+	annotationTimestamp = "k-rail/tainted-timestamp"
+	// annotationReason is used to define any additional reason in a human readable form
+	annotationReason = "k-rail/tainted-reason"
+)
+
+const defaultEvictionReason = "Tainted"
+const noEvictionNote = "Pod was marked as tainted"
+
+// evictPod is the business logic of the controller. it checks the the eviction rules and conditions before calling the pod provisioner.
+func (c *Controller) evictPod(key string) error {
+	obj, exists, err := c.podStore.GetByKey(key)
+	switch {
+	case err != nil:
+		return err
+	case !exists:
+		return nil
+	}
+	pod, ok := obj.(*v1.Pod)
+	if !ok {
+		return fmt.Errorf("unsupported type: %T", obj)
+	}
+	if !canEvict(pod, c.incubationPeriodSeconds) {
+		return nil
+	}
+
+	msg, ok := pod.Annotations[annotationReason]
+	if !ok || msg == "" {
+		msg = noEvictionNote
+	}
+	return c.podProvisioner.Evict(pod, defaultEvictionReason, msg)
+}
+
+func canEvict(pod *v1.Pod, incubationPeriod time.Duration) bool {
+	if pod == nil {
+		return false
+	}
+	switch pod.Annotations[annotationPreventEviction] {
+	case "yes", "true", "1", "YES", "TRUE", "Yes", "True":
+		return false
+	}
+
+	val, ok := pod.Annotations[annotationTimestamp]
+	if ok {
+		i, err := strconv.ParseInt(val, 10, 64)
+		if err != nil {
+			// todo: log
+			return true
+		}
+		timestamp := time.Unix(i, 0)
+		if time.Since(timestamp) < incubationPeriod {
+			return false
+		}
+	}
+	return true
+}
+
+const maxWorkerRetries = 5
+
+// handleErr checks if an error happened and makes sure we will retry later.
+func (c *Controller) handleErr(err error, key interface{}) {
+	if err == nil {
+		// Forget about the #AddRateLimited history of the key on every successful synchronization.
+		// This ensures that future processing of updates for this key is not delayed because of
+		// an outdated error history.
+		c.queue.Forget(key)
+		return
+	}
+
+	// This controller retries 5 times if something goes wrong. After that, it stops trying.
+	if c.queue.NumRequeues(key) < maxWorkerRetries {
+		klog.Infof("Error syncing pod %v: %v", key, err)
+
+		// Re-enqueue the key rate limited. Based on the rate limiter on the
+		// queue and the re-enqueue history, the key will be processed later again.
+		c.queue.AddRateLimited(key)
+		return
+	}
+
+	c.queue.Forget(key)
+	// Report to an external entity that, even after several retries, we could not successfully process this key
+	runtime.HandleError(err)
+	klog.Infof("Dropping pod %q out of the queue: %v", key, err)
+}
+
+const reconciliationTick = 30 * time.Second
+const startupGracePeriod = 90 * time.Second
+
+func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) {
+	defer runtime.HandleCrash()
+
+	// Let the workers stop when we are done
+	defer c.queue.ShutDown()
+	klog.Info("Starting Pod controller")
+
+	go c.informer.Run(stopCh)
+
+	// Wait for all involved caches to be synced, before processing items from the queue is started
+	if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
+		runtime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
+		return
+	}
+
+	for i := 0; i < threadiness; i++ {
+		go wait.Until(c.runWorker, time.Second, stopCh)
+	}
+
+	wait.Until(func() {
+		if time.Since(c.started) < startupGracePeriod {
+			return
+		}
+		if err := c.doReconciliation(); err != nil {
+			klog.Errorf("Reconciliation failed: %s", err)
+		}
+	}, reconciliationTick, stopCh)
+
+	klog.Info("Stopping Pod controller")
+}
+
+func (c *Controller) runWorker() {
+	for c.processNextItem() {
+	}
+}
+
+func (c *Controller) doReconciliation() error {
+	klog.Info("Reconciliation started")
+	for _, key := range c.podStore.ListKeys() {
+		if err := c.evictPod(key); err != nil {
+			return errors.Wrapf(err, "pod %q", key)
+		}
+	}
+	klog.Info("Reconciliation completed")
+	return nil
+}