{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/193933505","pull_request_review_id":127009683,"id":193933505,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDE5MzkzMzUwNQ==","diff_hunk":"@@ -0,0 +1,139 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License.  You may obtain a copy of the License at\n+ *\n+ *    http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+package org.apache.spark.scheduler.cluster.k8s\n+\n+import java.util.concurrent.atomic.{AtomicInteger, AtomicLong}\n+\n+import io.fabric8.kubernetes.api.model.PodBuilder\n+import io.fabric8.kubernetes.client.KubernetesClient\n+import scala.collection.mutable\n+\n+import org.apache.spark.{SparkConf, SparkException}\n+import org.apache.spark.deploy.k8s.Config._\n+import org.apache.spark.deploy.k8s.Constants._\n+import org.apache.spark.deploy.k8s.KubernetesConf\n+import org.apache.spark.internal.Logging\n+import org.apache.spark.util.{Clock, Utils}\n+\n+private[spark] class ExecutorPodsAllocator(\n+    conf: SparkConf,\n+    executorBuilder: KubernetesExecutorBuilder,\n+    kubernetesClient: KubernetesClient,\n+    snapshotsStore: ExecutorPodsSnapshotsStore,\n+    clock: Clock) extends Logging {\n+\n+  private val EXECUTOR_ID_COUNTER = new AtomicLong(0L)\n+\n+  private val totalExpectedExecutors = new AtomicInteger(0)\n+\n+  private val podAllocationSize = conf.get(KUBERNETES_ALLOCATION_BATCH_SIZE)\n+\n+  private val podAllocationDelay = conf.get(KUBERNETES_ALLOCATION_BATCH_DELAY)\n+\n+  private val podCreationTimeout = math.max(podAllocationDelay * 5, 60000)\n+\n+  private val kubernetesDriverPodName = conf\n+    .get(KUBERNETES_DRIVER_POD_NAME)\n+    .getOrElse(throw new SparkException(\"Must specify the driver pod name\"))\n+\n+  private val driverPod = kubernetesClient.pods()\n+    .withName(kubernetesDriverPodName)\n+    .get()\n+\n+  // Executor IDs that have been requested from Kubernetes but have not been detected in any\n+  // snapshot yet. Mapped to the timestamp when they were created.\n+  private val newlyCreatedExecutors = mutable.Map.empty[Long, Long]\n+\n+  def start(applicationId: String): Unit = {\n+    snapshotsStore.addSubscriber(podAllocationDelay) {\n+      processSnapshot(applicationId, _)\n+    }\n+  }\n+\n+  def setTotalExpectedExecutors(total: Int): Unit = totalExpectedExecutors.set(total)\n+\n+  private def processSnapshot(applicationId: String, snapshot: ExecutorPodsSnapshot): Unit = {\n+    snapshot.executorPods.filter {\n+      case (_, PodPending(_)) | (_, PodUnknown(_)) => false\n+      case _ => true\n+    }.keys.foreach {\n+      newlyCreatedExecutors -= _\n+    }\n+\n+    // For all executors we've created against the API but have not seen in a snapshot\n+    // yet - check the current time. If the current time has exceeded some threshold,\n+    // assume that the pod was either never created (the API server never properly\n+    // handled the creation request), or the API server created the pod but we missed\n+    // both the creation and deletion events. In either case, delete the missing pod\n+    // if possible, and mark such a pod to be rescheduled below.\n+    (newlyCreatedExecutors.keySet -- snapshot.executorPods.keySet).foreach { execId =>","path":"resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala","position":83,"original_position":83,"commit_id":"a97fc5d5b87a8caf14ae923f89a7bc106c48d411","original_commit_id":"a97fc5d5b87a8caf14ae923f89a7bc106c48d411","user":{"login":"mccheah","id":8975385,"node_id":"MDQ6VXNlcjg5NzUzODU=","avatar_url":"https://avatars1.githubusercontent.com/u/8975385?v=4","gravatar_id":"","url":"https://api.github.com/users/mccheah","html_url":"https://github.com/mccheah","followers_url":"https://api.github.com/users/mccheah/followers","following_url":"https://api.github.com/users/mccheah/following{/other_user}","gists_url":"https://api.github.com/users/mccheah/gists{/gist_id}","starred_url":"https://api.github.com/users/mccheah/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mccheah/subscriptions","organizations_url":"https://api.github.com/users/mccheah/orgs","repos_url":"https://api.github.com/users/mccheah/repos","events_url":"https://api.github.com/users/mccheah/events{/privacy}","received_events_url":"https://api.github.com/users/mccheah/received_events","type":"User","site_admin":false},"body":"A specific example where the previous approach wouldn't have worked. That said this case seems very unlikely - we'd have to miss all events for the newly created executor for a decent chunk of time. @foxish - would like to confirm that this is the right idea. Would specifically like feedback on if this is done correctly @foxish @liyinan926.","created_at":"2018-06-08T01:42:25Z","updated_at":"2018-06-08T01:53:45Z","html_url":"https://github.com/apache/spark/pull/21366#discussion_r193933505","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/21366","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/193933505"},"html":{"href":"https://github.com/apache/spark/pull/21366#discussion_r193933505"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/21366"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/21366","id":189138488,"node_id":"MDExOlB1bGxSZXF1ZXN0MTg5MTM4NDg4","html_url":"https://github.com/apache/spark/pull/21366","diff_url":"https://github.com/apache/spark/pull/21366.diff","patch_url":"https://github.com/apache/spark/pull/21366.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/21366","number":21366,"state":"open","locked":false,"title":"[SPARK-24248][K8S] Use the Kubernetes API to populate an event queue for scheduling","user":{"login":"mccheah","id":8975385,"node_id":"MDQ6VXNlcjg5NzUzODU=","avatar_url":"https://avatars1.githubusercontent.com/u/8975385?v=4","gravatar_id":"","url":"https://api.github.com/users/mccheah","html_url":"https://github.com/mccheah","followers_url":"https://api.github.com/users/mccheah/followers","following_url":"https://api.github.com/users/mccheah/following{/other_user}","gists_url":"https://api.github.com/users/mccheah/gists{/gist_id}","starred_url":"https://api.github.com/users/mccheah/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mccheah/subscriptions","organizations_url":"https://api.github.com/users/mccheah/orgs","repos_url":"https://api.github.com/users/mccheah/repos","events_url":"https://api.github.com/users/mccheah/events{/privacy}","received_events_url":"https://api.github.com/users/mccheah/received_events","type":"User","site_admin":false},"body":"## What changes were proposed in this pull request?\r\n\r\nPreviously, the scheduler backend was maintaining state in many places, not only for reading state but also writing to it. For example, state had to be managed in both the watch and in the executor allocator runnable. Furthermore, one had to keep track of multiple hash tables.\r\n\r\nWe can do better here by:\r\n\r\n1. Consolidating the places where we manage state. Here, we take inspiration from traditional Kubernetes controllers. These controllers tend to implement an event queue which is populated by two sources: a watch connection, and a periodic poller. Controllers typically use both mechanisms for redundancy; the watch connection may drop, so the periodic polling serves as a backup. Both sources write pod updates to a single event queue and then a processor periodically processes the current state of pods as reported by the two sources.\r\n\r\n2. Storing less specialized in-memory state in general. Previously we were creating hash tables to represent the state of executors. Instead, it's easier to represent state solely by the event queue, which has predictable read/write patterns and is more or less just a local up-to-date cache of the cluster's status.\r\n\r\n## How was this patch tested?\r\n\r\nIntegration tests should test there's no regressions end to end. Unit tests to be updated, in particular focusing on different orderings of events, particularly accounting for when events come in unexpected ordering.\r\n","created_at":"2018-05-18T21:24:15Z","updated_at":"2018-06-08T01:53:45Z","closed_at":null,"merged_at":null,"merge_commit_sha":"bad569d44188e2e741fd0ae902b5926be3a64494","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/21366/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/21366/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/21366/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/a97fc5d5b87a8caf14ae923f89a7bc106c48d411","head":{"label":"palantir:event-queue-driven-scheduling","ref":"event-queue-driven-scheduling","sha":"a97fc5d5b87a8caf14ae923f89a7bc106c48d411","user":{"login":"palantir","id":303157,"node_id":"MDEyOk9yZ2FuaXphdGlvbjMwMzE1Nw==","avatar_url":"https://avatars0.githubusercontent.com/u/303157?v=4","gravatar_id":"","url":"https://api.github.com/users/palantir","html_url":"https://github.com/palantir","followers_url":"https://api.github.com/users/palantir/followers","following_url":"https://api.github.com/users/palantir/following{/other_user}","gists_url":"https://api.github.com/users/palantir/gists{/gist_id}","starred_url":"https://api.github.com/users/palantir/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/palantir/subscriptions","organizations_url":"https://api.github.com/users/palantir/orgs","repos_url":"https://api.github.com/users/palantir/repos","events_url":"https://api.github.com/users/palantir/events{/privacy}","received_events_url":"https://api.github.com/users/palantir/received_events","type":"Organization","site_admin":false},"repo":{"id":30919203,"node_id":"MDEwOlJlcG9zaXRvcnkzMDkxOTIwMw==","name":"spark","full_name":"palantir/spark","owner":{"login":"palantir","id":303157,"node_id":"MDEyOk9yZ2FuaXphdGlvbjMwMzE1Nw==","avatar_url":"https://avatars0.githubusercontent.com/u/303157?v=4","gravatar_id":"","url":"https://api.github.com/users/palantir","html_url":"https://github.com/palantir","followers_url":"https://api.github.com/users/palantir/followers","following_url":"https://api.github.com/users/palantir/following{/other_user}","gists_url":"https://api.github.com/users/palantir/gists{/gist_id}","starred_url":"https://api.github.com/users/palantir/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/palantir/subscriptions","organizations_url":"https://api.github.com/users/palantir/orgs","repos_url":"https://api.github.com/users/palantir/repos","events_url":"https://api.github.com/users/palantir/events{/privacy}","received_events_url":"https://api.github.com/users/palantir/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/palantir/spark","description":"Palantir Distribution of Apache Spark","fork":true,"url":"https://api.github.com/repos/palantir/spark","forks_url":"https://api.github.com/repos/palantir/spark/forks","keys_url":"https://api.github.com/repos/palantir/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/palantir/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/palantir/spark/teams","hooks_url":"https://api.github.com/repos/palantir/spark/hooks","issue_events_url":"https://api.github.com/repos/palantir/spark/issues/events{/number}","events_url":"https://api.github.com/repos/palantir/spark/events","assignees_url":"https://api.github.com/repos/palantir/spark/assignees{/user}","branches_url":"https://api.github.com/repos/palantir/spark/branches{/branch}","tags_url":"https://api.github.com/repos/palantir/spark/tags","blobs_url":"https://api.github.com/repos/palantir/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/palantir/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/palantir/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/palantir/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/palantir/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/palantir/spark/languages","stargazers_url":"https://api.github.com/repos/palantir/spark/stargazers","contributors_url":"https://api.github.com/repos/palantir/spark/contributors","subscribers_url":"https://api.github.com/repos/palantir/spark/subscribers","subscription_url":"https://api.github.com/repos/palantir/spark/subscription","commits_url":"https://api.github.com/repos/palantir/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/palantir/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/palantir/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/palantir/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/palantir/spark/contents/{+path}","compare_url":"https://api.github.com/repos/palantir/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/palantir/spark/merges","archive_url":"https://api.github.com/repos/palantir/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/palantir/spark/downloads","issues_url":"https://api.github.com/repos/palantir/spark/issues{/number}","pulls_url":"https://api.github.com/repos/palantir/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/palantir/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/palantir/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/palantir/spark/labels{/name}","releases_url":"https://api.github.com/repos/palantir/spark/releases{/id}","deployments_url":"https://api.github.com/repos/palantir/spark/deployments","created_at":"2015-02-17T13:57:35Z","updated_at":"2018-06-07T17:08:47Z","pushed_at":"2018-06-08T01:40:48Z","git_url":"git://github.com/palantir/spark.git","ssh_url":"git@github.com:palantir/spark.git","clone_url":"https://github.com/palantir/spark.git","svn_url":"https://github.com/palantir/spark","homepage":"","size":272541,"stargazers_count":24,"watchers_count":24,"language":"Scala","has_issues":true,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":20,"mirror_url":null,"archived":false,"open_issues_count":16,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":20,"open_issues":16,"watchers":24,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"e76b0124fbe463def00b1dffcfd8fd47e04772fe","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"node_id":"MDEwOlJlcG9zaXRvcnkxNzE2NTY1OA==","name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","deployments_url":"https://api.github.com/repos/apache/spark/deployments","created_at":"2014-02-25T08:00:08Z","updated_at":"2018-06-08T01:28:47Z","pushed_at":"2018-06-08T01:40:52Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":271371,"stargazers_count":17566,"watchers_count":17566,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":16044,"mirror_url":null,"archived":false,"open_issues_count":562,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":16044,"open_issues":562,"watchers":17566,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/21366"},"html":{"href":"https://github.com/apache/spark/pull/21366"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/21366"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/21366/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/21366/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/21366/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/a97fc5d5b87a8caf14ae923f89a7bc106c48d411"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"mccheah"}}