libinfluxdb (empty) → 0.0.1

raw patch · 4 files changed

+267/−0 lines, 4 filesdep +basedep +clockdep +containerssetup-changed

Dependencies added: base, clock, containers, http-client, http-client-tls, http-types, resource-pool, stm, text

Files

LICENSE +20/−0
Setup.hs +2/−0
libinfluxdb.cabal +40/−0
src/Database/InfluxDB/Writer.hs +205/−0

+ LICENSE view

@@ -0,0 +1,20 @@+Copyright (c) 2015 Tomas Carnecky++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

+ Setup.hs view

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

+ libinfluxdb.cabal view

@@ -0,0 +1,40 @@+name:                libinfluxdb+version:             0.0.1++synopsis:            libinfluxdb+description:         libinfluxdb++license:             MIT+license-file:        LICENSE++author:              Tomas Carnecky+maintainer:          tomas.carnecky@gmail.com++category:            Database++build-type:          Simple+cabal-version:       >=1.10+++source-repository head+  type:     git+  location: git://github.com/wereHamster/libinfluxdb.git+++library+  hs-source-dirs:      src+  default-language:    Haskell2010++  exposed-modules:+     Database.InfluxDB.Writer++  build-depends:+     base >=4.8 && <4.9+   , text+   , clock+   , containers+   , resource-pool+   , stm+   , http-types+   , http-client+   , http-client-tls

+ src/Database/InfluxDB/Writer.hs view

@@ -0,0 +1,205 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards   #-}++module Database.InfluxDB.Writer+    ( Config, Handle+    , createHandle, newHandle++    , Value(..), Tags, Fields+    , writePoint++    , test+    ) where+++import           Data.Int+import           Data.Monoid+import           Data.Pool++import           Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T++import           Data.Map (Map)+import qualified Data.Map as M++import           Control.Monad+import           Control.Concurrent+import           Control.Concurrent.STM++import           System.Clock++import           Network.HTTP.Types.Header+import           Network.HTTP.Client+import           Network.HTTP.Client.TLS (tlsManagerSettings)++import           Prelude++++data Config = Config+    { cURL :: !String+    , cDB  :: !String+    }++data Handle = Handle+    { hConfig  :: !Config+    , hManager :: !Manager++    , hRequest :: !Request+      -- ^ A template for a request which includes everything but the request+      -- body.++    , hPool :: Pool (TQueue (Maybe Point))+      -- ^ A pool of threads which consume messages from 'TQueue's and write+      -- them in batches to InfluxDB.+    }+++createHandle :: Config -> IO (Either () Handle)+createHandle c = do+    manager <- newManager tlsManagerSettings+    newHandle c manager+++newHandle :: Config -> Manager -> IO (Either () Handle)+newHandle c manager = do+    req  <- mkRequestTemplate <$> parseUrl (cURL c)+    pool <- createPool (allocateResource req) releaseResource 1 600 3++    return $ Right $ Handle c manager req pool++  where+    mkRequestTemplate req = req+        { method      = "POST"+        , path        = "/write"+        , queryString = "?db=" <> T.encodeUtf8 (T.pack (cDB c))+        }+++    dequeuePoints :: [Point] -> TQueue (Maybe Point) -> Int -> STM ([Point], Bool)+    dequeuePoints acc queue n+        | n <= 0    = return $ (reverse acc, False)+        | otherwise = do+            mbPoint <- tryReadTQueue queue+            case mbPoint of+                Nothing       -> return $ (reverse acc, False)+                Just Nothing  -> return $ (reverse acc, True)+                Just (Just p) -> dequeuePoints (p:acc) queue (n - 1)+++    dequeueBatch :: [Point] -> TQueue (Maybe Point) -> TimeSpec -> Int -> IO ([Point], Bool)+    dequeueBatch acc queue start n = do+        mbNextBatch <- atomically $ dequeuePoints [] queue n+        case mbNextBatch of+            (points, True) -> return (points, True)+            (points, False) -> do+                -- We can collect more points if we still have space AND time+                -- left. Otherwise return what we have.+                now <- getTime Monotonic+                if length points < n && timeSpecAsNanoSecs (diffTimeSpec start now) < (10 * 1000000000)+                    then dequeueBatch (acc ++ points) queue start (n - length points)+                    else return $ (acc ++ points, False)+++    flushQueue :: Manager -> Request -> TQueue (Maybe Point) -> IO ()+    flushQueue manager req queue = do+        -- Dequeue the next batch of points. This operation will block until+        -- 20 points are available or a timeout is reached, whichever comes+        -- first.+        start <- getTime Monotonic+        (points, isLast) <- dequeueBatch [] queue start 20++        let sleep    = threadDelay $ 5 * 1000000+            flush    = flushPoints manager req points+            continue = flushQueue manager req queue++        -- Deciding what to do next is a bit tricky.+        case (length points, isLast) of+            (0, True)  -> return ()+            (0, False) -> sleep >> continue+            (_, True)  -> flush+            (_, False) -> flush >> continue+++    allocateResource :: Request -> IO (TQueue (Maybe Point))+    allocateResource req = do+        queue <- newTQueueIO++        -- Fork off a thread which periodically flushes the queue.+        void $ forkIO $ flushQueue manager req queue++        return queue+++    releaseResource queue =+        atomically $ writeTQueue queue Nothing+++++-- | A 'Value' is either an integer, a floating point number, a boolean or+-- string.+data Value = I !Int64 | F !Double | B !Bool | S !Text+    deriving (Show, Eq)++type Tags = Map Text Text+type Fields = Map Text Value++data Point = Point+    { pMeasurement :: !Text+    , pTags :: !Tags+    , pFields :: !Fields+    , pTimestamp :: !(Maybe Int64)+    } deriving (Show, Eq)+++-- | Write a point to the database. Generates a timestamp from the local clock.+writePoint :: Handle -> Text -> Tags -> Fields -> IO ()+writePoint h measurement tags fields = do+    timestamp <- fromIntegral . timeSpecAsNanoSecs <$> getTime Realtime+    writePoint' h measurement tags fields timestamp++-- | Same as 'writePoint' but allows the caller to supply the timestamp.+writePoint' :: Handle -> Text -> Tags -> Fields -> Int64 -> IO ()+writePoint' h measurement tags fields timestamp =+    withResource (hPool h) $ \queue -> atomically $ writeTQueue queue $ Just $+        Point measurement tags fields (Just timestamp)++++flushPoints :: Manager -> Request -> [Point] -> IO ()+flushPoints manager requestTemplate points = do+    putStrLn $ "Flusing points: " ++ show (length points)+    print body+    void $ httpLbs req manager++  where+    renderValue :: Value -> Text+    renderValue (I x) = (T.pack $ show x) <> "i"+    renderValue (F x) = (T.pack $ show x)+    renderValue (B x) = (T.pack $ if x then "true" else "false")+    renderValue (S x) = "\"" <> x <> "\""++    tagsList :: Tags -> [Text]+    tagsList tags = map (\(k,v) -> k <> "=" <> v) $ M.toList tags+    fieldsList fields = map (\(k,v) -> k <> "=" <> renderValue v) $ M.toList fields++    line :: Point -> Text+    line Point{..} = mconcat $+        [ T.intercalate "," ([pMeasurement] ++ (tagsList pTags))+        , " "+        , T.intercalate "," (fieldsList pFields)+        ] ++ maybe [] (\x -> [" ", T.pack $ show x]) pTimestamp++    body = T.encodeUtf8 $ T.intercalate "\n" $ map line points+    req = requestTemplate { requestBody = RequestBodyBS body }+++test = do+    Right h <- createHandle (Config "http://localhost:8086/write" "rmx")+    writePoint h "test" (M.empty) ((M.singleton "value" (F 134)))+    threadDelay $ 1000000+    writePoint h "test" (M.empty) ((M.singleton "value" (F 234)))+    threadDelay $ 1000000+    writePoint h "test" (M.empty) ((M.singleton "value" (F 334)))