//******************************************************************************************************//  HadoopReplicationProvider.cs - Gbtc////  Copyright © 2012, Grid Protection Alliance.  All Rights Reserved.////  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See//  the NOTICE file distributed with this work for additional information regarding copyright ownership.//  The GPA licenses this file to you under the MIT License (MIT), 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.opensource.org/licenses/MIT////  Unless agreed to in writing, the subject software distributed under the License is distributed on an//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the//  License for the specific language governing permissions and limitations.////  Code Modification History://  ----------------------------------------------------------------------------------------------------//  11/12/2009 - Pinal C. Patel//       Generated original version of source code.//  11/18/2009 - Pinal C. Patel//       Localized the use of replication log file to ReplicateArchive() method.//  12/04/2009 - Pinal C. Patel//       Added more information to replication log file that can be useful for troubleshooting.//  12/22/2009 - Pinal C. Patel//       Added try...catch block to the code that performs FTP file delete when an exception is //       encountered in ReplicateArchive() method.//  12/23/2009 - Pinal C. Patel//       Added HashRequestAttempts and HashRequestWaitTime properties for better control over HDFS file //       hash request process.//  12/24/2009 - Pinal C. Patel//       Added error handling around the hash request process due to the flaky nature of FTP server when//       dealing with large files.//  03/03/2010 - Pinal C. Patel//       Modified to include files in sub directories of the root directory to be included in the //       replication process.//  05/20/2010 - Pinal C. Patel//       Added the option to allow for deletion of the original files after being replicated.//  06/10/2010 - Pinal C. Patel//       Added the option to allow multiple archive locations to be specified delimited by ';' in //       ArchiveLocation for convenience.//       Added extensive debug messages to enhance the debugging experience of the adapter.//  09/13/2010 - Pinal C. Patel//       Modified the replication process to attempt deletion of the original file again if deletion was //       unsuccessful during previous attempt.//  10/11/2010 - Mihir Brahmbhatt//       Updated header and license agreement.//  12/20/2012 - Starlynn Danyelle Gilliam//       Modified Header.////****************************************************************************************************** using System;using System.Collections.Generic;using System.Data;using System.Diagnostics;using System.IO;using System.Security.Cryptography;using System.Threading;using GSF;using GSF.Collections;using GSF.Configuration;using GSF.Historian.Replication;using GSF.IO;using GSF.IO.Checksums;using GSF.Net.Ftp;using GSF.Units; namespace Hadoop.Replication{    /// <summary>    /// Represents a provider of replication for the <see cref="GSF.Historian.IArchive"/> to Hadoop using FTP channel.    /// </summary>    public class HadoopReplicationProvider : ReplicationProviderBase    {        #region [ Members ]         // Constants         /// <summary>        /// Specifies the default value for the <see cref="BytesPerCrc32"/> property.        /// </summary>        public const int DefaultBytesPerCrc32 = 512;         /// <summary>        /// Specifies the default value for the <see cref="HdfsBlockSize"/> property.        /// </summary>        public const int DefaultHdfsBlockSize = 64;         /// <summary>        /// Specifies the default value for the <see cref="ApplyBufferPadding"/> property.        /// </summary>        public const bool DefaultApplyBufferPadding = true;         /// <summary>        /// Specifies the default value for the <see cref="HashRequestAttempts"/> property.        /// </summary>        public const int DefaultHashRequestAttempts = 3;         /// <summary>        /// Specifies the default value for the <see cref="HashRequestWaitTime"/> property.        /// </summary>        public const int DefaultHashRequestWaitTime = 3000;         /// <summary>        /// Specifies the default value for the <see cref="DeleteOriginalFiles"/> property.        /// </summary>        public const bool DefaultDeleteOriginalFiles = false;         /// <summary>        /// Length to be used for the <see cref="FilePath.TrimFileName"/> method.        /// </summary>        private const int FilePathTrimLength = 30;         /// <summary>        /// Name of the file where replication history information is to be serialized.        /// </summary>        private const string ReplicationLogFile = "HadoopReplicationLog.xml";         // Fields        private int m_bytesPerCrc32;        private int m_hdfsBlockSize;        private bool m_applyBufferPadding;        private int m_hashRequestAttempts;        private int m_hashRequestWaitTime;        private bool m_deleteOriginalFiles;         #endregion         #region [ Constructors ]         /// <summary>        /// Initializes a new instance of the <see cref="HadoopReplicationProvider"/> class.        /// </summary>        public HadoopReplicationProvider()        {            m_bytesPerCrc32 = DefaultBytesPerCrc32;            m_hdfsBlockSize = DefaultHdfsBlockSize;            m_applyBufferPadding = DefaultApplyBufferPadding;            m_hashRequestAttempts = DefaultHashRequestAttempts;            m_hashRequestWaitTime = DefaultHashRequestWaitTime;            m_deleteOriginalFiles = DefaultDeleteOriginalFiles;        }         #endregion         #region [ Properties ]         /// <summary>        /// Gets or sets the number of bytes at which HDFS is configured to compute a CRC32.        /// </summary>        /// <exception cref="ArgumentException">Value being assigned is zero or negative.</exception>        public int BytesPerCrc32        {            get            {                return m_bytesPerCrc32;            }            set            {                if (value < 1)                    throw new ArgumentException("Value must be at least 1.");                 m_bytesPerCrc32 = value;            }        }         /// <summary>        /// Gets or sets the size, in MB, of the data blocks for HDFS where the file resides.        /// </summary>        /// <exception cref="ArgumentException">Value being assigned is zero or negative.</exception>        public int HdfsBlockSize        {            get            {                return m_hdfsBlockSize;            }            set            {                if (value < 1)                    throw new ArgumentException("Value must be at least 1.");                 m_hdfsBlockSize = value;            }        }         /// <summary>        /// Gets or sets a boolean value that indicates whether the buffer used for computing file hash is to be padded with null bytes for replicating HDFS hashing bug.        /// </summary>        public bool ApplyBufferPadding        {            get            {                return m_applyBufferPadding;            }            set            {                m_applyBufferPadding = value;            }        }         /// <summary>        /// Gets or sets the maximum number of requests to be made to the FTP server for HDFS file hash.        /// </summary>        /// <exception cref="ArgumentException">The value being assigned is negative or zero.</exception>        public int HashRequestAttempts        {            get            {                return m_hashRequestAttempts;            }            set            {                if (value < 0)                    throw new ArgumentException("Value must be positive");                 m_hashRequestAttempts = value;            }        }         /// <summary>        /// Gets or set the time, in milliseconds, to wait between requests to the FTP server for HDFS file hash.        /// </summary>        /// <exception cref="ArgumentException">The value being assigned is negative or zero.</exception>        public int HashRequestWaitTime        {            get            {                return m_hashRequestWaitTime;            }            set            {                if (value < 0)                    throw new ArgumentException("Value must be positive");                 m_hashRequestWaitTime = value;            }        }         /// <summary>        /// Gets or sets a boolean value that indicates whether the original files are to be deleted after being replicated successfully.        /// </summary>        public bool DeleteOriginalFiles        {            get            {                return m_deleteOriginalFiles;            }            set            {                m_deleteOriginalFiles = value;            }        }         #endregion         #region [ Methods ]         /// <summary>        /// Saves <see cref="HadoopReplicationProvider"/> settings to the config file if the <see cref="GSF.Adapters.Adapter.PersistSettings"/> property is set to true.        /// </summary>        public override void SaveSettings()        {            base.SaveSettings();            if (PersistSettings)            {                // Save settings under the specified category.                ConfigurationFile config = ConfigurationFile.Current;                CategorizedSettingsElementCollection settings = config.Settings[SettingsCategory];                settings["BytesPerCrc32", true].Update(m_bytesPerCrc32);                settings["HdfsBlockSize", true].Update(m_hdfsBlockSize);                settings["ApplyBufferPadding", true].Update(m_applyBufferPadding);                settings["HashRequestAttempts", true].Update(m_hashRequestAttempts);                settings["HashRequestWaitTime", true].Update(m_hashRequestWaitTime);                settings["DeleteOriginalFiles", true].Update(m_deleteOriginalFiles);                config.Save();            }        }         /// <summary>        /// Loads saved <see cref="HadoopReplicationProvider"/> settings from the config file if the <see cref="GSF.Adapters.Adapter.PersistSettings"/> property is set to true.        /// </summary>          public override void LoadSettings()        {            base.LoadSettings();            if (PersistSettings)            {                // Load settings from the specified category.                ConfigurationFile config = ConfigurationFile.Current;                CategorizedSettingsElementCollection settings = config.Settings[SettingsCategory];                settings.Add("BytesPerCrc32", m_bytesPerCrc32, "Number of bytes at which HDFS is configured to compute a CRC32.");                settings.Add("HdfsBlockSize", m_hdfsBlockSize, "Size (in MB) of the data blocks for HDFS where the file resides.");                settings.Add("ApplyBufferPadding", m_applyBufferPadding, "True if the buffer used for computing file hash is to be padded with null bytes for replicating HDFS hashing bug, otherwise False.");                settings.Add("HashRequestAttempts", m_hashRequestAttempts, "Maximum number of requests to be made to the FTP server for HDFS file hash.");                settings.Add("HashRequestWaitTime", m_hashRequestWaitTime, "Time (in milliseconds) to wait between requests to the FTP server for HDFS file hash.");                settings.Add("DeleteOriginalFiles", m_deleteOriginalFiles, "True if the original files are to be deleted after being replicated successfully; otherwise False.");                BytesPerCrc32 = settings["BytesPerCrc32"].ValueAs(m_bytesPerCrc32);                HdfsBlockSize = settings["HdfsBlockSize"].ValueAs(m_hdfsBlockSize);                ApplyBufferPadding = settings["ApplyBufferPadding"].ValueAs(m_applyBufferPadding);                HashRequestAttempts = settings["HashRequestAttempts"].ValueAs(m_hashRequestAttempts);                HashRequestWaitTime = settings["HashRequestWaitTime"].ValueAs(m_hashRequestWaitTime);                DeleteOriginalFiles = settings["DeleteOriginalFiles"].ValueAs(m_deleteOriginalFiles);            }        }         /// <summary>        /// Replicates the <see cref="GSF.Historian.IArchive"/>.        /// </summary>        protected override void ReplicateArchive()        {            WriteTrace("Archive replication started");             // Parse FTP client information.            Uri replicaUri = new Uri(ReplicaLocation);            string[] credentials = replicaUri.UserInfo.Split(':');             // Ensure credentials are supplied.            if (credentials.Length != 2)                throw new ArgumentException("FTP credentials are missing in ReplicaLocation.");             // Create FTP client for uploading.            FtpClient ftpClient = new FtpClient();            ftpClient.Server = replicaUri.Host;            ftpClient.Port = replicaUri.Port;            ftpClient.FileTransferProgress += FtpClient_FileTransferProgress;             // Initialize the replication log.            WriteTrace("Initializing archive replication log");            DataTable replicationLog = new DataTable("ReplicationRecord");            replicationLog.Columns.Add("DateTime");            replicationLog.Columns.Add("FileName");            replicationLog.Columns.Add("FileHash");            replicationLog.Columns.Add("FileSync");            replicationLog.Columns.Add("HashingTime");            replicationLog.Columns.Add("TransferTime");            replicationLog.Columns.Add("TransferRate");            replicationLog.Columns.Add("ServerRequests");            replicationLog.Columns.Add("ServerResponse");            if (File.Exists(FilePath.GetAbsolutePath(ReplicationLogFile)))                replicationLog.ReadXml(FilePath.GetAbsolutePath(ReplicationLogFile));            WriteTrace("Archive replication log initialized");             try            {                // Connect FTP client to server.                WriteTrace("Connecting to ftp://{0}:{1}", ftpClient.Server, ftpClient.Port);                ftpClient.Connect(credentials[0], credentials[1]);                WriteTrace("Connection successful");                WriteTrace("Changing current directory to '{0}'", replicaUri.AbsolutePath);                ftpClient.SetCurrentDirectory(replicaUri.LocalPath);                WriteTrace("Current directory changed to '{0}'", ftpClient.CurrentDirectory.FullPath);                 // Process all archive location(s).                foreach (string folder in ArchiveLocation.Split(';'))                {                    if (!string.IsNullOrEmpty(folder))                        ReplicateToHadoop(ftpClient, folder.Trim(), replicationLog);                }            }            finally            {                ftpClient.Dispose();                replicationLog.WriteXml(FilePath.GetAbsolutePath(ReplicationLogFile));            }             WriteTrace("Archive replication complete");        }         private void ReplicateToHadoop(FtpClient ftpClient, string replicationFolder, DataTable replicationLog)        {            WriteTrace("Replicating folder '{0}'", FilePath.TrimFileName(replicationFolder, FilePathTrimLength));             // Create list of files to be replicated.            List<string> files = new List<string>(Directory.GetFiles(replicationFolder, "*_to_*.d", SearchOption.AllDirectories));            files.Sort();             // Process all the files in the list.            WriteTrace("Found {0} files in folder", files.Count);            foreach (string file in files)            {                // Initialize local variables.                int requests = int.MinValue;                bool uploading = false;                double hashingStartTime = double.MinValue;                double hashingTotalTime = double.MinValue;                double transferStartTime = double.MinValue;                double transferTotalTime = double.MinValue;                string justFileName = FilePath.GetFileName(file);                DataRow record = null;                DataRow[] filter = replicationLog.Select(string.Format("FileName ='{0}'", justFileName));                FileInfo fileInfo = new FileInfo(file);                 WriteTrace("Replicating file '{0}' of size {1:0,0} KB", FilePath.TrimFileName(file, FilePathTrimLength), Convert.ToInt32(fileInfo.Length / 1024D));                try                {                    // Continue to "ping" FTP server so that it knows we are alive and well                    ftpClient.ControlChannel.Command("NOOP");                     // Compute HDFS file hash.                    WriteTrace("Hashing file");                    hashingStartTime = Common.SystemTimer;                    byte[] localHash = ComputeHdfsFileHash(file, m_bytesPerCrc32, m_hdfsBlockSize, m_applyBufferPadding);                    hashingTotalTime = Common.SystemTimer - hashingStartTime;                    WriteTrace("File hashed in {0} seconds", Convert.ToInt32(hashingTotalTime));                     // Check if file is to be uploaded.                    if (filter.Length == 0 ||                        filter[0]["FileSync"].ToString() == "Fail" ||                        localHash.CompareTo(ByteEncoding.Hexadecimal.GetBytes(filter[0]["FileHash"].ToString())) != 0)                    {                        // Upload file to HDFS since:                        // 1) File has not been replicated previously.                        // OR                        // 2) File has been replicated in the past, but its content has changed since then.                        uploading = true;                        WriteTrace("Uploading file");                        transferStartTime = Common.SystemTimer;                        ftpClient.CurrentDirectory.PutFile(file);                        transferTotalTime = Common.SystemTimer - transferStartTime;                        WriteTrace("File uploaded in {0} seconds", Convert.ToInt32(transferTotalTime));                         // Request file hash from HDFS.                        for (requests = 1; requests <= m_hashRequestAttempts; requests++)                        {                            try                            {                                // Wait before request.                                WriteTrace("Waiting {0} seconds before HDFS hash request", m_hashRequestWaitTime / 1000);                                Thread.Sleep(m_hashRequestWaitTime);                                // Request file hash.                                WriteTrace("Requesting HDFS hash (Attempt {0})", requests);                                ftpClient.ControlChannel.Command(string.Format("HDFSCHKSM {0}{1}", ftpClient.CurrentDirectory.FullPath, justFileName));                                WriteTrace("Hash request response - {0}", ftpClient.ControlChannel.LastResponse.Message.RemoveCrLfs());                                // Exit when successful.                                if (ftpClient.ControlChannel.LastResponse.Code == 200)                                    break;                            }                            catch (Exception ex)                            {                                // Try again - Apache MINA FTP server acts funny with updoad & hash check of large files.                                WriteTrace("Hash request error - {0}", ex.Message);                            }                        }                         // Initialize replication log entry.                        if (filter.Length > 0)                        {                            record = filter[0];                        }                        else                        {                            record = replicationLog.NewRow();                            replicationLog.Rows.Add(record);                        }                         // Update replication log entry.                        record["DateTime"] = DateTime.UtcNow;                        record["FileName"] = justFileName;                        record["FileHash"] = ByteEncoding.Hexadecimal.GetString(localHash);                        record["HashingTime"] = hashingTotalTime.ToString("0.000");                        record["TransferTime"] = transferTotalTime.ToString("0.000");                        record["TransferRate"] = ((new FileInfo(file).Length / SI2.Kilo) / transferTotalTime).ToString("0.00");                        record["ServerRequests"] = requests < m_hashRequestAttempts ? requests : m_hashRequestAttempts;                        record["ServerResponse"] = ftpClient.ControlChannel.LastResponse.Message.RemoveCrLfs();                         // Compare local and HDFS hash.                        if (ftpClient.ControlChannel.LastResponse.Code == 200 &&                            localHash.CompareTo(ByteEncoding.Hexadecimal.GetBytes(ftpClient.ControlChannel.LastResponse.Message.RemoveCrLfs().Split(':')[1])) == 0)                        {                            // File uploaded and hashes match.                            record["FileSync"] = "Pass";                            WriteTrace("Replication successful");                             // Deleted original file after replication.                            if (m_deleteOriginalFiles)                                DeleteOriginalFile(file, fileInfo);                             // Notify about the successful replication.                            OnReplicationProgress(new ProcessProgress<int>("ReplicateArchive", justFileName, 1, 1));                        }                        else                        {                            // Hashes are different - possible causes:                            // 1) Local file got modified after hash was computed locally.                            // OR                            // 2) Local and remote hashing algorithms are not the same.                            record["FileSync"] = "Fail";                            WriteTrace("Replication unsuccessful");                            throw new InvalidDataException("File hash mismatch");                        }                         // Write replication entry to the log file.                        WriteTrace("Updating replication log file");                        replicationLog.WriteXml(FilePath.GetAbsolutePath(ReplicationLogFile));                        WriteTrace("Replication log file updated");                    }                    else                    {                        WriteTrace("Replication skipped - file content unchanged"); 

Lines 1–500 of 660. Load more