funflow 1.1.0 → 1.3.0

raw patch · 11 files changed

+156/−31 lines, 11 filesdep +Glob

Dependencies added: Glob

Files

• TestFunflow.hs +2/−0
• funflow.cabal +2/−1
• src/Control/Funflow.hs +3/−0
• src/Control/Funflow/Base.hs +1/−0
• src/Control/Funflow/ContentStore.hs +5/−0
• src/Control/Funflow/External.hs +2/−0
• src/Control/Funflow/External/Docker.hs +97/−27
• src/Control/Funflow/External/Executor.hs +6/−2
• src/Control/Funflow/Steps.hs +22/−1
• test/Funflow/SQLiteCoordinator.hs +2/−0
• test/Funflow/TestFlows.hs +14/−0

@@ -56,6 +56,7 @@       { _etCommand = "/run/current-system/sw/bin/echo"       , _etParams = [textParam t]       , _etWriteToStdOut = StdOutCapture+      , _etEnv = []       }     flow = exFlow >>> readString_   in withSystemTempDir "test_output_external_" $ \storeDir -> do@@ -73,6 +74,7 @@       { _etCommand = "/run/current-system/sw/bin/cat"       , _etParams = [contentParam a, contentParam b]       , _etWriteToStdOut = StdOutCapture+      , _etEnv = []       }     flow = proc (s1, s2) -> do       f1 <- writeString_ -< s1

@@ -1,5 +1,5 @@ Name:                funflow-Version:             1.1.0+Version:             1.3.0 Synopsis:            Workflows with arrows Description:         An arrow with resumable computations and logging@@ -49,6 +49,7 @@                    , Control.Funflow.ContentStore.Notify    Build-depends:                  base                    >= 4.6 && <5+               , Glob                , aeson                   >= 1.2.3.0                , async                , bytestring

@@ -1,9 +1,12 @@+{-# LANGUAGE TypeOperators #-}+ -- | Central Funflow module. -- --   This module just re-exports various other modules for convenience. module Control.Funflow   ( Base.Flow   , Base.SimpleFlow+  , type (Base.==>)   , Base.NoEffect   , Base.Flow'(..)   , Base.Cacher(..)

@@ -115,6 +115,7 @@ runNoEffect = error "Impossible!"  type SimpleFlow = Flow NoEffect SomeException+type (==>) = SimpleFlow  -- | Convert a flow to a diagram, for inspection/pretty printing toDiagram :: Flow eff ex a b -> Diagram ex a b

@@ -97,6 +97,7 @@   -- * Accessors   , itemHash   , itemPath+  , itemRelPath   , contentPath   , contentItem   , contentFilename@@ -314,6 +315,10 @@ -- | The root directory of the store. root :: ContentStore -> Path Abs Dir root = storeRoot++-- | The scoped path to a content item within the store.+itemRelPath :: Item -> Path Rel Dir+itemRelPath (Item x) = prefixHashPath itemPrefix x  -- | The store path of a completed item. itemPath :: ContentStore -> Item -> Path Abs Dir

@@ -177,6 +177,8 @@ data ExternalTask = ExternalTask {     _etCommand       :: T.Text   , _etParams        :: [Param]+    -- ^ Environment variables to set for the scope of the execution.+  , _etEnv           :: [(T.Text, Param)]   , _etWriteToStdOut :: OutputCapture } deriving (Generic, Show) 

@@ -2,6 +2,7 @@ {-# LANGUAGE FlexibleInstances     #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings     #-}+{-# LANGUAGE ViewPatterns          #-}  -- | Module supporting the use of docker containers as external tasks. --@@ -9,39 +10,106 @@ --   easier to specify certain tasks which will run inside docker containers. It --   handles constructing the call to docker, mounting input and output --   directories, and specifying the docker image and version to use.-module Control.Funflow.External.Docker where+module Control.Funflow.External.Docker+  ( Config (..)+  , toExternal+  ) where +import           Control.Arrow                    (Kleisli (..), second) import           Control.Funflow.ContentHashable import           Control.Funflow.External-import           Data.Map.Strict                 (Map)-import qualified Data.Map.Strict                 as Map-import           Data.Semigroup                  ((<>))-import qualified Data.Text                       as T-import           GHC.Generics                    (Generic)-import           System.FilePath+import           Control.Monad.Trans.State.Strict+import           Data.Semigroup                   (Semigroup, (<>))+import qualified Data.Text                        as T+import           GHC.Generics                     (Generic)  data Bind-  -- | Single input, will get mounted to @/input@ on the image.-  = SingleInput InputPath-  -- | Multiple inputs, each gets mouted into a subdirectory under-  -- @/input@ as described by the given mapping.-  | MultiInput (Map FilePath InputPath)+  -- | No inputs+  = NoInput+  -- | Single input, and the path to its mountpoint within the system.+  | SingleInput (InputPath, FilePath)+  -- | Multiple inputs.+  | MultiInput [(InputPath, FilePath)]   deriving Generic  instance ContentHashable IO Bind +instance Semigroup Bind where+  NoInput <> x = x+  x <> NoInput = x+  (SingleInput x) <> (SingleInput y) =+    MultiInput [x, y]+  (SingleInput x) <> (MultiInput m) =+    MultiInput $ x : m+  (MultiInput m) <> (SingleInput x) =+    MultiInput $ x : m+  (MultiInput m) <> (MultiInput m') =+    MultiInput $ m <> m'++instance Monoid Bind where+  mempty = NoInput+  mappend = (<>)+ data Config = Config   { image      :: T.Text   , optImageID :: Maybe T.Text-  , input      :: Bind-  , command    :: FilePath-  , args       :: [T.Text]+  , command    :: Param+  , args       :: [Param]+  , env        :: [(T.Text, Param)]+  , stdout     :: OutputCapture   } deriving Generic -instance ContentHashable IO Config+data Docker = Docker+  { dImage      :: T.Text+  , dOptImageID :: Maybe T.Text+  , dInput      :: Bind+  , dCommand    :: Param+  , dArgs       :: [Param]+  , dEnv        :: [(T.Text, Param)]+  , dStdout     :: OutputCapture+  } deriving Generic +instance ContentHashable IO Docker++-- | Convertion state for mapping from 'Config' to 'Docker'+data ConvertState = ConvertState+  { _csInput :: Bind+    -- | Fresh name generator+  , _csFresh ::  Int+  }++toDocker :: Config -> Docker+toDocker cfg = Docker+    { dImage = image cfg+    , dOptImageID = optImageID cfg+    , dInput = input'+    , dCommand = command'+    , dArgs = args'+    , dEnv = env'+    , dStdout = stdout cfg+    }+  where+    initState = ConvertState NoInput 0+    ((command', args', env'), ConvertState input' _) = flip runState initState $ do+      command'' <- transformParam (command cfg)+      args'' <- mapM transformParam (args cfg)+      env'' <- mapM (runKleisli $ second $ Kleisli transformParam) (env cfg)+      return (command'', args'', env'')+    transformParam :: Param -> State ConvertState Param+    transformParam (Param pfs) = Param <$> mapM transformParamField pfs+    transformParamField :: ParamField -> State ConvertState ParamField+    transformParamField (ParamPath ip) = do+      ConvertState input fresh <- get+      put $ ConvertState+        (input <> SingleInput (ip, mkInputPath fresh))+        (fresh + 1)+      return $ ParamText (T.pack $ mkInputPath fresh)+    transformParamField po = return po+    mkInputPath :: Int -> String+    mkInputPath x = "/input/" <> show x <> "/"+ toExternal :: Config -> ExternalTask-toExternal cfg = ExternalTask+toExternal (toDocker -> cfg) = ExternalTask   -- XXX: Allow to configure the path to the docker executable.   { _etCommand = "docker"   , _etParams =@@ -49,21 +117,23 @@       , "--user=" <> uidParam       ] ++ mounts ++       [ imageArg-      , stringParam (command cfg)-      ] ++ map textParam (args cfg)-  , _etWriteToStdOut = NoOutputCapture+      , dCommand cfg+      ] ++ dArgs cfg+  , _etEnv = dEnv cfg+  , _etWriteToStdOut = dStdout cfg   }   where     mounts = outputMount : inputMounts     mount src dst =       "--volume=" <> pathParam src <> ":" <> stringParam dst     outputMount = "--volume=" <> outParam <> ":/output"-    inputMounts = case input cfg of-      SingleInput chash -> [ mount chash "/input" ]+    inputMounts = case dInput cfg of+      NoInput -> []+      SingleInput (chash, dest) -> [ mount chash dest ]       MultiInput cmap ->-        [ mount chash ("/input" </> dest)-        | (dest, chash) <- Map.toList cmap+        [ mount chash dest+        | (chash, dest) <- cmap         ]-    imageArg = textParam $ case optImageID cfg of-      Nothing  -> image cfg-      Just id' -> image cfg <> ":" <> id'+    imageArg = textParam $ case dOptImageID cfg of+      Nothing  -> dImage cfg+      Just id' -> dImage cfg <> ":" <> id'

@@ -11,6 +11,7 @@ --   You probably want to start with 'executeLoop'. module Control.Funflow.External.Executor where +import           Control.Arrow                        (second) import           Control.Concurrent                   (threadDelay) import           Control.Concurrent.Async import           Control.Concurrent.MVar@@ -75,11 +76,12 @@         = withFollowFile fpErr stderr         . withFollowFile fpOut stdout       cmd = T.unpack $ td ^. tdTask . etCommand-      procSpec params = (proc cmd $ T.unpack <$> params)+      procSpec params textEnv = (proc cmd $ T.unpack <$> params)         { cwd = Just (fromAbsDir fp)         , close_fds = True         , std_err = UseHandle hErr         , std_out = UseHandle hOut+        , env = map (bimap T.unpack T.unpack) <$> textEnv         }       convParam = ConvParam         { convPath = pure . CS.itemPath store@@ -90,6 +92,8 @@         }     mbParams <- lift $ runMaybeT $       traverse (paramToText convParam) (td ^. tdTask . etParams)+    mbEnv <- lift $ runMaybeT $+      traverse (sequence . second (paramToText convParam))  (td ^. tdTask . etEnv)     params <- case mbParams of       Nothing     -> fail "A parameter was not ready"       Just params -> return params@@ -110,7 +114,7 @@       (Json.encode (td ^. tdTask))      start <- lift $ getTime Monotonic-    let theProc = procSpec params+    let theProc = procSpec params mbEnv     katipAddNamespace "process" . katipAddContext (sl "processId" $ show theProc) $ do       $(logTM) InfoS "Executing"       res <- lift $ tryIO $ withCreateProcess theProc $ \_ _ _ ph ->

@@ -16,6 +16,8 @@   , copyDirToStore   , copyFileToStore   , listDirContents+  , globDir+  , globDirPattern   , lookupAliasInStore   , mergeDirs   , mergeFiles@@ -59,6 +61,7 @@ import           GHC.Conc                        (threadDelay) import           Path import           Path.IO+import qualified System.FilePath.Glob            as Glob import           System.Posix.Files              (accessModes, createLink,                                                   setFileMode) import           System.Random@@ -166,6 +169,25 @@                )   ) +-- | Search for files in the directory matching the given text string, as a glob+-- pattern.+globDir :: ArrowFlow eff ex arr+        => arr (CS.Content Dir, String) [CS.Content File]+globDir = globDirPattern <<< second (arr $ Glob.simplify . Glob.compile)++-- | Search for files in the directory matching the given pattern.+globDirPattern :: ArrowFlow eff ex arr+               => arr (CS.Content Dir, Glob.Pattern) [CS.Content File]+globDirPattern = internalManipulateStore+  ( \store (dir, patt) -> let+      item = CS.contentItem dir+      itemRoot = CS.itemPath store item+    in do+      files <- mapM parseAbsFile =<< Glob.globDir1 patt (toFilePath itemRoot)+      relFiles <- for files (stripProperPrefix itemRoot)+      return ( (item :</>) <$> relFiles )+  )+ -- | Merge a number of store directories together into a single output directory. --   This uses hardlinks to avoid duplicating the data on disk. mergeDirs :: ArrowFlow eff ex arr => arr [CS.Content Dir] (CS.Content Dir)@@ -192,7 +214,6 @@     (\d inFiles -> for_ inFiles $ \inFile ->       createLink (toFilePath inFile) (toFilePath $ d </> filename inFile)     ) -< absFiles-  -- | Read the contents of the given file in the store. readString :: ArrowFlow eff ex arr => arr (CS.Content File) String

@@ -71,6 +71,7 @@   { _etCommand = "echo"   , _etWriteToStdOut = StdOutCapture   , _etParams = ["-n", fromString msg]+  , _etEnv = []   }  sleepEcho :: SimpleFlow (Double, String) CS.Item@@ -82,6 +83,7 @@       , "sleep " <> fromString (show time) <> ";"         <> "echo -n " <> fromString msg       ]+  , _etEnv = []   }  flow :: SimpleFlow () String

@@ -89,6 +89,7 @@     { _etCommand = "non-existent-executable-39fd1e85a0a05113938e0"     , _etParams = []     , _etWriteToStdOut = StdOutCapture+    , _etEnv = []     }))     `catch` arr (Left @SomeException . snd)     -< ()@@ -96,6 +97,18 @@     Left _ -> Left ()     Right _ -> Right () +-- | Test that we can provide an environment variable to an external step.+externalEnvVar :: SimpleFlow () (Either String ())+externalEnvVar = proc () -> do+  r <- readString_ <<< external (\() -> ExternalTask+    { _etCommand = "bash"+    , _etParams = [textParam "-c", textParam "echo -n $FOO"]+    , _etWriteToStdOut = StdOutCapture+    , _etEnv = [("FOO", textParam "testing")]+    }) -< ()+  returnA -< case r of+    "testing" -> Right ()+    x -> Left x  flowAssertions :: [FlowAssertion] flowAssertions =@@ -108,6 +121,7 @@   , FlowAssertion "cachingFlow" () flowCached (Just True) (return ())   , FlowAssertion "mergingStoreItems" () flowMerge (Just True) (return ())   , FlowAssertion "missingExecutable" () flowMissingExecutable (Just (Left ())) (return ())+  , FlowAssertion "externalEnvVar" () externalEnvVar (Just (Right ())) (return ())   ]  setup :: IO ()