diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,31 @@ +Copyright Well-Typed LLP, 2011-2012 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of the owner nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/demos/Echo.hs b/demos/Echo.hs new file mode 100644 --- /dev/null +++ b/demos/Echo.hs @@ -0,0 +1,49 @@ +{-# LANGUAGE TemplateHaskell #-} + +import System.IO (hFlush, stdout) +import System.Environment (getArgs) +import Control.Monad (unless, forever) +import Control.Monad.IO.Class (liftIO) +import Control.Distributed.Process (Process, expect) +import Control.Distributed.Process.Closure (remotable, mkClosure) +import Control.Distributed.Process.Backend.Azure + +echoRemote :: () -> Backend -> Process () +echoRemote () _backend = forever $ do + str <- expect + remoteSend (str :: String) + +remotable ['echoRemote] + +echoLocal :: LocalProcess () +echoLocal = do + str <- liftIO $ putStr "# " >> hFlush stdout >> getLine + unless (null str) $ do + localSend str + liftIO $ putStr "Echo: " >> hFlush stdout + echo <- localExpect + liftIO $ putStrLn echo + echoLocal + +main :: IO () +main = do + args <- getArgs + case args of + "onvm":args' -> + -- Pass execution to 'onVmMain' if we are running on the VM + -- ('callOnVM' will provide the right arguments) + onVmMain __remoteTable args' + + sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do + -- Initialize the Azure backend + params <- defaultAzureParameters sid x509 pkey + let params' = params { azureSshUserName = user } + backend <- initializeBackend params' cloudService + + -- Find the specified virtual machine + Just vm <- findNamedVM backend virtualMachine + + -- Run the echo client proper + callOnVM backend vm port $ + ProcessPair ($(mkClosure 'echoRemote) ()) + echoLocal diff --git a/demos/Fib.hs b/demos/Fib.hs new file mode 100644 --- /dev/null +++ b/demos/Fib.hs @@ -0,0 +1,74 @@ +{-# LANGUAGE TemplateHaskell #-} +import System.Environment (getArgs) +import System.Random (randomIO) +import Control.Monad (forM) +import Control.Monad.IO.Class (liftIO) +import Control.Distributed.Process + ( Process + , NodeId + , SendPort + , newChan + , sendChan + , spawn + , receiveChan + , spawnLocal + ) +import Control.Distributed.Process.Backend.Azure +import Control.Distributed.Process.Closure + ( remotable + , remotableDecl + , mkClosure + ) + +randomElement :: [a] -> IO a +randomElement xs = do + ix <- randomIO + return (xs !! (ix `mod` length xs)) + +remotableDecl [ + [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ; + dfib (_, reply, 0) = sendChan reply 0 + dfib (_, reply, 1) = sendChan reply 1 + dfib (nids, reply, n) = do + nid1 <- liftIO $ randomElement nids + nid2 <- liftIO $ randomElement nids + (sport, rport) <- newChan + _ <- spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2) + _ <- spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1) + n1 <- receiveChan rport + n2 <- receiveChan rport + sendChan reply $ n1 + n2 + |] + ] + +remoteFib :: ([NodeId], Integer) -> Backend -> Process () +remoteFib (nids, n) _backend = do + (sport, rport) <- newChan + _ <- spawnLocal $ dfib (nids, sport, n) + fib_n <- receiveChan rport + mapM_ terminateNode nids + remoteSend fib_n + +remotable ['remoteFib] + +printResult :: LocalProcess () +printResult = do + result <- localExpect :: LocalProcess Integer + liftIO $ print result + +main :: IO () +main = do + args <- getArgs + case args of + "onvm":args' -> onVmMain (__remoteTable . __remoteTableDecl) args' + [sid, x509, pkey, user, cloudService, n] -> do + params <- defaultAzureParameters sid x509 pkey + let params' = params { azureSshUserName = user } + backend <- initializeBackend params' cloudService + vms <- findVMs backend + nids <- forM vms $ \vm -> spawnNodeOnVM backend vm "8080" + callOnVM backend (head vms) "8081" $ + ProcessPair ($(mkClosure 'remoteFib) (nids, read n :: Integer)) + printResult + _ -> + error "Invalid command line arguments" diff --git a/demos/Ping.hs b/demos/Ping.hs new file mode 100644 --- /dev/null +++ b/demos/Ping.hs @@ -0,0 +1,85 @@ +{-# LANGUAGE TemplateHaskell #-} + +import System.Environment (getArgs) +import Data.Binary (encode, decode) +import Control.Monad (forever) +import Control.Monad.IO.Class (liftIO) +import Control.Exception (try, IOException) +import Control.Distributed.Process + ( Process + , getSelfPid + , expect + , send + , monitor + , receiveWait + , match + , ProcessMonitorNotification(..) + ) +import Control.Distributed.Process.Closure (remotable, mkClosure) +import Control.Distributed.Process.Backend.Azure +import qualified Data.ByteString.Lazy as BSL (readFile, writeFile) + +pingServer :: () -> Backend -> Process () +pingServer () _backend = do + us <- getSelfPid + liftIO $ BSL.writeFile "pingServer.pid" (encode us) + forever $ do + them <- expect + send them () + +pingClientRemote :: () -> Backend -> Process () +pingClientRemote () _backend = do + mPingServerEnc <- liftIO $ try (BSL.readFile "pingServer.pid") + case mPingServerEnc of + Left err -> + remoteSend $ "Ping server not found: " ++ show (err :: IOException) + Right pingServerEnc -> do + let pingServerPid = decode pingServerEnc + pid <- getSelfPid + _ref <- monitor pingServerPid + send pingServerPid pid + gotReply <- receiveWait + [ match (\() -> return True) + , match (\(ProcessMonitorNotification {}) -> return False) + ] + if gotReply + then remoteSend $ "Ping server at " ++ show pingServerPid ++ " ok" + else remoteSend $ "Ping server at " ++ show pingServerPid ++ " failure" + +remotable ['pingClientRemote, 'pingServer] + +pingClientLocal :: LocalProcess () +pingClientLocal = localExpect >>= liftIO . putStrLn + +main :: IO () +main = do + args <- getArgs + case args of + "onvm":args' -> + -- Pass execution to 'onVmMain' if we are running on the VM + onVmMain __remoteTable args' + + "list":sid:x509:pkey:_ -> do + -- List all available cloud services + -- (useful, but not strictly necessary for the example) + params <- defaultAzureParameters sid x509 pkey + css <- cloudServices (azureSetup params) + mapM_ print css + + cmd:sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do + -- Initialize the backend and find the right VM + params <- defaultAzureParameters sid x509 pkey + let params' = params { azureSshUserName = user } + backend <- initializeBackend params' cloudService + Just vm <- findNamedVM backend virtualMachine + + -- The same binary can behave as the client or the server, + -- depending on the command line arguments + case cmd of + "server" -> do + pid <- spawnOnVM backend vm port ($(mkClosure 'pingServer) ()) + putStrLn $ "Ping server started at " ++ show pid + "client" -> + callOnVM backend vm port $ + ProcessPair ($(mkClosure 'pingClientRemote) ()) + pingClientLocal diff --git a/distributed-process-azure.cabal b/distributed-process-azure.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-azure.cabal @@ -0,0 +1,102 @@ +Name: distributed-process-azure +Version: 0.1.0 +Cabal-Version: >=1.8 +Build-Type: Simple +License: BSD3 +License-File: LICENSE +Copyright: Well-Typed LLP +Author: Duncan Coutts, Nicolas Wu, Edsko de Vries +Maintainer: edsko@well-typed.com, duncan@well-typed.com +Stability: experimental +Homepage: http://github.com/haskell-distributed/distributed-process +Bug-Reports: mailto:edsko@well-typed.com +Synopsis: Microsoft Azure backend for Cloud Haskell +Description: This is a proof of concept Azure backend for Cloud Haskell. It + provides just enough functionality to run Cloud Haskell + applications on Azure virtual machines. You need to create your + virtual machines in the Azure management portal; you can then + use this backend to copy or verify your executable to the + virtual machine, start or terminate Cloud Haskell nodes on those + virtual machines, and communicate with those virtual machines + from your local machine. +Category: Control + +Source-Repository head + Type: git + Location: https://github.com/haskell-distributed/distributed-process + SubDir: distributed-process-azure + +Flag build-demos + description: Build the demos + default: False + +Library + Build-Depends: base >= 4.4 && < 5, + azure-service-api >= 0.1 && < 0.2, + filepath >= 1.3 && < 1.4, + executable-path >= 0.0.3 && < 0.1, + libssh2 >= 0.2 && < 0.3, + pureMD5 >= 2.1 && < 2.2, + bytestring >= 0.9 && < 0.11, + distributed-process >= 0.3.2 && < 0.5, + binary >= 0.5 && < 0.7, + network-transport >= 0.3 && < 0.4, + network-transport-tcp >= 0.3 && < 0.4, + transformers >= 0.3 && < 0.4, + certificate >= 1.3 && < 1.4, + unix >= 2.5 && < 2.7, + mtl >= 2.1 && < 2.2, + rank1dynamic >= 0.1 && < 0.2, + distributed-static >= 0.2 && < 0.3 + Exposed-modules: Control.Distributed.Process.Backend.Azure + Extensions: ViewPatterns, + RankNTypes, + ExistentialQuantification, + ScopedTypeVariables, + DeriveDataTypeable, + GeneralizedNewtypeDeriving + ghc-options: -Wall + HS-Source-Dirs: src + +Executable cloud-haskell-azure-echo + if flag(build-demos) + Build-Depends: base >= 4.4 && < 5, + distributed-process-azure >= 0.1 && < 0.2, + distributed-process >= 0.3.2 && < 0.5, + transformers >= 0.3 && < 0.4 + else + buildable: False + Main-Is: demos/Echo.hs + ghc-options: -Wall + +Executable cloud-haskell-azure-ping + if flag(build-demos) + Build-Depends: base >= 4.4 && < 5, + distributed-process-azure >= 0.1 && < 0.2, + distributed-process >= 0.3.2 && < 0.5, + transformers >= 0.3 && < 0.4, + bytestring >= 0.9 && < 0.11, + binary >= 0.5 && < 0.7, + mtl, + libssh2 + else + buildable: False + Main-Is: demos/Ping.hs + ghc-options: -Wall + +Executable cloud-haskell-azure-fib + if flag(build-demos) + Build-Depends: base >= 4.4 && < 5, + distributed-process-azure >= 0.1 && < 0.2, + distributed-process >= 0.3.2 && < 0.5, + transformers >= 0.3 && < 0.4, + bytestring >= 0.9 && < 0.11, + binary >= 0.5 && < 0.7, + binary-generic >= 0.2 && < 0.3, + rank1dynamic >= 0.1 && < 0.2, + distributed-static >= 0.2 && < 0.3, + random >= 1.0 && < 1.1 + else + buildable: False + Main-Is: demos/Fib.hs + ghc-options: -Wall diff --git a/src/Control/Distributed/Process/Backend/Azure.hs b/src/Control/Distributed/Process/Backend/Azure.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Backend/Azure.hs @@ -0,0 +1,964 @@ +-- | This module provides the API for running Cloud Haskell on Microsoft Azure +-- virtual machines (). Virtual machines within an +-- Azure cloud service can talk to each other directly using standard Cloud +-- Haskell primitives (using TCP/IP under the hood); to talk to the remote +-- machines from your local machine you can use the primitives provided in this +-- module (which use ssh under the hood). It looks something like +-- +-- > _ _ +-- > ( ` )_ +-- > ( ) `) Azure cloud service +-- > (_ (_ . _) _) +-- > +-- > | +-- > | ssh connection +-- > | +-- > +-- > +---+ +-- > | | Local machine +-- > +---+ +-- +-- /NOTE/: It is unfortunate that the local machine cannot talk to the remote +-- machine using the standard Cloud Haskell primitives. In an ideal world, we +-- could just start a Cloud Haskell node on the local machine, too. +-- Unfortunately, Cloud Haskell does not yet support using multiple network +-- transports within the same system (i.e. both TCP/IP and SSH). This is a +-- temporary workaround. +-- +-- [Azure Setup] +-- +-- In this section we describe how to set up an Azure Cloud Service for use +-- with Cloud Haskell, starting from a brand new Azure account. It is not +-- intended as an Azure tutorial, but as a guide to making the right choices to +-- get Cloud Haskell up and running as quickly as possible. +-- +-- An Azure /Cloud Service/ is a set of virtual machines that can talk to each +-- other directly over TCP/IP (they are part of the same private network). You +-- don't create the cloud service directly; instead, after you have set up your +-- first virtual machine as a /stand alone/ virtual machine, you can /connect/ +-- subsequent virtual machines to the first virtual machine, thereby implicitly +-- setting up a Cloud Service. +-- +-- We have only tested Cloud Haskell with Linux based virtual machines; +-- Windows based virtual machines /might/ work, but you'll be entering +-- unchartered territory. Cloud Haskell assumes that all nodes run the same +-- binary code; hence, you must use the same OS on all virtual machines, +-- /as well as on your local machine/. We use Ubuntu Server 12.04 LTS for our +-- tests (running on VirtualBox on our local machine). +-- +-- When you set up your virtual machine, you can pick an arbitrary virtual +-- machine name; these names are for your own use only and do not need to be +-- globally unique. Set a username and password; you should use the same +-- username on all virtual machines. You should also upload an SSH key for +-- authentication (see +-- /Converting OpenSSH keys for use on Windows Azure Linux VM's/, +-- , for +-- information on how to convert a standard Linux @id_rsa.pub@ public key to +-- X509 format suitable for Azure). For the first VM you create select +-- /Standalone Virtual Machine/, and pick an appropriate DNS name. The DNS name +-- /does/ have to be globally unique, and will also be the name of the Cloud +-- Service. For subsequent virtual machines, select +-- /Connect to Existing Virtual Machine/ instead and then select the first VM +-- you created. +-- +-- Once your virtual machines have been set up, you have to make sure that the +-- user you created when you created the VM can ssh from any virtual machine to +-- any other using public key authentication. Moreover, you have to make sure +-- that @libssh2@ is installed; if you are using the Ubuntu image you can +-- install @libssh2@ using +-- +-- > sudo apt-get install libssh2-1 +-- +-- (TODO: if you don't install libssh2 things will break without a clear error +-- message.) +-- +-- In these notes, we assume three virtual machines called @CHDemo1@, +-- @CHDemo2@, and @CHDemo3@, all part of a @CloudHaskellDemo@ cloud service. +-- +-- [Obtaining a Management Certificate] +-- +-- Azure authentication is by means of an X509 certificate and corresponding +-- private key. /Create management certificates for Linux in Windows Azure/, +-- , +-- describes how you can create a management certificate for Azure, download it +-- as a @.publishsettings@ file, and extract an @.pfx@ file from it. You cannot +-- use this @.pfx@ directly; instead, you will need to extract an X509 +-- certificate from it and a private key in suitable format. You can use the +-- @openssl@ command line tool for both tasks; assuming that you stored the +-- @.pfx@ file as @credentials.pfx@, to extract the X509 certificate: +-- +-- > openssl pkcs12 -in credentials.pfx -nokeys -out credentials.x509 +-- +-- And to extract the private key: +-- +-- > openssl pkcs12 -in credentials.pfx -nocerts -nodes | openssl rsa -out credentials.private +-- +-- (@openssl pkcs12@ outputs the private key in PKCS#8 format (BEGIN PRIVATE +-- KEY), but we need it in PKCS#1 format (BEGIN RSA PRIVATE KEY). +-- +-- [Testing the Setup] +-- +-- Build and install the @distributed-process-azure@ package, making sure to +-- pass the @build-demos@ flag to Cabal. +-- +-- > cabal-dev install distributed-process-azure -f build-demos +-- +-- We can the @cloud-haskell-azure-ping@ demo to test our setup: +-- +-- > cloud-haskell-azure-ping list \ +-- > <> \ +-- > /path/to/credentials.x509 \ +-- > /path/to/credentials.private +-- +-- (you can find your subscription ID in the @.publishsettings@ file from the previous step). +-- If everything went well, this will output something like +-- +-- > Cloud Service "CloudHaskellDemo" +-- > VIRTUAL MACHINES +-- > Virtual Machine "CHDemo3" +-- > IP 10.119.182.127 +-- > INPUT ENDPOINTS +-- > Input endpoint "SSH" +-- > Port 50136 +-- > VIP 168.63.31.38 +-- > Virtual Machine "CHDemo2" +-- > IP 10.59.238.125 +-- > INPUT ENDPOINTS +-- > Input endpoint "SSH" +-- > Port 63365 +-- > VIP 168.63.31.38 +-- > Virtual Machine "CHDemo1" +-- > IP 10.59.224.122 +-- > INPUT ENDPOINTS +-- > Input endpoint "SSH" +-- > Port 22 +-- > VIP 168.63.31.38 +-- +-- The IP addresses listed are /internal/ IP addresses; they can be used by the +-- virtual machines to talk to each other, but not by the outside world to talk +-- to the virtual machines. To do that, you will need to use the VIP (Virtual +-- IP) address instead, which you will notice is the same for all virtual +-- machines that are part of the cloud service. The corresponding DNS name +-- (here @CloudHaskellDemo.cloudapp.net@) will also resolve to this (V)IP +-- address. To login to individual machines (through SSH) you will need to use +-- the specific port mentioned under INPUT ENDPOINTS. +-- +-- [Overview of the API] +-- +-- The Azure 'Backend' provides low-level functionality for interacting with +-- Azure virtual machines. 'findVMs' finds all currently available virtual +-- machines; 'copyToVM' copies the executable to a specified VM (recall that +-- all VMs, as well as the local machine, are assumed to run the same OS so +-- that they can all run the same binary), and 'checkMD5' checks the MD5 hash +-- of the executable on a remote machine. +-- +-- 'callOnVM' and 'spawnOnVM' deal with setting up Cloud Haskell nodes. +-- 'spawnOnVM' takes a virtual machine and a port number, as well as a +-- @RemoteProcess ()@, starts the executable on the remote node, sets up a new +-- Cloud Haskell node, and then runs the specified process. The Cloud Haskell +-- node will be shut down when the given process terminates. 'RemoteProcess' is +-- defined as +-- +-- > type RemoteProcess a = Closure (Backend -> Process a) +-- +-- (If you don't know what a 'Closure' is you should read +-- "Control.Distributed.Process.Closure".); the remote process will be supplied +-- with an Azure backend initialized with the same parameters. 'spawnOnVM' +-- returns once the Cloud Haskell node has been set up. +-- +-- 'callOnVM' is similar to 'spawnOnVM', but it takes a /pair/ of processes: +-- one to run on the remote host (on a newly created Cloud Haskell node), and +-- one to run on the local machine. In this case, the new Cloud Haskell node +-- will be terminated when the /local/ process terminates. 'callOnVM' is useful +-- because the remote process and the local process can communicate through a +-- set of primitives provided in this module ('localSend', 'localExpect', and +-- 'remoteSend' -- there is no 'remoteExpect'; instead the remote process can +-- use the standard Cloud Haskell 'expect' primitive). +-- +-- [First Example: Echo] +-- +-- When we run the @cloud-haskell-azure-echo@ demo on our local machine, it +-- starts a new Cloud Haskell node on the specified remote virtual machine. It +-- then repeatedly waits for input from the user on the local machine, sends +-- this to the remote virtual machine which will echo it back, and wait for and +-- show the echo. +-- +-- Before you can try it you will first need to copy the executable (for +-- example, using scp, although the Azure backend also provides this natively +-- in Haskell). Once that's done, you can run the demo as follows: +-- +-- > cloud-haskell-azure-echo \ +-- > <> \ +-- > /path/to/credentials.x509 \ +-- > /path/to/credentials.private \ +-- > <> \ +-- > <> \ +-- > <> \ +-- > <> +-- > # Everything I type gets echoed back +-- > Echo: Everything I type gets echoed back +-- > # Until I enter a blank line +-- > Echo: Until I enter a blank line +-- > # +-- +-- The full @echo@ demo is +-- +-- > {-# LANGUAGE TemplateHaskell #-} +-- > +-- > import System.IO (hFlush, stdout) +-- > import System.Environment (getArgs) +-- > import Control.Monad (unless, forever) +-- > import Control.Monad.IO.Class (liftIO) +-- > import Control.Distributed.Process (Process, expect) +-- > import Control.Distributed.Process.Closure (remotable, mkClosure) +-- > import Control.Distributed.Process.Backend.Azure +-- > +-- > echoRemote :: () -> Backend -> Process () +-- > echoRemote () _backend = forever $ do +-- > str <- expect +-- > remoteSend (str :: String) +-- > +-- > remotable ['echoRemote] +-- > +-- > echoLocal :: LocalProcess () +-- > echoLocal = do +-- > str <- liftIO $ putStr "# " >> hFlush stdout >> getLine +-- > unless (null str) $ do +-- > localSend str +-- > liftIO $ putStr "Echo: " >> hFlush stdout +-- > echo <- localExpect +-- > liftIO $ putStrLn echo +-- > echoLocal +-- > +-- > main :: IO () +-- > main = do +-- > args <- getArgs +-- > case args of +-- > "onvm":args' -> +-- > -- Pass execution to 'onVmMain' if we are running on the VM +-- > -- ('callOnVM' will provide the right arguments) +-- > onVmMain __remoteTable args' +-- > +-- > sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do +-- > -- Initialize the Azure backend +-- > params <- defaultAzureParameters sid x509 pkey +-- > let params' = params { azureSshUserName = user } +-- > backend <- initializeBackend params' cloudService +-- > +-- > -- Find the specified virtual machine +-- > Just vm <- findNamedVM backend virtualMachine +-- > +-- > -- Run the echo client proper +-- > callOnVM backend vm port $ +-- > ProcessPair ($(mkClosure 'echoRemote) ()) +-- > echoLocal +-- +-- The most important part of this code is the last three lines +-- +-- > callOnVM backend vm port $ +-- > ProcessPair ($(mkClosure 'echoRemote) ()) +-- > echoLocal +-- +-- 'callOnVM' creats a new Cloud Haskell node on the specified virtual machine, +-- then runs @echoRemote@ on the remote machine and @echoLocal@ on the local +-- machine. +-- +-- [Second Example: Ping] +-- +-- The second example differs from the @echo@ demo in that it uses both +-- 'callOnVM' and 'spawnOnVM'. It uses the latter to +-- install a ping server which keeps running in the background; it uses the +-- former to run a ping client which sends a request to the ping server and +-- outputs the response. +-- +-- As with the @echo@ demo, make sure to copy the executable to the remote server first. +-- Once that is done, you can start a ping server on a virtual machine using +-- +-- > cloud-haskell-azure-ping server \ +-- > <> \ +-- > /path/to/credentials.x509 \ +-- > /path/to/credentials.private \ +-- > <> \ +-- > <> \ +-- > <> \ +-- > <> +-- +-- As before, when we execute this on our local machine, it starts a new Cloud +-- Haskell node on the specified remote virtual machine and then executes the +-- ping server. Unlike with the echo example, however, this command will +-- terminate once the Cloud Haskell node has been set up, leaving the ping +-- server running in the background. +-- +-- Once the ping server is running we can run the ping client: +-- +-- > cloud-haskell-azure-ping client \ +-- > <> \ +-- > /path/to/credentials.x509 \ +-- > /path/to/credentials.private \ +-- > <> \ +-- > <> \ +-- > <> \ +-- > <> +-- > Ping server at pid://10.59.224.122:8080:0:2 ok +-- +-- Note that we must pass a different port number, because the client will run +-- within its own Cloud Haskell instance. +-- +-- The code for the @ping@ demo is similar to the @echo@ demo, but uses both +-- 'callOnVM' and 'spawnOnVM' and demonstrates a way to discover processes (in +-- this case, through a PID file). +-- +-- > {-# LANGUAGE TemplateHaskell #-} +-- > +-- > import System.Environment (getArgs) +-- > import Data.Binary (encode, decode) +-- > import Control.Monad (void, forever) +-- > import Control.Monad.IO.Class (liftIO) +-- > import Control.Exception (try, IOException) +-- > import Control.Distributed.Process +-- > ( Process +-- > , getSelfPid +-- > , expect +-- > , send +-- > , monitor +-- > , receiveWait +-- > , match +-- > , ProcessMonitorNotification(..) +-- > ) +-- > import Control.Distributed.Process.Closure (remotable, mkClosure) +-- > import Control.Distributed.Process.Backend.Azure +-- > import qualified Data.ByteString.Lazy as BSL (readFile, writeFile) +-- > +-- > pingServer :: () -> Backend -> Process () +-- > pingServer () _backend = do +-- > us <- getSelfPid +-- > liftIO $ BSL.writeFile "pingServer.pid" (encode us) +-- > forever $ do +-- > them <- expect +-- > send them () +-- > +-- > pingClientRemote :: () -> Backend -> Process () +-- > pingClientRemote () _backend = do +-- > mPingServerEnc <- liftIO $ try (BSL.readFile "pingServer.pid") +-- > case mPingServerEnc of +-- > Left err -> +-- > remoteSend $ "Ping server not found: " ++ show (err :: IOException) +-- > Right pingServerEnc -> do +-- > let pingServerPid = decode pingServerEnc +-- > pid <- getSelfPid +-- > _ref <- monitor pingServerPid +-- > send pingServerPid pid +-- > gotReply <- receiveWait +-- > [ match (\() -> return True) +-- > , match (\(ProcessMonitorNotification {}) -> return False) +-- > ] +-- > if gotReply +-- > then remoteSend $ "Ping server at " ++ show pingServerPid ++ " ok" +-- > else remoteSend $ "Ping server at " ++ show pingServerPid ++ " failure" +-- > +-- > remotable ['pingClientRemote, 'pingServer] +-- > +-- > pingClientLocal :: LocalProcess () +-- > pingClientLocal = localExpect >>= liftIO . putStrLn +-- > +-- > main :: IO () +-- > main = do +-- > args <- getArgs +-- > case args of +-- > "onvm":args' -> +-- > -- Pass execution to 'onVmMain' if we are running on the VM +-- > onVmMain __remoteTable args' +-- > +-- > "list":sid:x509:pkey:_ -> do +-- > -- List all available cloud services +-- > -- (useful, but not strictly necessary for the example) +-- > params <- defaultAzureParameters sid x509 pkey +-- > css <- cloudServices (azureSetup params) +-- > mapM_ print css +-- > +-- > cmd:sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do +-- > -- Initialize the backend and find the right VM +-- > params <- defaultAzureParameters sid x509 pkey +-- > let params' = params { azureSshUserName = user } +-- > backend <- initializeBackend params' cloudService +-- > Just vm <- findNamedVM backend virtualMachine +-- > +-- > -- The same binary can behave as the client or the server, +-- > -- depending on the command line arguments +-- > case cmd of +-- > "server" -> void $ spawnOnVM backend vm port ($(mkClosure 'pingServer) ()) +-- > "client" -> callOnVM backend vm port $ +-- > ProcessPair ($(mkClosure 'pingClientRemote) ()) +-- > pingClientLocal +module Control.Distributed.Process.Backend.Azure + ( -- * Initialization + Backend(..) + , AzureParameters(..) + , defaultAzureParameters + , initializeBackend + -- * Utilities + , findNamedVM + -- * On-VM main + , onVmMain + -- * Re-exports from Azure Service Management + , CloudService(..) + , VirtualMachine(..) + , Endpoint(..) + , AzureSetup + , Azure.cloudServices + -- * Remote and local processes + , ProcessPair(..) + , RemoteProcess + , LocalProcess + , localSend + , localExpect + , remoteSend + -- * High-level API + , spawnNodeOnVM + , terminateNode + ) where + +import Prelude hiding (catch) +import System.Environment (getEnv) +import System.FilePath ((), takeFileName) +import System.Environment.Executable (getExecutablePath) +import System.IO + ( stdout + , hFlush + , hSetBinaryMode + , stdin + , stdout + , stderr + , Handle + , hClose + ) +import qualified System.Posix.Process as Posix (forkProcess, createSession) +import Data.Maybe (listToMaybe) +import Data.Binary (Binary(get, put), encode, decode, getWord8, putWord8) +import Data.Digest.Pure.MD5 (md5, MD5Digest) +import qualified Data.ByteString as BSS + ( ByteString + , length + , concat + , hPut + , hGet + ) +import qualified Data.ByteString.Char8 as BSSC (pack) +import qualified Data.ByteString.Lazy as BSL + ( ByteString + , readFile + , length + , fromChunks + , toChunks + , hPut + , hGet + ) +import Data.Typeable (Typeable) +import Data.Foldable (forM_) +import Control.Applicative ((<$>), (<*>)) +import Control.Monad (void, when) +import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask) +import Control.Exception + ( Exception + , catches + , Handler(Handler) + , throwIO + , SomeException + ) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, putMVar) + +-- Azure +import Network.Azure.ServiceManagement + ( CloudService(..) + , VirtualMachine(..) + , Endpoint(..) + , AzureSetup + ) +import qualified Network.Azure.ServiceManagement as Azure + ( cloudServices + , azureSetup + , vmSshEndpoint + ) + +-- SSH +import qualified Network.SSH.Client.LibSSH2 as SSH + ( withSSH2 + , scpSendFile + , withChannelBy + , Session + , readAllChannel + , writeAllChannel + , Channel + ) +import qualified Network.SSH.Client.LibSSH2.Foreign as SSH + ( openChannelSession + , channelExecute + , writeChannel + , readChannel + , channelSendEOF + ) +import qualified Network.SSH.Client.LibSSH2.Errors as SSH + ( ErrorCode + , NULL_POINTER + , getLastError + ) + +-- CH +import Control.Distributed.Process + ( Process + , Closure + , RemoteTable + , catch + , unClosure + , ProcessId + , getSelfPid + , NodeId + , processNodeId + , register + , expect + , nsendRemote + ) +import Control.Distributed.Process.Serializable (Serializable) +import qualified Control.Distributed.Process.Internal.Types as CH + ( LocalNode + , LocalProcess(processQueue) + , Message + , payloadToMessage + , messageToPayload + , createMessage + ) +import Control.Distributed.Process.Node + ( runProcess + , forkProcess + , newLocalNode + , initRemoteTable + ) +import Control.Distributed.Process.Internal.CQueue (CQueue, enqueue) +import Network.Transport.TCP (createTransport, defaultTCPParameters) +import Network.Transport.Internal (encodeInt32, decodeInt32, prependLength) + +-- Static +import Control.Distributed.Static + ( Static + , registerStatic + , staticClosure + , staticLabel + ) +import Data.Rank1Dynamic (toDynamic) + +-- | Azure backend +data Backend = Backend { + -- | Find virtual machines + findVMs :: IO [VirtualMachine] + -- | Copy the executable to a virtual machine + , copyToVM :: VirtualMachine -> IO () + -- | Check the MD5 hash of the remote executable + , checkMD5 :: VirtualMachine -> IO Bool + -- | @runOnVM vm port pp@ starts a new CH node on machine @vm@ and then + -- runs the specified process pair. The CH node will shut down when the + -- /local/ process exists. @callOnVM@ returns the returned by the local + -- process on exit. + , callOnVM :: forall a. VirtualMachine -> String -> ProcessPair a -> IO a + -- | Create a new CH node and run the specified process. + -- The CH node will shut down when the /remote/ process exists. @spawnOnVM@ + -- returns as soon as the process has been spawned. + , spawnOnVM :: VirtualMachine -> String -> RemoteProcess () -> IO ProcessId + } deriving (Typeable) + +-- | Azure connection parameters +data AzureParameters = AzureParameters { + azureSetup :: AzureSetup + , azureSshUserName :: FilePath + , azureSshPublicKey :: FilePath + , azureSshPrivateKey :: FilePath + , azureSshPassphrase :: String + , azureSshKnownHosts :: FilePath + , azureSshRemotePath :: FilePath + , azureSshLocalPath :: FilePath + } + +instance Binary AzureParameters where + put params = do + put (azureSetup params) + put (azureSshUserName params) + put (azureSshPublicKey params) + put (azureSshPrivateKey params) + put (azureSshPassphrase params) + put (azureSshKnownHosts params) + put (azureSshRemotePath params) + put (azureSshLocalPath params) + get = + AzureParameters <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get + +-- | Create default azure parameters +defaultAzureParameters :: String -- ^ Azure subscription ID + -> FilePath -- ^ Path to X509 certificate + -> FilePath -- ^ Path to private key + -> IO AzureParameters +defaultAzureParameters sid x509 pkey = do + home <- getEnv "HOME" + user <- getEnv "USER" + self <- getExecutablePath + setup <- Azure.azureSetup sid x509 pkey + return AzureParameters + { azureSetup = setup + , azureSshUserName = user + , azureSshPublicKey = home ".ssh" "id_rsa.pub" + , azureSshPrivateKey = home ".ssh" "id_rsa" + , azureSshPassphrase = "" + , azureSshKnownHosts = home ".ssh" "known_hosts" + , azureSshRemotePath = takeFileName self + , azureSshLocalPath = self + } + +-- | Initialize the backend +initializeBackend :: AzureParameters -- ^ Connection parameters + -> String -- ^ Cloud service name + -> IO Backend +initializeBackend params cloudService = + return Backend { + findVMs = apiFindVMs params cloudService + , copyToVM = apiCopyToVM params + , checkMD5 = apiCheckMD5 params + , callOnVM = apiCallOnVM params cloudService + , spawnOnVM = apiSpawnOnVM params cloudService + } + +-- | Find virtual machines +apiFindVMs :: AzureParameters -> String -> IO [VirtualMachine] +apiFindVMs params cloudService = do + css <- Azure.cloudServices (azureSetup params) + case filter ((== cloudService) . cloudServiceName) css of + [cs] -> return $ cloudServiceVMs cs + _ -> return [] + +-- | Start a CH node on the given virtual machine +apiCopyToVM :: AzureParameters -> VirtualMachine -> IO () +apiCopyToVM params vm = + void . withSSH2 params vm $ \s -> catchSshError s $ + SSH.scpSendFile s 0o700 (azureSshLocalPath params) (azureSshRemotePath params) + +-- | Call a process on a VM +apiCallOnVM :: AzureParameters + -> String + -> VirtualMachine + -> String + -> ProcessPair a + -> IO a +apiCallOnVM = runOnVM False + +apiSpawnOnVM :: AzureParameters + -> String + -> VirtualMachine + -> String + -> Closure (Backend -> Process ()) + -> IO ProcessId +apiSpawnOnVM params cloudService vm port rproc = + runOnVM True params cloudService vm port $ + ProcessPair rproc localExpect + +-- | Internal generalization of 'spawnOnVM' and 'callOnVM' +runOnVM :: Bool + -> AzureParameters + -> String + -> VirtualMachine + -> String + -> ProcessPair a + -> IO a +runOnVM bg params cloudService vm port ppair = + withSSH2 params vm $ \s -> do + -- TODO: reduce duplication with apiCallOnVM + let exe = "PATH=. " ++ azureSshRemotePath params + ++ " onvm" + ++ " " ++ vmIpAddress vm + ++ " " ++ port + ++ " " ++ cloudService + ++ " " ++ show bg + ++ " 2>&1" + let paramsEnc = encode params + let rprocEnc = encode (ppairRemote ppair) + (status, r) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do + SSH.channelExecute ch exe + SSH.writeChannel ch (encodeInt32 (BSL.length rprocEnc)) + SSH.writeAllChannel ch rprocEnc + SSH.writeChannel ch (encodeInt32 (BSL.length paramsEnc)) + SSH.writeAllChannel ch paramsEnc + runLocalProcess (ppairLocal ppair) ch + if status == 0 + then return r + else error "runOnVM: Non-zero exit status" -- This would a bug + +-- | Check the MD5 hash of the executable on the remote machine +apiCheckMD5 :: AzureParameters -> VirtualMachine -> IO Bool +apiCheckMD5 params vm = do + hash <- localHash params + withSSH2 params vm $ \s -> do + (r, _) <- SSH.withChannelBy (SSH.openChannelSession s) id $ \ch -> do + SSH.channelExecute ch "md5sum -c --status" + SSH.writeChannel ch . BSSC.pack $ show hash ++ " " ++ azureSshRemotePath params + SSH.channelSendEOF ch + SSH.readAllChannel ch + return (r == 0) + +withSSH2 :: AzureParameters -> VirtualMachine -> (SSH.Session -> IO a) -> IO a +withSSH2 params (Azure.vmSshEndpoint -> Just ep) = + SSH.withSSH2 (azureSshKnownHosts params) + (azureSshPublicKey params) + (azureSshPrivateKey params) + (azureSshPassphrase params) + (azureSshUserName params) + (endpointVip ep) + (read $ endpointPort ep) +withSSH2 _ vm = + error $ "withSSH2: No SSH endpoint for virtual machine " ++ vmName vm + +catchSshError :: SSH.Session -> IO a -> IO a +catchSshError s io = + catches io [ Handler handleErrorCode + , Handler handleNullPointer + ] + where + handleErrorCode :: SSH.ErrorCode -> IO a + handleErrorCode _ = do + (_, str) <- SSH.getLastError s + error str + + handleNullPointer :: SSH.NULL_POINTER -> IO a + handleNullPointer _ = do + (_, str) <- SSH.getLastError s + error str + +localHash :: AzureParameters -> IO MD5Digest +localHash params = md5 <$> BSL.readFile (azureSshLocalPath params) + +-------------------------------------------------------------------------------- +-- Utilities -- +-------------------------------------------------------------------------------- + +-- | Find a virtual machine with a particular name +findNamedVM :: Backend -> String -> IO (Maybe VirtualMachine) +findNamedVM backend vm = + listToMaybe . filter ((== vm) . vmName) <$> findVMs backend + +-------------------------------------------------------------------------------- +-- Local and remote processes -- +-------------------------------------------------------------------------------- + +-- | A process pair consists of a remote process and a local process. The local +-- process can send messages to the remote process using 'localSend' and wait +-- for messages from the remote process using 'localExpect'. The remote process +-- can send messages to the local process using 'remoteSend', and wait for +-- messages from the local process using the standard Cloud Haskell primitives. +-- +-- See also 'callOnVM'. +data ProcessPair a = ProcessPair { + ppairRemote :: RemoteProcess () + , ppairLocal :: LocalProcess a + } + +-- | The process to run on the remote node (see 'ProcessPair' and 'callOnVM'). +type RemoteProcess a = Closure (Backend -> Process a) + +-- | The process to run on the local node (see 'ProcessPair' and 'callOnVM'). +newtype LocalProcess a = LocalProcess { unLocalProcess :: ReaderT SSH.Channel IO a } + deriving (Functor, Monad, MonadIO, MonadReader SSH.Channel) + +runLocalProcess :: LocalProcess a -> SSH.Channel -> IO a +runLocalProcess = runReaderT . unLocalProcess + +-- | Send a messages from the local process to the remote process +-- (see 'ProcessPair') +localSend :: Serializable a => a -> LocalProcess () +localSend x = LocalProcess $ do + ch <- ask + liftIO $ mapM_ (SSH.writeChannel ch) + . prependLength + . CH.messageToPayload + . CH.createMessage + $ x + +-- | Wait for a message from the remote process (see 'ProcessPair'). +-- Note that unlike for the standard Cloud Haskell 'expect' it will result in a +-- runtime error if the remote process sends a message of type other than @a@. +-- +-- Since it is relatively easy for the remote process to mess up the +-- communication protocol (for instance, by doing a putStr) we ask for the +-- length twice, as some sort of sanity check. +localExpect :: Serializable a => LocalProcess a +localExpect = LocalProcess $ do + ch <- ask + liftIO $ do + isE <- readIntChannel ch + len <- readIntChannel ch + lenAgain <- readIntChannel ch + when (len /= lenAgain) $ throwIO (userError "Internal error: protocol violation (perhaps the remote binary is not installed correctly?)") + msg <- readSizeChannel ch len + if isE /= 0 + then error (decode msg) + else return (decode msg) + +-- | Send a message from the remote process to the local process (see +-- 'ProcessPair'). Note that the remote process can use the standard Cloud +-- Haskell primitives to /receive/ messages from the local process. +remoteSend :: Serializable a => a -> Process () +remoteSend = liftIO . remoteSend' + +remoteSend' :: Serializable a => a -> IO () +remoteSend' = remoteSendFlagged 0 + +-- | If the remote process encounters an error it can use 'remoteThrow'. This +-- will cause the exception to be raised (as a user-exception, not as the +-- original type) in the local process (as well as in the remote process). +remoteThrow :: Exception e => e -> IO () +remoteThrow e = remoteSendFlagged 1 (show e) >> throwIO e + +remoteSendFlagged :: Serializable a => Int -> a -> IO () +remoteSendFlagged flags x = do + let enc = encode x + BSS.hPut stdout (encodeInt32 flags) + -- See 'localExpect' for why we send the length twice + BSS.hPut stdout (encodeInt32 (BSL.length enc)) + BSS.hPut stdout (encodeInt32 (BSL.length enc)) + BSL.hPut stdout enc + hFlush stdout + +-------------------------------------------------------------------------------- +-- On-VM main -- +-------------------------------------------------------------------------------- + +-- | Program main when run on the VM. A typical 'main' function looks like +-- +-- > main :: IO () +-- > main = do +-- > args <- getArgs +-- > case args of +-- > "onvm":args' -> onVmMain __remoteTable args' +-- > _ -> -- your normal main +onVmMain :: (RemoteTable -> RemoteTable) -> [String] -> IO () +onVmMain rtable [host, port, cloudService, bg] = do + hSetBinaryMode stdin True + hSetBinaryMode stdout True + Just rprocEnc <- getWithLength stdin + Just paramsEnc <- getWithLength stdin + backend <- initializeBackend (decode paramsEnc) cloudService + let rproc = decode rprocEnc + lprocMVar <- newEmptyMVar :: IO (MVar CH.LocalProcess) + if read bg + then + void . Posix.forkProcess $ do + -- We inherit the file descriptors from the parent, so the SSH + -- session will not be terminated until we close them + void Posix.createSession + startCH rproc lprocMVar backend + (\node proc -> runProcess node $ do + us <- getSelfPid + liftIO $ do + remoteSend' us + mapM_ hClose [stdin, stdout, stderr] + proc) + else do + startCH rproc lprocMVar backend forkProcess + lproc <- readMVar lprocMVar + queueFromHandle stdin (CH.processQueue lproc) + where + startCH :: RemoteProcess () + -> MVar CH.LocalProcess + -> Backend + -> (CH.LocalNode -> Process () -> IO a) + -> IO () + startCH rproc lprocMVar backend go = do + mTransport <- createTransport host port defaultTCPParameters + case mTransport of + Left err -> remoteThrow err + Right transport -> do + node <- newLocalNode transport (rtable . __remoteTable $ initRemoteTable) + void . go node $ do + ask >>= liftIO . putMVar lprocMVar + proc <- unClosure rproc :: Process (Backend -> Process ()) + catch (proc backend) + (liftIO . (remoteThrow :: SomeException -> IO ())) +onVmMain _ _ + = error "Invalid arguments passed on onVmMain" + +-- | Read a 4-byte length @l@ and then an @l@-byte payload +-- +-- Returns Nothing on EOF +getWithLength :: Handle -> IO (Maybe BSL.ByteString) +getWithLength h = do + lenEnc <- BSS.hGet h 4 + if BSS.length lenEnc < 4 + then return Nothing + else do + let len = decodeInt32 lenEnc + bs <- BSL.hGet h len + if BSL.length bs < fromIntegral len + then return Nothing + else return (Just bs) + +queueFromHandle :: Handle -> CQueue CH.Message -> IO () +queueFromHandle h q = do + mPayload <- getWithLength stdin + forM_ mPayload $ \payload -> do + enqueue q $ CH.payloadToMessage (BSL.toChunks payload) + queueFromHandle h q + +-------------------------------------------------------------------------------- +-- SSH utilities -- +-------------------------------------------------------------------------------- + +readSizeChannel :: SSH.Channel -> Int -> IO BSL.ByteString +readSizeChannel ch = go [] + where + go :: [BSS.ByteString] -> Int -> IO BSL.ByteString + go acc 0 = return (BSL.fromChunks $ reverse acc) + go acc size = do + bs <- SSH.readChannel ch (fromIntegral (0x400 `min` size)) + go (bs : acc) (size - BSS.length bs) + +readIntChannel :: SSH.Channel -> IO Int +readIntChannel ch = + decodeInt32 . BSS.concat . BSL.toChunks <$> readSizeChannel ch 4 + +-------------------------------------------------------------------------------- +-- High-level API -- +-------------------------------------------------------------------------------- + +data ServiceProcessMsg = + ServiceProcessTerminate + deriving Typeable + +instance Binary ServiceProcessMsg where + put ServiceProcessTerminate = putWord8 0 + get = do + header <- getWord8 + case header of + 0 -> return ServiceProcessTerminate + _ -> fail "ServiceProcessMsg.get" + +serviceProcess :: Backend -> Process () +serviceProcess _backend = do + us <- getSelfPid + register "$azureBackendServiceProcess" us + go + where + go = do + msg <- expect + case msg of + ServiceProcessTerminate -> + return () + +serviceProcessStatic :: Static (Backend -> Process ()) +serviceProcessStatic = staticLabel "serviceProcess" + +-- | Start a new Cloud Haskell node on the given virtual machine +spawnNodeOnVM :: Backend -> VirtualMachine -> String -> IO NodeId +spawnNodeOnVM backend vm port = + processNodeId <$> spawnOnVM backend vm port (staticClosure serviceProcessStatic) + +-- | Terminate a node started with 'spawnNodeOnVM' +terminateNode :: NodeId -> Process () +terminateNode nid = nsendRemote nid "$azureBackendServiceProcess" ServiceProcessTerminate + +__remoteTable :: RemoteTable -> RemoteTable +__remoteTable = registerStatic "serviceProcess" (toDynamic serviceProcess)