packages feed

angel 0.2 → 0.2.1

raw patch · 8 files changed

+385/−2 lines, 8 filesdep ~base

Dependency ranges changed: base

Files

+ Angel/Config.hs view
@@ -0,0 +1,57 @@+module Angel.Config where++import qualified Data.Map as M+import Control.Concurrent.STM+import Control.Concurrent.STM.TVar (readTVar, writeTVar)+import Text.ParserCombinators.Parsec.Error (ParseError)++import Angel.Parse (parseConfig)+import Angel.Job (syncSupervisors)+import Angel.Data+import Angel.Log (logger)+import Angel.Util (waitForWake)++-- |produce a mapping of name -> program for every program+buildConfigMap :: Spec -> SpecKey+buildConfigMap cfg = M.fromList [(name p, p) | p <- cfg]++-- |invoke the parser to process the file at configPath+-- |produce a SpecKey+processConfig :: String -> IO (Either String SpecKey)+processConfig configPath = do +    mc <- catch ( do+            c <- readFile configPath +            return $ Just c+            ) (\e-> return Nothing)++    res <- case mc of +        Just c -> do+            let cfg = parseConfig c+            case cfg of+                Left e -> return $ Left $ show e+                Right cfg -> do return $ Right $ buildConfigMap cfg+        Nothing -> return $ Left ("could not read config file at " ++ configPath)+    return res++-- |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"
+ Angel/Data.hs view
@@ -0,0 +1,46 @@+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 :: String,+    delay :: Int,+    stdout :: String,+    stderr :: String,+    workingDir :: 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{+    name="",+    exec="",+    delay=5,+    stdout="/dev/null",+    stderr="/dev/null",+    workingDir = Nothing+}
+ Angel/Files.hs view
@@ -0,0 +1,33 @@+module Angel.Files (getFile, startFileManager) where++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 <- catch  (openFile path AppendMode >>= \h-> return $ Just h) (\e-> return Nothing)+    case mh of+        Just hand -> do+            hand' <- hDuplicate hand+            hClose hand+            atomically $ writeTChan resp (Just hand')+        Nothing -> 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 view
@@ -0,0 +1,130 @@+module Angel.Job where++import Data.String.Utils (split, strip)+import Data.Maybe (isJust, fromJust)+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)++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)" >> deleteRunning) +        +        (do+            (attachOut, attachErr) <- makeFiles my_spec cfg++            let (cmd, args) = cmdSplit $ exec my_spec+            +            (_, _, _, p) <- createProcess (proc cmd args){+            std_out = attachOut,+            std_err = attachErr,+            cwd = workingDir my_spec +            }+            +            updateRunningPid my_spec (Just p)+            log "RUNNING"+            waitForProcess p+            log "ENDED"+            updateRunningPid my_spec (Nothing)+            +            cfg <- atomically $ readTVar sharedGroupConfig+            if M.notMember id (spec cfg) ++                then do +                log  "QUIT"+                deleteRunning++                else do +                log  "WAITING"+                sleepSecs $ 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)+            }+                +        deleteRunning = atomically $ do +            wcfg <- readTVar sharedGroupConfig+            writeTVar sharedGroupConfig wcfg{+                running=M.delete id (running wcfg)+            }+            +        makeFiles my_spec cfg = do  +            attachOut <- case stdout my_spec of+                ""    -> return Inherit+                other -> UseHandle `fmap` getFile other cfg+  +            attachErr <- case stderr my_spec of+                ""    -> return Inherit+                other -> UseHandle `fmap` getFile other cfg+            return $ (attachOut, attachErr)+++-- |send a TERM signal to all provided process handles+killProcesses :: [ProcessHandle] -> IO ()+killProcesses pids = mapM_ terminateProcess pids++-- |fire up new supervisors for new program ids+startProcesses :: TVar GroupConfig -> [String] -> IO ()+startProcesses sharedGroupConfig starts = mapM_ spawnWatcher starts+    where+        spawnWatcher s = forkIO $ supervise sharedGroupConfig s++-- |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 kills = mustKill cfg+   let starts = mustStart cfg+   when (length kills > 0 || length starts > 0) $ log (+         "Must kill=" ++ (show $ length kills)+                ++ ", must start=" ++ (show $ length starts))+   killProcesses kills+   startProcesses sharedGroupConfig starts+                                         +    where+        mustKill cfg = map (fromJust . snd . snd) $ filter (runningAndDifferent $ spec cfg) $ M.assocs (running cfg)+        runningAndDifferent spec (id, (pg, pid)) = (isJust pid && (M.notMember id spec +                                           || M.findWithDefault defaultProgram id spec `cmp` pg))+            where cmp one two = one /= two++        mustStart cfg = map fst $ filter (isNew $ running cfg) $ M.assocs (spec cfg)+        isNew running (id, pg) = M.notMember id running++-- |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
+ Angel/Log.hs view
@@ -0,0 +1,17 @@+module Angel.Log where++import Text.Printf (printf)+import System.Time (getClockTime, toCalendarTime, CalendarTime(..), formatCalendarTime)+import System.Locale (defaultTimeLocale)++-- |provide a clean, ISO-ish format for timestamps in logs+cleanCalendar :: CalendarTime -> String+cleanCalendar ct = formatCalendarTime defaultTimeLocale "%Y/%m/%d %H:%M:%S" ct++-- |log a line to stdout; indented for use with partial application for +-- |"local log"-type macroing+logger :: String -> String -> IO ()+logger lname msg = do +    tm <- getClockTime+    ct <- toCalendarTime tm+    printf "[%s] {%s} %s\n" (cleanCalendar ct) lname msg
+ Angel/Parse.hs view
@@ -0,0 +1,74 @@+-- |Parse Angel configuration files--a parsec parser+module Angel.Parse where++import Data.String.Utils (strip)+import Text.ParserCombinators.Parsec+import Data.Maybe (isJust)+import Data.Either (Either(..))+import Control.Monad (foldM)++import Angel.Data++type Kw = Maybe (String, String)++reqInt = manyTill (oneOf ['0'..'9']) (char '\n') <?> "integer"+configString = manyTill anyChar $ char '\n'++configLine :: GenParser Char st Kw+configLine = do +    name <- manyTill (noneOf "[") (char ' ')+    val <- case name of +            "exec" -> configString+            "delay" -> reqInt+            "stdout" -> configString+            "stderr" -> configString+            "directory" -> configString+            otherwise -> fail $ "unknown config verb '" ++ name ++ "'"+    return $ Just (strip name, strip val)++commentLine :: GenParser Char st Kw+commentLine = do +    manyTill (oneOf " \t") $ char '#'+    manyTill anyChar $ char '\n'+    return Nothing++emptyLine :: GenParser Char st Kw+emptyLine = manyTill (oneOf " \t\r") (char '\n') >> return Nothing++header = do +    char '[' <?> "start program id"+    name <- manyTill anyChar $ char ']'+    manyTill (oneOf " \t\r") $ char '\n'+    return name++program = do +    given_id <- header+    kw <- many $ (emptyLine <|> commentLine <|> configLine)+    let real_kw = filter isJust kw+    let prg = defaultProgram{name=given_id}+    prg <- foldM setAttr prg real_kw+    mapM_ (check_set prg) [("exec", exec), ("name", name)] +    return prg++    where +    +        setAttr prg (Just (n, v))  = case n of +                                    "exec" -> return prg{exec=v}+                                    "delay" -> return prg{delay=(read v)::Int}+                                    "stdout" -> return prg{stdout=v}+                                    "stderr" -> return prg{stderr=v}+                                    "directory" -> return prg { workingDir = Just v }+                                    otherwise -> fail $ "unknown config keyword " ++ n+        setAttr _ _ = fail "non-just in setAttr argument??"++        check_set prg (name, f) = if f prg == f defaultProgram +                            then fail $ name ++ " must be set for all programs" +                            else return ()++configFile :: GenParser Char st Spec+configFile = do +    many $ (emptyLine <|> commentLine)+    return =<< manyTill program eof++parseConfig :: String -> Either ParseError Spec+parseConfig input = parse configFile "(config)" input
+ Angel/Util.hs view
@@ -0,0 +1,19 @@+-- |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)++-- |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
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.2+Version:             0.2.1  -- A short (one-line) description of the package. -- Synopsis:            @@ -79,7 +79,14 @@       -- Modules not exported by this package.-  -- Other-modules:       +  Other-modules: Angel.Files,+                 Angel.Config,+                 Angel.Data,+                 Angel.Files,+                 Angel.Job,+                 Angel.Log,+                 Angel.Parse,+                 Angel.Util      -- Extra tools (e.g. alex, hsc2hs, ...) needed to build the source.   -- Build-tools: