diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,29 @@ +BSD 3-Clause License + +Copyright (c) 2017, SeatGeek +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + +* Redistributions of source code must retain the above copyright notice, this + list of conditions and the following disclaimer. + +* 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. + +* 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. diff --git a/src/Invoker.hs b/src/Invoker.hs new file mode 100644 --- /dev/null +++ b/src/Invoker.hs @@ -0,0 +1,134 @@ +{-# LANGUAGE RecordWildCards, NamedFieldPuns #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} + +module Invoker where + +import Control.Exception (SomeException, try) +import Control.Monad (foldM) +import Data.Aeson (eitherDecodeStrict) +import Data.Binary (Binary) +import qualified Data.ByteString as BS +import Data.Either (lefts, rights) +import Data.Text (Text, pack, unpack) +import Data.Typeable (Typeable) +import GHC.Generics +import Model (Rollup(..), WreckerRun(..)) +import System.Environment (lookupEnv) +import System.IO.Temp (withSystemTempFile) +import System.Process (callProcess, readCreateProcess, shell) + +-- | Represents the concurrency (number of threads) used to execute a run +newtype Concurrency = + Concurrency Int + deriving (Eq, Show, Typeable, Generic) + +-- | The number of seconds to spend on a particular run. +newtype Seconds = + Seconds Int + deriving (Eq, Show, Typeable, Generic) + +-- | Specifies a wrecker CLI command +data Command = Command + { title :: Text + , seconds :: Seconds + , concurrency :: Concurrency + } deriving (Eq, Show, Generic, Typeable) + +-- We need these intances to serialize commands over the network +instance Binary Concurrency + +instance Binary Seconds + +instance Binary Command + +-- | Represents the exit status of a wrecker invocation. +data Result + = Success + | ExecError [Text] + | TooManyErrors + deriving (Show) + +-- | A helper function used to execute a group of wrecker invocations, given a list of concurrency levels. +-- This function is higly polymorphic, it only encodes the idea of repeatedly building a command, executing +-- it and recording its results. You have to provide the implementation of each of those individual steps +escalate :: + Monad m + => (Concurrency -> Command) -- ^ A function that given the concurrecy, will create a Command record + -> (Command -> m [Either String WreckerRun]) -- ^ A function to execute the tests in wrecker + -> (Command -> m key) -- ^ A function to create the run identifier + -> (key -> [Either String WreckerRun] -> m ()) -- ^ A function to record the run + -> [Concurrency] -- ^ The list of concurrency steps to execute in wrecker + -> m Result +escalate builder executer keyGen recorder steps = foldM run Success steps + where + run prev step = do + case prev of + Success -> execution step + err -> return err + execution step = do + let command = builder step + key <- keyGen command + results <- executer command + recorder key results + case lefts results of + [] -> + if isUnAcceptable (rights results) + then return TooManyErrors + else return Success + errors -> return (ExecError (fmap pack errors)) + +-- | Whether or not there were too many errors when invoking a wrecker run +isUnAcceptable :: [WreckerRun] -> Bool +isUnAcceptable [] = False +-- | A run is not acceptable if more than 1% of the hits were errors +isUnAcceptable results = + let (errors, hits) = foldr extract (0, 0) results + in errors / hits > (0.01 :: Double) + where + extract (WreckerRun {rollup = Rollup {..}}) (errors, hits) = + ( errors + fromIntegral (rollupUserErrorHits + rollupServerErrorHits + rollupFailedHits) + , hits + fromIntegral rollupHits) + +-- | Invokes the wrecker command line tool with the given options +wrecker :: Command -> IO (Either String WreckerRun) +wrecker (Command title (Seconds secs) (Concurrency conc)) = do + res <- try runScript + case res :: Either SomeException (Either String WreckerRun) of + Left err -> return (Left (show err)) + Right runResult -> return runResult + where + runScript = do + executable <- findExecutable + withSystemTempFile "wrecker.results" $ \file handle -> do + callProcess + executable + [ "--concurrency" + , show conc + , "--log-level" + , "LevelInfo" + , "--match" + , unpack title + , "--run-timed" + , show secs + , "--output-path" + , file + ] + contents <- BS.hGetContents handle + return (eitherDecodeStrict contents) + +-- | Returns the list of tests that are executable in the wrecker CLI +listGroups :: IO [String] +listGroups = do + executable <- findExecutable + output <- readCreateProcess (shell $ executable ++ " --list-test-groups") "" + let each = lines output + return (fmap (dropWhile (\c -> c `elem` notMeaningfulChars)) each) + where + notMeaningfulChars = '.' : '>' : ' ' : ['0' .. '9'] + +-- | Returns the default executable name or the one present in the WRECKER_EXECUTABLE env var +findExecutable :: IO String +findExecutable = do + executable <- lookupEnv "WRECKER_EXECUTABLE" + return (maybe "wrecker" id executable) diff --git a/src/Main.hs b/src/Main.hs new file mode 100644 --- /dev/null +++ b/src/Main.hs @@ -0,0 +1,399 @@ +{-# LANGUAGE RecordWildCards, NamedFieldPuns, DeriveGeneric, + TypeApplications #-} + +import Control.Concurrent.Async (race) +import Control.Monad (void) +import Control.Monad.IO.Class (liftIO) +import Control.Monad.IO.Class (MonadIO) +import Control.Monad.Reader (ask) +import Data.Aeson hiding (json) +import Data.Either (lefts, rights) +import qualified Data.Map as Map +import Data.Maybe (catMaybes, listToMaybe) +import Data.Monoid ((<>)) +import Data.Text (Text) +import qualified Data.Text as Text +import Data.Text.Encoding (encodeUtf8) +import qualified Data.Text.Internal.Lazy as LText +import Data.Time.Clock (getCurrentTime) +import Database.PostgreSQL.Simple.URL (parseDatabaseUrl) +import Network.HTTP.Types +import Network.HostAndPort (hostAndPort) +import Network.Wai.Middleware.Cors (simpleCors) +import qualified System.Directory as Dir +import System.Environment (lookupEnv) +import Text.Read (readMaybe) +import Web.Scotty + +import qualified Database.Persist as P +import qualified Database.Persist.Sql as Sql +import Model + (Database, DbBackend(..), Key, LogLevel(..), Page(..), Rollup(..), + Run(..), TransactionMonad, WreckerRun(..), findPageStats, + findPagesList, findRunStats, findRuns, findRunsByMatch, + runDbAction, runMigrations, storeRunResults, withDb) + +import Control.Distributed.Process (NodeId(..), Process) +import qualified Control.Distributed.Process.Backend.SimpleLocalnet + as LocalNet +import qualified Control.Distributed.Process.Node as Node +import Control.Distributed.Static (RemoteTable) +import Network.Transport (EndPointAddress(..)) + +import qualified Scheduler + +-- | CLI options +data Config = Config + { uiPort :: Int -- ^ The port for the HTTP interface + , dbType :: DbBackend -- ^ Either Sqlite or Postgres + , folder :: String -- ^ The path to the assets folder + , logLevel :: LogLevel -- ^ Either Debug or Silent. Currently ony used for db queries + , hostName :: String -- ^ The host this node is running on. Defaults to 127.0.0.1. + , staticSlaves :: [NodeId] + , testsList :: Scheduler.RunSchedule -- ^ Private config holding the run scheduler + } + +-- | The list of functions that are allowed to be called on remote nodes. +networkFunctions :: RemoteTable +networkFunctions = Scheduler.__remoteTable Node.initRemoteTable + +main :: IO () +main = do + role <- lookupEnv "WRECKER_ROLE" + case role of + Just "slave" -> createSlave + Just "master" -> mainProcess =<< getConfig + Nothing -> mainProcess =<< getConfig + _ -> error "Only master and slave roles are valid" + +-- | Initializes the CLI config +getConfig :: IO Config +getConfig = do + uiPort <- readPort "WRECKER_UI_PORT" 3000 + maybeLevel <- lookupEnv "WRECKER_LOG_LEVEL" + let parsedLevel = maybe Nothing readMaybe maybeLevel + logLevel = maybe Silent id parsedLevel + dbType <- selectDatabaseType + folder <- findAssetsFolder + staticSlaves <- getStaticSlaves + testsList <- Scheduler.emptyRunSchedule + hostName <- maybe "127.0.0.1" id <$> lookupEnv "WRECKER_HOST" + return (Config {..}) + +-- | Starts a slave worker node. Slaves are used to execute the wrecker cli and transmit back the results +createSlave :: IO () +createSlave = do + port <- readPort "WRECKER_PORT" 10501 + hostName <- maybe "127.0.0.1" id <$> lookupEnv "WRECKER_HOST" + backend <- LocalNet.initializeBackend hostName (show @Int port) networkFunctions + putStrLn "Started slave process" + LocalNet.startSlave backend + +-- | Starts the local worker node. This node will always receive part of the load, specially when the +-- cocurrency level for a run is low. +mainProcess :: Config -> IO () +mainProcess config = do + port <- readPort "WRECKER_PORT" 10500 + hasStaticSlaves <- lookupEnv "WRECKER_SLAVES" + putStrLn "Starting local node" + backend <- LocalNet.initializeBackend (hostName config) (show @Int port) networkFunctions + case hasStaticSlaves of + Nothing -> LocalNet.startMaster backend (startUI config) + _ -> do + putStrLn "Found a static list of slaves" + node <- (LocalNet.newLocalNode backend) + Node.runProcess node (startUI config []) + +-- | Starts the program itself. That is, the http interface and the run scheduler +startUI :: Config -> [NodeId] -> Process () +startUI (Config {..}) discoveredSlaves = do + liftIO $ putStrLn "Starting Web UI" + let slaves = staticSlaves ++ discoveredSlaves + localProcess <- ask -- Get the context that the Process monad is enclosing + liftIO $ do + putStrLn ("Found the following slave servers: " ++ show slaves) + withDb logLevel dbType $ \db -> + liftIO $ -- We're inside the DbMonad, so the results needs to be converted back to IO + do + runMigrations db -- Run the database migrations first + void $ -- Disregard the return value of `race` + race -- start both function in different threads and stop the other if one dies + (Scheduler.runScheduler Scheduler.Config {..}) -- start accepting new jobs + (routes uiPort folder db testsList) -- start accepting http requests + +---------------------------------- +-- App initialization +---------------------------------- +findAssetsFolder :: IO String +findAssetsFolder = do + f <- lookupEnv "WRECKER_ASSETS" + let folder = + case f of + Nothing -> "assets/" + Just path -> path <> "/" + exists <- Dir.doesFileExist (folder <> "app.js") + if not exists + then error + "Could not find the assets folder. Set the WRECKER_ASSETS env variable with the full path to it" + else return folder + +selectDatabaseType :: IO DbBackend +selectDatabaseType = do + dbInfo <- lookupEnv "WRECKER_DB" + case dbInfo of + Nothing -> do + putStrLn "Using SQLite" + return Sqlite + Just info -> + case parseDatabaseUrl info of + Nothing -> error "Invalid WRECKER_DB url" + Just connDetails -> do + putStrLn "Using PostgreSQL" + return (Postgresql connDetails) + +getStaticSlaves :: IO [NodeId] +getStaticSlaves = do + maybeString <- lookupEnv "WRECKER_SLAVES" + case maybeString of + Nothing -> return [] + Just "" -> return [] + Just str -> do + let parsed = parseNodes (Text.pack str) + builder s = NodeId (EndPointAddress (encodeUtf8 (s <> ":0"))) + mapM_ (putStrLn . Text.unpack) (lefts parsed) -- Let the user know there were errors in slaves + return (fmap (builder . Text.pack) (rights parsed)) + where + parseNodes str = + let splitted = Text.splitOn "," str + in fmap (ensurePort . validate) splitted + validate s = (s, hostAndPort (Text.unpack s)) + ensurePort (s, (Left _)) = Left ("Invalid slave address: " <> s) + ensurePort (s, (Right (_, Nothing))) = Left ("Invalid slave address. Missing port for: " <> s) + ensurePort (_, (Right ((host, Just port)))) = Right (host <> ":" <> port) + +---------------------------------- +-- Routes and middleware +---------------------------------- +routes :: Int -> String -> Database -> Scheduler.RunSchedule -> IO () +routes port folder db testsList = do + scotty port $ -- Let's declare the routes and handlers + do + middleware simpleCors + -- ^ Useful when developing the elm app using a separate elm-live server + get "/" $ do + setHeader "Content-Type" "text/html; charset=utf-8" + file (folder <> "index.html") + get "/app.js" $ do + setHeader "Content-Type" "application/javascript" + file (folder <> "app.js") + get "/main.css" $ do + setHeader "Content-Type" "text/css" + file (folder <> "main.css") + -- ^ Serves the few static files we have + get "/runs" (listRuns db) + -- ^ Gets the list of runs that have been stored. Optionally filtering by "match" + post "/runs" (createRun db) + -- ^ Creates a new run and returns the id + post "/runs/:id" (storeResults db) + -- ^ Gets the JSON blob from a wrecker run and stores them in the DB + get "/runs/rollup" (getManyRuns db) + -- ^ Returns the basic info for the passed runs, the list of pages and the general stats + get "/runs/page" (getPageStats db) + -- ^ Returns the statistics for a single or multiple runs + get "/runs/:id" (getRun db) + -- ^ Returns the basic info for the run, the list of pages and the general stats + get "/test-list" (getTestList testsList) + -- ^ Returns the list of tests with the status of the last run for them if any + post "/test-list" (scheduleTest testsList) + +---------------------------------- +-- Controllers +---------------------------------- +listRuns :: Database -> ActionM () +listRuns db = do + match <- optionalParam "match" + result <- liftAndCatchIO (fetchRuns match) + json $ object ["runs" .= result, "success" .= True] + where + fetchRuns :: Text -> IO [P.Entity Run] + fetchRuns match = runDbAction db $ findRunsByMatch match + +createRun :: Database -> ActionM () +createRun db = do + conc <- readEither <$> param "concurrency" + title <- param "title" + group <- optionalParam "groupName" + case conc of + Left err -> do + json $ object ["error" .= ("Invalid concurrency number" :: Text), "reason" .= err] + status badRequest400 + Right concurrency -> do + now <- liftAndCatchIO getCurrentTime + let run = Run title group concurrency now + newId <- liftAndCatchIO (doStoreRun run) + json $ object ["success" .= True, "id" .= newId] + status created201 + where + doStoreRun :: Run -> IO (Key Run) + doStoreRun run = runDbAction db (P.insert run) + +storeResults :: Database -> ActionM () +storeResults db = do + runId <- param "id" + jsonPayload <- body + let wreckerRun = eitherDecode jsonPayload :: Either String WreckerRun + case wreckerRun of + Left err -- In case we can't parse the json, respond with error + -> do + json $ object ["error" .= ("Invalid JSON payload" :: Text), "reason" .= err] + status badRequest400 + Right run -- Otherwise, we have the run and we can store it in the DB + -> do + liftAndCatchIO (storeStats runId run) + json $ object ["success" .= True] + status ok200 + where + storeStats :: Int -> WreckerRun -> IO () + storeStats runId run = runDbAction db $ storeRunResults (toSqlKey runId) run + +getRun :: Database -> ActionM () +getRun db = do + rId <- readEither <$> param "id" + case rId of + Right runId -> do + (stats, pages) <- liftAndCatchIO (runDbAction db $ fetchRunStats [runId]) + -- if the list is empty, errorResponse, otherwise call sendResult with the first in the list + maybe (errorResponse errorTxt) (sendResult pages) (listToMaybe stats) + Left err -> errorResponse err + where + errorTxt :: Text + errorTxt = "No query result" + -- | Transforms the SQL result into a JSON response + -- + sendResult list (run, stats) = do + json $ object ["run" .= run, "stats" .= stats, "pages" .= list] + status ok200 + -- | Responds with a JSON error + -- + errorResponse err = do + json $ object ["error" .= ("Invalid run id" :: Text), "reason" .= err] + status badRequest400 + +-- | Fetches the run stats in the datatabase. It retuns a pair: +-- The first positionin the pair is a list of (run, rollup) +-- The second position is a reverse index of pages participating in the run {page: [run id, ...]} +fetchRunStats :: + MonadIO m => [Int] -> TransactionMonad m ([(P.Entity Run, Rollup)], Map.Map Text [Key Run]) +fetchRunStats runId = do + let runKey = toSqlKey <$> runId + stats <- findRunStats runKey + list <- findPagesList runKey + runs <- findRuns (fmap fst stats) + return + ( [ (run, rollup) -- Locally doing a join by the same run key + | (sKey, rollup) <- stats + , run@(P.Entity rKey _) <- runs + , sKey == rKey + ] + , reverseIndex list) + where + reverseIndex pages = Map.fromListWith (++) [(page, [key]) | (key, page) <- pages] + +getManyRuns :: Database -> ActionM () +getManyRuns db = do + allIds <- idsOrMatch db + case allIds of + [] -> errorResponse "Invalid list of ids" + _ -> do + result <- liftAndCatchIO (runDbAction db $ fetchRunStats allIds) + sendResult result + where + sendResult (results, pages) = do + let buildObject (run, stats) = object ["run" .= run, "stats" .= stats] + json $ object ["runs" .= fmap buildObject results, "pages" .= pages] + status ok200 + -- | Responds with a JSON error + -- + errorResponse :: Text -> ActionM () + errorResponse err = do + json $ object ["reason" .= err] + status badRequest400 + +getPageStats :: Database -> ActionM () +getPageStats db = do + allIds <- idsOrMatch db + pageName <- param "name" + page <- liftAndCatchIO (fetchPageStats allIds pageName) + json page + where + fetchPageStats :: [Int] -> Text -> IO [Page] + fetchPageStats runIds pageName = + runDbAction db $ do + let runKeys = toSqlKey <$> runIds + findPageStats runKeys pageName + +getTestList :: Scheduler.RunSchedule -> ActionM () +getTestList testsList = do + tests <- liftAndCatchIO $ Scheduler.getRunSchedule testsList + json $ object ["status" .= True, "tests" .= tests] + +scheduleTest :: Scheduler.RunSchedule -> ActionM () +scheduleTest testsList = do + testTitle <- param "testTitle" + groupName <- param "groupName" + cStart <- readEither <$> param "concurrencyStart" + cEnd <- readEither <$> param "concurrencyEnd" + sSize <- readEither <$> param "stepSize" + time <- Right . readMaybe <$> param "runTime" -- Read into a Maybe, then wrap with Right + -- + -- The lazy way of checking for conversion errors + -- We "unpack" the "Right" value from each var. If any "Left" is found + -- The operation is aborted and the whole subroutine retuns "Left" + let allParams = (Scheduler.ScheduleOptions groupName) <$> cStart <*> cEnd <*> sSize <*> time + case allParams of + Left err -> handleError err + Right schedule -> do + result <- liftAndCatchIO (Scheduler.addToQueue testTitle schedule testsList) + either handleError handleSucccess result + where + handleError err = do + json $ object ["error" .= ("Invalid argument" :: Text), "reason" .= err] + status badRequest400 + handleSucccess _ = do + json $ object ["success" .= True] + status created201 + +---------------------------------- +-- Utilities +---------------------------------- +readPort :: Read port => String -> port -> IO port +readPort name defaultPort = do + taintedPort <- lookupEnv name + -- Reading the port from the ENV. Abort with an error on bad port + let varReader = maybe (error $ "Bad " <> name) id + return (maybe defaultPort (varReader . readMaybe) taintedPort) + +optionalParam :: (Monoid a, Parsable a) => LText.Text -> ActionM a +optionalParam name = param name `rescue` (\_ -> return mempty) + +parseIdList :: ActionM [Int] +parseIdList = do + longString <- optionalParam "ids" + let idsList = fmap Text.unpack (Text.splitOn "," longString) + parsedList = fmap readMaybe idsList + return (catMaybes parsedList) + +idsOrMatch :: Database -> ActionM [Int] +idsOrMatch db = do + ids <- parseIdList + case ids of + [] -> idsFromMatch + _ -> return ids + where + idsFromMatch = do + match <- param "match" + runs <- liftAndCatchIO (runDbAction db $ findRunsByMatch match) + return (fmap (fromIntegral . Sql.fromSqlKey . P.entityKey) runs) + +toSqlKey :: Int -> Key Run +toSqlKey = Sql.toSqlKey . fromIntegral diff --git a/src/Model.hs b/src/Model.hs new file mode 100644 --- /dev/null +++ b/src/Model.hs @@ -0,0 +1,367 @@ +{-# LANGUAGE EmptyDataDecls #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE QuasiQuotes #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE ConstraintKinds #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE StrictData #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# OPTIONS_GHC -fno-warn-orphans #-} + +module Model where + +import Control.Monad.IO.Class (MonadIO) +import Control.Monad.Logger + (LoggingT(..), filterLogger, runStdoutLoggingT) +import Control.Monad.Trans.Control (MonadBaseControl(..)) +import Control.Monad.Trans.Reader (ReaderT) +import qualified Control.Monad.Trans.Resource as R +import Data.Aeson hiding (Value) +import Data.Aeson.Types (Options(..), Parser) +import Data.Binary (Binary(..)) +import Data.Char (toLower) +import qualified Data.Map as Map +import Data.Map (Map) +import Data.Monoid ((<>)) +import Data.Pool (Pool) +import Data.Ratio (numerator) +import Data.Text (Text) +import Data.Time.Clock (UTCTime) +import Data.Typeable (Typeable) +import Database.Esqueleto +import Database.Persist (Key(..)) +import qualified Database.Persist as P +import Database.Persist.Postgresql (withPostgresqlPool) +import qualified Database.Persist.Sql as Sql +import Database.Persist.Sqlite (withSqlitePool) +import Database.Persist.TH +import Database.PostgreSQL.Simple.Internal + (ConnectInfo, postgreSQLConnectionString) +import GHC.Generics hiding (from) + +------------------------------------- +-- Schema Definition +------------------------------------- +share + [mkPersist sqlSettings, mkMigrate "migrateAll"] + [persistLowerCase| + Run + match Text + groupName Text + concurrency Int + created UTCTime + deriving Eq Show Generic + + Rollup + runId RunId Maybe + hits Int + maxTime Double + meanTime Double + minTime Double + totalTime Double + variance Double + successHits Int + userErrorHits Int + serverErrorHits Int + failedHits Int + quantile95 Double + deriving Eq Show Generic Typeable + + Page + runId RunId Maybe + url Text Maybe + hits Int + maxTime Double + meanTime Double + minTime Double + totalTime Double + variance Double + successHits Int + userErrorHits Int + serverErrorHits Int + failedHits Int + quantile95 Double + deriving Eq Show Generic Typeable +|] + +---------------------------------- +-- Convenient type aliases +---------------------------------- +type DbMonad m = (MonadBaseControl IO m, MonadIO m) + +type TransactionMonad m a = ReaderT SqlBackend (R.ResourceT m) a + +type Database = Pool SqlBackend + +type DbAction m = ReaderT SqlBackend m + +data DbBackend + = Postgresql ConnectInfo + | Sqlite + +data LogLevel + = Silent + | Debug + deriving (Show, Read, Eq) + +data WreckerRun = WreckerRun + { rollup :: Rollup + , pages :: [Page] + } deriving (Eq, Show, Generic, Typeable) + +instance Binary WreckerRun + +instance Binary Rollup + +instance Binary Page + +instance Binary (Key Run) where + get = fmap Sql.toSqlKey Data.Binary.get + put = put . Sql.fromSqlKey + +instance Binary (Key Page) where + get = fmap Sql.toSqlKey Data.Binary.get + put = put . Sql.fromSqlKey + +------------------------------------- +-- Functions for working with the DB +------------------------------------- +withDb :: DbMonad m => LogLevel -> DbBackend -> (Database -> LoggingT m a) -> m a +withDb logLevel dbType = + case logLevel of + Debug -> runStdoutLoggingT . withPool 10 + Silent -> runStdoutLoggingT . filterLogger (\_ _ -> False) . withPool 10 + where + withPool size = + case dbType of + Sqlite -> withSqlitePool "wrecker.db" size + Postgresql info -> withPostgresqlPool (postgreSQLConnectionString info) size + +runDbAction :: DbMonad m => Database -> TransactionMonad m a -> m a +runDbAction db = R.runResourceT . flip Sql.runSqlPool db + +runMigrations :: DbMonad m => Database -> m () +runMigrations db = runDbAction db (Sql.runMigration migrateAll) + +------------------------------------- +-- Finder queries +------------------------------------- +findRunsByMatch :: MonadIO m => Text -> SqlReadT m [Entity Run] +findRunsByMatch match = + let matchWithExpanse = val ("%" <> match <> "%") + in select $ + from $ \r -> do + where_ (r ^. RunMatch `like` matchWithExpanse) + return r + +findRuns :: MonadIO m => [Key Run] -> SqlReadT m [Entity Run] +findRuns keys = + select $ + from $ \r -> do + where_ (r ^. RunId `in_` valList keys) + orderBy [asc (r ^. RunCreated)] + return r + +findPagesList :: MonadIO m => [Key Run] -> SqlReadT m [(Key Run, Text)] +findPagesList runIds = do + rows <- + select $ + distinct $ + from $ \p -> do + where_ (p ^. PageRunId `in_` justList (valList runIds)) + return + ( coalesceDefault [p ^. PageRunId] (val (toKey 0)) + , coalesceDefault [p ^. PageUrl] (val "")) + return (fmap (\(k, p) -> (unValue k, unValue p)) rows) + +findRunStats :: MonadIO m => [Key Run] -> SqlReadT m [(Key Run, Rollup)] +findRunStats runIds = do + rows <- + select $ + from $ \rollup -> do + where_ (rollup ^. RollupRunId `in_` justList (valList (runIds))) + groupBy (rollup ^. RollupRunId) + orderBy [asc (rollup ^. RollupRunId)] + return + ( coalesceDefault [rollup ^. RollupRunId] (val (toKey 0)) + , coalesceDefault [sum_ (rollup ^. RollupHits)] (val 0) + , coalesceDefault [max_ (rollup ^. RollupMaxTime)] (val 0) + , coalesceDefault [avg_ (rollup ^. RollupMeanTime)] (val 0) + , coalesceDefault [min_ (rollup ^. RollupMinTime)] (val 0) + , coalesceDefault [sum_ (rollup ^. RollupTotalTime)] (val 0) + , coalesceDefault [avg_ (rollup ^. RollupVariance)] (val 0) + , coalesceDefault [sum_ (rollup ^. RollupSuccessHits)] (val 0) + , coalesceDefault [sum_ (rollup ^. RollupUserErrorHits)] (val 0) + , coalesceDefault [sum_ (rollup ^. RollupServerErrorHits)] (val 0) + , coalesceDefault [sum_ (rollup ^. RollupFailedHits)] (val 0) + , coalesceDefault [avg_ (rollup ^. RollupQuantile95)] (val 0)) + return (fmap buildRollup rows) + where + buildRollup fields = set11Fields (Rollup Nothing) fields + +findPageStats :: MonadIO m => [Key Run] -> Text -> SqlReadT m [Page] +findPageStats runIds url = do + rows <- + select $ + from $ \page -> do + where_ (page ^. PageRunId `in_` justList (valList runIds)) + where_ (page ^. PageUrl ==. just (val url)) + groupBy (page ^. PageRunId) + return + ( coalesceDefault [page ^. PageRunId] (val (toKey 0)) + , coalesceDefault [sum_ (page ^. PageHits)] (val 0) + , coalesceDefault [max_ (page ^. PageMaxTime)] (val 0) + , coalesceDefault [avg_ (page ^. PageMeanTime)] (val 0) + , coalesceDefault [min_ (page ^. PageMinTime)] (val 0) + , coalesceDefault [sum_ (page ^. PageTotalTime)] (val 0) + , coalesceDefault [avg_ (page ^. PageVariance)] (val 0) + , coalesceDefault [sum_ (page ^. PageSuccessHits)] (val 0) + , coalesceDefault [sum_ (page ^. PageUserErrorHits)] (val 0) + , coalesceDefault [sum_ (page ^. PageServerErrorHits)] (val 0) + , coalesceDefault [sum_ (page ^. PageFailedHits)] (val 0) + , coalesceDefault [avg_ (page ^. PageQuantile95)] (val 0)) + return (fmap buildPage rows) + where + buildPage fields = setRunId . buildIntermediate $ fields + buildIntermediate fields = set11Fields (Page Nothing (Just url)) fields + setRunId (rId, page) = page {pageRunId = Just rId} + +toKey :: Int -> Key Run +toKey key = Sql.toSqlKey . fromIntegral $ key + +fromKey :: Key Run -> Int +fromKey = fromIntegral . Sql.fromSqlKey + +set11Fields :: + (Int -> t7 -> t6 -> t5 -> t4 -> t3 -> Int -> Int -> Int -> Int -> t2 -> t1) + -> ( Value t + , Value Rational + , Value t7 + , Value t6 + , Value t5 + , Value t4 + , Value t3 + , Value Rational + , Value Rational + , Value Rational + , Value Rational + , Value t2) + -> (t, t1) +set11Fields func (Value key, Value a, Value b, Value c, Value d, Value e, Value f, Value g, Value h, Value i, Value j, Value k) = + ( key + , func -- Some fields need casting + (castInt a) + b + c + d + e + f + (castInt g) + (castInt h) + (castInt i) + (castInt j) + k) + where + castInt :: Rational -> Int + castInt = fromIntegral . numerator + +storeRunResults :: DbMonad m => Key Run -> WreckerRun -> DbAction m () +storeRunResults runKey WreckerRun {..} = do + let fullRollup = rollup {rollupRunId = Just runKey} + fullPage page = page {pageRunId = Just runKey} + -- + -- We insert both the rollup and the pages in the same transaction + _ <- P.insert fullRollup + mapM_ (P.insert . fullPage) pages + +------------------------------------- +-- JSON conversions +------------------------------------- +instance ToJSON Run where + toJSON = genericToJSON (defaultOptions {fieldLabelModifier = lcFirst . drop 3}) + +instance FromJSON Run where + parseJSON = genericParseJSON (defaultOptions {fieldLabelModifier = lcFirst . drop 3}) + +instance ToJSON Rollup where + toJSON = genericToJSON (defaultOptions {fieldLabelModifier = lcFirst . drop 6}) + +instance ToJSON Page where + toJSON = genericToJSON (defaultOptions {fieldLabelModifier = lcFirst . drop 4}) + +lcFirst :: String -> String +lcFirst "" = "" +lcFirst (f:rest) = toLower f : rest + +instance FromJSON Rollup where + parseJSON = + withObject "Rollup" $ \o -> do + let rollupRunId = Nothing + rollup <- o .: "rollup" + rollupMaxTime <- rollup .: "max" + rollupMeanTime <- rollup .: "mean" + rollupMinTime <- rollup .: "min" + rollupHits <- rollup .: "count" + rollupVariance <- rollup .: "variance" + rollupTotalTime <- rollup .: "total" + rollupQuantile95 <- rollup .: "quantile95" + successData <- o .: "2xx" + rollupSuccessHits <- successData .: "count" + userErrorData <- o .: "4xx" + rollupUserErrorHits <- userErrorData .: "count" + serverErrorData <- o .: "5xx" + rollupServerErrorHits <- serverErrorData .: "count" + failureData <- o .: "failed" + rollupFailedHits <- failureData .: "count" + return Rollup {..} + +instance FromJSON Page where + parseJSON = + withObject "Page" $ \o -> do + let pageRunId = Nothing + let pageUrl = Nothing + rollup <- o .: "rollup" + pageMaxTime <- rollup .: "max" + pageMeanTime <- rollup .: "mean" + pageMinTime <- rollup .: "min" + pageHits <- rollup .: "count" + pageVariance <- rollup .: "variance" + pageTotalTime <- rollup .: "total" + pageQuantile95 <- rollup .: "quantile95" + successData <- o .: "2xx" + pageSuccessHits <- successData .: "count" + userErrorData <- o .: "4xx" + pageUserErrorHits <- userErrorData .: "count" + serverErrorData <- o .: "5xx" + pageServerErrorHits <- serverErrorData .: "count" + failureData <- o .: "failed" + pageFailedHits <- failureData .: "count" + return Page {..} + +instance FromJSON WreckerRun where + parseJSON = + withObject "WreckerRun" $ \o -> do + rollup <- o .: "rollup" + allPages <- o .: "per-request" :: Parser (Map Text Page) + -- Now we convert the pages dictionary into a list of 'Page' + -- by traversing all the structure with a accumulator function + let pages = + Map.foldlWithKey' + (\list url page -> page {pageUrl = Just url} : list) + [] -- The initial accumulator value + allPages + return WreckerRun {..} + +instance ToJSON a => ToJSON (Entity a) where + toJSON (Entity k run) = + let (Object encoded) = toJSON run -- first encode the run + (Object eKey) = object ["id" .= k] -- convert the key to json + in Object (encoded <> eKey) -- Add both parts together as an object diff --git a/src/Recorder.hs b/src/Recorder.hs new file mode 100644 --- /dev/null +++ b/src/Recorder.hs @@ -0,0 +1,25 @@ +{-# LANGUAGE RecordWildCards, NamedFieldPuns #-} + +module Recorder where + +import Data.Either (lefts, rights) +import Data.Text (Text) +import Data.Time.Clock (getCurrentTime) +import Database.Persist (insert) +import Invoker (Command(..), Concurrency(..)) +import Model + (Database, Key, Run(..), WreckerRun(..), runDbAction, + storeRunResults) + +createRun :: Database -> Text -> Command -> IO (Key Run) +createRun db gName (Command title _ (Concurrency concurrency)) = do + now <- getCurrentTime + runDbAction db $ insert $ Run title gName concurrency now + +record :: Database -> Key Run -> [Either String WreckerRun] -> IO () +record db runId result = do + case lefts result of + [] -> runDbAction db $ mapM_ (storeRunResults runId) (rights result) + errors -> do + putStrLn "Got an error when processing wrecker results:" + print errors diff --git a/src/Scheduler.hs b/src/Scheduler.hs new file mode 100644 --- /dev/null +++ b/src/Scheduler.hs @@ -0,0 +1,274 @@ +{-# LANGUAGE RecordWildCards, NamedFieldPuns #-} +{-# LANGUAGE TemplateHaskell #-} + +module Scheduler + ( RunSchedule + , Config(..) + , ScheduleOptions(..) + , emptyRunSchedule + , runScheduler + , getRunSchedule + , addToQueue + , __remoteTable -- This is created by the call to remotable using TemplateHaskell + ) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (Async, async, pollSTM) +import qualified Control.Concurrent.STM as STM +import Control.Monad (guard) +import Data.Aeson +import Data.List.NonEmpty (NonEmpty(..), (!!), fromList) +import qualified Data.Map as Map +import Data.Map (Map) +import qualified Data.Maybe as Maybe +import Data.Maybe (fromMaybe) +import Data.Monoid ((<>)) +import Data.Text (Text, pack) +import Data.Time.Clock (getCurrentTime) +import Data.Time.ISO8601 (formatISO8601) +import qualified Invoker as Wrecker +import Model (Database, WreckerRun) +import Prelude hiding ((!!)) +import qualified Recorder + +import Control.Distributed.Process + (NodeId, Process, getSelfNode, liftIO) +import Control.Distributed.Process.Async + (AsyncResult(..), AsyncTask, asyncLinked, remoteTask, wait) +import qualified Control.Distributed.Process.Async as DAsync +import Control.Distributed.Process.Closure +import Control.Distributed.Process.Internal.Types + (LocalProcess, runLocalProcess) + +-- | Represents each of the stages a run can be at +data RunStatus + = Running (Maybe (Async Wrecker.Result)) + | Done + | Scheduled ScheduleOptions + | None + +-- | Avilable options for executing a run +data ScheduleOptions = ScheduleOptions + { gName :: Text -- ^ The run group name, used to tag many similar runs with different concurrency + , cStart :: Int -- ^ Concurrency start + , cEnd :: Int -- ^ Concurrency end + , sStep :: Int -- ^ Step size. This is the increment to use to get from start concurrency to end + , time :: Maybe Int -- ^ The amount of time to spend on each of the individual runs + } deriving (Show) + +-- | A transactional memory variable containing the status fo the current run and the schedule for future runs +type RunSchedule = STM.TVar (Map Text RunStatus) + +-- | Scheduler configuration +data Config = Config + { db :: Database -- ^ A database instance, used to store the run results + , testsList :: RunSchedule -- ^ The varibale to store the current run and schedule for future runs + , localProcess :: LocalProcess -- ^ This node description + , slaves :: [NodeId] -- ^ A list of worker slaves used to distribute runs accross them + } + +-- | Wraps the wrecker function into the Process monad +-- This is used for remotely executing wrecker in other nodes +wrecker :: Wrecker.Command -> Process (Either String WreckerRun) +wrecker = liftIO . Wrecker.wrecker + +-- Make the wrecker function ready to be called over the wire on slave nodes +remotable ['wrecker] + +emptyRunSchedule :: IO RunSchedule +emptyRunSchedule = do + list <- Wrecker.listGroups + STM.newTVarIO (Map.fromList [(pack t, None) | t <- list]) + +getRunSchedule :: STM.TVar a -> IO a +getRunSchedule testsList = STM.atomically (STM.readTVar testsList) + +-- | A blocking functions that refreshes the schedule state each 5 seconds. +-- This functions is responsible for executing the run jobs as they become available +runScheduler :: Config -> IO () +runScheduler config@(Config {..}) = do + threadDelay $ 5 * 1000 * 1000 -- wait 5 seconds + maybeSchedule <- + STM.atomically $ do + tests <- STM.readTVar testsList + cleaned <- cleanupFinished tests + STM.writeTVar testsList cleaned + pickOneToRun + process maybeSchedule + runScheduler (Config {..}) + where + cleanupFinished tests = do + let running = filter isRunning (Map.toList tests) + polled <- mapM pollTest running + let updated = Map.fromList (Maybe.mapMaybe markAsDone polled) + return (Map.union updated tests) + -- + -- | In the pair (title, job) poll the job for a status and return it + pollTest (title, Running (Just stat)) = do + res <- pollSTM stat + return (title, res) + pollTest (title, _) = return (title, Nothing) + -- + -- | In the pair (title, status) if status is not Nothing, replace + -- the status with "Done", otherwise returns Nothing + markAsDone (_, Nothing) = Nothing + markAsDone (title, _) = Just (title, Done) + pickOneToRun = do + tests <- STM.readTVar testsList + return (take 1 $ filter isScheduled (Map.toList tests)) + -- + -- | Selects elements having Scheduled as status + isScheduled element = + case element of + (_, Scheduled _) -> True + _ -> False + -- + -- | Selects elements having Running as status + isRunning element = + case element of + (_, Running _) -> True + _ -> False + -- + -- | If any test was selected for running, then run it + process maybeSchedule = + case maybeSchedule of + (name, Scheduled schedule):_ -> tryRunningNow config name schedule + _ -> return () + +-- | Adds a new run job to the queue. This function will wither queue the job +-- or return a reason for not being able to do so. +addToQueue :: Text -> ScheduleOptions -> RunSchedule -> IO (Either Text Bool) +addToQueue name schedule testsList = + STM.atomically $ do + tests <- STM.readTVar testsList + case canAddToSchedule name tests of + notPossible@(Left _) -> return notPossible + isPossible -> do + updateStatus testsList name (Scheduled schedule) + return isPossible + +-- | Checks wheter or not it is possible to executing the give run right now and +-- if possible, it executes it immediately. +tryRunningNow :: Config -> Text -> ScheduleOptions -> IO () +tryRunningNow config@(Config {..}) name ScheduleOptions {..} = do + canRun <- STM.atomically (markAsRunning testsList name) + guard canRun + now <- getCurrentTime + let steps = Wrecker.Concurrency <$> createSteps + fTime = pack . formatISO8601 $ now + groupName = gName <> " - " <> fTime + runner = escalateWithRecorder config name time groupName + job <- async (runLocalProcess localProcess $ runner steps) + STM.atomically (updateStatus testsList name (Running $ Just job)) + where + createSteps = + if cStart == 1 || cStart == 0 + then tail [0,sStep .. cEnd] + else [cStart,(cStart + sStep) .. cEnd] + +-- | A helper function used to execute a batch of run from a starting to an ending concurrency level, +-- this helper function will try to execute wrecker remotely if the right conditions are given, and +-- will also record the results in the database. +escalateWithRecorder :: + Config -- ^ The scheduler configuration + -> Text -- ^ The test name to execute + -> Maybe Int -- ^ Ampunt of seconds to spend on each test client + -> Text -- ^ The group name to assign to this test run + -> ([Wrecker.Concurrency] -> Process Wrecker.Result) +escalateWithRecorder Config {db, slaves} name time groupName = do + let secs = Wrecker.Seconds (fromMaybe 10 time) -- Run for 10 seconds if no preference was given + builder = Wrecker.Command name secs + keyGen command = liftIO (Recorder.createRun db groupName command) + executer = remoteWrecker slaves + recorder key result = liftIO (Recorder.record db key result) + Wrecker.escalate builder executer keyGen recorder + +-- | Executes a wrecker command in the remote sleves, if any are provided. The local node will always +-- participate in executing some of the load. If no slaves are available, the local node executes the whole run. +remoteWrecker :: [NodeId] -> Wrecker.Command -> Process [Either String WreckerRun] +remoteWrecker slaves command@(Wrecker.Command {concurrency}) = do + thisNode <- getSelfNode + tasks <- mapM asyncLinked (divideWork (fromList $ thisNode : slaves)) -- Execute each task remotely and async + mapM waitFor tasks + where + divideWork :: NonEmpty NodeId -> [AsyncTask (Either String WreckerRun)] + -- Try to comfortably accommodate as much work on a single node as possible + -- and return a list of async tasks that should be executed on the network + divideWork nodes@(selfNode :| _) = + let Wrecker.Concurrency conc_ = concurrency + conc = fromIntegral conc_ + in if length slaves == 0 || conc < 2000 + then [buildTask command selfNode] + else let (first:rest) = + [ nodes !! (i - 1) -- Return the node at position i + | i <- [1 .. length nodes] -- Select one more node + , (conc / fromIntegral i) >= 1000 -- If there are at least 1000 threads to run on it + ] + -- Calculate the number of threads per task to execute + total = fromIntegral (length (first : rest)) -- Get the total available workers + fairDivision = fromRational (conc / total) :: Double -- Divide the number of threads equally + avgConcurrency = floor fairDivision -- Round to integer + remainderConcurrency = + avgConcurrency + -- One of the workers need to get the remainder threads + (ceiling (total * ((conc / total) - fromIntegral avgConcurrency))) + setConcurrency c = command {Wrecker.concurrency = Wrecker.Concurrency c} + -- + -- Build the async tasks that should be executed + firstTask = buildTask (setConcurrency remainderConcurrency) first -- First task gets teh remainder + restTasks = fmap (buildTask (setConcurrency avgConcurrency)) rest -- But the rest get the same amount of threads + in firstTask : restTasks + +-- | Builds the async task for the givend node and command +-- the "task" is executing a wrecker command on a remote machine +buildTask :: Wrecker.Command -> NodeId -> AsyncTask (Either String WreckerRun) +buildTask command node = remoteTask $(functionTDict 'wrecker) node ($(mkClosure 'wrecker) command) + +-- | Waits for a remote Asyn job to finish, and returns its result once it dies or it's done. +waitFor :: DAsync.Async (Either String WreckerRun) -> Process (Either String WreckerRun) +waitFor task = do + asyncResult <- wait task -- Wait for the task to be done + case asyncResult of + AsyncDone res -> return res + AsyncFailed reason -> return (Left (show reason)) + AsyncLinkFailed reason -> return (Left (show reason)) + _ -> return (Left "Error spawning async wrecker task") + +-- | Modifies the test list if it is valid to insert the new schedule +markAsRunning :: RunSchedule -> Text -> STM.STM Bool +markAsRunning testsList name = do + tests <- STM.readTVar testsList + if isRunnable name tests + then do + updateStatus testsList name (Running Nothing) + return True + else return False + +-- | Updates the status of a single run in the give run schedule +updateStatus :: RunSchedule -> Text -> RunStatus -> STM.STM () +updateStatus testsList name s = STM.modifyTVar' testsList (Map.insert name s) + +-- | Returns whehter it is ok to schedule or run the given test +-- Left is returned when it is not possible to run the test. +-- Otherwise the test is ok to be set for scheduling +canAddToSchedule :: Text -> Map Text RunStatus -> Either Text Bool +canAddToSchedule name list = + case Map.lookup name list of + Nothing -> Left "Test name does not exist" + Just (Running _) -> Left "This test is still running" + Just _ -> Right True + +-- | Checks whether or not a particular run (given its name) can be added to the queue. +isRunnable :: Text -> Map Text RunStatus -> Bool +isRunnable name list = + case Map.lookup name list of + Just (Scheduled _) -> True + _ -> False + +---------------------------------- +-- JSON Conversions +---------------------------------- +instance ToJSON RunStatus where + toJSON (Running _) = String "running" + toJSON Done = String "done" + toJSON (Scheduled _) = String "scheduled" + toJSON None = String "none" diff --git a/wrecker-ui.cabal b/wrecker-ui.cabal new file mode 100644 --- /dev/null +++ b/wrecker-ui.cabal @@ -0,0 +1,63 @@ +name: wrecker-ui +version: 2.3.0 +synopsis: A web interface for Wrecker, the HTTP Performance Benchmarker +description: + 'wrecker-ui' is a web based interface to visualize performance tests built using the wrecker library and schedule test runs. +cabal-version: >= 1.10 +build-type: Simple +license: BSD3 +license-file: LICENSE +category: Web +author: José Lorenzo Rodríguez +maintainer: lorenzo@seatgeek.com + +executable wrecker-ui + default-language: Haskell2010 + hs-source-dirs: src + main-is: Main.hs + other-modules: Model + , Invoker + , Scheduler + , Recorder + ghc-options: -Wall -threaded -O2 -rtsopts -with-rtsopts=-N + default-extensions: OverloadedStrings + build-depends: base >= 4 && < 5 + , scotty + , text + , time + , aeson + , containers + , http-types + , wai-cors + , directory + , persistent + , persistent-template + , persistent-sqlite + , persistent-postgresql + , esqueleto + , monad-control + , monad-logger + , resourcet + , transformers + , resource-pool + , postgresql-simple-url + , postgresql-simple + , process + , temporary + , bytestring + , iso8601-time + , stm + , async + , distributed-process + , network-transport + , network-transport-tcp + , distributed-process-simplelocalnet + , distributed-process-async + , distributed-static + , binary + , mtl + , HostAndPort + +source-repository head + type: git + location: https://github.com/seatgeek/wrecker-ui