Collect error in the index

schemas/monocle/protob/crawler.proto

@@ -28,6 +28,29 @@ enum EntityType {
   ENTITY_TYPE_USER = 3;
 }
 
+message CrawlerError {
+  string message = 1;
+  string body = 2;
+  google.protobuf.Timestamp created_at = 3;
+  Entity entity = 4;
+}
+
+message ErrorsRequest {
+  string index = 1;
+  string query = 2;
+}
+
+message ErrorsList {
+  repeated CrawlerError errors = 1;
+}
+
+message ErrorsResponse {
+  oneof result {
+    ErrorsList success = 1;
+    string error = 2;
+  }
+}
+
 message AddDocRequest {
   string index = 1;
   string crawler = 2;
@@ -45,6 +68,7 @@ message AddDocRequest {
   repeated monocle_issue.Issue issues = 10;
   // issue_events are added when Entity is project_issue_name
   repeated monocle_issue.IssueEvent issue_events = 11;
+  repeated CrawlerError errors = 12;
 }
 
 enum AddDocError {

schemas/monocle/protob/http.proto

@@ -172,6 +172,14 @@ service Crawler {
       body: "*"
     };
   }
+  // Get crawler errors
+  rpc Errors(monocle_crawler.ErrorsRequest)
+      returns (monocle_crawler.ErrorsResponse) {
+    option (google.api.http) = {
+      post: "/api/2/crawler/errors"
+      body: "*"
+    };
+  }
 }
 
 // The monocle HTTP API

src/Lentille.hs

@@ -90,12 +90,10 @@ data GraphQLError = GraphQLError
   deriving (Show, Generic, ToJSON)
 
 data LentilleError
-  = DecodeError [Text]
-  | GraphError GraphQLError
+  = DecodeError UTCTime [Text]
+  | GraphError UTCTime GraphQLError
   deriving (Show, Generic, ToJSON)
 
-instance Exception LentilleError
-
 type LentilleStream es a = Stream (Of (Either LentilleError a)) (Eff es) ()
 
 -------------------------------------------------------------------------------
@@ -143,7 +141,7 @@ type Changes = (Change, [ChangeEvent])
 -- We don't care about the rest so we replace it with ()
 -- See: https://hackage.haskell.org/package/streaming-0.2.4.0/docs/Streaming-Prelude.html#v:break
 --
--- >>> let stream = S.yield (Left (DecodeError ["oops"]))
+-- >>> let stream = S.yield (Left (DecodeError [utctime|2021-05-31 00:00:00|] ["oops"]))
 -- >>> runEff $ S.length_ $ streamDropBefore [utctime|2021-05-31 00:00:00|] stream
 -- 1
 streamDropBefore :: UTCTime -> LentilleStream es Changes -> LentilleStream es Changes

src/Lentille/GraphQL.hs

@@ -238,7 +238,9 @@ streamFetch client@GraphClient {..} mkArgs StreamFetchOptParams {..} transformRe
       Nothing -> pure Nothing
 
     case mErr of
-      Just err -> S.yield (Left $ GraphError err)
+      Just err -> do
+        now <- lift mGetCurrentTime
+        S.yield (Left $ GraphError now err)
       Nothing -> go Nothing 0
 
   go pageInfoM totalFetched = do
@@ -254,15 +256,18 @@ streamFetch client@GraphClient {..} mkArgs StreamFetchOptParams {..} transformRe
 
     -- Handle the response
     case respE of
-      Left e ->
+      Left e -> do
         -- Yield the error and stop the stream
-        S.yield (Left $ GraphError e)
+        now <- lift mGetCurrentTime
+        S.yield (Left $ GraphError now e)
       Right (pageInfo, rateLimitM, decodingErrors, xs) -> do
         -- Log crawling status
         logStep pageInfo rateLimitM xs totalFetched
 
         case decodingErrors of
-          _ : _ -> S.yield (Left $ DecodeError decodingErrors)
+          _ : _ -> do
+            now <- lift mGetCurrentTime
+            S.yield (Left $ DecodeError now decodingErrors)
           [] -> do
             -- Yield the results
             S.each (Right <$> xs)

src/Macroscope/Test.hs

@@ -1,6 +1,7 @@
 -- | Tests for the macroscope process
 module Macroscope.Test where
 
+import Data.ByteString.Base64.Lazy qualified as B64
 import Effectful.Env
 import Effectful.Prometheus
 import Effectful.Reader.Static qualified as E
@@ -17,6 +18,7 @@ import Monocle.Backend.Queries qualified as Q
 import Monocle.Backend.Test (fakeChangePB, withTenantConfig)
 import Monocle.Backend.Test qualified as BT (fakeChange, fakeDate, fakeDateAlt)
 import Monocle.Client
+import Monocle.Client.Api (crawlerErrors)
 import Monocle.Config qualified as Config
 import Monocle.Effects
 import Monocle.Entity (CrawlerName (..), Entity (Project))
@@ -53,9 +55,18 @@ testCrawlingPoint = do
       Macroscope.runStream apiKey indexName (CrawlerName crawlerName) (Macroscope.Changes badStream)
 
       (currentOldestAge, _) <- getOldest
-      liftIO $ assertEqual "Commit date is not updated on failure" oldestAge currentOldestAge
+      liftIO $ assertBool "Commit date is updated on failure" (currentOldestAge > oldestAge)
 
-      Macroscope.runStream apiKey indexName (CrawlerName crawlerName) (Macroscope.Changes goodStream)
+      -- Check that the error got indexed
+      errorResponse <- crawlerErrors client (CrawlerPB.ErrorsRequest (from indexName) "from:2020")
+      case errorResponse of
+        CrawlerPB.ErrorsResponse (Just (CrawlerPB.ErrorsResponseResultSuccess (CrawlerPB.ErrorsList (toList -> [e])))) -> liftIO do
+          e.crawlerErrorMessage @?= "decode"
+          (B64.decode . encodeUtf8 $ e.crawlerErrorBody) @?= Right "[\"Oops\"]"
+          (from <$> e.crawlerErrorEntity) @?= Just (Project "opendev/neutron")
+        _ -> error $ "Expected one error, got: " <> show errorResponse
+
+      Macroscope.runStream apiKey indexName (CrawlerName crawlerName) (Macroscope.Changes $ goodStream currentOldestAge)
 
       (newOldestAge, _) <- getOldest
       liftIO $ assertBool "Commit date updated" (newOldestAge > oldestAge)
@@ -64,12 +75,12 @@ testCrawlingPoint = do
   badStream date name
     | date == BT.fakeDateAlt && name == "opendev/neutron" = do
         Streaming.yield $ Right (fakeChangePB, [])
-        Streaming.yield $ Left (DecodeError ["Oops"])
+        Streaming.yield $ Left (DecodeError BT.fakeDateAlt ["Oops"])
     | otherwise = error "Bad crawling point"
 
   -- A document stream that yield a change
-  goodStream date name
-    | date == BT.fakeDateAlt && name == "opendev/neutron" = do
+  goodStream expected date name
+    | date == expected && name == "opendev/neutron" = do
         Streaming.yield $ Right (fakeChangePB, [])
     | otherwise = error "Bad crawling point"
 

src/Macroscope/Worker.hs

@@ -55,11 +55,6 @@ streamName = \case
   TaskDatas _ -> "TaskDatas"
   UserChanges _ -> "UserChanges"
 
-isTDStream :: DocumentStream m -> Bool
-isTDStream = \case
-  TaskDatas _ -> True
-  _anyOtherStream -> False
-
 -------------------------------------------------------------------------------
 -- Adapter between protobuf api and crawler stream
 -------------------------------------------------------------------------------
@@ -78,16 +73,17 @@ data DocumentType
   | DTChanges (Change, [ChangeEvent])
   | DTTaskData TaskData
   | DTIssues (Issue, [IssueEvent])
+  | DTError CrawlerError
   deriving (Generic, ToJSON)
 
 data ProcessError es
   = CommitError Text
   | AddError Text
-  | StreamError (LentilleError, LentilleStream es DocumentType)
 
 -- | 'processStream' read the stream of document and post to the monocle API
 processStream ::
   forall es.
+  Entity ->
   -- | Funtion to log about the processing
   (Int -> Eff es ()) ->
   -- | Function to post on the Monocle API
@@ -96,7 +92,7 @@ processStream ::
   LentilleStream es DocumentType ->
   -- | The processing results
   Eff es [Maybe (ProcessError es)]
-processStream logFunc postFunc = go (0 :: Word) [] []
+processStream entity logFunc postFunc = go (0 :: Word) [] []
  where
   go count acc results stream = do
     eDocument <- S.next stream
@@ -105,12 +101,12 @@ processStream logFunc postFunc = go (0 :: Word) [] []
         -- The end of the stream
         res <- processBatch acc
         pure $ reverse (res : results)
-      Right (Left err, rest) -> do
-        -- An error occured in the stream, abort now
-        let res = Just (StreamError (err, rest))
-        pure $ reverse (res : results)
-      Right (Right doc, rest) -> do
+      Right (edoc, rest) -> do
         -- We got a new document
+        let doc = case edoc of
+              Right x -> x
+              Left (DecodeError ts err) -> DTError $ CrawlerError "decode" (encodeBlob err) (Just $ from ts) (Just $ from entity)
+              Left (GraphError ts err) -> DTError $ CrawlerError "graph" (encodeBlob err) (Just $ from ts) (Just $ from entity)
         let newAcc = doc : acc
         if count == 499
           then do
@@ -149,16 +145,6 @@ runStream apiKey indexName crawlerName documentStream = do
       forM_ errors \case
         AddError err -> logWarn "Could not add documents" ["err" .= err]
         CommitError err -> logWarn "Could not commit update date" ["err" .= err]
-        StreamError (err, rest) -> do
-          logWarn "Error occured when consuming the document stream" ["err" .= err]
-          S.toList_ rest >>= \case
-            [] -> pure ()
-            items -> logWarn "Left over documents found after error" ["items" .= items]
-
-          -- TODO: explains why TDStream don't support offset?
-          unless (isTDStream documentStream) do
-            -- Try the next entity by incrementing the offset
-            go startTime (offset + 1)
 
 -- | 'runStreamError' is the stream processor
 runStreamError ::
@@ -195,6 +181,7 @@ runStreamError startTime apiKey indexName (CrawlerName crawlerName) documentStre
     -- Run the document stream for that entity
     postResult <-
       processStream
+        entity
         (\c -> logInfo "Posting documents" ["count" .= c])
         (httpRetry "api/commit/add" . mCrawlerAddDoc . mkRequest entity)
         (getStream oldestAge entity)
@@ -246,6 +233,7 @@ runStreamError startTime apiKey indexName (CrawlerName crawlerName) documentStre
         addDocRequestTaskDatas = V.fromList $ mapMaybe getTD xs
         addDocRequestIssues = V.fromList $ mapMaybe getIssue xs
         addDocRequestIssueEvents = V.fromList $ concat $ mapMaybe getIssueEvent xs
+        addDocRequestErrors = V.fromList $ mapMaybe getError xs
      in AddDocRequest {..}
    where
     getIssue = \case
@@ -266,6 +254,9 @@ runStreamError startTime apiKey indexName (CrawlerName crawlerName) documentStre
     getTD = \case
       DTTaskData td -> Just td
       _ -> Nothing
+    getError = \case
+      DTError e -> Just e
+      _ -> Nothing
 
   -- 'commitTimestamp' post the commit date.
   commitTimestamp entity = do