diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -7,36 +7,41 @@ {-# LANGUAGE TupleSections #-} module Main where -import Control.Arrow (second) -import Control.Monad (when) -import Control.Monad.Logger (LogLevel (..), filterLogger, logInfoS, - runStderrLoggingT) -import Data.Char (toLower) -import Data.Default (Default (..)) -import Data.List (intercalate) -import Data.Maybe (fromMaybe) -import Data.String.Conversions (cs) -import Data.Word (Word32) -import Haskoin (Network (..), allNets, bch, - bchRegTest, bchTest, btc, btcRegTest, - btcTest, eitherToMaybe) -import Haskoin.Node (withConnection) -import Haskoin.Store (StoreConfig (..), WebConfig (..), - WebLimits (..), WebTimeouts (..), - runWeb, withStore) -import Options.Applicative (Parser, auto, eitherReader, - execParser, flag, fullDesc, header, - help, helper, info, long, many, - metavar, option, progDesc, short, - showDefault, strOption, switch, value) -import System.Exit (exitSuccess) -import System.FilePath (()) -import System.IO.Unsafe (unsafePerformIO) -import Text.Read (readMaybe) -import UnliftIO (MonadIO) -import UnliftIO.Directory (createDirectoryIfMissing, - getAppUserDataDirectory) -import UnliftIO.Environment (lookupEnv) +import Control.Applicative ((<|>)) +import Control.Arrow (second) +import Control.Monad (when) +import Control.Monad.Logger (LogLevel (..), filterLogger, + logInfoS, runStderrLoggingT) +import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) +import Data.Char (toLower) +import Data.Default (Default (..)) +import Data.List (intercalate) +import Data.Maybe (fromMaybe) +import Data.String.Conversions (cs) +import qualified Data.Text as T +import Data.Word (Word32) +import Haskoin (Network (..), allNets, bch, + bchRegTest, bchTest, btc, + btcRegTest, btcTest, eitherToMaybe) +import Haskoin.Node (withConnection) +import Haskoin.Store (StoreConfig (..), WebConfig (..), + WebLimits (..), WebTimeouts (..), + runWeb, withStore) +import Haskoin.Store.Stats (withStats) +import Options.Applicative (Parser, auto, eitherReader, + execParser, flag, fullDesc, header, + help, helper, info, long, many, + metavar, option, progDesc, short, + showDefault, strOption, switch, + value) +import System.Exit (exitSuccess) +import System.FilePath (()) +import System.IO.Unsafe (unsafePerformIO) +import Text.Read (readMaybe) +import UnliftIO (MonadIO) +import UnliftIO.Directory (createDirectoryIfMissing, + getAppUserDataDirectory) +import UnliftIO.Environment (lookupEnv) version :: String #ifdef CURRENT_PACKAGE_VERSION @@ -69,9 +74,12 @@ , configMaxPeers :: !Int , configMaxDiff :: !Int , configCacheRefresh :: !Int - , configCacheRetries :: !Int - , configCacheRetryDelay :: Int + , configCacheRetryDelay :: !Int , configNumTxId :: !Bool + , configStatsd :: !Bool + , configStatsdHost :: !String + , configStatsdPort :: !Int + , configStatsdPrefix :: !String } instance Default Config where @@ -98,9 +106,12 @@ , configMaxPeers = defMaxPeers , configMaxDiff = defMaxDiff , configCacheRefresh = defCacheRefresh - , configCacheRetries = defCacheRetries , configCacheRetryDelay = defCacheRetryDelay , configNumTxId = defNumTxId + , configStatsd = defStatsd + , configStatsdHost = defStatsdHost + , configStatsdPort = defStatsdPort + , configStatsdPrefix = defStatsdPrefix } defEnv :: MonadIO m => String -> a -> (String -> Maybe a) -> m a @@ -113,11 +124,6 @@ defEnv "CACHE_REFRESH" 750 readMaybe {-# NOINLINE defCacheRefresh #-} -defCacheRetries :: Int -defCacheRetries = unsafePerformIO $ - defEnv "CACHE_RETRIES" 100 readMaybe -{-# NOINLINE defCacheRetries #-} - defCacheRetryDelay :: Int defCacheRetryDelay = unsafePerformIO $ defEnv "CACHE_RETRY_DELAY" 100000 readMaybe @@ -250,6 +256,36 @@ defEnv "MAX_DIFF" 2 readMaybe {-# NOINLINE defMaxDiff #-} +defStatsd :: Bool +defStatsd = unsafePerformIO $ + defEnv "STATSD" False parseBool +{-# NOINLINE defStatsd #-} + +defStatsdHost :: String +defStatsdHost = unsafePerformIO $ + defEnv "STATSD_HOST" "localhost" pure +{-# NOINLINE defStatsdHost #-} + +defStatsdPort :: Int +defStatsdPort = unsafePerformIO $ + defEnv "STATSD_PORT" 8125 readMaybe +{-# NOINLINE defStatsdPort #-} + +defStatsdPrefix :: String +defStatsdPrefix = + unsafePerformIO $ + runMaybeT go >>= \case + Nothing -> return "haskoin_store" + Just x -> return x + where + go = prefix <|> nomad + prefix = MaybeT $ lookupEnv "STATSD_PREFIX" + nomad = do + task <- MaybeT $ lookupEnv "NOMAD_TASK_NAME" + service <- MaybeT $ lookupEnv "NOMAD_ALLOC_INDEX" + return $ task <> "." <> service +{-# NOINLINE defStatsdPrefix #-} + netNames :: String netNames = intercalate "|" (map getNetworkName allNets) @@ -439,13 +475,6 @@ <> help "Refresh cache this frequently" <> showDefault <> value (configCacheRefresh def) - configCacheRetries <- - option auto $ - metavar "INT" - <> long "cache-retries" - <> help "Retry getting cache lock to index xpub" - <> showDefault - <> value (configCacheRetries def) configCacheRetryDelay <- option auto $ metavar "MICROSECONDS" @@ -472,6 +501,31 @@ flag (configNumTxId def) True $ long "numtxid" <> help "Numeric tx_index field" + configStatsd <- + flag (configStatsd def) True $ + long "statsd" + <> help "Enable statsd metrics" + configStatsdHost <- + strOption $ + metavar "HOST" + <> long "statsd-host" + <> help "Host to send statsd metrics" + <> showDefault + <> value (configStatsdHost def) + configStatsdPort <- + option auto $ + metavar "PORT" + <> long "statsd-port" + <> help "Port to send statsd metrics" + <> showDefault + <> value (configStatsdPort def) + configStatsdPrefix <- + strOption $ + metavar "PREFIX" + <> long "statsd-prefix" + <> help "Prefix for statsd metrics" + <> showDefault + <> value (configStatsdPrefix def) pure Config { configWebLimits = WebLimits {..} @@ -543,11 +597,14 @@ , configMaxDiff = maxdiff , configNoMempool = nomem , configCacheRefresh = crefresh - , configCacheRetries = cretries , configCacheRetryDelay = cretrydelay , configNumTxId = numtxid + , configStatsd = statsd + , configStatsdHost = statsdhost + , configStatsdPort = statsdport + , configStatsdPrefix = statsdpfx } = - runStderrLoggingT . filterLogger l $ do + runStderrLoggingT . filterLogger l . with_stats $ \stats -> do $(logInfoS) "Main" $ "Creating working directory if not found: " <> cs wd createDirectoryIfMissing True wd @@ -573,8 +630,8 @@ , storeConfPeerMaxLife = fromIntegral peerlife , storeConfConnect = withConnection , storeConfCacheRefresh = crefresh - , storeConfCacheRetries = cretries , storeConfCacheRetryDelay = cretrydelay + , storeConfStats = stats } withStore scfg $ \st -> runWeb @@ -589,8 +646,21 @@ , webMaxDiff = maxdiff , webNoMempool = nomem , webNumTxId = numtxid + , webStats = stats } where + with_stats go + | statsd = do + $(logInfoS) "Main" $ + "Sending stats to " <> T.pack statsdhost <> + ":" <> cs (show statsdport) <> + " with prefix: " <> T.pack statsdpfx + withStats + (T.pack statsdhost) + statsdport + (T.pack statsdpfx) + (go . Just) + | otherwise = go Nothing net' | asert == 0 = net | otherwise = net { getAsertActivationTime = Just asert } l _ lvl diff --git a/haskoin-store.cabal b/haskoin-store.cabal --- a/haskoin-store.cabal +++ b/haskoin-store.cabal @@ -1,11 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.34.2. +-- This file has been generated from package.yaml by hpack version 0.33.0. -- -- see: https://github.com/sol/hpack +-- +-- hash: 46a8bc9a10e9bbf0faff3e73de4e05ef0a483024f92087f321ee3be8c67f053e name: haskoin-store -version: 0.40.18 +version: 0.40.19 synopsis: Storage and index for Bitcoin and Bitcoin Cash description: Please see the README on GitHub at category: Bitcoin, Finance, Network @@ -32,6 +34,7 @@ Haskoin.Store.Database.Writer Haskoin.Store.Logic Haskoin.Store.Manager + Haskoin.Store.Stats Haskoin.Store.Web other-modules: Paths_haskoin_store @@ -47,6 +50,9 @@ , containers >=0.6.2.1 , data-default >=0.7.1.1 , deepseq >=1.4.4.0 + , ekg-core >=0.1.1.7 + , ekg-statsd >=0.2.5 + , foldl >=1.4.10 , hashable >=1.3.0.0 , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 @@ -54,6 +60,7 @@ , hedis >=0.12.13 , http-types >=0.12.3 , lens >=4.18.1 + , monad-control >=1.0.2.3 , monad-logger >=0.3.32 , mtl >=2.2.2 , network >=3.1.1.1 @@ -62,6 +69,7 @@ , rocksdb-haskell-jprupp >=2.1.3 , rocksdb-query >=0.4.0 , scotty >=0.11.5 + , stm >=2.5.0.0 , string-conversions >=0.4.0.1 , text >=1.2.3.0 , time >=1.9.3 @@ -90,7 +98,10 @@ , containers >=0.6.2.1 , data-default >=0.7.1.1 , deepseq >=1.4.4.0 + , ekg-core >=0.1.1.7 + , ekg-statsd >=0.2.5 , filepath + , foldl >=1.4.10 , hashable >=1.3.0.0 , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 @@ -99,6 +110,7 @@ , hedis >=0.12.13 , http-types >=0.12.3 , lens >=4.18.1 + , monad-control >=1.0.2.3 , monad-logger >=0.3.32 , mtl >=2.2.2 , network >=3.1.1.1 @@ -108,6 +120,7 @@ , rocksdb-haskell-jprupp >=2.1.3 , rocksdb-query >=0.4.0 , scotty >=0.11.5 + , stm >=2.5.0.0 , string-conversions >=0.4.0.1 , text >=1.2.3.0 , time >=1.9.3 @@ -141,6 +154,9 @@ , containers >=0.6.2.1 , data-default >=0.7.1.1 , deepseq >=1.4.4.0 + , ekg-core >=0.1.1.7 + , ekg-statsd >=0.2.5 + , foldl >=1.4.10 , hashable >=1.3.0.0 , haskoin-core >=0.19.0 , haskoin-node >=0.17.0 @@ -150,6 +166,7 @@ , hspec >=2.7.1 , http-types >=0.12.3 , lens >=4.18.1 + , monad-control >=1.0.2.3 , monad-logger >=0.3.32 , mtl >=2.2.2 , network >=3.1.1.1 @@ -158,6 +175,7 @@ , rocksdb-haskell-jprupp >=2.1.3 , rocksdb-query >=0.4.0 , scotty >=0.11.5 + , stm >=2.5.0.0 , string-conversions >=0.4.0.1 , text >=1.2.3.0 , time >=1.9.3 diff --git a/src/Haskoin/Store/BlockStore.hs b/src/Haskoin/Store/BlockStore.hs --- a/src/Haskoin/Store/BlockStore.hs +++ b/src/Haskoin/Store/BlockStore.hs @@ -82,10 +82,14 @@ getOldMempool, importBlock, initBest, newMempoolTx, revertBlock) +import Haskoin.Store.Stats import NQE (Listen, Mailbox, Publisher, inboxToMailbox, newInbox, publish, query, receive, send, sendSTM) +import qualified System.Metrics as Metrics +import qualified System.Metrics.Gauge as Metrics (Gauge) +import qualified System.Metrics.Gauge as Metrics.Gauge import System.Random (randomRIO) import UnliftIO (Exception, MonadIO, MonadUnliftIO, STM, TVar, async, @@ -136,8 +140,73 @@ , myPeer :: !(TVar (Maybe Syncing)) , myTxs :: !(TVar (HashMap TxHash PendingTx)) , requested :: !(TVar (HashSet TxHash)) + , myMetrics :: !(Maybe StoreMetrics) } +data StoreMetrics = StoreMetrics + { storeHeight :: !Metrics.Gauge + , headersHeight :: !Metrics.Gauge + , storePendingTxs :: !Metrics.Gauge + , storePeersConnected :: !Metrics.Gauge + , storeMempoolSize :: !Metrics.Gauge + } + +newStoreMetrics :: MonadIO m => Metrics.Store -> m StoreMetrics +newStoreMetrics s = liftIO $ do + storeHeight <- g "height" + headersHeight <- g "headers" + storePendingTxs <- g "pending_txs" + storePeersConnected <- g "peers_connected" + storeMempoolSize <- g "mempool_size" + return StoreMetrics{..} + where + g x = Metrics.createGauge ("store." <> x) s + +setStoreHeight :: MonadIO m => BlockT m () +setStoreHeight = + asks myMetrics >>= \case + Nothing -> return () + Just m -> + getBestBlock >>= \case + Nothing -> setit m 0 + Just bb -> getBlock bb >>= \case + Nothing -> setit m 0 + Just b -> setit m (blockDataHeight b) + where + setit m i = liftIO $ storeHeight m `Metrics.Gauge.set` fromIntegral i + +setHeadersHeight :: MonadIO m => BlockT m () +setHeadersHeight = + asks myMetrics >>= \case + Nothing -> return () + Just m -> do + h <- fmap nodeHeight $ chainGetBest =<< asks (blockConfChain . myConfig) + liftIO $ headersHeight m `Metrics.Gauge.set` fromIntegral h + +setPendingTxs :: MonadIO m => BlockT m () +setPendingTxs = + asks myMetrics >>= \case + Nothing -> return () + Just m -> do + s <- asks myTxs >>= \t -> atomically (HashMap.size <$> readTVar t) + liftIO $ storePendingTxs m `Metrics.Gauge.set` fromIntegral s + +setPeersConnected :: MonadIO m => BlockT m () +setPeersConnected = + asks myMetrics >>= \case + Nothing -> return () + Just m -> do + ps <- fmap length $ getPeers =<< asks (blockConfManager . myConfig) + liftIO $ storePeersConnected m `Metrics.Gauge.set` fromIntegral ps + +setMempoolSize :: MonadIO m => BlockT m () +setMempoolSize = + asks myMetrics >>= \case + Nothing -> return () + Just m -> do + s <- length <$> getMempool + liftIO $ storeMempoolSize m `Metrics.Gauge.set` fromIntegral s + -- | Configuration for a block store. data BlockStoreConfig = BlockStoreConfig @@ -157,6 +226,7 @@ -- ^ wipe mempool at start , blockConfPeerTimeout :: !NominalDiffTime -- ^ disconnect syncing peer if inactive for this long + , blockConfStats :: !(Maybe Metrics.Store) } type BlockT m = ReaderT BlockStore m @@ -216,11 +286,13 @@ ts <- newTVarIO HashMap.empty rq <- newTVarIO HashSet.empty inbox <- newInbox + metrics <- mapM newStoreMetrics (blockConfStats cfg) let r = BlockStore { myMailbox = inboxToMailbox inbox , myConfig = cfg , myPeer = pb , myTxs = ts , requested = rq + , myMetrics = metrics } withAsync (runReaderT (go inbox) r) $ \a -> do link a @@ -279,7 +351,6 @@ mempool :: (MonadUnliftIO m, MonadLoggerIO m) => Peer -> BlockT m () mempool p = guardMempool $ void $ async $ do - threadDelay 23849118 isInSync >>= \s -> when s $ do $(logDebugS) "BlockStore" $ "Requesting mempool from peer: " <> peerText p @@ -310,7 +381,7 @@ "Processing block: " <> blockText node Nothing <> " from peer: " <> peerText peer lift $ runImport (importBlock block node) >>= \case - Left e -> failure e + Left e -> failure e Right () -> success node where header = blockHeader block @@ -416,6 +487,8 @@ atomically $ do modifyTVar ts $ HashMap.insert th p modifyTVar rq $ HashSet.delete th + HashMap.size <$> readTVar ts + setPendingTxs where th = txHash (pendingTx p) @@ -438,10 +511,13 @@ || th `HashSet.member` rs pendingTxs :: MonadIO m => Int -> BlockT m [PendingTx] -pendingTxs i = asks myTxs >>= \box -> atomically $ do - pending <- readTVar box - let (selected, rest) = select pending - writeTVar box rest +pendingTxs i = do + selected <- asks myTxs >>= \box -> atomically $ do + pending <- readTVar box + let (selected, rest) = select pending + writeTVar box rest + return (selected) + setPendingTxs return selected where select pend = @@ -560,7 +636,7 @@ block_read (pendingTxTime p) (pendingTx p) >>= \case - True -> return $ Just (txHash (pendingTx p)) + True -> return $ Just (txHash (pendingTx p)) False -> return Nothing import_txs block_read txs = let r = mapM (run_import block_read) txs @@ -795,7 +871,7 @@ box <- asks myPeer readTVarIO box >>= \case Just Syncing { syncingPeer = p' } | p == p' -> reset_it box - _ -> return () + _ -> return () where reset_it box = do atomically $ writeTVar box Nothing @@ -805,7 +881,7 @@ trySetPeer :: MonadLoggerIO m => Peer -> BlockT m Bool trySetPeer p = getSyncingState >>= \case - Just _ -> return False + Just _ -> return False Nothing -> set_it where set_it = @@ -828,14 +904,14 @@ isInSync >>= \case True -> return () False -> getSyncingState >>= \case - Just _ -> return () + Just _ -> return () Nothing -> online_peer where recurse [] = return () recurse (p : ps) = trySetPeer p >>= \case False -> recurse ps - True -> syncMe + True -> syncMe online_peer = do ops <- getPeers =<< asks (blockConfManager . myConfig) let ps = map onlinePeerMailbox ops @@ -847,7 +923,7 @@ True -> mempool p False -> trySetPeer p >>= \case False -> return () - True -> syncMe + True -> syncMe getSyncingState :: (MonadIO m, MonadReader BlockStore m) => m (Maybe Syncing) @@ -858,7 +934,7 @@ :: (MonadLoggerIO m, MonadReader BlockStore m) => m () clearSyncingState = asks myPeer >>= readTVarIO >>= \case - Nothing -> return () + Nothing -> return () Just Syncing { syncingPeer = p } -> finishPeer p processBlockStoreMessage :: (MonadUnliftIO m, MonadLoggerIO m) @@ -891,6 +967,11 @@ pruneOrphans checkTime pruneMempool + setStoreHeight + setHeadersHeight + setPendingTxs + setPeersConnected + setMempoolSize atomically (r ()) pingMe :: MonadLoggerIO m => Mailbox BlockStoreMessage -> m () diff --git a/src/Haskoin/Store/Cache.hs b/src/Haskoin/Store/Cache.hs --- a/src/Haskoin/Store/Cache.hs +++ b/src/Haskoin/Store/Cache.hs @@ -6,12 +6,15 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiWayIf #-} {-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TupleSections #-} module Haskoin.Store.Cache ( CacheConfig(..) + , CacheMetrics , CacheT , CacheError(..) + , newCacheMetrics , withCache , connectRedis , blockRefScore @@ -24,71 +27,86 @@ , evictFromCache ) where -import Control.DeepSeq (NFData) -import Control.Monad (forM, forM_, forever, guard, unless, - void, when) -import Control.Monad.Logger (MonadLoggerIO, logDebugS, logErrorS, - logInfoS, logWarnS) -import Control.Monad.Reader (ReaderT (..), ask, asks) -import Control.Monad.Trans (lift) -import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) -import Data.Bits (shift, (.&.), (.|.)) -import Data.ByteString (ByteString) -import qualified Data.ByteString.Short as BSS -import Data.Default (def) -import Data.Either (rights) -import Data.HashMap.Strict (HashMap) -import qualified Data.HashMap.Strict as HashMap -import qualified Data.HashSet as HashSet -import qualified Data.IntMap.Strict as I -import Data.List (sort) -import qualified Data.Map.Strict as Map -import Data.Maybe (catMaybes, isNothing, mapMaybe) -import Data.Serialize (Serialize, decode, encode) -import Data.String.Conversions (cs) -import Data.Text (Text) -import Data.Time.Clock.System (getSystemTime, systemSeconds) -import Data.Word (Word32, Word64) -import Database.Redis (Connection, Redis, RedisCtx, Reply, - checkedConnect, defaultConnectInfo, - hgetall, parseConnectInfo, zadd, - zrangeWithscores, - zrangebyscoreWithscoresLimit, zrem) -import qualified Database.Redis as Redis -import GHC.Generics (Generic) -import Haskoin (Address, BlockHash, - BlockHeader (..), BlockNode (..), - DerivPathI (..), KeyIndex, - OutPoint (..), Tx (..), TxHash, - TxIn (..), TxOut (..), XPubKey, - blockHashToHex, derivePubPath, - eitherToMaybe, headerHash, - pathToList, scriptToAddressBS, - txHash, xPubAddr, - xPubCompatWitnessAddr, xPubExport, - xPubWitnessAddr) -import Haskoin.Node (Chain, chainBlockMain, - chainGetAncestor, chainGetBest, - chainGetBlock) +import Control.DeepSeq (NFData) +import Control.Monad (forM, forM_, forever, guard, + unless, void, when) +import Control.Monad.Logger (MonadLoggerIO, logDebugS, + logErrorS, logInfoS, logWarnS) +import Control.Monad.Reader (ReaderT (..), ask, asks) +import Control.Monad.Trans (lift) +import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) +import Data.Bits (shift, (.&.), (.|.)) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Short as BSS +import Data.Default (def) +import Data.Either (rights) +import Data.HashMap.Strict (HashMap) +import qualified Data.HashMap.Strict as HashMap +import Data.HashSet (HashSet) +import qualified Data.HashSet as HashSet +import qualified Data.IntMap.Strict as I +import Data.List (sort) +import qualified Data.Map.Strict as Map +import Data.Maybe (catMaybes, isNothing, mapMaybe) +import Data.Serialize (Serialize, decode, encode) +import Data.String.Conversions (cs) +import Data.Text (Text) +import Data.Time.Clock (NominalDiffTime, diffUTCTime) +import Data.Time.Clock.System (getSystemTime, systemSeconds, + systemToUTCTime) +import Data.Word (Word32, Word64) +import Database.Redis (Connection, Redis, RedisCtx, + Reply, checkedConnect, + defaultConnectInfo, hgetall, + parseConnectInfo, zadd, + zrangeWithscores, + zrangebyscoreWithscoresLimit, + zrem) +import qualified Database.Redis as Redis +import GHC.Generics (Generic) +import Haskoin (Address, BlockHash, + BlockHeader (..), BlockNode (..), + DerivPathI (..), KeyIndex, + OutPoint (..), Tx (..), TxHash, + TxIn (..), TxOut (..), XPubKey, + blockHashToHex, derivePubPath, + eitherToMaybe, headerHash, + pathToList, scriptToAddressBS, + txHash, xPubAddr, + xPubCompatWitnessAddr, xPubExport, + xPubWitnessAddr) +import Haskoin.Node (Chain, chainBlockMain, + chainGetAncestor, chainGetBest, + chainGetBlock) import Haskoin.Store.Common import Haskoin.Store.Data -import NQE (Inbox, Listen, Mailbox, - inboxToMailbox, query, receive, - send) -import System.Random (randomIO, randomRIO) -import UnliftIO (Exception, MonadIO, MonadUnliftIO, - async, atomically, bracket, liftIO, - link, throwIO, withAsync) -import UnliftIO.Concurrent (threadDelay) +import Haskoin.Store.Stats +import NQE (Inbox, Listen, Mailbox, + inboxToMailbox, query, receive, + send) +import qualified System.Metrics as Metrics +import qualified System.Metrics.Counter as Metrics (Counter) +import qualified System.Metrics.Counter as Metrics.Counter +import qualified System.Metrics.Distribution as Metrics (Distribution) +import qualified System.Metrics.Distribution as Metrics.Distribution +import qualified System.Metrics.Gauge as Metrics (Gauge) +import qualified System.Metrics.Gauge as Metrics.Gauge +import System.Random (randomIO, randomRIO) +import UnliftIO (Exception, MonadIO, MonadUnliftIO, + TQueue, TVar, async, atomically, + bracket, liftIO, link, readTQueue, + readTVar, throwIO, withAsync, + writeTQueue, writeTVar) +import UnliftIO.Concurrent (threadDelay) runRedis :: MonadLoggerIO m => Redis (Either Reply a) -> CacheX m a runRedis action = asks cacheConn >>= \conn -> liftIO (Redis.runRedis conn action) >>= \case - Right x -> return x - Left e -> do - $(logErrorS) "Cache" $ "Got error from Redis: " <> cs (show e) - throwIO (RedisError e) + Right x -> return x + Left e -> do + $(logErrorS) "Cache" $ "Got error from Redis: " <> cs (show e) + throwIO (RedisError e) data CacheConfig = CacheConfig { cacheConn :: !Connection @@ -96,10 +114,56 @@ , cacheMax :: !Integer , cacheChain :: !Chain , cacheRefresh :: !Int -- millisenconds - , cacheRetries :: !Int , cacheRetryDelay :: !Int -- microseconds + , cacheMetrics :: !(Maybe CacheMetrics) } +data CacheMetrics = CacheMetrics + { cacheHits :: !Metrics.Counter + , cacheMisses :: !Metrics.Counter + , cacheIgnore :: !Metrics.Counter + , cacheRefreshes :: !Metrics.Counter + , cacheIndexTime :: !StatDist + } + +newCacheMetrics :: MonadIO m => Metrics.Store -> m CacheMetrics +newCacheMetrics s = liftIO $ do + cacheHits <- c "hits" + cacheMisses <- c "misses" + cacheIgnore <- c "ignore" + cacheRefreshes <- c "refreshes" + cacheIndexTime <- d "index_time_ms" + return CacheMetrics{..} + where + c x = Metrics.createCounter ("cache." <> x) s + d x = createStatDist ("cache." <> x) s + +withMetrics :: MonadUnliftIO m + => (CacheMetrics -> StatDist) + -> CacheX m a + -> CacheX m a +withMetrics df go = + asks cacheMetrics >>= \case + Nothing -> go + Just m -> + bracket + (systemToUTCTime <$> liftIO getSystemTime) + (end m) + (const go) + where + end metrics t1 = do + t2 <- systemToUTCTime <$> liftIO getSystemTime + let diff = round $ diffUTCTime t2 t1 * 1000 + df metrics `addStatEntry` StatEntry diff 1 + +incrementCounter :: MonadIO m + => (CacheMetrics -> Metrics.Counter) + -> CacheX m () +incrementCounter cf = + asks cacheMetrics >>= \case + Nothing -> return () + Just m -> liftIO $ Metrics.Counter.inc (cf m) + type CacheT = ReaderT (Maybe CacheConfig) type CacheX = ReaderT CacheConfig @@ -139,15 +203,15 @@ getAddressesUnspents addrs = lift . getAddressesUnspents addrs xPubBals xpub = ask >>= \case - Nothing -> lift (xPubBals xpub) + Nothing -> lift (xPubBals xpub) Just cfg -> lift (runReaderT (getXPubBalances xpub) cfg) xPubUnspents xpub limits = ask >>= \case - Nothing -> lift (xPubUnspents xpub limits) + Nothing -> lift (xPubUnspents xpub limits) Just cfg -> lift (runReaderT (getXPubUnspents xpub limits) cfg) xPubTxs xpub limits = ask >>= \case - Nothing -> lift (xPubTxs xpub limits) + Nothing -> lift (xPubTxs xpub limits) Just cfg -> lift (runReaderT (getXPubTxs xpub limits) cfg) getMaxGap = lift getMaxGap getInitialGap = lift getInitialGap @@ -171,6 +235,7 @@ xpubtxt <- xpubText xpub isXPubCached xpub >>= \case True -> do + incrementCounter cacheHits txs <- cacheGetXPubTxs xpub limits return txs False -> do @@ -185,6 +250,7 @@ xpubtxt <- xpubText xpub isXPubCached xpub >>= \case True -> do + incrementCounter cacheHits bals <- cacheGetXPubBalances xpub process bals False -> do @@ -221,8 +287,8 @@ xpubtxt <- xpubText xpub isXPubCached xpub >>= \case True -> do - bals <- cacheGetXPubBalances xpub - return bals + incrementCounter cacheHits + cacheGetXPubBalances xpub False -> do bals <- lift $ xPubBals xpub newXPubC xpub bals @@ -231,7 +297,7 @@ isInCache :: MonadLoggerIO m => XPubSpec -> CacheT m Bool isInCache xpub = ask >>= \case - Nothing -> return False + Nothing -> return False Just cfg -> runReaderT (isXPubCached xpub) cfg isXPubCached :: MonadLoggerIO m => XPubSpec -> CacheX m Bool @@ -383,7 +449,7 @@ data CacheWriterMessage = CacheNewBlock - | CachePing (Listen ()) + | CachePing !(Listen ()) type CacheWriterInbox = Inbox CacheWriterMessage type CacheWriter = Mailbox CacheWriterMessage @@ -489,7 +555,7 @@ -> CacheX m (Maybe a) withLock f = bracket lockIt unlockIt $ \case - Just _ -> Just <$> f + Just _ -> Just <$> f Nothing -> return Nothing smallDelay :: MonadUnliftIO m => CacheX m () @@ -561,7 +627,9 @@ inSync >>= \s -> when s newBlockC cacheWriterReact (CachePing respond) = do s <- inSync - when s $ syncMempoolC >> void pruneDB + when s $ do + syncMempoolC + void pruneDB atomically (respond ()) lenNotNull :: [XPubBal] -> Int @@ -571,10 +639,12 @@ (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => XPubSpec -> [XPubBal] -> CacheX m () newXPubC xpub bals = - should_index >>= \i -> when i $ - asks cacheRetries >>= \r -> - void . async . withLockRetry r $ - isXPubCached xpub >>= \c -> when (not c) index + should_index >>= \i -> + if i + then do + incrementCounter cacheMisses + withMetrics cacheIndexTime index + else incrementCounter cacheIgnore where op XPubUnspent {xPubUnspent = u} = (unspentPoint u, unspentBlock u) should_index = do @@ -583,7 +653,6 @@ then inSync else return False index = do - bals <- lift (xPubBals xpub) xpubtxt <- xpubText xpub $(logDebugS) "Cache" $ "Caching " <> xpubtxt <> ": " <> cs (show (length bals)) <> @@ -964,13 +1033,14 @@ syncMempoolC :: (MonadUnliftIO m, MonadLoggerIO m, StoreReadExtra m) => CacheX m () -syncMempoolC = do - refresh <- toInteger <$> asks cacheRefresh +syncMempoolC = + toInteger <$> asks cacheRefresh >>= \refresh -> void . withLockForever . withCool "cool" (refresh * 9 `div` 10) $ do - nodepool <- HashSet.fromList . map snd <$> lift getMempool - cachepool <- HashSet.fromList . map snd <$> cacheGetMempool - getem (HashSet.difference nodepool cachepool) - getem (HashSet.difference cachepool nodepool) + nodepool <- HashSet.fromList . map snd <$> lift getMempool + cachepool <- HashSet.fromList . map snd <$> cacheGetMempool + getem (HashSet.difference nodepool cachepool) + getem (HashSet.difference cachepool nodepool) + incrementCounter cacheRefreshes where getem tset = do let tids = HashSet.toList tset @@ -1017,7 +1087,7 @@ => [XPubSpec] -> CacheT m () evictFromCache xpubs = ask >>= \case - Nothing -> return () + Nothing -> return () Just cfg -> void (runReaderT (withLockRetry 100 (delXPubKeys xpubs)) cfg) delXPubKeys :: diff --git a/src/Haskoin/Store/Manager.hs b/src/Haskoin/Store/Manager.hs --- a/src/Haskoin/Store/Manager.hs +++ b/src/Haskoin/Store/Manager.hs @@ -36,7 +36,7 @@ blockStoreTxSTM, withBlockStore) import Haskoin.Store.Cache (CacheConfig (..), CacheWriter, cacheNewBlock, cacheWriter, - connectRedis) + connectRedis, newCacheMetrics) import Haskoin.Store.Common (StoreEvent (..)) import Haskoin.Store.Database.Reader (DatabaseReader (..), DatabaseReaderT, @@ -46,6 +46,7 @@ publishSTM, receive, withProcess, withPublisher, withSubscription) +import qualified System.Metrics as Metrics (Store) import UnliftIO (MonadIO, MonadUnliftIO, STM, atomically, link, withAsync) import UnliftIO.Concurrent (threadDelay) @@ -97,10 +98,10 @@ -- ^ connect to peers using the function 'withConnection' , storeConfCacheRefresh :: !Int -- ^ refresh the cache this often (milliseconds) - , storeConfCacheRetries :: !Int - -- ^ retry count for getting cache lock to index xpub , storeConfCacheRetryDelay :: !Int -- ^ delay in microseconds to retry getting cache lock + , storeConfStats :: !(Maybe Metrics.Store) + -- ^ stats store } withStore :: (MonadLoggerIO m, MonadUnliftIO m) @@ -146,6 +147,7 @@ , blockConfNoMempool = storeConfNoMempool cfg , blockConfWipeMempool = storeConfWipeMempool cfg , blockConfPeerTimeout = storeConfPeerTimeout cfg + , blockConfStats = storeConfStats cfg } nodeCfg :: StoreConfig @@ -182,24 +184,25 @@ Nothing -> action Nothing Just redisurl -> + mapM newCacheMetrics (storeConfStats cfg) >>= \metrics -> connectRedis redisurl >>= \conn -> withSubscription pub $ \evts -> - withProcess (f conn) $ \p -> + withProcess (f conn metrics) $ \p -> cacheWriterProcesses crefresh evts (getProcessMailbox p) $ - action (Just (c conn)) + action (Just (c conn metrics)) where crefresh = storeConfCacheRefresh cfg - f conn cwinbox = - runReaderT (cacheWriter (c conn) cwinbox) db - c conn = + f conn metrics cwinbox = + runReaderT (cacheWriter (c conn metrics) cwinbox) db + c conn metrics = CacheConfig { cacheConn = conn , cacheMin = storeConfCacheMin cfg , cacheChain = chain , cacheMax = storeConfMaxKeys cfg , cacheRefresh = storeConfCacheRefresh cfg - , cacheRetries = storeConfCacheRetries cfg , cacheRetryDelay = storeConfCacheRetryDelay cfg + , cacheMetrics = metrics } cacheWriterProcesses :: MonadUnliftIO m diff --git a/src/Haskoin/Store/Stats.hs b/src/Haskoin/Store/Stats.hs new file mode 100644 --- /dev/null +++ b/src/Haskoin/Store/Stats.hs @@ -0,0 +1,141 @@ +{-# LANGUAGE OverloadedStrings #-} +module Haskoin.Store.Stats + ( StatDist + , StatEntry(..) + , withStats + , createStatDist + , addStatEntry + ) where + +import Control.Concurrent.STM.TQueue (TQueue, flushTQueue, + writeTQueue) +import qualified Control.Foldl as L +import Control.Monad (forever) +import Data.Function (on) +import Data.HashMap.Strict (HashMap) +import qualified Data.HashMap.Strict as HashMap +import Data.Int (Int64) +import Data.List (sort, sortBy) +import Data.Maybe (fromMaybe) +import Data.Ord (Down (..), comparing) +import Data.String.Conversions (cs) +import Data.Text (Text) +import System.Metrics (Store, Value (..), newStore, + registerGcMetrics, + registerGroup, sampleAll) +import System.Remote.Monitoring.Statsd (defaultStatsdOptions, + flushInterval, forkStatsd, + host, port, prefix) +import UnliftIO (MonadIO, atomically, liftIO, + newTQueueIO, withAsync) +import UnliftIO.Concurrent (threadDelay) + +withStats :: MonadIO m => Text -> Int -> Text -> (Store -> m a) -> m a +withStats h p pfx go = do + store <- liftIO newStore + _statsd <- liftIO $ + forkStatsd + defaultStatsdOptions + { prefix = pfx + , host = h + , port = p + } store + liftIO $ registerGcMetrics store + go store + +data StatEntry = StatEntry + { statValue :: !Int64 + , statCount :: !Int64 + } +type StatDist = TQueue StatEntry + +createStatDist :: MonadIO m => Text -> Store -> m StatDist +createStatDist t store = liftIO $ do + q <- newTQueueIO + let metrics = HashMap.fromList + [ (t <> ".query_count", Gauge . fromIntegral . length) + , (t <> ".item_count", Gauge . sum . map count) + , (t <> ".per_query.mean", Gauge . mean . map value) + , (t <> ".per_query.avg", Gauge . avg . map value) + , (t <> ".per_query.max", Gauge . maxi . map value) + , (t <> ".per_query.min", Gauge . mini . map value) + , (t <> ".per_query.p90max", Gauge . p90max . map value) + , (t <> ".per_query.p90min", Gauge . p90min . map value) + , (t <> ".per_query.p90avg", Gauge . p90avg . map value) + , (t <> ".per_query.var", Gauge . var . map value) + , (t <> ".per_item.mean", Gauge . mean . normalize) + , (t <> ".per_item.avg", Gauge . avg . normalize) + , (t <> ".per_item.max", Gauge . maxi . normalize) + , (t <> ".per_item.min", Gauge . mini . normalize) + , (t <> ".per_item.p90max", Gauge . p90max . normalize) + , (t <> ".per_item.p90min", Gauge . p90min . normalize) + , (t <> ".per_item.p90avg", Gauge . p90avg . normalize) + , (t <> ".per_item.var", Gauge . var . normalize) + ] + registerGroup metrics (flush q) store + return q + where + count = statCount + value = statValue + +toDouble :: Int64 -> Double +toDouble = fromIntegral + +addStatEntry :: MonadIO m => StatDist -> StatEntry -> m () +addStatEntry q = liftIO . atomically . writeTQueue q + +flush :: MonadIO m => StatDist -> m [StatEntry] +flush = atomically . flushTQueue + +average :: Fractional a => L.Fold a a +average = (/) <$> L.sum <*> L.genericLength + +avg :: [Int64] -> Int64 +avg = round . L.fold average . map toDouble + +mean :: [Int64] -> Int64 +mean = round . L.fold L.mean . map toDouble + +maxi :: [Int64] -> Int64 +maxi = fromMaybe 0 . L.fold L.maximum + +mini :: [Int64] -> Int64 +mini = fromMaybe 0 . L.fold L.minimum + +var :: [Int64] -> Int64 +var = round . L.fold L.variance . map toDouble + +p90max :: [Int64] -> Int64 +p90max ls = + case chopped of + [] -> 0 + h:_ -> h + where + sorted = sortBy (comparing Down) ls + len = length sorted + chopped = drop (length sorted * 1 `div` 10) sorted + +p90min :: [Int64] -> Int64 +p90min ls = + case chopped of + [] -> 0 + h:_ -> h + where + sorted = sort ls + len = length sorted + chopped = drop (length sorted * 1 `div` 10) sorted + +p90avg :: [Int64] -> Int64 +p90avg ls = + avg chopped + where + sorted = sortBy (comparing Down) ls + len = length sorted + chopped = drop (length sorted * 1 `div` 10) sorted + +normalize :: [StatEntry] -> [Int64] +normalize = + concatMap $ + \(StatEntry x i) -> + replicate (fromIntegral i) (round (toDouble x / toDouble i)) + diff --git a/src/Haskoin/Store/Web.hs b/src/Haskoin/Store/Web.hs --- a/src/Haskoin/Store/Web.hs +++ b/src/Haskoin/Store/Web.hs @@ -28,6 +28,7 @@ import Control.Monad.Reader (ReaderT, ask, asks, local, runReaderT) import Control.Monad.Trans (lift) +import Control.Monad.Trans.Control (liftWith, restoreT) import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) import Data.Aeson (Encoding, ToJSON (..), Value) import Data.Aeson.Encode.Pretty (Config (..), defConfig, @@ -43,6 +44,7 @@ import Data.HashMap.Strict (HashMap) import qualified Data.HashMap.Strict as HashMap import qualified Data.HashSet as HashSet +import Data.Int (Int64) import Data.List (nub) import qualified Data.Map.Strict as Map import Data.Maybe (catMaybes, fromJust, fromMaybe, @@ -57,9 +59,9 @@ import qualified Data.Text.Encoding as T import Data.Text.Lazy (toStrict) import qualified Data.Text.Lazy as TL -import Data.Time.Clock (NominalDiffTime, diffUTCTime, - getCurrentTime) -import Data.Time.Clock.System (getSystemTime, systemSeconds) +import Data.Time.Clock (NominalDiffTime, diffUTCTime) +import Data.Time.Clock.System (getSystemTime, systemSeconds, + systemToUTCTime) import Data.Word (Word32, Word64) import Database.RocksDB (Property (..), getProperty) import Haskoin.Address @@ -76,31 +78,39 @@ import Haskoin.Store.Data import Haskoin.Store.Database.Reader import Haskoin.Store.Manager +import Haskoin.Store.Stats import Haskoin.Store.WebCommon import Haskoin.Transaction import Haskoin.Util +import NQE (Inbox, receive, + withSubscription) import Network.HTTP.Types (Status (..), status400, status403, status404, status500, - status503, statusIsSuccessful) + status503, statusIsClientError, + statusIsServerError, + statusIsSuccessful) import Network.Wai (Middleware, Request (..), responseStatus) import Network.Wai.Handler.Warp (defaultSettings, setHost, setPort) import qualified Network.Wreq as Wreq -import NQE (Inbox, receive, - withSubscription) +import qualified System.Metrics as Metrics +import qualified System.Metrics.Counter as Metrics (Counter) +import qualified System.Metrics.Counter as Metrics.Counter +import qualified System.Metrics.Gauge as Metrics (Gauge) +import qualified System.Metrics.Gauge as Metrics.Gauge import Text.Printf (printf) import UnliftIO (MonadIO, MonadUnliftIO, TVar, - askRunInIO, atomically, - handleAny, liftIO, newTVarIO, - readTVarIO, timeout, withAsync, - writeTVar) + askRunInIO, atomically, bracket, + bracket_, handleAny, liftIO, + newTVarIO, readTVarIO, timeout, + withAsync, writeTVar) import UnliftIO.Concurrent (threadDelay) import Web.Scotty.Internal.Types (ActionT) import Web.Scotty.Trans (Parsable) import qualified Web.Scotty.Trans as S -type WebT m = ActionT Except (ReaderT WebConfig m) +type WebT m = ActionT Except (ReaderT WebState m) data WebLimits = WebLimits { maxLimitCount :: !Word32 @@ -134,8 +144,139 @@ , webVersion :: !String , webNoMempool :: !Bool , webNumTxId :: !Bool + , webStats :: !(Maybe Metrics.Store) } +data WebState = WebState + { webConfig :: !WebConfig + , webTicker :: !(TVar (HashMap Text BinfoTicker)) + , webMetrics :: !(Maybe WebMetrics) + } + +data ErrorCounter = ErrorCounter + { clientErrors :: Metrics.Counter + , serverErrors :: Metrics.Counter + } + +createErrorCounter :: MonadIO m + => Text + -> Metrics.Store + -> m ErrorCounter +createErrorCounter t s = liftIO $ do + clientErrors <- Metrics.createCounter (t <> ".client_errors") s + serverErrors <- Metrics.createCounter (t <> ".server_errors") s + return ErrorCounter{..} + +data WebMetrics = WebMetrics + { everyResponseTime :: !StatDist + , everyError :: !ErrorCounter + , blockResponseTime :: !StatDist + , rawBlockResponseTime :: !StatDist + , blockErrors :: !ErrorCounter + , txResponseTime :: !StatDist + , txsBlockResponseTime :: !StatDist + , txErrors :: !ErrorCounter + , txAfterResponseTime :: !StatDist + , postTxResponseTime :: !StatDist + , postTxErrors :: !ErrorCounter + , mempoolResponseTime :: !StatDist + , addrTxResponseTime :: !StatDist + , addrTxFullResponseTime :: !StatDist + , addrBalanceResponseTime :: !StatDist + , addrUnspentResponseTime :: !StatDist + , xPubResponseTime :: !StatDist + , xPubTxResponseTime :: !StatDist + , xPubTxFullResponseTime :: !StatDist + , xPubUnspentResponseTime :: !StatDist + , xPubEvictResponseTime :: !StatDist + , multiaddrResponseTime :: !StatDist + , multiaddrErrors :: !ErrorCounter + , rawtxResponseTime :: !StatDist + , rawtxErrors :: !ErrorCounter + , peerResponseTime :: !StatDist + , healthResponseTime :: !StatDist + , dbStatsResponseTime :: !StatDist + , eventsConnected :: !Metrics.Gauge + } + +createMetrics :: MonadIO m => Metrics.Store -> m WebMetrics +createMetrics s = liftIO $ do + everyResponseTime <- d "all.response_time_ms" + everyError <- e "all.errors" + blockResponseTime <- d "block.response_time_ms" + blockErrors <- e "block.errors" + rawBlockResponseTime <- d "block_raw.response_time_ms" + txResponseTime <- d "tx.response_time" + txsBlockResponseTime <- d "txs_block.response_time" + txErrors <- e "tx.errors" + txAfterResponseTime <- d "tx_after.response_time_ms" + postTxResponseTime <- d "tx_post.response_time_ms" + postTxErrors <- e "tx_post.errors" + mempoolResponseTime <- d "mempool.response_time_ms" + addrBalanceResponseTime <- d "addr_balance.response_time_ms" + addrTxResponseTime <- d "addr_tx.response_time_ms" + addrTxFullResponseTime <- d "addr_tx_full.response_time_ms" + addrUnspentResponseTime <- d "addr_unspent.response_time_ms" + xPubResponseTime <- d "xpub_balance.response_time_ms" + xPubTxResponseTime <- d "xpub_tx.response_time_ms" + xPubTxFullResponseTime <- d "xpub_tx_full.response_time_ms" + xPubUnspentResponseTime <- d "xpub_unspent.response_time_ms" + xPubEvictResponseTime <- d "xpub_evict.response_time_ms" + multiaddrResponseTime <- d "multiaddr.response_time_ms" + multiaddrErrors <- e "multiaddr.errors" + rawtxResponseTime <- d "rawtx.response_time_ms" + rawtxErrors <- e "rawtx.errors" + peerResponseTime <- d "peer.response_time_ms" + healthResponseTime <- d "health.response_time_ms" + dbStatsResponseTime <- d "dbstats.response_time_ms" + eventsConnected <- g "events.connected" + return WebMetrics{..} + where + d x = createStatDist ("web." <> x) s + g x = Metrics.createGauge ("web." <> x) s + e x = createErrorCounter ("web." <> x) s + +withGaugeIncrease :: MonadUnliftIO m + => (WebMetrics -> Metrics.Gauge) + -> WebT m a + -> WebT m a +withGaugeIncrease gf go = + lift (asks webMetrics) >>= \case + Nothing -> go + Just m -> do + s <- liftWith $ \run -> + bracket_ + (start m) + (end m) + (run go) + restoreT $ return s + where + start m = liftIO $ Metrics.Gauge.inc (gf m) + end m = liftIO $ Metrics.Gauge.dec (gf m) + +withMetrics :: MonadUnliftIO m + => (WebMetrics -> StatDist) + -> Int + -> WebT m a + -> WebT m a +withMetrics df i go = + lift (asks webMetrics) >>= \case + Nothing -> go + Just m -> do + x <- + liftWith $ \run -> + bracket + (systemToUTCTime <$> liftIO getSystemTime) + (end m) + (const (run go)) + restoreT $ return x + where + end metrics t1 = do + t2 <- systemToUTCTime <$> liftIO getSystemTime + let diff = round $ diffUTCTime t2 t1 * 1000 + df metrics `addStatEntry` StatEntry diff (fromIntegral i) + everyResponseTime metrics `addStatEntry` StatEntry diff (fromIntegral i) + data WebTimeouts = WebTimeouts { txTimeout :: !Word64 , blockTimeout :: !Word64 @@ -149,7 +290,7 @@ def = WebTimeouts {txTimeout = 300, blockTimeout = 7200} instance (MonadUnliftIO m, MonadLoggerIO m) => - StoreReadBase (ReaderT WebConfig m) where + StoreReadBase (ReaderT WebState m) where getNetwork = runInWebReader getNetwork getBestBlock = runInWebReader getBestBlock getBlocksAtHeight height = runInWebReader (getBlocksAtHeight height) @@ -161,7 +302,7 @@ getMempool = runInWebReader getMempool instance (MonadUnliftIO m, MonadLoggerIO m) => - StoreReadExtra (ReaderT WebConfig m) where + StoreReadExtra (ReaderT WebState m) where getMaxGap = runInWebReader getMaxGap getInitialGap = runInWebReader getInitialGap getBalances as = runInWebReader (getBalances as) @@ -201,15 +342,19 @@ runWeb :: (MonadUnliftIO m, MonadLoggerIO m) => WebConfig -> m () runWeb cfg@WebConfig{ webHost = host , webPort = port - , webStore = store } = do + , webStore = store + , webStats = stats + } = do ticker <- newTVarIO HashMap.empty + metrics <- mapM createMetrics stats + let st = WebState { webConfig = cfg, webTicker = ticker, webMetrics = metrics } withAsync (price (storeNetwork store) ticker) $ \_ -> do reqLogger <- logIt runner <- askRunInIO - S.scottyOptsT opts (runner . (`runReaderT` cfg)) $ do + S.scottyOptsT opts (runner . (`runReaderT` st)) $ do S.middleware reqLogger S.defaultHandler defHandler - handlePaths ticker + handlePaths S.notFound $ S.raise ThingNotFound where opts = def {S.settings = settings defaultSettings} @@ -236,24 +381,53 @@ atomically . writeTVar v $ r ^. Wreq.responseBody threadDelay $ 5 * 60 * 1000 * 1000 -- five minutes +raise_ :: MonadIO m => Except -> WebT m a +raise_ err = + lift (asks webMetrics) >>= \case + Nothing -> S.raise err + Just metrics -> do + let status = errStatus err + if | statusIsClientError status -> liftIO $ + Metrics.Counter.inc (clientErrors (everyError metrics)) + | statusIsServerError status -> liftIO $ + Metrics.Counter.inc (serverErrors (everyError metrics)) + | otherwise -> + return () + S.raise err +raise :: MonadIO m => (WebMetrics -> ErrorCounter) -> Except -> WebT m a +raise metric err = + lift (asks webMetrics) >>= \case + Nothing -> S.raise err + Just metrics -> do + let status = errStatus err + if | statusIsClientError status -> liftIO $ do + Metrics.Counter.inc (clientErrors (everyError metrics)) + Metrics.Counter.inc (clientErrors (metric metrics)) + | statusIsServerError status -> liftIO $ do + Metrics.Counter.inc (serverErrors (everyError metrics)) + Metrics.Counter.inc (serverErrors (metric metrics)) + | otherwise -> + return () + S.raise err + +errStatus :: Except -> Status +errStatus ThingNotFound = status404 +errStatus BadRequest = status400 +errStatus UserError{} = status400 +errStatus StringError{} = status400 +errStatus ServerError = status500 +errStatus BlockTooLarge = status403 + defHandler :: Monad m => Except -> WebT m () defHandler e = do proto <- setupContentType False - case e of - ThingNotFound -> S.status status404 - BadRequest -> S.status status400 - UserError _ -> S.status status400 - StringError _ -> S.status status400 - ServerError -> S.status status500 - BlockTooLarge -> S.status status403 + S.status $ errStatus e S.raw $ protoSerial proto toEncoding toJSON e -handlePaths :: - (MonadUnliftIO m, MonadLoggerIO m) - => TVar (HashMap Text BinfoTicker) - -> S.ScottyT Except (ReaderT WebConfig m) () -handlePaths ticker = do +handlePaths :: (MonadUnliftIO m, MonadLoggerIO m) + => S.ScottyT Except (ReaderT WebState m) () +handlePaths = do -- Block Paths pathPretty (GetBlock <$> paramLazy <*> paramDef) @@ -452,7 +626,7 @@ S.get "/events" scottyEvents S.get "/dbstats" scottyDbStats -- Blockchain.info - S.post "/blockchain/multiaddr" (scottyMultiAddr ticker) + S.post "/blockchain/multiaddr" scottyMultiAddr S.get "/blockchain/rawtx/:txid" scottyBinfoTx where json_list f net = toJSONList . map (f net) @@ -463,7 +637,7 @@ -> (a -> WebT m b) -> (Network -> b -> Encoding) -> (Network -> b -> Value) - -> S.ScottyT Except (ReaderT WebConfig m) () + -> S.ScottyT Except (ReaderT WebState m) () pathPretty parser action encJson encValue = pathCommon parser action encJson encValue True @@ -473,7 +647,7 @@ -> (a -> WebT m b) -> (Network -> b -> Encoding) -> (Network -> b -> Value) - -> S.ScottyT Except (ReaderT WebConfig m) () + -> S.ScottyT Except (ReaderT WebState m) () pathCompact parser action encJson encValue = pathCommon parser action encJson encValue False @@ -484,12 +658,12 @@ -> (Network -> b -> Encoding) -> (Network -> b -> Value) -> Bool - -> S.ScottyT Except (ReaderT WebConfig m) () + -> S.ScottyT Except (ReaderT WebState m) () pathCommon parser action encJson encValue pretty = S.addroute (resourceMethod proxy) (capturePath proxy) $ do setHeaders proto <- setupContentType pretty - net <- lift $ asks (storeNetwork . webStore) + net <- lift $ asks (storeNetwork . webStore . webConfig) apiRes <- parser res <- action apiRes S.raw $ protoSerial proto (encJson net) (encValue net) res @@ -533,12 +707,21 @@ scottyBlock :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlock -> WebT m BlockData scottyBlock (GetBlock h (NoTx noTx)) = - maybe (S.raise ThingNotFound) (return . pruneTx noTx) =<< getBlock h + withMetrics blockResponseTime 1 $ + maybe (raise blockErrors ThingNotFound) (return . pruneTx noTx) =<< + getBlock h +getBlocks :: (MonadUnliftIO m, MonadLoggerIO m) + => [H.BlockHash] + -> Bool + -> WebT m [BlockData] +getBlocks hs notx = + (pruneTx notx <$>) . catMaybes <$> mapM getBlock (nub hs) + scottyBlocks :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlocks -> WebT m [BlockData] -scottyBlocks (GetBlocks hs (NoTx noTx)) = - (pruneTx noTx <$>) . catMaybes <$> mapM getBlock (nub hs) +scottyBlocks (GetBlocks hs (NoTx notx)) = + withMetrics blockResponseTime (length hs) $ getBlocks hs notx pruneTx :: Bool -> BlockData -> BlockData pruneTx False b = b @@ -550,13 +733,16 @@ (MonadUnliftIO m, MonadLoggerIO m) => GetBlockRaw -> WebT m (RawResult H.Block) -scottyBlockRaw (GetBlockRaw h) = RawResult <$> getRawBlock h +scottyBlockRaw (GetBlockRaw h) = + withMetrics rawBlockResponseTime 1 $ + RawResult <$> getRawBlock h getRawBlock :: (MonadUnliftIO m, MonadLoggerIO m) => H.BlockHash -> WebT m H.Block -getRawBlock h = do - b <- maybe (S.raise ThingNotFound) return =<< getBlock h - refuseLargeBlock b +getRawBlock h = + withMetrics rawBlockResponseTime 1 $ do + b <- maybe (raise blockErrors ThingNotFound) return =<< getBlock h + refuseLargeBlock blockErrors b toRawBlock b toRawBlock :: (Monad m, StoreReadBase m) => BlockData -> m H.Block @@ -565,25 +751,34 @@ txs <- map transactionData . catMaybes <$> mapM getTransaction ths return H.Block {H.blockHeader = blockDataHeader b, H.blockTxns = txs} -refuseLargeBlock :: Monad m => BlockData -> WebT m () -refuseLargeBlock BlockData {blockDataTxs = txs} = do - WebLimits {maxLimitFull = f} <- lift $ asks webMaxLimits - when (length txs > fromIntegral f) $ S.raise BlockTooLarge +refuseLargeBlock :: MonadIO m + => (WebMetrics -> ErrorCounter) + -> BlockData + -> WebT m () +refuseLargeBlock metric BlockData {blockDataTxs = txs} = do + WebLimits {maxLimitFull = f} <- lift $ asks (webMaxLimits . webConfig) + when (length txs > fromIntegral f) $ raise metric BlockTooLarge -- GET BlockBest / BlockBestRaw -- scottyBlockBest :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockBest -> WebT m BlockData -scottyBlockBest (GetBlockBest noTx) = do - bestM <- getBestBlock - maybe (S.raise ThingNotFound) (scottyBlock . (`GetBlock` noTx)) bestM +scottyBlockBest (GetBlockBest (NoTx notx)) = + withMetrics blockResponseTime 1 $ + getBestBlock >>= \case + Nothing -> raise blockErrors ThingNotFound + Just bb -> getBlock bb >>= \case + Nothing -> raise blockErrors ThingNotFound + Just b -> return $ pruneTx notx b scottyBlockBestRaw :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockBestRaw -> WebT m (RawResult H.Block) scottyBlockBestRaw _ = - RawResult <$> (maybe (S.raise ThingNotFound) getRawBlock =<< getBestBlock) + withMetrics rawBlockResponseTime 1 $ + fmap RawResult $ + maybe (raise blockErrors ThingNotFound) getRawBlock =<< getBestBlock -- GET BlockLatest -- @@ -592,7 +787,8 @@ => GetBlockLatest -> WebT m [BlockData] scottyBlockLatest (GetBlockLatest (NoTx noTx)) = - maybe (S.raise ThingNotFound) (go [] <=< getBlock) =<< getBestBlock + withMetrics blockResponseTime 100 $ + maybe (raise blockErrors ThingNotFound) (go [] <=< getBlock) =<< getBestBlock where go acc Nothing = return acc go acc (Just b) @@ -606,95 +802,114 @@ scottyBlockHeight :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockHeight -> WebT m [BlockData] -scottyBlockHeight (GetBlockHeight h noTx) = - scottyBlocks . (`GetBlocks` noTx) =<< getBlocksAtHeight (fromIntegral h) +scottyBlockHeight (GetBlockHeight h (NoTx notx)) = + withMetrics blockResponseTime 1 $ + (`getBlocks` notx) =<< getBlocksAtHeight (fromIntegral h) scottyBlockHeights :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockHeights -> WebT m [BlockData] -scottyBlockHeights (GetBlockHeights (HeightsParam heights) noTx) = do +scottyBlockHeights (GetBlockHeights (HeightsParam heights) (NoTx notx)) = + withMetrics blockResponseTime (length heights) $ do bhs <- concat <$> mapM getBlocksAtHeight (fromIntegral <$> heights) - scottyBlocks (GetBlocks bhs noTx) + getBlocks bhs notx scottyBlockHeightRaw :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockHeightRaw -> WebT m (RawResultList H.Block) scottyBlockHeightRaw (GetBlockHeightRaw h) = + withMetrics rawBlockResponseTime 1 $ RawResultList <$> (mapM getRawBlock =<< getBlocksAtHeight (fromIntegral h)) -- GET BlockTime / BlockTimeRaw -- scottyBlockTime :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockTime -> WebT m BlockData -scottyBlockTime (GetBlockTime (TimeParam t) (NoTx noTx)) = do - ch <- lift $ asks (storeChain . webStore) +scottyBlockTime (GetBlockTime (TimeParam t) (NoTx notx)) = + withMetrics blockResponseTime 1 $ do + ch <- lift $ asks (storeChain . webStore . webConfig) m <- blockAtOrBefore ch t - maybe (S.raise ThingNotFound) (return . pruneTx noTx) m + maybe (raise blockErrors ThingNotFound) (return . pruneTx notx) m scottyBlockMTP :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockMTP -> WebT m BlockData -scottyBlockMTP (GetBlockMTP (TimeParam t) (NoTx noTx)) = do - ch <- lift $ asks (storeChain . webStore) +scottyBlockMTP (GetBlockMTP (TimeParam t) (NoTx noTx)) = + withMetrics blockResponseTime 1 $ do + ch <- lift $ asks (storeChain . webStore . webConfig) m <- blockAtOrAfterMTP ch t - maybe (S.raise ThingNotFound) (return . pruneTx noTx) m + maybe (raise blockErrors ThingNotFound) (return . pruneTx noTx) m scottyBlockTimeRaw :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockTimeRaw -> WebT m (RawResult H.Block) -scottyBlockTimeRaw (GetBlockTimeRaw (TimeParam t)) = do - ch <- lift $ asks (storeChain . webStore) +scottyBlockTimeRaw (GetBlockTimeRaw (TimeParam t)) = + withMetrics rawBlockResponseTime 1 $ do + ch <- lift $ asks (storeChain . webStore . webConfig) m <- blockAtOrBefore ch t - b <- maybe (S.raise ThingNotFound) return m - refuseLargeBlock b + b <- maybe (raise blockErrors ThingNotFound) return m + refuseLargeBlock blockErrors b RawResult <$> toRawBlock b scottyBlockMTPRaw :: (MonadUnliftIO m, MonadLoggerIO m) => GetBlockMTPRaw -> WebT m (RawResult H.Block) -scottyBlockMTPRaw (GetBlockMTPRaw (TimeParam t)) = do - ch <- lift $ asks (storeChain . webStore) +scottyBlockMTPRaw (GetBlockMTPRaw (TimeParam t)) = + withMetrics rawBlockResponseTime 1 $ do + ch <- lift $ asks (storeChain . webStore . webConfig) m <- blockAtOrAfterMTP ch t - b <- maybe (S.raise ThingNotFound) return m - refuseLargeBlock b + b <- maybe (raise blockErrors ThingNotFound) return m + refuseLargeBlock blockErrors b RawResult <$> toRawBlock b -- GET Transactions -- scottyTx :: (MonadUnliftIO m, MonadLoggerIO m) => GetTx -> WebT m Transaction scottyTx (GetTx txid) = - maybe (S.raise ThingNotFound) return =<< getTransaction txid + withMetrics txResponseTime 1 $ + maybe (raise txErrors ThingNotFound) return =<< getTransaction txid scottyTxs :: (MonadUnliftIO m, MonadLoggerIO m) => GetTxs -> WebT m [Transaction] -scottyTxs (GetTxs txids) = catMaybes <$> mapM getTransaction (nub txids) +scottyTxs (GetTxs txids) = + withMetrics txResponseTime (length txids) $ + catMaybes <$> mapM getTransaction (nub txids) scottyTxRaw :: (MonadUnliftIO m, MonadLoggerIO m) => GetTxRaw -> WebT m (RawResult Tx) -scottyTxRaw (GetTxRaw txid) = do - tx <- maybe (S.raise ThingNotFound) return =<< getTransaction txid +scottyTxRaw (GetTxRaw txid) = + withMetrics txResponseTime 1 $ do + tx <- maybe (raise txErrors ThingNotFound) return =<< getTransaction txid return $ RawResult $ transactionData tx scottyTxsRaw :: (MonadUnliftIO m, MonadLoggerIO m) => GetTxsRaw -> WebT m (RawResultList Tx) -scottyTxsRaw (GetTxsRaw txids) = do +scottyTxsRaw (GetTxsRaw txids) = + withMetrics txResponseTime (length txids) $ do txs <- catMaybes <$> mapM getTransaction (nub txids) return $ RawResultList $ transactionData <$> txs +getTxsBlock :: (MonadUnliftIO m, MonadLoggerIO m) + => H.BlockHash + -> WebT m [Transaction] +getTxsBlock h = do + b <- maybe (raise txErrors ThingNotFound) return =<< getBlock h + refuseLargeBlock txErrors b + catMaybes <$> mapM getTransaction (blockDataTxs b) + scottyTxsBlock :: (MonadUnliftIO m, MonadLoggerIO m) => GetTxsBlock -> WebT m [Transaction] -scottyTxsBlock (GetTxsBlock h) = do - b <- maybe (S.raise ThingNotFound) return =<< getBlock h - refuseLargeBlock b - catMaybes <$> mapM getTransaction (blockDataTxs b) +scottyTxsBlock (GetTxsBlock h) = + withMetrics txsBlockResponseTime 1 $ getTxsBlock h scottyTxsBlockRaw :: (MonadUnliftIO m, MonadLoggerIO m) => GetTxsBlockRaw -> WebT m (RawResultList Tx) scottyTxsBlockRaw (GetTxsBlockRaw h) = - RawResultList . fmap transactionData <$> scottyTxsBlock (GetTxsBlock h) + withMetrics txsBlockResponseTime 1 $ + RawResultList . fmap transactionData <$> getTxsBlock h -- GET TransactionAfterHeight -- @@ -703,6 +918,7 @@ => GetTxAfter -> WebT m (GenericResult (Maybe Bool)) scottyTxAfter (GetTxAfter txid height) = + withMetrics txAfterResponseTime 1 $ GenericResult <$> cbAfterHeight (fromIntegral height) txid -- | Check if any of the ancestors of this transaction is a coinbase after the @@ -743,10 +959,11 @@ scottyPostTx :: (MonadUnliftIO m, MonadLoggerIO m) => PostTx -> WebT m TxId scottyPostTx (PostTx tx) = - lift ask >>= \cfg -> lift (publishTx cfg tx) >>= \case - Right () -> return $ TxId (txHash tx) - Left e@(PubReject _) -> S.raise $ UserError $ show e - _ -> S.raise ServerError + withMetrics postTxResponseTime 1 $ + lift (asks webConfig) >>= \cfg -> lift (publishTx cfg tx) >>= \case + Right () -> return $ TxId (txHash tx) + Left e@(PubReject _) -> raise postTxErrors $ UserError $ show e + _ -> raise postTxErrors ServerError -- | Publish a new transaction to the network. publishTx :: @@ -757,7 +974,7 @@ publishTx cfg tx = withSubscription pub $ \s -> getTransaction (txHash tx) >>= \case - Just _ -> return $ Right () + Just _ -> return $ Right () Nothing -> go s where pub = storePublisher (webStore cfg) @@ -781,8 +998,8 @@ | webNoMempool cfg = return $ Right () | otherwise = liftIO (timeout t (g p s)) >>= \case - Nothing -> return $ Left PubTimeout - Just (Left e) -> return $ Left e + Nothing -> return $ Left PubTimeout + Just (Left e) -> return $ Left e Just (Right ()) -> return $ Right () g p s = receive s >>= \case @@ -798,17 +1015,19 @@ scottyMempool :: (MonadUnliftIO m, MonadLoggerIO m) => GetMempool -> WebT m [TxHash] -scottyMempool (GetMempool limitM (OffsetParam o)) = do - wl <- lift $ asks webMaxLimits +scottyMempool (GetMempool limitM (OffsetParam o)) = + withMetrics mempoolResponseTime 1 $ do + wl <- lift $ asks (webMaxLimits . webConfig) let wl' = wl { maxLimitCount = 0 } l = Limits (validateLimit wl' False limitM) (fromIntegral o) Nothing map snd . applyLimits l <$> getMempool -scottyEvents :: MonadLoggerIO m => WebT m () -scottyEvents = do +scottyEvents :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () +scottyEvents = + withGaugeIncrease eventsConnected $ do setHeaders proto <- setupContentType False - pub <- lift $ asks (storePublisher . webStore) + pub <- lift $ asks (storePublisher . webStore . webConfig) S.stream $ \io flush' -> withSubscription pub $ \sub -> forever $ @@ -834,43 +1053,53 @@ scottyAddrTxs :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrTxs -> WebT m [TxRef] scottyAddrTxs (GetAddrTxs addr pLimits) = + withMetrics addrTxResponseTime 1 $ getAddressTxs addr =<< paramToLimits False pLimits scottyAddrsTxs :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrsTxs -> WebT m [TxRef] scottyAddrsTxs (GetAddrsTxs addrs pLimits) = + withMetrics addrTxResponseTime (length addrs) $ getAddressesTxs addrs =<< paramToLimits False pLimits scottyAddrTxsFull :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrTxsFull -> WebT m [Transaction] -scottyAddrTxsFull (GetAddrTxsFull addr pLimits) = do +scottyAddrTxsFull (GetAddrTxsFull addr pLimits) = + withMetrics addrTxFullResponseTime 1 $ do txs <- getAddressTxs addr =<< paramToLimits True pLimits catMaybes <$> mapM (getTransaction . txRefHash) txs scottyAddrsTxsFull :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrsTxsFull -> WebT m [Transaction] -scottyAddrsTxsFull (GetAddrsTxsFull addrs pLimits) = do +scottyAddrsTxsFull (GetAddrsTxsFull addrs pLimits) = + withMetrics addrTxFullResponseTime (length addrs) $ do txs <- getAddressesTxs addrs =<< paramToLimits True pLimits catMaybes <$> mapM (getTransaction . txRefHash) txs scottyAddrBalance :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrBalance -> WebT m Balance -scottyAddrBalance (GetAddrBalance addr) = getDefaultBalance addr +scottyAddrBalance (GetAddrBalance addr) = + withMetrics addrBalanceResponseTime 1 $ + getDefaultBalance addr scottyAddrsBalance :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrsBalance -> WebT m [Balance] -scottyAddrsBalance (GetAddrsBalance addrs) = getBalances addrs +scottyAddrsBalance (GetAddrsBalance addrs) = + withMetrics addrBalanceResponseTime (length addrs) $ + getBalances addrs scottyAddrUnspent :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrUnspent -> WebT m [Unspent] scottyAddrUnspent (GetAddrUnspent addr pLimits) = + withMetrics addrUnspentResponseTime 1 $ getAddressUnspents addr =<< paramToLimits False pLimits scottyAddrsUnspent :: (MonadUnliftIO m, MonadLoggerIO m) => GetAddrsUnspent -> WebT m [Unspent] scottyAddrsUnspent (GetAddrsUnspent addrs pLimits) = + withMetrics addrUnspentResponseTime (length addrs) $ getAddressesUnspents addrs =<< paramToLimits False pLimits -- GET XPubs -- @@ -878,26 +1107,35 @@ scottyXPub :: (MonadUnliftIO m, MonadLoggerIO m) => GetXPub -> WebT m XPubSummary scottyXPub (GetXPub xpub deriv (NoCache noCache)) = + withMetrics xPubResponseTime 1 $ lift . runNoCache noCache $ xPubSummary $ XPubSpec xpub deriv +getXPubTxs :: (MonadUnliftIO m, MonadLoggerIO m) + => XPubKey -> DeriveType -> LimitsParam -> Bool -> WebT m [TxRef] +getXPubTxs xpub deriv plimits nocache = do + limits <- paramToLimits False plimits + lift . runNoCache nocache $ xPubTxs (XPubSpec xpub deriv) limits + scottyXPubTxs :: (MonadUnliftIO m, MonadLoggerIO m) => GetXPubTxs -> WebT m [TxRef] -scottyXPubTxs (GetXPubTxs xpub deriv pLimits (NoCache noCache)) = do - limits <- paramToLimits False pLimits - lift . runNoCache noCache $ xPubTxs (XPubSpec xpub deriv) limits +scottyXPubTxs (GetXPubTxs xpub deriv plimits (NoCache nocache)) = + withMetrics xPubTxResponseTime 1 $ + getXPubTxs xpub deriv plimits nocache scottyXPubTxsFull :: (MonadUnliftIO m, MonadLoggerIO m) => GetXPubTxsFull -> WebT m [Transaction] -scottyXPubTxsFull (GetXPubTxsFull xpub deriv pLimits n@(NoCache noCache)) = do - refs <- scottyXPubTxs (GetXPubTxs xpub deriv pLimits n) - txs <- lift . runNoCache noCache $ mapM (getTransaction . txRefHash) refs +scottyXPubTxsFull (GetXPubTxsFull xpub deriv plimits (NoCache nocache)) = + withMetrics xPubTxFullResponseTime 1 $ do + refs <- getXPubTxs xpub deriv plimits nocache + txs <- lift . runNoCache nocache $ mapM (getTransaction . txRefHash) refs return $ catMaybes txs scottyXPubBalances :: (MonadUnliftIO m, MonadLoggerIO m) => GetXPubBalances -> WebT m [XPubBal] scottyXPubBalances (GetXPubBalances xpub deriv (NoCache noCache)) = + withMetrics xPubResponseTime 1 $ filter f <$> lift (runNoCache noCache (xPubBals spec)) where spec = XPubSpec xpub deriv @@ -907,7 +1145,8 @@ (MonadUnliftIO m, MonadLoggerIO m) => GetXPubUnspent -> WebT m [XPubUnspent] -scottyXPubUnspent (GetXPubUnspent xpub deriv pLimits (NoCache noCache)) = do +scottyXPubUnspent (GetXPubUnspent xpub deriv pLimits (NoCache noCache)) = + withMetrics xPubUnspentResponseTime 1 $ do limits <- paramToLimits False pLimits lift . runNoCache noCache $ xPubUnspents (XPubSpec xpub deriv) limits @@ -915,8 +1154,9 @@ (MonadUnliftIO m, MonadLoggerIO m) => GetXPubEvict -> WebT m (GenericResult Bool) -scottyXPubEvict (GetXPubEvict xpub deriv) = do - cache <- lift $ asks (storeCache . webStore) +scottyXPubEvict (GetXPubEvict xpub deriv) = + withMetrics xPubEvictResponseTime 1 $ do + cache <- lift $ asks (storeCache . webStore . webConfig) lift . withCache cache $ evictFromCache [XPubSpec xpub deriv] return $ GenericResult True @@ -969,20 +1209,22 @@ x -> x scottyMultiAddr :: (MonadUnliftIO m, MonadLoggerIO m) - => TVar (HashMap Text BinfoTicker) - -> WebT m () -scottyMultiAddr ticker = do + => WebT m () +scottyMultiAddr = + get_addrs >>= \(addrs, xpubs, saddrs, sxpubs, xspecs) -> + let len = HashSet.size addrs + HashSet.size xpubs + in withMetrics multiaddrResponseTime len $ do prune <- get_prune n <- get_count offset <- get_offset cashaddr <- get_cashaddr - numtxid <- lift $ asks webNumTxId - (addrs, xpubs, saddrs, sxpubs, xspecs) <- get_addrs + numtxid <- lift $ asks (webNumTxId . webConfig) xbals <- get_xbals xspecs let sxbals = compute_sxbals sxpubs xbals sxtrs <- get_sxtrs sxpubs xspecs sabals <- get_abals saddrs satrs <- get_atrs n offset saddrs + ticker <- lift $ asks webTicker local <- get_price ticker let sxtns = HashMap.map length sxtrs sxtrset = Set.fromList . concat $ HashMap.elems sxtrs @@ -996,14 +1238,14 @@ stxids = compute_txids n offset salltrs stxs <- get_txs stxids let etxids = if numtxid - then compute_etxids prune abook stxs - else [] + then compute_etxids prune abook stxs + else [] etxs <- if numtxid then Just <$> get_etxs etxids else return Nothing - best <- scottyBlockBest (GetBlockBest (NoTx True)) + best <- get_best_block peers <- get_peers - net <- lift $ asks (storeNetwork . webStore) + net <- lift $ asks (storeNetwork . webStore . webConfig) let ibal = fromIntegral bal btxs = binfo_txs etxs abook salladdrs prune ibal stxs ftxs = take n $ drop offset btxs @@ -1046,6 +1288,12 @@ , getBinfoMultiAddrCashAddr = cashaddr } where + get_best_block = + getBestBlock >>= \case + Nothing -> raise multiaddrErrors ThingNotFound + Just bh -> getBlock bh >>= \case + Nothing -> raise multiaddrErrors ThingNotFound + Just b -> return b get_price ticker = do code <- T.toUpper <$> S.param "local" `S.rescue` const (return "USD") prices <- readTVarIO ticker @@ -1057,19 +1305,19 @@ get_cashaddr = S.param "cashaddr" `S.rescue` const (return False) get_count = do - d <- lift (asks (maxLimitDefault . webMaxLimits)) - x <- lift (asks (maxLimitFull . webMaxLimits)) + d <- lift (asks (maxLimitDefault . webMaxLimits . webConfig)) + x <- lift (asks (maxLimitFull . webMaxLimits . webConfig)) i <- min x <$> (S.param "n" `S.rescue` const (return d)) return (fromIntegral i :: Int) get_offset = do - x <- lift (asks (maxLimitOffset . webMaxLimits)) + x <- lift (asks (maxLimitOffset . webMaxLimits . webConfig)) o <- min x <$> (S.param "offset" `S.rescue` const (return 0)) return (fromIntegral o :: Int) get_addrs_param name = do - net <- lift (asks (storeNetwork . webStore)) + net <- lift (asks (storeNetwork . webStore . webConfig)) p <- S.param name `S.rescue` const (return "") case parseBinfoAddr net p of - Nothing -> S.raise (UserError "invalid active address") + Nothing -> raise multiaddrErrors (UserError "invalid active address") Just xs -> return $ HashSet.fromList xs addr (BinfoAddr a) = Just a addr (BinfoXpub x) = Nothing @@ -1125,7 +1373,8 @@ g = HashMap.fromList . map f . catMaybes in fmap g . mapM getTransaction get_peers = do - ps <- lift $ getPeersInformation =<< asks (storeManager . webStore) + ps <- lift $ getPeersInformation =<< + asks (storeManager . webStore . webConfig) return (fromIntegral (length ps)) compute_txids n offset = map txRefHash . take (n + offset) . Set.toDescList compute_etxids prune abook = @@ -1168,12 +1417,13 @@ scottyBinfoTx :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () scottyBinfoTx = - lift (asks webNumTxId) >>= \num -> S.param "txid" >>= \case + withMetrics rawtxResponseTime 1 $ + lift (asks (webNumTxId . webConfig)) >>= \num -> S.param "txid" >>= \case BinfoTxIdHash h -> go num h BinfoTxIdIndex i -> - if | not num || isBinfoTxIndexNull i -> S.raise ThingNotFound - | isBinfoTxIndexBlock i -> block i - | isBinfoTxIndexHash i -> hash i + if | not num || isBinfoTxIndexNull i -> raise rawtxErrors ThingNotFound + | isBinfoTxIndexBlock i -> block i + | isBinfoTxIndexHash i -> hash i where get_format = S.param "format" `S.rescue` const (return ("json" :: Text)) js num t = do @@ -1185,27 +1435,27 @@ else return Nothing let f t = (txHash (transactionData t), t) etxs = HashMap.fromList . map f <$> etxs' - net <- lift $ asks (storeNetwork . webStore) + net <- lift $ asks (storeNetwork . webStore . webConfig) setHeaders S.json . binfoTxToJSON net $ toBinfoTxSimple etxs t hex t = do setHeaders S.text . TL.fromStrict . encodeHex . encode $ transactionData t go num h = getTransaction h >>= \case - Nothing -> S.raise ThingNotFound + Nothing -> raise rawtxErrors ThingNotFound Just t -> get_format >>= \case "hex" -> hex t - _ -> js num t + _ -> js num t block i = let Just (height, pos) = binfoTxIndexBlock i in getBlocksAtHeight height >>= \case - [] -> S.raise ThingNotFound + [] -> raise rawtxErrors ThingNotFound h:_ -> getBlock h >>= \case - Nothing -> S.raise ThingNotFound + Nothing -> raise rawtxErrors ThingNotFound Just BlockData{..} -> if length blockDataTxs > fromIntegral pos then go True (blockDataTxs !! fromIntegral pos) - else S.raise ThingNotFound + else raise rawtxErrors ThingNotFound hmatch h = listToMaybe . filter (matchBinfoTxHash h) hmem h = hmatch h . map snd <$> getMempool hblock h = @@ -1219,18 +1469,22 @@ Just x -> return (Just x) in getBestBlock >>= \case Nothing -> return Nothing - Just k -> g 10 k + Just k -> g 10 k hash h = hmem h >>= \case Nothing -> hblock h >>= \case - Nothing -> S.raise ThingNotFound - Just x -> go True x + Nothing -> raise rawtxErrors ThingNotFound + Just x -> go True x Just x -> go True x -- GET Network Information -- -scottyPeers :: MonadLoggerIO m => GetPeers -> WebT m [PeerInformation] -scottyPeers _ = lift $ - getPeersInformation =<< asks (storeManager . webStore) +scottyPeers :: (MonadUnliftIO m, MonadLoggerIO m) + => GetPeers + -> WebT m [PeerInformation] +scottyPeers _ = + withMetrics peerResponseTime 1 $ + lift $ + getPeersInformation =<< asks (storeManager . webStore . webConfig) -- | Obtain information about connected peers from peer manager process. getPeersInformation @@ -1256,8 +1510,9 @@ scottyHealth :: (MonadUnliftIO m, MonadLoggerIO m) => GetHealth -> WebT m HealthCheck -scottyHealth _ = do - h <- lift $ ask >>= healthCheck +scottyHealth _ = + withMetrics healthResponseTime 1 $ do + h <- lift $ asks webConfig >>= healthCheck unless (isOK h) $ S.status status503 return h @@ -1328,10 +1583,11 @@ $(logErrorS) "Web" $ "Health check failed: " <> t return hc -scottyDbStats :: MonadLoggerIO m => WebT m () -scottyDbStats = do +scottyDbStats :: (MonadUnliftIO m, MonadLoggerIO m) => WebT m () +scottyDbStats = + withMetrics dbStatsResponseTime 1 $ do setHeaders - db <- lift $ asks (databaseHandle . storeDB . webStore) + db <- lift $ asks (databaseHandle . storeDB . webStore . webConfig) statsM <- lift (getProperty db Stats) S.text $ maybe "Could not get stats" cs statsM @@ -1341,42 +1597,42 @@ -- | Returns @Nothing@ if the parameter is not supplied. Raises an exception on -- parse failure. -paramOptional :: (Param a, Monad m) => WebT m (Maybe a) +paramOptional :: (Param a, MonadIO m) => WebT m (Maybe a) paramOptional = go Proxy where - go :: (Param a, Monad m) => Proxy a -> WebT m (Maybe a) + go :: (Param a, MonadIO m) => Proxy a -> WebT m (Maybe a) go proxy = do - net <- lift $ asks (storeNetwork . webStore) + net <- lift $ asks (storeNetwork . webStore . webConfig) tsM :: Maybe [Text] <- p `S.rescue` const (return Nothing) case tsM of Nothing -> return Nothing -- Parameter was not supplied - Just ts -> maybe (S.raise err) (return . Just) $ parseParam net ts + Just ts -> maybe (raise_ err) (return . Just) $ parseParam net ts where l = proxyLabel proxy p = Just <$> S.param (cs l) err = UserError $ "Unable to parse param " <> cs l -- | Raises an exception if the parameter is not supplied -param :: (Param a, Monad m) => WebT m a +param :: (Param a, MonadIO m) => WebT m a param = go Proxy where - go :: (Param a, Monad m) => Proxy a -> WebT m a + go :: (Param a, MonadIO m) => Proxy a -> WebT m a go proxy = do resM <- paramOptional case resM of Just res -> return res _ -> - S.raise . UserError $ + raise_ . UserError $ "The param " <> cs (proxyLabel proxy) <> " was not defined" -- | Returns the default value of a parameter if it is not supplied. Raises an -- exception on parse failure. -paramDef :: (Default a, Param a, Monad m) => WebT m a +paramDef :: (Default a, Param a, MonadIO m) => WebT m a paramDef = fromMaybe def <$> paramOptional -- | Does not raise exceptions. Will call @Scotty.next@ if the parameter is -- not supplied or if parsing fails. -paramLazy :: (Param a, Monad m) => WebT m a +paramLazy :: (Param a, MonadIO m) => WebT m a paramLazy = do resM <- paramOptional `S.rescue` const (return Nothing) maybe S.next return resM @@ -1385,18 +1641,18 @@ parseBody = do b <- S.body case hex b <|> bin (L.toStrict b) of - Nothing -> S.raise $ UserError "Failed to parse request body" + Nothing -> raise_ $ UserError "Failed to parse request body" Just x -> return x where bin = eitherToMaybe . Serialize.decode hex = bin <=< decodeHex . cs . C.filter (not . isSpace) -parseOffset :: Monad m => WebT m OffsetParam +parseOffset :: MonadIO m => WebT m OffsetParam parseOffset = do res@(OffsetParam o) <- paramDef - limits <- lift $ asks webMaxLimits + limits <- lift $ asks (webMaxLimits . webConfig) when (maxLimitOffset limits > 0 && fromIntegral o > maxLimitOffset limits) $ - S.raise . UserError $ + raise_ . UserError $ "offset exceeded: " <> show o <> " > " <> show (maxLimitOffset limits) return res @@ -1420,12 +1676,12 @@ _ <- MaybeT $ getTxData (TxHash h) return $ AtTx (TxHash h) start_time q = do - ch <- lift $ asks (storeChain . webStore) + ch <- lift $ asks (storeChain . webStore . webConfig) b <- MaybeT $ blockAtOrBefore ch q let g = blockDataHeight b return $ AtBlock g -parseLimits :: Monad m => WebT m LimitsParam +parseLimits :: MonadIO m => WebT m LimitsParam parseLimits = LimitsParam <$> paramOptional <*> parseOffset <*> paramOptional paramToLimits :: @@ -1434,7 +1690,7 @@ -> LimitsParam -> WebT m Limits paramToLimits full (LimitsParam limitM o startM) = do - wl <- lift $ asks webMaxLimits + wl <- lift $ asks (webMaxLimits . webConfig) Limits (validateLimit wl full limitM) (fromIntegral o) <$> parseStart startM validateLimit :: WebLimits -> Bool -> Maybe LimitParam -> Word32 @@ -1455,24 +1711,27 @@ runInWebReader :: MonadIO m => CacheT (DatabaseReaderT m) a - -> ReaderT WebConfig m a + -> ReaderT WebState m a runInWebReader f = do - bdb <- asks (storeDB . webStore) - mc <- asks (storeCache . webStore) + bdb <- asks (storeDB . webStore . webConfig) + mc <- asks (storeCache . webStore . webConfig) lift $ runReaderT (withCache mc f) bdb -runNoCache :: MonadIO m => Bool -> ReaderT WebConfig m a -> ReaderT WebConfig m a +runNoCache :: MonadIO m => Bool -> ReaderT WebState m a -> ReaderT WebState m a runNoCache False f = f -runNoCache True f = - local (\s -> s {webStore = (webStore s) {storeCache = Nothing}}) f +runNoCache True f = local g f + where + g s = s { webConfig = h (webConfig s) } + h c = c { webStore = i (webStore c) } + i s = s { storeCache = Nothing } logIt :: (MonadUnliftIO m, MonadLoggerIO m) => m Middleware logIt = do runner <- askRunInIO return $ \app req respond -> do - t1 <- getCurrentTime + t1 <- systemToUTCTime <$> liftIO getSystemTime app req $ \res -> do - t2 <- getCurrentTime + t2 <- systemToUTCTime <$> liftIO getSystemTime let d = diffUTCTime t2 t1 s = responseStatus res msg = fmtReq req <> " [" <> fmtStatus s <> " / " <> fmtDiff d <> "]" diff --git a/test/Haskoin/StoreSpec.hs b/test/Haskoin/StoreSpec.hs --- a/test/Haskoin/StoreSpec.hs +++ b/test/Haskoin/StoreSpec.hs @@ -97,8 +97,8 @@ , storeConfPeerMaxLife = 48 * 3600 , storeConfConnect = dummyPeerConnect net ad , storeConfCacheRefresh = 750 - , storeConfCacheRetries = 100 , storeConfCacheRetryDelay = 100000 + , storeConfStats = Nothing } withStore cfg $ \Store {..} -> withSubscription storePublisher $ \sub ->