# Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. """Application Insights Telemetry Client for Bots.""" import traceback from typing import Dict, Callable from applicationinsights import TelemetryClient # pylint: disable=no-name-in-module from botbuilder.core.bot_telemetry_client import ( BotTelemetryClient, Severity, TelemetryDataPointType, ) from .bot_telemetry_processor import BotTelemetryProcessor def bot_telemetry_processor(data, context) -> bool: """Bot Telemetry Processor as a method for backward compatibility. Refer to callable object :class:`BotTelemetryProcessor` for details. :param data: Data from Application Insights :type data: telemetry item :param context: Context from Application Insights :type context: context object :return: determines if the event is passed to the server (False = Filtered). :rtype: bool """ processor = BotTelemetryProcessor() return processor(data, context) class ApplicationInsightsTelemetryClient(BotTelemetryClient): """Application Insights Telemetry Client.""" def __init__( self, instrumentation_key: str, telemetry_client: TelemetryClient = None, telemetry_processor: Callable[[object, object], bool] = None, client_queue_size: int = None, ): self._instrumentation_key = instrumentation_key self._client = ( telemetry_client if telemetry_client is not None else TelemetryClient(self._instrumentation_key) ) if client_queue_size: self._client.channel.queue.max_queue_length = client_queue_size # Telemetry Processor processor = ( telemetry_processor if telemetry_processor is not None else bot_telemetry_processor ) self._client.add_telemetry_processor(processor) def track_pageview( self, name: str, url: str, duration: int = 0, properties: Dict[str, object] = None, measurements: Dict[str, object] = None, ) -> None: """ Send information about the page viewed in the application (a web page for instance). :param name: the name of the page that was viewed. :type name: str :param url: the URL of the page that was viewed. :type url: str :param duration: the duration of the page view in milliseconds. (defaults to: 0) :duration: int :param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) :type properties: :class:`typing.Dict[str, object]` :param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: None) :type measurements: :class:`typing.Dict[str, object]` """ self._client.track_pageview(name, url, duration, properties, measurements) def track_exception( self, exception_type: type = None, value: Exception = None, trace: traceback = None, properties: Dict[str, object] = None, measurements: Dict[str, object] = None, ) -> None: """ Send information about a single exception that occurred in the application. :param exception_type: the type of the exception that was thrown. :param value: the exception that the client wants to send. :param trace: the traceback information as returned by :func:`sys.exc_info`. :param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) :type properties: :class:`typing.Dict[str, object]` :param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: None) :type measurements: :class:`typing.Dict[str, object]` """ self._client.track_exception( exception_type, value, trace, properties, measurements ) def track_event( self, name: str, properties: Dict[str, object] = None, measurements: Dict[str, object] = None, ) -> None: """ Send information about a single event that has occurred in the context of the application. :param name: the data to associate to this event. :type name: str :param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) :type properties: :class:`typing.Dict[str, object]` :param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: None) :type measurements: :class:`typing.Dict[str, object]` """ self._client.track_event(name, properties=properties, measurements=measurements) def track_metric( self, name: str, value: float, tel_type: TelemetryDataPointType = None, count: int = None, min_val: float = None, max_val: float = None, std_dev: float = None, properties: Dict[str, object] = None, ) -> NotImplemented: """ Send information about a single metric data point that was captured for the application. :param name: The name of the metric that was captured. :type name: str :param value: The value of the metric that was captured. :type value: float :param tel_type: The type of the metric. (defaults to: TelemetryDataPointType.aggregation`) :param count: the number of metrics that were aggregated into this data point. (defaults to: None) :type count: int :param min_val: the minimum of all metrics collected that were aggregated into this data point. (defaults to: None) :type min_val: float :param max_val: the maximum of all metrics collected that were aggregated into this data point. (defaults to: None) :type max_val: float :param std_dev: the standard deviation of all metrics collected that were aggregated into this data point. (defaults to: None) :type std_dev: float :param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) :type properties: :class:`typing.Dict[str, object]` """ self._client.track_metric( name, value, tel_type, count, min_val, max_val, std_dev, properties ) def track_trace( self, name: str, properties: Dict[str, object] = None, severity: Severity = None ): """ Sends a single trace statement. :param name: the trace statement. :type name: str :param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) :type properties: :class:`typing.Dict[str, object]` :param severity: the severity level of this trace, one of DEBUG, INFO, WARNING, ERROR, CRITICAL """ self._client.track_trace(name, properties, severity) def track_request( self, name: str, url: str, success: bool, start_time: str = None, duration: int = None, response_code: str = None, http_method: str = None, properties: Dict[str, object] = None, measurements: Dict[str, object] = None, request_id: str = None, ): """ Sends a single request that was captured for the application. :param name: The name for this request. All requests with the same name will be grouped together. :type name: str :param url: The actual URL for this request (to show in individual request instances). :type url: str :param success: True if the request ended in success, False otherwise. :type success: bool :param start_time: the start time of the request. The value should look the same as the one returned by :func:`datetime.isoformat`. (defaults to: None) :type start_time: str :param duration: the number of milliseconds that this request lasted. (defaults to: None) :type duration: int :param response_code: the response code that this request returned. (defaults to: None) :type response_code: str :param http_method: the HTTP method that triggered this request. (defaults to: None) :type http_method: str :param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) :type properties: :class:`typing.Dict[str, object]` :param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: None) :type measurements: :class:`typing.Dict[str, object]` :param request_id: the id for this request. If None, a new uuid will be generated. (defaults to: None) :type request_id: str """ self._client.track_request( name, url, success, start_time, duration, response_code, http_method, properties, measurements, request_id, ) def track_dependency( self, name: str, data: str, type_name: str = None, target: str = None, duration: int = None, success: bool = None, result_code: str = None, properties: Dict[str, object] = None, measurements: Dict[str, object] = None, dependency_id: str = None, ): """ Sends a single dependency telemetry that was captured for the application. :param name: the name of the command initiated with this dependency call. Low cardinality value. Examples are stored procedure name and URL path template. :type name: str :param data: the command initiated by this dependency call. Examples are SQL statement and HTTP URL with all query parameters. :type data: str :param type_name: the dependency type name. Low cardinality value for logical grouping of dependencies and interpretation of other fields like commandName and resultCode. Examples are SQL, Azure table, and HTTP. (default to: None) :type type_name: str :param target: the target site of a dependency call. Examples are server name, host address. (default to: None) :type target: str :param duration: the number of milliseconds that this dependency call lasted. (defaults to: None) :type duration: int :param success: true if the dependency call ended in success, false otherwise. (defaults to: None) :type success: bool :param result_code: the result code of a dependency call. Examples are SQL error code and HTTP status code. (defaults to: None) :type result_code: str :param properties: the set of custom properties the client wants attached to this data item. (defaults to: None) :type properties: :class:`typing.Dict[str, object]` :param measurements: the set of custom measurements the client wants to attach to this data item. (defaults to: None) :type measurements: :class:`typing.Dict[str, object]` :param dependency_id: the id for this dependency call. If None, a new uuid will be generated. (defaults to: None) :type dependency_id: str """ self._client.track_dependency( name, data, type_name, target, duration, success, result_code, properties, measurements, dependency_id, ) def flush(self): """Flushes data in the queue. Data in the queue will be sent either immediately irrespective of what sender is being used. """ self._client.flush()