diff --git a/hadoop-tools.cabal b/hadoop-tools.cabal --- a/hadoop-tools.cabal +++ b/hadoop-tools.cabal @@ -1,5 +1,5 @@ name: hadoop-tools -version: 0.3 +version: 0.4 synopsis: Fast command line tools for working with Hadoop. description: Fast command line tools for working with Hadoop. homepage: http://github.com/jystic/hadoop-tools @@ -35,7 +35,9 @@ , old-locale >= 1.0 , optparse-applicative >= 0.10 , protobuf >= 0.2.0.4 + , regex-pcre-builtin >= 0.94 , split >= 0.2 + , stm >= 2.4 , text >= 1.1 , time >= 1.4 , transformers >= 0.4 diff --git a/src/Main.hs b/src/Main.hs --- a/src/Main.hs +++ b/src/Main.hs @@ -3,7 +3,8 @@ module Main (main) where -import Control.Exception (SomeException, bracket) +import Control.Concurrent.STM +import Control.Exception (SomeException, throwIO, bracket, fromException) import Control.Monad import Control.Monad.Catch (handle, throwM) import Control.Monad.IO.Class (MonadIO, liftIO) @@ -20,25 +21,27 @@ import Data.Time import Data.Time.Clock.POSIX import qualified Data.Vector as V -import Data.Word (Word16, Word32, Word64) +import Data.Version (showVersion) +import Data.Word (Word16, Word64) import qualified Data.Configurator as C import Data.Configurator.Types (Worth(..)) +import Options.Applicative hiding (Success) import System.Environment (getEnv) -import System.FilePath.Posix +import qualified System.FilePath as FilePath +import qualified System.FilePath.Posix as Posix import System.IO import System.IO.Unsafe (unsafePerformIO) import System.Locale (defaultTimeLocale) import Text.PrettyPrint.Boxes hiding ((<>), (//)) import Data.Hadoop.Configuration (getHadoopConfig) +import Data.Hadoop.HdfsPath import Data.Hadoop.Types import Network.Hadoop.Hdfs hiding (runHdfs) import Network.Hadoop.Read -import Options.Applicative hiding (Success) - -import Data.Version (showVersion) +import qualified Glob as Glob import Paths_hadoop_tools (version) ------------------------------------------------------------------------ @@ -77,7 +80,7 @@ configPath :: FilePath configPath = unsafePerformIO $ do home <- getEnv "HOME" - return (home ".hh") + return (home `FilePath.combine` ".hh") {-# NOINLINE configPath #-} getHdfsUser :: IO (Maybe User) @@ -103,11 +106,11 @@ workingDirConfigPath :: FilePath workingDirConfigPath = unsafePerformIO $ do home <- getEnv "HOME" - return (home ".hhwd") + return (home `FilePath.combine` ".hhwd") {-# NOINLINE workingDirConfigPath #-} getDefaultWorkingDir :: MonadIO m => m HdfsPath -getDefaultWorkingDir = liftIO $ (("/user" //) . T.encodeUtf8 . hcUser) <$> getConfig +getDefaultWorkingDir = liftIO $ (("/user" ) . T.encodeUtf8 . hcUser) <$> getConfig getWorkingDir :: MonadIO m => m HdfsPath getWorkingDir = liftIO $ handle onError @@ -126,7 +129,7 @@ where getPath = if "/" `B.isPrefixOf` path then return path - else getWorkingDir >>= \pwd -> return (pwd // path) + else getWorkingDir >>= \pwd -> return (pwd path) normalizePath :: HdfsPath -> HdfsPath normalizePath = B.intercalate "/" . dropAbsParentDir . B.split '/' @@ -137,6 +140,7 @@ where go [] (".." : ys) = go [] ys go (_ : xs) (".." : ys) = go xs ys + go xs ("." : ys) = go xs ys go xs (y : ys) = go (y : xs) ys go xs [] = (xs, []) @@ -163,7 +167,7 @@ , subChDir , subChMod , subDiskUsage - -- , subFind + , subFind , subGet , subList , subMkDir @@ -218,8 +222,8 @@ modifyPerms :: Word16 -> HdfsPath -> Hdfs () modifyPerms mode path = do absPath <- getAbsolute path - info <- getFileInfo absPath - case info of + minfo <- getFileInfo absPath + case minfo of Nothing -> fail $ unwords ["No such file", B.unpack absPath] Just FileStatus{..} -> do {- @@ -238,14 +242,18 @@ subFind :: SubCommand subFind = SubCommand "find" "Recursively search a directory tree" go where - go = find <$> (argument bstr (completeDir <> help "the path to recursively search")) + go = find <$> (optional (argument bstr (completeDir <> help "the path to recursively search"))) <*> (optional (option bstr (long "name" <> metavar "FILENAME" - <> help "the file name to match exactly"))) - find path mexpr = SubHdfs $ do - absPath <- getAbsolute path - printFindResults absPath $ maybe (const True) feq mexpr + <> help "the file name to match"))) + find mpath mexpr = SubHdfs $ do + matcher <- liftIO (mkMatcher mexpr) + printFindResults (fromMaybe "" mpath) matcher - feq expr FileStatus{..} = expr == fsPath + mkMatcher :: Maybe ByteString -> IO (FileStatus -> Bool) + mkMatcher Nothing = return (const True) + mkMatcher (Just expr) = do + glob <- Glob.compile expr + return (Glob.matches glob . fsPath) subGet :: SubCommand subGet = SubCommand "get" "Get a file" go @@ -253,7 +261,7 @@ go = get <$> argument bstr (completePath <> help "source file") <*> optional (argument str (completePath <> help "destination file")) get src mdst = SubHdfs $ do - let dst = fromMaybe (takeFileName $ B.unpack src) mdst + let dst = fromMaybe (Posix.takeFileName $ B.unpack src) mdst absSrc <- getAbsolute src mReadHandle <- openRead absSrc let doRead readHandle = liftIO $ bracket @@ -265,7 +273,7 @@ subList :: SubCommand subList = SubCommand "ls" "List the contents of a directory" go where - go = ls <$> optional (argument bstr (completeDir <> help "the directory to list")) + go = ls <$> optional (argument bstr (completePath <> help "the directory to list")) ls path = SubHdfs $ printListing =<< getAbsolute (fromMaybe "" path) subMkDir :: SubCommand @@ -312,9 +320,9 @@ ------------------------------------------------------------------------ fileCompletion :: (FileType -> Bool) -> Completer -fileCompletion p = mkCompleter $ \spath -> handle ignore $ runHdfs $ do - let dir = B.pack $ fst $ splitFileName' spath - path = B.pack spath +fileCompletion p = mkCompleter $ \strPath -> handle ignore $ runHdfs $ do + let path = B.pack strPath + dir = takeParent path ls <- getListing' =<< getAbsolute dir @@ -327,12 +335,14 @@ where ignore (RemoteError _ _) = return [] - splitFileName' x = case splitFileName x of - ("./", f) -> ("", f) - (d, f) -> (d, f) +takeParent :: HdfsPath -> HdfsPath +takeParent bs = case B.elemIndexEnd '/' bs of + Nothing -> B.empty + Just 0 -> "/" + Just ix -> B.take ix bs displayPath :: HdfsPath -> FileStatus -> HdfsPath -displayPath parent file = parent // fsPath file <> suffix +displayPath parent file = parent fsPath file <> suffix where suffix = case fsFileType file of Dir -> "/" @@ -373,21 +383,50 @@ <+> col left (T.unpack . fsGroup) <+> col right (formatSize . fsLength) <+> col right (formatUTC . getModTime) - <+> col left (T.unpack . T.decodeUtf8 . fsPath) + <+> col left (ifEmpty basePath . T.unpack . T.decodeUtf8 . fsPath) + where + ifEmpty def x = if x=="" then def else x + basePath = Posix.takeFileName (B.unpack path) printFindResults :: HdfsPath -> (FileStatus -> Bool) -> Hdfs () -printFindResults path cond = handle (liftIO . printError) $ do - ls <- getListingOrFail path - V.mapM_ printMatch ls +printFindResults path cond = do + absPath <- getAbsolute path + q <- getListingRecursive absPath + liftIO $ loop q $ printMatch (replaceFullPathWithInputPath absPath path) where - printMatch :: FileStatus -> Hdfs () - printMatch fs@FileStatus{..} = do - let path' = displayPath path fs - when (cond fs) (liftIO $ B.putStrLn path') - case fsFileType of - Dir -> printFindResults path' cond - _ -> return () + loop :: TBQueue (Maybe (HdfsPath, Either SomeException (V.Vector FileStatus))) + -> (HdfsPath -> FileStatus -> IO ()) + -> IO () + loop q io = do + mx <- atomically (readTBQueue q) + case mx of + Nothing -> return () + Just (parent, x) -> do + case x of + (Left err) -> printOrThrow err + (Right ls) -> V.mapM_ (io parent) ls + loop q io + printOrThrow :: SomeException -> IO () + printOrThrow ex = case fromException ex of + Nothing -> throwIO ex + Just err -> printError err + + printMatch :: (HdfsPath -> HdfsPath) -> HdfsPath -> FileStatus -> IO () + printMatch fixup parent fs@FileStatus{..} = do + let path' = fixup (displayPath parent fs) + when (cond fs) (B.putStrLn path') + +replaceFullPathWithInputPath :: HdfsPath -> HdfsPath -> HdfsPath -> HdfsPath +replaceFullPathWithInputPath fullPath inputPath path + | fullPath' `B.isPrefixOf` path = inputPath B.drop (B.length fullPath') path + | otherwise = path + where + fullPath' = addDirSlash fullPath + + addDirSlash dir | "/" `B.isSuffixOf` dir = dir + | otherwise = dir <> "/" + ------------------------------------------------------------------------ formatSize :: Word64 -> String @@ -426,9 +465,9 @@ printError :: RemoteError -> IO () printError (RemoteError subject body) - | oneLiner = T.putStrLn firstLine - | T.null body = T.putStrLn subject - | otherwise = T.putStrLn subject >> T.putStrLn body + | oneLiner = T.hPutStrLn stderr firstLine + | T.null body = T.hPutStrLn stderr subject + | otherwise = T.hPutStrLn stderr subject >> T.putStrLn body where oneLiner = subject `elem` [ "org.apache.hadoop.security.AccessControlException" , "org.apache.hadoop.fs.FileAlreadyExistsException" @@ -437,22 +476,6 @@ isAccessDenied :: RemoteError -> Bool isAccessDenied (RemoteError s _) = s == "org.apache.hadoop.security.AccessControlException" - ------------------------------------------------------------------------- - -infixr 5 // - -(//) :: HdfsPath -> HdfsPath -> HdfsPath -(//) xs ys | B.null xs = ys - | B.null ys = xs - | B.head ys == '/' = ys - | B.last xs == '/' = xs <> ys - | otherwise = xs <> "/" <> ys - -trimEnd :: Char -> ByteString -> ByteString -trimEnd b bs | B.null bs = B.empty - | B.last bs == b = trimEnd b (B.init bs) - | otherwise = bs ------------------------------------------------------------------------