keter 1.2.0 → 1.2.1
raw patch · 6 files changed
+581/−5 lines, 6 filesdep +hspecdep −unix-process-conduitdep ~basedep ~conduitdep ~unix
Dependencies added: hspec
Dependencies removed: unix-process-conduit
Dependency ranges changed: base, conduit, unix
Files
- Data/Conduit/LogFile.hs +141/−0
- Data/Conduit/Process/Unix.hs +318/−0
- Keter/App.hs +4/−3
- cbits/process-tracker.c +100/−0
- keter.cabal +17/−2
- test/Spec.hs +1/−0
+ Data/Conduit/LogFile.hs view
@@ -0,0 +1,141 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}+module Data.Conduit.LogFile+ ( RotatingLog+ , openRotatingLog+ , addChunk+ , close+ , defaultMaxTotal+ , dummy+ ) where++import Control.Concurrent (forkIO)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TBQueue+import Control.Concurrent.STM.TVar+import Control.Exception (bracket, bracketOnError,+ finally)+import Control.Monad (void, when)+import qualified Data.ByteString as S+import Data.Time (UTCTime, getCurrentTime)+import Data.Word (Word)+import System.Directory (createDirectoryIfMissing,+ doesFileExist, renameFile)+import System.FilePath ((<.>), (</>))+import qualified System.IO as SIO+import System.IO.Unsafe (unsafePerformIO)+import System.Mem.Weak (addFinalizer)++data Command = AddChunk !S.ByteString+ | Close++-- | Represents a folder used for totating log files.+--+-- Since 0.2.1+data RotatingLog = RotatingLog !(TVar State)+-- Use a data instead of a newtype so that we can attach a finalizer.++-- | A @RotatingLog@ which performs no logging.+--+-- Since 0.2.1+dummy :: RotatingLog+dummy = RotatingLog $! unsafePerformIO $! newTVarIO Closed++data State = Closed+ | Running !SIO.Handle !(TBQueue Command)++queue :: Command -> RotatingLog -> IO ()+queue cmd (RotatingLog ts) = atomically $ do+ s <- readTVar ts+ case s of+ Closed -> return ()+ Running _ q -> writeTBQueue q cmd++addChunk :: RotatingLog -> S.ByteString -> IO ()+addChunk lf bs = queue (AddChunk bs) lf++close :: RotatingLog -> IO ()+close = queue Close++-- | Create a new @RotatingLog@.+--+-- Since 0.2.1+openRotatingLog :: FilePath -- ^ folder to contain logs+ -> Word -- ^ maximum log file size, in bytes+ -> IO RotatingLog+openRotatingLog dir maxTotal = do+ createDirectoryIfMissing True dir+ bracketOnError (moveCurrent dir) SIO.hClose $ \handle -> do+ queue <- newTBQueueIO 5+ let s = Running handle queue+ ts <- newTVarIO s+ void $ forkIO $ loop dir ts maxTotal+ let rl = RotatingLog ts+ addFinalizer rl (atomically (writeTBQueue queue Close))+ return rl++current :: FilePath -- ^ folder containing logs+ -> FilePath+current = (</> "current.log")++moveCurrent :: FilePath -- ^ folder containing logs+ -> IO SIO.Handle -- ^ new handle+moveCurrent dir = do+ let curr = current dir+ x <- doesFileExist curr+ when x $ do+ now <- getCurrentTime+ renameFile curr $ dir </> suffix now+ SIO.openFile curr SIO.WriteMode++suffix :: UTCTime -> FilePath+suffix now =+ (concatMap fix $ takeWhile (/= '.') $ show now) <.> "log"+ where+ fix ' ' = "_"+ fix c | '0' <= c && c <= '9' = [c]+ fix _ = ""++loop :: FilePath -- ^ folder containing logs+ -> TVar State+ -> Word -- ^ maximum total log size+ -> IO ()+loop dir ts maxTotal =+ go 0 `finally` (closeCurrentHandle `finally` atomically (writeTVar ts Closed))+ where+ closeCurrentHandle = bracket+ (atomically $ do+ s <- readTVar ts+ case s of+ Closed -> return Nothing+ Running h _ -> return $! Just h)+ (maybe (return ()) SIO.hClose)+ (const $ return ())++ go total = do+ res <- atomically $ do+ s <- readTVar ts+ case s of+ Closed -> return Nothing+ Running handle queue -> do+ cmd <- readTBQueue queue+ case cmd of+ Close -> return Nothing+ AddChunk bs -> return $! Just (handle, queue, bs)+ case res of+ Nothing -> return ()+ Just (handle, queue, bs) -> do+ let total' = total + fromIntegral (S.length bs)+ S.hPut handle bs+ SIO.hFlush handle+ if total' > maxTotal+ then do+ bracket+ (SIO.hClose handle >> moveCurrent dir)+ (\handle' -> atomically $ writeTVar ts $ Running handle' queue)+ (const $ return ())+ go 0+ else go total'++defaultMaxTotal :: Word+defaultMaxTotal = 5 * 1024 * 1024 -- 5 MB
+ Data/Conduit/Process/Unix.hs view
@@ -0,0 +1,318 @@+{-# LANGUAGE CPP #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Data.Conduit.Process.Unix+ ( -- * Process tracking+ -- $processTracker++ -- ** Types+ ProcessTracker+ -- ** Functions+ , initProcessTracker++ -- * Monitored process+ , MonitoredProcess+ , monitorProcess+ , terminateMonitoredProcess+ ) where++import Control.Applicative ((<$>), (<*>))+import Control.Arrow ((***))+import Control.Concurrent (forkIO)+import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (MVar, modifyMVar, modifyMVar_,+ newEmptyMVar, newMVar,+ putMVar, readMVar, swapMVar,+ takeMVar)+import Control.Exception (Exception, SomeException,+ bracketOnError, finally,+ handle, mask_,+ throwIO, try)+import Control.Monad (void)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as S8+import Data.Conduit (Source, ($$))+import Data.Conduit.Binary (sinkHandle, sourceHandle)+import qualified Data.Conduit.List as CL+import Data.IORef (IORef, newIORef, readIORef,+ writeIORef)+import Data.Time (getCurrentTime)+import Data.Time (diffUTCTime)+import Data.Typeable (Typeable)+import Foreign.C.Types+import Prelude (Bool (..), Either (..), IO,+ Maybe (..), Monad (..), Show,+ const, error, flip, fmap,+ fromIntegral, fst, head, id,+ length, map, maybe, show, snd,+ ($), ($!), (*), (.), (<),+ (==))+import System.Exit (ExitCode)+import System.IO (hClose)+import System.Posix.IO.ByteString ( closeFd, createPipe,+ fdToHandle)+import System.Posix.Signals (sigKILL, signalProcess)+import System.Posix.Types (CPid (..))+import System.Process+import System.Process.Internals (ProcessHandle (..),+ ProcessHandle__ (..))++processHandleMVar :: ProcessHandle -> MVar ProcessHandle__+#if MIN_VERSION_process(1, 2, 0)+processHandleMVar (ProcessHandle m _) = m+#else+processHandleMVar (ProcessHandle m) = m+#endif++withProcessHandle_+ :: ProcessHandle+ -> (ProcessHandle__ -> IO ProcessHandle__)+ -> IO ()+withProcessHandle_ ph io = modifyMVar_ (processHandleMVar ph) io++-- | Kill a process by sending it the KILL (9) signal.+--+-- Since 0.1.0+killProcess :: ProcessHandle -> IO ()+killProcess ph = withProcessHandle_ ph $ \p_ ->+ case p_ of+ ClosedHandle _ -> return p_+ OpenHandle h -> do+ signalProcess sigKILL h+ return p_++ignoreExceptions :: IO () -> IO ()+ignoreExceptions = handle (\(_ :: SomeException) -> return ())++-- $processTracker+--+-- Ensure that child processes are killed, regardless of how the parent process exits.+--+-- The technique used here is:+--+-- * Create a pipe.+--+-- * Fork a new child process that listens on the pipe.+--+-- * In the current process, send updates about processes that should be auto-killed.+--+-- * When the parent process dies, listening on the pipe in the child process will get an EOF.+--+-- * When the child process receives that EOF, it kills all processes it was told to auto-kill.+--+-- This code was originally written for Keter, but was moved to unix-process+-- conduit in the 0.2.1 release.++foreign import ccall unsafe "launch_process_tracker"+ c_launch_process_tracker :: IO CInt++foreign import ccall unsafe "track_process"+ c_track_process :: ProcessTracker -> CPid -> CInt -> IO ()++-- | Represents the child process which handles process cleanup.+--+-- Since 0.2.1+newtype ProcessTracker = ProcessTracker CInt++-- | Represents a child process which is currently being tracked by the cleanup+-- child process.+--+-- Since 0.2.1+data TrackedProcess = TrackedProcess !ProcessTracker !(IORef MaybePid) !(IO ExitCode)++data MaybePid = NoPid | Pid !CPid++-- | Fork off the child cleanup process.+--+-- This will ideally only be run once for your entire application.+--+-- Since 0.2.1+initProcessTracker :: IO ProcessTracker+initProcessTracker = do+ i <- c_launch_process_tracker+ if i == -1+ then throwIO CannotLaunchProcessTracker+ else return $! ProcessTracker i++-- | Since 0.2.1+data ProcessTrackerException = CannotLaunchProcessTracker+ deriving (Show, Typeable)+instance Exception ProcessTrackerException++-- | Begin tracking the given process. If the 'ProcessHandle' refers to a+-- closed process, no tracking will occur. If the process is closed, then it+-- will be untracked automatically.+--+-- Note that you /must/ compile your program with @-threaded@; see+-- 'waitForProcess'.+--+-- Since 0.2.1+trackProcess :: ProcessTracker -> ProcessHandle -> IO TrackedProcess+trackProcess pt ph = mask_ $ do+ mpid <- readMVar $ processHandleMVar ph+ mpid' <- case mpid of+ ClosedHandle{} -> return NoPid+ OpenHandle pid -> do+ c_track_process pt pid 1+ return $ Pid pid+ ipid <- newIORef mpid'+ baton <- newEmptyMVar+ let tp = TrackedProcess pt ipid (takeMVar baton)+ case mpid' of+ NoPid -> return ()+ Pid _ -> void $ forkIO $ do+ waitForProcess ph >>= putMVar baton+ untrackProcess tp+ return $! tp++-- | Explicitly remove the given process from the tracked process list in the+-- cleanup process.+--+-- Since 0.2.1+untrackProcess :: TrackedProcess -> IO ()+untrackProcess (TrackedProcess pt ipid _) = mask_ $ do+ mpid <- readIORef ipid+ case mpid of+ NoPid -> return ()+ Pid pid -> do+ c_track_process pt pid 0+ writeIORef ipid NoPid++-- | Fork and execute a subprocess, sending stdout and stderr to the specified+-- rotating log.+--+-- Since 0.2.1+forkExecuteLog :: ByteString -- ^ command+ -> [ByteString] -- ^ args+ -> Maybe [(ByteString, ByteString)] -- ^ environment+ -> Maybe ByteString -- ^ working directory+ -> Maybe (Source IO ByteString) -- ^ stdin+ -> (ByteString -> IO ()) -- ^ both stdout and stderr will be sent to this location+ -> IO ProcessHandle+forkExecuteLog cmd args menv mwdir mstdin rlog = bracketOnError+ setupPipe+ cleanupPipes+ usePipes+ where+ setupPipe = bracketOnError+ createPipe+ (\(x, y) -> closeFd x `finally` closeFd y)+ (\(x, y) -> (,) <$> fdToHandle x <*> fdToHandle y)+ cleanupPipes (x, y) = hClose x `finally` hClose y++ usePipes pipes@(readerH, writerH) = do+ (min, _, _, ph) <- createProcess CreateProcess+ { cmdspec = RawCommand (S8.unpack cmd) (map S8.unpack args)+ , cwd = S8.unpack <$> mwdir+ , env = map (S8.unpack *** S8.unpack) <$> menv+ , std_in = maybe Inherit (const CreatePipe) mstdin+ , std_out = UseHandle writerH+ , std_err = UseHandle writerH+ , close_fds = True+ , create_group = True+#if MIN_VERSION_process(1, 2, 0)+ , delegate_ctlc = False+#endif+ }+ ignoreExceptions $ addAttachMessage pipes ph+ void $ forkIO $ ignoreExceptions $+ (sourceHandle readerH $$ CL.mapM_ rlog) `finally` hClose readerH+ case (min, mstdin) of+ (Just h, Just source) -> void $ forkIO $ ignoreExceptions $+ (source $$ sinkHandle h) `finally` hClose h+ (Nothing, Nothing) -> return ()+ _ -> error $ "Invariant violated: Data.Conduit.Process.Unix.forkExecuteLog"+ return ph++ addAttachMessage pipes ph = withProcessHandle_ ph $ \p_ -> do+ now <- getCurrentTime+ case p_ of+ ClosedHandle ec -> do+ rlog $ S8.concat+ [ "\n\n"+ , S8.pack $ show now+ , ": Process immediately died with exit code "+ , S8.pack $ show ec+ , "\n\n"+ ]+ cleanupPipes pipes+ OpenHandle h -> do+ rlog $ S8.concat+ [ "\n\n"+ , S8.pack $ show now+ , ": Attached new process "+ , S8.pack $ show h+ , "\n\n"+ ]+ return p_++data Status = NeedsRestart | NoRestart | Running ProcessHandle++-- | Run the given command, restarting if the process dies.+monitorProcess+ :: (ByteString -> IO ()) -- ^ log+ -> ProcessTracker+ -> Maybe S8.ByteString -- ^ setuid+ -> S8.ByteString -- ^ executable+ -> S8.ByteString -- ^ working directory+ -> [S8.ByteString] -- ^ command line parameter+ -> [(S8.ByteString, S8.ByteString)] -- ^ environment+ -> (ByteString -> IO ())+ -> (ExitCode -> IO Bool) -- ^ should we restart?+ -> IO MonitoredProcess+monitorProcess log processTracker msetuid exec dir args env' rlog shouldRestart = do+ mstatus <- newMVar NeedsRestart+ let loop mlast = do+ next <- modifyMVar mstatus $ \status ->+ case status of+ NoRestart -> return (NoRestart, return ())+ _ -> do+ now <- getCurrentTime+ case mlast of+ Just last | diffUTCTime now last < 5 -> do+ log $ "Process restarting too quickly, waiting before trying again: " `S8.append` exec+ threadDelay $ 5 * 1000 * 1000+ _ -> return ()+ let (cmd, args') =+ case msetuid of+ Nothing -> (exec, args)+ Just setuid -> ("sudo", "-E" : "-u" : setuid : "--" : exec : args)+ res <- try $ forkExecuteLog+ cmd+ args'+ (Just env')+ (Just dir)+ (Just $ return ())+ rlog+ case res of+ Left e -> do+ log $ "Data.Conduit.Process.Unix.monitorProcess: " `S8.append` S8.pack (show (e :: SomeException))+ return (NeedsRestart, return ())+ Right pid -> do+ log $ "Process created: " `S8.append` exec+ return (Running pid, do+ TrackedProcess _ _ wait <- trackProcess processTracker pid+ ec <- wait+ shouldRestart' <- shouldRestart ec+ if shouldRestart'+ then loop (Just now)+ else return ())+ next+ _ <- forkIO $ loop Nothing+ return $ MonitoredProcess mstatus++-- | Abstract type containing information on a process which will be restarted.+newtype MonitoredProcess = MonitoredProcess (MVar Status)++-- | Terminate the process and prevent it from being restarted.+terminateMonitoredProcess :: MonitoredProcess -> IO ()+terminateMonitoredProcess (MonitoredProcess mstatus) = do+ status <- swapMVar mstatus NoRestart+ case status of+ Running pid -> do+ terminateProcess pid+ threadDelay 1000000+ killProcess pid+ _ -> return ()
Keter/App.hs view
@@ -20,8 +20,9 @@ import Control.Exception (IOException, try) import Control.Monad (void, when) import qualified Data.Conduit.LogFile as LogFile+import Data.Conduit.LogFile (RotatingLog) import Data.Conduit.Process.Unix (MonitoredProcess, ProcessTracker,- RotatingLog, monitorProcess,+ monitorProcess, terminateMonitoredProcess) import Data.IORef import qualified Data.Map as Map@@ -271,7 +272,7 @@ (maybe "/tmp" (encodeUtf8 . either id id . F.toText) mdir) (map encodeUtf8 $ V.toList waconfigArgs) (map (encodeUtf8 *** encodeUtf8) env)- rlog+ (LogFile.addChunk rlog) (const $ return True)) terminateMonitoredProcess $ \mp -> f RunningWebApp@@ -358,7 +359,7 @@ (maybe "/tmp" (encodeUtf8 . either id id . F.toText) mdir) (map encodeUtf8 $ V.toList bgconfigArgs) (map (encodeUtf8 *** encodeUtf8) env)- rlog+ (LogFile.addChunk rlog) (const shouldRestart)) terminateMonitoredProcess (f . RunningBackgroundApp)
+ cbits/process-tracker.c view
@@ -0,0 +1,100 @@+#include <unistd.h>+#include <stdlib.h>+#include <stdio.h>+#include <signal.h>++struct node {+ pid_t pid;+ struct node *next;+};++static struct node * add_node(pid_t pid, struct node *n) {+ struct node *n2 = malloc(sizeof(struct node));+ n2->pid = pid;+ n2->next = n;+ return n2;+}++static struct node * remove_node(pid_t pid, struct node *n) {+ if (!n) {+ return n;+ }+ else if (n->pid == pid) {+ struct node *n2 = n->next;+ free(n);+ return remove_node(pid, n2);+ }+ else {+ n->next = remove_node(pid, n->next);+ return n;+ }+}++extern void track_process(int fd, pid_t pid, int b) {+ unsigned int buffer[2];++ //printf("Tracking process %d, %d\n", pid, b);++ buffer[0] = pid;+ buffer[1] = b;+ if (! write(fd, buffer, sizeof(unsigned int) * 2)) {+ //printf("Error writing to fd %d\n", fd);+ }+}++// Returns FD to write to, or -1 on failure.+extern int launch_process_tracker(void) {+ int pipes[2];+ pid_t child;++ if (pipe(pipes)) {+ return -1;+ }++ child = fork();++ if (child < 0) {+ return -1;+ }+ else if (child == 0) {+ unsigned int buffer[2];+ struct node *n = 0, *n2;++ close(pipes[1]);++ // Prevent monitoring programs like Upstart from killing this+ // new process along with the parent+ setpgid(0, 0);++ while (read(pipes[0], buffer, sizeof(unsigned int) * 2) > 0) {+ if (buffer[1]) {+ //printf("Adding node %d\n", buffer[0]);+ n = add_node(buffer[0], n);+ }+ else {+ //printf("Removing node %d\n", buffer[0]);+ n = remove_node(buffer[0], n);+ }+ }++ for (n2 = n; n2; n2 = n2->next) {+ //printf("Sending process %d TERM signal\n", n2->pid);+ kill(n2->pid, SIGTERM);+ }++ sleep(2);++ while (n) {+ //printf("Sending process %d KILL signal\n", n->pid);+ kill(n2->pid, SIGKILL);++ n2 = n;+ n = n->next;+ free(n2);+ }+ }+ else {+ close(pipes[0]);+ return pipes[1];+ }+}
keter.cabal view
@@ -1,5 +1,5 @@ Name: keter-Version: 1.2.0+Version: 1.2.1 Synopsis: Web application deployment manager, focusing on Haskell web frameworks Description: Handles deployment of web apps, providing a reverse proxy to achieve zero downtime deployments. For more information, please see the README on Github: <https://github.com/snoyberg/keter#readme>@@ -52,7 +52,6 @@ , network-conduit >= 0.6 , network-conduit-tls >= 1.0.0.2 , http-reverse-proxy >= 0.3.1 && < 0.4- , unix-process-conduit >= 0.2.2 && < 0.3 , unix >= 2.5 , wai-app-static >= 2.0 && < 2.1 , wai >= 2.1 && < 2.2@@ -88,7 +87,10 @@ Network.HTTP.ReverseProxy.Rewrite Data.Yaml.FilePath Codec.Archive.TempTarball+ Data.Conduit.LogFile+ Data.Conduit.Process.Unix ghc-options: -Wall+ c-sources: cbits/process-tracker.c Executable keter Main-is: keter.hs@@ -112,6 +114,19 @@ , transformers ghc-options: -threaded -Wall buildable: False++test-suite test+ hs-source-dirs: test+ main-is: Spec.hs+ type: exitcode-stdio-1.0+ build-depends: base+ , transformers+ , conduit+ , bytestring+ , hspec >= 1.3+ , unix+ , keter+ ghc-options: -Wall -threaded source-repository head type: git
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}