You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by we...@apache.org on 2015/09/19 06:12:56 UTC
incubator-reef git commit: [REEF-663] Make Http Endpoint available
for client after submit
Repository: incubator-reef
Updated Branches:
refs/heads/master 6eefbf1cc -> 189c5acba
[REEF-663] Make Http Endpoint available for client after submit
- This PR is to return HttpEndPoint to the client after Job is submitted
- At Java side, the http endpoint is written to JobSubmission folder in HDFS
and Driver folder
- At REEFClient side, HttpClientHelper is added to pull the http endpoint from
the file and return it in Submit()
- E2e test cases are updated to get the URI and verify on it.
- The new Submit API is noted as [Ustable] as we might change it in future
- The change is part of the change in tlc branch with some modifications
JIRA:
[REEF-663](https://issues.apache.org/jira/browse/REEF-663)
Pull Request:
This closes #505
Project: http://git-wip-us.apache.org/repos/asf/incubator-reef/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-reef/commit/189c5acb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-reef/tree/189c5acb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-reef/diff/189c5acb
Branch: refs/heads/master
Commit: 189c5acba1bb5a1f8e1393a02dbe643676f959fe
Parents: 6eefbf1
Author: Julia Wang <ju...@apache.org>
Authored: Fri Sep 18 12:47:01 2015 -0700
Committer: Markus Weimer <we...@apache.org>
Committed: Fri Sep 18 21:10:52 2015 -0700
----------------------------------------------------------------------
.../Org.Apache.REEF.Client/API/IREEFClient.cs | 13 +
.../Common/HttpClientHelper.cs | 344 +++++++++++++++++++
.../Common/IDriverHttpEndpoint.cs | 31 ++
.../Org.Apache.REEF.Client/Local/LocalClient.cs | 53 ++-
.../Org.Apache.REEF.Client.csproj | 7 +
.../Properties/AssemblyInfo.cs | 7 +
.../Org.Apache.REEF.Client/YARN/YARNClient.cs | 33 +-
.../Files/REEFFileNames.cs | 8 +
.../AllHandlers.cs | 11 +-
.../Functional/Bridge/TestBridgeClient.cs | 13 +-
.../Functional/ReefFunctionalTest.cs | 38 +-
.../apache/reef/bridge/client/LocalClient.java | 8 +-
.../bridge/client/LocalSubmissionFromCS.java | 2 +
.../bridge/client/YarnJobSubmissionClient.java | 66 +++-
.../reef/javabridge/generic/JobDriver.java | 21 ++
.../runtime/common/files/REEFFileNames.java | 10 +-
.../client/HDInsightDriverConfiguration.java | 2 +-
.../yarn/client/YarnSubmissionHelper.java | 8 +
.../yarn/driver/UploaderToJobfolder.java | 2 +-
.../yarn/driver/YarnContainerManager.java | 17 +-
.../yarn/driver/YarnDriverConfiguration.java | 2 +-
.../parameters/JobSubmissionDirectory.java | 5 +-
22 files changed, 669 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Client/API/IREEFClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/API/IREEFClient.cs b/lang/cs/Org.Apache.REEF.Client/API/IREEFClient.cs
index e4f4c88..74b8d95 100644
--- a/lang/cs/Org.Apache.REEF.Client/API/IREEFClient.cs
+++ b/lang/cs/Org.Apache.REEF.Client/API/IREEFClient.cs
@@ -17,6 +17,10 @@
* under the License.
*/
+using System;
+using Org.Apache.REEF.Client.Common;
+using Org.Apache.REEF.Common.Attributes;
+
namespace Org.Apache.REEF.Client.API
{
/// <summary>
@@ -30,5 +34,14 @@ namespace Org.Apache.REEF.Client.API
/// </summary>
/// <param name="jobSubmission"></param>
void Submit(IJobSubmission jobSubmission);
+
+ /// <summary>
+ /// Submit the job described in jobSubmission to the cluster.
+ /// Expect IDriverHttpEndpoint returned after the call.
+ /// </summary>
+ /// <param name="jobSubmission"></param>
+ /// <returns>IDriverHttpEndpoint</returns>
+ [Unstable("0.13", "Working in progress for what to return after submit")]
+ IDriverHttpEndpoint SubmitAndGetDriverUrl(IJobSubmission jobSubmission);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Client/Common/HttpClientHelper.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Common/HttpClientHelper.cs b/lang/cs/Org.Apache.REEF.Client/Common/HttpClientHelper.cs
new file mode 100644
index 0000000..1a00493
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/Common/HttpClientHelper.cs
@@ -0,0 +1,344 @@
+/**
+ * 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.
+ */
+
+using System.Collections.Generic;
+using Newtonsoft.Json;
+using Org.Apache.REEF.Utilities.Logging;
+using System;
+using System.IO;
+using System.Net;
+using System.Net.Http;
+using System.Net.Http.Headers;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Org.Apache.REEF.Client.Common
+{
+ internal class HttpClientHelper : IDriverHttpEndpoint
+ {
+ private static readonly Logger LOGGER = Logger.GetLogger(typeof (HttpClientHelper));
+ private const int MaxConnectAttemptCount = 20;
+ private const int MilliSecondsToWaitBeforeNextConnectAttempt = 1000;
+ private const int SecondsForHttpClientTimeout = 120;
+ private const string UnAssigned = "UNASSIGNED";
+ private const string TrackingUrlKey = "trackingUrl";
+ private const string AppKey = "app";
+ private const string ThisIsStandbyRm = "This is standby RM";
+ private const string AppJson = "application/json";
+
+ private string _driverUrl;
+
+ private readonly HttpClient _client;
+
+ internal HttpClientHelper()
+ {
+ _client = new HttpClient
+ {
+ Timeout = TimeSpan.FromSeconds(SecondsForHttpClientTimeout),
+ };
+ _client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue(AppJson));
+ }
+
+ public string DriverUrl { get { return _driverUrl; } }
+
+ public string GetUrlResult(string url)
+ {
+ var task = Task.Run(() => CallUrl(url));
+ task.Wait();
+ return task.Result;
+ }
+
+ enum UrlResultKind
+ {
+ WasNotAbleToTalkToRm,
+ BackupRm,
+ AppIdNotThereYet,
+ UrlNotAssignedYet,
+ GotAppIdUrl,
+ }
+
+ internal static List<string> GetRmUri(string filePath)
+ {
+ using (var sr = new StreamReader(File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.Read)))
+ {
+ sr.ReadLine(); // appid
+ sr.ReadLine(); // trackingUrl
+ var rmList = new List<string>();
+ var rmUri = sr.ReadLine();
+ while (rmUri != null)
+ {
+ rmList.Add(rmUri);
+ rmUri = sr.ReadLine();
+ }
+ return rmList;
+ }
+ }
+
+ internal static string GetAppId(string filePath)
+ {
+ using (var sr = new StreamReader(File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.Read)))
+ {
+ var appId = sr.ReadLine();
+ return appId;
+ }
+ }
+
+ internal static string GetTrackingUrl(string filePath)
+ {
+ using (var sr = new StreamReader(File.Open(filePath, FileMode.Open, FileAccess.Read, FileShare.Read)))
+ {
+ sr.ReadLine(); // appid
+ var trackingUrl = sr.ReadLine();
+ return "http://" + trackingUrl + "/";
+ }
+ }
+
+ internal async Task<string> CallUrl (string url)
+ {
+ var result = await TryGetUri(url);
+ if (HasCommandFailed(result))
+ {
+ return null;
+ }
+ LOGGER.Log(Level.Warning, "CallUrl result " + result.Item2);
+ return result.Item2;
+ }
+
+ internal string GetDriverUrlForYarn(String filePath)
+ {
+ _driverUrl = GetTrackingUrl(filePath);
+ return _driverUrl;
+ }
+
+ internal string GetDriverUrlForLocalRuntime(string filePath)
+ {
+ _driverUrl = null;
+ for (int i = 0; i < 10; i++)
+ {
+ var driverUrl = TryReadHttpServerIpAndPortFromFile(filePath);
+ if (!string.IsNullOrEmpty(driverUrl))
+ {
+ _driverUrl = "http://" + driverUrl + "/";
+ break;
+ }
+ Thread.Sleep(1000);
+ }
+ return _driverUrl;
+ }
+
+ private string TryReadHttpServerIpAndPortFromFile(String fileName)
+ {
+ string httpServerIpAndPort = null;
+ try
+ {
+ LOGGER.Log(Level.Info, "try open " + fileName);
+ using (var rdr = new StreamReader(File.OpenRead(fileName)))
+ {
+ httpServerIpAndPort = rdr.ReadLine();
+ LOGGER.Log(Level.Info, "httpServerIpAndPort is " + httpServerIpAndPort);
+ }
+ }
+ catch (FileNotFoundException)
+ {
+ LOGGER.Log(Level.Info, "File does not exist: " + fileName);
+ }
+ return httpServerIpAndPort;
+ }
+
+ internal async Task<string> GetAppIdTrackingUrl(string url)
+ {
+ var result = await TryGetUri(url);
+ if (HasCommandFailed(result) ||
+ result.Item2 == null)
+ {
+ return null;
+ }
+
+ LOGGER.Log(Level.Info, "GetAppIdTrackingUrl: " + result.Item2);
+ return result.Item2;
+ }
+
+ private static bool ShouldRetry(HttpRequestException httpRequestException)
+ {
+ var shouldRetry = false;
+ if (httpRequestException.Message.IndexOf(((int)(HttpStatusCode.NotFound)).ToString(), StringComparison.Ordinal) != -1 ||
+ httpRequestException.Message.IndexOf(((int)(HttpStatusCode.BadGateway)).ToString(), StringComparison.Ordinal) != -1)
+ {
+ shouldRetry = true;
+ }
+ else
+ {
+ var webException = httpRequestException.InnerException as System.Net.WebException;
+ if (webException != null)
+ {
+ if (webException.Status == System.Net.WebExceptionStatus.ConnectFailure)
+ {
+ shouldRetry = true;
+ }
+ }
+ }
+ return shouldRetry;
+ }
+
+ private static Tuple<bool, string> CommandFailed(String reason)
+ {
+ return new Tuple<bool, string>(false, null);
+ }
+
+ private static Tuple<bool, string> CommandSucceeded(string commandResult)
+ {
+ return new Tuple<bool, string>(true, commandResult);
+ }
+
+ private bool HasCommandFailed(Tuple<bool, string> httpCallResult)
+ {
+ return !httpCallResult.Item1;
+ }
+
+ internal async Task<Tuple<bool, string>> TryGetUri(string commandUri)
+ {
+ var connectAttemptCount = 0;
+ Tuple<bool, string> result;
+
+ while (true)
+ {
+ try
+ {
+ string strResult = null;
+ LOGGER.Log(Level.Warning, "Try url [" + commandUri + "] connectAttemptCount " + connectAttemptCount + ".");
+ strResult = await _client.GetStringAsync(commandUri);
+ result = CommandSucceeded(strResult);
+ LOGGER.Log(Level.Warning, "Connection succeeded. connectAttemptCount was " + connectAttemptCount + ".");
+ break;
+ }
+ catch (HttpRequestException httpRequestException)
+ {
+ if (!ShouldRetry(httpRequestException))
+ {
+ LOGGER.Log(Level.Error,
+ commandUri + " exception " + httpRequestException.Message + "\n" +
+ httpRequestException.StackTrace);
+ result = CommandFailed(httpRequestException.Message);
+ LOGGER.Log(Level.Warning, "Connection failed. connectAttemptCount was " + connectAttemptCount + ".");
+ break;
+ }
+ }
+ catch (Exception ex)
+ {
+ LOGGER.Log(Level.Error, commandUri + " exception " + ex.Message + "\n" + ex.StackTrace);
+ result = CommandFailed(ex.Message);
+ LOGGER.Log(Level.Warning, "Connection failed. connectAttemptCount was " + connectAttemptCount + ".");
+ break;
+ }
+
+ ++connectAttemptCount;
+ if (connectAttemptCount >= MaxConnectAttemptCount)
+ {
+ result = CommandFailed("Could not connect to " + commandUri + " after " + MaxConnectAttemptCount.ToString() + "attempts.");
+ LOGGER.Log(Level.Warning, "Connection failed. connectAttemptCount was " + connectAttemptCount + ".");
+ break;
+ }
+
+ Thread.Sleep(MilliSecondsToWaitBeforeNextConnectAttempt);
+ }
+
+ return result;
+ }
+
+ internal async Task<string> TryUntilNoConnection(string commandUri)
+ {
+ var connectAttemptCount = 0;
+ while (true)
+ {
+ try
+ {
+ var strResult = await _client.GetStringAsync(commandUri);
+ LOGGER.Log(Level.Info,
+ "Connection succeeded. connectAttemptCount was " + connectAttemptCount + ".");
+ }
+ catch (HttpRequestException httpRequestException)
+ {
+ LOGGER.Log(Level.Info, httpRequestException.Message);
+ break;
+ }
+ catch (Exception e)
+ {
+ LOGGER.Log(Level.Info, e.Message);
+ break;
+ }
+
+ ++connectAttemptCount;
+ if (connectAttemptCount >= MaxConnectAttemptCount)
+ {
+ LOGGER.Log(Level.Info, "Can still connect to " + commandUri + " after " + MaxConnectAttemptCount.ToString() + "attempts.");
+ break;
+ }
+
+ Thread.Sleep(MilliSecondsToWaitBeforeNextConnectAttempt);
+ }
+
+ return null;
+ }
+
+ private static bool ShouldRetry(HttpStatusCode httpStatusCode)
+ {
+ return httpStatusCode == HttpStatusCode.NotFound;
+ }
+
+ private UrlResultKind CheckUrlAttempt(string result)
+ {
+ UrlResultKind resultKind = UrlResultKind.WasNotAbleToTalkToRm;
+ if (string.IsNullOrEmpty(result))
+ {
+ resultKind = UrlResultKind.WasNotAbleToTalkToRm;
+ }
+ else if (result.StartsWith(ThisIsStandbyRm))
+ {
+ resultKind = UrlResultKind.BackupRm;
+ }
+ else
+ {
+ dynamic deserializedValue = JsonConvert.DeserializeObject(result);
+ var values = deserializedValue[AppKey];
+ if (values == null || values[TrackingUrlKey] == null)
+ {
+ resultKind = UrlResultKind.AppIdNotThereYet;
+ }
+ else
+ {
+ _driverUrl = values[TrackingUrlKey].ToString();
+ LOGGER.Log(Level.Info, "trackingUrl[" + _driverUrl + "]");
+
+ if (0 == String.Compare(_driverUrl, UnAssigned))
+ {
+ resultKind = UrlResultKind.UrlNotAssignedYet;
+ }
+ else
+ {
+ resultKind = UrlResultKind.GotAppIdUrl;
+ }
+
+ }
+ }
+
+ LOGGER.Log(Level.Info, "CheckUrlAttempt " + resultKind);
+ return resultKind;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Client/Common/IDriverHttpEndpoint.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Common/IDriverHttpEndpoint.cs b/lang/cs/Org.Apache.REEF.Client/Common/IDriverHttpEndpoint.cs
new file mode 100644
index 0000000..c128ad0
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/Common/IDriverHttpEndpoint.cs
@@ -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.
+ */
+
+using Org.Apache.REEF.Common.Attributes;
+
+namespace Org.Apache.REEF.Client.Common
+{
+ [Unstable("0.13", "Working in progress for what to return after submit")]
+ public interface IDriverHttpEndpoint
+ {
+ string GetUrlResult(string url);
+
+ string DriverUrl { get; }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Client/Local/LocalClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Local/LocalClient.cs b/lang/cs/Org.Apache.REEF.Client/Local/LocalClient.cs
index 518533e..af2d47b 100644
--- a/lang/cs/Org.Apache.REEF.Client/Local/LocalClient.cs
+++ b/lang/cs/Org.Apache.REEF.Client/Local/LocalClient.cs
@@ -20,9 +20,11 @@
using System;
using System.IO;
using System.Linq;
+using System.Threading.Tasks;
using Org.Apache.REEF.Client.API;
using Org.Apache.REEF.Client.Common;
using Org.Apache.REEF.Client.Local.Parameters;
+using Org.Apache.REEF.Common.Files;
using Org.Apache.REEF.Tang.Annotations;
using Org.Apache.REEF.Tang.Implementations.Tang;
using Org.Apache.REEF.Utilities.Logging;
@@ -49,29 +51,37 @@ namespace Org.Apache.REEF.Client.Local
private readonly JavaClientLauncher _javaClientLauncher;
private readonly int _numberOfEvaluators;
private readonly string _runtimeFolder;
+ private string _driverUrl;
+ private REEFFileNames _fileNames;
[Inject]
private LocalClient(DriverFolderPreparationHelper driverFolderPreparationHelper,
[Parameter(typeof(LocalRuntimeDirectory))] string runtimeFolder,
- [Parameter(typeof(NumberOfEvaluators))] int numberOfEvaluators, JavaClientLauncher javaClientLauncher)
+ [Parameter(typeof(NumberOfEvaluators))] int numberOfEvaluators,
+ JavaClientLauncher javaClientLauncher,
+ REEFFileNames fileNames)
{
_driverFolderPreparationHelper = driverFolderPreparationHelper;
_runtimeFolder = runtimeFolder;
_numberOfEvaluators = numberOfEvaluators;
_javaClientLauncher = javaClientLauncher;
+ _fileNames = fileNames;
}
/// <summary>
/// Uses Path.GetTempPath() as the runtime execution folder.
/// </summary>
/// <param name="driverFolderPreparationHelper"></param>
- /// <param name="reefJarPath"></param>
/// <param name="numberOfEvaluators"></param>
+ /// <param name="javaClientLauncher"></param>
+ /// <param name="fileNames"></param>
[Inject]
private LocalClient(
DriverFolderPreparationHelper driverFolderPreparationHelper,
- [Parameter(typeof(NumberOfEvaluators))] int numberOfEvaluators, JavaClientLauncher javaClientLauncher)
- : this(driverFolderPreparationHelper, Path.GetTempPath(), numberOfEvaluators, javaClientLauncher)
+ [Parameter(typeof(NumberOfEvaluators))] int numberOfEvaluators,
+ JavaClientLauncher javaClientLauncher,
+ REEFFileNames fileNames)
+ : this(driverFolderPreparationHelper, Path.GetTempPath(), numberOfEvaluators, javaClientLauncher, fileNames)
{
// Intentionally left blank.
}
@@ -99,6 +109,41 @@ namespace Org.Apache.REEF.Client.Local
Logger.Log(Level.Info, "Submitted the Driver for execution.");
}
+ public IDriverHttpEndpoint SubmitAndGetDriverUrl(IJobSubmission jobSubmission)
+ {
+ // Prepare the job submission folder
+ var jobFolder = CreateJobFolder(jobSubmission.JobIdentifier);
+ var driverFolder = Path.Combine(jobFolder, DriverFolderName);
+ Logger.Log(Level.Info, "Preparing driver folder in " + driverFolder);
+
+ _driverFolderPreparationHelper.PrepareDriverFolder(jobSubmission, driverFolder);
+
+ //TODO: Remove this when we have a generalized way to pass config to java
+ var javaParams = TangFactory.GetTang()
+ .NewInjector(jobSubmission.DriverConfigurations.ToArray())
+ .GetInstance<ClrClient2JavaClientCuratedParameters>();
+
+ Task.Run(() =>
+ _javaClientLauncher.Launch(JavaClassName, driverFolder, jobSubmission.JobIdentifier,
+ _numberOfEvaluators.ToString(),
+ javaParams.TcpPortRangeStart.ToString(),
+ javaParams.TcpPortRangeCount.ToString(),
+ javaParams.TcpPortRangeTryCount.ToString()
+ ));
+
+ var fileName = Path.Combine(driverFolder, _fileNames.DriverHttpEndpoint);
+ HttpClientHelper helper = new HttpClientHelper();
+ _driverUrl = helper.GetDriverUrlForLocalRuntime(fileName);
+
+ Logger.Log(Level.Info, "Submitted the Driver for execution. Returned driverUrl is: " + _driverUrl);
+ return helper;
+ }
+
+ public string DriverUrl
+ {
+ get { return _driverUrl; }
+ }
+
/// <summary>
/// Creates the temporary directory to hold the job submission.
/// </summary>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Client/Org.Apache.REEF.Client.csproj
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Org.Apache.REEF.Client.csproj b/lang/cs/Org.Apache.REEF.Client/Org.Apache.REEF.Client.csproj
index ee19ca6..a21c2e2 100644
--- a/lang/cs/Org.Apache.REEF.Client/Org.Apache.REEF.Client.csproj
+++ b/lang/cs/Org.Apache.REEF.Client/Org.Apache.REEF.Client.csproj
@@ -37,10 +37,15 @@ under the License.
<UseVSHostingProcess>false</UseVSHostingProcess>
</PropertyGroup>
<ItemGroup>
+ <Reference Include="Microsoft.CSharp" />
+ <Reference Include="Newtonsoft.Json">
+ <HintPath>$(PackagesDir)\Newtonsoft.Json.$(NewtonsoftJsonVersion)\lib\net45\Newtonsoft.Json.dll</HintPath>
+ </Reference>
<Reference Include="System" />
<Reference Include="System.Core" />
<Reference Include="System.IO.Compression" />
<Reference Include="System.IO.Compression.FileSystem" />
+ <Reference Include="System.Net.Http" />
</ItemGroup>
<ItemGroup>
<Compile Include="API\ClientFactory.cs" />
@@ -58,6 +63,8 @@ under the License.
<Compile Include="Common\ClrClient2JavaClientCuratedParameters.cs" />
<Compile Include="Common\DriverFolderPreparationHelper.cs" />
<Compile Include="Common\FileSets.cs" />
+ <Compile Include="Common\HttpClientHelper.cs" />
+ <Compile Include="Common\IDriverHttpEndpoint.cs" />
<Compile Include="Common\JavaClientLauncher.cs" />
<Compile Include="Common\ResourceHelper.cs" />
<Compile Include="Local\LocalClient.cs" />
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Client/Properties/AssemblyInfo.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Properties/AssemblyInfo.cs b/lang/cs/Org.Apache.REEF.Client/Properties/AssemblyInfo.cs
index 8ea0bff..b503884 100644
--- a/lang/cs/Org.Apache.REEF.Client/Properties/AssemblyInfo.cs
+++ b/lang/cs/Org.Apache.REEF.Client/Properties/AssemblyInfo.cs
@@ -18,6 +18,7 @@
*/
using System.Reflection;
+using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
// General Information about an assembly is controlled through the following
@@ -52,3 +53,9 @@ using System.Runtime.InteropServices;
// [assembly: AssemblyVersion("1.0.*")]
[assembly: AssemblyVersion("0.13.0.0")]
[assembly: AssemblyFileVersion("0.13.0.0")]
+
+[assembly: InternalsVisibleTo("Org.Apache.REEF.Tests, publickey=" +
+ "00240000048000009400000006020000002400005253413100040000010001005df3e621d886a9" +
+ "9c03469d0f93a9f5d45aa2c883f50cd158759e93673f759ec4657fd84cc79d2db38ef1a2d914cc" +
+ "b7c717846a897e11dd22eb260a7ce2da2dccf0263ea63e2b3f7dac24f28882aa568ef544341d17" +
+ "618392a1095f4049ad079d4f4f0b429bb535699155fd6a7652ec7d6c1f1ba2b560f11ef3a86b5945d288cf")]
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Client/YARN/YARNClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/YARN/YARNClient.cs b/lang/cs/Org.Apache.REEF.Client/YARN/YARNClient.cs
index f7e27ad..cda887d 100644
--- a/lang/cs/Org.Apache.REEF.Client/YARN/YARNClient.cs
+++ b/lang/cs/Org.Apache.REEF.Client/YARN/YARNClient.cs
@@ -22,6 +22,7 @@ using System.IO;
using System.Linq;
using Org.Apache.REEF.Client.API;
using Org.Apache.REEF.Client.Common;
+using Org.Apache.REEF.Common.Files;
using Org.Apache.REEF.Tang.Annotations;
using Org.Apache.REEF.Tang.Implementations.Tang;
using Org.Apache.REEF.Utilities.Logging;
@@ -38,15 +39,19 @@ namespace Org.Apache.REEF.Client.YARN
private static readonly Logger Logger = Logger.GetLogger(typeof(YARNClient));
private readonly DriverFolderPreparationHelper _driverFolderPreparationHelper;
private readonly JavaClientLauncher _javaClientLauncher;
+ private String _driverUrl;
+ private REEFFileNames _fileNames;
[Inject]
internal YARNClient(JavaClientLauncher javaClientLauncher,
DriverFolderPreparationHelper driverFolderPreparationHelper,
+ REEFFileNames fileNames,
YarnCommandLineEnvironment yarn)
{
_javaClientLauncher = javaClientLauncher;
_javaClientLauncher.AddToClassPath(yarn.GetYarnClasspathList());
_driverFolderPreparationHelper = driverFolderPreparationHelper;
+ _fileNames = fileNames;
}
public void Submit(IJobSubmission jobSubmission)
@@ -55,6 +60,27 @@ namespace Org.Apache.REEF.Client.YARN
var driverFolderPath = CreateDriverFolder(jobSubmission.JobIdentifier);
Logger.Log(Level.Info, "Preparing driver folder in " + driverFolderPath);
+ Launch(jobSubmission, driverFolderPath);
+ }
+
+ public IDriverHttpEndpoint SubmitAndGetDriverUrl(IJobSubmission jobSubmission)
+ {
+ // Prepare the job submission folder
+ var driverFolderPath = CreateDriverFolder(jobSubmission.JobIdentifier);
+ Logger.Log(Level.Info, "Preparing driver folder in " + driverFolderPath);
+
+ Launch(jobSubmission, driverFolderPath);
+
+ var pointerFileName = Path.Combine(driverFolderPath, _fileNames.DriverHttpEndpoint);
+
+ var httpClient = new HttpClientHelper();
+ _driverUrl = httpClient.GetDriverUrlForYarn(pointerFileName);
+
+ return httpClient;
+ }
+
+ private void Launch(IJobSubmission jobSubmission, string driverFolderPath)
+ {
_driverFolderPreparationHelper.PrepareDriverFolder(jobSubmission, driverFolderPath);
//TODO: Remove this when we have a generalized way to pass config to java
@@ -71,7 +97,12 @@ namespace Org.Apache.REEF.Client.YARN
javaParams.MaxApplicationSubmissions.ToString(),
javaParams.DriverRestartEvaluatorRecoverySeconds.ToString()
);
- Logger.Log(Level.Info, "Submitted the Driver for execution.");
+ Logger.Log(Level.Info, "Submitted the Driver for execution." + jobSubmission.JobIdentifier);
+ }
+
+ public string DriverUrl
+ {
+ get { return _driverUrl; }
}
/// <summary>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Common/Files/REEFFileNames.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Common/Files/REEFFileNames.cs b/lang/cs/Org.Apache.REEF.Common/Files/REEFFileNames.cs
index 5316301..15ddd02 100644
--- a/lang/cs/Org.Apache.REEF.Common/Files/REEFFileNames.cs
+++ b/lang/cs/Org.Apache.REEF.Common/Files/REEFFileNames.cs
@@ -19,6 +19,7 @@
using System.Diagnostics.CodeAnalysis;
using System.IO;
+using Org.Apache.REEF.Common.Attributes;
using Org.Apache.REEF.Tang.Annotations;
namespace Org.Apache.REEF.Common.Files
@@ -48,6 +49,7 @@ namespace Org.Apache.REEF.Common.Files
private const string DRIVER_CONFIGURATION_NAME = "driver.conf";
private const string EVALUATOR_CONFIGURATION_NAME = "evaluator.conf";
private const string CLR_DRIVER_CONFIGURATION_NAME = "clrdriver.conf";
+ private const string DRIVER_HTTP_ENDPOINT_FILE_NAME = "DriverHttpEndpoint.txt";
private const string BRIDGE_EXE_NAME = "Org.Apache.REEF.Bridge.exe";
private const string BRIDGE_EXE_CONFIG_NAME = "Org.Apache.REEF.Bridge.exe.config";
@@ -229,6 +231,12 @@ namespace Org.Apache.REEF.Common.Files
return Path.Combine(REEF_BASE_FOLDER, BRIDGE_EXE_CONFIG_NAME);
}
+ /// <summary>
+ /// </summary>
+ /// <returns>File name that contains the dfs path for the DriverHttpEndpoint</returns>
+ [Unstable("0.13", "Working in progress for what to return after submit")]
+ public string DriverHttpEndpoint { get { return DRIVER_HTTP_ENDPOINT_FILE_NAME; } }
+
private static readonly string GLOBAL_FOLDER_PATH = Path.Combine(REEF_BASE_FOLDER, GLOBAL_FOLDER);
private static readonly string LOCAL_FOLDER_PATH = Path.Combine(REEF_BASE_FOLDER, LOCAL_FOLDER);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Examples.AllHandlers/AllHandlers.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Examples.AllHandlers/AllHandlers.cs b/lang/cs/Org.Apache.REEF.Examples.AllHandlers/AllHandlers.cs
index 27a553b..0eeb7d5 100644
--- a/lang/cs/Org.Apache.REEF.Examples.AllHandlers/AllHandlers.cs
+++ b/lang/cs/Org.Apache.REEF.Examples.AllHandlers/AllHandlers.cs
@@ -20,6 +20,7 @@
using System;
using System.IO;
using Org.Apache.REEF.Client.API;
+using Org.Apache.REEF.Client.Common;
using Org.Apache.REEF.Client.Local;
using Org.Apache.REEF.Client.YARN;
using Org.Apache.REEF.Common.Evaluator;
@@ -50,7 +51,7 @@ namespace Org.Apache.REEF.Examples.AllHandlers
_jobSubmissionBuilderFactory = jobSubmissionBuilderFactory;
}
- private void Run()
+ private IDriverHttpEndpoint Run()
{
var helloDriverConfiguration = DriverConfiguration.ConfigurationModule
.Set(DriverConfiguration.OnEvaluatorAllocated, GenericType<HelloAllocatedEvaluatorHandler>.Class)
@@ -82,7 +83,8 @@ namespace Org.Apache.REEF.Examples.AllHandlers
.SetJobIdentifier("HelloDriver")
.Build();
- _reefClient.Submit(helloJobSubmission);
+ IDriverHttpEndpoint driverHttpEndpoint = _reefClient.SubmitAndGetDriverUrl(helloJobSubmission);
+ return driverHttpEndpoint;
}
/// <summary></summary>
@@ -122,11 +124,12 @@ namespace Org.Apache.REEF.Examples.AllHandlers
/// args[0] specify either running local or YARN. Default is local
/// args[1] specify running folder. Default is REEF_LOCAL_RUNTIME
/// </remarks>
- public static void Run(string[] args)
+ public static IDriverHttpEndpoint Run(string[] args)
{
string runOnYarn = args.Length > 0 ? args[0] : Local;
string runtimeFolder = args.Length > 1 ? args[1] : "REEF_LOCAL_RUNTIME";
- TangFactory.GetTang().NewInjector(GetRuntimeConfiguration(runOnYarn, runtimeFolder)).GetInstance<AllHandlers>().Run();
+ IDriverHttpEndpoint driverEndPoint = TangFactory.GetTang().NewInjector(GetRuntimeConfiguration(runOnYarn, runtimeFolder)).GetInstance<AllHandlers>().Run();
+ return driverEndPoint;
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Tests/Functional/Bridge/TestBridgeClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/Bridge/TestBridgeClient.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/Bridge/TestBridgeClient.cs
index aa582c4..6a4e15c 100644
--- a/lang/cs/Org.Apache.REEF.Tests/Functional/Bridge/TestBridgeClient.cs
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/Bridge/TestBridgeClient.cs
@@ -19,6 +19,7 @@
using System;
using Microsoft.VisualStudio.TestTools.UnitTesting;
+using Org.Apache.REEF.Client.Common;
using Org.Apache.REEF.Examples.AllHandlers;
using Org.Apache.REEF.Utilities.Logging;
@@ -62,11 +63,19 @@ namespace Org.Apache.REEF.Tests.Functional.Bridge
RunClrBridgeClient(false, testRuntimeFolder);
}
- private void RunClrBridgeClient(bool runOnYarn, string testRuntimeFolder)
+ private async void RunClrBridgeClient(bool runOnYarn, string testRuntimeFolder)
{
string[] a = new[] { runOnYarn ? "yarn" : "local", testRuntimeFolder };
- AllHandlers.Run(a);
+ IDriverHttpEndpoint driverHttpEndpoint = AllHandlers.Run(a);
+
+ var uri = driverHttpEndpoint.DriverUrl + "NRT/status?a=1&b=2";
+ var strStatus = driverHttpEndpoint.GetUrlResult(uri);
+ Assert.IsTrue(strStatus.Equals("Byte array returned from HelloHttpHandler in CLR!!!\r\n"));
+
+ await ((HttpClientHelper)driverHttpEndpoint).TryUntilNoConnection(uri);
+
ValidateSuccessForLocalRuntime(2, testRuntimeFolder);
+
CleanUp(testRuntimeFolder);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/cs/Org.Apache.REEF.Tests/Functional/ReefFunctionalTest.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Tests/Functional/ReefFunctionalTest.cs b/lang/cs/Org.Apache.REEF.Tests/Functional/ReefFunctionalTest.cs
index e19f53f..0a8935d 100644
--- a/lang/cs/Org.Apache.REEF.Tests/Functional/ReefFunctionalTest.cs
+++ b/lang/cs/Org.Apache.REEF.Tests/Functional/ReefFunctionalTest.cs
@@ -51,6 +51,7 @@ namespace Org.Apache.REEF.Tests.Functional
private const string Local = "local";
private const string YARN = "yarn";
+ private const int SleepTime = 1000;
private readonly static Logger Logger = Logger.GetLogger(typeof(ReefFunctionalTest));
private const string StorageAccountKeyEnvironmentVariable = "REEFTestStorageAccountKey";
@@ -139,14 +140,35 @@ namespace Org.Apache.REEF.Tests.Functional
const string successIndication = "EXIT: ActiveContextClr2Java::Close";
const string failedTaskIndication = "Java_com_microsoft_reef_javabridge_NativeInterop_clrSystemFailedTaskHandlerOnNext";
const string failedEvaluatorIndication = "Java_com_microsoft_reef_javabridge_NativeInterop_clrSystemFailedEvaluatorHandlerOnNext";
- string[] lines = File.ReadAllLines(GetLogFile(_stdout, testFolder));
- Console.WriteLine("Lines read from log file : " + lines.Count());
- string[] successIndicators = lines.Where(s => s.Contains(successIndication)).ToArray();
- string[] failedTaskIndicators = lines.Where(s => s.Contains(failedTaskIndication)).ToArray();
- string[] failedIndicators = lines.Where(s => s.Contains(failedEvaluatorIndication)).ToArray();
- Assert.IsTrue(successIndicators.Count() == numberOfEvaluatorsToClose);
- Assert.IsFalse(failedTaskIndicators.Any());
- Assert.IsFalse(failedIndicators.Any());
+ string[] lines = null;
+ for (int i = 0; i < 60; i++)
+ {
+ try
+ {
+ lines = File.ReadAllLines(GetLogFile(_stdout, testFolder));
+ break;
+ }
+ catch (Exception)
+ {
+ Thread.Sleep(SleepTime);
+ }
+ }
+
+ if (lines != null)
+ {
+ Console.WriteLine("Lines read from log file : " + lines.Count());
+ string[] successIndicators = lines.Where(s => s.Contains(successIndication)).ToArray();
+ string[] failedTaskIndicators = lines.Where(s => s.Contains(failedTaskIndication)).ToArray();
+ string[] failedIndicators = lines.Where(s => s.Contains(failedEvaluatorIndication)).ToArray();
+ Assert.IsTrue(successIndicators.Count() == numberOfEvaluatorsToClose);
+ Assert.IsFalse(failedTaskIndicators.Any());
+ Assert.IsFalse(failedIndicators.Any());
+ }
+ else
+ {
+ Console.WriteLine("Cannot read from log file");
+ Assert.Fail();
+ }
}
protected void PeriodicUploadLog(object source, ElapsedEventArgs e)
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalClient.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalClient.java b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalClient.java
index 19a4055..7a08ee2 100644
--- a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalClient.java
+++ b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalClient.java
@@ -19,6 +19,7 @@
package org.apache.reef.bridge.client;
import org.apache.reef.client.parameters.DriverConfigurationProviders;
+import org.apache.reef.driver.parameters.JobSubmissionDirectory;
import org.apache.reef.runtime.common.driver.parameters.ClientRemoteIdentifier;
import org.apache.reef.runtime.common.files.REEFFileNames;
import org.apache.reef.runtime.local.client.DriverConfigurationProvider;
@@ -79,8 +80,11 @@ public final class LocalClient {
final Configuration providedConfigurations = configurationBuilder.build();
final Configuration driverConfiguration = Configurations.merge(
driverConfiguration1,
+ Tang.Factory.getTang()
+ .newConfigurationBuilder()
+ .bindNamedParameter(JobSubmissionDirectory.class, driverFolder.toString())
+ .build(),
providedConfigurations);
-
final File driverConfigurationFile = new File(driverFolder, fileNames.getDriverConfigurationPath());
configurationSerializer.toFile(driverConfiguration, driverConfigurationFile);
launcher.launch(driverFolder, localSubmissionFromCS.getJobId(), CLIENT_REMOTE_ID);
@@ -97,4 +101,4 @@ public final class LocalClient {
client.submit(localSubmissionFromCS);
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalSubmissionFromCS.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalSubmissionFromCS.java b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalSubmissionFromCS.java
index 58c5fea..ee37590 100644
--- a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalSubmissionFromCS.java
+++ b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/LocalSubmissionFromCS.java
@@ -20,6 +20,7 @@ package org.apache.reef.bridge.client;
import org.apache.commons.lang.Validate;
import org.apache.reef.client.parameters.DriverConfigurationProviders;
+import org.apache.reef.driver.parameters.JobSubmissionDirectory;
import org.apache.reef.io.TcpPortConfigurationProvider;
import org.apache.reef.runtime.common.files.REEFFileNames;
import org.apache.reef.runtime.common.launch.parameters.DriverLaunchCommandPrefix;
@@ -93,6 +94,7 @@ final class LocalSubmissionFromCS {
.bindNamedParameter(TcpPortRangeBegin.class, Integer.toString(tcpBeginPort))
.bindNamedParameter(TcpPortRangeCount.class, Integer.toString(tcpRangeCount))
.bindNamedParameter(TcpPortRangeTryCount.class, Integer.toString(tcpTryCount))
+ .bindNamedParameter(JobSubmissionDirectory.class, runtimeRootFolder.getAbsolutePath())
.bindList(DriverLaunchCommandPrefix.class, driverLaunchCommandPrefixList)
.build();
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/YarnJobSubmissionClient.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/YarnJobSubmissionClient.java b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/YarnJobSubmissionClient.java
index be5c24b..fa90602 100644
--- a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/YarnJobSubmissionClient.java
+++ b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/YarnJobSubmissionClient.java
@@ -18,6 +18,7 @@
*/
package org.apache.reef.bridge.client;
+import org.apache.hadoop.fs.*;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
@@ -43,11 +44,14 @@ import org.apache.reef.tang.annotations.Parameter;
import org.apache.reef.tang.exceptions.InjectionException;
import org.apache.reef.tang.formats.ConfigurationSerializer;
import org.apache.reef.util.JARFileMaker;
-
import javax.inject.Inject;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.FileWriter;
import java.util.List;
import java.util.Set;
import java.util.logging.Level;
@@ -173,7 +177,6 @@ public final class YarnJobSubmissionClient {
try (final YarnSubmissionHelper submissionHelper =
new YarnSubmissionHelper(yarnConfiguration, fileNames, classpath, tokenProvider, commandPrefixList)) {
-
// ------------------------------------------------------------------------
// Prepare the JAR
final JobFolder jobFolderOnDFS = this.uploader.createJobFolder(submissionHelper.getApplicationId());
@@ -183,7 +186,6 @@ public final class YarnJobSubmissionClient {
final File jarFile = makeJar(yarnSubmission.getDriverFolder());
LOG.log(Level.INFO, "Created job submission jar file: {0}", jarFile);
-
// ------------------------------------------------------------------------
// Upload the JAR
LOG.info("Uploading job submission JAR");
@@ -207,7 +209,65 @@ public final class YarnJobSubmissionClient {
} catch (InjectionException ie) {
throw new RuntimeException("Unable to submit job due to " + ie);
}
+ writeDriverHttpEndPoint(yarnSubmission.getDriverFolder(),
+ submissionHelper.getStringApplicationId(), jobFolderOnDFS.getPath());
+ }
+ }
+
+ /**
+ * We leave a file behind in job submission directory so that clr client can figure out
+ * the applicationId and yarn rest endpoint.
+ * @param driverFolder
+ * @param applicationId
+ * @throws IOException
+ */
+ private void writeDriverHttpEndPoint(final File driverFolder,
+ final String applicationId,
+ final Path dfsPath) throws IOException {
+ final FileSystem fs = FileSystem.get(yarnConfiguration);
+ final Path httpEndpointPath = new Path(dfsPath, fileNames.getDriverHttpEndpoint());
+
+ String trackingUri = null;
+ for (int i = 0; i < 60; i++) {
+ try {
+ LOG.log(Level.INFO, "Attempt " + i + " reading " + httpEndpointPath.toString());
+ if (fs.exists(httpEndpointPath)) {
+ FSDataInputStream input = fs.open(httpEndpointPath);
+ BufferedReader reader = new BufferedReader(new InputStreamReader(input, "UTF-8"));
+ trackingUri = reader.readLine();
+ reader.close();
+ break;
+ }
+ } catch (Exception ex) {
+ }
+ try{
+ Thread.sleep(1000);
+ } catch(InterruptedException ex2) {
+ break;
+ }
+ }
+
+ if (null == trackingUri) {
+ trackingUri = "";
+ LOG.log(Level.WARNING, "Failed reading " + httpEndpointPath.toString());
+ }
+
+ final File driverHttpEndpointFile = new File(driverFolder, fileNames.getDriverHttpEndpoint());
+ BufferedWriter out = new BufferedWriter(new FileWriter(driverHttpEndpointFile));
+ out.write(applicationId + "\n");
+ out.write(trackingUri + "\n");
+ String addr = yarnConfiguration.get("yarn.resourcemanager.webapp.address");
+ if (null == addr || addr.startsWith("0.0.0.0")) {
+ String str2 = yarnConfiguration.get("yarn.resourcemanager.ha.rm-ids");
+ if (null != str2) {
+ for (String rm : str2.split(",")) {
+ out.write(yarnConfiguration.get("yarn.resourcemanager.webapp.address." + rm) +"\n");
+ }
+ }
+ } else {
+ out.write(addr +"\n");
}
+ out.close();
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-bridge-java/src/main/java/org/apache/reef/javabridge/generic/JobDriver.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-bridge-java/src/main/java/org/apache/reef/javabridge/generic/JobDriver.java b/lang/java/reef-bridge-java/src/main/java/org/apache/reef/javabridge/generic/JobDriver.java
index 054408e..f619f5f 100644
--- a/lang/java/reef-bridge-java/src/main/java/org/apache/reef/javabridge/generic/JobDriver.java
+++ b/lang/java/reef-bridge-java/src/main/java/org/apache/reef/javabridge/generic/JobDriver.java
@@ -31,6 +31,7 @@ import org.apache.reef.javabridge.*;
import org.apache.reef.driver.restart.DriverRestartCompleted;
import org.apache.reef.runtime.common.driver.DriverStatusManager;
import org.apache.reef.driver.evaluator.EvaluatorProcess;
+import org.apache.reef.runtime.common.files.REEFFileNames;
import org.apache.reef.tang.annotations.Unit;
import org.apache.reef.util.Optional;
import org.apache.reef.util.logging.CLRBufferedLogHandler;
@@ -48,6 +49,9 @@ import org.apache.reef.webserver.*;
import javax.inject.Inject;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletResponse;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
@@ -110,6 +114,8 @@ public final class JobDriver {
*/
private final Map<String, ActiveContext> contexts = new HashMap<>();
+ private final REEFFileNames reefFileNames;
+ private final LocalAddressProvider localAddressProvider;
/**
* Logging scope factory that provides LoggingScope.
*/
@@ -140,6 +146,7 @@ public final class JobDriver {
new HashMap<String, AllocatedEvaluatorBridge>();
private EvaluatorRequestorBridge evaluatorRequestorBridge;
+
/**
* Job driver constructor.
* All parameters are injected from TANG automatically.
@@ -159,6 +166,7 @@ public final class JobDriver {
final LoggingScopeFactory loggingScopeFactory,
final LocalAddressProvider localAddressProvider,
final ActiveContextBridgeFactory activeContextBridgeFactory,
+ final REEFFileNames reefFileNames,
final AllocatedEvaluatorBridgeFactory allocatedEvaluatorBridgeFactory,
final CLRProcessFactory clrProcessFactory) {
this.clock = clock;
@@ -171,6 +179,8 @@ public final class JobDriver {
this.allocatedEvaluatorBridgeFactory = allocatedEvaluatorBridgeFactory;
this.nameServerInfo = localAddressProvider.getLocalAddress() + ":" + this.nameServer.getPort();
this.loggingScopeFactory = loggingScopeFactory;
+ this.reefFileNames = reefFileNames;
+ this.localAddressProvider = localAddressProvider;
this.clrProcessFactory = clrProcessFactory;
}
@@ -188,6 +198,17 @@ public final class JobDriver {
}
final String portNumber = httpServer == null ? null : Integer.toString((httpServer.getPort()));
+ if (portNumber != null){
+ try {
+ final File outputFileName = new File(reefFileNames.getDriverHttpEndpoint());
+ BufferedWriter out = new BufferedWriter(new FileWriter(outputFileName));
+ out.write(localAddressProvider.getLocalAddress() + ":" + portNumber + "\n");
+ out.close();
+ } catch (IOException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
this.evaluatorRequestorBridge =
new EvaluatorRequestorBridge(JobDriver.this.evaluatorRequestor, false, loggingScopeFactory);
final long[] handlers = initializer.getClrHandlers(portNumber, evaluatorRequestorBridge);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/files/REEFFileNames.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/files/REEFFileNames.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/files/REEFFileNames.java
index 68a6fb0..2c93e22 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/files/REEFFileNames.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/files/REEFFileNames.java
@@ -47,9 +47,9 @@ public final class REEFFileNames {
private static final String DRIVER_STDOUT = "driver.stdout";
private static final String EVALUATOR_STDERR = "evaluator.stderr";
private static final String EVALUATOR_STDOUT = "evaluator.stdout";
+ private static final String DRIVER_HTTP_ENDPOINT_FILE_NAME = "DriverHttpEndpoint.txt";
private static final String BRIDGE_EXE_NAME = "Org.Apache.REEF.Bridge.exe";
-
@Inject
public REEFFileNames() {
}
@@ -208,4 +208,12 @@ public final class REEFFileNames {
public String getEvaluatorStdoutFileName() {
return EVALUATOR_STDOUT;
}
+
+ /**
+ * @return File name that contains the dfs path for the DriverHttpEndpoint.
+ */
+ public String getDriverHttpEndpoint() {
+ return DRIVER_HTTP_ENDPOINT_FILE_NAME;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-runtime-hdinsight/src/main/java/org/apache/reef/runtime/hdinsight/client/HDInsightDriverConfiguration.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-runtime-hdinsight/src/main/java/org/apache/reef/runtime/hdinsight/client/HDInsightDriverConfiguration.java b/lang/java/reef-runtime-hdinsight/src/main/java/org/apache/reef/runtime/hdinsight/client/HDInsightDriverConfiguration.java
index 1a79505..72ec88b 100644
--- a/lang/java/reef-runtime-hdinsight/src/main/java/org/apache/reef/runtime/hdinsight/client/HDInsightDriverConfiguration.java
+++ b/lang/java/reef-runtime-hdinsight/src/main/java/org/apache/reef/runtime/hdinsight/client/HDInsightDriverConfiguration.java
@@ -35,7 +35,7 @@ import org.apache.reef.runtime.common.parameters.JVMHeapSlack;
import org.apache.reef.runtime.hdinsight.HDInsightClasspathProvider;
import org.apache.reef.runtime.hdinsight.HDInsightJVMPathProvider;
import org.apache.reef.runtime.yarn.driver.*;
-import org.apache.reef.runtime.yarn.driver.parameters.JobSubmissionDirectory;
+import org.apache.reef.driver.parameters.JobSubmissionDirectory;
import org.apache.reef.runtime.yarn.driver.parameters.YarnHeartbeatPeriod;
import org.apache.reef.runtime.yarn.util.YarnConfigurationConstructor;
import org.apache.reef.tang.formats.ConfigurationModule;
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/client/YarnSubmissionHelper.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/client/YarnSubmissionHelper.java b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/client/YarnSubmissionHelper.java
index 8723ac3..e21d81b 100644
--- a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/client/YarnSubmissionHelper.java
+++ b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/client/YarnSubmissionHelper.java
@@ -102,6 +102,14 @@ public final class YarnSubmissionHelper implements Closeable{
}
/**
+ *
+ * @return the application ID string representation assigned by YARN.
+ */
+ public String getStringApplicationId() {
+ return this.applicationId.toString();
+ }
+
+ /**
* Set the name of the application to be submitted.
* @param applicationName
* @return
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/UploaderToJobfolder.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/UploaderToJobfolder.java b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/UploaderToJobfolder.java
index c6d8291..b8467d5 100644
--- a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/UploaderToJobfolder.java
+++ b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/UploaderToJobfolder.java
@@ -28,7 +28,7 @@ import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
-import org.apache.reef.runtime.yarn.driver.parameters.JobSubmissionDirectory;
+import org.apache.reef.driver.parameters.JobSubmissionDirectory;
import org.apache.reef.tang.annotations.Parameter;
import javax.inject.Inject;
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java
index e088649..028744f 100644
--- a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java
+++ b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnContainerManager.java
@@ -20,6 +20,7 @@ package org.apache.reef.runtime.yarn.driver;
import com.google.protobuf.ByteString;
import org.apache.commons.collections.ListUtils;
+import org.apache.hadoop.fs.*;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.AMRMClient;
@@ -29,6 +30,7 @@ import org.apache.hadoop.yarn.client.api.async.NMClientAsync;
import org.apache.hadoop.yarn.client.api.async.impl.NMClientAsyncImpl;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.reef.driver.parameters.JobSubmissionDirectory;
import org.apache.reef.exception.DriverFatalRuntimeException;
import org.apache.reef.proto.ReefServiceProtos;
import org.apache.reef.runtime.common.driver.DriverStatusManager;
@@ -36,6 +38,7 @@ import org.apache.reef.runtime.common.driver.resourcemanager.NodeDescriptorEvent
import org.apache.reef.runtime.common.driver.resourcemanager.ResourceEventImpl;
import org.apache.reef.runtime.common.driver.resourcemanager.ResourceStatusEventImpl;
import org.apache.reef.runtime.common.driver.resourcemanager.RuntimeStatusEventImpl;
+import org.apache.reef.runtime.common.files.REEFFileNames;
import org.apache.reef.runtime.yarn.driver.parameters.YarnHeartbeatPeriod;
import org.apache.reef.tang.annotations.Parameter;
import org.apache.reef.util.Optional;
@@ -75,6 +78,8 @@ final class YarnContainerManager
private final ContainerRequestCounter containerRequestCounter;
private final DriverStatusManager driverStatusManager;
private final TrackingURLProvider trackingURLProvider;
+ private final String jobSubmissionDirectory;
+ private final REEFFileNames reefFileNames;
private final RackNameFormatter rackNameFormatter;
@Inject
@@ -86,9 +91,10 @@ final class YarnContainerManager
final ApplicationMasterRegistration registration,
final ContainerRequestCounter containerRequestCounter,
final DriverStatusManager driverStatusManager,
+ final REEFFileNames reefFileNames,
+ @Parameter(JobSubmissionDirectory.class) final String jobSubmissionDirectory,
final TrackingURLProvider trackingURLProvider,
final RackNameFormatter rackNameFormatter) throws IOException {
-
this.reefEventHandlers = reefEventHandlers;
this.driverStatusManager = driverStatusManager;
@@ -104,6 +110,8 @@ final class YarnContainerManager
this.resourceManager = AMRMClientAsync.createAMRMClientAsync(yarnRMHeartbeatPeriod, this);
this.nodeManager = new NMClientAsyncImpl(this);
+ this.jobSubmissionDirectory = jobSubmissionDirectory;
+ this.reefFileNames = reefFileNames;
LOG.log(Level.FINEST, "Instantiated YarnContainerManager");
}
@@ -250,7 +258,12 @@ final class YarnContainerManager
this.registration.setRegistration(this.resourceManager.registerApplicationMaster(
"", 0, this.trackingURLProvider.getTrackingUrl()));
LOG.log(Level.FINE, "YARN registration: {0}", registration);
-
+ final FileSystem fs = FileSystem.get(this.yarnConf);
+ final Path outputFileName = new Path(this.jobSubmissionDirectory, this.reefFileNames.getDriverHttpEndpoint());
+ final FSDataOutputStream out = fs.create(outputFileName);
+ out.writeBytes(this.trackingURLProvider.getTrackingUrl() + "\n");
+ out.flush();
+ out.close();
} catch (final YarnException | IOException e) {
LOG.log(Level.WARNING, "Unable to register application master.", e);
onRuntimeError(e);
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnDriverConfiguration.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnDriverConfiguration.java b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnDriverConfiguration.java
index e3cade9..a7de1d3 100644
--- a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnDriverConfiguration.java
+++ b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/YarnDriverConfiguration.java
@@ -30,7 +30,7 @@ import org.apache.reef.runtime.common.launch.parameters.ErrorHandlerRID;
import org.apache.reef.runtime.common.launch.parameters.LaunchID;
import org.apache.reef.runtime.common.parameters.JVMHeapSlack;
import org.apache.reef.runtime.yarn.YarnClasspathProvider;
-import org.apache.reef.runtime.yarn.driver.parameters.JobSubmissionDirectory;
+import org.apache.reef.driver.parameters.JobSubmissionDirectory;
import org.apache.reef.runtime.yarn.driver.parameters.YarnHeartbeatPeriod;
import org.apache.reef.runtime.yarn.util.YarnConfigurationConstructor;
import org.apache.reef.tang.formats.*;
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/189c5acb/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/parameters/JobSubmissionDirectory.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/parameters/JobSubmissionDirectory.java b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/parameters/JobSubmissionDirectory.java
index 9c36503..6213bf9 100644
--- a/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/parameters/JobSubmissionDirectory.java
+++ b/lang/java/reef-runtime-yarn/src/main/java/org/apache/reef/runtime/yarn/driver/parameters/JobSubmissionDirectory.java
@@ -16,10 +16,11 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.reef.runtime.yarn.driver.parameters;
-import org.apache.reef.tang.annotations.Name;
+package org.apache.reef.driver.parameters;
+
import org.apache.reef.tang.annotations.NamedParameter;
+import org.apache.reef.tang.annotations.Name;
/**
* The job submission directory.