Revert "Parallel watch processing. (#2169)" (#2174)

This reverts commit 637b734.

Author: scrosby

scheduler/java/com/twosigma/cook/kubernetes/ParallelWatchQueue.java

@@ -27,7 +27,7 @@
             [metrics.timers :as timers]
             [opentracing-clj.core :as tracing])
   (:import (com.google.auth.oauth2 GoogleCredentials)
-           (com.twosigma.cook.kubernetes TokenRefreshingAuthenticator ParallelWatchQueue)
+           (com.twosigma.cook.kubernetes TokenRefreshingAuthenticator)
            (io.kubernetes.client.openapi ApiClient)
            (io.kubernetes.client.openapi.models V1Node V1Pod)
            (io.kubernetes.client.util ClientBuilder KubeConfig)
@@ -214,23 +214,15 @@
 
 (defn make-cook-pod-watch-callback
   "Make a callback function that is passed to the pod-watch callback. This callback forwards changes to the cook.kubernetes.controller."
-  [{:keys [^ParallelWatchQueue parallel-watch-queue] :as kcc}]
+  [kcc]
   (fn pod-watch-callback
-    [_ ^V1Pod prev-pod ^V1Pod pod]
+    [_ prev-pod pod]
     (try
-      (let [name (or (some-> prev-pod .getMetadata .getName)
-                     (some-> pod .getMetadata .getName))
-            shardNum (mod (.hashCode name) (.getShardCount parallel-watch-queue))
-            ^Runnable event (fn []
-                              (try
-                                (if (nil? pod)
-                                  (controller/pod-deleted kcc prev-pod)
-                                  (controller/pod-update kcc pod))
-                                (catch Exception e
-                                  (log/error e "Error processing status update on" name))))]
-        (.submitEvent parallel-watch-queue event shardNum))
+      (if (nil? pod)
+        (controller/pod-deleted kcc prev-pod)
+        (controller/pod-update kcc pod))
       (catch Exception e
-        (log/error e "Error submitting pod status update")))))
+        (log/error e "Error processing status update")))))
 
 (defn task-ents->map-by-task-id
   "Given seq of task entities from datomic, generate a map of task-id -> entity."
@@ -416,8 +408,7 @@
                                      compute-cluster-launch-rate-limiter cook-pool-taint-name cook-pool-taint-prefix
                                      cook-pool-taint2-name cook-pool-taint2-value
                                      cook-pool-label-name cook-pool-label-prefix
-                                     controller-lock-objects kill-lock-object
-                                     parallel-watch-queue]
+                                     controller-lock-objects kill-lock-object]
   cc/ComputeCluster
   (launch-tasks [this pool-name matches process-task-post-launch-fn]
     (let [task-metadata-seq (mapcat :task-metadata-seq matches)]
@@ -895,8 +886,6 @@
            name
            namespace
            node-blocklist-labels
-           parallel-watch-max-outstanding
-           parallel-watch-shards
            read-timeout-seconds
            scan-frequency-seconds
            state
@@ -917,8 +906,6 @@
          state :running
          state-locked? false
          use-google-service-account? true
-         parallel-watch-max-outstanding 1000
-         parallel-watch-shards 200
          cook-pool-taint-prefix ""
          cook-pool-label-prefix ""
          use-token-refreshing-authenticator? false}
@@ -976,7 +963,6 @@
                                                     (with-meta (vec (repeatedly lock-shard-count #(ReentrantLock.)))
                                                                {:json-value (str "<count of " lock-shard-count " ReentrantLocks>")})
                                                     ; cluster-level kill-lock. See cc/kill-lock-object
-                                                    (ReentrantReadWriteLock. true)
-                                                    (ParallelWatchQueue. controller-executor-service parallel-watch-max-outstanding parallel-watch-shards))]
+                                                    (ReentrantReadWriteLock. true))]
     (cc/register-compute-cluster! compute-cluster)
     compute-cluster))

scheduler/src/cook/kubernetes/compute_cluster.clj

@@ -43,7 +43,6 @@
             [ring.middleware.params :refer [wrap-params]])
   (:import (com.google.common.cache CacheBuilder)
            (com.netflix.fenzo SimpleAssignmentResult)
-           (com.twosigma.cook.kubernetes ParallelWatchQueue)
            (io.kubernetes.client.custom Quantity Quantity$Format)
            (io.kubernetes.client.openapi.models V1Container V1Node V1NodeSpec V1NodeStatus V1ObjectMeta V1Pod V1PodSpec V1ResourceRequirements V1Taint)
            (java.util.concurrent Executors TimeUnit)
@@ -662,5 +661,4 @@
                                     "some-random-label-A"
                                     "some-random-label-val-B"
                                     (repeatedly 16 #(ReentrantLock.))
-                                    (ReentrantReadWriteLock. true)
-                                    (ParallelWatchQueue. (Executors/newSingleThreadExecutor) 1000 100))))
+                                    (ReentrantReadWriteLock. true))))

scheduler/src/cook/test/testutil.clj

@@ -12,7 +12,6 @@
             [cook.test.testutil :as tu]
             [datomic.api :as d])
   (:import (clojure.lang ExceptionInfo)
-           (com.twosigma.cook.kubernetes ParallelWatchQueue)
            (io.kubernetes.client.openapi.models V1NodeSelectorRequirement V1Pod V1PodSecurityContext)
            (java.util.concurrent Executors)
            (java.util.concurrent.locks ReentrantLock ReentrantReadWriteLock)
@@ -81,8 +80,7 @@
                                                               {} (atom :running) (atom false) false
                                                               cook.rate-limit/AllowAllRateLimiter "t-a" "p-a" "t2-a" "p2-a" "l-p" "l-v1"
                                                               (repeatedly 16 #(ReentrantLock.))
-                                                              (ReentrantReadWriteLock. true)
-                                                              (ParallelWatchQueue. (Executors/newSingleThreadExecutor) 1000 10))
+                                                              (ReentrantReadWriteLock. true))
               task-metadata (task/TaskAssignmentResult->task-metadata (d/db conn)
                                                                       nil
                                                                       compute-cluster
@@ -105,8 +103,7 @@
                                                               {} (atom :running) (atom false) false
                                                               cook.rate-limit/AllowAllRateLimiter "t-b" "p-b" "t2-a" "p2-a" "l-p" "l-v2"
                                                               (repeatedly 16 #(ReentrantLock.))
-                                                              (ReentrantReadWriteLock. true)
-                                                              (ParallelWatchQueue. (Executors/newSingleThreadExecutor) 1000 10))
+                                                              (ReentrantReadWriteLock. true))
               task-metadata (task/TaskAssignmentResult->task-metadata (d/db conn)
                                                                       nil
                                                                       compute-cluster
@@ -137,8 +134,7 @@
                                                             {} (atom :running) (atom false) false
                                                             cook.rate-limit/AllowAllRateLimiter "t-c" "p-c" "t2-a" "p2-a" "l-p" "l-c2"
                                                             (repeatedly 16 #(ReentrantLock.))
-                                                            (ReentrantReadWriteLock. true)
-                                                            (ParallelWatchQueue. (Executors/newSingleThreadExecutor) 1000 10))
+                                                            (ReentrantReadWriteLock. true))
             node-name->node {"nodeA" (tu/node-helper "nodeA" 1.0 1000.0 10 "nvidia-tesla-p100" nil nil)
                              "nodeB" (tu/node-helper "nodeB" 1.0 1000.0 25 "nvidia-tesla-p100" nil nil)
                              "nodeC" (tu/node-helper "nodeC" 1.0 1000.0 nil nil nil nil)