You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by af...@apache.org on 2015/11/13 03:26:46 UTC
incubator-reef git commit: [REEF-902] Create .NET job resource
uploader
Repository: incubator-reef
Updated Branches:
refs/heads/master 748421aa1 -> 0a78fcac5
[REEF-902] Create .NET job resource uploader
This addressed the issue by:
* Creating interface IJobResourceUploader
* Adding a legacy implementation by calling into Java
* Add test for the the legacy implementation
JIRA:
[REEF-902](https://issues.apache.org/jira/browse/REEF-902)
Pull Request:
Closes #627
Project: http://git-wip-us.apache.org/repos/asf/incubator-reef/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-reef/commit/0a78fcac
Tree: http://git-wip-us.apache.org/repos/asf/incubator-reef/tree/0a78fcac
Diff: http://git-wip-us.apache.org/repos/asf/incubator-reef/diff/0a78fcac
Branch: refs/heads/master
Commit: 0a78fcac528d3caec8a64cf2147641eed15b8883
Parents: 748421a
Author: Anupam <an...@gmail.com>
Authored: Tue Nov 10 16:43:36 2015 -0800
Committer: Andrew Chung <af...@gmail.com>
Committed: Thu Nov 12 18:25:46 2015 -0800
----------------------------------------------------------------------
.../LegacyJobResourceUploaderTests.cs | 131 +++++++++++++++++++
.../Org.Apache.REEF.Client.Tests.csproj | 1 +
.../Org.Apache.REEF.Client/Common/DotNetFile.cs | 50 +++++++
lang/cs/Org.Apache.REEF.Client/Common/IFile.cs | 35 +++++
.../Common/IJavaClientLauncher.cs | 37 ++++++
.../Common/IResourceArchiveFileGenerator.cs | 32 +++++
.../Common/JavaClientLauncher.cs | 11 +-
.../Common/ResourceArchiveFileGenerator.cs | 60 +++++++++
.../Org.Apache.REEF.Client/Local/LocalClient.cs | 6 +-
.../Org.Apache.REEF.Client.csproj | 12 +-
.../YARN/IJobResourceUploader.cs | 32 +++++
.../YARN/IJobSubmissionDirectoryProvider.cs | 34 +++++
.../Org.Apache.REEF.Client/YARN/JobResource.cs | 27 ++++
.../YARN/JobSubmissionDirectoryProvider.cs | 48 +++++++
.../YARN/LegacyJobResourceUploader.cs | 100 ++++++++++++++
.../YARN/YARNREEFClient.cs | 4 +-
.../reef/bridge/client/JobResourceUploader.java | 84 ++++++++++++
17 files changed, 695 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client.Tests/LegacyJobResourceUploaderTests.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client.Tests/LegacyJobResourceUploaderTests.cs b/lang/cs/Org.Apache.REEF.Client.Tests/LegacyJobResourceUploaderTests.cs
new file mode 100644
index 0000000..e76ae66
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client.Tests/LegacyJobResourceUploaderTests.cs
@@ -0,0 +1,131 @@
+// 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;
+using System.IO;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+using NSubstitute;
+using Org.Apache.REEF.Client.Common;
+using Org.Apache.REEF.Client.Yarn;
+using Org.Apache.REEF.Tang.Implementations.Tang;
+using Org.Apache.REEF.Tang.Util;
+
+namespace Org.Apache.REEF.Client.Tests
+{
+ [TestClass]
+ public class LegacyJobResourceUploaderTests
+ {
+ private const string AnyDriverLocalFolderPath = @"Any\Local\Folder\Path";
+ private const string AnyDriverResourceUploadPath = "/vol1/tmp";
+ private const string AnyUploadedResourcePath = "hdfs://foo/vol1/tmp/driver.zip";
+ private const long AnyModificationTime = 1446161745550;
+ private const long AnyResourceSize = 53092;
+
+ [TestMethod]
+ public void LegacyJobResourceUploaderCanInstantiateWithDefaultBindings()
+ {
+ TangFactory.GetTang().NewInjector().GetInstance<LegacyJobResourceUploader>();
+ }
+
+ [TestMethod]
+ public void UploadJobResourceCreatesResourceArchive()
+ {
+ var testContext = new TestContext();
+ var jobResourceUploader = testContext.GetJobResourceUploader();
+
+ jobResourceUploader.UploadJobResource(AnyDriverLocalFolderPath);
+
+ // Archive file generator recieved exactly one call with correct driver local folder path with trailing \
+ testContext.ResourceArchiveFileGenerator.Received(1).CreateArchiveToUpload(AnyDriverLocalFolderPath + @"\");
+ }
+
+ [TestMethod]
+ public void UploadJobResourceJavaLauncherCalledWithCorrectArguments()
+ {
+ var testContext = new TestContext();
+ var jobResourceUploader = testContext.GetJobResourceUploader();
+ const string anyLocalArchivePath = @"Any\Local\Archive\Path.zip";
+ testContext.ResourceArchiveFileGenerator.CreateArchiveToUpload(AnyDriverLocalFolderPath + @"\")
+ .Returns(anyLocalArchivePath);
+ testContext.JobSubmissionDirectoryProvider.GetJobSubmissionDirectory().Returns(AnyDriverResourceUploadPath);
+ jobResourceUploader.UploadJobResource(AnyDriverLocalFolderPath);
+
+ const string javaClassNameForResourceUploader = @"org.apache.reef.bridge.client.JobResourceUploader";
+ Guid notUsed;
+
+ // Clientlauncher is called with correct class name, local archive path, upload path and temp file.
+ testContext.JavaClientLauncher.Received()
+ .Launch(javaClassNameForResourceUploader,
+ anyLocalArchivePath,
+ AnyDriverResourceUploadPath + "/",
+ Arg.Is<string>(
+ outputFilePath =>
+ Path.GetDirectoryName(outputFilePath) + @"\" == Path.GetTempPath()
+ && Guid.TryParse(Path.GetFileName(outputFilePath), out notUsed)));
+ }
+
+ [TestMethod]
+ [ExpectedException(typeof(FileNotFoundException))]
+ public void UploadJobResourceNoFileCreatedByJavaCallThrowsException()
+ {
+ var testContext = new TestContext();
+ var jobResourceUploader = testContext.GetJobResourceUploader(fileExistsReturnValue: false);
+
+ // throws filenotfound exception
+ jobResourceUploader.UploadJobResource(AnyDriverLocalFolderPath);
+ }
+
+ [TestMethod]
+ public void UploadJobResourceReturnsJobResourceDetails()
+ {
+ var testContext = new TestContext();
+ var jobResourceUploader = testContext.GetJobResourceUploader();
+
+ var jobResource = jobResourceUploader.UploadJobResource(AnyDriverLocalFolderPath);
+
+ Assert.AreEqual(AnyModificationTime, jobResource.LastModificationUnixTimestamp);
+ Assert.AreEqual(AnyResourceSize, jobResource.ResourceSize);
+ Assert.AreEqual(AnyUploadedResourcePath, jobResource.RemoteUploadPath);
+ }
+
+ private class TestContext
+ {
+ public readonly IJavaClientLauncher JavaClientLauncher = Substitute.For<IJavaClientLauncher>();
+ public readonly IJobSubmissionDirectoryProvider JobSubmissionDirectoryProvider =
+ Substitute.For<IJobSubmissionDirectoryProvider>();
+ public readonly IResourceArchiveFileGenerator ResourceArchiveFileGenerator =
+ Substitute.For<IResourceArchiveFileGenerator>();
+ public readonly IFile File = Substitute.For<IFile>();
+
+ public LegacyJobResourceUploader GetJobResourceUploader(bool fileExistsReturnValue = true,
+ string uploadedResourcePath = AnyUploadedResourcePath,
+ long modificationTime = AnyModificationTime,
+ long resourceSize = AnyResourceSize)
+ {
+ var injector = TangFactory.GetTang().NewInjector();
+ File.Exists(Arg.Any<string>()).Returns(fileExistsReturnValue);
+ File.ReadAllText(Arg.Any<string>())
+ .Returns(string.Format("{0};{1};{2}", uploadedResourcePath, modificationTime, resourceSize));
+ injector.BindVolatileInstance(GenericType<IJavaClientLauncher>.Class, JavaClientLauncher);
+ injector.BindVolatileInstance(GenericType<IJobSubmissionDirectoryProvider>.Class, JobSubmissionDirectoryProvider);
+ injector.BindVolatileInstance(GenericType<IResourceArchiveFileGenerator>.Class, ResourceArchiveFileGenerator);
+ injector.BindVolatileInstance(GenericType<IFile>.Class, File);
+ return injector.GetInstance<LegacyJobResourceUploader>();
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client.Tests/Org.Apache.REEF.Client.Tests.csproj
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client.Tests/Org.Apache.REEF.Client.Tests.csproj b/lang/cs/Org.Apache.REEF.Client.Tests/Org.Apache.REEF.Client.Tests.csproj
index 261021c..e7af026 100644
--- a/lang/cs/Org.Apache.REEF.Client.Tests/Org.Apache.REEF.Client.Tests.csproj
+++ b/lang/cs/Org.Apache.REEF.Client.Tests/Org.Apache.REEF.Client.Tests.csproj
@@ -47,6 +47,7 @@ under the License.
<Reference Include="System.ServiceProcess" />
</ItemGroup>
<ItemGroup>
+ <Compile Include="LegacyJobResourceUploaderTests.cs" />
<Compile Include="MultipleRMUrlProviderTests.cs" />
<Compile Include="WindowsHadoopEmulatorYarnClientTests.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/Common/DotNetFile.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Common/DotNetFile.cs b/lang/cs/Org.Apache.REEF.Client/Common/DotNetFile.cs
new file mode 100644
index 0000000..d040dc5
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/Common/DotNetFile.cs
@@ -0,0 +1,50 @@
+// 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.IO;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.Common
+{
+ /// <summary>
+ /// Pass through implementation of IFile calling into
+ /// System.IO.File.
+ /// Avoid adding any extra logic in this class.
+ /// </summary>
+ internal sealed class DotNetFile : IFile
+ {
+ [Inject]
+ private DotNetFile()
+ {
+ }
+
+ public bool Exists(string path)
+ {
+ return File.Exists(path);
+ }
+
+ public string ReadAllText(string path)
+ {
+ return File.ReadAllText(path);
+ }
+
+ public void Delete(string path)
+ {
+ File.Delete(path);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/Common/IFile.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Common/IFile.cs b/lang/cs/Org.Apache.REEF.Client/Common/IFile.cs
new file mode 100644
index 0000000..f7b9019
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/Common/IFile.cs
@@ -0,0 +1,35 @@
+// 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.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.Common
+{
+ /// <summary>
+ /// Wrapper interface around static implementation of System.IO.File
+ /// to allow unittestability without using actual files
+ /// </summary>
+ [DefaultImplementation(typeof(DotNetFile))]
+ internal interface IFile
+ {
+ bool Exists(string path);
+
+ string ReadAllText(string path);
+
+ void Delete(string path);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/Common/IJavaClientLauncher.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Common/IJavaClientLauncher.cs b/lang/cs/Org.Apache.REEF.Client/Common/IJavaClientLauncher.cs
new file mode 100644
index 0000000..293813a
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/Common/IJavaClientLauncher.cs
@@ -0,0 +1,37 @@
+// 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 Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.Common
+{
+ /// <summary>
+ /// Helper interface to launch the java side of the various clients.
+ /// </summary>
+ [DefaultImplementation(typeof(JavaClientLauncher))]
+ internal interface IJavaClientLauncher
+ {
+ void Launch(string javaClassName, params string[] parameters);
+
+ /// <summary>
+ /// Add entries to the end of the classpath of the java client.
+ /// </summary>
+ /// <param name="entries"></param>
+ void AddToClassPath(IEnumerable<string> entries);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/Common/IResourceArchiveFileGenerator.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Common/IResourceArchiveFileGenerator.cs b/lang/cs/Org.Apache.REEF.Client/Common/IResourceArchiveFileGenerator.cs
new file mode 100644
index 0000000..847884c
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/Common/IResourceArchiveFileGenerator.cs
@@ -0,0 +1,32 @@
+// 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.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.Common
+{
+ [DefaultImplementation(typeof(ResourceArchiveFileGenerator))]
+ internal interface IResourceArchiveFileGenerator
+ {
+ /// <summary>
+ /// Creates an archive file from the assembled folder path.
+ /// </summary>
+ /// <param name="folderPath"></param>
+ /// <returns>Path to the archive file.</returns>
+ string CreateArchiveToUpload(string folderPath);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/Common/JavaClientLauncher.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Common/JavaClientLauncher.cs b/lang/cs/Org.Apache.REEF.Client/Common/JavaClientLauncher.cs
index d12646e..25b6fe7 100644
--- a/lang/cs/Org.Apache.REEF.Client/Common/JavaClientLauncher.cs
+++ b/lang/cs/Org.Apache.REEF.Client/Common/JavaClientLauncher.cs
@@ -33,7 +33,7 @@ namespace Org.Apache.REEF.Client.Common
/// <summary>
/// Helper class to launch the java side of the various clients.
/// </summary>
- internal class JavaClientLauncher
+ internal class JavaClientLauncher : IJavaClientLauncher
{
/// <summary>
/// The folder in which we search for the client jar.
@@ -48,7 +48,12 @@ namespace Org.Apache.REEF.Client.Common
{
}
- internal void Launch(string javaClassName, params string[] parameters)
+ /// <summary>
+ /// Launch a java class in ClientConstants.ClientJarFilePrefix with provided parameters.
+ /// </summary>
+ /// <param name="javaClassName"></param>
+ /// <param name="parameters"></param>
+ public void Launch(string javaClassName, params string[] parameters)
{
var startInfo = new ProcessStartInfo
{
@@ -174,7 +179,7 @@ namespace Org.Apache.REEF.Client.Common
/// Add entries to the end of the classpath of the java client.
/// </summary>
/// <param name="entries"></param>
- internal void AddToClassPath(IEnumerable<string> entries)
+ public void AddToClassPath(IEnumerable<string> entries)
{
foreach (var entry in entries)
{
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/Common/ResourceArchiveFileGenerator.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/Common/ResourceArchiveFileGenerator.cs b/lang/cs/Org.Apache.REEF.Client/Common/ResourceArchiveFileGenerator.cs
new file mode 100644
index 0000000..e343be3
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/Common/ResourceArchiveFileGenerator.cs
@@ -0,0 +1,60 @@
+// 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;
+using System.IO;
+using System.IO.Compression;
+using Org.Apache.REEF.Common.Files;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Diagnostics;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Client.Common
+{
+ /// <summary>
+ /// Creates and returns zip file for assembled driver folder.
+ /// </summary>
+ internal sealed class ResourceArchiveFileGenerator : IResourceArchiveFileGenerator
+ {
+ private static readonly Logger Log = Logger.GetLogger(typeof(ResourceArchiveFileGenerator));
+ private readonly REEFFileNames _reefFileNames;
+
+ [Inject]
+ private ResourceArchiveFileGenerator (REEFFileNames reefFileNames)
+ {
+ _reefFileNames = reefFileNames;
+ }
+
+ public string CreateArchiveToUpload(string folderPath)
+ {
+ string archivePath = Path.Combine(folderPath, Path.GetDirectoryName(folderPath) + ".zip");
+ string reefFolder = Path.Combine(folderPath, _reefFileNames.GetReefFolderName());
+ if (!Directory.Exists(reefFolder))
+ {
+ Exceptions.Throw(new DirectoryNotFoundException("Cannot find directory " + reefFolder), Log);
+ }
+
+ if (File.Exists(archivePath))
+ {
+ Exceptions.Throw(new InvalidOperationException("Archive file already exists " + archivePath), Log);
+ }
+
+ ZipFile.CreateFromDirectory(reefFolder, archivePath);
+ return archivePath;
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/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 a1e8f91..b11bf87 100644
--- a/lang/cs/Org.Apache.REEF.Client/Local/LocalClient.cs
+++ b/lang/cs/Org.Apache.REEF.Client/Local/LocalClient.cs
@@ -55,7 +55,7 @@ namespace Org.Apache.REEF.Client.Local
private static readonly Logger Logger = Logger.GetLogger(typeof(LocalClient));
private readonly DriverFolderPreparationHelper _driverFolderPreparationHelper;
- private readonly JavaClientLauncher _javaClientLauncher;
+ private readonly IJavaClientLauncher _javaClientLauncher;
private readonly int _maxNumberOfConcurrentEvaluators;
private readonly string _runtimeFolder;
private REEFFileNames _fileNames;
@@ -64,7 +64,7 @@ namespace Org.Apache.REEF.Client.Local
private LocalClient(DriverFolderPreparationHelper driverFolderPreparationHelper,
[Parameter(typeof(LocalRuntimeDirectory))] string runtimeFolder,
[Parameter(typeof(NumberOfEvaluators))] int maxNumberOfConcurrentEvaluators,
- JavaClientLauncher javaClientLauncher,
+ IJavaClientLauncher javaClientLauncher,
REEFFileNames fileNames)
{
_driverFolderPreparationHelper = driverFolderPreparationHelper;
@@ -85,7 +85,7 @@ namespace Org.Apache.REEF.Client.Local
private LocalClient(
DriverFolderPreparationHelper driverFolderPreparationHelper,
[Parameter(typeof(NumberOfEvaluators))] int numberOfEvaluators,
- JavaClientLauncher javaClientLauncher,
+ IJavaClientLauncher javaClientLauncher,
REEFFileNames fileNames)
: this(driverFolderPreparationHelper, Path.GetTempPath(), numberOfEvaluators, javaClientLauncher, fileNames)
{
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/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 9b41ea5..117017a 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
@@ -71,12 +71,17 @@ under the License.
<Compile Include="Avro\Local\AvroLocalJobSubmissionParameters.cs" />
<Compile Include="Avro\YARN\AvroYarnJobSubmissionParameters.cs" />
<Compile Include="Avro\YARN\AvroYarnClusterJobSubmissionParameters.cs" />
+ <Compile Include="Common\DotNetFile.cs" />
<Compile Include="Common\ClientConstants.cs" />
<Compile Include="Common\DriverFolderPreparationHelper.cs" />
+ <Compile Include="Common\IFile.cs" />
+ <Compile Include="Common\IResourceArchiveFileGenerator.cs" />
<Compile Include="Common\FileSets.cs" />
- <Compile Include="Common\JobSubmissionResult.cs" />
+ <Compile Include="Common\IJavaClientLauncher.cs" />
<Compile Include="Common\IJobSubmissionResult.cs" />
+ <Compile Include="Common\JobSubmissionResult.cs" />
<Compile Include="Common\JavaClientLauncher.cs" />
+ <Compile Include="Common\ResourceArchiveFileGenerator.cs" />
<Compile Include="Common\ResourceHelper.cs" />
<Compile Include="Local\LocalClient.cs" />
<Compile Include="Local\LocalJobSubmissionResult.cs" />
@@ -84,6 +89,11 @@ under the License.
<Compile Include="Local\Parameters\LocalRuntimeDirectory.cs" />
<Compile Include="Local\Parameters\NumberOfEvaluators.cs" />
<Compile Include="Properties\AssemblyInfo.cs" />
+ <Compile Include="YARN\IJobResourceUploader.cs" />
+ <Compile Include="YARN\IJobSubmissionDirectoryProvider.cs" />
+ <Compile Include="YARN\JobResource.cs" />
+ <Compile Include="YARN\JobSubmissionDirectoryProvider.cs" />
+ <Compile Include="YARN\LegacyJobResourceUploader.cs" />
<Compile Include="YARN\Parameters\JobSubmissionDirectoryPrefixParameter.cs" />
<Compile Include="YARN\Parameters\SecurityTokenParameters.cs" />
<Compile Include="YARN\RestClient\DataModel\Acls.cs" />
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/YARN/IJobResourceUploader.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/YARN/IJobResourceUploader.cs b/lang/cs/Org.Apache.REEF.Client/YARN/IJobResourceUploader.cs
new file mode 100644
index 0000000..2366f01
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/YARN/IJobResourceUploader.cs
@@ -0,0 +1,32 @@
+// 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.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.Yarn
+{
+ [DefaultImplementation(typeof(LegacyJobResourceUploader))]
+ public interface IJobResourceUploader
+ {
+ /// <summary>
+ /// Upload archived local driver folder to DFS destination path.
+ /// </summary>
+ /// <param name="driverLocalFolderPath">Local folder where REEF application resources are staged</param>
+ /// <returns>Path, modification time and size of uploaded file as JobResource</returns>
+ JobResource UploadJobResource(string driverLocalFolderPath);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/YARN/IJobSubmissionDirectoryProvider.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/YARN/IJobSubmissionDirectoryProvider.cs b/lang/cs/Org.Apache.REEF.Client/YARN/IJobSubmissionDirectoryProvider.cs
new file mode 100644
index 0000000..b702e79
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/YARN/IJobSubmissionDirectoryProvider.cs
@@ -0,0 +1,34 @@
+// 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.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.Yarn
+{
+ /// <summary>
+ /// Provides path to job submission directory.
+ /// </summary>
+ [DefaultImplementation(typeof(JobSubmissionDirectoryProvider))]
+ public interface IJobSubmissionDirectoryProvider
+ {
+ /// <summary>
+ /// Returns path to job submission directory.
+ /// </summary>
+ /// <returns></returns>
+ string GetJobSubmissionDirectory();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/YARN/JobResource.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/YARN/JobResource.cs b/lang/cs/Org.Apache.REEF.Client/YARN/JobResource.cs
new file mode 100644
index 0000000..fa10b64
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/YARN/JobResource.cs
@@ -0,0 +1,27 @@
+// 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.
+namespace Org.Apache.REEF.Client.Yarn
+{
+ public class JobResource
+ {
+ public string RemoteUploadPath { get; set; }
+
+ public long LastModificationUnixTimestamp { get; set; }
+
+ public long ResourceSize { get; set; }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/YARN/JobSubmissionDirectoryProvider.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/YARN/JobSubmissionDirectoryProvider.cs b/lang/cs/Org.Apache.REEF.Client/YARN/JobSubmissionDirectoryProvider.cs
new file mode 100644
index 0000000..4a5c59a
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/YARN/JobSubmissionDirectoryProvider.cs
@@ -0,0 +1,48 @@
+// 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;
+using Org.Apache.REEF.Client.YARN.Parameters;
+using Org.Apache.REEF.Common.Files;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.Client.Yarn
+{
+ public class JobSubmissionDirectoryProvider : IJobSubmissionDirectoryProvider
+ {
+ private readonly string _jobSubmissionDirectoryPrefix;
+ private readonly REEFFileNames _filenames;
+
+ [Inject]
+ private JobSubmissionDirectoryProvider(
+ [Parameter(typeof(JobSubmissionDirectoryPrefixParameter))]
+ string jobSubmissionDirectoryPrefix,
+ REEFFileNames filenames)
+ {
+ _filenames = filenames;
+ _jobSubmissionDirectoryPrefix = jobSubmissionDirectoryPrefix;
+ }
+
+ public string GetJobSubmissionDirectory()
+ {
+ return string.Format(@"{0}/{1}{2}/",
+ _jobSubmissionDirectoryPrefix,
+ DateTime.Now.ToString("yyyy_MM_dd_HH_mm_ss"),
+ _filenames.GetJobFolderPrefix());
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/YARN/LegacyJobResourceUploader.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/YARN/LegacyJobResourceUploader.cs b/lang/cs/Org.Apache.REEF.Client/YARN/LegacyJobResourceUploader.cs
new file mode 100644
index 0000000..22e32a1
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.Client/YARN/LegacyJobResourceUploader.cs
@@ -0,0 +1,100 @@
+// 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;
+using System.IO;
+using Org.Apache.REEF.Client.Common;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Utilities.Diagnostics;
+using Org.Apache.REEF.Utilities.Logging;
+
+namespace Org.Apache.REEF.Client.Yarn
+{
+ /// <summary>
+ /// Uploads assembled driver folder to DFS as an archive file.
+ /// Uploading of DFS is done via calling into org.apache.reef.bridge.client.JobResourceUploader
+ /// Uploading via Java is necessary to support Hadoop versions before v2.7.0 because
+ /// we need to find modification time with accuracy of seconds to use archive as resource for YARNRM
+ /// <see cref="Org.Apache.REEF.IO.FileSystem.Hadoop.HadoopFileSystem"/> implementation uses HDFS
+ /// commandline shell which does not provide modification time with accuracy of seconds until v2.7.1
+ /// </summary>
+ internal sealed class LegacyJobResourceUploader : IJobResourceUploader
+ {
+ private static readonly Logger Log = Logger.GetLogger(typeof(LegacyJobResourceUploader));
+
+ private static readonly string JavaClassNameForResourceUploader =
+ @"org.apache.reef.bridge.client.JobResourceUploader";
+
+ private readonly IJavaClientLauncher _javaLauncher;
+ private readonly IJobSubmissionDirectoryProvider _jobSubmissionDirectoryProvider;
+ private readonly IResourceArchiveFileGenerator _resourceArchiveFileGenerator;
+ private readonly IFile _file;
+
+ [Inject]
+ private LegacyJobResourceUploader(IJavaClientLauncher javaLauncher,
+ IJobSubmissionDirectoryProvider jobSubmissionDirectoryProvider,
+ IResourceArchiveFileGenerator resourceArchiveFileGenerator,
+ IFile file)
+ {
+ _file = file;
+ _resourceArchiveFileGenerator = resourceArchiveFileGenerator;
+ _jobSubmissionDirectoryProvider = jobSubmissionDirectoryProvider;
+ _javaLauncher = javaLauncher;
+ }
+
+ public JobResource UploadJobResource(string driverLocalFolderPath)
+ {
+ driverLocalFolderPath = driverLocalFolderPath.TrimEnd('\\') + @"\";
+ string driverUploadPath = _jobSubmissionDirectoryProvider.GetJobSubmissionDirectory().TrimEnd('/') + @"/";
+ Log.Log(Level.Info, "DriverFolderPath: {0} DriverUploadPath: {1}", driverLocalFolderPath, driverUploadPath);
+
+ var archivePath = _resourceArchiveFileGenerator.CreateArchiveToUpload(driverLocalFolderPath);
+
+ var resourceDetailsOutputPath = Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString("N"));
+ _javaLauncher.Launch(JavaClassNameForResourceUploader,
+ archivePath,
+ driverUploadPath,
+ resourceDetailsOutputPath);
+
+ return ParseGeneratedOutputFile(resourceDetailsOutputPath);
+ }
+
+ private JobResource ParseGeneratedOutputFile(string resourceDetailsOutputPath)
+ {
+ if (!_file.Exists(resourceDetailsOutputPath))
+ {
+ Exceptions.Throw(
+ new FileNotFoundException("Could not find resource details file " + resourceDetailsOutputPath),
+ Log);
+ }
+
+ string fileContent = _file.ReadAllText(resourceDetailsOutputPath).Trim();
+
+ Log.Log(Level.Info, "Java uploader returned content: " + fileContent);
+
+ _file.Delete(resourceDetailsOutputPath);
+ string[] tokens = fileContent.Split(';');
+
+ return new JobResource
+ {
+ RemoteUploadPath = tokens[0],
+ LastModificationUnixTimestamp = long.Parse(tokens[1]),
+ ResourceSize = long.Parse(tokens[2])
+ };
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/cs/Org.Apache.REEF.Client/YARN/YARNREEFClient.cs
----------------------------------------------------------------------
diff --git a/lang/cs/Org.Apache.REEF.Client/YARN/YARNREEFClient.cs b/lang/cs/Org.Apache.REEF.Client/YARN/YARNREEFClient.cs
index acec5e7..ce242bc 100644
--- a/lang/cs/Org.Apache.REEF.Client/YARN/YARNREEFClient.cs
+++ b/lang/cs/Org.Apache.REEF.Client/YARN/YARNREEFClient.cs
@@ -49,7 +49,7 @@ namespace Org.Apache.REEF.Client.Yarn
private static readonly Logger Logger = Logger.GetLogger(typeof(YarnREEFClient));
private readonly DriverFolderPreparationHelper _driverFolderPreparationHelper;
- private readonly JavaClientLauncher _javaClientLauncher;
+ private readonly IJavaClientLauncher _javaClientLauncher;
private readonly string _securityTokenKind;
private readonly string _securityTokenService;
private readonly string _jobSubmissionPrefix;
@@ -57,7 +57,7 @@ namespace Org.Apache.REEF.Client.Yarn
private readonly IYarnRMClient _yarnClient;
[Inject]
- internal YarnREEFClient(JavaClientLauncher javaClientLauncher,
+ internal YarnREEFClient(IJavaClientLauncher javaClientLauncher,
DriverFolderPreparationHelper driverFolderPreparationHelper,
REEFFileNames fileNames,
YarnCommandLineEnvironment yarn,
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/0a78fcac/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/JobResourceUploader.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/JobResourceUploader.java b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/JobResourceUploader.java
new file mode 100644
index 0000000..1e9fb25
--- /dev/null
+++ b/lang/java/reef-bridge-client/src/main/java/org/apache/reef/bridge/client/JobResourceUploader.java
@@ -0,0 +1,84 @@
+/*
+ * 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.reef.bridge.client;
+
+import org.apache.commons.lang.Validate;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.URL;
+import org.apache.reef.runtime.yarn.client.YarnClientConfiguration;
+import org.apache.reef.runtime.yarn.client.uploader.JobUploader;
+import org.apache.reef.runtime.yarn.driver.parameters.JobSubmissionDirectoryPrefix;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.Configurations;
+import org.apache.reef.tang.Tang;
+import org.apache.reef.tang.exceptions.InjectionException;
+
+import java.io.*;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Helper class that uploads job resource to HDFS.
+ */
+public final class JobResourceUploader {
+ private static final Logger LOG = Logger.getLogger(JobResourceUploader.class.getName());
+
+ private JobResourceUploader(){}
+
+ /**
+ * This class is invoked from Org.Apache.REEF.Client.Yarn.LegacyJobResourceUploader in .NET code.
+ * Arguments:
+ * [0] : Local path for already generated archive
+ * [1] : Path of job submission directory
+ * [2] : File path for output with details of uploaded resource
+ */
+ public static void main(final String[] args) throws InjectionException, IOException {
+ Validate.isTrue(args.length == 3, "Job resource uploader requires 3 args");
+ final File localFile = new File(args[0]);
+ Validate.isTrue(localFile.exists(), "Local archive does not exist " + localFile.getAbsolutePath());
+ final String jobSubmissionDirectory = args[1];
+ final String localOutputPath = args[2];
+
+ LOG.log(Level.INFO, "Received args: LocalPath " + localFile.getAbsolutePath() + " Submission directory " +
+ jobSubmissionDirectory + " LocalOutputPath " + localOutputPath);
+ final Configuration configuration = Configurations.merge(
+ Tang.Factory.getTang().newConfigurationBuilder()
+ .bindNamedParameter(JobSubmissionDirectoryPrefix.class, jobSubmissionDirectory)
+ .build(),
+ YarnClientConfiguration.CONF.build());
+
+ final JobUploader jobUploader = Tang.Factory.getTang()
+ .newInjector(configuration)
+ .getInstance(JobUploader.class);
+ final LocalResource localResource = jobUploader.createJobFolder(jobSubmissionDirectory)
+ .uploadAsLocalResource(localFile);
+
+ // Output: <UploadedPath>;<LastModificationUnixTimestamp>;<ResourceSize>
+ final URL resource = localResource.getResource();
+ final String outputString = String.format("%s://%s:%d%s;%d;%d", resource.getScheme(), resource.getHost(),
+ resource.getPort(), resource.getFile(), localResource.getTimestamp(), localResource.getSize());
+ LOG.log(Level.INFO, "Writing output: " + outputString);
+ try (Writer writer = new BufferedWriter(new OutputStreamWriter(
+ new FileOutputStream(localOutputPath), "utf-8"))) {
+ writer.write(outputString);
+ }
+
+ LOG.log(Level.FINER, "Done writing output file");
+ }
+}