Merge pull request #4328 from hove-io/task_NAV_3691_otlp_grafana

📈 Integration of Opentelemetry (without instrument)

Author: azime

.pre-commit-config.yaml

@@ -33,4 +33,4 @@ repos:
         args: [--ignore-missing-imports, --py2, --follow-imports, skip]
         files: source
         language_version: python3.9
-        exclude: /monitor/|/third_party/|/tests/|/sql/|env\.py$|setup\.py$
+        exclude: /monitor/|/third_party/|/tests/|/sql/|env\.py$|setup\.py|otlp\.py$

source/jormungandr/jormungandr/__init__.py

@@ -47,6 +47,8 @@
 if app.config.get(str('PATCH_WITH_GEVENT_SOCKET'), False):
     init.patch_http(patch_level=app.config.get(str('PATCH_WITH_GEVENT_SOCKET_LEVEL'), "socket"))
 
+from jormungandr import otlp
+
 from jormungandr import new_relic
 
 new_relic.init(app.config.get(str('NEWRELIC_CONFIG_PATH'), None))

source/jormungandr/jormungandr/api.py

@@ -33,7 +33,7 @@
 from __future__ import absolute_import, print_function, unicode_literals, division
 import importlib
 from flask_restful.representations import json
-from flask import request, make_response, abort
+from flask import request, make_response, abort, g
 from jormungandr import rest_api, app, i_manager
 from jormungandr.index import index
 from jormungandr.modules_loader import ModulesLoader
@@ -44,6 +44,8 @@
 from jormungandr.authentication import get_user, get_token, get_app_name, get_used_coverages
 from jormungandr._version import __version__
 import six
+from jormungandr.otlp import otlp_instance
+import time
 
 
 @rest_api.representation("text/jsonp")
@@ -126,6 +128,49 @@ def add_info_newrelic(response, *args, **kwargs):
     return response
 
 
+def __get_otlp_coverages_label():
+    used_coverages = get_used_coverages()
+
+    return ", ".join(sorted(used_coverages)) if used_coverages else "unknown"
+
+
+@app.after_request
+def record_request_call_to_otlp(response, *args, **kwargs):
+    try:
+        duration = time.time() - g.start
+        token = get_token() if get_token() else "unknown"
+        user = get_user(token=token, abort_if_no_token=False) if token != "unknown" else None
+        user_id = str(user.id) if user else "unknown"
+        token_name = get_app_name(token)
+        token_name = token_name if token_name else "unknown"
+        labels = {
+            "token": token,
+            "user_id": user_id,
+            "token_name": token_name,
+            "status": response.status_code,
+            "coverages": __get_otlp_coverages_label(),
+        }
+        otlp_instance.send_request_call_metrics(duration, labels)
+    except:
+        logger = logging.getLogger(__name__)
+        logger.exception('error while reporting to otlp from app.after_request')
+
+    return response
+
+
+@app.before_request
+def set_request_id():
+    try:
+        g.start = time.time()
+
+        otlp_instance.record_label("api", request.endpoint)
+        otlp_instance.record_label("version", __version__)
+        otlp_instance.record_label("coverage", __get_otlp_coverages_label())
+    except:
+        logger = logging.getLogger(__name__)
+        logger.exception('error while reporting to otlp from app.before_request')
+
+
 # If modules are configured, then load and run them
 if 'MODULES' in rest_api.app.config:
     rest_api.module_loader = ModulesLoader(rest_api)

source/jormungandr/jormungandr/autocomplete/abstract_autocomplete.py

@@ -34,6 +34,7 @@
 from jormungandr.exceptions import UnknownObject, TechnicalError, log_exception
 import six
 from jormungandr.new_relic import record_custom_event
+from jormungandr.otlp import otlp_instance
 
 
 class AutocompleteError(RuntimeError):
@@ -79,6 +80,7 @@ def record_status(self, status, exc=None):
         if exc is not None:
             data["cause"] = str(exc)
         record_custom_event('autocomplete_status', data)
+        otlp_instance.send_event_metrics('autocomplete_status', data)
 
     def get_object_by_uri(self, uri, request_id=None, instances=None, current_datetime=None):
         """

source/jormungandr/jormungandr/equipments/sytral.py

@@ -30,6 +30,7 @@
 from __future__ import absolute_import, print_function, unicode_literals, division
 
 from jormungandr import cache, app, new_relic
+from jormungandr.otlp import otlp_instance
 from navitiacommon import type_pb2
 from dateutil import parser
 from jormungandr.utils import date_to_timestamp, PY3
@@ -118,6 +119,7 @@ def record_call(self, status, **kwargs):
         params = {'parking_system_id': "SytralRT", 'dataset': "sytral", 'status': status}
         params.update(kwargs)
         new_relic.record_custom_event('parking_status', params)
+        otlp_instance.send_event_metrics('parking_status', params)
 
     def _fill_equipment_details(self, equipment_form_web_service, equipment_details):
         equipment_details.id = equipment_form_web_service['id']

source/jormungandr/jormungandr/exceptions.py

@@ -32,6 +32,8 @@
 from werkzeug.exceptions import HTTPException
 import logging
 from jormungandr.new_relic import record_exception
+from jormungandr.otlp import otlp_instance
+from typing import Dict
 
 __all__ = [
     "RegionNotFound",
@@ -49,6 +51,10 @@ def format_error(code, message):
     return error
 
 
+def format_otlp_error(data):
+    return {"error_id": data["error"]["id"], "error_message": data["error"]["message"]}
+
+
 class RegionNotFound(HTTPException):
     def __init__(self, region=None, lon=None, lat=None, object_id=None, custom_msg=None):
         super(RegionNotFound, self).__init__()
@@ -76,6 +82,7 @@ def __init__(self, region=None, lon=None, lat=None, object_id=None, custom_msg=N
             self.data = format_error("unknown_object", "Invalid id : {id}".format(id=object_id))
         else:
             self.data = format_error("unknown_object", "Unable to parse region")
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
     def __str__(self):
         return repr(self.data['message'])
@@ -87,6 +94,7 @@ def __init__(self, region, path):
         error = 'The region {} is dead'.format(region)
         self.data = format_error("dead_socket", error)
         self.code = 503
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 class ApiNotFound(HTTPException):
@@ -95,6 +103,7 @@ def __init__(self, api):
         error = 'The api {} doesn\'t exist'.format(api)
         self.data = format_error("unknown_object", error)
         self.code = 404
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 class UnknownObject(HTTPException):
@@ -103,27 +112,31 @@ def __init__(self, msg):
         error = 'The object {} doesn\'t exist'.format(msg)
         self.data = format_error("unknown_object", error)
         self.code = 404
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 class InvalidArguments(HTTPException):
     def __init__(self, arg):
         super(InvalidArguments, self).__init__()
         self.data = format_error("unknown_object", "Invalid arguments " + arg)
         self.code = 400
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 class UnableToParse(HTTPException):
     def __init__(self, msg):
         super(UnableToParse, self).__init__()
         self.data = format_error("unable_to_parse", msg)
         self.code = 400
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 class TechnicalError(HTTPException):
     def __init__(self, msg):
         super(TechnicalError, self).__init__()
         self.data = format_error("technical_error", msg)
         self.code = 500
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 # Only used by geovelo streetnetwork
@@ -132,27 +145,31 @@ def __init__(self, msg):
         super(GeoveloTechnicalError, self).__init__()
         self.data = format_error("technical_error", msg)
         self.code = 500
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 class HandimapTechnicalError(HTTPException):
     def __init__(self, msg):
         super(HandimapTechnicalError, self).__init__()
         self.data = format_error("technical_error", msg)
         self.code = 500
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 class AndyamoTechnicalError(HTTPException):
     def __init__(self, msg):
         super(AndyamoTechnicalError, self).__init__()
         self.data = format_error("technical_error", msg)
         self.code = 500
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 class ConfigException(Exception):
     def __init__(self, arg):
         super(ConfigException, self).__init__(arg)
         self.data = format_error("config_exception", "Invalid config " + arg)
         self.code = 400
+        otlp_instance.record_exception(self, format_otlp_error(self.data))
 
 
 def log_exception(sender, exception, **extra):
@@ -166,9 +183,11 @@ def log_exception(sender, exception, **extra):
         logger.debug(error)
         if exception.code >= 500:
             record_exception()
+            otlp_instance.record_exception(exception)
     else:
         logger.exception(error)
         record_exception()
+        otlp_instance.record_exception(exception)
 
 
 class StatManagerError(RuntimeError):

source/jormungandr/jormungandr/external_services/external_service.py

@@ -36,6 +36,7 @@
 import logging
 import requests as requests
 from six.moves.urllib.parse import urlencode
+from jormungandr.otlp import otlp_instance
 
 
 class ExternalServiceError(RuntimeError):
@@ -79,6 +80,7 @@ def record_call(self, url, status, **kwargs):
         params = {'external_service_id': "Forseti", 'status': status, 'external_service_url': url}
         params.update(kwargs)
         new_relic.record_custom_event('external_service_status', params)
+        otlp_instance.send_event_metrics('external_service_status', params)
 
     @abc.abstractmethod
     def get_response(self, arguments):

source/jormungandr/jormungandr/interfaces/v1/Journeys.py

@@ -484,7 +484,6 @@ def wrapper(*args, **kwargs):
 class Journeys(JourneyCommon):
     def __init__(self):
         # journeys must have a custom authentication process
-
         super(Journeys, self).__init__(output_type_serializer=api.JourneysSerializer)
 
         parser_get = self.parsers["get"]

source/jormungandr/jormungandr/new_relic.py

@@ -36,6 +36,7 @@
 from typing import Text, Callable
 from contextlib import contextmanager
 from jormungandr import app
+from jormungandr.otlp import otlp_instance
 
 try:
     from newrelic import agent
@@ -130,9 +131,10 @@ def get_common_event_params(service_name, call_name, status="ok"):
     }
 
 
+# TODO: Update and move this function into otlp.py when we will remove newrelic
 def distributedEvent(call_name, group_name):
     """
-    Custom event that we publish to New Relic for distributed scenario
+    Custom event that we publish to New Relic and Grafana for distributed scenario
     """
 
     def wrap(func):
@@ -148,13 +150,16 @@ def wrapper(obj, service, *args, **kwargs):
             except Exception as e:
                 event_params["status"] = "failed"
                 event_params.update({"exception": e})
+                otlp_instance.record_exception(e, event_params)
                 raise
 
             duration = timeit.default_timer() - start_time
             event_params.update({"duration": duration})
 
             # Send the custom event to newrelic !
             record_custom_event("distributed", event_params)
+            # Send metrics to otlp
+            otlp_instance.send_event_metrics("distributed", event_params)
 
             return result
 
@@ -163,6 +168,7 @@ def wrapper(obj, service, *args, **kwargs):
     return wrap
 
 
+# TODO: Update and move this function into otlp.py when we will remove newrelic
 @contextmanager
 def record_streetnetwork_call(call_name, connector_name, mode, coverage_name):
     """
@@ -179,15 +185,19 @@ def record_streetnetwork_call(call_name, connector_name, mode, coverage_name):
     except Exception as e:
         event_params["status"] = "failed"
         event_params.update({"exception": e})
+        otlp_instance.record_exception(e, event_params)
         raise
 
     duration = timeit.default_timer() - start_time
     event_params.update({"duration": duration})
 
     # Send the custom event to newrelic !
     record_custom_event(newrelic_service_name, event_params)
+    # Send metrics to otlp
+    otlp_instance.send_event_metrics(newrelic_service_name, event_params)
 
 
+# TODO: Update and move this function into otlp.py when we will remove newrelic
 def statManagerEvent(call_name, group_name):
     """
     Custom event that we publish to New Relic for stat_manager
@@ -205,13 +215,16 @@ def wrapper(obj, service, *args, **kwargs):
             except Exception as e:
                 event_params["status"] = "failed"
                 event_params.update({"reason": str(e)})
+                otlp_instance.record_exception(e, event_params)
                 raise
             finally:
                 duration = timeit.default_timer() - start_time
                 event_params.update({"duration": duration})
 
                 # Send the custom event to newrelic !
                 record_custom_event("stat_manager", event_params)
+                # Send metrics to otlp
+                otlp_instance.send_event_metrics("stat_manager", event_params)
 
         return wrapper