hadoop-rpc 1.0.0.1 → 1.1.0.0
raw patch · 10 files changed
+337/−64 lines, 10 filesdep +gsasldep ~transformers
Dependencies added: gsasl
Dependency ranges changed: transformers
Files
- hadoop-rpc.cabal +7/−4
- src/Data/Hadoop/Configuration.hs +58/−11
- src/Data/Hadoop/Types.hs +16/−2
- src/Network/Hadoop/Hdfs.hs +8/−6
- src/Network/Hadoop/Read.hs +31/−7
- src/Network/Hadoop/Rpc.hs +118/−31
- src/Network/Hadoop/Sasl.hs +67/−0
- src/Network/Hadoop/Stream.hs +5/−1
- src/Network/Hadoop/Types.hs +25/−0
- test/Main.hs +2/−2
hadoop-rpc.cabal view
@@ -1,5 +1,5 @@ name: hadoop-rpc-version: 1.0.0.1+version: 1.1.0.0 synopsis: Use the Hadoop RPC interface from Haskell.@@ -16,7 +16,7 @@ homepage: http://github.com/jystic/hadoop-rpc license: Apache-2.0 license-file: LICENSE-author: Jacob Stanley, Conrad Parker+author: Jacob Stanley, Conrad Parker, Luke Clifton maintainer: Jacob Stanley <jacob@stanley.io> category: Data build-type: Simple@@ -41,8 +41,10 @@ Network.Hadoop.Hdfs Network.Hadoop.Read Network.Hadoop.Rpc+ Network.Hadoop.Sasl Network.Hadoop.Socket Network.Hadoop.Stream+ Network.Hadoop.Types build-depends: base >= 4.7 && < 5@@ -50,6 +52,7 @@ , bytestring >= 0.10 , cereal >= 0.4 , exceptions >= 0.6+ , gsasl >= 0.3.6 , hashable >= 1.2.1 , monad-loops >= 0.4 , network >= 2.5@@ -58,7 +61,7 @@ , socks >= 0.5 , stm >= 2.4 , text >= 1.1- , transformers >= 0.4+ , transformers >= 0.3 , unix >= 2.7 , unordered-containers >= 0.2 , uuid >= 1.3.4@@ -80,7 +83,7 @@ buildable: False build-depends:- base >= 4.7 && < 5+ base >= 4.7 && < 5 , hadoop-rpc , protobuf , tasty
src/Data/Hadoop/Configuration.hs view
@@ -2,13 +2,17 @@ {-# LANGUAGE RecordWildCards #-} module Data.Hadoop.Configuration- ( getHadoopConfig+ ( authUser+ , getHadoopConfig , getHadoopUser , getNameNodes+ , readPrincipal+ , writePrincipal ) where -import Control.Applicative ((<$>), (<*>))+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)@@ -19,17 +23,23 @@ import System.Environment (lookupEnv) import System.Posix.User (getEffectiveUserName)-import Text.XmlHtml +import Text.XmlHtml (Document, parseXML, docContent)+import Text.XmlHtml (nodeText, childElementTag, descendantElementsTag)++import Prelude+ import Data.Hadoop.Types ------------------------------------------------------------------------ getHadoopConfig :: IO HadoopConfig getHadoopConfig = do- hcUser <- getHadoopUser+ udUser <- getHadoopUser+ let udAuthUser = Nothing hcNameNodes <- getNameNodes let hcProxy = Nothing+ let hcUser = UserDetails{..} return HadoopConfig{..} ------------------------------------------------------------------------@@ -45,6 +55,28 @@ ------------------------------------------------------------------------ +-- Extract the name to be used for authentication+authUser :: UserDetails -> User+authUser UserDetails{..} = maybe udUser id udAuthUser++------------------------------------------------------------------------++readPrincipal :: Text -> HostName -> Maybe Principal+readPrincipal p host =+ case T.split (`elem` ['/', '@']) p of+ [pService,"_HOST",pRealm] -> let pHost = host in Just Principal{..}+ [pService,pHost,pRealm] -> Just Principal{..}+ _ ->+ case T.split (`elem` ['@']) p of+ [pService,pRealm] -> let pHost = "" in Just Principal{..}+ _ -> Nothing++writePrincipal :: Principal -> Text+writePrincipal Principal{..} = case pHost of+ "" -> pService <> "@" <> pRealm+ _ -> pService <> "/" <> pHost <> "@" <> pRealm++------------------------------------------------------------------------ type HadoopXml = H.HashMap Text Text getNameNodes :: IO [NameNode]@@ -54,10 +86,11 @@ return $ fromMaybe [] $ resolveNameNode cfg <$> (stripProto =<< H.lookup fsDefaultNameKey cfg) where- proto = "hdfs://"- fsDefaultNameKey = "fs.defaultFS"- nameNodesPrefix = "dfs.ha.namenodes."- rpcAddressPrefix = "dfs.namenode.rpc-address."+ proto = "hdfs://"+ fsDefaultNameKey = "fs.defaultFS"+ nameNodesPrefix = "dfs.ha.namenodes."+ rpcAddressPrefix = "dfs.namenode.rpc-address."+ namenodePrincipal = "dfs.namenode.kerberos.principal" stripProto :: Text -> Maybe Text stripProto uri | proto `T.isPrefixOf` uri = Just (T.drop (T.length proto) uri)@@ -65,9 +98,23 @@ 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)+ -- contains "host:port" directly+ Just ep@Endpoint{..} ->+ [ NameNode+ { nnEndPoint = ep+ , nnPrincipal = lookupPrincipal cfg epHost+ }+ ]+ Nothing -> mapMaybe (\nn -> do+ ep <- lookupAddress cfg $ name <> "." <> nn+ let pr = lookupPrincipal cfg (epHost ep)+ return $ NameNode ep pr+ ) (lookupNameNodes cfg name)++ lookupPrincipal :: HadoopXml -> HostName -> Maybe Principal+ lookupPrincipal cfg host = do+ p <- H.lookup namenodePrincipal cfg+ readPrincipal p host lookupNameNodes :: HadoopXml -> Text -> [Text] lookupNameNodes cfg name = fromMaybe []
src/Data/Hadoop/Types.hs view
@@ -12,9 +12,18 @@ ------------------------------------------------------------------------ -type NameNode = Endpoint+data NameNode = NameNode+ { nnEndPoint :: !Endpoint+ , nnPrincipal :: !(Maybe Principal)+ } deriving (Eq, Ord, Show) type SocksProxy = Endpoint +data Principal = Principal+ { pService :: !Text+ , pHost :: !Text+ , pRealm :: !Text+ } deriving (Eq, Ord, Show)+ data Endpoint = Endpoint { epHost :: !HostName , epPort :: !Port@@ -26,9 +35,14 @@ ------------------------------------------------------------------------ data HadoopConfig = HadoopConfig- { hcUser :: !User+ { hcUser :: !UserDetails , hcNameNodes :: ![NameNode] , hcProxy :: !(Maybe SocksProxy)+ } deriving (Eq, Ord, Show)++data UserDetails = UserDetails+ { udUser :: !User+ , udAuthUser :: !(Maybe User) } deriving (Eq, Ord, Show) ------------------------------------------------------------------------
src/Network/Hadoop/Hdfs.hs view
@@ -25,27 +25,29 @@ , setPermissions ) where -import Control.Applicative (Applicative(..), (<$>))+import Control.Applicative import Control.Concurrent.STM import Control.Exception (SomeException(..), throw) import Control.Monad (ap, when) import Control.Monad.Catch (MonadMask(..), MonadThrow(..), MonadCatch(..)) import Control.Monad.IO.Class (MonadIO(..))+ import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B import Data.Maybe (fromMaybe) import Data.Monoid ((<>))+import Data.ProtocolBuffers+import Data.ProtocolBuffers.Orphans () import Data.Text (Text) import qualified Data.Text as T import qualified Data.Text.Encoding as T import qualified Data.Vector as V import Data.Word (Word32) +import Prelude+ 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.HdfsPath import Data.Hadoop.Types@@ -103,10 +105,10 @@ runHdfs' config hdfs runHdfs' :: HadoopConfig -> Hdfs a -> IO a-runHdfs' config@HadoopConfig{..} hdfs = S.bracketSocket hcProxy nameNode session+runHdfs' config@HadoopConfig{..} hdfs = S.bracketSocket hcProxy (nnEndPoint nameNode) session where session socket = do- conn <- initConnectionV9 config hdfsProtocol socket+ conn <- initConnectionV9 config nameNode hdfsProtocol socket unHdfs hdfs conn nameNode = case hcNameNodes of
src/Network/Hadoop/Read.hs view
@@ -3,6 +3,29 @@ {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TupleSections #-} +{-|++ Example for accessing HDFS files:++>import qualified Data.ByteString.Char8 as BC+>import Network.Hadoop.Hdfs+>import Network.Hadoop.Read+>import System.IO (IOMode(..), withBinaryFile)+>+> main :: IO ()+> main = do+> (Just readHandle) <- runHdfs $ openRead $ BC.pack+> withBinaryFile "local-file" WriteMode $ \handle ->+> withHdfsReader (BC.hPut handle) "/absolute/path/to/file"+>+> withHdfsReader :: (BC.ByteString -> IO ()) -> String -> IO ()+> withHdfsReader action path = do+> readHandle_ <- runHdfs $ openRead $ BC.pack path+> case readHandle_ of+> (Just readHandle) -> hdfsMapM_ action readHandle+> Nothing -> error "no read handle++-} module Network.Hadoop.Read ( HdfsReadHandle , openRead@@ -13,26 +36,30 @@ , hdfsCat ) where -import Control.Applicative ((<$>), (<$))+import Control.Applicative import Control.Exception (IOException, throwIO) import Control.Monad (foldM) import Control.Monad.Catch (MonadMask, catch) import Control.Monad.IO.Class (MonadIO(..)) import Control.Monad.Loops (iterateUntilM)-import Control.Monad.Trans.State import Control.Monad.Trans.Class (lift)+import Control.Monad.Trans.State (StateT(..), execStateT, get, put)+ import Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as B import Data.Maybe (fromMaybe) import Data.ProtocolBuffers import Data.ProtocolBuffers.Orphans ()+import qualified Data.Serialize.Get as Get+import qualified Data.Serialize.Put as Put import qualified Data.Text as T import qualified Data.Text.Encoding as T import Data.Word (Word64) -import qualified Data.Serialize.Get as Get-import qualified Data.Serialize.Put as Put+import Text.Printf (printf) +import Prelude hiding (rem)+ import Data.Hadoop.Protobuf.ClientNameNode import Data.Hadoop.Protobuf.DataTransfer import Data.Hadoop.Protobuf.Hdfs@@ -41,9 +68,6 @@ import Network.Hadoop.Rpc import qualified Network.Hadoop.Socket as S import qualified Network.Hadoop.Stream as Stream-import Text.Printf (printf)--import Prelude hiding (rem) ------------------------------------------------------------------------
src/Network/Hadoop/Rpc.hs view
@@ -14,50 +14,41 @@ , invoke ) where -import Control.Applicative ((<$>))+import Control.Applicative import Control.Concurrent (ThreadId, forkIO, newEmptyMVar, putMVar, takeMVar) import Control.Concurrent.STM import Control.Exception (SomeException(..), throwIO, handle) import Control.Monad (forever, when)+import Control.Monad.IO.Class (liftIO)+ import Data.ByteString (ByteString) import qualified Data.ByteString.Lazy as L import qualified Data.HashMap.Strict as H import Data.Hashable (Hashable) import Data.Maybe (fromMaybe, isNothing)-import Data.Monoid ((<>))-import Data.Monoid (mempty)-import Data.Text (Text)-import qualified Data.Text as T-import qualified Data.UUID as UUID-import System.Random (randomIO)-+import Data.Monoid import Data.ProtocolBuffers import Data.ProtocolBuffers.Orphans () import Data.Serialize.Get import Data.Serialize.Put+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Data.UUID as UUID -import qualified Data.Hadoop.Protobuf.Headers as P-import Data.Hadoop.Types-import qualified Network.Hadoop.Stream as S+import qualified Network.Protocol.SASL.GNU as GS import Network.Socket (Socket) ---------------------------------------------------------------------------data Connection = Connection- { cnVersion :: !Int- , cnConfig :: !HadoopConfig- , cnProtocol :: !Protocol- , invokeRaw :: !(Method -> RawRequest -> (RawResponse -> IO ()) -> IO ())- }+import System.Random (randomIO) -data Protocol = Protocol- { prName :: !Text- , prVersion :: !Int- } deriving (Eq, Ord, Show)+import Prelude -type Method = Text-type RawRequest = ByteString-type RawResponse = Either SomeException ByteString+import Data.Hadoop.Configuration+import qualified Data.Hadoop.Protobuf.Headers as P+import Data.Hadoop.Types+import Network.Hadoop.Sasl+import qualified Network.Hadoop.Stream as S+import Network.Hadoop.Types type CallId = Int @@ -77,8 +68,8 @@ , csFatalError :: !(TVar (Maybe SomeException)) } -initConnectionV9 :: HadoopConfig -> Protocol -> Socket -> IO Connection-initConnectionV9 config@HadoopConfig{..} protocol sock = do+initConnectionV9 :: HadoopConfig -> NameNode -> Protocol -> Socket -> IO Connection+initConnectionV9 config@HadoopConfig{..} nameNode protocol sock = do csStream <- S.mkSocketStream sock csClientId <- mkClientId @@ -86,7 +77,14 @@ putByteString "hrpc" putWord8 9 -- version putWord8 0 -- rpc service class (0 = default/protobuf, 1 = built-in, 2 = writable, 3 = protobuf- putWord8 0 -- auth protocol (0 = none, -33/0xDF = sasl)+ -- auth protocol (0 = none, -33/0xDF = sasl)+ putWord8 $ maybe 0 (const 0xDF) (nnPrincipal nameNode)++ case nnPrincipal nameNode of+ Nothing -> return ()+ Just principal -> saslAuth principal csStream++ S.runPut csStream $ putMessage $ delimitedBytesL (rpcRequestHeaderProto csClientId (-3)) <> delimitedBytesL (contextProto protocol hcUser) @@ -102,6 +100,95 @@ return (Connection 7 config protocol (enqueue cs)) where+ invokeSasl :: (Encode a, Decode b) => S.Stream -> a -> IO b+ invokeSasl stream msgr = do+ let message = delimitedBytesL (rpcRequestHeaderProto (ClientId "") (-33))+ <> delimitedBytesL msgr++ S.runPut stream (putMessage message)+ mget <- S.maybeGet stream $ do+ n <- fromIntegral <$> getWord32be+ -- TODO Would be nice if we didn't have to isolate here+ -- TODO and could stream instead. We could stream if we+ -- TODO were able to read the varint length prefix+ -- TODO ourselves and keep track of how many bytes were+ -- TODO remaining instead of calling `getRemaining`.+ isolate n $ do+ hdr <- decodeLengthPrefixedMessage+ msg <- case getField (P.rspStatus hdr) of+ P.Success -> Right <$> getRemaining+ _ -> return . Left . SomeException $ rspError hdr+ return (hdr, msg)++ case mget of+ Nothing -> throwIO ConnectionClosed+ Just (_, msg) ->+ case msg of+ Left ex -> throwIO ex+ Right m -> case fromDelimitedBytes m of+ Left ex -> throwIO ex+ Right x -> return x++ saslAuth :: Principal -> S.Stream -> IO ()+ saslAuth Principal{..} c = do+ negResp <- liftIO $ invokeSasl c RpcSaslProto+ { spState = putField Negotiate+ , spVersion = mempty+ , spToken = mempty+ , spAuths = mempty+ }+ case getField $ spState negResp of+ Negotiate -> return ()+ _ -> throwIO $ SomeException FailedNegotiation++ let auths = getField $ spAuths negResp+ authsMap = zip (map (getField . saMechanism) auths) auths++ res <- GS.runSASL $ do+ -- Select a common mechanism+ mSelectedMech <- GS.clientSuggestMechanism $ map (GS.Mechanism . fst) authsMap+ GS.Mechanism selectedMech <- liftIO $+ maybe (throwIO $ SomeException NoSharedMechanism) return mSelectedMech+ -- Safe because `selectedMech` was selected from one of the keys in `authsMap`+ let Just selectedAuth = lookup selectedMech authsMap+ GS.runClient (GS.Mechanism selectedMech) $ do+ GS.setProperty GS.PropertyService $ T.encodeUtf8 pService+ GS.setProperty GS.PropertyHostname $ T.encodeUtf8 pHost+ GS.setProperty GS.PropertyAuthID $ T.encodeUtf8 (authUser hcUser)+ (token,_) <- GS.step ""+ r <- liftIO $ invokeSasl c RpcSaslProto+ { spState = putField Initiate+ , spVersion = mempty+ , spToken = putField $ Just token+ , spAuths = putField $ [selectedAuth]+ }+ saslAuthLoop c r++ case res of+ Left err -> throwIO $ SomeException $ SaslException err+ Right _ -> return ()++ saslAuthLoop :: S.Stream -> RpcSaslProto -> GS.Session ()+ saslAuthLoop c r = do+ case getField $ spState r of+ Challenge -> do+ let mToken = getField $ spToken r+ case mToken of+ Nothing -> liftIO $ throwIO $ SomeException NoToken+ Just token -> do+ (resToken, _) <- GS.step token+ resp <- liftIO $ invokeSasl c RpcSaslProto+ { spToken = putField $ Just resToken+ , spState = putField $ Response+ , spVersion = mempty+ , spAuths = mempty+ }+ saslAuthLoop c resp+ Success -> return ()+ _ -> liftIO $ throwIO $ SomeException UnexptededState+++ enqueue :: ConnectionState -> Method -> RawRequest@@ -188,11 +275,11 @@ ------------------------------------------------------------------------ -contextProto :: Protocol -> User -> P.IpcConnectionContext+contextProto :: Protocol -> UserDetails -> P.IpcConnectionContext contextProto protocol user = P.IpcConnectionContext { P.ctxProtocol = putField (Just (prName protocol)) , P.ctxUserInfo = putField (Just P.UserInformation- { P.effectiveUser = putField (Just user)+ { P.effectiveUser = putField (Just (authUser user)) , P.realUser = mempty }) }
+ src/Network/Hadoop/Sasl.hs view
@@ -0,0 +1,67 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Network.Hadoop.Sasl where++import Network.Protocol.SASL.GNU+import Data.ProtocolBuffers++import Control.Exception (Exception)++import GHC.Generics+import Data.ByteString+import Data.Text+import Data.Typeable+import Data.Word++------------------------------------------------------------------------++newtype SaslException = SaslException Error+ deriving (Show, Typeable)++instance Exception SaslException++data SaslNegotiationError+ = FailedNegotiation+ | NoSharedMechanism+ | NoToken+ | UnexptededState+ deriving (Show, Typeable)++instance Exception SaslNegotiationError++rpcSaslProto :: Text+rpcSaslProto = "RpcSaslProto"++data RpcSaslProto = RpcSaslProto+ { spVersion :: Optional 1 (Value Word32)+ , spState :: Required 2 (Enumeration SaslState)+ , spToken :: Optional 3 (Value ByteString)+ , spAuths :: Repeated 4 (Message SaslAuth)+ } deriving (Eq, Ord, Show, Generic)++data SaslAuth = SaslAuth+ { saMethod :: Required 1 (Value ByteString)+ , saMechanism :: Required 2 (Value ByteString)+ , saProtocol :: Optional 3 (Value ByteString)+ , saServerId :: Optional 4 (Value ByteString)+ , saChallenge :: Optional 5 (Value ByteString)+ } deriving (Eq, Ord, Show, Generic)++data SaslState+ = Success+ | Negotiate+ | Initiate+ | Challenge+ | Response+ | Wrap+ deriving (Eq, Ord, Show, Enum, Bounded)++instance Encode SaslAuth+instance Decode SaslAuth+instance Encode RpcSaslProto+instance Decode RpcSaslProto+
src/Network/Hadoop/Stream.hs view
@@ -17,19 +17,23 @@ , close ) where -import Control.Applicative ((<$>))+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 Prelude import Data.Hadoop.Types
+ src/Network/Hadoop/Types.hs view
@@ -0,0 +1,25 @@+module Network.Hadoop.Types where++import Data.Text+import Data.ByteString+import Control.Exception+import Data.Hadoop.Types++------------------------------------------------------------------------++data Connection = Connection+ { cnVersion :: !Int+ , cnConfig :: !HadoopConfig+ , cnProtocol :: !Protocol+ , invokeRaw :: !(Method -> RawRequest -> (RawResponse -> IO ()) -> IO ())+ }++data Protocol = Protocol+ { prName :: !Text+ , prVersion :: !Int+ } deriving (Eq, Ord, Show)++type Method = Text+type RawRequest = ByteString+type RawResponse = Either SomeException ByteString+
test/Main.hs view
@@ -21,7 +21,7 @@ config :: HadoopConfig config = HadoopConfig {- hcUser = "hdfs"- , hcNameNodes = [(Endpoint "127.0.0.1" 8020)]+ hcUser = UserDetails "hdfs" Nothing+ , hcNameNodes = [NameNode (Endpoint "127.0.0.1" 8020) Nothing] , hcProxy = Just (Endpoint "127.0.0.1" 2080) }