git.iko.soy/distributed-process
on d54d67e help sign in
distributed-process/distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs
378 lines 13 kB history raw
1module Control.Distributed.Process.Backend.Azure
2 ( -- * Initialization
3 Backend(..)
4 , AzureParameters(..)
5 , defaultAzureParameters
6 , initializeBackend
7 -- * Re-exports from Azure Service Management
8 , CloudService(..)
9 , VirtualMachine(..)
10 , Endpoint(..)
11 , AzureSetup
12 , Azure.cloudServices
13 -- * Remote and local processes
14 , ProcessPair(..)
15 , RemoteProcess
16 , LocalProcess
17 , localSend
18 , localExpect
19 , remoteSend
20 , remoteThrow
21 ) where
23import System.Environment (getEnv)
24import System.FilePath ((</>), takeFileName)
25import System.Environment.Executable (getExecutablePath)
26import System.IO (stdout, hFlush)
27import Data.Binary (Binary(get, put), encode, decode)
28import Data.Digest.Pure.MD5 (md5, MD5Digest)
29import qualified Data.ByteString as BSS
30 ( ByteString
31 , length
32 , concat
33 , hPut
34 )
35import qualified Data.ByteString.Char8 as BSSC (pack)
36import qualified Data.ByteString.Lazy as BSL
37 ( ByteString
38 , readFile
39 , length
40 , fromChunks
41 , toChunks
42 , hPut
43 )
44import qualified Data.ByteString.Lazy.Char8 as BSLC (unpack)
45import Data.Typeable (Typeable)
46import Control.Applicative ((<$>), (<*>))
47import Control.Monad (void, unless)
48import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
49import Control.Exception (Exception, catches, Handler(Handler), throwIO)
50import Control.Monad.IO.Class (MonadIO, liftIO)
52-- Azure
53import Network.Azure.ServiceManagement
54 ( CloudService(..)
55 , VirtualMachine(..)
56 , Endpoint(..)
57 , AzureSetup
58 )
59import qualified Network.Azure.ServiceManagement as Azure
60 ( cloudServices
61 , azureSetup
62 , vmSshEndpoint
63 )
65-- SSH
66import qualified Network.SSH.Client.LibSSH2 as SSH
67 ( withSSH2
68 , scpSendFile
69 , withChannelBy
70 , Session
71 , readAllChannel
72 , writeAllChannel
73 , Channel
74 )
75import qualified Network.SSH.Client.LibSSH2.Foreign as SSH
76 ( openChannelSession
77 , channelExecute
78 , writeChannel
79 , readChannel
80 , channelSendEOF
81 )
82import qualified Network.SSH.Client.LibSSH2.Errors as SSH
83 ( ErrorCode
84 , NULL_POINTER
85 , getLastError
86 )
88-- CH
89import Control.Distributed.Process (Process, Closure)
90import Control.Distributed.Process.Serializable (Serializable)
91import Control.Distributed.Process.Internal.Types
92 ( messageToPayload
93 , createMessage
94 )
95import Network.Transport.Internal (encodeInt32, decodeInt32, prependLength)
97-- | Azure backend
98data Backend = Backend {
99 -- | Find virtual machines
100 findVMs :: IO [VirtualMachine]
101 -- | Copy the executable to a virtual machine
102 , copyToVM :: VirtualMachine -> IO ()
103 -- | Check the MD5 hash of the remote executable
104 , checkMD5 :: VirtualMachine -> IO Bool
105 -- | @runOnVM vm port pp@ starts a new CH node on machine @vm@ and then
106 -- runs the specified process pair. The CH node will shut down when the
107 -- /local/ process exists. @callOnVM@ returns the returned by the local
108 -- process on exit.
109 , callOnVM :: forall a. VirtualMachine -> String -> ProcessPair a -> IO a
110 -- | Create a new CH node and run the specified process.
111 -- The CH node will shut down when the /remote/ process exists. @spawnOnVM@
112 -- returns as soon as the process has been spawned.
113 , spawnOnVM :: VirtualMachine -> String -> RemoteProcess () -> IO ()
114 } deriving (Typeable)
116-- | Azure connection parameters
117data AzureParameters = AzureParameters {
118 azureSetup :: AzureSetup
119 , azureSshUserName :: FilePath
120 , azureSshPublicKey :: FilePath
121 , azureSshPrivateKey :: FilePath
122 , azureSshPassphrase :: String
123 , azureSshKnownHosts :: FilePath
124 , azureSshRemotePath :: FilePath
125 , azureSshLocalPath :: FilePath
126 }
128instance Binary AzureParameters where
129 put params = do
130 put (azureSetup params)
131 put (azureSshUserName params)
132 put (azureSshPublicKey params)
133 put (azureSshPrivateKey params)
134 put (azureSshPassphrase params)
135 put (azureSshKnownHosts params)
136 put (azureSshRemotePath params)
137 put (azureSshLocalPath params)
138 get =
139 AzureParameters <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get
141-- | Create default azure parameters
142defaultAzureParameters :: String -- ^ Azure subscription ID
143 -> FilePath -- ^ Path to X509 certificate
144 -> FilePath -- ^ Path to private key
145 -> IO AzureParameters
146defaultAzureParameters sid x509 pkey = do
147 home <- getEnv "HOME"
148 user <- getEnv "USER"
149 self <- getExecutablePath
150 setup <- Azure.azureSetup sid x509 pkey
151 return AzureParameters
152 { azureSetup = setup
153 , azureSshUserName = user
154 , azureSshPublicKey = home </> ".ssh" </> "id_rsa.pub"
155 , azureSshPrivateKey = home </> ".ssh" </> "id_rsa"
156 , azureSshPassphrase = ""
157 , azureSshKnownHosts = home </> ".ssh" </> "known_hosts"
158 , azureSshRemotePath = takeFileName self
159 , azureSshLocalPath = self
160 }
162-- | Initialize the backend
163initializeBackend :: AzureParameters -- ^ Connection parameters
164 -> String -- ^ Cloud service name
165 -> IO Backend
166initializeBackend params cloudService =
167 return Backend {
168 findVMs = apiFindVMs params cloudService
169 , copyToVM = apiCopyToVM params
170 , checkMD5 = apiCheckMD5 params
171 , callOnVM = apiCallOnVM params cloudService
172 , spawnOnVM = apiSpawnOnVM params cloudService
173 }
175-- | Find virtual machines
176apiFindVMs :: AzureParameters -> String -> IO [VirtualMachine]
177apiFindVMs params cloudService = do
178 css <- Azure.cloudServices (azureSetup params)
179 case filter ((== cloudService) . cloudServiceName) css of
180 [cs] -> return $ cloudServiceVMs cs
181 _ -> return []
183-- | Start a CH node on the given virtual machine
184apiCopyToVM :: AzureParameters -> VirtualMachine -> IO ()
185apiCopyToVM params vm =
186 void . withSSH2 params vm $ \s -> catchSshError s $
187 SSH.scpSendFile s 0o700 (azureSshLocalPath params) (azureSshRemotePath params)
189-- | Call a process on a VM
190apiCallOnVM :: AzureParameters
191 -> String
192 -> VirtualMachine
193 -> String
194 -> ProcessPair a
195 -> IO a
196apiCallOnVM params cloudService vm port ppair =
197 withSSH2 params vm $ \s -> do
198 let exe = "PATH=. " ++ azureSshRemotePath params
199 ++ " onvm run "
200 ++ " --host " ++ vmIpAddress vm
201 ++ " --port " ++ port
202 ++ " --cloud-service " ++ cloudService
203 ++ " 2>&1"
204 let paramsEnc = encode params
205 (status, r) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do
206 SSH.channelExecute ch exe
207 SSH.writeChannel ch (encodeInt32 (BSL.length rprocEnc))
208 SSH.writeAllChannel ch rprocEnc
209 SSH.writeChannel ch (encodeInt32 (BSL.length paramsEnc))
210 SSH.writeAllChannel ch paramsEnc
211 runLocalProcess (ppairLocal ppair) ch
212 if status == 0
213 then return r
214 else error "callOnVM: Non-zero exit status"
215 where
216 rprocEnc :: BSL.ByteString
217 rprocEnc = encode (ppairRemote ppair)
219apiSpawnOnVM :: AzureParameters
220 -> String
221 -> VirtualMachine
222 -> String
223 -> Closure (Backend -> Process ())
224 -> IO ()
225apiSpawnOnVM params cloudService vm port proc =
226 withSSH2 params vm $ \s -> do
227 -- TODO: reduce duplication with apiCallOnVM
228 let exe = "PATH=. " ++ azureSshRemotePath params
229 ++ " onvm run "
230 ++ " --host " ++ vmIpAddress vm
231 ++ " --port " ++ port
232 ++ " --cloud-service " ++ cloudService
233 ++ " --background "
234 ++ " 2>&1"
235 let paramsEnc = encode params
236 (status, r) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do
237 SSH.channelExecute ch exe
238 SSH.writeChannel ch (encodeInt32 (BSL.length procEnc))
239 SSH.writeAllChannel ch procEnc
240 SSH.writeChannel ch (encodeInt32 (BSL.length paramsEnc))
241 SSH.writeAllChannel ch paramsEnc
242 SSH.channelSendEOF ch
243 SSH.readAllChannel ch
244 unless (status == 0) $ error (BSLC.unpack r)
245 where
246 procEnc :: BSL.ByteString
247 procEnc = encode proc
249-- | Check the MD5 hash of the executable on the remote machine
250apiCheckMD5 :: AzureParameters -> VirtualMachine -> IO Bool
251apiCheckMD5 params vm = do
252 hash <- localHash params
253 withSSH2 params vm $ \s -> do
254 (r, _) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do
255 SSH.channelExecute ch "md5sum -c --status"
256 SSH.writeChannel ch . BSSC.pack $ show hash ++ " " ++ azureSshRemotePath params
257 SSH.channelSendEOF ch
258 SSH.readAllChannel ch
259 return (r == 0)
261withSSH2 :: AzureParameters -> VirtualMachine -> (SSH.Session -> IO a) -> IO a
262withSSH2 params (Azure.vmSshEndpoint -> Just ep) =
263 SSH.withSSH2 (azureSshKnownHosts params)
264 (azureSshPublicKey params)
265 (azureSshPrivateKey params)
266 (azureSshPassphrase params)
267 (azureSshUserName params)
268 (endpointVip ep)
269 (read $ endpointPort ep)
270withSSH2 _ vm =
271 error $ "withSSH2: No SSH endpoint for virtual machine " ++ vmName vm
273catchSshError :: SSH.Session -> IO a -> IO a
274catchSshError s io =
275 catches io [ Handler handleErrorCode
276 , Handler handleNullPointer
277 ]
278 where
279 handleErrorCode :: SSH.ErrorCode -> IO a
280 handleErrorCode _ = do
281 (_, str) <- SSH.getLastError s
282 error str
284 handleNullPointer :: SSH.NULL_POINTER -> IO a
285 handleNullPointer _ = do
286 (_, str) <- SSH.getLastError s
287 error str
289localHash :: AzureParameters -> IO MD5Digest
290localHash params = md5 <$> BSL.readFile (azureSshLocalPath params)
292--------------------------------------------------------------------------------
293-- Local and remote processes --
294--------------------------------------------------------------------------------
296-- | A process pair consists of a remote process and a local process. The local
297-- process can send messages to the remote process using 'localSend' and wait
298-- for messages from the remote process using 'localExpect'. The remote process
299-- can send messages to the local process using 'remoteSend', and wait for
300-- messages from the local process using the standard Cloud Haskell primitives.
301--
302-- See also 'callOnVM'.
303data ProcessPair a = ProcessPair {
304 ppairRemote :: RemoteProcess ()
305 , ppairLocal :: LocalProcess a
306 }
308-- | The process to run on the remote node (see 'ProcessPair' and 'callOnVM').
309type RemoteProcess a = Closure (Backend -> Process a)
311-- | The process to run on the local node (see 'ProcessPair' and 'callOnVM').
312newtype LocalProcess a = LocalProcess { unLocalProcess :: ReaderT SSH.Channel IO a }
313 deriving (Functor, Monad, MonadIO, MonadReader SSH.Channel)
315runLocalProcess :: LocalProcess a -> SSH.Channel -> IO a
316runLocalProcess = runReaderT . unLocalProcess
318-- | Send a messages from the local process to the remote process
319-- (see 'ProcessPair')
320localSend :: Serializable a => a -> LocalProcess ()
321localSend x = LocalProcess $ do
322 ch <- ask
323 liftIO $ mapM_ (SSH.writeChannel ch)
324 . prependLength
325 . messageToPayload
326 . createMessage
327 $ x
329-- | Wait for a message from the remote process (see 'ProcessPair').
330-- Note that unlike for the standard Cloud Haskell 'expect' it will result in a
331-- runtime error if the remote process sends a message of type other than @a@.
332localExpect :: Serializable a => LocalProcess a
333localExpect = LocalProcess $ do
334 ch <- ask
335 liftIO $ do
336 isE <- readIntChannel ch
337 len <- readIntChannel ch
338 msg <- readSizeChannel ch len
339 if isE /= 0
340 then error (decode msg)
341 else return (decode msg)
343-- | Send a message from the remote process to the local process (see
344-- 'ProcessPair'). Note that the remote process can use the standard Cloud
345-- Haskell primitives to /receive/ messages from the local process.
346remoteSend :: Serializable a => a -> Process ()
347remoteSend = liftIO . remoteSend' 0
349-- | If the remote process encounters an error it can use 'remoteThrow'. This
350-- will cause the exception to be raised (as a user-exception, not as the
351-- original type) in the local process (as well as in the remote process).
352remoteThrow :: Exception e => e -> IO ()
353remoteThrow e = remoteSend' 1 (show e) >> throwIO e
355remoteSend' :: Serializable a => Int -> a -> IO ()
356remoteSend' flags x = do
357 let enc = encode x
358 BSS.hPut stdout (encodeInt32 flags)
359 BSS.hPut stdout (encodeInt32 (BSL.length enc))
360 BSL.hPut stdout enc
361 hFlush stdout
363--------------------------------------------------------------------------------
364-- SSH utilities --
365--------------------------------------------------------------------------------
367readSizeChannel :: SSH.Channel -> Int -> IO BSL.ByteString
368readSizeChannel ch = go []
369 where
370 go :: [BSS.ByteString] -> Int -> IO BSL.ByteString
371 go acc 0 = return (BSL.fromChunks $ reverse acc)
372 go acc size = do
373 bs <- SSH.readChannel ch (fromIntegral (0x400 `min` size))
374 go (bs : acc) (size - BSS.length bs)
376readIntChannel :: SSH.Channel -> IO Int
377readIntChannel ch =
378 decodeInt32 . BSS.concat . BSL.toChunks <$> readSizeChannel ch 4