diff --git a/TestFunflow.hs b/TestFunflow.hs --- a/TestFunflow.hs +++ b/TestFunflow.hs @@ -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 diff --git a/funflow.cabal b/funflow.cabal --- a/funflow.cabal +++ b/funflow.cabal @@ -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 diff --git a/src/Control/Funflow.hs b/src/Control/Funflow.hs --- a/src/Control/Funflow.hs +++ b/src/Control/Funflow.hs @@ -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(..) diff --git a/src/Control/Funflow/Base.hs b/src/Control/Funflow/Base.hs --- a/src/Control/Funflow/Base.hs +++ b/src/Control/Funflow/Base.hs @@ -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 diff --git a/src/Control/Funflow/ContentStore.hs b/src/Control/Funflow/ContentStore.hs --- a/src/Control/Funflow/ContentStore.hs +++ b/src/Control/Funflow/ContentStore.hs @@ -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 diff --git a/src/Control/Funflow/External.hs b/src/Control/Funflow/External.hs --- a/src/Control/Funflow/External.hs +++ b/src/Control/Funflow/External.hs @@ -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) diff --git a/src/Control/Funflow/External/Docker.hs b/src/Control/Funflow/External/Docker.hs --- a/src/Control/Funflow/External/Docker.hs +++ b/src/Control/Funflow/External/Docker.hs @@ -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' diff --git a/src/Control/Funflow/External/Executor.hs b/src/Control/Funflow/External/Executor.hs --- a/src/Control/Funflow/External/Executor.hs +++ b/src/Control/Funflow/External/Executor.hs @@ -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 -> diff --git a/src/Control/Funflow/Steps.hs b/src/Control/Funflow/Steps.hs --- a/src/Control/Funflow/Steps.hs +++ b/src/Control/Funflow/Steps.hs @@ -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 diff --git a/test/Funflow/SQLiteCoordinator.hs b/test/Funflow/SQLiteCoordinator.hs --- a/test/Funflow/SQLiteCoordinator.hs +++ b/test/Funflow/SQLiteCoordinator.hs @@ -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 diff --git a/test/Funflow/TestFlows.hs b/test/Funflow/TestFlows.hs --- a/test/Funflow/TestFlows.hs +++ b/test/Funflow/TestFlows.hs @@ -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 ()