Merge pull request #119 from michalschott/annotationTTL

Adding --annotation-ttl for automatic unlock

README.md

@@ -16,6 +16,7 @@
 	* [Testing](#testing)
 	* [Disabling Reboots](#disabling-reboots)
 	* [Manual Unlock](#manual-unlock)
+  * [Automatic Unlock](#automatic-unlock)
 * [Building](#building)
 * [Frequently Asked/Anticipated Questions](#frequently-askedanticipated-questions)
 * [Getting Help](#getting-help)
@@ -73,6 +74,7 @@ The following arguments can be passed to kured via the daemonset pod template:
 
 ```
 Flags:
+      --annotation-ttl time                  force clean annotation after this ammount of time (default 0, disabled)
       --alert-filter-regexp regexp.Regexp   alert names to ignore when checking for active alerts
       --blocking-pod-selector stringArray   label selector identifying pods whose presence should prevent reboots
       --ds-name string                      name of daemonset on which to place lock (default "kured")
@@ -259,6 +261,13 @@ kubectl -n kube-system annotate ds kured weave.works/kured-node-lock-
 > NB the `-` at the end of the command is important - it instructs
 > `kubectl` to remove that annotation entirely.
 
+### Automatic Unlock
+
+In exceptional circumstances (especially when used with cluster-autoscaler) a node
+which holds lock might be killed thus annotation will stay there for ever.
+
+Using `--annotation-ttl=30m` will allow other nodes to take over if TTL has expired (in this case 30min) and continue reboot process.
+
 ## Building
 
 See the [CircleCI config](.circleci/config.yml) for the preferred

cmd/kured/main.go

@@ -46,6 +46,8 @@ var (
 	rebootEnd   string
 	timezone    string
 
+	annotationTTL time.Duration
+
 	// Metrics
 	rebootRequiredGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Subsystem: "kured",
@@ -98,6 +100,9 @@ func main() {
 	rootCmd.PersistentFlags().StringVar(&timezone, "time-zone", "UTC",
 		"use this timezone for schedule inputs")
 
+	rootCmd.PersistentFlags().DurationVar(&annotationTTL, "annotation-ttl", 0,
+		"force clean annotation after this ammount of time (default 0, disabled)")
+
 	if err := rootCmd.Execute(); err != nil {
 		log.Fatal(err)
 	}
@@ -205,8 +210,8 @@ func holding(lock *daemonsetlock.DaemonSetLock, metadata interface{}) bool {
 	return holding
 }
 
-func acquire(lock *daemonsetlock.DaemonSetLock, metadata interface{}) bool {
-	holding, holder, err := lock.Acquire(metadata)
+func acquire(lock *daemonsetlock.DaemonSetLock, metadata interface{}, TTL time.Duration) bool {
+	holding, holder, err := lock.Acquire(metadata, TTL)
 	switch {
 	case err != nil:
 		log.Fatalf("Error acquiring lock: %v", err)
@@ -284,7 +289,7 @@ type nodeMeta struct {
 	Unschedulable bool `json:"unschedulable"`
 }
 
-func rebootAsRequired(nodeID string, window *timewindow.TimeWindow) {
+func rebootAsRequired(nodeID string, window *timewindow.TimeWindow, TTL time.Duration) {
 	config, err := rest.InClusterConfig()
 	if err != nil {
 		log.Fatal(err)
@@ -315,7 +320,7 @@ func rebootAsRequired(nodeID string, window *timewindow.TimeWindow) {
 			}
 			nodeMeta.Unschedulable = node.Spec.Unschedulable
 
-			if acquire(lock, &nodeMeta) {
+			if acquire(lock, &nodeMeta, TTL) {
 				if !nodeMeta.Unschedulable {
 					drain(nodeID)
 				}
@@ -347,8 +352,13 @@ func root(cmd *cobra.Command, args []string) {
 	log.Infof("Reboot Sentinel: %s every %v", rebootSentinel, period)
 	log.Infof("Blocking Pod Selectors: %v", podSelectors)
 	log.Infof("Reboot on: %v", window)
+	if annotationTTL > 0 {
+		log.Infof("Force annotation cleanup after: %v", annotationTTL)
+	} else {
+		log.Info("Force annotation cleanup disabled.")
+	}
 
-	go rebootAsRequired(nodeID, window)
+	go rebootAsRequired(nodeID, window, annotationTTL)
 	go maintainRebootRequiredMetric(nodeID)
 
 	http.Handle("/metrics", promhttp.Handler())

pkg/daemonsetlock/daemonsetlock.go

@@ -20,15 +20,17 @@ type DaemonSetLock struct {
 }
 
 type lockAnnotationValue struct {
-	NodeID   string      `json:"nodeID"`
-	Metadata interface{} `json:"metadata,omitempty"`
+	NodeID   string        `json:"nodeID"`
+	Metadata interface{}   `json:"metadata,omitempty"`
+	Created  time.Time     `json:"created"`
+	TTL      time.Duration `json:"TTL"`
 }
 
 func New(client *kubernetes.Clientset, nodeID, namespace, name, annotation string) *DaemonSetLock {
 	return &DaemonSetLock{client, nodeID, namespace, name, annotation}
 }
 
-func (dsl *DaemonSetLock) Acquire(metadata interface{}) (acquired bool, owner string, err error) {
+func (dsl *DaemonSetLock) Acquire(metadata interface{}, TTL time.Duration) (acquired bool, owner string, err error) {
 	for {
 		ds, err := dsl.client.AppsV1().DaemonSets(dsl.namespace).Get(context.TODO(), dsl.name, metav1.GetOptions{})
 		if err != nil {
@@ -41,13 +43,18 @@ func (dsl *DaemonSetLock) Acquire(metadata interface{}) (acquired bool, owner st
 			if err := json.Unmarshal([]byte(valueString), &value); err != nil {
 				return false, "", err
 			}
+
+			if ttlExpired(value.Created, value.TTL) {
+				return true, value.NodeID, nil
+			}
+
 			return value.NodeID == dsl.nodeID, value.NodeID, nil
 		}
 
 		if ds.ObjectMeta.Annotations == nil {
 			ds.ObjectMeta.Annotations = make(map[string]string)
 		}
-		value := lockAnnotationValue{NodeID: dsl.nodeID, Metadata: metadata}
+		value := lockAnnotationValue{NodeID: dsl.nodeID, Metadata: metadata, Created: time.Now().UTC(), TTL: TTL}
 		valueBytes, err := json.Marshal(&value)
 		if err != nil {
 			return false, "", err
@@ -80,6 +87,11 @@ func (dsl *DaemonSetLock) Test(metadata interface{}) (holding bool, err error) {
 		if err := json.Unmarshal([]byte(valueString), &value); err != nil {
 			return false, err
 		}
+
+		if ttlExpired(value.Created, value.TTL) {
+			return true, nil
+		}
+
 		return value.NodeID == dsl.nodeID, nil
 	}
 
@@ -99,7 +111,7 @@ func (dsl *DaemonSetLock) Release() error {
 			if err := json.Unmarshal([]byte(valueString), &value); err != nil {
 				return err
 			}
-			if value.NodeID != dsl.nodeID {
+			if value.NodeID != dsl.nodeID && !ttlExpired(value.Created, value.TTL) {
 				return fmt.Errorf("Not lock holder: %v", value.NodeID)
 			}
 		} else {
@@ -121,3 +133,10 @@ func (dsl *DaemonSetLock) Release() error {
 		return nil
 	}
 }
+
+func ttlExpired(created time.Time, ttl time.Duration) bool {
+	if ttl > 0 && time.Since(created) >= ttl {
+		return true
+	}
+	return false
+}

pkg/daemonsetlock/daemonsetlock_test.go

@@ -0,0 +1,28 @@
+package daemonsetlock
+
+import (
+	"testing"
+	"time"
+)
+
+func TestTtlExpired(t *testing.T) {
+	d := time.Date(2020, 05, 05, 14, 15, 0, 0, time.UTC)
+	second, _ := time.ParseDuration("1s")
+	zero, _ := time.ParseDuration("0m")
+
+	tests := []struct {
+		created time.Time
+		ttl     time.Duration
+		result  bool
+	}{
+		{d, second, true},
+		{time.Now(), second, false},
+		{d, zero, false},
+	}
+
+	for i, tst := range tests {
+		if ttlExpired(tst.created, tst.ttl) != tst.result {
+			t.Errorf("Test %d failed, expected %v but got %v", i, tst.result, !tst.result)
+		}
+	}
+}