dbmonitor (empty) → 0.1.0

raw patch · 15 files changed

+934/−0 lines, 15 filesdep +ansi-terminaldep +asyncdep +basesetup-changed

Dependencies added: ansi-terminal, async, base, bytestring, dbmonitor, dhall, directory, filepath, fsnotify, hasql, lifted-base, monad-control, mtl, optparse-applicative, stm, telegram-bot-simple, text, time, transformers-base, unordered-containers, vector

Files

• CHANGELOG.md +4/−0
• LICENSE +29/−0
• README.md +120/−0
• Setup.hs +2/−0
• app/Main.hs +9/−0
• dbmonitor.cabal +96/−0
• src/Monitor/Configuration/Config.hs +103/−0
• src/Monitor/Configuration/Options.hs +29/−0
• src/Monitor/Configuration/Read.hs +82/−0
• src/Monitor/DB.hs +68/−0
• src/Monitor/DataModel.hs +51/−0
• src/Monitor/Entry.hs +93/−0
• src/Monitor/Loader.hs +44/−0
• src/Monitor/Queue.hs +114/−0
• src/Monitor/Telegram.hs +90/−0

@@ -0,0 +1,4 @@+0.1.0+---++Initial publication.

@@ -0,0 +1,29 @@+BSD 3-Clause License++Copyright (c) 2021, Vitalii Guzeev+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice, this+   list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright notice,+   this list of conditions and the following disclaimer in the documentation+   and/or other materials provided with the distribution.++3. Neither the name of the copyright holder nor the names of its+   contributors may be used to endorse or promote products derived from+   this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,120 @@+# db-monitor: data consistency monitoring for PostgreSQL.++[![Hackage](https://img.shields.io/hackage/v/dbmonitor.svg)](http://hackage.haskell.org/package/dbmonitor)++This simple tool periodically runs provided SQL-queries and alerts DBAs about failed checks via Telegram.+It's highly configurable and tracks configuration changes.++We believe such an instrument may help those who are administrating one or several databases on a single server to find data malformations early. Hence promptly notify developers about possible bugs and other oddities like inaccurate manual data manipulation. Other use case -- executing periodic actions like vacuuming on database with failure alerts.++And we hope it allows more structured monitoring process than arbitrary one built on shell scripts.++All unnecessary restrictions (PostgreSQL, Telegram) are coming from our limited resources and current conditions. They are not ideological, so we will be glad to accept pull requests broadening domain of the tool.++### Configuration and usage.++To be run `dbmonitor` requires a configuration directory. By default it will be searched in current directory under name `.monitor`, but you may pass it as an option (`--dir <path>` or `-D <path>`).++Structure of the directory is the following (in terms of `ls -R` command output):+```+<config-dir>:+<database1-dir>+<database2-dir>+<database3-dir>+...++<config-dir>/<database1-dir>:+conf.dhall check1.sql check2.sql check3.sql job1.sql ...+```++`conf.dhall` is a configuration file for single database. Name of the file is fixed. It stores connection string, list of Telegram IDs of alert-receivers (channels) and some other default settings.++Example contents of `conf.dhall`:+```+{ connection = "host=localhost user=user port=5432 dbname=postgres password=password"+, channels = [-1001408342381, ...]+, frequency = 1 -- Default period in minutes between checks.+, assertion = "null" -- Default assertion made against result of any query.+}+```++Files of checks and jobs are plain SQL files with optional comments required for setting check behavior. These files may have arbitrary names and are searched recursively in configuration directory. Hidden files are never tracked.++Proper check file contents may look like that:+```+-- frequency = 30+-- assertion = null+SELECT id FROM table1 WHERE NOT id = ANY(SELECT DISTINCT reference_column FROM table2);+```++or+```+SELECT id FROM table1 WHERE NOT id = ANY(SELECT DISTINCT reference_column FROM table2);+```++or+```+-- frequency = 10+-- assertion = zero+-- description = Some text+-- description = Some other text+SELECT count(id) FROM table1 WHERE obligatory_field IS NULL;+```++If `assertion` or `frequency` comment is omitted, default value applies.+These comments must satisfy the following regular expression: `^--\s*[:field:]\s*=\s*[:value:]\s*$`. `description` lines are optional and will appear in alerts.++Complex assertions can be made on SQL side or on Haskell side and it's more native to do them on SQL side, so `assertion` field may have only one of these values -- `null`, `not null`, `true`, `false`, `zero`, `resultless`. Last value is reserved for jobs.++Frequency is expected to be positive integer. I.e any value less than 1 will be treated as 1, decimal numbers will be truncated.++Incorrect assertion or syntactically wrong query will result in messages to maintainers.++Telegram token is expected to be stored in environmental variable `TG_TOKEN`. You can pass name of the variable as an option. `--token <variable-name>` or `-T <variable-name>`.++Log is written to stdout with immediate buffer flushes.++**Options Reference:**++* `--token <variable-name>` or `-T <variable-name>` -- name of environmental variable where Telegram token is stored.+* `--dir <path>` or `-D <path>` -- path to configuration directory.+* `--help` or `-h` -- see options reference in usual optparse-applicative format.++### Installation.++There are several options:+- Tool is now available at https://github.com/pandora-mccme/dbmonitor/pkgs/container/dbmonitor.+- It can be installed via `cabal install dbmonitor`.+- It can be cloned or downloaded  fron Github and built from source using Stack.++### Behavior details++#### Persistence++Tool is designed to be stable under changes of configuration.++This stability is achived in a following manner:+* If a new database directory is created in `.monitor`, it's automatically tracked.+* System tracks changes in check files and automatically reload them.+* New check files are automatically tracked and run as jobs.+* On removal of check file tool stops executing related job. Same for database directories - directory deletion causes monitor for this directory to stop.+* `conf.dhall` changes are tracked. If config is invalid, old jobs are removed, but database monitor is not stopped, when config is fixed everything will work again.++#### Other implementation details++* If assertion cannot be parsed from `conf.dhall` or check file, it is treated as `not null` instead of throwing error.+* Database queries are based on `hasql` library, which has quite strict control over what query is expected to return. So you may encounter some unfamiliar errors in Telegram. In all cases we know they are reasonable against provided assertions. Also usage of this library as a backend limits us to PostgeSQL databases. It's possible to extend to other backends.+* File watch was never tested on Mac and Windows.+* Hopefully you will find our logs detailed but not floody.++### Telegram caveat.++We expect users to be familiar with SQL, but there is a caveat in Telegram.++Usually monitoring channels must be private. It's impossible to send message to private channel by it's name, that's why we restricted channel field of `conf.dhall` to chat ids. Id of a private channel may be extracted from URL in a web version of Telegram.++### Attribution.++This tool was initially written for HyperMath team at Moscow Center for Continuous Mathematical Education.++**_Issues and pull requests are welcome._**

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,9 @@+module Main where++import Options.Applicative (execParser)++import Monitor.Configuration.Options (options)+import Monitor.Entry (runApp)++main :: IO ()+main = execParser options >>= runApp

@@ -0,0 +1,96 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.34.4.+--+-- see: https://github.com/sol/hpack++name:           dbmonitor+version:        0.1.0+synopsis:       Data consistency alerting for PostgreSQL+description:    Tool performing periodic checks of data invariants which are hard or impossible to be imposed by constraints. For Postgres only.+category:       Monitoring, Database+homepage:       https://github.com/viviag/dbmonitor#readme+bug-reports:    https://github.com/viviag/dbmonitor/issues+author:         Vitalii Guzeev+maintainer:     viviag@yandex.ru+copyright:      2021-2022 Vitalii Guzeev+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md+    CHANGELOG.md++source-repository head+  type: git+  location: https://github.com/viviag/dbmonitor++library+  exposed-modules:+      Monitor.Configuration.Config+      Monitor.Configuration.Options+      Monitor.Configuration.Read+      Monitor.DataModel+      Monitor.DB+      Monitor.Entry+      Monitor.Loader+      Monitor.Queue+      Monitor.Telegram+  other-modules:+      Paths_dbmonitor+  hs-source-dirs:+      src+  ghc-options: -Wall+  build-depends:+      ansi-terminal >=0.11.3 && <0.12+    , async >=2.2.2 && <2.3+    , base >=4.12.0 && <=4.15.1.0+    , bytestring >=0.10.8 && <0.11+    , dhall >=1.40.2 && <1.42+    , directory >=1.3.3 && <1.4+    , filepath >=1.4.2 && <1.5+    , fsnotify >=0.3.0 && <0.4+    , hasql >=1.4.4 && <=1.5.0.5+    , lifted-base >=0.2.3 && <0.3+    , monad-control >=1.0.3 && <1.1+    , mtl >=2.2.2 && <2.3+    , optparse-applicative >=0.16.1.0 && <0.18+    , stm >=2.5.0 && <2.6+    , telegram-bot-simple >=0.4.5 && <0.6+    , text >=1.2.3 && <1.3+    , time >=1.8.0 && <=1.9.3+    , transformers-base >=0.4.6 && <0.5+    , unordered-containers >=0.2.10 && <0.3+    , vector >=0.12.0 && <0.13+  default-language: Haskell2010++executable dbmonitor+  main-is: Main.hs+  other-modules:+      Paths_dbmonitor+  hs-source-dirs:+      app+  ghc-options: -threaded -rtsopts -with-rtsopts=-N -Wall+  build-depends:+      ansi-terminal >=0.11.3 && <0.12+    , async >=2.2.2 && <2.3+    , base >=4.12.0 && <=4.15.1.0+    , bytestring >=0.10.8 && <0.11+    , dbmonitor+    , dhall >=1.40.2 && <1.42+    , directory >=1.3.3 && <1.4+    , filepath >=1.4.2 && <1.5+    , fsnotify >=0.3.0 && <0.4+    , hasql >=1.4.4 && <=1.5.0.5+    , lifted-base >=0.2.3 && <0.3+    , monad-control >=1.0.3 && <1.1+    , mtl >=2.2.2 && <2.3+    , optparse-applicative >=0.16.1.0 && <0.18+    , stm >=2.5.0 && <2.6+    , telegram-bot-simple >=0.4.5 && <0.6+    , text >=1.2.3 && <1.3+    , time >=1.8.0 && <=1.9.3+    , transformers-base >=0.4.6 && <0.5+    , unordered-containers >=0.2.10 && <0.3+    , vector >=0.12.0 && <0.13+  default-language: Haskell2010

@@ -0,0 +1,103 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingVia #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE ImplicitParams #-}+{-# LANGUAGE BangPatterns #-}+module Monitor.Configuration.Config where++import System.Console.ANSI+import System.IO (hFlush, stdout)++import Control.Concurrent+import Control.Concurrent.STM.TVar+import Control.Exception+import Control.Monad.IO.Class++import qualified Data.ByteString.Char8 as BSC+import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HM+import Data.Time++import qualified Hasql.Connection as HaSQL++import Dhall ( Generic, auto, inputFile, FromDhall, Natural )+import Dhall.Deriving++logMessage :: (?mutex :: Mutexes) => MonadIO m => String -> m ()+logMessage event = liftIO $ do+  takeMVar (stdoutMutex ?mutex)+  !time <- getCurrentTime+  setSGR [SetColor Foreground Dull Green]+  putStr $ (show time) <> ": "+  setSGR [SetDefaultColor Foreground]+  putStrLn $ event+  hFlush stdout+  putMVar (stdoutMutex ?mutex) ()++data Mutexes = Mutexes {+    stdoutMutex :: MVar ()+  } deriving (Eq)++data Config = Config+  { configConnection :: String+  , configChannels :: [Int]+  , configFrequency :: Natural+  , configAssertion :: String+  }+  deriving (Eq, Show, Generic)+  deriving+    (FromDhall)+    via Codec (Dhall.Deriving.Field (SnakeCase <<< DropPrefix "config")) Config++-- NOTE: resultless is for periodic actions.+data Assertion = AssertNull | AssertNotNull | AssertTrue | AssertFalse | AssertZero | AssertResultless+  deriving (Eq, Show)++data Settings = Settings+  { dbConnection :: HaSQL.Connection+  , channels :: [Integer]+  , defaultFrequency :: Int+  , defaultAssertion :: Assertion+  , telegramTokenVar :: String+  , databaseDirectory :: FilePath+  , jobQueue :: TVar (HashMap FilePath ThreadId)+  }++readAssertion :: String -> Assertion+readAssertion "null" = AssertNull+readAssertion "true" = AssertTrue+readAssertion "false" = AssertFalse+readAssertion "zero" = AssertZero+readAssertion "resultless" = AssertResultless+-- NOTE: mention in README.+readAssertion _ = AssertNotNull++readSettings :: (?mutex :: Mutexes) => FilePath -> String -> FilePath -> IO (Maybe Settings)+readSettings dbDir tokenVar configPath = do+  cfg <- try $ inputFile auto configPath+  case cfg of+    Left ex -> do+      logMessage ("Config for " <> dbDir <> " cannot be read. See exception below.")+      putStrLn ("Exception: " <> show @SomeException ex)+      return Nothing+    Right Config{..} -> do+      dbConnection <- HaSQL.acquire (BSC.pack configConnection)+      case dbConnection of+        Left _ -> do+          logMessage ("Config error: connection string for " <> dbDir <> " directory does not provide connection to any database")+          return Nothing+        Right conn -> do+          queue <- newTVarIO HM.empty+          return . Just $ Settings+            { dbConnection = conn+            , channels = map fromIntegral configChannels+            , defaultFrequency = fromIntegral configFrequency+            , defaultAssertion = readAssertion configAssertion+            , telegramTokenVar = tokenVar+            , databaseDirectory = dbDir+            , jobQueue = queue+            }

@@ -0,0 +1,29 @@+module Monitor.Configuration.Options where++import Options.Applicative++data Options = Options {+    optionsDir   :: FilePath+  , optionsToken :: String+  } deriving (Show, Eq)++optionsParser :: Parser Options+optionsParser = Options+  <$> strOption+      ( long "dir"+     <> short 'D'+     <> metavar "CONFIG_DIR"+     <> value ".monitor"+     <> help "Path to configuration directory" )+  <*> strOption+      ( long "token"+     <> short 'T'+     <> metavar "TOKEN_VAR"+     <> value "TG_TOKEN"+     <> help "Variable with Telegram access token" )++options :: ParserInfo Options+options = info (optionsParser <**> helper)+   ( fullDesc+  <> progDesc "PostgreSQL data consistency monitoring tool"+  <> header "dbmonitor" )

@@ -0,0 +1,82 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ImplicitParams #-}+module Monitor.Configuration.Read+  ( readMonitor+  , notHidden+  , isCheck+  , collectMonitors+  )+  where++import Control.Concurrent++import System.Directory+import System.FilePath+import System.FSNotify++import Monitor.Configuration.Config+import Monitor.DataModel++collectMonitors :: FilePath -> IO [FilePath]+collectMonitors configDir = do+  mDatabaseDirs <- listDirectory configDir+  relativePaths <- filterM doesDirectoryExist+    (map (configDir </>) . filter notHidden $ mDatabaseDirs)+  currentDir <- getCurrentDirectory+  return $ map (currentDir </>) relativePaths++tryReadConfig :: (?mutex :: Mutexes) => FilePath -> MVar () -> String -> IO Settings+tryReadConfig dir configChange tgvar = do+  takeMVar configChange+  readConfig dir configChange tgvar++readConfig :: (?mutex :: Mutexes) => FilePath -> MVar () -> String -> IO Settings+readConfig dir configChange tgvar = do+  let configPath = dir </> configName+  configExists <- doesFileExist configPath+  if configExists+    then do+      mSettings <- readSettings dir tgvar configPath+      case mSettings of+        Nothing -> tryReadConfig dir configChange tgvar+        Just cfg -> return cfg+    else tryReadConfig dir configChange tgvar++modifiedCfg :: MVar () -> Event -> IO ()+modifiedCfg mvar _ = putMVar mvar ()++readConfigTillSuccess :: (?mutex :: Mutexes)+                      => WatchManager -> FilePath -> String -> MVar () -> IO Settings+readConfigTillSuccess cfgManager dir tgvar configChange = do+  removeWatch <- watchDir cfgManager dir (const True)+    (modifiedCfg configChange)+  cfg <- readConfig dir configChange tgvar+  logMessage $ "Successfully read configuration at " <> dir+  removeWatch+  return cfg++notHidden :: FilePath -> Bool+notHidden ('.':_) = False+notHidden _ = True++isCheck :: FilePath -> Bool+isCheck f = notHidden f && not (representsConfigName f) -- && takeExtension f == ".sql"++readInitialData :: FilePath -> IO [FilePath]+readInitialData dir = do+  contents <- listDirectory dir+  files <- filterM doesFileExist . map (dir </>) . filter isCheck $ contents+  subdirs <- filterM doesDirectoryExist . map (dir </>) . filter notHidden $ contents+  relativePaths <- case subdirs of+    [] -> return files+    lst -> mapM (readInitialData) lst+       >>= \nested -> return (files ++ concat nested)+  currentDir <- getCurrentDirectory+  return $ map (currentDir </>) relativePaths++readMonitor :: (?mutex :: Mutexes) => FilePath -> String -> IO (Settings, [FilePath])+readMonitor dir tgvar = do+  configReadMVar <- newEmptyMVar+  cfg <- withManager (\cfgManager -> readConfigTillSuccess cfgManager dir tgvar configReadMVar)+  checks <- readInitialData dir+  return (cfg, checks)

@@ -0,0 +1,68 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ImplicitParams #-}+{-# LANGUAGE BangPatterns #-}+module Monitor.DB where++import Data.ByteString (ByteString)+import qualified Data.Vector as V++import qualified Hasql.Session as HaSQL+import qualified Hasql.Statement as HaSQL+import qualified Hasql.Decoders as D+import qualified Hasql.Encoders as E++import Monitor.DataModel++decodeAssertResultless :: D.Result Bool+decodeAssertResultless = (\() -> True) <$> D.noResult++decodeAssertNull :: D.Result Bool+decodeAssertNull = test . V.toList <$> D.rowVector (D.column (D.nullable (D.custom (\_ _ -> Right ()))))+  where+    test [] = True+    test [Nothing] = True+    test _ = False++decodeAssertNotNull :: D.Result Bool+decodeAssertNotNull = not <$> decodeAssertNull++decodeAssertTrue :: D.Result Bool+decodeAssertTrue = test <$> D.rowMaybe (D.column (D.nullable D.bool))+  where+    test Nothing = False+    test (Just Nothing) = False+    test (Just (Just a)) = a++decodeAssertFalse :: D.Result Bool+decodeAssertFalse = test <$> D.rowMaybe (D.column (D.nullable D.bool))+  where+    test Nothing = False+    test (Just Nothing) = False+    test (Just (Just a)) = not a++decodeAssertZero :: D.Result Bool+decodeAssertZero = test <$> D.rowMaybe (D.column (D.nullable D.int4))+  where+    test Nothing = False+    test (Just Nothing) = False+    test (Just (Just a)) = a == 0++session :: Assertion -> ByteString -> HaSQL.Session Bool+session assertion sql = HaSQL.statement () $ case assertion of+  AssertNull -> HaSQL.Statement sql E.noParams decodeAssertNull False+  AssertNotNull -> HaSQL.Statement sql E.noParams decodeAssertNotNull False+  AssertZero -> HaSQL.Statement sql E.noParams decodeAssertZero False+  AssertTrue -> HaSQL.Statement sql E.noParams decodeAssertTrue False+  AssertFalse -> HaSQL.Statement sql E.noParams decodeAssertFalse False+  AssertResultless -> HaSQL.Statement sql E.noParams decodeAssertResultless False++runSQL :: PureJob -> Monitor JobFeedback+runSQL PureJob{..} = do+  conn <- asks dbConnection+  !result <- liftIO $ HaSQL.run (session pureJobAssertion pureJobSQL) conn+  return $ case result of+    Left (HaSQL.QueryError _ _ (HaSQL.ClientError err)) ->+      ConnectionError (show err)+    Left (HaSQL.QueryError _ _ (HaSQL.ResultError err)) ->+      QueryError (show err)+    Right assertionResult -> AssertionResult assertionResult

@@ -0,0 +1,51 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE DerivingVia #-}+{-# LANGUAGE UndecidableInstances #-}+module Monitor.DataModel (+    module Monitor.DataModel+  , module Control.Monad.Reader+  , Assertion(..)+  , Settings(..)+  , Mutexes(..)+  , readAssertion+  , logMessage+  ) where++import Control.Monad.Base+import Control.Monad.Reader+import Control.Monad.Trans.Control++import Data.ByteString (ByteString)++import Monitor.Configuration.Config++newtype Monitor a = Monitor {getMonitor :: ReaderT Settings IO a} deriving+  (Functor, Applicative, Monad, MonadIO, MonadReader Settings, MonadBase IO, MonadBaseControl IO)+  via ReaderT Settings IO++configName :: FilePath+configName = "conf.dhall"++representsConfigName :: FilePath -> Bool+representsConfigName path = path == configName++data JobFeedback = ConnectionError String+                 | QueryError String+                 | AssertionResult Bool+  deriving (Eq, Show)++data Job = Job+  { jobDescription :: Maybe String+  , jobFrequency :: Maybe Int+  , jobAssertion :: Maybe Assertion+  , jobSQL :: ByteString+  } deriving (Eq, Show)++data PureJob = PureJob+  { pureJobDescription :: String+  , pureJobAssertion :: Assertion+  , pureJobSQL :: ByteString+  } deriving (Eq, Show)

@@ -0,0 +1,93 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ImplicitParams #-}+module Monitor.Entry where++import GHC.Conc (labelThread, atomically)++import System.FilePath+import System.FSNotify++import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM.TVar++import Data.HashMap.Strict (HashMap)+import qualified Data.HashMap.Strict as HM++import Monitor.Configuration.Options (Options(..))+import Monitor.Configuration.Config+import Monitor.Configuration.Read+import Monitor.Queue+import Monitor.DataModel++watchTower :: (?mutex :: Mutexes)+           => MVar () -> FilePath -> String -> Settings+           -> TVar (HashMap FilePath (MVar ())) -> Event -> IO ()+watchTower monitorHolder dir tgvar cfg locksTVar event = do+  let path = eventPath event+      filename = takeFileName path+  if isCheck filename+    then do+      label path . async+                 . flip runReaderT cfg+                 . getMonitor $+        case event of+          Modified _ _ False -> restartJob path+          Removed _ _ False -> removeJob path+          Added _ _ False -> startJob path+          _ -> pure ()+    else when (representsConfigName filename) $ do+      putMVar monitorHolder ()+      flip runReaderT cfg . getMonitor $ destroyQueue+      logMessage ("Monitor at " <> dir <> " is stopped due to configuration change. All jobs are removed, monitor will be restarted.")+      trackDatabase tgvar dir locksTVar++trackDatabase :: (?mutex :: Mutexes) => String -> FilePath+              -> TVar (HashMap FilePath (MVar ())) -> IO ()+trackDatabase tgvar dbDir locksTVar = do+  (cfg, checks) <- readMonitor dbDir tgvar+  logMessage $ "Monitor at " <> dbDir <> " is started."+  withManager $+    \monitorManager -> do+      lock <- newEmptyMVar+      atomically $ modifyTVar locksTVar (HM.insert dbDir lock)+      void $ watchTree monitorManager dbDir (const True)+             (watchTower lock dbDir tgvar cfg locksTVar)+      mapM_ (\f -> void . async $ runReaderT (getMonitor $ startJob f) cfg) checks+      takeMVar lock++watchNewTrack :: (?mutex :: Mutexes) => String+              -> TVar (HashMap FilePath (MVar ())) -> Event -> IO ()+watchNewTrack _ locksTVar (Removed path _ True) = do+  locks <- liftIO $ readTVarIO locksTVar+  putMVar (locks HM.! path) ()+  atomically $ modifyTVar locksTVar (HM.delete path)+  logMessage $ "Monitor at " <> path <> " is stopped due to directory deletion."+watchNewTrack tgvar locksTVar (Added path _ True) =+  spawnMonitorThread tgvar locksTVar path+watchNewTrack _ _ _ = pure ()++label :: String -> IO (Async ()) -> IO ()+label lab action = do+  asyn <- action+  labelThread (asyncThreadId asyn) lab++spawnMonitorThread :: (?mutex :: Mutexes) => String+                   -> TVar (HashMap FilePath (MVar ())) -> FilePath -> IO ()+spawnMonitorThread tgvar locksTVar dir =+  label dir . async $ trackDatabase tgvar dir locksTVar++runApp :: Options -> IO ()+runApp Options{..} = do+  stdoutMutex <- newMVar ()+  let ?mutex = Mutexes{..} in do+    logMessage "dbmonitor process started."+    databaseDirs <- collectMonitors optionsDir+    eventChannel <- newChan+    locksTVar <- newTVarIO HM.empty+    withManager $ \mainWatcher -> do+      void $ watchTreeChan mainWatcher optionsDir (const True) eventChannel+      forM_ databaseDirs $ spawnMonitorThread optionsToken locksTVar+      forever $ do+        event <- readChan eventChannel+        watchNewTrack optionsToken locksTVar event

@@ -0,0 +1,44 @@+{-# LANGUAGE OverloadedStrings #-}+module Monitor.Loader where++import Data.List+import Data.Maybe+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T++import Monitor.DataModel++parseLine :: Text -> Maybe Text+parseLine space = case T.splitOn "=" space of+  [_key, value] -> Just . T.strip $ value+  _ -> Nothing++parseDescription :: [Text] -> Maybe String+parseDescription arg = case catMaybes (map parseLine arg) of+  [] -> Nothing+  a -> Just . intercalate "\n" . map T.unpack $ a++parseAssertion :: [Text] -> Maybe Assertion+parseAssertion arg = case catMaybes (map parseLine arg) of+  [] -> Nothing+  (a:_) -> Just . readAssertion . T.unpack $ a++parseFrequency :: [Text] -> Maybe Int+parseFrequency arg = case catMaybes (map parseLine arg) of+  [] -> Nothing+  (a:_) -> Just . read . T.unpack $ a++parseJob :: Text -> Job+parseJob txt = Job {+    jobSQL = sql+  , jobDescription = parseDescription mDescription+  , jobAssertion = parseAssertion mAssertion+  , jobFrequency = parseFrequency mFrequency+  }+  where+    comments = filter (T.isPrefixOf "--") . map T.strip . T.lines $ txt+    sql = T.encodeUtf8 . T.unlines . filter (not . T.isPrefixOf "--") . map T.strip . T.lines $ txt+    mDescription = filter (T.isInfixOf "description") comments+    mFrequency = filter (T.isInfixOf "frequency") comments+    mAssertion = filter (T.isInfixOf "assertion") comments

@@ -0,0 +1,114 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ImplicitParams #-}+module Monitor.Queue where++import GHC.Conc++import Control.Concurrent+import qualified Control.Concurrent.Lifted as Lifted+import Control.Concurrent.STM.TVar++import System.Directory+import System.FilePath++import qualified Data.HashMap.Strict as HM+import Data.Maybe+import qualified Data.Text.IO as T+import Data.Time++import Monitor.DataModel+import Monitor.Loader+import Monitor.DB+import Monitor.Telegram++-- This is a hack. On connection error all thread must try to restart.+touchConfig :: (?mutex :: Mutexes) => Monitor ()+touchConfig = do+  dir <- asks databaseDirectory+  logMessage ("Monitor at " <> dir <> "is restarted in order to reestablish db connection.")+  time <- liftIO getCurrentTime+  liftIO $ setModificationTime (dir </> configName) time++processQueryResult :: (?mutex :: Mutexes) => FilePath -> PureJob -> JobFeedback -> Monitor ()+processQueryResult _path _ (ConnectionError err) =+  alertConnectionError err >> touchConfig+processQueryResult path PureJob{..} (QueryError err) =+  alertQueryError path err pureJobSQL+processQueryResult path job (AssertionResult value) =+  if value+    then pure ()+    else alertFailedAssertion path job++purify :: Job -> Assertion -> FilePath -> PureJob+purify Job{..} assertion path = PureJob {+    pureJobDescription = fromMaybe ("Job at " <> path) jobDescription+  , pureJobAssertion = fromMaybe assertion jobAssertion+  , pureJobSQL = jobSQL+  }++periodicEvent :: (?mutex :: Mutexes) => Job -> FilePath -> Monitor ()+periodicEvent job@Job{..} path = forever $ do+  defFreq <- asks defaultFrequency+  defAssert <- asks defaultAssertion+  let pureJob = purify job defAssert path+      delay = 60 * 10^((6)::Int) * (fromMaybe defFreq jobFrequency)+  queryResult <- runSQL pureJob+  processQueryResult path pureJob queryResult+  logMessage ("Job at " <> path <> " is executed.")+  liftIO $ threadDelay delay++forkWaitable :: (?mutex :: Mutexes) => Monitor () -> Monitor (ThreadId, MVar ())+forkWaitable action = do+  handle <- liftIO newEmptyMVar+  thread <- Lifted.forkFinally action (\_ -> liftIO $ putMVar handle ())+  return (thread, handle)++startJob :: (?mutex :: Mutexes) => FilePath -> Monitor ()+startJob path = do+  queue <- asks jobQueue+  job <- liftIO $ parseJob <$> T.readFile path+  queueMap <- liftIO $ readTVarIO queue+  case HM.lookup path queueMap of+    Nothing -> pure ()+    Just accidental_thread -> do+      logMessage ("Job " <> path <> " was probably initiated by different monitor threads. Please report a bug.")+      liftIO $ killThread accidental_thread+  (thread, waitHandle) <- forkWaitable (periodicEvent job path)+  liftIO $ labelThread thread ("jobThread: " <> path)+  liftIO . atomically $ modifyTVar queue (HM.insert path thread)+  logMessage ("Job " <> path <> " is started.")+  void $ liftIO $ takeMVar waitHandle++removeJob :: (?mutex :: Mutexes) => FilePath -> Monitor ()+removeJob path = do+  queueTVar <- asks jobQueue+  queue <- liftIO $ readTVarIO queueTVar+  liftIO . killThread $ queue HM.! path+  liftIO . atomically $ modifyTVar queueTVar (HM.delete path)+  logMessage ("Job " <> path <> " is removed")++restartJob :: (?mutex :: Mutexes) => FilePath -> Monitor ()+restartJob path = do+  queueTVar <- asks jobQueue+  queue <- liftIO $ readTVarIO queueTVar+  liftIO . killThread $ queue HM.! path+  job <- liftIO $ parseJob <$> T.readFile path+  (thread, waitHandle) <- forkWaitable (periodicEvent job path)+  liftIO $ labelThread thread ("jobThread: " <> path)+  liftIO . atomically $ modifyTVar queueTVar (HM.adjust (\_ -> thread) path)+  logMessage ("Job " <> path <> " is restarted due to file modification.")+  void $ liftIO $ takeMVar waitHandle++destroyQueue :: Monitor ()+destroyQueue = do+  queueTVar <- asks jobQueue+  queue <- liftIO $ readTVarIO queueTVar+  mapM_ (liftIO . killThread) $ HM.elems queue+  liftIO . atomically $ modifyTVar queueTVar (\_ -> HM.empty)++destroyMonitor :: (?mutex :: Mutexes) => MVar () -> Monitor ()+destroyMonitor monitorHolder = do+  liftIO $ putMVar monitorHolder ()+  destroyQueue+  alertThreadDeath

@@ -0,0 +1,90 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ImplicitParams #-}+module Monitor.Telegram where++import Data.ByteString (ByteString)+import Data.Text (pack, Text)+import Data.Text.Encoding (decodeUtf8)++import Telegram.Bot.API.MakingRequests+import Telegram.Bot.API.Methods+import Telegram.Bot.API.Types+import Telegram.Bot.Simple.BotApp (getEnvToken)++import Monitor.DataModel++standardRequest :: SomeChatId -> Text -> SendMessageRequest+standardRequest chan txt = SendMessageRequest+  { sendMessageChatId                   = chan+  , sendMessageText                     = txt+  , sendMessageParseMode                = Just Markdown+  , sendMessageDisableWebPagePreview    = Just True+  , sendMessageDisableNotification      = Just False+  , sendMessageReplyToMessageId         = Nothing+  , sendMessageReplyMarkup              = Nothing+  , sendMessageEntities                 = Nothing+  , sendMessageProtectContent           = Nothing+  , sendMessageAllowSendingWithoutReply = Nothing+  }++postAlert :: SendMessageRequest -> Monitor ()+postAlert msg = asks telegramTokenVar >>= \tgvar -> liftIO $ do+  token <- getEnvToken tgvar+  resp <- defaultRunBot token (sendMessage msg)+  print resp++broadcast :: (SomeChatId -> SendMessageRequest) -> Monitor ()+broadcast f = do+  chans <- asks channels+  mapM_ (postAlert . f. SomeChatId . ChatId) chans++deathNote :: FilePath -> SomeChatId -> SendMessageRequest+deathNote dir chan = standardRequest chan msg+  where+    msg = "*Monitor at " <> (pack dir) <> " has stopped by deleting or moving it's working directory*"++alertThreadDeath :: (?mutex :: Mutexes) => Monitor ()+alertThreadDeath = do+  dir <- asks databaseDirectory+  broadcast $ deathNote dir+  logMessage ("Death alert sent for monitor at " <> dir)++connectionErrorMessage :: String -> FilePath -> SomeChatId -> SendMessageRequest+connectionErrorMessage err dir chan = standardRequest chan msg+  where+    msg = "*Cannot connect to database at " <> (pack dir) <> ".* \n\+          \It may indicate cluster restart, check all applications.\n\+          \_Error message_: " <> (pack err)++alertConnectionError :: (?mutex :: Mutexes) => String -> Monitor ()+alertConnectionError err = do+  dir <- asks databaseDirectory+  broadcast $ connectionErrorMessage err dir+  logMessage ("Database connection problem alert sent for monitor at " <> dir)++queryErrorMessage :: FilePath -> String -> ByteString -> SomeChatId -> SendMessageRequest+queryErrorMessage path err sql chan = standardRequest chan msg+  where+    msg = "*Query error while executing check `"  <> (pack path) <> "`.* \n\+          \*Error message: *\n```" <> (pack err) <> "```\n\+          \*SQL text*: ```\n" <> (decodeUtf8 sql) <> "```\n\+          \It means incorrect assertion (parse errors are treated as 'not null') or error in query."++alertQueryError :: (?mutex :: Mutexes) => FilePath -> String -> ByteString -> Monitor ()+alertQueryError path err sql = do+  broadcast $ queryErrorMessage path err sql+  logMessage ("Query error alert sent for " <> path)++assertionMessage :: FilePath -> Assertion -> ByteString -> String -> SomeChatId -> SendMessageRequest+assertionMessage path assertion sql desc chan = standardRequest chan msg+  where+    msg = "*Assertion failed*:\nCheck `" <> (pack path) <> "`. \n\n\+          \*Assertion: *" <> (pack (show assertion)) <> "\n\n\+          \*SQL text*: ```\n" <> (decodeUtf8 sql) <> "```\n\+          \_Check description_:\n" <> (pack desc)++alertFailedAssertion :: (?mutex :: Mutexes) => FilePath -> PureJob -> Monitor ()+alertFailedAssertion path PureJob{..} = do+  broadcast (assertionMessage path pureJobAssertion pureJobSQL pureJobDescription)+  logMessage ("Failed assertion alert sent for " <> path)