refactor: consolidate telemetry events (#4275)

* add specific Ray event

* group evaluation and training events

* consolidate pipeline run events

* fix send_event import

* review feedback

* typo

* send uptime

* track embeddingRetriever openai encoder

* track embeddingRetriever openai encoder

* pylitn

---------

Co-authored-by: Massimiliano Pippi <[email protected]>

Author: ZanSara

haystack/modeling/evaluation/eval.py

@@ -58,7 +58,9 @@ def eval(
         :return: all_results: A list of dictionaries, one for each prediction head. Each dictionary contains the metrics
                              and reports generated during evaluation.
         """
-        send_event("Evaluator.eval()")
+        send_event(
+            event_name="Evaluation", event_properties={"class": self.__class__.__name__, "function_name": "eval"}
+        )
         model.prediction_heads[0].use_confidence_scores_for_ranking = use_confidence_scores_for_ranking
         model.prediction_heads[0].use_no_answer_legacy_confidence = use_no_answer_legacy_confidence
         model.eval()

haystack/modeling/training/base.py

@@ -22,7 +22,7 @@
 from haystack.modeling.utils import GracefulKiller
 from haystack.utils.experiment_tracking import Tracker as tracker
 from haystack.utils.early_stopping import EarlyStopping
-from haystack.telemetry import send_event
+from haystack.telemetry_2 import send_event
 
 
 logger = logging.getLogger(__name__)
@@ -164,7 +164,7 @@ def train(self):
         :return: Returns the model after training. When you do ``early_stopping``
             with a ``save_dir`` the best model is loaded and returned.
         """
-        send_event("Trainer.train()")
+        send_event(event_name="Training", event_properties={"class": self.__class__.__name__, "function_name": "train"})
         # connect the prediction heads with the right output from processor
         self.model.connect_heads_with_processor(self.data_silo.processor.tasks, require_labels=True)
         # Check that the tokenizer(s) fits the language model(s)

haystack/nodes/reader/farm.py

@@ -435,7 +435,7 @@ def train(
         :param max_query_length: Maximum length of the question in number of tokens.
         :return: None
         """
-        send_event("FARMReader.train()")
+        send_event(event_name="Training", event_properties={"class": self.__class__.__name__, "function_name": "train"})
         return self._training_procedure(
             data_dir=data_dir,
             train_filename=train_filename,
@@ -557,7 +557,10 @@ def distil_prediction_layer_from(
         :param early_stopping: An initialized EarlyStopping object to control early stopping and saving of the best models.
         :return: None
         """
-        send_event("FARMReader.distil_prediction_layer_from()")
+        send_event(
+            event_name="Training",
+            event_properties={"class": self.__class__.__name__, "function_name": "distil_prediction_layer_from"},
+        )
         return self._training_procedure(
             data_dir=data_dir,
             train_filename=train_filename,
@@ -680,7 +683,10 @@ def distil_intermediate_layers_from(
         :param early_stopping: An initialized EarlyStopping object to control early stopping and saving of the best models.
         :return: None
         """
-        send_event("FARMReader.distil_intermediate_layers_from()")
+        send_event(
+            event_name="Training",
+            event_properties={"class": self.__class__.__name__, "function_name": "distil_intermediate_layers_from"},
+        )
         return self._training_procedure(
             data_dir=data_dir,
             train_filename=train_filename,
@@ -942,7 +948,10 @@ def eval_on_file(
             "Hence, results might slightly differ from those of `Pipeline.eval()`\n."
             "If you are just about starting to evaluate your model consider using `Pipeline.eval()` instead."
         )
-        send_event("FARMReader.eval_on_file()")
+        send_event(
+            event_name="Evaluation",
+            event_properties={"class": self.__class__.__name__, "function_name": "eval_on_file"},
+        )
         if device is None:
             device = self.devices[0]
         else:
@@ -1020,7 +1029,9 @@ def eval(
             "Hence, results might slightly differ from those of `Pipeline.eval()`\n."
             "If you are just about starting to evaluate your model consider using `Pipeline.eval()` instead."
         )
-        send_event("FARMReader.eval()")
+        send_event(
+            event_name="Evaluation", event_properties={"class": self.__class__.__name__, "function_name": "eval"}
+        )
         if device is None:
             device = self.devices[0]
         else:

haystack/nodes/retriever/_embedding_encoder.py

@@ -200,7 +200,7 @@ def train(
             reference the Sentence-Transformers [documentation](https://www.sbert.net/docs/training/overview.html#sentence_transformers.SentenceTransformer.fit)
             for a full list of keyword arguments.
         """
-        send_event("SentenceTransformersEmbeddingEncoder.train()")
+        send_event(event_name="Training", event_properties={"class": self.__class__.__name__, "function_name": "train"})
 
         if train_loss not in _TRAINING_LOSSES:
             raise ValueError(f"Unrecognized train_loss {train_loss}. Should be one of: {_TRAINING_LOSSES.keys()}")

haystack/nodes/retriever/_openai_encoder.py

@@ -12,6 +12,7 @@
 from haystack.nodes.retriever._base_embedding_encoder import _BaseEmbeddingEncoder
 from haystack.schema import Document
 from haystack.utils.openai_utils import USE_TIKTOKEN, count_openai_tokens, load_openai_tokenizer, openai_request
+from haystack.telemetry_2 import send_event
 
 if TYPE_CHECKING:
     from haystack.nodes.retriever import EmbeddingRetriever
@@ -23,6 +24,7 @@
 
 class _OpenAIEmbeddingEncoder(_BaseEmbeddingEncoder):
     def __init__(self, retriever: "EmbeddingRetriever"):
+        send_event("OpenAIEmbeddingEncoder initialized", event_properties={"model": retriever.embedding_model})
         # See https://platform.openai.com/docs/guides/embeddings and
         # https://learn.microsoft.com/en-us/azure/cognitive-services/openai/how-to/embeddings?tabs=console for more details
         self.using_azure = (

haystack/nodes/retriever/base.py

@@ -154,7 +154,9 @@ def eval(
                              contains the keys "predictions" and "metrics".
         :param headers: Custom HTTP headers to pass to document store client if supported (e.g. {'Authorization': 'Basic YWRtaW46cm9vdA=='} for basic authentication)
         """
-        send_event("BaseRetriever.eval()")
+        send_event(
+            event_name="Evaluation", event_properties={"class": self.__class__.__name__, "function_name": "eval"}
+        )
         # Extract all questions for evaluation
         filters: Dict = {"origin": [label_origin]}
 

haystack/nodes/retriever/dense.py

@@ -655,7 +655,7 @@ def train(
         Checkpoints can be stored via setting `checkpoint_every` to a custom number of steps.
         If any checkpoints are stored, a subsequent run of train() will resume training from the latest available checkpoint.
         """
-        send_event("DensePassageRetriever.train()")
+        send_event(event_name="Training", event_properties={"class": self.__class__.__name__, "function_name": "train"})
         self.processor.embed_title = embed_title
         self.processor.data_dir = Path(data_dir)
         self.processor.train_filename = train_filename
@@ -1307,7 +1307,7 @@ def train(
         :param checkpoints_to_keep: The maximum number of train checkpoints to save.
         :param early_stopping: An initialized EarlyStopping object to control early stopping and saving of the best models.
         """
-        send_event("TableTextRetriever.train()")
+        send_event(event_name="Training", event_properties={"class": self.__class__.__name__, "function_name": "train"})
         if embed_meta_fields is None:
             embed_meta_fields = ["page_title", "section_title", "caption"]
 
@@ -1923,7 +1923,7 @@ def train(
             reference the Sentence-Transformers [documentation](https://www.sbert.net/docs/training/overview.html#sentence_transformers.SentenceTransformer.fit)
             for a full list of keyword arguments.
         """
-        send_event("EmbeddingRetriever.train()")
+        send_event(event_name="Training", event_properties={"class": self.__class__.__name__, "function_name": "train"})
         self.embedding_encoder.train(
             training_data,
             learning_rate=learning_rate,

haystack/pipelines/base.py

@@ -83,6 +83,7 @@ def __init__(self):
         self.run_total = 0
         self.sent_event_in_window = False
         self.yaml_hash = False
+        self.last_run = None
 
     @property
     def root_node(self) -> Optional[str]:
@@ -496,7 +497,8 @@ def run(  # type: ignore
         """
         send_pipeline_run_event(
             pipeline=self,
-            event_name="Pipeline.run()",
+            classname=self.__class__.__name__,
+            function_name="run",
             query=query,
             file_paths=file_paths,
             labels=labels,
@@ -647,7 +649,8 @@ def run_batch(  # type: ignore
         """
         send_pipeline_run_event(
             pipeline=self,
-            event_name="Pipeline.run_batch()",
+            classname=self.__class__.__name__,
+            function_name="run_batch",
             queries=queries,
             file_paths=file_paths,
             labels=labels,
@@ -813,7 +816,7 @@ def eval_beir(
         Each metric is represented by a dictionary containing the scores for each top_k value.
         """
         send_event_2(
-            event_name="Pipeline.eval_beir()",
+            event_name=f"{cls.__name__}.eval_beir()",
             event_properties={
                 "dataset": dataset,
                 "index_pipeline": index_pipeline.yaml_hash,
@@ -1261,7 +1264,7 @@ def eval(
                                Additional information can be found here
                                https://huggingface.co/transformers/main_classes/model.html#transformers.PreTrainedModel.from_pretrained
         """
-        send_pipeline_event(pipeline=self, event_name="Pipeline.eval()")
+        send_pipeline_event(pipeline=self, event_name="Evaluation", event_properties={"function_name": "eval"})
 
         eval_result = EvaluationResult()
         if add_isolated_node_eval:
@@ -1380,7 +1383,7 @@ def eval_batch(
                                Additional information can be found here
                                https://huggingface.co/transformers/main_classes/model.html#transformers.PreTrainedModel.from_pretrained
         """
-        send_pipeline_event(pipeline=self, event_name="Pipeline.eval_batch()")
+        send_pipeline_event(pipeline=self, event_name=f"{self.__class__.__name__}.eval_batch()")
 
         eval_result = EvaluationResult()
         if add_isolated_node_eval:

haystack/pipelines/ray.py

@@ -23,6 +23,7 @@
 from haystack.nodes.base import BaseComponent, RootNode
 from haystack.pipelines.base import Pipeline
 from haystack.schema import Document, MultiLabel
+from haystack.telemetry_2 import send_pipeline_run_event
 
 
 logger = logging.getLogger(__name__)
@@ -311,6 +312,19 @@ async def run_async(  # type: ignore
                       about their execution. By default, this information includes the input parameters
                       the Nodes received and the output they generated. You can then find all debug information in the dictionary returned by this method under the key `_debug`.
         """
+        send_pipeline_run_event(
+            pipeline=self,
+            classname=self.__class__.__name__,
+            function_name="run_async",
+            query=query,
+            file_paths=file_paths,
+            labels=labels,
+            documents=documents,
+            meta=meta,
+            params=params,
+            debug=debug,
+        )
+
         # validate the node names
         self._validate_node_names_in_params(params=params)
 

haystack/telemetry_2.py

@@ -4,8 +4,9 @@
 import logging
 from pathlib import Path
 import json
-import yaml
+import datetime
 
+import yaml
 import posthog
 
 from haystack.environment import collect_static_system_specs, collect_dynamic_system_specs
@@ -103,7 +104,8 @@ def send_event(self, event_name: str, event_properties: Optional[Dict[str, Any]]
 
 
 def send_pipeline_run_event(  # type: ignore
-    event_name: str,
+    classname: str,
+    function_name: str,
     pipeline: "Pipeline",  # type: ignore
     query: Optional[str] = None,
     queries: Optional[List[str]] = None,
@@ -117,7 +119,8 @@ def send_pipeline_run_event(  # type: ignore
     """
     Sends a telemetry event about the execution of a pipeline, if telemetry is enabled.
 
-    :param event_name: The name of the event to show in PostHog.
+    :param classname: The name of the Pipeline class (Pipeline, RayPipeline, ...)
+    :param function_name: The name of the function that was invoked (run, run_batch, async_run, ...).
     :param pipeline: the pipeline that is running
     :param query: the value of the `query` input of the pipeline, if any
     :param queries: the value of the `queries` input of the pipeline, if any
@@ -130,15 +133,18 @@ def send_pipeline_run_event(  # type: ignore
     """
     try:
         if telemetry:
-            event_properties: Dict[str, Optional[Union[str, bool, int, Dict[str, Any]]]] = {}
+            event_properties: Dict[str, Optional[Union[str, bool, int, Dict[str, Any]]]] = {
+                "class": classname,
+                "function_name": function_name,
+            }
 
             # Check if it's the public demo
             exec_context = os.environ.get(HAYSTACK_EXECUTION_CONTEXT, "")
             if exec_context == "public_demo":
                 event_properties["pipeline.is_public_demo"] = True
                 event_properties["pipeline.run_parameters.query"] = query
                 event_properties["pipeline.run_parameters.params"] = params
-                telemetry.send_event(event_name=event_name, event_properties=event_properties)
+                telemetry.send_event(event_name=function_name, event_properties=event_properties)
                 return
 
             # Collect pipeline profile
@@ -185,26 +191,35 @@ def send_pipeline_run_event(  # type: ignore
             event_properties["pipeline.run_parameters.params"] = bool(params)
             event_properties["pipeline.run_parameters.debug"] = bool(debug)
 
-            telemetry.send_event(event_name=event_name, event_properties=event_properties)
+            telemetry.send_event(event_name="Pipeline run", event_properties=event_properties)
     except Exception as e:
         # Never let telemetry break things
-        logger.debug("There was an issue sending a %s telemetry event", event_name, exc_info=e)
+        logger.debug("There was an issue sending a '%s' telemetry event", function_name, exc_info=e)
 
 
-def send_pipeline_event(pipeline: "Pipeline", event_name: str):  # type: ignore
+def send_pipeline_event(pipeline: "Pipeline", event_name: str, event_properties: Optional[Dict[str, Any]] = None):  # type: ignore
     """
     Send a telemetry event related to a pipeline which is not a call to run(), if telemetry is enabled.
     """
     try:
         if telemetry:
-            telemetry.send_event(
-                event_name=event_name,
-                event_properties={
+            if not event_properties:
+                event_properties = {}
+            event_properties.update(
+                {
                     "pipeline.classname": pipeline.__class__.__name__,
                     "pipeline.fingerprint": pipeline.fingerprint,
                     "pipeline.yaml_hash": pipeline.yaml_hash,
-                },
+                }
             )
+            now = datetime.datetime.now()
+            if pipeline.last_run:
+                event_properties["pipeline.since_last_run"] = (now - pipeline.last_run).total_seconds()
+            else:
+                event_properties["pipeline.since_last_run"] = 0
+            pipeline.last_run = now
+
+            telemetry.send_event(event_name=event_name, event_properties=event_properties)
     except Exception as e:
         # Never let telemetry break things
         logger.debug("There was an issue sending a '%s' telemetry event", event_name, exc_info=e)
@@ -222,16 +237,6 @@ def send_event(event_name: str, event_properties: Optional[Dict[str, Any]] = Non
         logger.debug("There was an issue sending a '%s' telemetry event", event_name, exc_info=e)
 
 
-def _serializer(obj):
-    """
-    Small function used to build pipeline fingerprints and safely serialize any object.
-    """
-    try:
-        return str(obj)
-    except:
-        return "~ non serializable object ~"
-
-
 if os.environ.get("HAYSTACK_TELEMETRY_VERSION", "2") == "2":
     telemetry = Telemetry()
 else: