Ignoring json serialization errors. (#1403)

Add ability to use logs from jenkins.

beacon_chain/logtrace.nim

@@ -41,6 +41,21 @@ type
       name: "net-dir",
       defaultValue: "" }: string
 
+    logDir* {.
+      desc: "Specifies path with bunch of logs",
+      name: "log-dir",
+      defaultValue: "" }: string
+
+    ignoreSerializationErrors* {.
+      desc: "Ignore serialization errors while parsing log files",
+      name: "ignore-errors",
+      defaultValue: true }: bool
+
+    dumpSerializationErrors* {.
+      desc: "Dump full serialization errors while parsing log files",
+      name: "dump-errors",
+      defaultValue: false }: bool
+
     nodes* {.
       desc: "Specifies node names which logs will be used",
       name: "nodes" }: seq[string]
@@ -166,7 +181,9 @@ proc readLogFile(file: string): seq[JsonNode] =
   finally:
     stream.close()
 
-proc readLogFileForAttsMessages(file: string): seq[SlotAttMessage] =
+proc readLogFileForAttsMessages(file: string,
+                                ignoreErrors = true,
+                                dumpErrors = false): seq[SlotAttMessage] =
   var res = newSeq[SlotAttMessage]()
   var stream = newFileStream(file)
   var line: string
@@ -174,7 +191,22 @@ proc readLogFileForAttsMessages(file: string): seq[SlotAttMessage] =
   try:
     while not(stream.atEnd()):
       line = stream.readLine()
-      let m = Json.decode(line, LogMessage, allowUnknownFields = true)
+      inc(counter)
+      var m: LogMessage
+      try:
+        m = Json.decode(line, LogMessage, allowUnknownFields = true)
+      except SerializationError as exc:
+        if dumpErrors:
+          error "Serialization error while reading file, ignoring", file = file,
+                 line_number = counter, errorMsg = exc.formatMsg(line)
+        else:
+          error "Serialization error while reading file, ignoring", file = file,
+                 line_number = counter
+        if not(ignoreErrors):
+          raise exc
+        else:
+          continue
+
       if m.msg == "Attestation sent":
         let am = Json.decode(line, AttestationSentMessage,
                              allowUnknownFields = true)
@@ -187,30 +219,42 @@ proc readLogFileForAttsMessages(file: string): seq[SlotAttMessage] =
         let m = SlotAttMessage(kind: SaMessageType.SlotStart,
                                ssmsg: sm)
         res.add(m)
-      inc(counter)
+
       if counter mod 10_000 == 0:
         info "Processing file", file = extractFilename(file),
                                 lines_processed = counter,
                                 lines_filtered = len(res)
     result = res
 
-  except SerializationError as exc:
-    error "Serialization error while reading data from file", file = file,
-          errorMsg = exc.formatMsg(line)
   except CatchableError as exc:
     warn "Error reading data from file", file = file, errorMsg = exc.msg
   finally:
     stream.close()
 
-proc readLogFileForASRMessages(file: string,
-                               srnode: var SRANode) =
+proc readLogFileForASRMessages(file: string, srnode: var SRANode,
+                               ignoreErrors = true, dumpErrors = false) =
   var stream = newFileStream(file)
   var line: string
   var counter = 0
   try:
     while not(stream.atEnd()):
+      var m: LogMessage
       line = stream.readLine()
-      let m = Json.decode(line, LogMessage, allowUnknownFields = true)
+      inc(counter)
+      try:
+        m = Json.decode(line, LogMessage, allowUnknownFields = true)
+      except SerializationError as exc:
+        if dumpErrors:
+          error "Serialization error while reading file, ignoring", file = file,
+                 line_number = counter, errorMsg = exc.formatMsg(line)
+        else:
+          error "Serialization error while reading file, ignoring", file = file,
+                 line_number = counter
+        if not(ignoreErrors):
+          raise exc
+        else:
+          continue
+
       if m.msg == "Attestation sent":
         let sm = Json.decode(line, AttestationSentMessage,
                              allowUnknownFields = true)
@@ -219,15 +263,13 @@ proc readLogFileForASRMessages(file: string,
         let rm = Json.decode(line, AttestationReceivedMessage,
                              allowUnknownFields = true)
         discard srnode.recvs.hasKeyOrPut(rm.attestation.signature, rm)
-      inc(counter)
+
       if counter mod 10_000 == 0:
         info "Processing file", file = extractFilename(file),
                                 lines_processed = counter,
                                 sends_filtered = len(srnode.sends),
                                 recvs_filtered = len(srnode.recvs)
-  except SerializationError as exc:
-    error "Serialization error while reading data from file", file = file,
-          errorMsg = exc.formatMsg(line)
+
   except CatchableError as exc:
     warn "Error reading data from file", file = file, errorMsg = exc.msg
   finally:
@@ -292,6 +334,22 @@ proc getDirectoryLogFiles*(builddir: string,
         res.add(nodeDir)
   return res
 
+proc getLogFiles*(builddir: string,
+                  filter: seq[string]): seq[NodeDirectory] =
+  var res = newSeq[NodeDirectory]()
+  let dataPath = absolutePath(builddir)
+  if not dirExists(dataPath):
+    error "Logs directory did not exist", path = dataPath
+    quit(1)
+  for filePath in walkFiles(dataPath & DirSep & "*.*"):
+    let name = extractFilename(filePath)
+    if (len(filter) == 0) or (name in filter):
+      let nodeDir = NodeDirectory(name: extractFilename(filePath),
+                                  path: dataPath,
+                                  logs: @[extractFilename(filePath)])
+      res.add(nodeDir)
+  return res
+
 proc getMessage(logs: seq[GossipMessage],
                 msg: GossipMessage): Option[GossipMessage] =
   {.push warning[ProveInit]: off.}
@@ -450,6 +508,9 @@ proc run(conf: LogTraceConf) =
   if len(conf.netDir) > 0:
     logNodes = getDirectoryLogFiles(conf.netDir, conf.nodes)
 
+  if len(conf.logDir) > 0:
+    logNodes = getLogFiles(conf.logDir, conf.nodes)
+
   if len(logFiles) == 0 and len(logNodes) == 0:
     error "Log file sources not specified or not enough log files found"
     quit(1)