diff --git a/Changelog.md b/Changelog.md --- a/Changelog.md +++ b/Changelog.md @@ -1,3 +1,10 @@ +### 1.2.1 + +- Improvements and bug fixes in module *NgxExport.Tools.Resolve*. + + Added ability to set response timeout in service *collectUpstreams*. + + Use *handleAny* from *safe-exceptions* in service *collectUpstreams*. + + Improved type declarations and documentation. + ### 1.2.0 - Added module *NgxExport.Tools.Resolve* which can be used to manage service diff --git a/NgxExport/Tools/Aggregate.hs b/NgxExport/Tools/Aggregate.hs --- a/NgxExport/Tools/Aggregate.hs +++ b/NgxExport/Tools/Aggregate.hs @@ -127,7 +127,7 @@ -- -- updateStats :: ByteString -> IO C8L.ByteString -- __/updateStats/__ s = do --- let cbs = 'readFromByteString' \@Int s +-- let cbs = 'NgxExport.Tools.Read.readFromByteString' \@Int s -- modifyIORef\' stats $ \\(Stats bs rs _) -> -- let !nbs = bs + fromMaybe 0 cbs -- !nrs = rs + 1 diff --git a/NgxExport/Tools/Resolve.hs b/NgxExport/Tools/Resolve.hs --- a/NgxExport/Tools/Resolve.hs +++ b/NgxExport/Tools/Resolve.hs @@ -1,5 +1,5 @@ -{-# LANGUAGE TemplateHaskell, ScopedTypeVariables, RecordWildCards #-} -{-# LANGUAGE BangPatterns, OverloadedStrings #-} +{-# LANGUAGE TemplateHaskell, RecordWildCards, BangPatterns, NumDecimals #-} +{-# LANGUAGE OverloadedStrings #-} ----------------------------------------------------------------------------- -- | @@ -14,6 +14,25 @@ -- DNS resolve utilities from the more extra tools collection for -- . -- +-- __An important note.__ Currently, package /resolv/ at +-- [hackage.org](https://hackage.haskell.org/package/resolv) has +-- [a bug](https://github.com/haskell-hvr/resolv/pull/12) which leads to +-- memory leaks on every DNS query. This makes service /collectUpstreams/ +-- from module /NgxExport.Tools.Resolve/ leak as well, because it makes DNS +-- queries regularly. To prevent memory leaks, you can clone /resolv/ from +-- [this fork](https://github.com/lyokha/resolv) and /v1-install/ it from +-- the source. Or, if you prefer /v2-build/, simply put lines +-- +-- @ +-- source-repository-package +-- type: git +-- location: https://github.com/lyokha/resolv.git +-- tag: 6a46c2659f79e78defd974849a8120548257cadc +-- post-checkout-command: autoreconf -i +-- @ +-- +-- into the /cabal.project/ file. + ----------------------------------------------------------------------------- module NgxExport.Tools.Resolve ( @@ -24,10 +43,9 @@ UName ,SAddress ,UQuery (..) - ,PriorityList (..) + ,PriorityPolicy (..) ,UData (..) ,ServerData (..) - ,CollectedServerDataGen ,CollectedServerData -- * Exported functions ,collectA @@ -54,9 +72,11 @@ import Data.Bits import Control.Concurrent.Async import Control.Exception +import Control.Exception.Safe (handleAny) import Control.Arrow import Control.Monad import System.IO.Unsafe +import System.Timeout -- $dynamicUpstreams -- @@ -113,16 +133,17 @@ -- -- haskell_run_service __/simpleService_collectUpstreams/__ $hs_upstreams -- \'Conf { upstreams = --- [UData { uQuery = --- QuerySRV --- (Name \"_http._tcp.mycompany.com\") --- (SinglePriority \"__/utest/__\") --- , uMaxFails = 0 --- , uFailTimeout = 10 +-- ['UData' { 'uQuery' = +-- 'QuerySRV' +-- ('Name' \"_http._tcp.mycompany.com\") +-- ('SinglePriority' \"__/utest/__\") +-- , 'uMaxFails' = 1 +-- , 'uFailTimeout' = 10 -- } -- ] --- , maxWait = Sec 300 --- , waitOnException = Sec 2 +-- , maxWait = 'Sec' 300 +-- , waitOnException = 'Sec' 2 +-- , responseTimeout = 'Unset' -- }\'; -- -- haskell_service_var_ignore_empty $hs_upstreams; @@ -180,6 +201,10 @@ -- servers does not exceed the value of /maxWait/, then the service will restart -- in this time. -- +-- Too big response times may also cause exceptions during the collection of the +-- servers. The timeout is defined by the value of /responseTimeout/. In our +-- example, the timeout is not set. +-- -- Notice that we used /QuerySRV/ and /SinglePriority \"utest\"/. The latter -- means that all collected servers will inhabit upstream /utest/ regardless of -- their priority values. To distribute collected servers among a number of @@ -189,16 +214,17 @@ -- @ -- haskell_run_service __/simpleService_collectUpstreams/__ $hs_upstreams -- \'Conf { upstreams = --- [UData { uQuery = --- QuerySRV --- (Name \"_http._tcp.mycompany.com\") --- (PriorityList [\"__/utest/__\", \"__/utest1/__\"]) --- , uMaxFails = 0 --- , uFailTimeout = 10 +-- ['UData' { 'uQuery' = +-- 'QuerySRV' +-- ('Name' \"_http._tcp.mycompany.com\") +-- ('PriorityList' [\"__/utest/__\", \"__/utest1/__\"]) +-- , 'uMaxFails' = 1 +-- , 'uFailTimeout' = 10 -- } -- ] --- , maxWait = Sec 300 --- , waitOnException = Sec 2 +-- , maxWait = 'Sec' 300 +-- , waitOnException = 'Sec' 2 +-- , responseTimeout = 'Unset' -- }\'; -- @ -- @@ -255,37 +281,47 @@ -- for the collected list of domain names, -- - the same as the previous, but distribute collected servers among a list of -- upstreams according to the collected priorities. -data UQuery = QueryA [Name] UName -- ^ Query /A/ records - | QuerySRV Name PriorityList -- ^ Query an /SRV/ record +data UQuery = QueryA [Name] UName -- ^ Query /A/ records + | QuerySRV Name (PriorityPolicy UName) -- ^ Query an /SRV/ record deriving Read --- | Specifies how to distribute collected servers among the given upstreams. -data PriorityList = SinglePriority UName -- ^ All servers to one upstream - | PriorityList [UName] -- ^ Distribute servers by priorities - deriving Read +-- | Priority policy. +-- +-- Specifies how to distribute collected items by priorities. In particular, +-- /PriorityPolicy UName/ specifies how to distribute collected servers among +-- the given upstreams. +data PriorityPolicy a = SinglePriority a -- ^ All items go to a single element + | PriorityList [a] -- ^ Distribute items by priorities + deriving Read -- | Upstream configuration. -- --- Includes DNS query model and data for Nginx /server/ description. +-- Includes DNS query model and parameters for Nginx /server/ description. +-- Values of /uMaxFails/ and /uFailTimeout/ get assigned to each collected +-- server as /max_fails/ and /fail_timeout/ respectively. The weight of an +-- individual server gets picked from the value of 'srvWeight' collected in +-- /SRV/ queries. Note that setting of parameters /max_conns/, /backup/ and +-- /down/ is not supported. data UData = UData { uQuery :: UQuery -- ^ DNS query model - , uMaxFails :: Int -- ^ /maxFails/ - , uFailTimeout :: Int -- ^ /failTimeout/ + , uMaxFails :: Int -- ^ /max_fails/ + , uFailTimeout :: Int -- ^ /fail_timeout/ } deriving Read data Conf = Conf { upstreams :: [UData] , maxWait :: TimeInterval , waitOnException :: TimeInterval + , responseTimeout :: TimeInterval } deriving Read newtype Upconf = Upconf { upconfAddr :: (SUrl, SAddress) } deriving Read -- | Server data. -- --- The fields map exactly to data from Nginx /server/ description. +-- The fields map exactly to parameters from Nginx /server/ description. data ServerData = ServerData { sAddr :: SAddress -- ^ Server address , sWeight :: Maybe Int -- ^ /weight/ - , sMaxFails :: Maybe Int -- ^ /maxFails/ - , sFailTimeout :: Maybe Int -- ^ /failTimeout/ + , sMaxFails :: Maybe Int -- ^ /max_fails/ + , sFailTimeout :: Maybe Int -- ^ /fail_timeout/ } deriving (Show, Eq, Ord) instance FromJSON ServerData where @@ -304,22 +340,10 @@ , ("fail_timeout" .=) <$> sFailTimeout ] --- | Generic type to collect and store server data. --- --- Type /a/ is instantiated either by 'TTL' (to collect) or 'TimeInterval' --- (to store). -type CollectedServerDataGen a = (a, Map UName [ServerData]) - -- | Collected server data. --- --- The first element of the tuple gets transformed into the time interval before --- the next run of the /collectUpstreams/ service. The second element contains --- the collected data. -type CollectedServerData = CollectedServerDataGen TTL - -type CollectedServerDataStore = CollectedServerDataGen TimeInterval +type CollectedServerData = Map UName [ServerData] -collectedServerData :: IORef CollectedServerDataStore +collectedServerData :: IORef (TimeInterval, CollectedServerData) collectedServerData = unsafePerformIO $ newIORef (Unset, M.empty) {-# NOINLINE collectedServerData #-} @@ -359,7 +383,7 @@ -- -- After getting the /SRV/ record, runs 'collectA' for each collected element. -- --- Returns a list of IP addresses wrapped in 'SRV' container and the minimum +-- Returns a list of IP addresses wrapped in an 'SRV' container and the minimum -- value of their TTLs. If the list is empty, then the returned TTL value gets -- taken from the first argument. collectSRV @@ -398,12 +422,17 @@ where showAddr i p = showIPv4 i ++ ':' : show p -- | Collects server data for the given upstream configuration. +-- +-- Returns the collected server data and the minimum value of all the collected +-- TTLs. If this TTL value, having been converted into a 'TimeInterval', is not +-- bigger than /maxWait/, then it defines in how long time service +-- /collectUpstreams/, which calls this function, will restart again. collectServerData :: TTL -- ^ Fallback TTL value -> UData -- ^ Upstream configuration - -> IO CollectedServerData -collectServerData lTTL (UData (QueryA [] u) _ _) = - return (lTTL, M.singleton u []) + -> IO (TTL, CollectedServerData) +collectServerData lTTL (UData (QueryA [] _) _ _) = + return (lTTL, M.empty) collectServerData lTTL ud@(UData (QueryA ns u) _ _) = do a <- mapConcurrently (collectA lTTL) ns return $ @@ -430,22 +459,14 @@ groupBy ((==) `on` srvPriority) . sortOn srvPriority withTrail = uncurry (++) . (id &&& repeat . last) -handleCollectErrors :: TimeInterval -> IO [CollectedServerData] -> - IO [CollectedServerData] -handleCollectErrors wt = - handle (\(e :: SomeException) -> do - writeIORef collectedServerData (wt, M.empty) - throwIO e - ) - collectUpstreams :: Conf -> Bool -> IO L.ByteString collectUpstreams Conf {..} = const $ do (wt, old) <- readIORef collectedServerData when (wt /= Unset) $ threadDelaySec $ toSec wt let (lTTL, hTTL) = (toTTL waitOnException, toTTL maxWait) - srv <- handleCollectErrors waitOnException $ - mapConcurrently (collectServerData lTTL) upstreams - let nwt = fromTTL $ min hTTL $ minimumTTL lTTL $ map fst srv + srv <- handleCollectErrors $ + mapConcurrently (withTimeout . collectServerData lTTL) upstreams + let nwt = fromTTL $ max (TTL 1) $ min hTTL $ minimumTTL lTTL $ map fst srv new = mconcat $ map snd srv if new == old then do @@ -457,6 +478,17 @@ return $ encode new where toTTL = TTL . fromIntegral . toSec fromTTL (TTL ttl) = Sec $ fromIntegral ttl + handleCollectErrors = handleAny $ \e -> do + writeIORef collectedServerData (waitOnException, M.empty) + throwIO e + withTimeout act = do + r <- timeout (toTimeout responseTimeout) act + case r of + Nothing -> throwIO $ + userError "Collection of server data was timed out" + Just r' -> return r' + toTimeout Unset = -1 + toTimeout v = toSec v * 1e6 ngxExportSimpleServiceTyped 'collectUpstreams ''Conf $ PersistentService Nothing diff --git a/ngx-export-tools-extra.cabal b/ngx-export-tools-extra.cabal --- a/ngx-export-tools-extra.cabal +++ b/ngx-export-tools-extra.cabal @@ -1,5 +1,5 @@ name: ngx-export-tools-extra -version: 1.2.0 +version: 1.2.1 synopsis: More extra tools for Nginx haskell module description: More extra tools for . @@ -47,6 +47,7 @@ , network >= 2.4.0.0 , async >= 2.0.1.0 , aeson >= 1.0.0.0 + , safe-exceptions , resolv , time , binary