Fix the Spark remote debugging issue on Spark 2.1 cluster

Signed-off-by: Wei Zhang <[email protected]>

Author: wezhang

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/azure/hdinsight/spark/run/SparkBatchJobDebuggerRunner.java

@@ -85,7 +85,6 @@ public class SparkBatchJobDebuggerRunner extends GenericDebuggerRunner {
 
     // More complex pattern, please use grok
     private final Pattern simpleLogPattern = Pattern.compile("\\d{1,2}[/-]\\d{1,2}[/-]\\d{1,2} \\d{1,2}:\\d{1,2}:\\d{1,2} (INFO|WARN|ERROR) .*", Pattern.DOTALL);
-    private final Pattern executorLogUrlPattern = Pattern.compile("^\\s+SPARK_LOG_URL_STDERR -> https?://([^:]+):?\\d*/node/containerlogs/(container.*)/livy/stderr.*");
 
     public void setDebugJob(SparkBatchRemoteDebugJob debugJob) {
         this.debugJob = debugJob;
@@ -158,6 +157,7 @@ protected void execute(@NotNull ExecutionEnvironment environment, @Nullable Call
 
                         // Create Driver debug process
                         createDebugProcess(
+                                remoteDebugJob,
                                 environment,
                                 callback,
                                 submissionState,
@@ -383,7 +383,8 @@ public SparkBatchDebugSession createSparkBatchDebugSession(
     /*
      * Create a debug process, if it's a Driver process, the following Executor processes will be created
      */
-    private void createDebugProcess(@NotNull ExecutionEnvironment environment,
+    private void createDebugProcess(SparkBatchRemoteDebugJob remoteDebugJob,
+                                    @NotNull ExecutionEnvironment environment,
                                     @Nullable Callback callback,
                                     @NotNull SparkBatchJobSubmissionState submissionState,
                                     boolean isDriver,
@@ -392,7 +393,7 @@ private void createDebugProcess(@NotNull ExecutionEnvironment environment,
                                     String remoteHost,
                                     int remotePort,
                                     String logUrl,
-                                    final CredentialsProvider credentialsProvider ) {
+                                    final CredentialsProvider credentialsProvider) {
         SparkBatchDebugSession session = getDebugSession().orElse(null);
         if (session == null) {
             return;
@@ -412,7 +413,7 @@ private void createDebugProcess(@NotNull ExecutionEnvironment environment,
                 if (isDriver) {
                     debugProcessOb = debugProcessOb.share();
 
-                    matchExecutorFromDebugProcessObservable(debugProcessOb)
+                    getExecutorsObservable(remoteDebugJob)
                             .subscribe(hostContainerPair -> {
                                 String host = hostContainerPair.getKey();
                                 String containerId = hostContainerPair.getValue();
@@ -448,7 +449,7 @@ private void createDebugProcess(@NotNull ExecutionEnvironment environment,
                                     }
 
                                     // create debug process for the Spark job executor
-                                    createDebugProcess( environment,
+                                    createDebugProcess(remoteDebugJob, environment,
                                                         callback,
                                                         newExecutorState,
                                                         false,
@@ -515,6 +516,18 @@ public void processTerminated(ProcessEvent processEvent) {
         .subscribe(debugProcessConsole::onNext, debugSessionSubscriber::onError, debugPhaser::arriveAndDeregister);
     }
 
+    /**
+     * To get Executor from Yarn UI App Attempt page
+     */
+    private Observable<SimpleEntry<String, String>> getExecutorsObservable(@NotNull SparkBatchRemoteDebugJob sparkDebugJob) {
+        return sparkDebugJob
+                .getSparkJobYarnCurrentAppAttempt()
+                .flatMap(appAttempt -> sparkDebugJob.getSparkJobYarnContainersObservable(appAttempt)
+                                                    .filter(hostContainerPair -> !StringUtils.equals(
+                                                            hostContainerPair.getValue(), appAttempt.getContainerId())))
+                .map(kv -> new SimpleEntry<>(kv.getKey(), kv.getValue()));
+    }
+
     /*
      * Create an Observable for a debug process, the Yarn log 'stderr' will be considered as the events
      * with its type key.
@@ -547,39 +560,6 @@ private Observable<SimpleEntry<String, Key>> createDebugProcessObservable(
                 .filter(lineKeyPair -> lineKeyPair.getKey() != null);
     }
 
-    /**
-     * To match Executor lunch content from debug process Observable
-     *
-     * @param debugProcessOb the debug process Observable to match
-     * @return matched Executor Observable, the event is SimpleEntry with host, containerId pair
-     */
-    private Observable<SimpleEntry<String, String>> matchExecutorFromDebugProcessObservable(
-                                                        Observable<SimpleEntry<String, Key>> debugProcessOb) {
-        PublishSubject<String> closeSubject = PublishSubject.create();
-        PublishSubject<String> openSubject = PublishSubject.create();
-
-        return debugProcessOb
-                .map(lineKeyPair -> {
-                    String line = lineKeyPair.getKey();
-
-                    if (line.matches("^YARN executor launch context:$")) {
-                        openSubject.onNext("YARN executor launch");
-                    }
-
-                    if (line.matches("^={5,}$")) {
-                        closeSubject.onNext("=====");
-                    }
-
-                    return line;
-                })
-                .window(openSubject, s -> closeSubject)
-                .flatMap(executorLunchContextOb -> executorLunchContextOb
-                                                    .map(executorLogUrlPattern::matcher)
-                                                    .filter(Matcher::matches)
-                                                    .map(matcher -> new SimpleEntry<>(matcher.group(1), matcher.group(2)))
-                );
-    }
-
     protected int getLogReadBlockSize() {
         return 4096;
     }

Utils/hdinsight-node-common/Test/java/com/microsoft/azure/hdinsight/spark/common/SparkBatchRemoteDebugJobScenario.java

@@ -22,12 +22,14 @@
 
 package com.microsoft.azure.hdinsight.spark.common;
 
-import com.microsoft.azure.hdinsight.sdk.common.HttpResponse;
+import com.microsoft.azure.hdinsight.sdk.rest.yarn.rm.AppAttempt;
 import cucumber.api.java.Before;
+import cucumber.api.java.en.And;
 import cucumber.api.java.en.Given;
 import cucumber.api.java.en.Then;
 import org.mockito.ArgumentCaptor;
 import org.slf4j.Logger;
+import rx.Observable;
 
 import java.net.URI;
 import java.util.HashMap;
@@ -228,4 +230,21 @@ public void checkGetSparkDriverHost(
             assertEquals(expectedHost, "__exception_got__");
         }
     }
+
+    @And("^mock method getSparkJobApplicationIdObservable to return '(.+)' Observable$")
+    public void mockMethodGetSparkJobApplicationIdObservable(String appIdMock) {
+        when(debugJobMock.getSparkJobApplicationIdObservable()).thenReturn(Observable.just(appIdMock));
+    }
+
+    @Then("^getting current Yarn App attempt should be '(.+)'$")
+    public void checkGetCurrentYarnAppAttemptResult(String appAttemptIdExpect) {
+        when(debugJobMock.getConnectUri()).thenReturn(URI.create(httpServerMock.completeUrl("/")));
+
+        AppAttempt appAttempt = debugJobMock
+                .getSparkJobYarnCurrentAppAttempt()
+                .toBlocking()
+                .first();
+
+        assertEquals(appAttemptIdExpect, appAttempt.getAppAttemptId());
+    }
 }

Utils/hdinsight-node-common/Test/resources/com/microsoft/azure/hdinsight/spark/common/SparkBatchRemoteDebugJobScenario.feature

@@ -65,4 +65,9 @@ Feature: Spark Batch Remote Debug Job Testing
     Given setup a mock livy service for GET request '/batch/9' to return '{"id":9,"state":"starting","appId":"application_1492415936046_0015","appInfo":{"driverLogUrl":"http://127.0.0.1:$port/yarnui/10.0.0.15/node/containerlogs/container_e02_1492415936046_0015_01_000001/livy","sparkUiUrl":"https://spkdbg.azurehdinsight.net/yarnui/hn/proxy/application_1492415936046_0015/"},"log":["\\t ApplicationMaster RPC port: -1","\\t queue: default","\\t start time: 1492569369011","\\t final status: UNDEFINED","\\t tracking URL: https://spkdbg.azurehdinsight.net/yarnui/hn/proxy/application_1492415936046_0015/","\\t user: livy","17/04/19 02:36:09 INFO ShutdownHookManager: Shutdown hook called","17/04/19 02:36:09 INFO ShutdownHookManager: Deleting directory /tmp/spark-1984dc9d-acd4-4648-9104-398431590f8e","YARN Diagnostics:","AM container is launched, waiting for AM container to Register with RM"]}' with status code 200
     And setup a mock livy service for GET request '/yarnui/ws/v1/cluster/apps/application_1492415936046_0015' to return '{ "app": { "amNodeLabelExpression": "", "finishedTime": 1493100184345, "startedTime": 1493097873053, "priority": 0, "applicationTags": "livy-batch-15-bcixsxv0", "applicationType": "SPARK", "clusterId": 1492780173422, "diagnostics": "Application application_1492780173422_0003 failed 5 times due to ApplicationMaster for attempt appattempt_1492780173422_0003_000005 timed out. Failing the application.", "trackingUrl": "http://hn0-zhwe-s.uhunwunss5gupibv1jib3beicb.lx.internal.cloudapp.net:8088/cluster/app/application_1492780173422_0003", "id": "application_1492780173422_0003", "user": "livy", "name": "SparkCore_WasbIOTest", "queue": "default", "state": "FAILED", "finalStatus": "FAILED", "progress": 100, "trackingUI": "History", "elapsedTime": 2311292, "amContainerLogs": "http://10.0.0.15:30060/node/containerlogs/container_e03_1492780173422_0003_05_000001/livy", "amHostHttpAddress": "10.0.0.15:30060", "allocatedMB": -1, "allocatedVCores": -1, "runningContainers": -1, "memorySeconds": 3549035, "vcoreSeconds": 2308, "queueUsagePercentage": 0, "clusterUsagePercentage": 0, "preemptedResourceMB": 0, "preemptedResourceVCores": 0, "numNonAMContainerPreempted": 0, "numAMContainerPreempted": 0, "logAggregationStatus": "SUCCEEDED", "unmanagedApplication": false }}' with status code 200
     Then getting Spark driver host from URL '/batch', batch ID 9 should be '__exception_got__'
-    Then throw exception 'java.net.UnknownServiceException' with message 'The Livy job 9 on yarn is not running.'
+    Then throw exception 'java.net.UnknownServiceException' with message 'The Livy job 9 on yarn is not running.'
+
+  Scenario: getSparkJobYarnCurrentAppAttempt integration test with response
+    Given setup a mock livy service for GET request '/yarnui/ws/v1/cluster/apps/application_1513565654634_0011/appattempts' to return '{"appAttempts":{"appAttempt":[{"id":1,"startTime":1513673984219,"finishedTime":0,"containerId":"container_1513565654634_0011_01_000001","nodeHttpAddress":"10.0.0.6:30060","nodeId":"10.0.0.6:30050","logsLink":"http://10.0.0.6:30060/node/containerlogs/container_1513565654634_0011_01_000001/livy","blacklistedNodes":"","appAttemptId":"appattempt_1513565654634_0011_000001"},{"id":2,"startTime":1513673985219,"finishedTime":0,"containerId":"container_1513565654634_0011_01_000002","nodeHttpAddress":"10.0.0.7:30060","nodeId":"10.0.0.7:30050","logsLink":"http://10.0.0.7:30060/node/containerlogs/container_1513565654634_0011_01_000002/livy","blacklistedNodes":"","appAttemptId":"appattempt_1513565654634_0011_000002"}]}}' with status code 200
+   And mock method getSparkJobApplicationIdObservable to return 'application_1513565654634_0011' Observable
+   Then getting current Yarn App attempt should be 'appattempt_1513565654634_0011_000002'

Utils/hdinsight-node-common/src/com/microsoft/azure/hdinsight/sdk/rest/yarn/rm/AppAttempt.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright (c) Microsoft Corporation
+ * <p/>
+ * All rights reserved.
+ * <p/>
+ * MIT License
+ * <p/>
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
+ * to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ * <p/>
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
+ * the Software.
+ * <p/>
+ * THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
+ * THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+ * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.microsoft.azure.hdinsight.sdk.rest.yarn.rm;
+
+public class AppAttempt {
+    private int id;                  // The app attempt id
+    private String nodeId;           // The node id of the node the attempt ran on
+    private String nodeHttpAddress;  // The node http address of the node the attempt ran on
+    private String logsLink;         // The http link to the app attempt logs
+    private String containerId;      // The id of the container for the app attempt
+    private long startTime;          // The start time of the attempt (in ms since epoch)
+    private long finishedTime;       // The end time of the attempt (in ms since epoch), 0 for not end
+    private String blacklistedNodes; // Nodes blacklist
+    private String appAttemptId;     // App Attempt Id
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public String getNodeId() {
+        return nodeId;
+    }
+
+    public void setNodeId(String nodeId) {
+        this.nodeId = nodeId;
+    }
+
+    public String getNodeHttpAddress() {
+        return nodeHttpAddress;
+    }
+
+    public void setNodeHttpAddress(String nodeHttpAddress) {
+        this.nodeHttpAddress = nodeHttpAddress;
+    }
+
+    public String getLogsLink() {
+        return logsLink;
+    }
+
+    public void setLogsLink(String logsLink) {
+        this.logsLink = logsLink;
+    }
+
+    public String getContainerId() {
+        return containerId;
+    }
+
+    public void setContainerId(String containerId) {
+        this.containerId = containerId;
+    }
+
+    public long getStartTime() {
+        return startTime;
+    }
+
+    public void setStartTime(long startTime) {
+        this.startTime = startTime;
+    }
+
+    public long getFinishedTime() {
+        return finishedTime;
+    }
+
+    public void setFinishedTime(long finishedTime) {
+        this.finishedTime = finishedTime;
+    }
+
+    public String getBlacklistedNodes() {
+        return blacklistedNodes;
+    }
+
+    public void setBlacklistedNodes(String blacklistedNodes) {
+        this.blacklistedNodes = blacklistedNodes;
+    }
+
+    public String getAppAttemptId() {
+        return appAttemptId;
+    }
+
+    public void setAppAttemptId(String appAttemptId) {
+        this.appAttemptId = appAttemptId;
+    }
+}

Utils/hdinsight-node-common/src/com/microsoft/azure/hdinsight/sdk/rest/yarn/rm/AppAttemptsResponse.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) Microsoft Corporation
+ * <p/>
+ * All rights reserved.
+ * <p/>
+ * MIT License
+ * <p/>
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
+ * to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ * <p/>
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
+ * the Software.
+ * <p/>
+ * THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
+ * THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+ * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.microsoft.azure.hdinsight.sdk.rest.yarn.rm;
+
+import java.util.List;
+
+public class AppAttemptsResponse {
+    public class AppAttempts {
+        public List<AppAttempt> appAttempt;
+    }
+
+    private AppAttempts appAttempts;
+
+    public AppAttempts getAppAttempts() {
+        return appAttempts;
+    }
+
+    public void setAppAttempts(AppAttempts appAttempts) {
+        this.appAttempts = appAttempts;
+    }
+}