diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/hadoop-rpc.cabal b/hadoop-rpc.cabal new file mode 100644 --- /dev/null +++ b/hadoop-rpc.cabal @@ -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 +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 diff --git a/src/Data/Hadoop/Configuration.hs b/src/Data/Hadoop/Configuration.hs new file mode 100644 --- /dev/null +++ b/src/Data/Hadoop/Configuration.hs @@ -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 diff --git a/src/Data/Hadoop/Protobuf/ClientNameNode.hs b/src/Data/Hadoop/Protobuf/ClientNameNode.hs new file mode 100644 --- /dev/null +++ b/src/Data/Hadoop/Protobuf/ClientNameNode.hs @@ -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 diff --git a/src/Data/Hadoop/Protobuf/DataTransfer.hs b/src/Data/Hadoop/Protobuf/DataTransfer.hs new file mode 100644 --- /dev/null +++ b/src/Data/Hadoop/Protobuf/DataTransfer.hs @@ -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 diff --git a/src/Data/Hadoop/Protobuf/Hdfs.hs b/src/Data/Hadoop/Protobuf/Hdfs.hs new file mode 100644 --- /dev/null +++ b/src/Data/Hadoop/Protobuf/Hdfs.hs @@ -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 diff --git a/src/Data/Hadoop/Protobuf/Headers.hs b/src/Data/Hadoop/Protobuf/Headers.hs new file mode 100644 --- /dev/null +++ b/src/Data/Hadoop/Protobuf/Headers.hs @@ -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 +-} diff --git a/src/Data/Hadoop/Protobuf/ProtocolInfo.hs b/src/Data/Hadoop/Protobuf/ProtocolInfo.hs new file mode 100644 --- /dev/null +++ b/src/Data/Hadoop/Protobuf/ProtocolInfo.hs @@ -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 + } + +-} diff --git a/src/Data/Hadoop/Protobuf/Security.hs b/src/Data/Hadoop/Protobuf/Security.hs new file mode 100644 --- /dev/null +++ b/src/Data/Hadoop/Protobuf/Security.hs @@ -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 diff --git a/src/Data/Hadoop/Types.hs b/src/Data/Hadoop/Types.hs new file mode 100644 --- /dev/null +++ b/src/Data/Hadoop/Types.hs @@ -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) diff --git a/src/Network/Hadoop/Hdfs.hs b/src/Network/Hadoop/Hdfs.hs new file mode 100644 --- /dev/null +++ b/src/Network/Hadoop/Hdfs.hs @@ -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 diff --git a/src/Network/Hadoop/Read.hs b/src/Network/Hadoop/Read.hs new file mode 100644 --- /dev/null +++ b/src/Network/Hadoop/Read.hs @@ -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) diff --git a/src/Network/Hadoop/Rpc.hs b/src/Network/Hadoop/Rpc.hs new file mode 100644 --- /dev/null +++ b/src/Network/Hadoop/Rpc.hs @@ -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") diff --git a/src/Network/Hadoop/Socket.hs b/src/Network/Hadoop/Socket.hs new file mode 100644 --- /dev/null +++ b/src/Network/Hadoop/Socket.hs @@ -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 diff --git a/src/Network/Hadoop/Stream.hs b/src/Network/Hadoop/Stream.hs new file mode 100644 --- /dev/null +++ b/src/Network/Hadoop/Stream.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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) + }