hadoop-rpc (empty) → 0.1.0.0
raw patch · 17 files changed
+2315/−0 lines, 17 filesdep +attoparsecdep +basedep +bytestringsetup-changed
Dependencies added: attoparsec, base, bytestring, cereal, exceptions, hadoop-rpc, network, protobuf, socks, text, transformers, unix, unordered-containers, vector, xmlhtml
Files
- LICENSE +202/−0
- Setup.hs +2/−0
- hadoop-rpc.cabal +71/−0
- src/Data/Hadoop/Configuration.hs +103/−0
- src/Data/Hadoop/Protobuf/ClientNameNode.hs +152/−0
- src/Data/Hadoop/Protobuf/DataTransfer.hs +319/−0
- src/Data/Hadoop/Protobuf/Hdfs.hs +211/−0
- src/Data/Hadoop/Protobuf/Headers.hs +170/−0
- src/Data/Hadoop/Protobuf/ProtocolInfo.hs +81/−0
- src/Data/Hadoop/Protobuf/Security.hs +66/−0
- src/Data/Hadoop/Types.hs +90/−0
- src/Network/Hadoop/Hdfs.hs +245/−0
- src/Network/Hadoop/Read.hs +217/−0
- src/Network/Hadoop/Rpc.hs +141/−0
- src/Network/Hadoop/Socket.hs +64/−0
- src/Network/Hadoop/Stream.hs +161/−0
- test/Main.hs +20/−0
+ LICENSE view
@@ -0,0 +1,202 @@++ Apache License+ Version 2.0, January 2004+ http://www.apache.org/licenses/++ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION++ 1. Definitions.++ "License" shall mean the terms and conditions for use, reproduction,+ and distribution as defined by Sections 1 through 9 of this document.++ "Licensor" shall mean the copyright owner or entity authorized by+ the copyright owner that is granting the License.++ "Legal Entity" shall mean the union of the acting entity and all+ other entities that control, are controlled by, or are under common+ control with that entity. For the purposes of this definition,+ "control" means (i) the power, direct or indirect, to cause the+ direction or management of such entity, whether by contract or+ otherwise, or (ii) ownership of fifty percent (50%) or more of the+ outstanding shares, or (iii) beneficial ownership of such entity.++ "You" (or "Your") shall mean an individual or Legal Entity+ exercising permissions granted by this License.++ "Source" form shall mean the preferred form for making modifications,+ including but not limited to software source code, documentation+ source, and configuration files.++ "Object" form shall mean any form resulting from mechanical+ transformation or translation of a Source form, including but+ not limited to compiled object code, generated documentation,+ and conversions to other media types.++ "Work" shall mean the work of authorship, whether in Source or+ Object form, made available under the License, as indicated by a+ copyright notice that is included in or attached to the work+ (an example is provided in the Appendix below).++ "Derivative Works" shall mean any work, whether in Source or Object+ form, that is based on (or derived from) the Work and for which the+ editorial revisions, annotations, elaborations, or other modifications+ represent, as a whole, an original work of authorship. For the purposes+ of this License, Derivative Works shall not include works that remain+ separable from, or merely link (or bind by name) to the interfaces of,+ the Work and Derivative Works thereof.++ "Contribution" shall mean any work of authorship, including+ the original version of the Work and any modifications or additions+ to that Work or Derivative Works thereof, that is intentionally+ submitted to Licensor for inclusion in the Work by the copyright owner+ or by an individual or Legal Entity authorized to submit on behalf of+ the copyright owner. For the purposes of this definition, "submitted"+ means any form of electronic, verbal, or written communication sent+ to the Licensor or its representatives, including but not limited to+ communication on electronic mailing lists, source code control systems,+ and issue tracking systems that are managed by, or on behalf of, the+ Licensor for the purpose of discussing and improving the Work, but+ excluding communication that is conspicuously marked or otherwise+ designated in writing by the copyright owner as "Not a Contribution."++ "Contributor" shall mean Licensor and any individual or Legal Entity+ on behalf of whom a Contribution has been received by Licensor and+ subsequently incorporated within the Work.++ 2. Grant of Copyright License. Subject to the terms and conditions of+ this License, each Contributor hereby grants to You a perpetual,+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable+ copyright license to reproduce, prepare Derivative Works of,+ publicly display, publicly perform, sublicense, and distribute the+ Work and such Derivative Works in Source or Object form.++ 3. Grant of Patent License. Subject to the terms and conditions of+ this License, each Contributor hereby grants to You a perpetual,+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable+ (except as stated in this section) patent license to make, have made,+ use, offer to sell, sell, import, and otherwise transfer the Work,+ where such license applies only to those patent claims licensable+ by such Contributor that are necessarily infringed by their+ Contribution(s) alone or by combination of their Contribution(s)+ with the Work to which such Contribution(s) was submitted. If You+ institute patent litigation against any entity (including a+ cross-claim or counterclaim in a lawsuit) alleging that the Work+ or a Contribution incorporated within the Work constitutes direct+ or contributory patent infringement, then any patent licenses+ granted to You under this License for that Work shall terminate+ as of the date such litigation is filed.++ 4. Redistribution. You may reproduce and distribute copies of the+ Work or Derivative Works thereof in any medium, with or without+ modifications, and in Source or Object form, provided that You+ meet the following conditions:++ (a) You must give any other recipients of the Work or+ Derivative Works a copy of this License; and++ (b) You must cause any modified files to carry prominent notices+ stating that You changed the files; and++ (c) You must retain, in the Source form of any Derivative Works+ that You distribute, all copyright, patent, trademark, and+ attribution notices from the Source form of the Work,+ excluding those notices that do not pertain to any part of+ the Derivative Works; and++ (d) If the Work includes a "NOTICE" text file as part of its+ distribution, then any Derivative Works that You distribute must+ include a readable copy of the attribution notices contained+ within such NOTICE file, excluding those notices that do not+ pertain to any part of the Derivative Works, in at least one+ of the following places: within a NOTICE text file distributed+ as part of the Derivative Works; within the Source form or+ documentation, if provided along with the Derivative Works; or,+ within a display generated by the Derivative Works, if and+ wherever such third-party notices normally appear. The contents+ of the NOTICE file are for informational purposes only and+ do not modify the License. You may add Your own attribution+ notices within Derivative Works that You distribute, alongside+ or as an addendum to the NOTICE text from the Work, provided+ that such additional attribution notices cannot be construed+ as modifying the License.++ You may add Your own copyright statement to Your modifications and+ may provide additional or different license terms and conditions+ for use, reproduction, or distribution of Your modifications, or+ for any such Derivative Works as a whole, provided Your use,+ reproduction, and distribution of the Work otherwise complies with+ the conditions stated in this License.++ 5. Submission of Contributions. Unless You explicitly state otherwise,+ any Contribution intentionally submitted for inclusion in the Work+ by You to the Licensor shall be under the terms and conditions of+ this License, without any additional terms or conditions.+ Notwithstanding the above, nothing herein shall supersede or modify+ the terms of any separate license agreement you may have executed+ with Licensor regarding such Contributions.++ 6. Trademarks. This License does not grant permission to use the trade+ names, trademarks, service marks, or product names of the Licensor,+ except as required for reasonable and customary use in describing the+ origin of the Work and reproducing the content of the NOTICE file.++ 7. Disclaimer of Warranty. Unless required by applicable law or+ agreed to in writing, Licensor provides the Work (and each+ Contributor provides its Contributions) on an "AS IS" BASIS,+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or+ implied, including, without limitation, any warranties or conditions+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A+ PARTICULAR PURPOSE. You are solely responsible for determining the+ appropriateness of using or redistributing the Work and assume any+ risks associated with Your exercise of permissions under this License.++ 8. Limitation of Liability. In no event and under no legal theory,+ whether in tort (including negligence), contract, or otherwise,+ unless required by applicable law (such as deliberate and grossly+ negligent acts) or agreed to in writing, shall any Contributor be+ liable to You for damages, including any direct, indirect, special,+ incidental, or consequential damages of any character arising as a+ result of this License or out of the use or inability to use the+ Work (including but not limited to damages for loss of goodwill,+ work stoppage, computer failure or malfunction, or any and all+ other commercial damages or losses), even if such Contributor+ has been advised of the possibility of such damages.++ 9. Accepting Warranty or Additional Liability. While redistributing+ the Work or Derivative Works thereof, You may choose to offer,+ and charge a fee for, acceptance of support, warranty, indemnity,+ or other liability obligations and/or rights consistent with this+ License. However, in accepting such obligations, You may act only+ on Your own behalf and on Your sole responsibility, not on behalf+ of any other Contributor, and only if You agree to indemnify,+ defend, and hold each Contributor harmless for any liability+ incurred by, or claims asserted against, such Contributor by reason+ of your accepting any such warranty or additional liability.++ END OF TERMS AND CONDITIONS++ APPENDIX: How to apply the Apache License to your work.++ To apply the Apache License to your work, attach the following+ boilerplate notice, with the fields enclosed by brackets "[]"+ replaced with your own identifying information. (Don't include+ the brackets!) The text should be enclosed in the appropriate+ comment syntax for the file format. We also recommend that a+ file or class name and description of purpose be included on the+ same "printed page" as the copyright notice for easier+ identification within third-party archives.++ Copyright [yyyy] [name of copyright owner]++ 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++ 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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ hadoop-rpc.cabal view
@@ -0,0 +1,71 @@+name: hadoop-rpc+version: 0.1.0.0++synopsis:+ Use the Hadoop RPC interface from Haskell.++description:+ Use the Hadoop RPC interface from Haskell.+ .+ Currently we only support v7 of the RPC protocol (< CDH5).+ .+ Support for v9 (>= CDH5) is coming soon.+ .+ /The API is still evolving and is highly likely to change in the future./++homepage: http://github.com/jystic/hadoop-rpc+license: Apache-2.0+license-file: LICENSE+author: Jacob Stanley, Conrad Parker+maintainer: Jacob Stanley <jacob@stanley.io>+category: Data+build-type: Simple+cabal-version: >= 1.10++library+ hs-source-dirs: src+ default-language: Haskell2010++ ghc-options: -funbox-strict-fields++ exposed-modules:+ Data.Hadoop.Configuration+ Data.Hadoop.Protobuf.ClientNameNode+ Data.Hadoop.Protobuf.DataTransfer+ Data.Hadoop.Protobuf.Hdfs+ Data.Hadoop.Protobuf.Headers+ Data.Hadoop.Protobuf.ProtocolInfo+ Data.Hadoop.Protobuf.Security+ Data.Hadoop.Types+ Network.Hadoop.Hdfs+ Network.Hadoop.Read+ Network.Hadoop.Rpc+ Network.Hadoop.Socket+ Network.Hadoop.Stream++ build-depends:+ base >= 4.7 && < 5+ , attoparsec >= 0.12+ , bytestring >= 0.10+ , cereal >= 0.4+ , exceptions >= 0.6+ , network >= 2.5+ , protobuf >= 0.2.0.4+ , socks >= 0.5+ , text >= 1.1+ , transformers >= 0.4+ , unix >= 2.7+ , unordered-containers >= 0.2+ , vector >= 0.10+ , xmlhtml >= 0.2++executable test-hadoop-rpc+ main-is: Main.hs+ hs-source-dirs: test+ default-language: Haskell2010++ build-depends:+ base >= 4.7 && < 5+ , hadoop-rpc+ , protobuf+ , vector
+ src/Data/Hadoop/Configuration.hs view
@@ -0,0 +1,103 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Data.Hadoop.Configuration+ ( getHadoopConfig+ , getHadoopUser+ , getNameNodes+ ) where++import Control.Applicative ((<$>), (<*>))+import Control.Exception (IOException, handle)+import qualified Data.ByteString.Char8 as B+import qualified Data.HashMap.Lazy as H+import Data.Maybe (fromMaybe, mapMaybe)+import Data.Monoid ((<>))+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Read as T++import System.Environment (lookupEnv)+import System.Posix.User (getEffectiveUserName)+import Text.XmlHtml++import Data.Hadoop.Types++------------------------------------------------------------------------++getHadoopConfig :: IO HadoopConfig+getHadoopConfig = do+ hcUser <- getHadoopUser+ hcNameNodes <- getNameNodes+ let hcProxy = Nothing+ return HadoopConfig{..}++------------------------------------------------------------------------++getHadoopUser :: IO User+getHadoopUser = maybe fromUnix return =<< fromEnv+ where+ fromEnv :: IO (Maybe User)+ fromEnv = fmap T.pack <$> lookupEnv "HADOOP_USER_NAME"++ fromUnix :: IO User+ fromUnix = T.pack <$> getEffectiveUserName++------------------------------------------------------------------------++type HadoopXml = H.HashMap Text Text++getNameNodes :: IO [NameNode]+getNameNodes = do+ cfg <- H.union <$> readHadoopConfig "/etc/hadoop/conf/core-site.xml"+ <*> readHadoopConfig "/etc/hadoop/conf/hdfs-site.xml"+ return $ fromMaybe []+ $ resolveNameNode cfg <$> (stripProto =<< H.lookup fsDefaultNameKey cfg)+ where+ proto = "hdfs://"+ fsDefaultNameKey = "fs.defaultFS"+ nameNodesPrefix = "dfs.ha.namenodes."+ rpcAddressPrefix = "dfs.namenode.rpc-address."++ stripProto :: Text -> Maybe Text+ stripProto uri | proto `T.isPrefixOf` uri = Just (T.drop (T.length proto) uri)+ | otherwise = Nothing++ resolveNameNode :: HadoopXml -> Text -> [NameNode]+ resolveNameNode cfg name = case parseEndpoint name of+ Just ep -> [ep] -- contains "host:port" directly+ Nothing -> mapMaybe (\nn -> lookupAddress cfg $ name <> "." <> nn)+ (lookupNameNodes cfg name)++ lookupNameNodes :: HadoopXml -> Text -> [Text]+ lookupNameNodes cfg name = fromMaybe []+ $ T.splitOn "," <$> H.lookup (nameNodesPrefix <> name) cfg++ lookupAddress :: HadoopXml -> Text -> Maybe Endpoint+ lookupAddress cfg name = parseEndpoint =<< H.lookup (rpcAddressPrefix <> name) cfg++ parseEndpoint :: Text -> Maybe Endpoint+ parseEndpoint ep = Endpoint host <$> port+ where+ host = T.takeWhile (/= ':') ep+ port = either (const Nothing) (Just . fst)+ $ T.decimal $ T.drop (T.length host + 1) ep++readHadoopConfig :: FilePath -> IO HadoopXml+readHadoopConfig path = do+ exml <- readXML path+ case exml of+ Left _ -> return H.empty+ Right xml -> return (toHashMap (docContent xml))+ where+ toHashMap = H.fromList . mapMaybe fromNode+ . concatMap (descendantElementsTag "property")++ fromNode n = (,) <$> (nodeText <$> childElementTag "name" n)+ <*> (nodeText <$> childElementTag "value" n)++readXML :: FilePath -> IO (Either String Document)+readXML path = handle onError (parseXML path <$> B.readFile path)+ where+ onError :: IOException -> IO (Either String Document)+ onError e = return $ Left $ show e
+ src/Data/Hadoop/Protobuf/ClientNameNode.hs view
@@ -0,0 +1,152 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}+{-# OPTIONS_GHC -fcontext-stack=37 #-}++module Data.Hadoop.Protobuf.ClientNameNode where++import Data.ByteString (ByteString)+import Data.ProtocolBuffers+import Data.Text (Text)+import Data.Int (Int64)+import Data.Word (Word64)+import GHC.Generics (Generic)++import Data.Hadoop.Protobuf.Hdfs++------------------------------------------------------------------------++data GetListingRequest = GetListingRequest+ { lsSrc :: Required 1 (Value Text) -- ^ the directory to list+ , lsStartAfter :: Required 2 (Value ByteString) -- ^ begin the listing after this file (encoded java utf8)+ , lsNeedLocation :: Required 3 (Value Bool) -- ^ return the location data in the cluster+ } deriving (Generic, Show)++instance Encode GetListingRequest+instance Decode GetListingRequest++data GetListingResponse = GetListingResponse+ { lsDirList :: Optional 1 (Message DirectoryListing)+ } deriving (Generic, Show)++instance Encode GetListingResponse+instance Decode GetListingResponse++------------------------------------------------------------------------++data GetFileInfoRequest = GetFileInfoRequest+ { fiSrc :: Required 1 (Value Text)+ } deriving (Generic, Show)++instance Encode GetFileInfoRequest+instance Decode GetFileInfoRequest++data GetFileInfoResponse = GetFileInfoResponse+ { fiFileStatus :: Optional 1 (Message FileStatus)+ } deriving (Generic, Show)++instance Encode GetFileInfoResponse+instance Decode GetFileInfoResponse++------------------------------------------------------------------------++data GetContentSummaryRequest = GetContentSummaryRequest+ { csPath :: Required 1 (Value Text)+ } deriving (Generic, Show)++instance Encode GetContentSummaryRequest+instance Decode GetContentSummaryRequest++data GetContentSummaryResponse = GetContentSummaryResponse+ { csSummary :: Required 1 (Message ContentSummary)+ } deriving (Generic, Show)++instance Encode GetContentSummaryResponse+instance Decode GetContentSummaryResponse++------------------------------------------------------------------------++data MkdirsRequest = MkdirsRequest+ { mdSrc :: Required 1 (Value Text)+ , mdMasked :: Required 2 (Message FilePermission)+ , mdCreateParent :: Required 3 (Value Bool)+ } deriving (Generic, Show)++instance Encode MkdirsRequest+instance Decode MkdirsRequest++data MkdirsResponse = MkdirsResponse+ { mdResult :: Required 1 (Value Bool)+ } deriving (Generic, Show)++instance Encode MkdirsResponse+instance Decode MkdirsResponse++------------------------------------------------------------------------++data DeleteRequest = DeleteRequest+ { dlSrc :: Required 1 (Value Text)+ , dlRecursive :: Required 2 (Value Bool)+ } deriving (Generic, Show)++instance Encode DeleteRequest+instance Decode DeleteRequest++data DeleteResponse = DeleteResponse+ { dlResult :: Required 1 (Value Bool)+ } deriving (Generic, Show)++instance Encode DeleteResponse+instance Decode DeleteResponse++------------------------------------------------------------------------++data Rename2Request = Rename2Request+ { mvSrc :: Required 1 (Value Text)+ , mvDst :: Required 2 (Value Text)+ , mvOverwrite :: Required 3 (Value Bool)+ } deriving (Generic, Show)++instance Encode Rename2Request+instance Decode Rename2Request++data Rename2Response = Rename2Response+ {+ } deriving (Generic, Show)++instance Encode Rename2Response+instance Decode Rename2Response++------------------------------------------------------------------------++data SetPermissionRequest = SetPermissionRequest+ { chmodPath :: Required 1 (Value Text)+ , chmodMode :: Required 2 (Message FilePermission)+ } deriving (Generic, Show)++instance Encode SetPermissionRequest+instance Decode SetPermissionRequest++data SetPermissionResponse = SetPermissionResponse+ {+ } deriving (Generic, Show)++instance Encode SetPermissionResponse+instance Decode SetPermissionResponse++------------------------------------------------------------------------++data GetBlockLocationsRequest = GetBlockLocationsRequest+ { catSrc :: Required 1 (Value Text)+ , catOffset :: Required 2 (Value Int64)+ , catLength :: Required 3 (Value Int64)+ } deriving (Generic, Show)++instance Encode GetBlockLocationsRequest+instance Decode GetBlockLocationsRequest++data GetBlockLocationsResponse = GetBlockLocationsResponse+ { locsLocations :: Optional 1 (Message LocatedBlocks)+ } deriving (Generic, Show)++instance Encode GetBlockLocationsResponse+instance Decode GetBlockLocationsResponse
+ src/Data/Hadoop/Protobuf/DataTransfer.hs view
@@ -0,0 +1,319 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}+{-# OPTIONS_GHC -fcontext-stack=37 #-}++-- | This module contains protocol buffers that are used to transfer data+-- to and from the datanode, as well as between datanodes.+module Data.Hadoop.Protobuf.DataTransfer where++import Data.ByteString (ByteString)+import Data.Int (Int32, Int64)+import Data.ProtocolBuffers+import Data.ProtocolBuffers.Orphans ()+import Data.Text (Text)+import Data.Word (Word32, Word64)+import GHC.Generics (Generic)++import Data.Hadoop.Protobuf.Hdfs+import Data.Hadoop.Protobuf.Security++------------------------------------------------------------------------++data DataTransferEncryptorStatus =+ DTES_SUCCESS | DTES_ERROR_UNKNOWN_KEY | DTES_ERROR+ deriving (Generic, Show, Eq)++instance Enum DataTransferEncryptorStatus where+ toEnum n = case n of+ 0 -> DTES_SUCCESS+ 1 -> DTES_ERROR_UNKNOWN_KEY+ 2 -> DTES_ERROR+ _ -> error $ "DataTransferEncryptorStatus.toEnum: invalid value <" ++ show n ++ ">"++ fromEnum e = case e of+ DTES_SUCCESS -> 0+ DTES_ERROR_UNKNOWN_KEY -> 1+ DTES_ERROR -> 2++instance Encode DataTransferEncryptorStatus+instance Decode DataTransferEncryptorStatus++data DataTransferEncryptorMessage = DataTransferEncryptorMessage+ { dtesStatus :: Required 1 (Enumeration DataTransferEncryptorStatus)+ , dtesPayload :: Optional 2 (Value ByteString)+ , dtesData :: Optional 3 (Value Text)+ } deriving (Generic, Show)++instance Encode DataTransferEncryptorMessage+instance Decode DataTransferEncryptorMessage++data BaseHeader = BaseHeader+ { bhBlock :: Required 1 (Message ExtendedBlock)+ , bhToken :: Optional 2 (Message Token)+ } deriving (Generic, Show)++instance Encode BaseHeader+instance Decode BaseHeader++data ClientOperationHeader = ClientOperationHeader+ { cohBaseHeader :: Required 1 (Message BaseHeader)+ , cohClientName :: Required 2 (Value Text)+ } deriving (Generic, Show)++instance Encode ClientOperationHeader+instance Decode ClientOperationHeader++data CachingStrategy = CachingStrategy+ { csDropBehind :: Optional 1 (Value Bool)+ , csReadahead :: Optional 2 (Value Int64)+ } deriving (Generic, Show)++instance Encode CachingStrategy+instance Decode CachingStrategy++data OpReadBlock = OpReadBlock+ { orbHeader :: Required 1 (Message ClientOperationHeader)+ , orbOffset :: Required 2 (Value Word64)+ , orbLen :: Required 3 (Value Word64)+ , orbSendChecksums :: Optional 4 (Value Bool)+ , orbCachingStrategy :: Optional 5 (Message CachingStrategy)+ } deriving (Generic, Show)++instance Encode OpReadBlock+instance Decode OpReadBlock++data Checksum = Checksum+ { csType :: Required 1 (Message ChecksumType)+ , csBytesPerChecksum :: Required 2 (Value Word32)+ } deriving (Generic, Show)++instance Encode Checksum+instance Decode Checksum+++data BlockConstructionStage =+ BCS_PIPELINE_SETUP_APPEND+ -- | pipeline set up for failed PIPELINE_SETUP_APPEND recovery+ | BCS_PIPELINE_SETUP_APPEND_RECOVERY+ -- | data streaming+ | BCS_DATA_STREAMING+ -- | pipeline setup for failed data streaming recovery+ | BCS_PIPELINE_SETUP_STREAMING_RECOVERY+ -- | close the block and pipeline+ | BCS_PIPELINE_CLOSE+ -- | Recover a failed PIPELINE_CLOSE+ | BCS_PIPELINE_CLOSE_RECOVERY+ -- | pipeline set up for block creation+ | BCS_PIPELINE_SETUP_CREATE+ -- | transfer RBW for adding datanodes+ | BCS_TRANSFER_RBW+ -- | transfer Finalized for adding datanodes+ | BCS_TRANSFER_FINALIZED+ deriving (Generic, Show, Eq)++instance Enum BlockConstructionStage where+ toEnum n = case n of+ 0 -> BCS_PIPELINE_SETUP_APPEND+ 1 -> BCS_PIPELINE_SETUP_APPEND_RECOVERY+ 2 -> BCS_DATA_STREAMING+ 3 -> BCS_PIPELINE_SETUP_STREAMING_RECOVERY+ 4 -> BCS_PIPELINE_CLOSE+ 5 -> BCS_PIPELINE_CLOSE_RECOVERY+ 6 -> BCS_PIPELINE_SETUP_CREATE+ 7 -> BCS_TRANSFER_RBW+ 8 -> BCS_TRANSFER_FINALIZED+ _ -> error $ "BlockConstructionStage.toEnum: invalid enum value <" ++ show n ++ ">"++ fromEnum e = case e of+ BCS_PIPELINE_SETUP_APPEND -> 0+ BCS_PIPELINE_SETUP_APPEND_RECOVERY -> 1+ BCS_DATA_STREAMING -> 2+ BCS_PIPELINE_SETUP_STREAMING_RECOVERY -> 3+ BCS_PIPELINE_CLOSE -> 4+ BCS_PIPELINE_CLOSE_RECOVERY -> 5+ BCS_PIPELINE_SETUP_CREATE -> 6+ BCS_TRANSFER_RBW -> 7+ BCS_TRANSFER_FINALIZED -> 8++instance Encode BlockConstructionStage+instance Decode BlockConstructionStage++data OpWriteBlock = OpWriteBlock+ { owbHeader :: Required 1 (Message ClientOperationHeader)+ , owbTargets :: Repeated 2 (Message DataNodeInfo)+ , owbSource :: Optional 3 (Message DataNodeInfo)+ , owbStage :: Required 4 (Enumeration BlockConstructionStage)+ , owbPipelineSize :: Required 5 (Value Word32)+ , owbMinBytesRcvd :: Required 6 (Value Word64)+ , owbMaxBytesRcvd :: Required 7 (Value Word64)+ , owbLatestGenerationStamp :: Required 8 (Value Word64)++ -- The requested checksum mechanism for this block write.+ , owbRequestedChecksum :: Required 9 (Message Checksum)+ , owbCachingStrategy :: Optional 10 (Message CachingStrategy)+ } deriving (Generic, Show)++instance Encode OpWriteBlock+instance Decode OpWriteBlock++data OpTransferBlock = OpTransferBlock+ { otbHeader :: Required 1 (Message ClientOperationHeader)+ , otbTargets :: Repeated 2 (Message DataNodeInfo)+ } deriving (Generic, Show)++instance Encode OpTransferBlock+instance Decode OpTransferBlock++data OpReplaceBlock = OpReplaceBlock+ { orpHeader :: Required 1 (Message BaseHeader)+ , orpDelHint :: Required 2 (Value Text)+ , orpSource :: Required 3 (Message DataNodeInfo)+ } deriving (Generic, Show)++instance Encode OpReplaceBlock+instance Decode OpReplaceBlock++data OpCopyBlock = OpCopyBlock+ { ocbHeader :: Required 1 (Message BaseHeader)+ } deriving (Generic, Show)++instance Encode OpCopyBlock+instance Decode OpCopyBlock++data OpBlockChecksum = OpBlockChecksum+ { obcHeader :: Required 1 (Message BaseHeader)+ } deriving (Generic, Show)++instance Encode OpBlockChecksum+instance Decode OpBlockChecksum++data OpRequestShortCircuitAccess = OpRequestShortCircuitAccess+ { orscHeader :: Required 1 (Message BaseHeader)++ -- | In order to get short-circuit access to block data, clients must+ -- set this to the highest version of the block data that they can+ -- understand. Currently 1 is the only version, but more versions+ -- may exist in the future if the on-disk format changes.+ , orscMaxVersion :: Required 2 (Value Word32)+ } deriving (Generic, Show)++instance Encode OpRequestShortCircuitAccess+instance Decode OpRequestShortCircuitAccess++-- All fields must be fixed-length!+data PacketHeader = PacketHeader+ { phOffsetInBlock :: Required 1 (Value (Fixed Int64))+ , phSeqno :: Required 2 (Value (Fixed Int64))+ , phLastPacketInBlock :: Required 3 (Value Bool)+ , phDataLen :: Required 4 (Value (Fixed Int32))+ , phSyncBlock :: Optional 5 (Value Bool) -- ^ default = false+ } deriving (Generic, Show)++instance Encode PacketHeader+instance Decode PacketHeader++data DataTransferStatus =+ DT_SUCCESS+ | DT_ERROR+ | DT_ERROR_CHECKSUM+ | DT_ERROR_INVALID+ | DT_ERROR_EXISTS+ | DT_ERROR_ACCESS_TOKEN+ | DT_CHECKSUM_OK+ | DT_ERROR_UNSUPPORTED+ deriving (Generic, Show, Eq)++instance Enum DataTransferStatus where+ toEnum n = case n of+ 0 -> DT_SUCCESS+ 1 -> DT_ERROR+ 2 -> DT_ERROR_CHECKSUM+ 3 -> DT_ERROR_INVALID+ 4 -> DT_ERROR_EXISTS+ 5 -> DT_ERROR_ACCESS_TOKEN+ 6 -> DT_CHECKSUM_OK+ 7 -> DT_ERROR_UNSUPPORTED+ _ -> error $ "DataTransferStatus.toEnum: invalid enum value <" ++ show n ++ ">"++ fromEnum e = case e of+ DT_SUCCESS -> 0+ DT_ERROR -> 1+ DT_ERROR_CHECKSUM -> 2+ DT_ERROR_INVALID -> 3+ DT_ERROR_EXISTS -> 4+ DT_ERROR_ACCESS_TOKEN -> 5+ DT_CHECKSUM_OK -> 6+ DT_ERROR_UNSUPPORTED -> 7++data PipelineAck = PipelineAck+ { psSeqno :: Required 1 (Value (Signed Int64))+ -- TODO paStatus should be a repeated field but there is a bug in protobuf+ -- TODO with repeated enumeration's+ , paStatus :: Optional 2 (Enumeration DataTransferStatus)+ , paDownstreamAckTimeNanos :: Optional 3 (Value Word64) -- ^ default = 0+ } deriving (Generic, Show)++instance Encode PipelineAck+instance Decode PipelineAck++-- | Sent as part of the BlockOpResponse+-- for READ_BLOCK and COPY_BLOCK operations.+data ReadOpChecksumInfo = ReadOpChecksumInfo+ { rociChecksum :: Required 1 (Message Checksum)++ -- | The offset into the block at which the first packet+ -- will start. This is necessary since reads will align+ -- backwards to a checksum chunk boundary.+ , rociChunkOffset :: Required 2 (Value Word64)+ } deriving (Generic, Show)++instance Encode ReadOpChecksumInfo+instance Decode ReadOpChecksumInfo++data BlockOpResponse = BlockOpResponse+ { borStatus :: Required 1 (Enumeration DataTransferStatus)+ , borFirstBadLink :: Optional 2 (Value String)+ , borChecksumResponse :: Optional 3 (Message OpBlockChecksumResponse)+ , borReadOpChecksumInfo :: Optional 4 (Message ReadOpChecksumInfo)++ -- | Explanatory text which may be useful to log on the client side+ , borData :: Optional 5 (Value Text)++ -- | If the server chooses to agree to the request of a client for+ -- short-circuit access, it will send a response data with the relevant+ -- file descriptors attached.+ --+ -- In the body of the data, this version number will be set to the+ -- specific version number of the block data that the client is about to+ -- read.+ , borShortCircuitAccessVersion :: Optional 6 (Value Word32)+ } deriving (Generic, Show)++instance Encode BlockOpResponse+instance Decode BlockOpResponse++-- | Message sent from the client to the DN after reading the entire+-- read request.+data ClientReadStatus = ClientReadStatus+ { crsStatus :: Required 1 (Enumeration DataTransferStatus)+ } deriving (Generic, Show)++instance Encode ClientReadStatus+instance Decode ClientReadStatus++data DNTransferAck = DNTransferAck+ { dntaStatus :: Required 1 (Enumeration DataTransferStatus)+ } deriving (Generic, Show)++instance Encode DNTransferAck+instance Decode DNTransferAck++data OpBlockChecksumResponse = OpBlockChecksumResponse+ { obcrBytesPerCrc :: Required 1 (Value Word32)+ , obcrCrcPerBlock :: Required 2 (Value Word64)+ , obcrMd5 :: Required 3 (Value ByteString)+ , obcrCrcType :: Optional 4 (Message ChecksumType)+ } deriving (Generic, Show)++instance Encode OpBlockChecksumResponse+instance Decode OpBlockChecksumResponse
+ src/Data/Hadoop/Protobuf/Hdfs.hs view
@@ -0,0 +1,211 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}++module Data.Hadoop.Protobuf.Hdfs where++import Data.ByteString (ByteString)+import Data.ProtocolBuffers+import Data.ProtocolBuffers.Orphans ()+import Data.Text (Text)+import Data.Word (Word32, Word64)+import GHC.Generics (Generic)++import Data.Hadoop.Protobuf.Security++------------------------------------------------------------------------++-- | Summary of a file or directory+data ContentSummary = ContentSummary+ { csLength :: Required 1 (Value Word64)+ , csFileCount :: Required 2 (Value Word64)+ , csDirectoryCount :: Required 3 (Value Word64)+ , csQuota :: Required 4 (Value Word64)+ , csSpaceConsumed :: Required 5 (Value Word64)+ , csSpaceQuota :: Required 6 (Value Word64)+ } deriving (Generic, Show)++instance Encode ContentSummary+instance Decode ContentSummary++------------------------------------------------------------------------++-- | Directory listing.+data DirectoryListing = DirectoryListing+ { dlPartialListing :: Repeated 1 (Message FileStatus)+ , dlRemaingEntries :: Required 2 (Value Word32)+ } deriving (Generic, Show)++instance Encode DirectoryListing+instance Decode DirectoryListing++------------------------------------------------------------------------++-- | Status of a file, directory or symbolic link. Optionally includes a+-- files block locations if requested by client on the RPC call.+data FileStatus = FileStatus+ { fsFileType :: Required 1 (Enumeration FileType)+ , fsPath :: Required 2 (Value ByteString) -- ^ local name of inode (encoded java utf8)+ , fsLength :: Required 3 (Value Word64)+ , fsPermission :: Required 4 (Message FilePermission)+ , fsOwner :: Required 5 (Value Text)+ , fsGroup :: Required 6 (Value Text)+ , fsModificationTime :: Required 7 (Value Word64)+ , fsAccessTime :: Required 8 (Value Word64)++ -- Optional fields for symlink+ , fsSymLink :: Optional 9 (Value ByteString) -- ^ if symlink, target (encoded java utf8)++ -- Optional fields for file+ , fsBlockReplication :: Optional 10 (Value Word32) -- ^ default = 0, only 16bits used+ , fsBlockSize :: Optional 11 (Value Word64) -- ^ default = 0+ , fsLocations :: Optional 12 (Message LocatedBlocks) -- ^ supplied only if asked by client+ } deriving (Generic, Show)++instance Encode FileStatus+instance Decode FileStatus++------------------------------------------------------------------------++-- | The type of a file (either directory, file or symbolic link)+data FileType = IS_DIR | IS_FILE | IS_SYMLINK+ deriving (Generic, Show, Eq)++instance Enum FileType where+ toEnum n = case n of+ 1 -> IS_DIR+ 2 -> IS_FILE+ 3 -> IS_SYMLINK+ _ -> error $ "FileType.toEnum: invalid enum value <" ++ show n ++ ">"++ fromEnum e = case e of+ IS_DIR -> 1+ IS_FILE -> 2+ IS_SYMLINK -> 3++------------------------------------------------------------------------++data ChecksumType = CHECKSUM_NULL | CHECKSUM_CRC32 | CHECKSUM_CRC32C+ deriving (Generic, Show, Eq)++instance Enum ChecksumType where+ toEnum n = case n of+ 0 -> CHECKSUM_NULL+ 1 -> CHECKSUM_CRC32+ 2 -> CHECKSUM_CRC32C+ _ -> error $ "ChecksumType.toEnum: invalid enum value <" ++ show n ++ ">"++ fromEnum e = case e of+ CHECKSUM_NULL -> 0+ CHECKSUM_CRC32 -> 1+ CHECKSUM_CRC32C -> 2++instance Encode ChecksumType+instance Decode ChecksumType++------------------------------------------------------------------------++-- | File or directory permission, same spec as POSIX.+data FilePermission = FilePermission+ { fpPerm :: Required 1 (Value Word32) -- ^ actually a short, only 16 bits used+ } deriving (Generic, Show)++instance Encode FilePermission+instance Decode FilePermission++------------------------------------------------------------------------++-- | A set of file blocks and their locations.+data LocatedBlocks = LocatedBlocks+ { lbFileLength :: Required 1 (Value Word64)+ , lbBlocks :: Repeated 2 (Message LocatedBlock)+ , lbUnderConstruction :: Required 3 (Value Bool)+ , lbLastBlock :: Optional 4 (Message LocatedBlock)+ , lbLastBlockComplete :: Required 5 (Value Bool)+ } deriving (Generic, Show)++instance Encode LocatedBlocks+instance Decode LocatedBlocks++------------------------------------------------------------------------++-- | Information about a block and its location.+data LocatedBlock = LocatedBlock+ { lbExtended :: Required 1 (Message ExtendedBlock)+ , lbOffset :: Required 2 (Value Word64) -- ^ offset of first byte of block in the file+ , lbLocations :: Repeated 3 (Message DataNodeInfo) -- ^ locations ordered by proximity to client IP++ -- | `True` if all replicas of a block are corrupt. If the block has a few corrupt replicas,+ -- they are filtered and their locations are not part of this object.+ , lbCorrupt :: Required 4 (Value Bool)+ , lbToken :: Required 5 (Message Token)+ } deriving (Generic, Show)++instance Encode LocatedBlock+instance Decode LocatedBlock++------------------------------------------------------------------------++-- | Identifies a block.+data ExtendedBlock = ExtendedBlock+ { ebPoolId :: Required 1 (Value Text) -- ^ block pool id - globally unique across clusters+ , ebBlockId :: Required 2 (Value Word64) -- ^ the local id within a pool+ , ebGenerationStamp :: Required 3 (Value Word64)+ , ebNumBytes :: Optional 4 (Value Word64) -- ^ does not belong, here for historical reasons+ } deriving (Generic, Show)++instance Encode ExtendedBlock+instance Decode ExtendedBlock++------------------------------------------------------------------------++-- | Status of a data node.+data DataNodeInfo = DataNodeInfo+ { dnId :: Required 1 (Message DataNodeId)+ , dnCapacity :: Optional 2 (Value Word64) -- ^ default = 0+ , dnDfsUsed :: Optional 3 (Value Word64) -- ^ default = 0+ , dnRemaining :: Optional 4 (Value Word64) -- ^ default = 0+ , dnBlockPoolUsed :: Optional 5 (Value Word64) -- ^ default = 0+ , dnLastUpdate :: Optional 6 (Value Word64) -- ^ default = 0+ , dnXceiverCount :: Optional 7 (Value Word32) -- ^ default = 0+ , dnLocation :: Optional 8 (Value Text)+ , dnAdminState :: Optional 9 (Enumeration AdminState) -- ^ default = NORMAL+ } deriving (Generic, Show)++instance Encode DataNodeInfo+instance Decode DataNodeInfo++------------------------------------------------------------------------++data AdminState = NORMAL | DECOMMISSION_INPROGRESS | DECOMMISSIONED+ deriving (Generic, Show, Eq)++instance Enum AdminState where+ toEnum n = case n of+ 0 -> NORMAL+ 1 -> DECOMMISSION_INPROGRESS+ 2 -> DECOMMISSIONED+ _ -> error $ "AdminState.toEnum: invalid enum value <" ++ show n ++ ">"++ fromEnum e = case e of+ NORMAL -> 0+ DECOMMISSION_INPROGRESS -> 1+ DECOMMISSIONED -> 2++------------------------------------------------------------------------++-- | Identifies a data node+data DataNodeId = DataNodeId+ { dnIpAddr :: Required 1 (Value Text) -- ^ IP address+ , dnHostName :: Required 2 (Value Text) -- ^ Host name+ , dnStorageId :: Required 3 (Value Text) -- ^ Storage ID+ , dnXferPort :: Required 4 (Value Word32) -- ^ Data streaming port+ , dnInfoPort :: Required 5 (Value Word32) -- ^ Info server port+ , dnIpcPort :: Required 6 (Value Word32) -- ^ IPC server port+ } deriving (Generic, Show)++instance Encode DataNodeId+instance Decode DataNodeId++------------------------------------------------------------------------++-- BlockTokenId was moved to Security.Token in protocol version 9
+ src/Data/Hadoop/Protobuf/Headers.hs view
@@ -0,0 +1,170 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}++module Data.Hadoop.Protobuf.Headers where++import Data.ByteString (ByteString)+import Data.ProtocolBuffers+import Data.ProtocolBuffers.Orphans ()+import Data.Text (Text)+import Data.Word (Word32, Word64)+import GHC.Generics (Generic)++------------------------------------------------------------------------++-- TODO From Hadoop source:+-- Spec for UserInformationProto is specified in ProtoUtil#makeIpcConnectionContext++-- | User information beyond what can be determined as part of the+-- security handshake at connection time (kerberos, tokens, etc)+data UserInformation = UserInformation+ { effectiveUser :: Optional 1 (Value Text)+ , realUser :: Optional 2 (Value Text)+ } deriving (Generic, Show)++instance Encode UserInformation+instance Decode UserInformation++-- | The connection context is sent as part of the connection+-- establishment. It establishes the context for ALL Rpc calls+-- within the connection.+data IpcConnectionContext = IpcConnectionContext+ { ctxUserInfo :: Optional 2 (Message UserInformation)+ , ctxProtocol :: Optional 3 (Value Text) -- ^ Name of the next RPC layer.+ } deriving (Generic, Show)++instance Encode IpcConnectionContext+instance Decode IpcConnectionContext++------------------------------------------------------------------------++-- | Determines the RPC Engine and the serialization of the RPC request.+data RpcKind = Builtin -- ^ Used for built-in calls by tests+ | Writable -- ^ Use WritableRpcEngine+ | ProtocolBuffer -- ^ Use ProtobufRpcEngine+ deriving (Generic, Show, Enum)++data RpcOperation = FinalPacket -- ^ The final RPC packet+ | ContinuationPacket -- ^ Not implemented yet+ | CloseConnection -- ^ Close the RPC connection+ deriving (Generic, Show, Enum)++data RpcRequestHeader = RpcRequestHeader+ { reqKind :: Optional 1 (Enumeration RpcKind)+ , reqOp :: Optional 2 (Enumeration RpcOperation)+ , reqCallId :: Required 3 (Value Word32) -- ^ A sequence number that is sent back in the response++ -- Fields below don't apply until v9+ --, reqClientId :: Required 4 (Value ByteString) -- ^ Globally unique client ID+ --, reqRetryCount :: Optional 5 (Value Int32) -- ^ Retry count, 1 means this is the first retry+ } deriving (Generic, Show)++instance Encode RpcRequestHeader+instance Decode RpcRequestHeader++-- | This message is used for Protobuf RPC Engine.+-- The message is used to marshal a RPC request from RPC client to the+-- RPC server. The response to the RPC call (including errors) are handled+-- as part of the standard RPC response.+data RpcRequest = RpcRequest+ { reqMethodName :: Required 1 (Value Text) -- ^ Name of the RPC method+ , reqBytes :: Optional 2 (Value ByteString) -- ^ Bytes corresponding to the client protobuf request+ , reqProtocolName :: Required 3 (Value Text) -- ^ Protocol name of class declaring the called method+ , reqProtocolVersion :: Required 4 (Value Word64) -- ^ Protocol version of class declaring the called method+ } deriving (Generic, Show)++instance Encode RpcRequest+instance Decode RpcRequest++------------------------------------------------------------------------++-- | Success or failure. The reponse header's error detail, exception+-- class name and error message contains further details on the error.+data RpcStatus = Success -- ^ Succeeded+ | Error -- ^ Non-fatal error, connection left open+ | Fatal -- ^ Fatal error, connection closed+ deriving (Generic, Show, Eq, Enum)++-- | Note that RPC response header is also used when connection setup fails.+-- (i.e. the response looks like an RPC response with a fake callId)+--+-- For v7:+-- - If successfull then the Respose follows after this header+-- - length (4 byte int), followed by the response+-- - If error or fatal - the exception info follow+-- - length (4 byte int) Class name of exception - UTF-8 string+-- - length (4 byte int) Stacktrace - UTF-8 string+-- - if the strings are null then the length is -1+--+-- In case of Fatal error then the respose contains the Serverside's IPC version.++data RpcResponseHeader = RpcResponseHeader+ { rspCallId :: Required 1 (Value Word32) -- ^ Call ID used in request+ , rspStatus :: Required 2 (Enumeration RpcStatus)+ , rspServerIpcVersion :: Optional 3 (Value Word32) -- ^ v7: Sent if fatal v9: Sent if success or fail++ -- Fields below don't apply until v9+ --, rspExceptionClassName :: Optional 4 (Value Text) -- ^ If the request fails+ --, rspErrorMsg :: Optional 5 (Value Text) -- ^ If the request fails, often contains stack trace+ --, rspErrorDetail :: Optional 6 (Enumeration Error) -- ^ In case of error+ --, rspClientId :: Optional 7 (Value ByteString) -- ^ Globally unique client ID+ --, rspRetryCount :: Optional 8 (Value Int32)+ } deriving (Generic, Show)++instance Encode RpcResponseHeader+instance Decode RpcResponseHeader++{-+-- Error doesn't apply until v9++-- | Describes why an RPC error occurred.+data Error = ErrorApplication -- ^ RPC failed - RPC app threw exception+ | ErrorNoSuchMethod -- ^ RPC error - no such method+ | ErrorNoSuchProtocol -- ^ RPC error - no such protocol+ | ErrorRpcServer -- ^ RPC error on server side+ | ErrorSerializingResponse -- ^ Error serializing response+ | ErrorRpcVersionMismatch -- ^ RPC protocol version mismatch+ | ErrorCode Int -- ^ RPC error that we don't know about++ -- starts at 10+ | FatalUnknown -- ^ Unknown fatal error+ | FatalUnsupportedSerialization -- ^ IPC layer serilization type invalid+ | FatalInvalidRpcHeader -- ^ Fields of RPC header are invalid+ | FatalDeserializingRequest -- ^ Could not deserialize RPC request+ | FatalVersionMismatch -- ^ IPC layer version mismatch+ | FatalUnauthorized -- ^ Auth failed+ | FatalCode Int -- ^ Fatal error that we don't know about+ deriving (Generic, Show)++instance Enum Error where+ toEnum n = case n of+ 1 -> ErrorApplication+ 2 -> ErrorNoSuchMethod+ 3 -> ErrorNoSuchProtocol+ 4 -> ErrorRpcServer+ 5 -> ErrorSerializingResponse+ 6 -> ErrorRpcVersionMismatch+ 10 -> FatalUnknown+ 11 -> FatalUnsupportedSerialization+ 12 -> FatalInvalidRpcHeader+ 13 -> FatalDeserializingRequest+ 14 -> FatalVersionMismatch+ 15 -> FatalUnauthorized+ _ -> if n < 10 then ErrorCode n else FatalCode n++ fromEnum e = case e of+ ErrorApplication -> 1+ ErrorNoSuchMethod -> 2+ ErrorNoSuchProtocol -> 3+ ErrorRpcServer -> 4+ ErrorSerializingResponse -> 5+ ErrorRpcVersionMismatch -> 6+ ErrorCode n -> n+ FatalUnknown -> 10+ FatalUnsupportedSerialization -> 11+ FatalInvalidRpcHeader -> 12+ FatalDeserializingRequest -> 13+ FatalVersionMismatch -> 14+ FatalUnauthorized -> 15+ FatalCode n -> n+-}
+ src/Data/Hadoop/Protobuf/ProtocolInfo.hs view
@@ -0,0 +1,81 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}++module Data.Hadoop.Protobuf.ProtocolInfo where++import Data.ProtocolBuffers+import Data.Text (Text)+import Data.Word (Word32, Word64)+import GHC.Generics (Generic)++------------------------------------------------------------------------++-- | Request to get protocol versions for all supported rpc kinds.+data GetProtocolVersionsRequest = GetProtocolVersionsRequest+ { pvProtocol :: Required 1 (Value Text) -- ^ Protocol name+ } deriving (Generic, Show)++instance Encode GetProtocolVersionsRequest+instance Decode GetProtocolVersionsRequest++-- | Protocol version with corresponding RPC kind.+data ProtocolVersion = ProtocolVersion+ { pvRpcKind :: Required 1 (Value Text) -- ^ RPC kind+ , pvVersions :: Repeated 2 (Value Word64) -- ^ Protocol version corresponding to the rpc kind+ } deriving (Generic, Show)++instance Encode ProtocolVersion+instance Decode ProtocolVersion++-- | Get protocol version response.+data GetProtocolVersionsResponse = GetProtocolVersionsResponse+ { pvProtocolVersions :: Repeated 1 (Message ProtocolVersion)+ } deriving (Generic, Show)++instance Encode GetProtocolVersionsResponse+instance Decode GetProtocolVersionsResponse++------------------------------------------------------------------------++-- | Get protocol signature request+data GetProtocolSignatureRequest = GetProtocolSignatureRequest+ { psProtocol :: Required 1 (Value Text) -- ^ Protocol name+ , psRpcKind :: Required 2 (Value Text) -- ^ RPC kind+ } deriving (Generic, Show)++instance Encode GetProtocolSignatureRequest+instance Decode GetProtocolSignatureRequest++-- | Get protocol signature response+data GetProtocolSignatureResponse = GetProtocolSignatureResponse+ { psSignatures :: Repeated 1 (Message ProtocolSignature)+ } deriving (Generic, Show)++instance Encode GetProtocolSignatureResponse+instance Decode GetProtocolSignatureResponse++data ProtocolSignature = ProtocolSignature+ { psVersion :: Required 1 (Value Word64)+ , psMethods :: Repeated 2 (Value Word32)+ } deriving (Generic, Show)++instance Encode ProtocolSignature+instance Decode ProtocolSignature++------------------------------------------------------------------------++{-+Example Request+===============++ req = RpcRequest+ { reqMethodName = putField "getProtocolSignature"+ , reqBytes = putField $ Just $ toBytes GetProtocolSignatureRequest+ { psProtocol = putField "org.apache.hadoop.hdfs.protocol.ClientProtocol"+ , psRpcKind = putField "RPC_PROTOCOL_BUFFER"+ }+ , reqProtocolName = putField "org.apache.hadoop.ipc.ProtocolMetaInfoPB"+ , reqProtocolVersion = putField 1+ }++-}
+ src/Data/Hadoop/Protobuf/Security.hs view
@@ -0,0 +1,66 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveGeneric #-}++module Data.Hadoop.Protobuf.Security where++import Data.ByteString (ByteString)+import Data.ProtocolBuffers+import Data.ProtocolBuffers.Orphans ()+import Data.Text (Text)+import Data.Word (Word64)+import GHC.Generics (Generic)++------------------------------------------------------------------------++-- | Security token identifier+data Token = Token+ { tokenIdentifier :: Required 1 (Value ByteString)+ , tokenPassword :: Required 2 (Value ByteString)+ , tokenKind :: Required 3 (Value Text)+ , tokenService :: Required 4 (Value Text)+ } deriving (Generic, Show)++instance Encode Token+instance Decode Token++data GetDelegationTokenRequest = GetDelegationTokenRequest+ { gdtRenewer :: Required 1 (Value Text)+ } deriving (Generic, Show)++instance Encode GetDelegationTokenRequest+instance Decode GetDelegationTokenRequest++data GetDelegationTokenResponse = GetDelegationTokenResponse+ { gdtToken :: Optional 1 (Message Token)+ } deriving (Generic, Show)++instance Encode GetDelegationTokenResponse+instance Decode GetDelegationTokenResponse++data RenewDelegationTokenRequest = RenewDelegationTokenRequest+ { rdtToken :: Required 1 (Message Token)+ } deriving (Generic, Show)++instance Encode RenewDelegationTokenRequest+instance Decode RenewDelegationTokenRequest++data RenewDelegationTokenResponse = RenewDelegationTokenResponse+ { rtdNewExpiryTime :: Required 1 (Value Word64)+ } deriving (Generic, Show)++instance Encode RenewDelegationTokenResponse+instance Decode RenewDelegationTokenResponse++data CancelDelegationTokenRequest = CancelDelegationTokenRequest+ { cdtToken :: Required 1 (Message Token)+ } deriving (Generic, Show)++instance Encode CancelDelegationTokenRequest+instance Decode CancelDelegationTokenRequest++data CancelDelegationTokenResponse = CancelDelegationTokenResponse+ { -- void response+ } deriving (Generic, Show)++instance Encode CancelDelegationTokenResponse+instance Decode CancelDelegationTokenResponse
+ src/Data/Hadoop/Types.hs view
@@ -0,0 +1,90 @@+{-# LANGUAGE DeriveDataTypeable #-}++module Data.Hadoop.Types where++import Control.Exception (Exception)+import Data.ByteString (ByteString)+import Data.Data (Data)+import Data.Text (Text)+import Data.Typeable (Typeable)+import Data.Word (Word16, Word64)++------------------------------------------------------------------------++type NameNode = Endpoint+type SocksProxy = Endpoint++data Endpoint = Endpoint+ { epHost :: !HostName+ , epPort :: !Port+ } deriving (Eq, Ord, Show)++type HostName = Text+type Port = Int++------------------------------------------------------------------------++data HadoopConfig = HadoopConfig+ { hcUser :: !User+ , hcNameNodes :: ![NameNode]+ , hcProxy :: !(Maybe SocksProxy)+ } deriving (Eq, Ord, Show)++------------------------------------------------------------------------++data ConfigError = ConfigError !Text+ deriving (Show, Eq, Data, Typeable)++instance Exception ConfigError++------------------------------------------------------------------------++data RemoteError = RemoteError !Text !Text+ deriving (Show, Eq, Data, Typeable)++instance Exception RemoteError++------------------------------------------------------------------------++type User = Text+type Group = Text+type Permission = Word16++-- | Path to a file/directory on HDFS, stored as UTF8.+type HdfsPath = ByteString++-- | Microseconds since 1 January 1970.+type HdfsTime = Word64++-- | The type of a file (either directory, file or symbolic link)+data FileType = Dir | File | SymLink+ deriving (Eq, Ord, Show)++-- | Summary of a file or directory+data ContentSummary = ContentSummary+ { csLength :: !Word64+ , csFileCount :: !Word64+ , csDirectoryCount :: !Word64+ , csQuota :: !Word64+ , csSpaceConsumed :: !Word64+ , csSpaceQuota :: !Word64+ } deriving (Eq, Ord, Show)++-- | Status of a file, directory or symbolic link.+data FileStatus = FileStatus+ { fsFileType :: !FileType+ , fsPath :: !HdfsPath+ , fsLength :: !Word64+ , fsPermission :: !Permission+ , fsOwner :: !User+ , fsGroup :: !Group+ , fsModificationTime :: !HdfsTime+ , fsAccessTime :: !HdfsTime++ -- Optional fields for symlink+ , fsSymLink :: !(Maybe HdfsPath) -- ^ if symlink, target++ -- Optional fields for file+ , fsBlockReplication :: !Word16+ , fsBlockSize :: !Word64+ } deriving (Eq, Ord, Show)
+ src/Network/Hadoop/Hdfs.hs view
@@ -0,0 +1,245 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Network.Hadoop.Hdfs+ ( Hdfs(..)+ , hdfsProtocol+ , getConnection+ , runHdfs+ , runHdfs'+ , hdfsInvoke++ , CreateParent+ , Recursive+ , Overwrite++ , getListing+ , getListing'+ , getFileInfo+ , getContentSummary+ , mkdirs+ , delete+ , rename+ , setPermissions+ ) where++import Control.Applicative (Applicative(..), (<$>))+import Control.Exception (throw)+import Control.Monad (ap)+import Control.Monad.Catch (MonadMask(..), MonadThrow(..), MonadCatch(..))+import Control.Monad.IO.Class (MonadIO(..))+import Data.ByteString (ByteString)+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import qualified Data.Text.Encoding as T+import qualified Data.Vector as V+import Data.Word (Word32)++import qualified Data.Hadoop.Protobuf.ClientNameNode as P+import qualified Data.Hadoop.Protobuf.Hdfs as P+import Data.ProtocolBuffers+import Data.ProtocolBuffers.Orphans ()++import Data.Hadoop.Configuration+import Data.Hadoop.Types+import Network.Hadoop.Rpc+import qualified Network.Hadoop.Socket as S++------------------------------------------------------------------------++newtype Hdfs a = Hdfs { unHdfs :: Connection -> IO a }++instance Functor Hdfs where+ fmap f m = Hdfs $ \c -> fmap f (unHdfs m c)++instance Applicative Hdfs where+ pure = return+ (<*>) = ap++instance Monad Hdfs where+ return x = Hdfs $ \_ -> return x+ m >>= k = Hdfs $ \c -> unHdfs m c >>= \x -> unHdfs (k x) c++instance MonadIO Hdfs where+ liftIO io = Hdfs $ const io++instance MonadThrow Hdfs where+ throwM = liftIO . throwM++instance MonadCatch Hdfs where+ catch m k = Hdfs $ \c -> unHdfs m c `catch` \e -> unHdfs (k e) c++instance MonadMask Hdfs where+ mask a = Hdfs $ \e -> mask $ \u -> unHdfs (a $ q u) e+ where q u (Hdfs b) = Hdfs (u . b)+ uninterruptibleMask a =+ Hdfs $ \e -> uninterruptibleMask $ \u -> unHdfs (a $ q u) e+ where q u (Hdfs b) = Hdfs (u . b)++------------------------------------------------------------------------++type CreateParent = Bool+type Recursive = Bool+type Overwrite = Bool++------------------------------------------------------------------------++getConnection :: Hdfs Connection+getConnection = Hdfs return++hdfsProtocol :: Protocol+hdfsProtocol = Protocol "org.apache.hadoop.hdfs.protocol.ClientProtocol" 1++runHdfs :: Hdfs a -> IO a+runHdfs hdfs = do+ config <- getHadoopConfig+ runHdfs' config hdfs++runHdfs' :: HadoopConfig -> Hdfs a -> IO a+runHdfs' config@HadoopConfig{..} hdfs = S.runTcp hcProxy nameNode session+ where+ session socket = do+ conn <- initConnectionV7 config hdfsProtocol socket+ unHdfs hdfs conn++ nameNode = case hcNameNodes of+ [] -> throw (ConfigError "Could not find name nodes in Hadoop configuration")+ (x:_) -> x++hdfsInvoke :: (Encode a, Decode b) => Text -> a -> Hdfs b+hdfsInvoke method arg = Hdfs $ \c -> invoke c method arg++------------------------------------------------------------------------++getListing :: HdfsPath -> Hdfs (Maybe (V.Vector FileStatus))+getListing path = do+ mDirList <- getPartialListing path ""+ case mDirList of+ Nothing -> return Nothing+ Just dirList -> do+ let p = partialListing dirList+ if hasRemainingEntries dirList+ then Just <$> loop [p] (lastFileName p)+ else return (Just p)+ where+ partialListing :: P.DirectoryListing -> V.Vector FileStatus+ partialListing = V.map fromProtoFileStatus+ . V.fromList+ . getField+ . P.dlPartialListing++ hasRemainingEntries :: P.DirectoryListing -> Bool+ hasRemainingEntries = (/= 0) . getField . P.dlRemaingEntries++ lastFileName :: V.Vector FileStatus -> ByteString+ lastFileName v | V.null v = ""+ | otherwise = fsPath . V.last $ v++ loop :: [V.Vector FileStatus] -> ByteString -> Hdfs (V.Vector FileStatus)+ loop ps startAfter = do+ dirList <- fromMaybe emptyListing <$> getPartialListing path startAfter++ let p = partialListing dirList+ ps' = ps ++ [p]++ if hasRemainingEntries dirList+ then loop ps' (lastFileName p)+ else return (V.concat ps')++ emptyListing = P.DirectoryListing (putField []) (putField 0)++getListing' :: HdfsPath -> Hdfs (V.Vector FileStatus)+getListing' path = fromMaybe V.empty <$> getListing path++------------------------------------------------------------------------++getPartialListing :: HdfsPath -> ByteString -> Hdfs (Maybe P.DirectoryListing)+getPartialListing path startAfter = getField . P.lsDirList <$>+ hdfsInvoke "getListing" P.GetListingRequest+ { P.lsSrc = putField (T.decodeUtf8 path)+ , P.lsStartAfter = putField startAfter+ , P.lsNeedLocation = putField False+ }++getFileInfo :: HdfsPath -> Hdfs (Maybe FileStatus)+getFileInfo path = fmap fromProtoFileStatus . getField . P.fiFileStatus <$>+ hdfsInvoke "getFileInfo" P.GetFileInfoRequest+ { P.fiSrc = putField (T.decodeUtf8 path)+ }++getContentSummary :: HdfsPath -> Hdfs ContentSummary+getContentSummary path = fromProtoContentSummary . getField . P.csSummary <$>+ hdfsInvoke "getContentSummary" P.GetContentSummaryRequest+ { P.csPath = putField (T.decodeUtf8 path)+ }++mkdirs :: CreateParent -> HdfsPath -> Hdfs Bool+mkdirs createParent path = getField . P.mdResult <$>+ hdfsInvoke "mkdirs" P.MkdirsRequest+ { P.mdSrc = putField (T.decodeUtf8 path)+ , P.mdMasked = putField (P.FilePermission (putField 0o755))+ , P.mdCreateParent = putField createParent+ }++delete :: Recursive -> HdfsPath -> Hdfs Bool+delete recursive path = getField . P.dlResult <$>+ hdfsInvoke "delete" P.DeleteRequest+ { P.dlSrc = putField (T.decodeUtf8 path)+ , P.dlRecursive = putField recursive+ }++rename :: Overwrite -> HdfsPath -> HdfsPath -> Hdfs ()+rename overwrite src dst = ignore <$>+ hdfsInvoke "rename2" P.Rename2Request+ { P.mvSrc = putField (T.decodeUtf8 src)+ , P.mvDst = putField (T.decodeUtf8 dst)+ , P.mvOverwrite = putField overwrite+ }+ where+ ignore :: P.Rename2Response -> ()+ ignore = const ()++setPermissions :: Word32 -> HdfsPath -> Hdfs ()+setPermissions mode path = ignore <$>+ hdfsInvoke "setPermission" P.SetPermissionRequest+ { P.chmodPath = putField (T.decodeUtf8 path)+ , P.chmodMode = putField $ P.FilePermission { fpPerm = putField mode }+ }+ where+ ignore :: P.SetPermissionResponse -> ()+ ignore = const ()++------------------------------------------------------------------------+-- Awesome Protobuf Mapping (╯°□°)╯︵ ┻━┻++fromProtoContentSummary :: P.ContentSummary -> ContentSummary+fromProtoContentSummary p = ContentSummary+ { csLength = getField $ P.csLength p+ , csFileCount = getField $ P.csFileCount p+ , csDirectoryCount = getField $ P.csDirectoryCount p+ , csQuota = getField $ P.csQuota p+ , csSpaceConsumed = getField $ P.csSpaceConsumed p+ , csSpaceQuota = getField $ P.csSpaceQuota p+ }++fromProtoFileStatus :: P.FileStatus -> FileStatus+fromProtoFileStatus p = FileStatus+ { fsFileType = fromProtoFileType $ getField $ P.fsFileType p+ , fsPath = getField $ P.fsPath p+ , fsLength = getField $ P.fsLength p+ , fsPermission = fromIntegral $ getField $ P.fpPerm $ getField $ P.fsPermission p+ , fsOwner = getField $ P.fsOwner p+ , fsGroup = getField $ P.fsGroup p+ , fsModificationTime = getField $ P.fsModificationTime p+ , fsAccessTime = getField $ P.fsAccessTime p+ , fsSymLink = getField $ P.fsSymLink p+ , fsBlockReplication = fromIntegral $ fromMaybe 0 $ getField $ P.fsBlockReplication p+ , fsBlockSize = fromMaybe 0 $ getField $ P.fsBlockSize p+ }++fromProtoFileType :: P.FileType -> FileType+fromProtoFileType p = case p of+ P.IS_DIR -> Dir+ P.IS_FILE -> File+ P.IS_SYMLINK -> SymLink
+ src/Network/Hadoop/Read.hs view
@@ -0,0 +1,217 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}++module Network.Hadoop.Read+ ( HdfsReadHandle+ , openRead+ , hdfsMapM_+ , hdfsFoldM++ -- * Convenience utilities+ , hdfsCat+ ) where++import Control.Applicative (Applicative(..), (<$>))+import Control.Exception (SomeException, throwIO)+import Control.Monad (guard, foldM)+import Control.Monad.Catch (MonadMask, catch)+import Control.Monad.IO.Class (MonadIO(..))+import Data.Attoparsec.Text (Parser, char, decimal, parseOnly)+import Data.Bits+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as B+import Data.List (intercalate)+import Data.Maybe (fromMaybe)+import Data.ProtocolBuffers+import Data.ProtocolBuffers.Orphans ()+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import Data.Word (Word32, Word64)++import Data.Serialize.Get+import Data.Serialize.Put++import Data.Hadoop.Protobuf.ClientNameNode+import Data.Hadoop.Protobuf.DataTransfer+import Data.Hadoop.Protobuf.Hdfs+import Data.Hadoop.Protobuf.Security+import Data.Hadoop.Types+import Network.Hadoop.Hdfs+import Network.Hadoop.Rpc+import qualified Network.Hadoop.Socket as S+import Network.Hadoop.Stream (Stream)+import qualified Network.Hadoop.Stream as Stream++------------------------------------------------------------------------++data HdfsReadHandle = HdfsReadHandle+ { readProxy :: Maybe SocksProxy+ , readLocations :: LocatedBlocks+ }++-- | Open an HDFS path for reading.+-- This retrieves the block locations for an HDFS path, but does not+-- yet initiate any data transfers.+openRead :: HdfsPath -> Hdfs (Maybe HdfsReadHandle)+openRead path = do+ locs <- getField . locsLocations <$> hdfsInvoke "getBlockLocations" GetBlockLocationsRequest+ { catSrc = putField (T.decodeUtf8 path)+ , catOffset = putField 0+ , catLength = putField maxBound+ }+ case locs of+ Nothing -> return Nothing+ Just ls -> do+ proxy <- hcProxy . cnConfig <$> getConnection+ return $ Just (HdfsReadHandle proxy ls)++-- | Print the contents of a file on HDFS to stdout+hdfsCat :: HdfsPath -> Hdfs ()+hdfsCat path = do+ h <- openRead path+ maybe (return ()) (hdfsMapM_ (liftIO . B.putStr)) h++-- | Map an action over the contents of a file on HDFS.+hdfsMapM_ :: (MonadIO m, MonadMask m) =>+ (ByteString -> m ()) -> HdfsReadHandle -> m ()+hdfsMapM_ f = hdfsFoldM (\_ x -> f x) ()++hdfsFoldM :: (MonadIO m, MonadMask m) =>+ (a -> ByteString -> m a) -> a -> HdfsReadHandle -> m a+hdfsFoldM f acc0 (HdfsReadHandle proxy l) = do+ let len = getField . lbFileLength $ l+ foldM (procBlock f proxy len) acc0 (getField . lbBlocks $ l)++procBlock :: (MonadIO m, MonadMask m) =>+ (a -> ByteString -> m a) -> Maybe SocksProxy -> Word64 -> a -> LocatedBlock -> m a+procBlock f proxy len acc0 b = do+ let extended = getField . lbExtended $ b+ token = getField . lbToken $ b+ case getField . lbLocations $ b of+ [] -> error $ "No locations for block " ++ show extended+ ls -> failover (error $ "All locations failed for block " ++ show extended)+ (map (getLoc proxy len extended token) ls)+ where+ failover err [] = err+ failover err (x:xs) = catch x f+ where f (_ :: SomeException) = failover err xs++ getLoc proxy len extended token l = do+ let i = getField (dnId l)+ Right addr = parseOnly parseIPv4 . getField . dnIpAddr $ i+ host = getField . dnHostName $ i+ port = getField . dnXferPort $ i+ let endpoint = Endpoint host (fromIntegral port)+ runBlock proxy endpoint 0 len extended token++ runBlock proxy endpoint offset len0 extended token = do+ let len = fromMaybe len0 . getField . ebNumBytes $ extended+ S.runTcp proxy endpoint $ readBlock offset len extended token++ readBlock offset len extended token sock = go 0 offset len acc0+ where+ go nread0 offset0 rem0 acc = do+ (len, acc') <- readBlockPart offset0 rem0 extended token sock acc+ let offset = offset0 + len+ nread = nread0 + len+ rem = rem0 - len+ if rem > 0 then go nread offset rem acc' else return acc'++ readBlockPart offset rem extended token sock acc = do+ stream <- liftIO $ Stream.mkSocketStream sock+ b <- liftIO $ do+ Stream.runPut stream putReadRequest+ Stream.runGet stream decodeLengthPrefixedMessage+ readPackets b rem stream acc+ where++ putReadRequest = do+ putWord16be 28 -- Data Transfer Protocol Version, 2 bytes+ putWord8 81 -- READ_BLOCK+ encodeLengthPrefixedMessage opReadBlock++ opReadBlock = OpReadBlock+ { orbHeader = putField coh+ , orbOffset = putField offset+ , orbLen = putField rem+ , orbSendChecksums = putField Nothing+ , orbCachingStrategy = putField Nothing+ }+ coh = ClientOperationHeader+ { cohBaseHeader = putField bh+ , cohClientName = putField (T.pack "hh")+ }+ bh = BaseHeader+ { bhBlock = putField extended+ , bhToken = putField (Just token)+ }++ showBlockOpResponse :: BlockOpResponse -> String+ showBlockOpResponse = show++ readPackets BlockOpResponse{..} len stream acc1 = go 0 len acc1+ where+ go nread0 rem0 acc = do+ (len, acc') <- readPacket b rem0 stream acc+ let rem = rem0 - len+ nread = nread0 + len+ if rem > 0 then go nread rem acc' else return (nread, acc')++ m = getField borReadOpChecksumInfo+ c = getField . rociChecksum <$> m+ b = fromIntegral . getField . csBytesPerChecksum <$> c++ readPacket bytesPerChecksum remaining stream acc = do+ (dataLen, d) <- liftIO $ do+ len <- Stream.runGet stream getWord32be+ sz <- Stream.runGet stream getWord16be++ bs <- Stream.runGet stream $ getByteString (fromIntegral sz)+ ph <- decodeBytes bs+ let numChunks = countChunks ph+ dataLen = fromIntegral . getField . phDataLen $ ph++ _ <- Stream.runGet stream (getByteString (4*numChunks))+ (fromIntegral dataLen,) <$>+ Stream.runGet stream (getByteString dataLen)++ acc' <- f acc d+ return (fromIntegral dataLen, acc')+ where+ showPacketHeader :: PacketHeader -> String+ showPacketHeader = show++ countChunks :: PacketHeader -> Int+ countChunks PacketHeader{..} = (dataLen + b - 1) `div` b+ where+ b = fromIntegral $ fromMaybe 512 bytesPerChecksum+ dataLen = fromIntegral $ getField phDataLen++ decodeBytes bs = case runGetState decodeMessage bs 0 of+ Left err -> throwIO (RemoteError "DecodeError" (T.pack err))+ Right (x, "") -> return x+ Right (_, _) -> throwIO (RemoteError "DecodeError" "decoded response but did not consume enough bytes")++------------------------------------------------------------------------++parseIPv4 :: Parser Word32+parseIPv4 = d >>= dd >>= dd >>= dd+ where+ d = do+ x <- decimal+ guard $ x < 256+ return x+ dd acc = do+ x <- char '.' *> d+ return $ (acc `shiftL` 8) .|. x++showIPv4 :: Word32 -> String+showIPv4 x = intercalate "." . map show $ [a,b,c,d]+ where+ a = (x .&. 0xFF000000) `shiftR` 24+ b = (x .&. 0xFF0000) `shiftR` 16+ c = (x .&. 0xFF00) `shiftR` 8+ d = (x .&. 0xFF)
+ src/Network/Hadoop/Rpc.hs view
@@ -0,0 +1,141 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Network.Hadoop.Rpc+ ( Connection(..)+ , Protocol(..)+ , User+ , Method+ , RawRequest+ , RawResponse++ , initConnectionV7+ , invoke+ ) where++import Control.Applicative ((<$>), (<*>))+import Control.Exception (throwIO)+import Control.Monad.IO.Class (liftIO)++import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as B+import Data.IORef+import Data.Monoid (mempty)+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T++import Data.ProtocolBuffers+import Data.ProtocolBuffers.Orphans ()+import Data.Serialize.Get+import Data.Serialize.Put++import Data.Hadoop.Protobuf.Headers+import Data.Hadoop.Types+import qualified Network.Hadoop.Stream as S+import Network.Socket (Socket)++------------------------------------------------------------------------++data Connection = Connection+ { cnVersion :: !Int+ , cnConfig :: !HadoopConfig+ , cnProtocol :: !Protocol+ , invokeRaw :: !(Method -> RawRequest -> IO RawResponse)+ }++data Protocol = Protocol+ { prName :: !Text+ , prVersion :: !Int+ } deriving (Eq, Ord, Show)++type Method = Text+type RawRequest = ByteString+type RawResponse = ByteString++------------------------------------------------------------------------++-- hadoop-2.1.0-beta is on version 9+-- see https://issues.apache.org/jira/browse/HADOOP-8990 for differences++initConnectionV7 :: HadoopConfig -> Protocol -> Socket -> IO Connection+initConnectionV7 config@HadoopConfig{..} protocol sock = do+ stream <- S.mkSocketStream sock+ S.runPut stream $ do+ putByteString "hrpc"+ putWord8 7 -- version+ putWord8 80 -- auth method (80 = simple, 81 = kerberos/gssapi, 82 = token/digest-md5)+ putWord8 0 -- ipc serialization type (0 = protobuf)++ let bs = runPut (encodeMessage context)+ putWord32be (fromIntegral (B.length bs))+ putByteString bs++ ref <- newIORef 0+ return (Connection 7 config protocol (sendAndWait stream ref))+ where+ sendAndWait :: S.Stream -> IORef Int -> Method -> ByteString -> IO ByteString+ sendAndWait stream ref method requestBytes = do+ callId <- atomicModifyIORef' ref (\x -> (succ x, x))++ S.runPut stream $ do+ let bs = runPut $ encodeLengthPrefixedMessage (requestHeader callId)+ >> encodeLengthPrefixedMessage (request method requestBytes)++ putWord32be (fromIntegral (B.length bs))+ putByteString bs++ responseHdr <- S.maybeGet stream decodeLengthPrefixedMessage+ case getField . rspStatus <$> responseHdr of+ Just Success -> S.runGet stream getResponse+ Just _ -> S.runGet stream getError >>= liftIO . throwIO+ Nothing -> throwClosed++ context = IpcConnectionContext+ { ctxProtocol = putField (Just (prName protocol))+ , ctxUserInfo = putField (Just UserInformation+ { effectiveUser = putField (Just hcUser)+ , realUser = mempty+ })+ }++ requestHeader callId = RpcRequestHeader+ { reqKind = putField (Just ProtocolBuffer)+ , reqOp = putField (Just FinalPacket)+ , reqCallId = putField (fromIntegral callId)+ }++ request method bytes = RpcRequest+ { reqMethodName = putField method+ , reqBytes = putField (Just bytes)+ , reqProtocolName = putField (prName protocol)+ , reqProtocolVersion = putField (fromIntegral (prVersion protocol))+ }++ throwClosed = throwIO (RemoteError "ConnectionClosed" "The socket connection was closed")++------------------------------------------------------------------------++getResponse :: Get ByteString+getResponse = do+ n <- fromIntegral <$> getWord32be+ getByteString n++getError :: Get RemoteError+getError = RemoteError <$> getText <*> getText+ where+ getText = do+ n <- fromIntegral <$> getWord32be+ T.decodeUtf8 <$> getByteString n++------------------------------------------------------------------------++invoke :: (Decode b, Encode a) => Connection -> Text -> a -> IO b+invoke Connection{..} method arg = decodeBytes =<< invokeRaw method (encodeBytes arg)+ where+ encodeBytes = runPut . encodeMessage+ decodeBytes bs = case runGetState decodeMessage bs 0 of+ Left err -> throwIO (RemoteError "DecodeError" (T.pack err))+ Right (x, "") -> return x+ Right (_, _) -> throwIO (RemoteError "DecodeError" "decoded response but did not consume enough bytes")
+ src/Network/Hadoop/Socket.hs view
@@ -0,0 +1,64 @@+module Network.Hadoop.Socket+ ( S.Socket+ , S.SockAddr(..)++ , runTcp++ , connectSocket+ , newSocket+ , closeSocket+ ) where++import Control.Applicative ((<$>))+import Control.Monad.Catch (MonadMask, bracket, bracketOnError)+import Control.Monad.IO.Class (MonadIO(..))+import Data.Hadoop.Types+import qualified Data.Text as T+import Network (PortID(PortNumber))+import qualified Network.Socket as S+import Network.Socks5 (defaultSocksConf, socksConnectWith)++------------------------------------------------------------------------++runTcp :: (MonadMask m, MonadIO m) => Maybe SocksProxy -> Endpoint -> (S.Socket -> m a) -> m a+runTcp Nothing = runTcp'+runTcp (Just proxy) = runSocks proxy++runTcp' :: (MonadMask m, MonadIO m) => Endpoint -> (S.Socket -> m a) -> m a+runTcp' endpoint = bracket+ (liftIO $ fst <$> connectSocket endpoint)+ (liftIO . closeSocket)++runSocks :: (MonadMask m, MonadIO m) => SocksProxy -> Endpoint -> (S.Socket -> m a) -> m a+runSocks proxy endpoint = bracket+ (liftIO $ socksConnectWith proxyConf host port)+ (liftIO . closeSocket)+ where+ proxyConf = defaultSocksConf (T.unpack $ epHost proxy)+ (fromIntegral $ epPort proxy)++ host = T.unpack $ epHost endpoint+ port = PortNumber $ fromIntegral $ epPort endpoint++------------------------------------------------------------------------++connectSocket :: Endpoint -> IO (S.Socket, S.SockAddr)+connectSocket endpoint = do+ (addr:_) <- S.getAddrInfo (Just hints) (Just host) (Just port)+ bracketOnError (newSocket addr) closeSocket $ \sock -> do+ let sockAddr = S.addrAddress addr+ S.connect sock sockAddr+ return (sock, sockAddr)+ where+ host = T.unpack (epHost endpoint)+ port = show (epPort endpoint)+ hints = S.defaultHints { S.addrFlags = [S.AI_ADDRCONFIG]+ , S.addrSocketType = S.Stream }++newSocket :: S.AddrInfo -> IO S.Socket+newSocket addr = S.socket (S.addrFamily addr)+ (S.addrSocketType addr)+ (S.addrProtocol addr)++closeSocket :: S.Socket -> IO ()+closeSocket = S.sClose
+ src/Network/Hadoop/Stream.hs view
@@ -0,0 +1,161 @@+{-# LANGUAGE OverloadedStrings #-}++--------------------------------------------------------------------------------+-- | Lightweight abstraction over an input/output stream.+-- (stolen from 'websockets' package)++module Network.Hadoop.Stream+ ( Stream+ , mkStream+ , mkSocketStream+ , mkEchoStream+ , parse+ , maybeGet+ , runGet+ , runPut+ , write+ , close+ ) where++import Control.Applicative ((<$>))+import qualified Control.Concurrent.Chan as Chan+import Control.Exception (throwIO)+import Control.Monad (forM_)+import qualified Data.Attoparsec.ByteString as Atto+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as L+import Data.IORef (IORef, newIORef, readIORef, writeIORef)+import qualified Data.Serialize.Get as Get+import qualified Data.Serialize.Put as Put+import qualified Network.Socket as S+import qualified Network.Socket.ByteString as B (recv)+import qualified Network.Socket.ByteString.Lazy as L (sendAll)++import Data.Hadoop.Types++--------------------------------------------------------------------------------++-- | State of the stream+data StreamState+ = Closed !B.ByteString -- Remainder+ | Open !B.ByteString -- Buffer++--------------------------------------------------------------------------------++-- | Lightweight abstraction over an input/output stream.+data Stream = Stream+ { streamIn :: IO (Maybe B.ByteString)+ , streamOut :: (Maybe L.ByteString -> IO ())+ , streamState :: !(IORef StreamState)+ }++--------------------------------------------------------------------------------++mkStream+ :: IO (Maybe B.ByteString) -- ^ Reading+ -> (Maybe L.ByteString -> IO ()) -- ^ Writing+ -> IO Stream -- ^ Resulting stream+mkStream i o = Stream i o <$> newIORef (Open B.empty)++--------------------------------------------------------------------------------++mkSocketStream :: S.Socket -> IO Stream+mkSocketStream socket = mkStream receive send+ where+ receive = do+ bs <- B.recv socket (1024*1024)+ return $ if B.null bs then Nothing else Just bs++ send Nothing = return ()+ send (Just bs) = L.sendAll socket bs++--------------------------------------------------------------------------------++mkEchoStream :: IO Stream+mkEchoStream = do+ chan <- Chan.newChan+ mkStream (Chan.readChan chan) $ \mbBs -> case mbBs of+ Nothing -> Chan.writeChan chan Nothing+ Just bs -> forM_ (L.toChunks bs) $ \c -> Chan.writeChan chan (Just c)++--------------------------------------------------------------------------------++parse :: Stream -> Atto.Parser a -> IO (Maybe a)+parse stream parser = do+ state <- readIORef (streamState stream)+ case state of+ Closed remainder+ | B.null remainder -> return Nothing+ | otherwise -> go (Atto.parse parser remainder) True+ Open buffer+ | B.null buffer -> do+ mbBs <- streamIn stream+ case mbBs of+ Nothing -> do+ writeIORef (streamState stream) (Closed B.empty)+ return Nothing+ Just bs -> go (Atto.parse parser bs) False+ | otherwise -> go (Atto.parse parser buffer) False+ where+ -- Buffer is empty when entering this function.+ go (Atto.Done remainder x) closed = do+ writeIORef (streamState stream) $+ if closed then Closed remainder else Open remainder+ return (Just x)+ go (Atto.Partial f) closed+ | closed = go (f B.empty) True+ | otherwise = do+ mbBs <- streamIn stream+ case mbBs of+ Nothing -> go (f B.empty) True+ Just bs -> go (f bs) False+ go (Atto.Fail _ _ err) _ = error ("parse: " ++ err)++maybeGet :: Stream -> Get.Get a -> IO (Maybe a)+maybeGet stream getter = do+ state <- readIORef (streamState stream)+ case state of+ Closed remainder+ | B.null remainder -> return Nothing+ | otherwise -> go (Get.runGetPartial getter remainder) True+ Open buffer+ | B.null buffer -> do+ mbBs <- streamIn stream+ case mbBs of+ Nothing -> do+ writeIORef (streamState stream) (Closed B.empty)+ return Nothing+ Just bs -> go (Get.runGetPartial getter bs) False+ | otherwise -> go (Get.runGetPartial getter buffer) False+ where+ -- Buffer is empty when entering this function.+ go (Get.Done x remainder) closed = do+ writeIORef (streamState stream) $+ if closed then Closed remainder else Open remainder+ return (Just x)+ go (Get.Partial f) closed+ | closed = go (f B.empty) True+ | otherwise = do+ mbBs <- streamIn stream+ case mbBs of+ Nothing -> go (f B.empty) True+ Just bs -> go (f bs) False+ go (Get.Fail err _) _ = error ("runGetStream: " ++ err)++runGet :: Stream -> Get.Get a -> IO a+runGet stream getter = maybe throwClosed return =<< maybeGet stream getter+ where+ throwClosed = throwIO (RemoteError "ConnectionClosed" "The socket connection was closed")++--------------------------------------------------------------------------------++runPut :: Stream -> Put.Put -> IO ()+runPut stream = write stream . Put.runPutLazy++write :: Stream -> L.ByteString -> IO ()+write stream = streamOut stream . Just++--------------------------------------------------------------------------------++close :: Stream -> IO ()+close stream = streamOut stream Nothing
+ test/Main.hs view
@@ -0,0 +1,20 @@+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import Data.Hadoop.Types+import Network.Hadoop.Hdfs++------------------------------------------------------------------------++main :: IO ()+main = do+ files <- runHdfs' config (getListing' "/")+ print files++config :: HadoopConfig+config = HadoopConfig {+ hcUser = "hdfs"+ , hcNameNodes = [(Endpoint "127.0.0.1" 8020)]+ , hcProxy = Just (Endpoint "127.0.0.1" 2080)+ }