267 small issues before release (#268)

* 267 - ensure that function does not error if the graph is empty

* 267 - added cli updates on which tasks are being performed for otel2pv

* 267 - ensured JSON decode errors are logeed instead of terminal

Author: FreddieMatherSmartDCSIT

tel2puml/otel_to_pv/data_sources/json_data_source/json_datasource.py

@@ -1,9 +1,10 @@
 """Module containing DataSource sub class responsible for JSON ingestion."""
 
 import os
-from typing import Iterator
+from typing import Generator, Iterator, Any
 import json
 from logging import getLogger
+from io import TextIOWrapper
 
 from tqdm import tqdm
 from pydantic import ValidationError
@@ -13,11 +14,10 @@
 from .json_jq_converter import (
     generate_records_from_compiled_jq,
     compile_jq_query,
-    get_jq_query_from_config
+    get_jq_query_from_config,
 )
 from .json_config import JSONDataSourceConfig
 
-
 LOGGER = getLogger(__name__)
 
 
@@ -40,9 +40,7 @@ def __init__(self, config: JSONDataSourceConfig) -> None:
             )
         self.file_list = self.get_file_list()
         self.jq_query = get_jq_query_from_config(self.config)
-        self.compiled_jq = compile_jq_query(
-            self.jq_query
-        )
+        self.compiled_jq = compile_jq_query(self.jq_query)
         self.file_pbar = tqdm(
             total=len(self.file_list),
             desc="Ingesting JSON files",
@@ -117,10 +115,9 @@ def parse_json_stream(self, filepath: str) -> Iterator[OTelEvent]:
         :rtype: `Iterator`[:class:`OTelEvent`]
         """
         with open(filepath, "r", encoding="utf-8") as file:
-            if self.config.json_per_line:
-                jsons = (json.loads(line, strict=False) for line in file)
-            else:
-                jsons = (data for data in [json.load(file, strict=False)])
+            jsons = get_jsons_from_file(
+                file, filepath, self.config.json_per_line
+            )
             for data in jsons:
                 for record in generate_records_from_compiled_jq(
                     data, self.compiled_jq
@@ -160,3 +157,38 @@ def __next__(self) -> OTelEvent:
         self.events_pbar.close()
         self.event_error_pbar.close()
         raise StopIteration
+
+
+def get_jsons_from_file(
+    file_io: TextIOWrapper, filepath: str, json_per_line: bool = False
+) -> Generator[Any, Any, None]:
+    """Generator function to yield JSON data from a file.
+
+    :param file_io: The file object to read from
+    :type file_io: :class:`TextIOWrapper`
+    :param filepath: The path to the file
+    :type filepath: `str`
+    :param json_per_line: Whether the JSON data is formatted with one JSON
+    object per line. Defaults to `False`.
+    :type json_per_line: `bool`
+    :return: A generator yielding JSON objects
+    :rtype: `Generator`[:class:`Any`, `Any`, `None`]
+    """
+    counter = 0
+    if json_per_line:
+        try:
+            for line in file_io:
+                yield json.loads(line, strict=False)
+                counter += 1
+            return
+        except json.JSONDecodeError:
+            pass
+    try:
+        yield json.load(file_io, strict=False)
+    except json.JSONDecodeError:
+        LOGGER.error(
+            f"Error decoding JSON data in file: {filepath}\n"
+            "However, if the current line counter for the file is"
+            "greater than 0 then that number of lines was able to"
+            f" be decoded: {counter} lines."
+        )

tel2puml/otel_to_pv/otel_to_pv.py

@@ -57,20 +57,26 @@ def otel_to_pv(
         tqdm.write("Data ingested.")
     else:
         data_holder = fetch_data_holder(config)
+    tqdm.write("Cleaning data...")
     # validate spans
+    tqdm.write("Removing inconsistent jobs...")
     data_holder.remove_inconsistent_jobs()
     # remove jobs totally within time buffer zones
+    tqdm.write("Removing jobs outside of time window...")
     data_holder.remove_jobs_outside_of_time_window()
     # make sure job names are consistent in traces
+    tqdm.write("Updating job names by root span...")
     data_holder.update_job_names_by_root_span()
+    tqdm.write("Finished performing data cleaning operations.")
     # find unique graphs if required
     if find_unique_graphs:
+        tqdm.write("Finding unique graphs within the data holder...")
         job_name_to_job_ids_map: dict[str, set[str]] | None = (
             data_holder.find_unique_graphs()
         )
+        tqdm.write("Finished finding unique graphs.")
     else:
         job_name_to_job_ids_map = None
-
     job_name_group_streams = data_holder.stream_data(job_name_to_job_ids_map)
     # get the async event groups from the config
     sequencer_config = config.sequencer

tel2puml/puml_graph.py

@@ -482,7 +482,10 @@ def write_uml_blocks(
         :rtype: `list[str]`
         """
         # get the head node of the graph
-        head_node = list(topological_sort(self))[0]
+        top_sort = list(topological_sort(self))
+        if len(top_sort) == 0:
+            return []
+        head_node = top_sort[0]
         # get the ordered nodes from the depth-first search successors
         sorted_nodes = self._order_nodes_from_dfs_successors_dict(
             head_node, dfs_successors(self, head_node)