Prototype a timeseries framework (WIP)

cabal.project

@@ -31,6 +31,7 @@ packages:
   trace-dispatcher
   trace-resources
   trace-forward
+  timeseries-io
 
 -- Needed when cross compiling
 extra-packages: alex

timeseries-io/CHANGELOG.md

@@ -0,0 +1,5 @@
+# Revision history for timeseries-io
+
+## 0.1.0.0 -- YYYY-mm-dd
+
+* First version. Released on an unsuspecting world.

timeseries-io/app/Main.hs

@@ -0,0 +1,67 @@
+{-# OPTIONS_GHC -Wno-deprecations #-}
+{-# OPTIONS_GHC -Wno-name-shadowing #-}
+
+module Main where
+
+import           Cardano.Timeseries.Import.PlainCBOR
+import           Cardano.Timeseries.Query (interp)
+import           Cardano.Timeseries.Query.Parser (expr)
+import           Cardano.Timeseries.Query.Value (Error, Value)
+import           Cardano.Timeseries.Store
+import           Cardano.Timeseries.Store.Flat (Flat, Point (instant, name))
+import           Cardano.Timeseries.Store.Parser (points)
+
+import           Control.Monad (forever)
+import           Control.Monad.Except (runExceptT)
+import           Control.Monad.State.Strict (evalState)
+import           Data.Attoparsec (skipMany)
+import           Data.Attoparsec.Text (decimal, endOfInput, parseOnly, space)
+import           Data.Foldable (for_, traverse_)
+import           Data.Text (pack)
+import           GHC.List (foldl')
+import           System.Exit (die)
+import           System.IO (hFlush, stdout)
+import Cardano.Timeseries.Store.Tree (fromFlat)
+import Cardano.Logging.Resources (readResourceStats)
+import Cardano.Logging (forHuman)
+import Cardano.Logging.Resources (ResourceStats)
+import Cardano.Logging.Resources (Resources(..))
+
+snapshotsFile :: String
+snapshotsFile = "6nodes_4hours_1mininterval.cbor"
+
+printStore :: Flat Double -> IO ()
+printStore = traverse_ print
+
+printQueryResult :: Either Error Value -> IO ()
+printQueryResult (Left err) = putStrLn ("Error: " <> err)
+printQueryResult (Right ok) = print ok
+
+printStats :: ResourceStats -> IO ()
+printStats stats =
+  putStrLn $ "Alloc: " <> show ((fromIntegral (rAlloc stats) :: Double) / 1024 / 1024) <> "MB\n"
+          <> "Live: " <> show ((fromIntegral (rLive stats) :: Double) / 1024 / 1024) <> "MB\n"
+          <> "Heap: " <> show ((fromIntegral (rHeap stats) :: Double) / 1024 / 1024) <> "MB\n"
+          <> "RSS: " <> show ((fromIntegral (rRSS stats) :: Double) / 1024 / 1024) <> "MB"
+
+interactive :: Store s Double => s -> IO ()
+interactive store = forever $ do
+ Just stats <- readResourceStats
+ putStrLn "----------"
+ printStats stats
+ putStrLn $ "Number of store entries: " <> show (count store)
+ putStrLn "----------"
+ putStr "> "
+ hFlush stdout
+ queryString <- getLine
+ case parseOnly (expr <* skipMany space <* endOfInput) (pack queryString) of
+   Left err -> putStrLn err
+   Right query -> do
+     putStrLn ("Expr: " <> show query)
+     printQueryResult (evalState (runExceptT $ interp store mempty query 0) 0)
+
+main :: IO ()
+main = do
+  content <- readFileSnapshots snapshotsFile
+  let store = snapshotsToFlatStore content
+  interactive (fromFlat store)

timeseries-io/bench/Bench.hs

@@ -0,0 +1,62 @@
+{-# OPTIONS_GHC -Wno-name-shadowing #-}
+{-# OPTIONS_GHC -Wno-incomplete-uni-patterns #-}
+{-# OPTIONS_GHC -Wno-deprecations #-}
+import           Cardano.Logging (forHuman)
+import           Cardano.Logging.Resources (ResourceStats, Resources (..), readResourceStats)
+import           Cardano.Timeseries.Import.PlainCBOR
+import           Cardano.Timeseries.Query (interp)
+import           Cardano.Timeseries.Query.Expr (Expr)
+import           Cardano.Timeseries.Query.Parser (expr)
+import           Cardano.Timeseries.Query.Value (Error, Value)
+import           Cardano.Timeseries.Store
+import           Cardano.Timeseries.Store.Flat (Flat, Point (instant, name))
+import           Cardano.Timeseries.Store.Parser (points)
+import           Cardano.Timeseries.Store.Tree (fromFlat)
+
+import           Control.Monad (forever)
+import           Control.Monad.Except (runExceptT)
+import           Control.Monad.State.Strict (evalState)
+import           Data.Attoparsec (skipMany)
+import           Data.Attoparsec.Text (decimal, endOfInput, parseOnly, space)
+import           Data.Either (fromRight)
+import           Data.Foldable (for_, traverse_)
+import           Data.Text (pack)
+import           GHC.List (foldl')
+import           System.Exit (die)
+import           System.IO (hFlush, stdout)
+
+import           Criterion.Main
+
+-- Given a snapshots file
+-- Given a query string
+-- Bench mark evaluation of the query
+
+snapshotsFile :: String
+snapshotsFile = "data/6nodes_4hours_1mininterval.cbor"
+
+query :: String
+query = "\
+  \let start = milliseconds 1762173870000 in \
+  \let period = minutes 10 in \
+  \let F = range Forge_forged_counter \
+  \(fast_forward epoch start) \
+  \(fast_forward epoch (add_duration start period)) in \
+  \increase F"
+
+action :: Store s Double => (s, Expr) -> Value
+action (store, query) =
+  let Right !x = evalState (runExceptT $ interp store mempty query 0) 0 in x
+
+main :: IO ()
+main = do
+  content <- readFileSnapshots snapshotsFile
+  let flatStore = snapshotsToFlatStore content
+  let treeStore = fromFlat flatStore
+  case parseOnly (expr <* skipMany space <* endOfInput) (pack query) of
+    Left err -> putStrLn err
+    Right !query -> defaultMain
+      [
+        bench "flat" $ nf action (flatStore, query),
+        bench "tree" $ nf action (treeStore, query)
+      ]
+

timeseries-io/doc.txt

@@ -0,0 +1,19 @@
+An instant of type `a` is the following data:
+— A set of labels
+— A timestamp
+— An element of `a`
+The set of labels identifies which timeseries the instant belongs to.
+We sometimes call this set of labels a "series identifier".
+
+An instant vector of type `a` is a set of `instant`s of type `a` where every two elements of the set have distinct:
+— Set of labels
+
+A timeseries of type `a` is:
+— A set of labels
+— a set of pairs of:
+  — A timestamp
+  — An element of `a`
+
+A timeseries vector of type `a` is a set of timeseries where every two elements of the set have distinct:
+— Set of labels
+

timeseries-io/elab.txt

@@ -0,0 +1,108 @@
+// WIP: In case we want some ambiguity in the surface language
+
+fst t
+snd t
+(t, t)
+t == t
+t < t
+t <= t
+t > t
+t >= t
+t + t
+t - t
+t * t
+t / t
+not a
+a && b
+a || b
+\x -> t
+minutes n
+hours n
+epoch
+now
+rewind t d
+fast_forward t d
+toScalar t
+s[t; t]
+s[t; t; t]
+v{ls}
+max t
+avg t
+filter t t
+t ⊗ t
+t t
+abs t : Scalar
+increase t
+rate t
+avg_over_time t
+sum_over_time t
+quantile_over_time t t
+unless t t
+quantile-by ls t t
+
+Γ |- t ~> t' ∈ (A, B)
+------------------------
+Γ |- fst t ~> fst t' ∈ A
+
+
+Γ |- t ~> t' ∈ (A, B)
+-------------------
+Γ |- snd t ~> snd t' ∈ B
+
+
+Γ |- a ~> a' ∈ A
+Γ |- b ~> b' ∈ B
+--------------------------------
+Γ |- (a, b) ~> (a', b') ∈ (A, B)
+
+
+Γ |- a₀ ~> a₀' ∈ A
+Γ |- a₁ ~> a₁' ∈ B
+Γ |- A = B = Scalar
+------------------------------------
+Γ |- a₀ == a₁ ~> eq_scalar a₀' a₁' ∈ Scalar
+
+
+Γ |- a₀ ~> a₀' ∈ A
+Γ |- a₁ ~> a₁' ∈ B
+Γ |- A = B = Bool
+---------------------------------------
+Γ |- a₀ == a₁ ~> eq_bool a₀' a₁' ∈ Bool
+
+
+Γ |- a₀ ~> a₀' ∈ Scalar
+Γ |- a₁ ~> a₁' ∈ Scalar
+-----------------------------------------
+Γ |- a₀ < a₁ ~> lt_scalar a₀' a₁' ∈ Scalar
+
+
+Γ |- a₀ ~> a₀' ∈ InstantVector Scalar
+Γ |- a₁ ~> a₁' ∈ Scalar
+-----------------------------------------------------------------------
+Γ |- a₀ < a₁ ~> lt_instant_vector_scalar a₀' a₁' ∈ InstantVector Scalar
+
+
+Γ |- a₀ ~> a₀' ∈ Scalar
+Γ |- a₁ ~> a₁' ∈ Scalar
+----------------------------------------
+Γ |- a₀ + a₁ ~> add_scalar a₀ a₁ ∈ Scalar
+
+
+Γ |- a₀ ~> a₀' ∈ InstantVector Scalar
+Γ |- a₁ ~> a₁' ∈ InstantVector Scalar
+----------------------------------------------------------------------
+Γ |- a₀ + a₁ ~> add_instant_vector_scalar a₀ a₁ ∈ InstantVector Scalar
+
+
+Γ |- v ∋ Timestamp -> InstantVector Scalar
+Γ |- a ∋ Timestamp
+Γ |- b ∋ Timestamp
+------------------------------------------
+Γ |- v[a, b] ∈ RangeVector Scalar
+
+Γ |- v ∋ Timestamp -> InstantVector Scalar
+Γ |- a ∋ Timestamp
+Γ |- b ∋ Timestamp
+Γ |- d ∋ Duration
+------------------------------------------
+Γ |- v[a, b, d] ∈ RangeVector Scalar

timeseries-io/src/Cardano/Timeseries/Domain/Identifier.hs

@@ -0,0 +1,6 @@
+module Cardano.Timeseries.Domain.Identifier(Identifier(..)) where
+
+-- | Identifiers come in two sorts: Userspace and Machine-generated.
+-- | The first kind comes from user-typed expressions.
+-- | The second kind is used for automatic code-generation for hygienic scoping (i.e. to avoid unintentional variable capture)
+data Identifier = User String | Machine Int deriving (Show, Ord, Eq)

timeseries-io/src/Cardano/Timeseries/Domain/Instant.hs

@@ -0,0 +1,38 @@
+module Cardano.Timeseries.Domain.Instant(Instant(..), InstantVector, mostRecent, share, toVector, prettyInstant, prettyInstantVector) where
+
+import           Cardano.Timeseries.Domain.Types (SeriesIdentifier, Timestamp)
+
+import           Control.DeepSeq (NFData)
+import qualified Data.Set as Set
+import           Data.Text (Text, intercalate, pack)
+import           Data.Vector
+import           GHC.Generics (Generic)
+
+-- | One datapoint in a series.
+data Instant a = Instant {
+  labels :: SeriesIdentifier,
+  timestamp :: Timestamp,
+  value :: a
+} deriving (Show, Eq, Functor, Foldable, Traversable, Generic)
+
+instance NFData a => NFData (Instant a)
+
+-- | Do the instant vectors share a series?
+share :: Instant a -> Instant b -> Bool
+share a b = labels a == labels b
+
+-- | Datapoints from different series. The vector must not contain datapoints sharing a series.
+type InstantVector a = [Instant a]
+
+mostRecent :: Instant a -> Instant a -> Instant a
+mostRecent u v = if timestamp u < timestamp v then v else u
+
+toVector :: InstantVector Double -> Vector Double
+toVector = fromList . fmap value
+
+prettyInstant :: Show a => Instant a -> Text
+prettyInstant (Instant ls t v) =
+  pack (show (Set.toList ls)) <> " " <> pack (show t) <> " " <> pack (show v)
+
+prettyInstantVector :: Show a => InstantVector a -> Text
+prettyInstantVector = intercalate "\n" . fmap prettyInstant

timeseries-io/src/Cardano/Timeseries/Domain/Interval.hs

@@ -0,0 +1,14 @@
+module Cardano.Timeseries.Domain.Interval(Interval(..), length) where
+
+import           Cardano.Timeseries.Domain.Types (Timestamp)
+
+import           Prelude hiding (length)
+
+-- | A time interval. Assumption: `start` ≤ `end`
+data Interval = Interval {
+  start :: Timestamp,
+  end :: Timestamp
+} deriving (Show, Eq)
+
+length :: Interval -> Double
+length (Interval s e) = fromIntegral (e - s) / 2