git.iko.soy/distributed-process
on master help sign in
Basic implementation of 'run' w/closure
6418283Edsko de Vries
distributed-process-azure/distributed-process-azure.cabal+6−2
compare with master
unifiedsplit
@@ -25,7 +25,9 @@ Library
25 executable-path >= 0.0.3 && < 0.1,
25 executable-path >= 0.0.3 && < 0.1,
26 libssh2 >= 0.2 && < 0.3,
26 libssh2 >= 0.2 && < 0.3,
27 pureMD5 >= 2.1 && < 2.2,
27 pureMD5 >= 2.1 && < 2.2,
28− bytestring >= 0.9 && < 0.11
28+ bytestring >= 0.9 && < 0.11,
29+ distributed-process >= 0.2 && < 0.3,
30+ binary >= 0.5 && < 0.6
29 Exposed-modules: Control.Distributed.Process.Backend.Azure
31 Exposed-modules: Control.Distributed.Process.Backend.Azure
30 Extensions: ViewPatterns
32 Extensions: ViewPatterns
31 ghc-options: -Wall
33 ghc-options: -Wall
@@ -41,4 +43,6 @@ Executable cloud-haskell-azure
41 optparse-applicative >= 0.2 && < 0.4,
43 optparse-applicative >= 0.2 && < 0.4,
42 distributed-process >= 0.2 && < 0.3,
44 distributed-process >= 0.2 && < 0.3,
43 transformers >= 0.3 && < 0.4,
45 transformers >= 0.3 && < 0.4,
44− network-transport-tcp >= 0.2 && < 0.3
46+ network-transport-tcp >= 0.2 && < 0.3,
47+ binary >= 0.5 && < 0.6,
48+ bytestring >= 0.9 && < 0.11
distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+32−9
@@ -13,11 +13,16 @@ import System.Environment (getEnv)
13import System.FilePath ((</>), takeFileName)
13import System.FilePath ((</>), takeFileName)
14import System.Environment.Executable (getExecutablePath)
14import System.Environment.Executable (getExecutablePath)
15import System.Posix.Types (Fd)
15import System.Posix.Types (Fd)
16+import Data.Binary (encode)
16import Data.Digest.Pure.MD5 (md5, MD5Digest)
17import Data.Digest.Pure.MD5 (md5, MD5Digest)
17import qualified Data.ByteString.Lazy as BSL (readFile)
18import qualified Data.ByteString.Lazy as BSL (readFile)
19+import qualified Data.ByteString.Lazy.Char8 as BSLC (unpack)
18import Control.Applicative ((<$>))
20import Control.Applicative ((<$>))
19import Control.Monad (void)
21import Control.Monad (void)
20import Control.Exception (catches, Handler(Handler))
22import Control.Exception (catches, Handler(Handler))
23+import GHC.IO.Encoding (setForeignEncoding, char8)
24+
25+-- Azure
21import Network.Azure.ServiceManagement
26import Network.Azure.ServiceManagement
22 ( CloudService(..)
27 ( CloudService(..)
23 , VirtualMachine(..)
28 , VirtualMachine(..)
@@ -28,6 +33,8 @@ import qualified Network.Azure.ServiceManagement as Azure
28 , azureSetup
33 , azureSetup
29 , vmSshEndpoint
34 , vmSshEndpoint
30 )
35 )
36+
37+-- SSH
31import qualified Network.SSH.Client.LibSSH2 as SSH
38import qualified Network.SSH.Client.LibSSH2 as SSH
32 ( withSSH2
39 ( withSSH2
33 , execCommands
40 , execCommands
@@ -49,13 +56,22 @@ import qualified Network.SSH.Client.LibSSH2.Errors as SSH
49 , getLastError
56 , getLastError
50 )
57 )
51
58
59+-- CH
60+import Control.Distributed.Process
61+ ( Closure
62+ , Process
63+ )
64+
52-- | Azure backend
65-- | Azure backend
53data Backend = Backend {
66data Backend = Backend {
54 -- | Find virtual machines
67 -- | Find virtual machines
55 cloudServices :: IO [CloudService]
68 cloudServices :: IO [CloudService]
69+ -- | Copy the executable to a virtual machine
56− , copyToVM :: VirtualMachine -> IO ()
70+ , copyToVM :: VirtualMachine -> IO ()
71+ -- | Check the MD5 hash of the remote executable
57− , checkMD5 :: VirtualMachine -> IO Bool
72+ , checkMD5 :: VirtualMachine -> IO Bool
73+ -- | @runOnVM vm port p@ starts a CH node on port 'port' and runs 'p'
58− , runOnVM :: VirtualMachine -> IO ()
74+ , runOnVM :: VirtualMachine -> String -> Closure (Process ()) -> IO ()
59 }
75 }
60
76
61data AzureParameters = AzureParameters {
77data AzureParameters = AzureParameters {
@@ -113,16 +129,23 @@ apiCopyToVM params vm =
113 SSH.scpSendFile fd s 0o700 (azureSshLocalPath params) (azureSshRemotePath params)
129 SSH.scpSendFile fd s 0o700 (azureSshLocalPath params) (azureSshRemotePath params)
114
130
115-- | Start the executable on the remote machine
131-- | Start the executable on the remote machine
116−apiRunOnVM :: AzureParameters -> VirtualMachine -> IO ()
132+apiRunOnVM :: AzureParameters -> VirtualMachine -> String -> Closure (Process ()) -> IO ()
117−apiRunOnVM params vm@(Azure.vmSshEndpoint -> Just ep) =
133+apiRunOnVM params vm port proc =
118 void . withSSH2 params vm $ \fd s -> do
134 void . withSSH2 params vm $ \fd s -> do
119 let exe = "/home/edsko/" ++ azureSshRemotePath params
135 let exe = "/home/edsko/" ++ azureSshRemotePath params
120 ++ " onvm run "
136 ++ " onvm run "
121− ++ " --host " ++ endpointVip ep
137+ ++ " --host " ++ vmIpAddress vm
122− ++ " --port 8080 "
138+ ++ " --port " ++ port
123− ++ "2>&1"
139+ ++ " 2>&1"
124 putStrLn $ "Executing " ++ show exe
140 putStrLn $ "Executing " ++ show exe
125− r <- SSH.execCommands fd s [exe] -- ++ " >/dev/null 2>&1 &"] /usr/bin/nohup
141+ r <- SSH.withChannel (SSH.openChannelSession s) id fd s $ \ch -> do
142+ SSH.retryIfNeeded fd s $ SSH.channelExecute ch exe
143+ let enc = encode proc
144+ putStrLn $ "Sending closure of size " ++ show (length (BSLC.unpack enc))
145+ setForeignEncoding char8
146+ SSH.writeChannel ch $ BSLC.unpack enc
147+ SSH.channelSendEOF ch
148+ SSH.readAllChannel fd ch
126 print r
149 print r
127
150
128-- | Check the MD5 hash of the executable on the remote machine
151-- | Check the MD5 hash of the executable on the remote machine
distributed-process-azure/src/Control/Distributed/Process/Backend/Azure/CommandLine.hs+117−89
@@ -1,10 +1,19 @@
1+{-# LANGUAGE TemplateHaskell #-}
2+import Prelude hiding (catch)
1import System.Environment (getArgs)
3import System.Environment (getArgs)
2import System.Exit (exitSuccess, exitFailure)
4import System.Exit (exitSuccess, exitFailure)
3−import System.IO (hFlush, stdout)
5+import System.IO
6+ ( hFlush
7+ , stdout
8+ , stdin
9+ , hSetBinaryMode
10+ )
11+import Data.Binary (decode)
12+import qualified Data.ByteString.Lazy as BSL (ByteString, length, getContents)
4−import Control.Monad (unless, forM, forM_)
13+import Control.Monad (unless, forM, forM_, join)
5import Control.Monad.IO.Class (liftIO)
14import Control.Monad.IO.Class (liftIO)
6import Control.Arrow (returnA)
15import Control.Arrow (returnA)
7−import Control.Exception (throwIO)
16+import Control.Exception (throwIO, SomeException)
8import Control.Distributed.Process.Backend.Azure
17import Control.Distributed.Process.Backend.Azure
9 ( AzureParameters(azureSshUserName)
18 ( AzureParameters(azureSshUserName)
10 , defaultAzureParameters
19 , defaultAzureParameters
@@ -37,10 +46,109 @@ import Options.Applicative
37 , switch
46 , switch
38 )
47 )
39import Options.Applicative.Arrows (runA, asA)
48import Options.Applicative.Arrows (runA, asA)
40−import Control.Distributed.Process (getSelfPid, RemoteTable)
49+import Control.Distributed.Process
50+ ( getSelfPid
51+ , RemoteTable
52+ , Closure
53+ , Process
54+ , unClosure
55+ , catch
56+ )
57+import Control.Distributed.Process.Closure (remotable, mkClosure)
41import Control.Distributed.Process.Node (newLocalNode, runProcess, initRemoteTable)
58import Control.Distributed.Process.Node (newLocalNode, runProcess, initRemoteTable)
42import Network.Transport.TCP (createTransport, defaultTCPParameters)
59import Network.Transport.TCP (createTransport, defaultTCPParameters)
43
60
61+cprint :: String -> Process ()
62+cprint = liftIO . putStrLn
63+
64+remotable ['cprint]
65+
66+--------------------------------------------------------------------------------
67+-- Main --
68+--------------------------------------------------------------------------------
69+
70+main :: IO ()
71+main = do
72+ _ <- SSH.initialize True
73+ cmd <- execParser opts
74+ case cmd of
75+ List {} -> do
76+ params <- azureParameters (azureOptions cmd) Nothing
77+ backend <- initializeBackend params
78+ css <- cloudServices backend
79+ mapM_ print css
80+ CopyTo {} -> do
81+ params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))
82+ backend <- initializeBackend params
83+ css <- cloudServices backend
84+ forM_ (findTarget (target cmd) css) $ \vm -> do
85+ putStr (vmName vm ++ ": ") >> hFlush stdout
86+ copyToVM backend vm
87+ putStrLn "Done"
88+ CheckMD5 {} -> do
89+ params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))
90+ backend <- initializeBackend params
91+ css <- cloudServices backend
92+ matches <- forM (findTarget (target cmd) css) $ \vm -> do
93+ unless (status cmd) $ putStr (vmName vm ++ ": ") >> hFlush stdout
94+ match <- checkMD5 backend vm
95+ unless (status cmd) $ putStrLn $ if match then "OK" else "FAILED"
96+ return match
97+ if and matches
98+ then exitSuccess
99+ else exitFailure
100+ RunOn {} -> do
101+ params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))
102+ backend <- initializeBackend params
103+ css <- cloudServices backend
104+ forM_ (findTarget (target cmd) css) $ \vm -> do
105+ putStr (vmName vm ++ ": ") >> hFlush stdout
106+ runOnVM backend vm (port cmd) ($(mkClosure 'cprint) "Hello world")
107+ OnVmCommand (vmCmd@OnVmRun {}) -> do
108+ let rtable = __remoteTable initRemoteTable
109+ onVmRun rtable (onVmIP vmCmd) (onVmPort vmCmd)
110+ SSH.exit
111+ where
112+ opts = info (helper <*> commandParser)
113+ ( fullDesc
114+ & header "Cloud Haskell backend for Azure"
115+ )
116+
117+findTarget :: Target -> [CloudService] -> [VirtualMachine]
118+findTarget (CloudService cs) css =
119+ concatMap cloudServiceVMs . filter ((== cs) . cloudServiceName) $ css
120+findTarget (VirtualMachine virtualMachine) css =
121+ [ vm | vm <- concatMap cloudServiceVMs css
122+ , vmName vm == virtualMachine
123+ ]
124+
125+azureParameters :: AzureOptions -> Maybe SshOptions -> IO AzureParameters
126+azureParameters opts Nothing =
127+ defaultAzureParameters (subscriptionId opts)
128+ (pathToCert opts)
129+ (pathToKey opts)
130+azureParameters opts (Just sshOpts) = do
131+ params <- defaultAzureParameters (subscriptionId opts)
132+ (pathToCert opts)
133+ (pathToKey opts)
134+ return params {
135+ azureSshUserName = remoteUser sshOpts
136+ }
137+
138+onVmRun :: RemoteTable -> String -> String -> IO ()
139+onVmRun rtable host port = do
140+ hSetBinaryMode stdin True
141+ proc <- BSL.getContents :: IO BSL.ByteString
142+ putStrLn $ "Got closure encoding of length " ++ show (BSL.length proc)
143+ mTransport <- createTransport host port defaultTCPParameters
144+ case mTransport of
145+ Left err -> throwIO err
146+ Right transport -> do
147+ node <- newLocalNode transport rtable
148+ runProcess node $ do
149+ liftIO $ putStrLn "Starting remote node"
150+ catch (join . unClosure . decode $ proc) (\e -> liftIO $ print (e :: SomeException))
151+
44--------------------------------------------------------------------------------
152--------------------------------------------------------------------------------
45-- Command line options --
153-- Command line options --
46--------------------------------------------------------------------------------
154--------------------------------------------------------------------------------
@@ -81,6 +189,7 @@ data Command =
81 azureOptions :: AzureOptions
189 azureOptions :: AzureOptions
82 , sshOptions :: SshOptions
190 , sshOptions :: SshOptions
83 , target :: Target
191 , target :: Target
192+ , port :: String
84 }
193 }
85 | OnVmCommand {
194 | OnVmCommand {
86 onVmCommand :: OnVmCommand
195 onVmCommand :: OnVmCommand
@@ -167,6 +276,10 @@ runOnParser = RunOn
167 <$> azureOptionsParser
276 <$> azureOptionsParser
168 <*> sshOptionsParser
277 <*> sshOptionsParser
169 <*> targetParser
278 <*> targetParser
279+ <*> strOption ( long "port"
280+ & metavar "PORT"
281+ & help "Port number of the CH instance"
282+ )
170
283
171onVmRunParser :: Parser OnVmCommand
284onVmRunParser :: Parser OnVmCommand
172onVmRunParser = OnVmRun
285onVmRunParser = OnVmRun
@@ -185,88 +298,3 @@ onVmCommandParser = OnVmCommand <$> subparser
185 (progDesc "Run the executable"))
298 (progDesc "Run the executable"))
186 )
299 )
187
300
188−--------------------------------------------------------------------------------
189−-- Main --
190−--------------------------------------------------------------------------------
191−
192−main :: IO ()
193−main = do
194− _ <- SSH.initialize True
195− cmd <- execParser opts
196− case cmd of
197− List {} -> do
198− params <- azureParameters (azureOptions cmd) Nothing
199− backend <- initializeBackend params
200− css <- cloudServices backend
201− mapM_ print css
202− CopyTo {} -> do
203− params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))
204− backend <- initializeBackend params
205− css <- cloudServices backend
206− forM_ (findTarget (target cmd) css) $ \vm -> do
207− putStr (vmName vm ++ ": ") >> hFlush stdout
208− copyToVM backend vm
209− putStrLn "Done"
210− CheckMD5 {} -> do
211− params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))
212− backend <- initializeBackend params
213− css <- cloudServices backend
214− matches <- forM (findTarget (target cmd) css) $ \vm -> do
215− unless (status cmd) $ putStr (vmName vm ++ ": ") >> hFlush stdout
216− match <- checkMD5 backend vm
217− unless (status cmd) $ putStrLn $ if match then "OK" else "FAILED"
218− return match
219− if and matches
220− then exitSuccess
221− else exitFailure
222− RunOn {} -> do
223− params <- azureParameters (azureOptions cmd) (Just (sshOptions cmd))
224− backend <- initializeBackend params
225− css <- cloudServices backend
226− forM_ (findTarget (target cmd) css) $ \vm -> do
227− putStr (vmName vm ++ ": ") >> hFlush stdout
228− runOnVM backend vm
229− putStrLn "Done"
230− OnVmCommand (vmCmd@OnVmRun {}) -> do
231− let rtable = initRemoteTable
232− onVmRun rtable (onVmIP vmCmd) (onVmPort vmCmd)
233− SSH.exit
234− where
235− opts = info (helper <*> commandParser)
236− ( fullDesc
237− & header "Cloud Haskell backend for Azure"
238− )
239−
240−findTarget :: Target -> [CloudService] -> [VirtualMachine]
241−findTarget (CloudService cs) css =
242− concatMap cloudServiceVMs . filter ((== cs) . cloudServiceName) $ css
243−findTarget (VirtualMachine virtualMachine) css =
244− [ vm | vm <- concatMap cloudServiceVMs css
245− , vmName vm == virtualMachine
246− ]
247−
248−azureParameters :: AzureOptions -> Maybe SshOptions -> IO AzureParameters
249−azureParameters opts Nothing =
250− defaultAzureParameters (subscriptionId opts)
251− (pathToCert opts)
252− (pathToKey opts)
253−azureParameters opts (Just sshOpts) = do
254− params <- defaultAzureParameters (subscriptionId opts)
255− (pathToCert opts)
256− (pathToKey opts)
257− return params {
258− azureSshUserName = remoteUser sshOpts
259− }
260−
261−onVmRun :: RemoteTable -> String -> String -> IO ()
262−onVmRun rtable host port = do
263− mTransport <- createTransport host port defaultTCPParameters
264− case mTransport of
265− Left err -> throwIO err
266− Right transport -> do
267− node <- newLocalNode transport rtable
268− runProcess node $ do
269− pid <- getSelfPid
270− liftIO . putStrLn $ "Azure controller has pid " ++ show pid
271−
272−