Log first successful container request (#20)

Call internal lyft API using reflection when the executor was successfully requested from K8s

(cherry picked from commit 860cb05)

Author: catalinii

core/src/main/scala/org/apache/spark/util/LyftUtils.scala

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * 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.
+ */
+
+package org.apache.spark.util
+
+private[spark] object LyftUtils {
+  def callObjectMethodNoArguments(objectName: String, method: String): Boolean = {
+    var ok = true
+    try {
+      val m = Utils.classForName(objectName).getField("MODULE$").get(null)
+      Utils.classForName(objectName).getDeclaredMethod(method).invoke(m)
+    } catch {
+      case e: Throwable => ok = false
+    }
+    ok
+  }
+}

core/src/test/scala/org/apache/spark/util/LyftUtilsSuite.scala

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * 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.
+ */
+
+package org.apache.spark.util
+
+import org.apache.spark.{SparkException, SparkFunSuite}
+import org.apache.spark.internal.Logging
+
+object TestObjectLyftUtils {
+  var testVar = 0L
+  def setVal() = {
+    testVar = 1L
+  }
+}
+
+class LyftUtilsSuite extends SparkFunSuite with ResetSystemProperties with Logging {
+
+  test("callObjectMethodNoArguments") {
+    // Test calling the method using reflection 1
+    val v = LyftUtils.callObjectMethodNoArguments("org.apache.spark.util.TestObjectLyftUtils$", "setVal")
+    assert(v === true)
+    assert(TestObjectLyftUtils.testVar === 1)
+    assert(false ==
+      LyftUtils.callObjectMethodNoArguments("org.apache.spark.util.TestObjectLyftUtils$", "setVal1"))
+  }
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala

@@ -466,8 +466,12 @@ class ExecutorPodsAllocator(
         .build()
       val resources = replacePVCsIfNeeded(
         podWithAttachedContainer, resolvedExecutorSpec.executorKubernetesResources, reusablePVCs)
-      val createdExecutorPod =
-        kubernetesClient.pods().inNamespace(namespace).resource(podWithAttachedContainer).create()
+      val createdExecutorPod = kubernetesClient.pods().create(podWithAttachedContainer)
+      
+      org.apache.spark.util.LyftUtils.callObjectMethodNoArguments(
+          "com.lyft.data.spark.AppMetrics$",
+          "setFirstExecutorAllocationTime")
+      
       try {
         addOwnerReference(createdExecutorPod, resources)
         resources