libinfluxdb 0.0.2 → 0.0.3

raw patch · 2 files changed

+65/−39 lines, 2 filesdep +bytestringPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: bytestring

API changes (from Hackage documentation)

Files

• libinfluxdb.cabal +2/−1
• src/Database/InfluxDB/Writer.hs +63/−38

@@ -1,5 +1,5 @@ name:                libinfluxdb-version:             0.0.2+version:             0.0.3  synopsis:            libinfluxdb description:         libinfluxdb@@ -30,6 +30,7 @@    build-depends:      base >=4.8 && <4.9+   , bytestring    , text    , clock    , containers

@@ -14,6 +14,8 @@ import           Data.Monoid import           Data.Pool +import qualified Data.ByteString as BS+ import           Data.Text (Text) import qualified Data.Text as T import qualified Data.Text.Encoding as T@@ -22,6 +24,7 @@ import qualified Data.Map as M  import           Control.Monad+import           Control.Exception import           Control.Concurrent import           Control.Concurrent.STM @@ -36,7 +39,11 @@  data Config = Config     { cURL :: !String-    , cDB  :: !String+      -- ^ The database name is extracted from the path (the leading slash is+      -- dropped, the tail is the database name). The rest of the URL (hostname,+      -- port, auth info) describes how to access the InfluxDB server.+      --+      -- Example: @http://localhost:8086/testdb@     }  data Handle = Handle@@ -63,62 +70,75 @@     mkRequestTemplate req = req         { method      = "POST"         , path        = "/write"-        , queryString = "?db=" <> T.encodeUtf8 (T.pack (cDB c))+        , queryString = "?db=" <> BS.tail (path req)         }  -    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)+    drainQueue :: TQueue (Maybe Point) -> Int -> Int -> IO [Point]+    drainQueue queue timeout count = do+        box <- newEmptyTMVarIO+        tmp <- newTVarIO [] +        void $ forkFinally (go tmp count) $+            (\_ -> atomically $ readTVar tmp >>= putTMVar box . reverse) -    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)+        atomically $ readTMVar box +      where+        go _   0 = return ()+        go tmp n = mask $ \restore -> do+            mbPoint <- restore $ atomically $ do+                mbPoint <- readTQueue queue+                case mbPoint of+                    Nothing -> return Nothing+                    Just p  -> do+                        modifyTVar' tmp (\x -> p : x)+                        return mbPoint +            case mbPoint of+                Nothing -> return ()+                Just _  -> do+                    when (n == count) $ do+                        threadId <- myThreadId+                        void $ forkFinally (threadDelay timeout) $ \_ -> killThread threadId++                    restore $ go tmp (n - 1)++     flushQueue :: Request -> TQueue (Maybe Point) -> IO ()     flushQueue req queue = do+        let batchSize = 50+        let timeout   = 10 * 1000 * 1000+         -- 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+        -- a batch of points is available.+        points <- drainQueue queue timeout batchSize -        let sleep    = threadDelay $ 5 * 1000000-            flush    = flushPoints manager req points-            continue = flushQueue req queue+        -- If the batch is non-empty, send the points to InfluxDB.+        when (length points > 0) $+            flushPoints manager req points -        -- 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+        -- If the batch is not full, it means there wasn't enough data in the+        -- queue. So we can sleep a bit.+        when (length points < batchSize) $+            threadDelay $ 5 * 1000000 +        -- Repeat.+        flushQueue req queue +     allocateResource :: Request -> IO (TQueue (Maybe Point))     allocateResource req = do         queue <- newTQueueIO          -- Fork off a thread which periodically flushes the queue.-        void $ forkIO $ flushQueue req queue+        --+        -- Async exceptions in the thread are completely masked, the only way+        -- to dispose of this thread is to write 'Nothing' into the queue. +        void $ mask_ $ forkIO $ flushQueue req queue++         return queue  @@ -158,11 +178,16 @@   +-- | Send a batch of points to the InfluxDB server. When the server is+-- unreachable, the batch is lost. Delivery is not retried. flushPoints :: Manager -> Request -> [Point] -> IO () flushPoints manager requestTemplate points = do-    void $ httpLbs req manager+    void send    where+    send :: IO (Either SomeException ())+    send = try $ void $ httpLbs req manager+     renderValue :: Value -> Text     renderValue (I x) = (T.pack $ show x) <> "i"     renderValue (F x) = (T.pack $ show x)