/* Copyright (c) Microsoft Corporation All rights reserved. Licensed 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 THIS CODE IS PROVIDED *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE, MERCHANTABLITY OR NON-INFRINGEMENT. See the Apache Version 2.0 License for specific language governing permissions and limitations under the License. */ using System.Security.Cryptography.X509Certificates; using System; using System.Collections.Generic; using System.IO; using System.Linq; using System.Net; using System.Threading; using Microsoft.Research.Peloponnese.Azure; using Microsoft.Research.Peloponnese.Hdfs; using Microsoft.Research.Peloponnese.Shared; using Microsoft.Research.Peloponnese.WebHdfs; using Microsoft.Research.Peloponnese.Yarn; using System.Xml.Linq; using Microsoft.Research.Tools; using Microsoft.WindowsAzure.Management.HDInsight; namespace Microsoft.Research.JobObjectModel { /// /// Serializable properties key-value pairs. /// [Serializable] public class PropertySetting { /// /// Property of a configuration. /// public string Property { get; set; } /// /// Value associated with property. /// public string Value { get; set; } /// /// Empty constructor, for serialization. /// public PropertySetting() { } /// /// Create a property setting. /// /// Property name. /// Value. public PropertySetting(String prop, string value) { this.Property = prop; this.Value = value; } } /// /// The serializable data part of a clusterConfiguration. /// [Serializable] public class ClusterConfigurationSerialization { /// /// Cluster name. /// public string Name { get; set; } /// /// Cluster type. /// public ClusterConfiguration.ClusterType Type { get; set; } /// /// The other properties. /// public List Properties { get; set; } /// /// Create a ClusterConfiguration from its serialization. /// /// The corresponding cluster configuration. public ClusterConfiguration Create() { var config = ClusterConfiguration.CreateConfiguration(this.Type); config.Name = this.Name; for (int i = 0; i < this.Properties.Count; i++) { var property = config.GetType().GetProperty(this.Properties[i].Property); string stringValue = this.Properties[i].Value; object value; if (property.PropertyType == typeof (int)) { value = int.Parse(stringValue); } else if (property.PropertyType == typeof (string)) { value = stringValue; } else if (property.PropertyType == typeof (Uri)) { value = new Uri(stringValue); } else { throw new InvalidCastException("Properties of type " + property.PropertyType + " not yet supported"); } property.SetValue(config, value); } return config; } } /// /// All configuration parameters descrbing the cluster setup should be here. /// public abstract class ClusterConfiguration { /// /// The type of runtime for the cluster. /// public enum ClusterType { /// /// Could not detect cluster version. /// Unknown, /// /// Cluster is running on the cosmos runtime. /// Cosmos, /// /// Cluster is running on the windows high-performance computing platform released by external research. /// ExternalResearchHPC, /// /// The taiga version of HPC. /// HPC, /// /// Cosmos cluster running scope. /// Scope, /// /// Fake cluster, used for post-mortem debugging; keeps some information about jobs in a local folder. /// Cache, /// /// Cluster emulated on a local machine /// LocalEmulator, /// /// Azure DFS client /// AzureDfs, /// /// Web-access to HDFS /// WebHdfs, /// /// Hdfs direct access. /// Hdfs, /// /// Error in creating configuration. /// Error, /// /// Max type, unused; for enumerating. /// MaxUnused, }; /// /// Set of cluster types available. /// public static HashSet Available; static ClusterConfiguration() { Available = new HashSet(); Available.Add(ClusterType.Cache); Available.Add(ClusterType.LocalEmulator); Available.Add(ClusterType.AzureDfs); Available.Add(ClusterType.WebHdfs); Available.Add(ClusterType.Hdfs); } /// /// Properties that can be edited. /// /// List of properties that can be edited. public abstract List GetPropertiesToEdit(); /// /// Must be called after setting all properties. /// Returns null if initialization succeeds, an error otherwise. /// public abstract string Initialize(); /// /// True if the cluster supports diagnosis. /// public bool SupportsDiagnosis { get; protected set; } /// /// Cluster description. /// public string Description { get; set; } private delegate object Work(); /// /// Enumerate all clusters this user is subscribed to. /// /// A list of clusters. public static IEnumerable EnumerateSubscribedClusters() { // ReSharper disable once JoinDeclarationAndInitializer IEnumerable list = null; try { Work work = AzureDfsClusterConfiguration.EnumerateAzureDfsSubscribedClusters; IAsyncResult result = work.BeginInvoke(null, null); if (result.IsCompleted == false) { result.AsyncWaitHandle.WaitOne(3000, false); if (result.IsCompleted == false) throw new ApplicationException("Timeout scanning Azure clusters"); } list = (List)work.EndInvoke(result); } catch (Exception ex) { Console.WriteLine("Exception enumerating DFS clusters: " + ex); } if (list != null) { foreach (var c in list) yield return c; } } /// /// Create serialization data structure for this configuration. /// /// The corresponding serialization. public ClusterConfigurationSerialization ExtractData() { ClusterConfigurationSerialization result = new ClusterConfigurationSerialization { Type = this.TypeOfCluster, Name = this.Name, Properties = new List() }; foreach (var prop in this.GetPropertiesToEdit()) { var property = this.GetType().GetProperty(prop); var value = property.GetValue(this); PropertySetting setting = new PropertySetting(prop, value != null ? value.ToString() : null); result.Properties.Add(setting); } return result; } /// /// The serialization of all known clusters. /// /// The serializations of all known clusters in a list. public static List KnownClustersSerialization() { var result = new List(); foreach (var clus in KnownClusters.Values) { result.Add(clus.ExtractData()); } return result; } /// /// Reconstruct the known clusters from the saved serialization. /// /// Serializations for each cluster. public static void ReconstructKnownCluster(List sers) { if (sers == null) return; foreach (var cs in sers) { try { var clus = cs.Create(); if (clus == null) continue; string error = clus.Initialize(); if (error != null) continue; AddKnownCluster(clus); } catch (Exception ex) { Console.WriteLine("Error reconstructing saved cluster; skipping " + cs.Name + ":" + ex.Message); } } } /// /// Create a cluster configuration of the specified type. /// /// Type of cluster. protected ClusterConfiguration(ClusterConfiguration.ClusterType type) { this.Description = ""; this.TypeOfCluster = type; this.SupportsDiagnosis = true; } /// /// Credentials to use for authentication. /// [NonSerialized] // ReSharper disable once NotAccessedField.Global protected NetworkCredential credentials; /// /// Set the credentials for connecting to this cluster. /// /// Credentials to use. public void SetCredential(NetworkCredential credential) { this.credentials = credential; } /// /// The name of this cluster. /// public string Name { get; set; } /// /// The type of the cluster /// public ClusterType TypeOfCluster { get; protected set; } /// /// Base directory on all cluster machines. /// public string DryadInstallDirectory { get; protected set; } /// /// Default domain for user account to use for connecting to cluster; empty to use the current domain. /// public string UserDomain { get; protected set; } /// /// The machine where the metadata for the copied jobs is stored. /// public virtual string MetaDataMachine { get; protected set; } /// /// Time zone of the analyzed cluster. We assume that the cluster is in the local time zone. /// /// Timezone infomation of the cluster /// Job we are interested in. public virtual TimeZoneInfo GetClusterTimeZone(DryadLinqJobSummary job) { return TimeZoneInfo.Local; } /// /// The directory where a specific process is created on the cluster. /// /// Process identifier /// Machine where process ran. /// Home directory containing the process information (not working directory of vertex). /// Job where the process belongs. /// True if vertex is terminated. public abstract IClusterResidentObject ProcessDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job); /// /// Work directory of a process vertex. /// /// Vertex guid. /// Machine where process ran. /// The path to the work directory of the vertex. /// Job where the process belongs. /// True if vertex is terminated. public abstract IClusterResidentObject ProcessWorkDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job); /// /// Given an input file identify the process that produced it. /// /// Input file of a process. /// Job that contained the process. /// The identity of the process that produced the file. // ReSharper disable UnusedParameter.Global public abstract DryadProcessIdentifier ProcessFromInputFile(IClusterResidentObject input, DryadLinqJobSummary job); // ReSharper restore UnusedParameter.Global /// /// File containing standard output of a process. /// /// Process identifier. /// Machine where process ran. /// Job containing process. /// The pathname to the standard output. /// True if vertex is terminated. public virtual IClusterResidentObject ProcessStdoutFile(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { IClusterResidentObject processdir = this.ProcessDirectory(identifier, terminated, machine, job); IClusterResidentObject file = processdir.GetFile("stdout.txt"); return file; } /// /// Create a cluster status for this cluster. /// /// The proper cluster status. public abstract ClusterStatus CreateClusterStatus(); static Dictionary KnownClusters = new Dictionary(); /// /// A known cluster configuration by name. /// /// Name of the cluster. /// The cluster configuration for that cluster. public static ClusterConfiguration KnownClusterByName(string name) { if (KnownClusters.ContainsKey(name)) return KnownClusters[name]; return null; } /// /// Add a new cluster to the list of known clusters. /// /// New config to add. public static void AddKnownCluster(ClusterConfiguration config) { if (KnownClusters.ContainsKey(config.Name)) { KnownClusters[config.Name] = config; } else { KnownClusters.Add(config.Name, config); } } /// /// Remove a cluster form the list of known clusters. /// /// Name of cluster to remove. /// The removed configuration. public static ClusterConfiguration RemoveKnownCluster(string name) { if (KnownClusters.ContainsKey(name)) { var config = KnownClusters[name]; KnownClusters.Remove(name); return config; } return null; } /// /// Get the list of known clusters. /// /// A list of cluster names. public static IEnumerable GetKnownClusterNames() { return KnownClusters.Keys; } /// /// File containing standard error of a process. Not available on all architectures. /// /// Process identifier. /// Machine where process ran. /// Job containing process. /// A reference to the standard output. /// Vertex state. public virtual IClusterResidentObject ProcessStderrFile(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { return this.ProcessStdoutFile(identifier, terminated, machine, job); } /// /// Log directory of a process vertex. /// /// Vertex guid. /// The path to the work directory of the vertex. /// Machine where process ran. /// Job where the process belongs. /// Vertex state. public abstract IClusterResidentObject ProcessLogDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job); /// /// A shell pattern matching (just the) log files produced by a job manager process. /// /// Pattern matching the log files. /// If true, return only the error logs. /// Job where the JM process belongs. // ReSharper disable once UnusedParameter.Global public abstract string JMLogFilesPattern(bool error, DryadLinqJobSummary job); /// /// A shell pattern matching (just the) log files produced by a vertex process. /// /// Pattern matching the log files. /// If true, return only the error logs. /// Job containing this vertex. public abstract string VertexLogFilesPattern(bool error, DryadLinqJobSummary job); /// /// Directory where the perfmon logs are being collected, relative to local machine. /// /// A directory containing the perfmon logs. public virtual string PerfmonLogDirectory() { return Path.Combine(this.DryadInstallDirectory, "Perfmon"); } /// /// Create an empty configuration of the specified type. /// /// Configuration type. /// The created configuration. public static ClusterConfiguration CreateConfiguration(ClusterType type) { switch (type) { case ClusterType.Cache: return new CacheClusterConfiguration(); case ClusterType.LocalEmulator: return new LocalEmulator(); case ClusterType.AzureDfs: return new AzureDfsClusterConfiguration(); case ClusterType.WebHdfs: return new WebHdfsClusterConfiguration(); case ClusterType.Hdfs: return new HdfsClusterConfiguration(); case ClusterType.Unknown: case ClusterType.MaxUnused: default: return new ErrorConfiguration("Unsupported cluster type " + type); } } /// /// Convert a GUID printed by the Dryad job manager into a process-id, which is platform dependent. /// /// Process guid. /// Process id.\ /// Job where guid is from. public abstract string ExtractPidFromGuid(string guid, DryadLinqJobSummary job); /// /// Navigate to a given url and return a stream with the corresponding web page. /// /// Url to navigate to. /// The web page. internal virtual Stream Navigate(string url) { return Utilities.Navigate(url, null); } /// /// The file containing the job query plan. /// /// Job whose plan is sought. /// An object containing the path, or null if it cannot be found. public virtual IClusterResidentObject JobQueryPlan(DryadLinqJobSummary job) { try { IClusterResidentObject dir = this.ProcessWorkDirectory(job.ManagerProcessGuid, true, job.Machine, job); // immutable var matchingfiles = dir.GetFilesAndFolders("DryadLinqProgram__*.xml").ToList(); if (matchingfiles.Count() != 1) throw new ClusterException("Could not find query plan file; got " + matchingfiles.Count() + " possible matches"); IClusterResidentObject result = matchingfiles.First(); result.ShouldCacheLocally = true; // immutable return result; } catch (Exception e) { return new UNCFile(e); } } } /// /// Represents an error in creating a cluster configuration. /// public sealed class ErrorConfiguration : ClusterConfiguration { /// /// Error message. /// public string ErrorMessage { get; private set; } /// /// Create an Error Cluster. /// internal ErrorConfiguration(string message) : base(ClusterType.Error) { this.ErrorMessage = message; } private static List properties = new List(); /// /// Not used. /// /// Exception. public override List GetPropertiesToEdit() { return properties; } /// /// Must be called after setting all properties. /// Returns null if initialization succeeds, an error otherwise. /// public override string Initialize() { return this.ErrorMessage; } /// /// Not used. /// /// Exception. public override IClusterResidentObject ProcessDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { throw new NotImplementedException(); } /// /// Not used. /// /// Exception. public override IClusterResidentObject ProcessWorkDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { throw new NotImplementedException(); } /// /// Not used. /// /// Exception. public override DryadProcessIdentifier ProcessFromInputFile(IClusterResidentObject input, DryadLinqJobSummary job) { throw new NotImplementedException(); } /// /// Not used. /// /// Exception. public override ClusterStatus CreateClusterStatus() { throw new NotImplementedException(); } /// /// Not used. /// /// Exception. public override IClusterResidentObject ProcessLogDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { throw new NotImplementedException(); } /// /// Not used. /// /// Exception. public override string JMLogFilesPattern(bool error, DryadLinqJobSummary job) { throw new NotImplementedException(); } /// /// Not used. /// /// Exception. public override string VertexLogFilesPattern(bool error, DryadLinqJobSummary job) { throw new NotImplementedException(); } /// /// Not used. /// /// Exception. public override string ExtractPidFromGuid(string guid, DryadLinqJobSummary job) { throw new NotImplementedException(); } } /// /// A cached cluster is just a set of files on a filesystem which are saved from old jobs; allows postmortem analysis. /// public sealed class CacheClusterConfiguration : ClusterConfiguration { /// /// Each cached job may originate from a completely different cluster; cache the actual configurations here. /// Dictionary jobConfig; /// /// Folder where the information is saved. /// public string LocalInformationFolder { get; set; } /// /// Create a fake cluster which stores the information in the specified folder. /// public CacheClusterConfiguration() : base(ClusterConfiguration.ClusterType.Cache) { this.LocalInformationFolder = null; this.Name = "Cache"; this.jobConfig = new Dictionary(); } /// /// Make this cluster the active cache. /// public void StartCaching() { string folder = String.IsNullOrEmpty(this.LocalInformationFolder) ? null : this.LocalInformationFolder; CachedClusterResidentObject.CacheDirectory = folder; // enables caching } /// /// Disable caching in this cluster. /// public void StopCaching() { CachedClusterResidentObject.CacheDirectory = null; // disables caching } private static List props = new List { "LocalInformationFolder" }; /// /// Properties that can be edited. /// /// List of properties that can be edited. public override List GetPropertiesToEdit() { return props; } /// /// Must be called after setting all properties. /// Returns null if initialization succeeds, an error otherwise. /// public override string Initialize() { return null; } /// /// Create a cluster status for this cluster. /// /// The proper cluster status. public override ClusterStatus CreateClusterStatus() { var stat = ClusterStatus.LookupStatus(this); if (stat != null) return stat; return new CacheClusterStatus(this); } /// /// Find the actual cluster configuration corresponding to the specified job. /// /// Cached job. /// A configuration that could have been used for the cluster running the job. public ClusterConfiguration ActualConfig(DryadLinqJobSummary job) { if (!this.jobConfig.ContainsKey(job.Cluster)) { ClusterConfiguration config = ClusterConfiguration.KnownClusterByName(job.Cluster); if (config == null) config = ClusterConfiguration.CreateConfiguration(job.ClusterType); this.jobConfig.Add(job.Cluster, config); } return this.jobConfig[job.Cluster]; } /// /// Object that can be used to access the process directory. /// /// Process identifier. /// True if process is terminated. /// Machine where the process ran. /// Job that contained the process. /// An object which can be used to access the process home directory. public override IClusterResidentObject ProcessDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); IClusterResidentObject pd = config.ProcessDirectory(identifier, terminated, machine, job); if (pd == null) return null; IClusterResidentObject result = new FolderInCachedCluster(pd as CachedClusterResidentObject); return result; } /// /// Object that can be used to access the process work directory. /// /// Process identifier. /// True if process is terminated. /// Machine where the process ran. /// Job that contained the process. /// An object which can be used to access the process work directory. public override IClusterResidentObject ProcessWorkDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); IClusterResidentObject wd = config.ProcessWorkDirectory(identifier, terminated, machine, job); if (wd == null) return null; IClusterResidentObject result = new FolderInCachedCluster(wd as CachedClusterResidentObject); return result; } /// /// Time zone of cluster. /// /// Job we are interested in. /// The time zome of the cluster. public override TimeZoneInfo GetClusterTimeZone(DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); return config.GetClusterTimeZone(job); } /// /// Given an input file identify the process that produced it. /// /// Input file of a process. /// Job that contained the process. /// The identity of the process that produced the file. public override DryadProcessIdentifier ProcessFromInputFile(IClusterResidentObject input, DryadLinqJobSummary job) { throw new InvalidOperationException(); } /// /// Object that can be used to access the process log directory. /// /// Process identifier. /// True if process is terminated. /// Machine where the process ran. /// Job that contained the process. /// An object which can be used to access the process log directory. public override IClusterResidentObject ProcessLogDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); IClusterResidentObject pld = config.ProcessLogDirectory(identifier, terminated, machine, job); if (pld == null) return null; IClusterResidentObject result = new FolderInCachedCluster(pld as CachedClusterResidentObject); return result; } /// /// Pattern which matches the log files. /// /// If true return only the log files containing errors. /// A string that can be used to match only log files. /// Job where process belongs. public override string JMLogFilesPattern(bool error, DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); return config.JMLogFilesPattern(error, job); } /// /// Pattern which matches the log files. /// /// If true return only the log files containing errors. /// A string that can be used to match only log files. /// Job containing the vertex. public override string VertexLogFilesPattern(bool error, DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); return config.VertexLogFilesPattern(error, job); } /// /// File containing standard output of a process. /// /// Process identifier. /// Machine where process ran. /// Job containing process. /// The pathname to the standard output. /// True if vertex is terminated. public override IClusterResidentObject ProcessStdoutFile(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); return config.ProcessStdoutFile(identifier, terminated, machine, job); } /// /// File containing standard error of a process. Not available on all architectures. /// /// Process identifier. /// Machine where process ran. /// Job containing process. /// A reference to the standard output. /// Vertex state. public override IClusterResidentObject ProcessStderrFile(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); return config.ProcessStderrFile(identifier, terminated, machine, job); } /// /// Extract the process id from a guid. /// /// Process guid. /// Job containing process. /// The process id. public override string ExtractPidFromGuid(string guid, DryadLinqJobSummary job) { ClusterConfiguration config = this.ActualConfig(job); return config.ExtractPidFromGuid(guid, job); } /// /// The file containing the job query plan. /// /// Job whose plan is sought. /// An object containing the path, or null if it cannot be found. public override IClusterResidentObject JobQueryPlan(DryadLinqJobSummary job) { try { // we have to handle the xml plan differently; the cached file is renamed to always be "0". IClusterResidentObject dir = this.ProcessWorkDirectory(job.ManagerProcessGuid, true, job.Machine, job); if (dir == null) return null; IClusterResidentObject result = dir.GetFile("DryadLinqProgram__0.xml"); return result; } catch (Exception e) { return new UNCFile(e); } } } /// /// A local machine used to run an emulated Yarn cluster. /// public sealed class LocalEmulator : ClusterConfiguration { /// /// Folder where job logs are stored. /// // ReSharper disable once UnusedAutoPropertyAccessor.Local public string JobsFolder { get; private set; } /// /// Create a cluster representing the local machine only. /// public LocalEmulator() : base(ClusterType.LocalEmulator) { } private static List props = new List {"JobsFolder"}; /// /// Must be called after setting all properties. /// Returns null if initialization succeeds, an error otherwise. /// public override string Initialize() { return null; } /// /// Properties that can be edited. /// /// List of properties that can be edited. public override List GetPropertiesToEdit() { return props; } /// /// Create a cluster status for this cluster. /// /// The proper cluster status. public override ClusterStatus CreateClusterStatus() { var stat = ClusterStatus.LookupStatus(this); if (stat != null) return stat; return new YarnEmulatedClusterStatus(this); } /// /// The directory where a specific process is created on the cluster. /// /// Process identifier /// Machine where process ran. /// Home directory containing the process information (not working directory of vertex). /// Job where the process belongs. /// True if vertex is terminated. public override IClusterResidentObject ProcessDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { if (identifier.ToString() == "jm") { // The job manager process is special return new LocalFile(Utilities.PathCombine(this.JobsFolder, job.ClusterJobId, identifier.ProcessIdentifier, "Process.000.001")); } else { string folder = Utilities.PathCombine(this.JobsFolder, job.ClusterJobId, "Worker"); return new LocalFile(Path.Combine(folder, identifier.ProcessIdentifier)); } } /// /// Given an input file, identify the process that has produced it. /// /// Input file. /// Job containing the process. /// The process that has produced the file. public override DryadProcessIdentifier ProcessFromInputFile(IClusterResidentObject input, DryadLinqJobSummary job) { throw new InvalidOperationException(); } /// /// File containing standard error of a process. Not available on all architectures. /// /// Process identifier. /// Machine where process ran. /// Job containing process. /// A reference to the standard output. /// Vertex state. public override IClusterResidentObject ProcessStderrFile(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { IClusterResidentObject processdir = this.ProcessDirectory(identifier, terminated, machine, job); IClusterResidentObject file = processdir.GetFile("stderr.txt"); return file; } /// /// The file containing the job query plan. /// /// Job whose plan is sought. /// An object containing the path, or null if it cannot be found. public override IClusterResidentObject JobQueryPlan(DryadLinqJobSummary job) { try { IClusterResidentObject dir = this.ProcessWorkDirectory(job.ManagerProcessGuid, true, job.Machine, job); // this is missing at this point //IClusterResidentObject dir = this.ProcessWorkDirectory(new DryadProcessIdentifier("Process.000.001"), true, job.Machine, job); var matchingfiles = dir.GetFilesAndFolders("DryadLinqProgram__*.xml").ToList(); if (matchingfiles.Count() != 1) throw new ClusterException("Could not find query plan file; got " + matchingfiles.Count() + " possible matches"); IClusterResidentObject result = matchingfiles.First(); result.ShouldCacheLocally = true; // immutable return result; } catch (Exception e) { return new UNCFile(e); } } /// /// Work directory of a process vertex. /// /// Vertex guid. /// Machine where process ran. /// The path to the work directory of the vertex. /// Job where the process belongs. /// True if vertex is terminated. public override IClusterResidentObject ProcessWorkDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { return this.ProcessDirectory(identifier, terminated, machine, job); } /// /// Log directory of a process vertex. /// /// Vertex guid. /// The path to the work directory of the vertex. /// Machine where process ran. /// True if vertex is terminated. /// Job where the process belongs. public override IClusterResidentObject ProcessLogDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { return this.ProcessDirectory(identifier, terminated, machine, job); } /// /// A shell pattern matching (just the) log files produced by a job manager process. /// /// Pattern matching the log files. /// If true, return only the error logs. /// Job where process belongs. public override string JMLogFilesPattern(bool error, DryadLinqJobSummary job) { return "stderr.txt"; } /// /// A shell pattern matching (just the) log files produced by a vertex process. /// /// Pattern matching the log files. /// If true, return only the error logs. /// Job where process belongs. public override string VertexLogFilesPattern(bool error, DryadLinqJobSummary job) { return this.JMLogFilesPattern(error, job); } /// /// Convert a GUID printed by the Dryad job manager into a process-id, which is platform dependent. /// /// Process guid. /// Process id. /// Job where process belongs. public override string ExtractPidFromGuid(string guid, DryadLinqJobSummary job) { return guid; } } /// /// Configuration for an AzureDfs cluster. /// public abstract class DfsClusterConfiguration : ClusterConfiguration { /// /// Create a cluster representing the local machine only. /// protected DfsClusterConfiguration(ClusterType type) : base(type) { this.SupportsDiagnosis = false; } /// /// Work directory of a process vertex. /// /// Vertex guid. /// Machine where process ran. /// The path to the work directory of the vertex. /// Job where the process belongs. /// True if vertex is terminated. public override IClusterResidentObject ProcessWorkDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { return this.ProcessDirectory(identifier, terminated, machine, job); } /// /// Given an input file identify the process that produced it. /// /// Input file of a process. /// Job that contained the process. /// The identity of the process that produced the file. // ReSharper disable UnusedParameter.Global public override DryadProcessIdentifier ProcessFromInputFile(IClusterResidentObject input, DryadLinqJobSummary job) { return null; } // ReSharper restore UnusedParameter.Global /// /// File containing standard output of a process. /// /// Process identifier. /// Machine where process ran. /// Job containing process. /// The pathname to the standard output. /// True if vertex is terminated. public override IClusterResidentObject ProcessStdoutFile(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { if (identifier.ToString() == "jm") { IClusterResidentObject processdir = this.ProcessDirectory(identifier, terminated, machine, job); IClusterResidentObject file = processdir.GetFile("calypso.log"); return file; } // vertices not supported return null; } /// /// Log directory of a process vertex. /// /// Vertex guid. /// The path to the work directory of the vertex. /// Machine where process ran. /// Job where the process belongs. /// Vertex state. public override IClusterResidentObject ProcessLogDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { return this.ProcessDirectory(identifier, terminated, machine, job); } /// /// A shell pattern matching (just the) log files produced by a job manager process. /// /// Pattern matching the log files. /// If true, return only the error logs. /// Job where the JM process belongs. // ReSharper disable once UnusedParameter.Global public override string JMLogFilesPattern(bool error, DryadLinqJobSummary job) { return "*.log"; } /// /// A shell pattern matching (just the) log files produced by a vertex process. /// /// Pattern matching the log files. /// If true, return only the error logs. /// Job containing this vertex. public override string VertexLogFilesPattern(bool error, DryadLinqJobSummary job) { return "*.log"; } /// /// Convert a GUID printed by the Dryad job manager into a process-id, which is platform dependent. /// /// Process guid. /// Process id.\ /// Job where guid is from. public override string ExtractPidFromGuid(string guid, DryadLinqJobSummary job) { return guid; } } /// /// Configuration for an AzureDfs cluster. /// public sealed class AzureDfsClusterConfiguration : DfsClusterConfiguration { /// /// Handle to client to enumerate logs. /// public AzureDfsClient AzureClient; /// /// Create a cluster representing the local machine only. /// public AzureDfsClusterConfiguration() : base(ClusterType.AzureDfs) { this.Description = "Container is usually `dryad-jobs'"; } /// /// Base Uri to access data in this Cluster. /// public Uri baseUri; /// /// Enumerate all the clusters this user is subscribed to. /// /// The list of clusters this user is subscribed to. public static List EnumerateAzureDfsSubscribedClusters() { List configList = new List(); var store = new X509Store(); store.Open(OpenFlags.ReadOnly); var configDir = Path.Combine( Environment.GetFolderPath(Environment.SpecialFolder.ApplicationData), "Windows Azure Powershell"); var defaultFile = Path.Combine(configDir, "WindowsAzureProfile.xml"); if (File.Exists(defaultFile)) { using (FileStream s = new FileStream(defaultFile, FileMode.Open, FileAccess.Read)) { XDocument doc = XDocument.Load(s); XNamespace ns = doc.Root.GetDefaultNamespace(); IEnumerable subs = doc.Descendants(ns + "AzureSubscriptionData"); foreach (XElement sub in subs) { string thumbprint = sub.Descendants(ns + "ManagementCertificate").Single().Value; string subId = sub.Descendants(ns + "SubscriptionId").Single().Value; Guid subGuid = new Guid(subId); X509Certificate2 cert = store.Certificates.Cast().First(item => item.Thumbprint == thumbprint); HDInsightCertificateCredential sCred = new HDInsightCertificateCredential(subGuid, cert); IHDInsightClient sClient = HDInsightClient.Connect(sCred); var clusters = sClient.ListClusters(); foreach (var cluster in clusters) { var account = cluster.DefaultStorageAccount; var accountName = account.Name.Split('.').First(); Console.WriteLine("Cluster " + cluster.Name + " uses account " + accountName + " with key " + account.Key); AzureDfsClusterConfiguration config = null; try { config = new AzureDfsClusterConfiguration(); config.AzureClient = new AzureDfsClient(accountName, account.Key, config.Container); config.Name = cluster.Name; } catch (Exception ex) { Console.WriteLine("Exception while reconstructing cluster " + cluster.Name + ": " + ex); } if (config != null) configList.Add(config); } } } } return configList; } /// /// The file containing the job query plan. /// /// Job whose plan is sought. /// An object containing the path, or null if it cannot be found. public override IClusterResidentObject JobQueryPlan(DryadLinqJobSummary job) { try { IClusterResidentObject dir = this.ProcessWorkDirectory(job.ManagerProcessGuid, true, job.Machine, job); // immutable var matchingfiles = dir.GetFilesAndFolders("DryadLinqProgram__*.xml").ToList(); if (matchingfiles.Count() != 1) throw new ClusterException("Could not find query plan file; got " + matchingfiles.Count() + " possible matches"); IClusterResidentObject result = matchingfiles.First(); (result as AzureDfsFile).IsDfsStream = true; result.ShouldCacheLocally = true; // immutable return result; } catch (Exception e) { return new UNCFile(e); } } /// /// Azure account name. /// public string AccountName { get; set; } /// /// Azure account key. /// public string AccountKey { get; set; } /// /// Azure container. /// public string Container { get; set; } private static List props = new List { "AccountName", "AccountKey", "Container" }; /// /// Must be called after setting all properties. /// Returns true if initialization succeeds. /// public override string Initialize() { try { this.AzureClient = new AzureDfsClient( this.AccountName, this.AccountKey, this.Container); this.baseUri = Microsoft.Research.Peloponnese.Azure.Utils.ToAzureUri(this.AccountName, this.Container, "", null, this.AccountKey); return null; } catch (Exception ex) { Console.WriteLine(ex); return ex.Message; } } /// /// The directory where a specific process is created on the cluster. /// /// Process identifier /// Machine where process ran. /// Home directory containing the process information (not working directory of vertex). /// Job where the process belongs. /// True if vertex is terminated. public override IClusterResidentObject ProcessDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { if (identifier.ToString() == "jm") { // The job manager process is special var result = new AzureDfsFile(this, job, this.AzureClient, job.ClusterJobId, terminated, true); return result; } // vertices not supported return null; } /// /// Create a cluster status for this cluster. /// /// The proper cluster status. public override ClusterStatus CreateClusterStatus() { var stat = ClusterStatus.LookupStatus(this); if (stat != null) return stat; return new AzureDfsClusterStatus(this); } /// /// Properties that can be edited. /// /// List of properties that can be edited. public override List GetPropertiesToEdit() { return props; } } /// /// Configuration for a WebHdfs cluster. /// public sealed class WebHdfsClusterConfiguration : DfsClusterConfiguration { /// /// Handle to client to access files. /// public HdfsClientBase DfsClient; /// /// Create a cluster representing the local machine only. /// public WebHdfsClusterConfiguration() : base(ClusterType.WebHdfs) { this.Description = "JobsFolderUri usually looks like hdfs://headnode:port/JobsFolder"; } /// /// WebHdfs user name. /// public string UserName { get; set; } /// /// WebHdfs port. /// public int WebHdfsPort { get; set; } /// /// Uri to folder containing jobs. /// public Uri JobsFolderUri { get; set; } /// /// Machine that supplies job status. /// public string StatusNode { get; set; } /// /// Port of status machine. /// public int StatusNodePort { get; set; } private static List props = new List { "UserName", "WebHdfsPort", "JobsFolderUri", "StatusNode", "StatusNodePort" }; /// /// Must be called after setting all properties. /// Returns true if initialization succeeds. /// public override string Initialize() { try { this.DfsClient = new WebHdfsClient(this.UserName, this.WebHdfsPort); return null; } catch (Exception ex) { Console.WriteLine(ex); return ex.Message; } } /// /// The file containing the job query plan. /// /// Job whose plan is sought. /// An object containing the path, or null if it cannot be found. public override IClusterResidentObject JobQueryPlan(DryadLinqJobSummary job) { try { IClusterResidentObject dir = this.ProcessWorkDirectory(job.ManagerProcessGuid, true, job.Machine, job); // immutable var matchingfiles = dir.GetFilesAndFolders("DryadLinqProgram__*.xml").ToList(); if (matchingfiles.Count() != 1) throw new ClusterException("Could not find query plan file; got " + matchingfiles.Count() + " possible matches"); IClusterResidentObject result = matchingfiles.First(); result.ShouldCacheLocally = true; // immutable return result; } catch (Exception e) { return new UNCFile(e); } } /// /// The directory where a specific process is created on the cluster. /// /// Process identifier /// Machine where process ran. /// Home directory containing the process information (not working directory of vertex). /// Job where the process belongs. /// True if vertex is terminated. public override IClusterResidentObject ProcessDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { if (identifier.ToString() == "jm") { // The job manager process is special var result = new DfsFile(this, this.JobsFolderUri, job, this.DfsClient, job.ClusterJobId, terminated, true); return result; } // vertices not supported return null; } /// /// Create a cluster status for this cluster. /// /// The proper cluster status. public override ClusterStatus CreateClusterStatus() { var stat = ClusterStatus.LookupStatus(this); if (stat != null) return stat; return new WebHdfsClusterStatus(this); } /// /// Properties that can be edited. /// /// List of properties that can be edited. public override List GetPropertiesToEdit() { return props; } } /// /// Configuration for an Hdfs cluster. /// public sealed class HdfsClusterConfiguration : DfsClusterConfiguration { /// /// Handle to client to access files. /// public HdfsClientBase DfsClient; /// /// Create a cluster representing the local machine only. /// public HdfsClusterConfiguration() : base(ClusterType.Hdfs) { this.Description = "JobsFolderUri should look like hdfs://headnode:port/JobsFolder"; } /// /// Path to cluster. /// public Uri JobsFolderUri { get; set; } /// /// Port to access HDFS. /// public string UserName { get; set; } /// /// Machine that supplies job status. /// public string StatusNode { get; set; } /// /// Port of status machine. /// public int StatusNodePort { get; set; } private static List props = new List { "UserName", "JobsFolderUri", "StatusNode", "StatusNodePort" }; /// /// Must be called after setting all properties. /// Returns true if initialization succeeds. /// public override string Initialize() { try { this.DfsClient = new HdfsClient(this.UserName); return null; } catch (Exception ex) { Console.WriteLine(ex); return ex.Message; } } /// /// The file containing the job query plan. /// /// Job whose plan is sought. /// An object containing the path, or null if it cannot be found. public override IClusterResidentObject JobQueryPlan(DryadLinqJobSummary job) { try { IClusterResidentObject dir = this.ProcessWorkDirectory(job.ManagerProcessGuid, true, job.Machine, job); // immutable var matchingfiles = dir.GetFilesAndFolders("DryadLinqProgram__*.xml").ToList(); if (matchingfiles.Count() != 1) throw new ClusterException("Could not find query plan file; got " + matchingfiles.Count() + " possible matches"); IClusterResidentObject result = matchingfiles.First(); result.ShouldCacheLocally = true; // immutable return result; } catch (Exception e) { return new UNCFile(e); } } /// /// The directory where a specific process is created on the cluster. /// /// Process identifier /// Machine where process ran. /// Home directory containing the process information (not working directory of vertex). /// Job where the process belongs. /// True if vertex is terminated. public override IClusterResidentObject ProcessDirectory(DryadProcessIdentifier identifier, bool terminated, string machine, DryadLinqJobSummary job) { if (identifier.ToString() == "jm") { // The job manager process is special var result = new DfsFile(this, this.JobsFolderUri, job, this.DfsClient, job.ClusterJobId, terminated, true); return result; } // vertices not supported return null; } /// /// Create a cluster status for this cluster. /// /// The proper cluster status. public override ClusterStatus CreateClusterStatus() { var stat = ClusterStatus.LookupStatus(this); if (stat != null) return stat; var result = new HdfsClusterStatus(this); return result; } /// /// Properties that can be edited. /// /// List of properties that can be edited. public override List GetPropertiesToEdit() { return props; } } }