Fix issue where k8s ingresses are sometimes not deleted (linkerd#1810)

Problem
Linkerd's io.l5d.ingress identifier watches ingress resources in order to function as an ingress controller in k8s. This watch receives events from k8s and relies on a resourceVersion to make sure the most up to date ingress is being used. The resourceVersion is supposedly a number that increases after a change event occurs on a resource. For a particular ingress resource, its resourceVersion increases after Add, Modify and Change events. Sometimes k8s returns a resourceVersion that is unchanged. Linkerd does not recognize this as a change since it expects a resourceVersion to always be larger than the previously observed resource and therefore does not update its IngressCache.

Solution
The solution is rather simple. When Linkerd encounters a resourceVersion that is the same as the
the previously observed we do not ignore the event and notify the ingress watch process to update the IngressCache

Validation
Tests were done on a locally running Linkerd instance connected to k8s running on docker. Linkerd was able to update its IngressCache when an ingress was deleted in k8s.

Fixes linkerd#1791

Signed-off-by: Dennis Adjei-Baah [email protected]

k8s/src/main/scala/io/buoyant/k8s/Watchable.scala

@@ -172,12 +172,24 @@ private[k8s] abstract class Watchable[O <: KubeObject: TypeReference, W <: Watch
 
         case Some((event, ws)) =>
           import Ordering.Implicits._
-          // Register the update only if its resource version is larger than the largest version
+          // Register the update only if its resource version is larger than or equal to the largest version
           // seen so far.
-          if (largestEvent.forall(_ < event)) {
+          if (largestEvent.forall(_ <= event)) {
+            log.trace(
+              "k8s watch on '%s' registered event with same or newer resource version %s (older resource version was %s)",
+              watchPath,
+              event.resourceVersion,
+              largestEvent.flatMap(_.resourceVersion)
+            )
             state.update(Activity.Ok(event))
             _processEventStream(ws, event.resourceVersion, Some(event))
           } else {
+            log.trace(
+              "k8s watch on '%s' skipping event with resource version %s (older than the most recent resource version %s)",
+              watchPath,
+              event.resourceVersion,
+              largestEvent.flatMap(_.resourceVersion)
+            )
             _processEventStream(ws, largestVersion, largestEvent)
           }
 

k8s/src/test/scala/io/buoyant/k8s/v1/ApiTest.scala

@@ -488,13 +488,13 @@ class ApiTest extends FunSuite
 
     assert(events.size == 0)
 
-    // repeat the event: no update since resource version is the same
+    // repeat the event: update since resource version is the same
     await(w.write(modified0))
-    assert(events.size == 0)
+    assert(events.size == 1)
 
     // write an earlier event: no update since resource version is too low
     await(w.write(modified2))
-    assert(events.size == 0)
+    assert(events.size == 1)
     closable.close()
     if (failure != null) throw failure
   }