angel 0.4.3 → 0.4.4
raw patch · 18 files changed
+770/−625 lines, 18 files
Files
- Angel/Config.hs +0/−171
- Angel/Data.hs +0/−44
- Angel/Files.hs +0/−34
- Angel/Job.hs +0/−190
- Angel/Log.hs +0/−19
- Angel/Main.hs +0/−71
- Angel/PidFile.hs +0/−52
- Angel/Util.hs +0/−32
- README.md +42/−9
- angel.cabal +4/−3
- src/Angel/Config.hs +199/−0
- src/Angel/Data.hs +68/−0
- src/Angel/Files.hs +44/−0
- src/Angel/Job.hs +218/−0
- src/Angel/Log.hs +20/−0
- src/Angel/Main.hs +85/−0
- src/Angel/PidFile.hs +52/−0
- src/Angel/Util.hs +38/−0
− Angel/Config.hs
@@ -1,171 +0,0 @@-module Angel.Config where--import Control.Exception (try, SomeException)-import qualified Data.Map as M-import Control.Monad (when, mapM_, (>=>))-import Control.Concurrent.STM-import Control.Concurrent.STM.TVar (readTVar, writeTVar)-import Data.Configurator (load, getMap, Worth(..))-import Data.Configurator.Types (Config, Value(..), Name)-import qualified Data.Traversable as T-import qualified Data.HashMap.Lazy as HM-import Data.String.Utils (split)-import Data.List (foldl')-import Data.Maybe (isJust, isNothing)-import Data.Monoid ((<>))-import qualified Data.Text as T--import Angel.Job (syncSupervisors)-import Angel.Data-import Angel.Log (logger)-import Angel.Util (waitForWake,- expandPath)--import Debug.Trace (trace)--void :: Monad m => m a -> m ()-void m = m >> return ()---- |produce a mapping of name -> program for every program-buildConfigMap :: HM.HashMap Name Value -> IO SpecKey-buildConfigMap cfg = - return $! HM.foldlWithKey' addToMap M.empty cfg- where- addToMap :: SpecKey -> Name -> Value -> SpecKey- addToMap m- (split "." . T.unpack -> [basekey, localkey])- value =- let !newprog = case M.lookup basekey m of- Just prog -> modifyProg prog localkey value- Nothing -> modifyProg defaultProgram{name=basekey} localkey value- in- M.insert basekey newprog m- addToMap m _ _ = m--checkConfigValues :: SpecKey -> IO SpecKey-checkConfigValues progs = (mapM_ checkProgram $ M.elems progs) >> (return progs)- where- checkProgram p = do- when (isNothing $ exec p) $ error $ name p ++ " does not have an 'exec' specification"- when ((isJust $ logExec p) &&- (isJust (stdout p) || isJust (stderr p) )) $ error $ name p ++ " cannot have both a logger process and stderr/stdout"--modifyProg :: Program -> String -> Value -> Program-modifyProg prog "exec" (String s) = prog{exec = Just (T.unpack s)}-modifyProg prog "exec" _ = error "wrong type for field 'exec'; string required"--modifyProg prog "delay" (Number n) | n < 0 = error "delay value must be >= 0"- | otherwise = prog{delay = Just $ round n}-modifyProg prog "delay" _ = error "wrong type for field 'delay'; integer"--modifyProg prog "stdout" (String s) = prog{stdout = Just (T.unpack s)}-modifyProg prog "stdout" _ = error "wrong type for field 'stdout'; string required"--modifyProg prog "stderr" (String s) = prog{stderr = Just (T.unpack s)}-modifyProg prog "stderr" _ = error "wrong type for field 'stderr'; string required"--modifyProg prog "directory" (String s) = prog{workingDir = (Just $ T.unpack s)}-modifyProg prog "directory" _ = error "wrong type for field 'directory'; string required"--modifyProg prog "logger" (String s) = prog{logExec = (Just $ T.unpack s)}-modifyProg prog "logger" _ = error "wrong type for field 'logger'; string required"--modifyProg prog "pidfile" (String s) = prog{pidFile = (Just $ T.unpack s)}-modifyProg prog "pidfile" _ = error "wrong type for field 'pidfile'; string required"--modifyProg prog n _ = prog----- |invoke the parser to process the file at configPath--- |produce a SpecKey-processConfig :: String -> IO (Either String SpecKey)-processConfig configPath = do - mconf <- try $ process =<< load [Required configPath]-- case mconf of- Right config -> return $ Right config- Left (e :: SomeException) -> return $ Left $ show e- where process = getMap >=>- return . expandByCount >=>- buildConfigMap >=>- expandPaths >=>- checkConfigValues---- |preprocess config into multiple programs if "count" is specified-expandByCount :: HM.HashMap Name Value -> HM.HashMap Name Value-expandByCount cfg = HM.unions expanded- where expanded :: [HM.HashMap Name Value]- expanded = concat $ HM.foldlWithKey' expand' [] groupedByProgram- expand' :: [[HM.HashMap Name Value]] -> Name -> HM.HashMap Name Value -> [[HM.HashMap Name Value]]- expand' acc = fmap (:acc) . expand- groupedByProgram :: HM.HashMap Name (HM.HashMap Name Value)- groupedByProgram = HM.foldlWithKey' binByProg HM.empty cfg- binByProg h- (T.split (== '.') -> [prog, k])- v = HM.insertWith HM.union prog (HM.singleton k v) h- binByProg h _ _ = h- expand :: Name -> HM.HashMap Name Value -> [HM.HashMap Name Value]- expand prog pcfg = maybe [reflatten prog pcfg]- expandWithCount- (HM.lookup "count" pcfg)- where expandWithCount (Number n)- | n >= 0 = [ reflatten (genProgName i) (rewriteConfig i pcfg) | i <- [1..n] ]- | otherwise = error "count must be >= 0"- expandWithCount _ = error "count must be a number or not specified"- genProgName i = prog <> "-" <> textNumber i--rewriteConfig :: Rational -> HM.HashMap Name Value -> HM.HashMap Name Value-rewriteConfig n = HM.adjust rewritePidfile "pidfile"- where rewritePidfile (String path) = String $ rewrittenFilename <> extension- where rewrittenFilename = filename <> "-" <> textNumber n- (filename, extension) = T.breakOn "." path- rewritePidfile x = x--textNumber :: Rational -> T.Text-textNumber = T.pack . show . truncate--reflatten :: Name -> HM.HashMap Name Value -> HM.HashMap Name Value-reflatten prog pcfg = HM.fromList asList- where asList = map prependKey $ filter notCount $ HM.toList pcfg- prependKey (k, v) = ((prog <> "." <> k), v)- notCount = not . (== "count") . fst---- |given a new SpecKey just parsed from the file, update the --- |shared state TVar-updateSpecConfig :: TVar GroupConfig -> SpecKey -> STM ()-updateSpecConfig sharedGroupConfig spec = do - cfg <- readTVar sharedGroupConfig- writeTVar sharedGroupConfig cfg{spec=spec}---- |read the config file, update shared state with current spec, --- |re-sync running supervisors, wait for the HUP TVar, then repeat!-monitorConfig :: String -> TVar GroupConfig -> TVar (Maybe Int) -> IO ()-monitorConfig configPath sharedGroupConfig wakeSig = do - let log = logger "config-monitor"- mspec <- processConfig configPath- case mspec of - Left e -> do - log $ " <<<< Config Error >>>>\n" ++ e- log " <<<< Config Error: Skipping reload >>>>"- Right spec -> do - print spec- atomically $ updateSpecConfig sharedGroupConfig spec- syncSupervisors sharedGroupConfig- waitForWake wakeSig- log "HUP caught, reloading config"--expandPaths :: SpecKey -> IO SpecKey-expandPaths = T.mapM expandProgramPaths--expandProgramPaths :: Program -> IO Program-expandProgramPaths prog = do exec' <- maybeExpand $ exec prog- stdout' <- maybeExpand $ stdout prog- stderr' <- maybeExpand $ stderr prog- workingDir' <- maybeExpand $ workingDir prog- pidFile' <- maybeExpand $ pidFile prog- return prog { exec = exec',- stdout = stdout',- stderr = stderr',- workingDir = workingDir',- pidFile = pidFile' }- where maybeExpand = T.traverse expandPath
− Angel/Data.hs
@@ -1,44 +0,0 @@-module Angel.Data where--import qualified Data.Map as M-import System.Process (createProcess, proc, waitForProcess, ProcessHandle)-import System.Process (terminateProcess, CreateProcess(..), StdStream(..))-import Control.Concurrent.STM.TChan (TChan)-import System.IO (Handle)---- |the whole shared state of the program; spec is what _should_--- |be running, while running is what actually _is_ running_ currently-data GroupConfig = GroupConfig {- spec :: SpecKey,- running :: RunKey,- fileRequest :: TChan FileRequest-}---- |map program ids to relevant structure-type SpecKey = M.Map ProgramId Program-type RunKey = M.Map ProgramId (Program, Maybe ProcessHandle)-type ProgramId = String-type FileRequest = (String, TChan (Maybe Handle))---- |the representation of a program is these 6 values, --- |read from the config file-data Program = Program {- name :: String,- exec :: Maybe String,- delay :: Maybe Int,- stdout :: Maybe String,- stderr :: Maybe String,- workingDir :: Maybe FilePath,- logExec :: Maybe String,- pidFile :: Maybe FilePath-} deriving (Show, Eq, Ord)---- |Lower-level atoms in the configuration process-type Spec = [Program]---- |a template for an empty program; the variable set to ""--- |are required, and must be overridden in the config file-defaultProgram = Program "" Nothing Nothing Nothing Nothing Nothing Nothing Nothing-defaultDelay = 5 :: Int-defaultStdout = "/dev/null"-defaultStderr = "/dev/null"
− Angel/Files.hs
@@ -1,34 +0,0 @@-module Angel.Files (getFile, startFileManager) where--import Control.Exception (try, SomeException)-import Control.Concurrent.STM-import Control.Concurrent.STM.TChan (readTChan, writeTChan, TChan, newTChan, newTChanIO)-import Control.Monad (forever)-import System.IO (Handle, hClose, openFile, IOMode(..), hIsClosed)-import GHC.IO.Handle (hDuplicate)-import Debug.Trace (trace)-import Angel.Data (GroupConfig(..), FileRequest)--startFileManager req = forever $ fileManager req--fileManager :: TChan FileRequest -> IO ()-fileManager req = do - (path, resp) <- atomically $ readTChan req- mh <- try $ openFile path AppendMode- case mh of- Right hand -> do- hand' <- hDuplicate hand- hClose hand- atomically $ writeTChan resp (Just hand')- Left (e :: SomeException) -> atomically $ writeTChan resp Nothing- fileManager req--getFile :: String -> GroupConfig -> IO Handle-getFile path cfg = do- resp <- newTChanIO- atomically $ writeTChan (fileRequest cfg) (path, resp)- mh <- atomically $ readTChan resp- hand <- case mh of- Just hand -> return hand- Nothing -> error $ "could not open stdout/stderr file " ++ path- return hand
− Angel/Job.hs
@@ -1,190 +0,0 @@-module Angel.Job where--import Control.Exception (finally)-import Data.String.Utils (split, strip)-import Data.Maybe (isJust, fromJust, fromMaybe, mapMaybe)-import System.Process (createProcess, proc, waitForProcess, ProcessHandle)-import System.Process (terminateProcess, CreateProcess(..), StdStream(..))-import Control.Concurrent-import Control.Concurrent.STM-import Control.Concurrent.STM.TVar (readTVar, writeTVar)-import qualified Data.Map as M-import Control.Monad (unless, when, forever)--import Angel.Log (logger)-import Angel.Data-import Angel.Util (sleepSecs)-import Angel.Files (getFile)-import Angel.PidFile (startMaybeWithPidFile, clearPIDFile)--ifEmpty :: String -> IO () -> IO () -> IO ()-ifEmpty s ioa iob = if s == "" then ioa else iob--- |launch the program specified by `id`, opening (and closing) the--- |appropriate fds for logging. When the process dies, either b/c it was--- |killed by a monitor, killed by a system user, or ended execution naturally,--- |re-examine the desired run config to determine whether to re-run it. if so,--- |tail call.-supervise sharedGroupConfig id = do - let log = logger $ "- program: " ++ id ++ " -"- log "START"- cfg <- atomically $ readTVar sharedGroupConfig- let my_spec = find_me cfg- ifEmpty (name my_spec) -- (log "QUIT (missing from config on restart)") - - (do- (attachOut, attachErr) <- makeFiles my_spec cfg-- let (cmd, args) = cmdSplit $ (fromJust $ exec my_spec)-- let procSpec = (proc cmd args) {- std_out = attachOut,- std_err = attachErr,- cwd = workingDir my_spec- }- let mPfile = pidFile my_spec-- - startMaybeWithPidFile procSpec mPfile $ \pHandle -> do- updateRunningPid my_spec (Just pHandle)- log "RUNNING"- waitForProcess pHandle- log "ENDED"- updateRunningPid my_spec (Nothing)- - cfg <- atomically $ readTVar sharedGroupConfig- if M.notMember id (spec cfg) -- then do - log "QUIT"-- else do - log "WAITING"- sleepSecs $ (fromMaybe defaultDelay $ delay my_spec)- log "RESTART"- supervise sharedGroupConfig id- )- - where- cmdSplit fullcmd = (head parts, tail parts) - where parts = (filter (/="") . map strip . split " ") fullcmd-- find_me cfg = M.findWithDefault defaultProgram id (spec cfg)- updateRunningPid my_spec mpid = atomically $ do - wcfg <- readTVar sharedGroupConfig- writeTVar sharedGroupConfig wcfg{- running=M.insertWith' (\n o-> n) id (my_spec, mpid) (running wcfg)- }- - makeFiles my_spec cfg = do- case (logExec my_spec) of- Just path -> logWithExec path- Nothing -> logWithFiles-- where- logWithFiles = do- let useout = fromMaybe defaultStdout $ stdout my_spec- attachOut <- UseHandle `fmap` getFile useout cfg-- let useerr = fromMaybe defaultStderr $ stderr my_spec- attachErr <- UseHandle `fmap` getFile useerr cfg-- return $ (attachOut, attachErr)-- logWithExec path = do- let (cmd, args) = cmdSplit path- - attachOut <- UseHandle `fmap` getFile "/dev/null" cfg-- (inPipe, _, _, p) <- createProcess (proc cmd args){- std_out = attachOut,- std_err = attachOut,- std_in = CreatePipe,- cwd = workingDir my_spec - }-- return $ (UseHandle (fromJust inPipe), - UseHandle (fromJust inPipe))---- |send a TERM signal to all provided process handles-killProcesses :: [ProcessHandle] -> IO ()-killProcesses pids = mapM_ terminateProcess pids--cleanPidfiles :: [Program] -> IO ()-cleanPidfiles progs = mapM_ clearPIDFile pidfiles- where pidfiles = mapMaybe pidFile progs---- |fire up new supervisors for new program ids-startProcesses :: TVar GroupConfig -> [String] -> IO ()-startProcesses sharedGroupConfig starts = mapM_ spawnWatcher starts- where- spawnWatcher s = forkIO $ wrapProcess sharedGroupConfig s--wrapProcess :: TVar GroupConfig -> String -> IO ()-wrapProcess sharedGroupConfig id = do- run <- createRunningEntry- when run $ finally (supervise sharedGroupConfig id) deleteRunning- where- deleteRunning = atomically $ do - wcfg <- readTVar sharedGroupConfig- writeTVar sharedGroupConfig wcfg{- running=M.delete id (running wcfg)- }-- createRunningEntry =- atomically $ do- cfg <- readTVar sharedGroupConfig- let specmap = spec cfg - case M.lookup id specmap of- Nothing -> return False- Just target -> do- let runmap = running cfg- case M.lookup id runmap of- Just _ -> return False- Nothing -> do- writeTVar sharedGroupConfig cfg{running=- M.insert id (target, Nothing) runmap}- return True---- |diff the requested config against the actual run state, and--- |do any start/kill action necessary-syncSupervisors :: TVar GroupConfig -> IO ()-syncSupervisors sharedGroupConfig = do - let log = logger "process-monitor"- cfg <- atomically $ readTVar sharedGroupConfig- let (killProgs, killHandles) = mustKill cfg- let starts = mustStart cfg- when (nnull killHandles || nnull starts) $ log (- "Must kill=" ++ (show $ length killHandles)- ++ ", must start=" ++ (show $ length starts))- killProcesses killHandles- cleanPidfiles killProgs- startProcesses sharedGroupConfig starts----TODO: make private-mustStart :: GroupConfig -> [String]-mustStart cfg = map fst $ filter (isNew $ running cfg) $ M.assocs (spec cfg)- where isNew running (id, pg) = M.notMember id running----TODO: make private-mustKill :: GroupConfig -> ([Program], [ProcessHandle])-mustKill cfg = unzip targets- where runningAndDifferent :: (ProgramId, (Program, Maybe ProcessHandle)) -> Maybe (Program, ProcessHandle)- runningAndDifferent (id, (pg, Nothing)) = Nothing- runningAndDifferent (id, (pg, (Just pid)))- | (M.notMember id specMap || M.findWithDefault defaultProgram id specMap /= pg) = Just (pg, pid)- | otherwise = Nothing- targets = mapMaybe runningAndDifferent allRunning- specMap = spec cfg- allRunning = M.assocs $ running cfg---- |periodically run the supervisor sync independent of config reload,--- |just in case state gets funky b/c of theoretically possible timing--- |issues on reload-pollStale :: TVar GroupConfig -> IO ()-pollStale sharedGroupConfig = forever $ sleepSecs 10 >> syncSupervisors sharedGroupConfig---nnull :: [a] -> Bool-nnull = not . null
− Angel/Log.hs
@@ -1,19 +0,0 @@-module Angel.Log where--import Data.Time.LocalTime (ZonedTime(..),- getZonedTime)-import Data.Time.Format (formatTime)--import Text.Printf (printf)-import System.Locale (defaultTimeLocale)---- |provide a clean, ISO-ish format for timestamps in logs-cleanCalendar :: ZonedTime -> String-cleanCalendar = formatTime defaultTimeLocale "%Y/%m/%d %H:%M:%S"---- |log a line to stdout; indented for use with partial application for --- |"local log"-type macroing-logger :: String -> String -> IO ()-logger lname msg = do - zt <- getZonedTime- printf "[%s] {%s} %s\n" (cleanCalendar zt) lname msg
− Angel/Main.hs
@@ -1,71 +0,0 @@-module Main where --import Control.Concurrent (threadDelay, forkIO, forkOS)-import Control.Concurrent-import Control.Concurrent.MVar (newEmptyMVar, MVar, takeMVar, putMVar)-import Control.Concurrent.STM-import Control.Concurrent.STM.TVar (readTVar, writeTVar)-import Control.Concurrent.STM.TChan (newTChan)-import Control.Monad (unless, when, forever)-import System.Environment (getArgs)-import System.Posix.Signals-import System.IO (hSetBuffering, BufferMode(..), stdout, stderr)--import qualified Data.Map as M--import Angel.Log (logger)-import Angel.Config (monitorConfig)-import Angel.Data (GroupConfig(..))-import Angel.Job (pollStale, syncSupervisors)-import Angel.Files (startFileManager)---- |Signal handler: when a HUP is trapped, write to the wakeSig Tvar--- |to make the configuration monitor loop cycle/reload-handleHup :: TVar (Maybe Int) -> IO ()-handleHup wakeSig = atomically $ writeTVar wakeSig $ Just 1--handleExit :: MVar Bool -> IO ()-handleExit mv = putMVar mv True--main = do - hSetBuffering stdout LineBuffering- hSetBuffering stderr LineBuffering- let log = logger "main" - log "Angel started"- args <- getArgs-- -- Exactly one argument required for the `angel` executable- unless (length args == 1) $ error "exactly one argument required: config file"- let configPath = head args- log $ "Using config file: " ++ configPath-- -- Create the TVar that represents the "global state" of running applications- -- and applications that _should_ be running- fileReqChan <- atomically $ newTChan- sharedGroupConfig <- newTVarIO $ GroupConfig M.empty M.empty fileReqChan-- -- The wake signal, set by the HUP handler to wake the monitor loop- wakeSig <- newTVarIO Nothing- installHandler sigHUP (Catch $ handleHup wakeSig) Nothing-- -- Handle dying- bye <- newEmptyMVar- installHandler sigTERM (Catch $ handleExit bye) Nothing- installHandler sigINT (Catch $ handleExit bye) Nothing-- -- Fork off an ongoing state monitor to watch for inconsistent state- forkIO $ pollStale sharedGroupConfig- forkIO $ startFileManager fileReqChan-- -- Finally, run the config load/monitor thread- forkIO $ forever $ monitorConfig configPath sharedGroupConfig wakeSig-- _ <- takeMVar bye- log "INT | TERM received; initiating shutdown..."- log " 1. Clearing config"- atomically $ do- cfg <- readTVar sharedGroupConfig- writeTVar sharedGroupConfig cfg {spec = M.empty}- log " 2. Forcing sync to kill running going"- syncSupervisors sharedGroupConfig- log "That's all folks!"
− Angel/PidFile.hs
@@ -1,52 +0,0 @@-module Angel.PidFile ( startMaybeWithPidFile- , startWithPidFile- , clearPIDFile) where--import Control.Applicative ( (<$>)- , (<*>) )-import Control.Exception.Base ( catch- , finally- , SomeException)-import Control.Monad (when)-import System.Process ( CreateProcess- , createProcess- , ProcessHandle )---- Wish I didn't have to do this :(-import System.Process.Internals ( PHANDLE- , ProcessHandle__(..)- , withProcessHandle- )-import System.Posix.Files ( removeLink- , fileExist)--startMaybeWithPidFile :: CreateProcess -> Maybe FilePath -> (ProcessHandle -> IO a) -> IO a-startMaybeWithPidFile procSpec (Just pidFile) = startWithPidFile procSpec pidFile-startMaybeWithPidFile procSpec Nothing = withPHandle procSpec--startWithPidFile :: CreateProcess -> FilePath -> (ProcessHandle -> IO a) -> IO a-startWithPidFile procSpec pidFile action = do- withPHandle procSpec $ \pHandle -> do- mPid <- getPID pHandle- maybe (return ()) write mPid- action pHandle `finally` clearPIDFile pidFile- where write = writePID pidFile--withPHandle :: CreateProcess -> (ProcessHandle -> IO a) -> IO a-withPHandle procSpec action = do- (_, _, _, pHandle) <- createProcess procSpec- action pHandle--writePID :: FilePath -> PHANDLE -> IO ()-writePID pidFile = writeFile pidFile . show--clearPIDFile :: FilePath -> IO ()-clearPIDFile pidFile = do ex <- fileExist pidFile- when ex rm- where exists = fileExist pidFile- rm = removeLink pidFile--getPID :: ProcessHandle -> IO (Maybe PHANDLE)-getPID pHandle = withProcessHandle pHandle getPID'- where getPID' h @ (OpenHandle t) = return (h, Just t)- getPID' h @ (ClosedHandle t) = return (h, Nothing)
− Angel/Util.hs
@@ -1,32 +0,0 @@--- |various utility functions-module Angel.Util where--import Control.Concurrent-import Control.Concurrent.STM-import Control.Concurrent.STM.TVar (readTVar, writeTVar)-import Control.Concurrent (threadDelay, forkIO, forkOS)-import System.Posix.User (getEffectiveUserName,- UserEntry(homeDirectory),- getUserEntryForName)---- |sleep for `s` seconds in an thread-sleepSecs :: Int -> IO ()-sleepSecs s = threadDelay $ s * 1000000---- |wait for the STM TVar to be non-nothing-waitForWake :: TVar (Maybe Int) -> IO ()-waitForWake wakeSig = atomically $ do - state <- readTVar wakeSig- case state of- Just x -> writeTVar wakeSig Nothing- Nothing -> retry--expandPath :: FilePath -> IO FilePath-expandPath ('~':rest) = do home <- getHome =<< getUser- return $ home ++ relativePath- where (userName, relativePath) = span (/= '/') rest- getUser = if null userName- then getEffectiveUserName- else return userName- getHome user = homeDirectory `fmap` getUserEntryForName user-expandPath path = return path
README.md view
@@ -90,16 +90,32 @@ Testing --------There are a few ways to run tests. The simplest is `make spec`. This requires-you have `cabal-dev` installed.+If you prefer to stick with haskell tools, use cabal to build the package. -If you have Ruby installed and bundler installed (`gem install bundler`), you-can use Guard to monitor source files and automatically run the test suite-after you save:+If you have Ruby installed, I've set up a Rakefile for assisting in the+build/testing/sandboxing/dependency process. This isn't necessary to build or+test Angel, but it makes it easier. Run: -1. Run `bundle install`-2. Run `guard start`. Hit enter to force rebuild.+```+gem install bundler # if you don't have it already+bundle install+rake --tasks+``` +If you're using cabal 0.17 or later, and I suggest you do, run++```+rake sandbox+```+Run the full test suite with+```+rake test+```++You can also use `guard start` which will watch for changes made to any source/test+files and re-run the tests for a rapid feedback cycle.++ Configuration and Usage Example ------------------------------- @@ -129,6 +145,10 @@ exec = "run_worker" count = 30 pidfile = "/path/to/pidfile.pid"+ env {+ FOO = "BAR"+ BAR = "BAZ"+ } } Each program that should be supervised starts a `program-id` block:@@ -156,10 +176,16 @@ to specify either stdout or stderr as well.* * `count` is an optional argument to specify the number of processes to spawn. For instance, if you specified a count of 2, it will spawn the program- twice, internally as `workers-1` and `workers-2`, for example.+ twice, internally as `workers-1` and `workers-2`, for example. Note that+ `count` will inject the environment variable `ANGEL_PROCESS_NUMBER` into the+ child process' environment variable. * `pidfile` is an optional argument to specify where a pidfile should be created. If you don't specify an absolute path, it will use the running- directory of angel.+ directory of angel. When combined with the `count` option, specifying a+ pidfile of `worker.pid`, it will generate `worker-1.pid`, `worker-2.pid`,+ etc.+ * `env` is a nested config of string key/value pairs. Non-string values are+ invalid. Assuming the above configuration was in a file called "example.conf", here's what a shell session might look like:@@ -260,6 +286,13 @@ CHANGELOG ---------++### 0.4.4++* Add `env` option to config.+* Inject `ANGEL_PROCESS_NUMBER` environment variable into processes started+ with `count`.+ ### 0.4.3 * Fix install failure from pidfile module not being accounted for.
angel.cabal view
@@ -7,7 +7,7 @@ -- The package version. See the Haskell package versioning policy -- (http://www.haskell.org/haskellwiki/Package_versioning_policy) for -- standards guiding when and how versions should be incremented.-Version: 0.4.3+Version: 0.4.4 -- A short (one-line) description of the package. -- Synopsis: @@ -63,6 +63,7 @@ Executable angel -- .hs or .lhs file containing the Main module.+ Hs-Source-Dirs: src Main-is: Angel/Main.hs -- Packages needed in order to build this package.@@ -96,12 +97,12 @@ Extensions: OverloadedStrings,ScopedTypeVariables,BangPatterns,ViewPatterns - Ghc-Options: -threaded + Ghc-Options: -threaded -fwarn-missing-import-lists test-suite spec Type: exitcode-stdio-1.0 Main-Is: Spec.hs- Hs-Source-Dirs: ., test+ Hs-Source-Dirs: src, test Build-Depends: base Build-Depends: hspec Build-depends: base >= 4.0 && < 5
+ src/Angel/Config.hs view
@@ -0,0 +1,199 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE OverloadedStrings #-}+module Angel.Config ( monitorConfig+ , modifyProg+ , expandByCount ) where++import Control.Exception ( try+ , SomeException )+import qualified Data.Map as M+import Control.Monad ( when+ , (>=>) )+import Control.Concurrent.STM ( STM+ , TVar+ , writeTVar+ , readTVar+ , atomically )+import Data.Configurator ( load+ , getMap+ , Worth(Required) )+import Data.Configurator.Types ( Value(Number, String)+ , Name )+import qualified Data.Traversable as T+import qualified Data.HashMap.Lazy as HM+import Data.Maybe ( isNothing+ , isJust )+import Data.Monoid ( (<>) )+import qualified Data.Text as T+import Angel.Job ( syncSupervisors )+import Angel.Data ( Program( exec+ , delay+ , stdout+ , stderr+ , logExec+ , pidFile+ , workingDir+ , name+ , env )+ , SpecKey+ , GroupConfig(..)+ , defaultProgram )+import Angel.Log ( logger )+import Angel.Util ( waitForWake+ , nnull+ , expandPath )++void :: Monad m => m a -> m ()+void m = m >> return ()++-- |produce a mapping of name -> program for every program+buildConfigMap :: HM.HashMap Name Value -> IO SpecKey+buildConfigMap cfg = do+ return $! HM.foldlWithKey' addToMap M.empty cfg+ where+ addToMap :: SpecKey -> Name -> Value -> SpecKey+ addToMap m name value+ | nnull basekey && nnull localkey = + let !newprog = case M.lookup basekey m of+ Just prog -> modifyProg prog localkey value+ Nothing -> modifyProg defaultProgram {name = basekey, env = []} localkey value+ in+ M.insert basekey newprog m+ | otherwise = m+ where (basekey, '.':localkey) = break (== '.') $ T.unpack name++checkConfigValues :: SpecKey -> IO SpecKey+checkConfigValues progs = (mapM_ checkProgram $ M.elems progs) >> (return progs)+ where+ checkProgram p = do+ when (isNothing $ exec p) $ error $ name p ++ " does not have an 'exec' specification"+ when ((isJust $ logExec p) &&+ (isJust (stdout p) || isJust (stderr p) )) $ error $ name p ++ " cannot have both a logger process and stderr/stdout"++modifyProg :: Program -> String -> Value -> Program+modifyProg prog "exec" (String s) = prog{exec = Just (T.unpack s)}+modifyProg prog "exec" _ = error "wrong type for field 'exec'; string required"++modifyProg prog "delay" (Number n) | n < 0 = error "delay value must be >= 0"+ | otherwise = prog{delay = Just $ round n}+modifyProg prog "delay" _ = error "wrong type for field 'delay'; integer"++modifyProg prog "stdout" (String s) = prog{stdout = Just (T.unpack s)}+modifyProg prog "stdout" _ = error "wrong type for field 'stdout'; string required"++modifyProg prog "stderr" (String s) = prog{stderr = Just (T.unpack s)}+modifyProg prog "stderr" _ = error "wrong type for field 'stderr'; string required"++modifyProg prog "directory" (String s) = prog{workingDir = (Just $ T.unpack s)}+modifyProg prog "directory" _ = error "wrong type for field 'directory'; string required"++modifyProg prog "logger" (String s) = prog{logExec = (Just $ T.unpack s)}+modifyProg prog "logger" _ = error "wrong type for field 'logger'; string required"++modifyProg prog "pidfile" (String s) = prog{pidFile = (Just $ T.unpack s)}+modifyProg prog "pidfile" _ = error "wrong type for field 'pidfile'; string required"++modifyProg prog ('e':'n':'v':'.':envVar) (String s) = prog{env = envVar'}+ where envVar' = (envVar, T.unpack s):(env prog)+modifyProg prog ('e':'n':'v':'.':_) _ = error "wrong type for env field; string required"++modifyProg prog n _ = prog+++-- |invoke the parser to process the file at configPath+-- |produce a SpecKey+processConfig :: String -> IO (Either String SpecKey)+processConfig configPath = do + mconf <- try $ process =<< load [Required configPath]++ case mconf of+ Right config -> return $ Right config+ Left (e :: SomeException) -> return $ Left $ show e+ where process = getMap >=>+ return . expandByCount >=>+ buildConfigMap >=>+ expandPaths >=>+ checkConfigValues++-- |preprocess config into multiple programs if "count" is specified+expandByCount :: HM.HashMap Name Value -> HM.HashMap Name Value+expandByCount cfg = HM.unions expanded+ where expanded :: [HM.HashMap Name Value]+ expanded = concat $ HM.foldlWithKey' expand' [] groupedByProgram+ expand' :: [[HM.HashMap Name Value]] -> Name -> HM.HashMap Name Value -> [[HM.HashMap Name Value]]+ expand' acc = fmap (:acc) . expand+ groupedByProgram :: HM.HashMap Name (HM.HashMap Name Value)+ groupedByProgram = HM.foldlWithKey' binByProg HM.empty cfg+ binByProg h fullKey v+ | prog /= "" && localKey /= "" = HM.insertWith HM.union prog (HM.singleton localKey v) h+ | otherwise = h+ where (prog, localKeyWithLeadingDot) = T.breakOn "." fullKey+ localKey = T.drop 1 localKeyWithLeadingDot+ expand :: Name -> HM.HashMap Name Value -> [HM.HashMap Name Value]+ expand prog pcfg = maybe [reflatten prog pcfg]+ expandWithCount+ (HM.lookup "count" pcfg)+ where expandWithCount (Number n)+ | n >= 0 = [ reflatten (genProgName i) (rewriteConfig i pcfg) | i <- [1..n] ]+ | otherwise = error "count must be >= 0"+ expandWithCount _ = error "count must be a number or not specified"+ genProgName i = prog <> "-" <> textNumber i++rewriteConfig :: Rational -> HM.HashMap Name Value -> HM.HashMap Name Value+rewriteConfig n = HM.insert "env.ANGEL_PROCESS_NUMBER" procNumber . HM.adjust rewritePidfile "pidfile"+ where procNumber = String n'+ n' = textNumber n+ rewritePidfile (String path) = String $ rewrittenFilename <> extension+ where rewrittenFilename = filename <> "-" <> n'+ (filename, extension) = T.breakOn "." path+ rewritePidfile x = x++textNumber :: Rational -> T.Text+textNumber = T.pack . show . truncate++reflatten :: Name -> HM.HashMap Name Value -> HM.HashMap Name Value+reflatten prog pcfg = HM.fromList asList+ where asList = map prependKey $ filter notCount $ HM.toList pcfg+ prependKey (k, v) = ((prog <> "." <> k), v)+ notCount = not . (== "count") . fst++-- |given a new SpecKey just parsed from the file, update the +-- |shared state TVar+updateSpecConfig :: TVar GroupConfig -> SpecKey -> STM ()+updateSpecConfig sharedGroupConfig spec = do + cfg <- readTVar sharedGroupConfig+ writeTVar sharedGroupConfig cfg{spec=spec}++-- |read the config file, update shared state with current spec, +-- |re-sync running supervisors, wait for the HUP TVar, then repeat!+monitorConfig :: String -> TVar GroupConfig -> TVar (Maybe Int) -> IO ()+monitorConfig configPath sharedGroupConfig wakeSig = do + let log = logger "config-monitor"+ mspec <- processConfig configPath+ case mspec of + Left e -> do + log $ " <<<< Config Error >>>>\n" ++ e+ log " <<<< Config Error: Skipping reload >>>>"+ Right spec -> do + atomically $ updateSpecConfig sharedGroupConfig spec+ syncSupervisors sharedGroupConfig+ waitForWake wakeSig+ log "HUP caught, reloading config"++expandPaths :: SpecKey -> IO SpecKey+expandPaths = T.mapM expandProgramPaths++expandProgramPaths :: Program -> IO Program+expandProgramPaths prog = do exec' <- maybeExpand $ exec prog+ stdout' <- maybeExpand $ stdout prog+ stderr' <- maybeExpand $ stderr prog+ workingDir' <- maybeExpand $ workingDir prog+ pidFile' <- maybeExpand $ pidFile prog+ return prog { exec = exec',+ stdout = stdout',+ stderr = stderr',+ workingDir = workingDir',+ pidFile = pidFile' }+ where maybeExpand = T.traverse expandPath
+ src/Angel/Data.hs view
@@ -0,0 +1,68 @@+module Angel.Data ( GroupConfig(..)+ , SpecKey+ , RunKey+ , ProgramId+ , FileRequest+ , Program(..)+ , Spec+ , defaultProgram+ , defaultDelay+ , defaultStdout+ , defaultStderr+ ) where++import qualified Data.Map as M+import System.Process ( createProcess+ , proc+ , waitForProcess+ , ProcessHandle )+import System.Process ( terminateProcess+ , CreateProcess(..)+ , StdStream(..) )+import Control.Concurrent.STM.TChan (TChan)+import System.IO (Handle)++-- |the whole shared state of the program; spec is what _should_+-- |be running, while running is what actually _is_ running_ currently+data GroupConfig = GroupConfig {+ spec :: SpecKey,+ running :: RunKey,+ fileRequest :: TChan FileRequest+}++-- |map program ids to relevant structure+type SpecKey = M.Map ProgramId Program+type RunKey = M.Map ProgramId (Program, Maybe ProcessHandle)+type ProgramId = String+type FileRequest = (String, TChan (Maybe Handle))++-- |the representation of a program is these 6 values, +-- |read from the config file+data Program = Program {+ name :: String,+ exec :: Maybe String,+ delay :: Maybe Int,+ stdout :: Maybe String,+ stderr :: Maybe String,+ workingDir :: Maybe FilePath,+ logExec :: Maybe String,+ pidFile :: Maybe FilePath,+ env :: [(String, String)]+} deriving (Show, Eq, Ord)++-- |Lower-level atoms in the configuration process+type Spec = [Program]++-- |a template for an empty program; the variable set to ""+-- |are required, and must be overridden in the config file+defaultProgram :: Program+defaultProgram = Program "" Nothing Nothing Nothing Nothing Nothing Nothing Nothing []++defaultDelay :: Int+defaultDelay = 5++defaultStdout :: FilePath+defaultStdout = "/dev/null"++defaultStderr :: FilePath+defaultStderr = "/dev/null"
+ src/Angel/Files.hs view
@@ -0,0 +1,44 @@+{-# LANGUAGE ScopedTypeVariables #-}+module Angel.Files ( getFile+ , startFileManager ) where++import Control.Exception ( try+ , SomeException )+import Control.Concurrent.STM ( readTChan+ , writeTChan+ , atomically+ , TChan+ , newTChan+ , newTChanIO )+import Control.Monad (forever)+import System.IO ( Handle+ , hClose+ , openFile+ , IOMode(..) )+import GHC.IO.Handle (hDuplicate)+import Angel.Data ( GroupConfig(..)+ , FileRequest )++startFileManager req = forever $ fileManager req++fileManager :: TChan FileRequest -> IO ()+fileManager req = do + (path, resp) <- atomically $ readTChan req+ mh <- try $ openFile path AppendMode+ case mh of+ Right hand -> do+ hand' <- hDuplicate hand+ hClose hand+ atomically $ writeTChan resp (Just hand')+ Left (e :: SomeException) -> atomically $ writeTChan resp Nothing+ fileManager req++getFile :: String -> GroupConfig -> IO Handle+getFile path cfg = do+ resp <- newTChanIO+ atomically $ writeTChan (fileRequest cfg) (path, resp)+ mh <- atomically $ readTChan resp+ hand <- case mh of+ Just hand -> return hand+ Nothing -> error $ "could not open stdout/stderr file " ++ path+ return hand
+ src/Angel/Job.hs view
@@ -0,0 +1,218 @@+module Angel.Job ( syncSupervisors+ , pollStale ) where++import Control.Exception ( finally )+import Data.String.Utils ( split+ , strip )+import Data.Maybe ( mapMaybe+ , fromMaybe+ , fromJust )+import System.Process ( createProcess+ , proc+ , waitForProcess+ , ProcessHandle )+import System.Process ( terminateProcess+ , CreateProcess(..)+ , StdStream(..) )+import Control.Concurrent ( forkIO )+import Control.Concurrent.STM ( TVar+ , writeTVar+ , readTVar+ , atomically )+import qualified Data.Map as M+import Control.Monad ( when+ , forever )+import Angel.Log ( logger )+import Angel.Data ( Program( delay+ , exec+ , logExec+ , name+ , pidFile+ , stderr+ , stdout+ , workingDir )+ , ProgramId+ , GroupConfig(..)+ , defaultProgram+ , defaultDelay+ , defaultStdout+ , defaultStderr )+import qualified Angel.Data as D+import Angel.Util ( sleepSecs+ , nnull )+import Angel.Files ( getFile )+import Angel.PidFile ( startMaybeWithPidFile+ , clearPIDFile )++ifEmpty :: String -> IO () -> IO () -> IO ()+ifEmpty s ioa iob = if s == "" then ioa else iob+-- |launch the program specified by `id`, opening (and closing) the+-- |appropriate fds for logging. When the process dies, either b/c it was+-- |killed by a monitor, killed by a system user, or ended execution naturally,+-- |re-examine the desired run config to determine whether to re-run it. if so,+-- |tail call.+supervise sharedGroupConfig id = do + log "START"+ cfg <- atomically $ readTVar sharedGroupConfig+ let my_spec = find_me cfg+ ifEmpty (name my_spec) ++ (log "QUIT (missing from config on restart)") + + (do+ (attachOut, attachErr) <- makeFiles my_spec cfg++ let (cmd, args) = cmdSplit $ (fromJust $ exec my_spec)++ let procSpec = (proc cmd args) {+ std_out = attachOut,+ std_err = attachErr,+ cwd = workingDir my_spec,+ env = Just $ D.env my_spec+ }++ let mPfile = pidFile my_spec++ log $ "Spawning process with env " ++ show (env procSpec)++ + startMaybeWithPidFile procSpec mPfile $ \pHandle -> do+ updateRunningPid my_spec (Just pHandle)+ log "RUNNING"+ waitForProcess pHandle+ log "ENDED"+ updateRunningPid my_spec (Nothing)+ + cfg <- atomically $ readTVar sharedGroupConfig+ if M.notMember id (spec cfg) ++ then do + log "QUIT"++ else do + log "WAITING"+ sleepSecs $ (fromMaybe defaultDelay $ delay my_spec)+ log "RESTART"+ supervise sharedGroupConfig id+ )+ + where+ log = logger $ "- program: " ++ id ++ " -"+ cmdSplit fullcmd = (head parts, tail parts) + where parts = (filter (/="") . map strip . split " ") fullcmd++ find_me cfg = M.findWithDefault defaultProgram id (spec cfg)+ updateRunningPid my_spec mpid = atomically $ do + wcfg <- readTVar sharedGroupConfig+ writeTVar sharedGroupConfig wcfg{+ running=M.insertWith' (\n o-> n) id (my_spec, mpid) (running wcfg)+ }+ + makeFiles my_spec cfg = do+ case (logExec my_spec) of+ Just path -> logWithExec path+ Nothing -> logWithFiles++ where+ logWithFiles = do+ let useout = fromMaybe defaultStdout $ stdout my_spec+ attachOut <- UseHandle `fmap` getFile useout cfg++ let useerr = fromMaybe defaultStderr $ stderr my_spec+ attachErr <- UseHandle `fmap` getFile useerr cfg++ return $ (attachOut, attachErr)++ logWithExec path = do+ let (cmd, args) = cmdSplit path+ + attachOut <- UseHandle `fmap` getFile "/dev/null" cfg++ log "Spawning process"+ (inPipe, _, _, p) <- createProcess (proc cmd args){+ std_out = attachOut,+ std_err = attachOut,+ std_in = CreatePipe,+ cwd = workingDir my_spec+ }++ return $ (UseHandle (fromJust inPipe), + UseHandle (fromJust inPipe))++-- |send a TERM signal to all provided process handles+killProcesses :: [ProcessHandle] -> IO ()+killProcesses pids = mapM_ terminateProcess pids++cleanPidfiles :: [Program] -> IO ()+cleanPidfiles progs = mapM_ clearPIDFile pidfiles+ where pidfiles = mapMaybe pidFile progs++-- |fire up new supervisors for new program ids+startProcesses :: TVar GroupConfig -> [String] -> IO ()+startProcesses sharedGroupConfig starts = mapM_ spawnWatcher starts+ where+ spawnWatcher s = forkIO $ wrapProcess sharedGroupConfig s++wrapProcess :: TVar GroupConfig -> String -> IO ()+wrapProcess sharedGroupConfig id = do+ run <- createRunningEntry+ when run $ finally (supervise sharedGroupConfig id) deleteRunning+ where+ deleteRunning = atomically $ do + wcfg <- readTVar sharedGroupConfig+ writeTVar sharedGroupConfig wcfg{+ running=M.delete id (running wcfg)+ }++ createRunningEntry =+ atomically $ do+ cfg <- readTVar sharedGroupConfig+ let specmap = spec cfg + case M.lookup id specmap of+ Nothing -> return False+ Just target -> do+ let runmap = running cfg+ case M.lookup id runmap of+ Just _ -> return False+ Nothing -> do+ writeTVar sharedGroupConfig cfg{running=+ M.insert id (target, Nothing) runmap}+ return True++-- |diff the requested config against the actual run state, and+-- |do any start/kill action necessary+syncSupervisors :: TVar GroupConfig -> IO ()+syncSupervisors sharedGroupConfig = do + let log = logger "process-monitor"+ cfg <- atomically $ readTVar sharedGroupConfig+ let (killProgs, killHandles) = mustKill cfg+ let starts = mustStart cfg+ when (nnull killHandles || nnull starts) $ log (+ "Must kill=" ++ (show $ length killHandles)+ ++ ", must start=" ++ (show $ length starts))+ killProcesses killHandles+ cleanPidfiles killProgs+ startProcesses sharedGroupConfig starts++--TODO: make private+mustStart :: GroupConfig -> [String]+mustStart cfg = map fst $ filter (isNew $ running cfg) $ M.assocs (spec cfg)+ where isNew running (id, pg) = M.notMember id running++--TODO: make private+mustKill :: GroupConfig -> ([Program], [ProcessHandle])+mustKill cfg = unzip targets+ where runningAndDifferent :: (ProgramId, (Program, Maybe ProcessHandle)) -> Maybe (Program, ProcessHandle)+ runningAndDifferent (id, (pg, Nothing)) = Nothing+ runningAndDifferent (id, (pg, (Just pid)))+ | (M.notMember id specMap || M.findWithDefault defaultProgram id specMap /= pg) = Just (pg, pid)+ | otherwise = Nothing+ targets = mapMaybe runningAndDifferent allRunning+ specMap = spec cfg+ allRunning = M.assocs $ running cfg++-- |periodically run the supervisor sync independent of config reload,+-- |just in case state gets funky b/c of theoretically possible timing+-- |issues on reload+pollStale :: TVar GroupConfig -> IO ()+pollStale sharedGroupConfig = forever $ sleepSecs 10 >> syncSupervisors sharedGroupConfig
+ src/Angel/Log.hs view
@@ -0,0 +1,20 @@+module Angel.Log ( cleanCalendar+ , logger) where++import Data.Time.LocalTime (ZonedTime,+ getZonedTime)+import Data.Time.Format (formatTime)++import Text.Printf (printf)+import System.Locale (defaultTimeLocale)++-- |provide a clean, ISO-ish format for timestamps in logs+cleanCalendar :: ZonedTime -> String+cleanCalendar = formatTime defaultTimeLocale "%Y/%m/%d %H:%M:%S"++-- |log a line to stdout; indented for use with partial application for +-- |"local log"-type macroing+logger :: String -> String -> IO ()+logger lname msg = do + zt <- getZonedTime+ printf "[%s] {%s} %s\n" (cleanCalendar zt) lname msg
+ src/Angel/Main.hs view
@@ -0,0 +1,85 @@+module Main (main) where ++import Control.Concurrent (threadDelay, forkIO, forkOS)+import Control.Concurrent+import Control.Concurrent.MVar (newEmptyMVar, MVar, takeMVar, putMVar)+import Control.Concurrent.STM+import Control.Concurrent.STM.TVar (readTVar, writeTVar)+import Control.Concurrent.STM.TChan (newTChan)+import Control.Monad (unless, when, forever)+import System.Environment (getArgs)+import System.Exit (exitFailure, exitSuccess)+import System.Posix.Signals+import System.IO (hSetBuffering, hPutStrLn, BufferMode(..), stdout, stderr)++import qualified Data.Map as M++import Angel.Log (logger)+import Angel.Config (monitorConfig)+import Angel.Data (GroupConfig(..))+import Angel.Job (pollStale, syncSupervisors)+import Angel.Files (startFileManager)++-- |Signal handler: when a HUP is trapped, write to the wakeSig Tvar+-- |to make the configuration monitor loop cycle/reload+handleHup :: TVar (Maybe Int) -> IO ()+handleHup wakeSig = atomically $ writeTVar wakeSig $ Just 1++handleExit :: MVar Bool -> IO ()+handleExit mv = putMVar mv True++main :: IO ()+main = handleArgs =<< getArgs++handleArgs :: [String] -> IO ()+handleArgs ["--help"] = printHelp+handleArgs ["-h"] = printHelp+handleArgs [] = printHelp+handleArgs [configPath] = runWithConfigPath configPath+handleArgs _ = errorExit "expected a single config file. Run with --help for usasge."++printHelp :: IO ()+printHelp = putStrLn "Usage: angel [--help] CONFIG_FILE" >> exitSuccess++runWithConfigPath :: FilePath -> IO ()+runWithConfigPath configPath = do+ hSetBuffering stdout LineBuffering+ hSetBuffering stderr LineBuffering+ let log = logger "main" + log "Angel started"++ log $ "Using config file: " ++ configPath++ -- Create the TVar that represents the "global state" of running applications+ -- and applications that _should_ be running+ fileReqChan <- atomically $ newTChan+ sharedGroupConfig <- newTVarIO $ GroupConfig M.empty M.empty fileReqChan++ -- The wake signal, set by the HUP handler to wake the monitor loop+ wakeSig <- newTVarIO Nothing+ installHandler sigHUP (Catch $ handleHup wakeSig) Nothing++ -- Handle dying+ bye <- newEmptyMVar+ installHandler sigTERM (Catch $ handleExit bye) Nothing+ installHandler sigINT (Catch $ handleExit bye) Nothing++ -- Fork off an ongoing state monitor to watch for inconsistent state+ forkIO $ pollStale sharedGroupConfig+ forkIO $ startFileManager fileReqChan++ -- Finally, run the config load/monitor thread+ forkIO $ forever $ monitorConfig configPath sharedGroupConfig wakeSig++ _ <- takeMVar bye+ log "INT | TERM received; initiating shutdown..."+ log " 1. Clearing config"+ atomically $ do+ cfg <- readTVar sharedGroupConfig+ writeTVar sharedGroupConfig cfg {spec = M.empty}+ log " 2. Forcing sync to kill running going"+ syncSupervisors sharedGroupConfig+ log "That's all folks!"++errorExit :: String -> IO ()+errorExit msg = hPutStrLn stderr msg >> exitFailure
+ src/Angel/PidFile.hs view
@@ -0,0 +1,52 @@+module Angel.PidFile ( startMaybeWithPidFile+ , startWithPidFile+ , clearPIDFile) where++import Control.Applicative ( (<$>)+ , (<*>) )+import Control.Exception.Base ( catch+ , finally+ , SomeException)+import Control.Monad (when)+import System.Process ( CreateProcess+ , createProcess+ , ProcessHandle )++-- Wish I didn't have to do this :(+import System.Process.Internals ( PHANDLE+ , ProcessHandle__(..)+ , withProcessHandle+ )+import System.Posix.Files ( removeLink+ , fileExist)++startMaybeWithPidFile :: CreateProcess -> Maybe FilePath -> (ProcessHandle -> IO a) -> IO a+startMaybeWithPidFile procSpec (Just pidFile) = startWithPidFile procSpec pidFile+startMaybeWithPidFile procSpec Nothing = withPHandle procSpec++startWithPidFile :: CreateProcess -> FilePath -> (ProcessHandle -> IO a) -> IO a+startWithPidFile procSpec pidFile action = do+ withPHandle procSpec $ \pHandle -> do+ mPid <- getPID pHandle+ maybe (return ()) write mPid+ action pHandle `finally` clearPIDFile pidFile+ where write = writePID pidFile++withPHandle :: CreateProcess -> (ProcessHandle -> IO a) -> IO a+withPHandle procSpec action = do+ (_, _, _, pHandle) <- createProcess procSpec+ action pHandle++writePID :: FilePath -> PHANDLE -> IO ()+writePID pidFile = writeFile pidFile . show++clearPIDFile :: FilePath -> IO ()+clearPIDFile pidFile = do ex <- fileExist pidFile+ when ex rm+ where exists = fileExist pidFile+ rm = removeLink pidFile++getPID :: ProcessHandle -> IO (Maybe PHANDLE)+getPID pHandle = withProcessHandle pHandle getPID'+ where getPID' h @ (OpenHandle t) = return (h, Just t)+ getPID' h @ (ClosedHandle t) = return (h, Nothing)
+ src/Angel/Util.hs view
@@ -0,0 +1,38 @@+-- |various utility functions+module Angel.Util ( sleepSecs+ , waitForWake+ , expandPath+ , nnull ) where++import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.STM.TVar (readTVar, writeTVar)+import Control.Concurrent (threadDelay, forkIO, forkOS)+import System.Posix.User (getEffectiveUserName,+ UserEntry(homeDirectory),+ getUserEntryForName)++-- |sleep for `s` seconds in an thread+sleepSecs :: Int -> IO ()+sleepSecs s = threadDelay $ s * 1000000++-- |wait for the STM TVar to be non-nothing+waitForWake :: TVar (Maybe Int) -> IO ()+waitForWake wakeSig = atomically $ do + state <- readTVar wakeSig+ case state of+ Just x -> writeTVar wakeSig Nothing+ Nothing -> retry++expandPath :: FilePath -> IO FilePath+expandPath ('~':rest) = do home <- getHome =<< getUser+ return $ home ++ relativePath+ where (userName, relativePath) = span (/= '/') rest+ getUser = if null userName+ then getEffectiveUserName+ else return userName+ getHome user = homeDirectory `fmap` getUserEntryForName user+expandPath path = return path++nnull :: [a] -> Bool+nnull = not . null