Improve logging for better debugging of named pipes (and less noise in logs)

Author: grafolean

netflowcollector.py

@@ -9,10 +9,13 @@
 
 from colors import color
 
+
 # python-netflow-v9-softflowd expects main.py to be the main entrypoint, but we only need
 # get_export_packets() iterator:
 sys.path.append(os.path.dirname(os.path.realpath(__file__)) + '/pynetflow')
 from pynetflow.main import get_export_packets
+# disable DEBUG logging on NetFlow collector library:
+logging.getLogger('pynetflow.main').setLevel(logging.INFO)
 
 
 logging.basicConfig(format='%(asctime)s.%(msecs)03d | %(levelname)s | %(message)s',
@@ -55,6 +58,7 @@ def process_netflow(netflow_port, named_pipe_filename):
                     entry = {
                         "ts": ts,
                         "client": client_ip,
+                        "seq": export.header.sequence,
                         "flows": [{
                             "IN_BYTES": data["IN_BYTES"],
                             "PROTOCOL": data["PROTOCOL"],
@@ -69,6 +73,7 @@ def process_netflow(netflow_port, named_pipe_filename):
                     }
                     line = json.dumps(entry).encode() + b'\n'
                     fp.write(line)
+                    log.debug(f"Wrote seq [{export.header.sequence}] from client [{client_ip}], ts [{ts}], n flows: [{len(flows_data)}]")
                     line = None
         except Exception as ex:
             log.exception(f"Exception: {str(ex)}")

netflowwriter.py

@@ -47,15 +47,15 @@ def write_record(j):
     with db.cursor() as c:
         # first save the flow record:
         ts = datetime.utcfromtimestamp(j['ts'])
-        log.info(f"Received record: {ts} from {j['client']}")
+        log.info(f"Received record [{j['seq']}]: {ts} from {j['client']}")
         c.execute(f"INSERT INTO {DB_PREFIX}records (ts, client_ip) VALUES (%s, %s) RETURNING seq;", (ts, j['client'],))
-        record_seq = c.fetchone()[0]
+        record_db_seq = c.fetchone()[0]
 
         # then save each of the flows within the record, but use execute_values() to perform bulk insert:
-        def _get_data(record_seq, flows):
+        def _get_data(record_db_seq, flows):
             for flow in flows:
                 yield (
-                    record_seq,
+                    record_db_seq,
                     flow.get('IN_BYTES'),
                     flow.get('PROTOCOL'),
                     flow.get('DIRECTION'),
@@ -66,7 +66,7 @@ def _get_data(record_seq, flows):
                     flow.get('IPV4_DST_ADDR'),
                     flow.get('IPV4_SRC_ADDR'),
                 )
-        data_iterator = _get_data(record_seq, j['flows'])
+        data_iterator = _get_data(record_db_seq, j['flows'])
         psycopg2.extras.execute_values(
             c,
             f"INSERT INTO {DB_PREFIX}flows (record, IN_BYTES, PROTOCOL, DIRECTION, L4_DST_PORT, L4_SRC_PORT, INPUT_SNMP, OUTPUT_SNMP, IPV4_DST_ADDR, IPV4_SRC_ADDR) VALUES %s",