Spaces:
Build error
Build error
File size: 12,086 Bytes
0827183 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 |
# 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()
|