two version of R2R are here HEAD master

4 files changed, 729 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/__init__.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/__init__.py
new file mode 100644
index 00000000..2bfa3fd6
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/__init__.py
@@ -0,0 +1,17 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)
+
+from .activity import ActivityType, log_activity, monitor_with_activity, monitor_with_telemetry_mixin
+from .logging_handler import AML_INTERNAL_LOGGER_NAMESPACE, configure_appinsights_logging
+
+__all__ = [
+    "monitor_with_activity",
+    "monitor_with_telemetry_mixin",
+    "log_activity",
+    "ActivityType",
+    "configure_appinsights_logging",
+    "AML_INTERNAL_LOGGER_NAMESPACE",
+]
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/_customtraceback.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/_customtraceback.py
new file mode 100644
index 00000000..f8b41393
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/_customtraceback.py
@@ -0,0 +1,183 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+# pylint: disable=protected-access
+
+"""A file for custom traceback for removing file path from the trace for compliance need."""
+
+import os
+import sys
+import traceback
+import types
+from typing import Any, Generator, List, Optional, Type
+
+
+class _CustomStackSummary(traceback.StackSummary):
+    """Subclass of StackSummary."""
+
+    def format(self) -> List[str]:
+        """Format the stack ready for printing.
+
+        :return: A list of strings ready for printing.
+          * Each string in the resulting list corresponds to a single frame from the stack.
+          * Each string ends in a newline; the strings may contain internal newlines as well, for those items with
+            source text lines.
+        :rtype: List[str]
+        """
+        result = []
+        for frame in self:
+            row = ['  File "{}", line {}, in {}\n'.format(os.path.basename(frame.filename), frame.lineno, frame.name)]
+            if frame.line:
+                row.append("    {}\n".format(frame.line.strip()))
+            if frame.locals:
+                for name, value in sorted(frame.locals.items()):
+                    row.append("    {name} = {value}\n".format(name=name, value=value))
+            result.append("".join(row))
+        return result
+
+
+# pylint: disable=too-many-instance-attributes
+class _CustomTracebackException(traceback.TracebackException):
+    # pylint: disable=super-init-not-called
+    def __init__(
+        self, exc_type, exc_value, exc_traceback, *, limit=None, lookup_lines=True, capture_locals=False, _seen=None
+    ):
+        if _seen is None:
+            _seen = set()
+        _seen.add(exc_value)
+        # Gracefully handle (the way Python 2.4 and earlier did) the case of
+        # being called with no type or value (None, None, None).
+        if exc_value and exc_value.__cause__ is not None and exc_value.__cause__ not in _seen:
+            cause = _CustomTracebackException(
+                type(exc_value.__cause__),
+                exc_value.__cause__,
+                exc_value.__cause__.__traceback__,
+                limit=limit,
+                lookup_lines=False,
+                capture_locals=capture_locals,
+                _seen=_seen,
+            )
+        else:
+            cause = None
+        if exc_value and exc_value.__context__ is not None and exc_value.__context__ not in _seen:
+            context = _CustomTracebackException(
+                type(exc_value.__context__),
+                exc_value.__context__,
+                exc_value.__context__.__traceback__,
+                limit=limit,
+                lookup_lines=False,
+                capture_locals=capture_locals,
+                _seen=_seen,
+            )
+        else:
+            context = None
+        self.exc_traceback = exc_traceback
+        self.__cause__ = cause
+        self.__context__ = context
+        self.__suppress_context__ = exc_value.__suppress_context__ if exc_value else False
+        # TODO: locals.
+        self.stack = _CustomStackSummary.extract(
+            traceback.walk_tb(exc_traceback),
+            limit=limit,
+            lookup_lines=lookup_lines,
+            capture_locals=capture_locals,
+        )
+        self.exc_type = exc_type
+        # Capture now to permit freeing resources: only complication is in the
+        # unofficial API _format_final_exc_line
+        self._str = traceback._some_str(exc_value)
+        if exc_type and issubclass(exc_type, SyntaxError):
+            # Handle SyntaxError's specially
+            self.filename = exc_value.filename
+            self.lineno = str(exc_value.lineno)
+            self.text = exc_value.text
+            self.offset = exc_value.offset
+            self.msg = exc_value.msg
+        if lookup_lines:
+            self._load_lines()
+
+    def format_exception_only(self) -> Generator:
+        """Format the exception part of the traceback.
+
+        :return: An iterable of strings, each ending in a newline. Normally, the generator emits a single
+           string; however, for SyntaxError exceptions, it emits several lines that (when printed) display
+           detailed information about where the syntax error occurred. The message indicating which exception occurred
+           is always the last string in the output.
+        :rtype: Iterable[str]
+        """
+        if self.exc_type is None:
+            yield traceback._format_final_exc_line(None, self._str)
+            return
+
+        stype = self.exc_type.__qualname__
+        smod = self.exc_type.__module__
+        if smod not in ("__main__", "builtins"):
+            stype = smod + "." + stype
+
+        if not issubclass(self.exc_type, SyntaxError):
+            yield traceback._format_final_exc_line(stype, self._str)  # type: ignore[attr-defined]
+            return
+
+        # It was a syntax error; show exactly where the problem was found.
+        filename = os.path.basename(self.filename) or "<string>"
+        lineno = str(self.lineno) or "?"
+        yield '  File "{}", line {}\n'.format(filename, lineno)
+
+        badline = self.text
+        offset = self.offset
+        if badline is not None:
+            yield "    {}\n".format(badline.strip())
+            if offset is not None:
+                caretspace: Any = badline.rstrip("\n")
+                offset = min(len(caretspace), offset) - 1
+                caretspace = caretspace[:offset].lstrip()
+                # non-space whitespace (likes tabs) must be kept for alignment
+                caretspace = ((c.isspace() and c or " ") for c in caretspace)
+                yield "    {}^\n".format("".join(caretspace))
+        msg = self.msg or "<no detail available>"
+        yield "{}: {}\n".format(stype, msg)
+
+
+def format_exc(limit: Optional[int] = None, chain: bool = True) -> str:
+    """Like print_exc() but return a string.
+
+    :param limit: None to include all frames or the number of frames to include.
+    :type limit: Optional[int]
+    :param chain: Whether to format __cause__ and __context__. Defaults to True
+    :type chain: bool
+    :return: The formatted exception string
+    :rtype: str
+    """
+    return "".join(format_exception(*sys.exc_info(), limit=limit, chain=chain))  # type: ignore[misc]
+
+
+def format_exception(  # pylint: disable=unused-argument
+    etype: Type[BaseException],
+    value: BaseException,
+    tb: types.TracebackType,
+    limit: Optional[int] = None,
+    chain: bool = True,
+) -> List[str]:
+    """Format a stack trace and the exception information.
+
+    The arguments have the same meaning as the corresponding arguments to print_exception().
+
+    :param etype: The type of the exception
+    :type etype: Type[BaseException]
+    :param value: The exception
+    :type value: BaseException
+    :param tb: The exception traceback
+    :type tb: types.TracebackType
+    :param limit: None to include all frames or the number of frames to include.
+    :type limit: Optional[int]
+    :param chain: Whether to format __cause__ and __context__. Defaults to True
+    :type chain: bool
+    :return: A list of strings, each ending in a newline and some containing internal newlines.
+      When these lines are concatenated and printed, exactly the same text is printed as does print_exception().
+    :rtype: List[str]
+    """
+    # format_exception has ignored etype for some time, and code such as cgitb
+    # passes in bogus values as a result. For compatibility with such code we
+    # ignore it here (rather than in the new TracebackException API).
+    return list(_CustomTracebackException(type(value), value, tb, limit=limit).format(chain=chain))
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/activity.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/activity.py
new file mode 100644
index 00000000..169fa5ef
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/activity.py
@@ -0,0 +1,379 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+# pylint: disable=protected-access
+
+"""Defines functionality for collecting telemetry about code.
+
+An activity is a logical block of code that consumers want to monitor.
+To monitor an activity, either wrap the logical block of code with the
+``log_activity()`` method or use the ``@monitor_with_activity``
+decorator.
+"""
+import contextlib
+import functools
+import inspect
+import logging
+import os
+import uuid
+from datetime import datetime
+from typing import Dict, Tuple
+from uuid import uuid4
+
+from marshmallow import ValidationError
+
+from azure.ai.ml._utils._logger_utils import OpsLogger
+from azure.ai.ml._utils.utils import _is_user_error_from_exception_type, _is_user_error_from_status_code, _str_to_bool
+from azure.ai.ml.exceptions import ErrorCategory, MlException
+from azure.core.exceptions import HttpResponseError
+
+# Get environment variable IS_IN_CI_PIPELINE to decide whether it's in CI test
+IS_IN_CI_PIPELINE = _str_to_bool(os.environ.get("IS_IN_CI_PIPELINE", "False"))
+
+ACTIVITY_SPAN = "activity_span"
+
+
+class ActivityType(object):
+    """The type of activity (code) monitored.
+
+    The default type is "PublicAPI".
+    """
+
+    PUBLICAPI = "PublicApi"  # incoming public API call (default)
+    INTERNALCALL = "InternalCall"  # internal (function) call
+    CLIENTPROXY = "ClientProxy"  # an outgoing service API call
+
+
+class ActivityCompletionStatus(object):
+    """The activity (code) completion status, success, or failure."""
+
+    SUCCESS = "Success"
+    FAILURE = "Failure"
+
+
+class ActivityLoggerAdapter(logging.LoggerAdapter):
+    """An adapter for loggers to keep activity contextual information in
+    logging output.
+
+    :param logger: The activity logger adapter.
+    :type logger: logging.LoggerAdapter
+    :param activity_info: The info to write to the logger.
+    :type activity_info: str
+    """
+
+    def __init__(self, logger: logging.Logger, activity_info: Dict):
+        """Initialize a new instance of the class.
+
+        :param logger: The activity logger.
+        :type logger: logging.Logger
+        :param activity_info: The info to write to the logger.
+        :type activity_info: Dict
+        """
+        self._activity_info = activity_info
+        super(ActivityLoggerAdapter, self).__init__(logger, None)  # type: ignore[arg-type]
+
+    @property
+    def activity_info(self) -> Dict:
+        """Return current activity info.
+
+        :return: The info to write to the logger
+        :rtype: Dict
+        """
+        return self._activity_info
+
+    def process(self, msg: str, kwargs: Dict) -> Tuple[str, Dict]:  # type: ignore[override]
+        """Process the log message.
+
+        :param msg: The log message.
+        :type msg: str
+        :param kwargs: The arguments with properties.
+        :type kwargs: dict
+        :return: A tuple of the processed msg and kwargs
+        :rtype: Tuple[str, Dict]
+        """
+        if "extra" not in kwargs:
+            kwargs["extra"] = {}
+
+        kwargs["extra"].update(self._activity_info)
+
+        return msg, kwargs
+
+
+def error_preprocess(activityLogger, exception):
+    """Try to update activityLogger if exception is HttpResponseError or
+    builtin error (such as KeyError, TypeError)
+
+    For HttpResponseError, will log exception message and classify error category according to http response status
+    code if have, otherwise classify as Unknown. For builtin error,  will log exception message and classify error
+    category as Unknown.
+
+    :param activityLogger: The logger adapter.
+    :type activityLogger: ActivityLoggerAdapter
+    :param exception: The raised exception to be preprocessed.
+    :type exception: BaseException
+    :return: The provided exception
+    :rtype: Exception
+    """
+
+    if isinstance(exception, HttpResponseError):
+        activityLogger.activity_info["errorMessage"] = exception.message
+        if exception.response:
+            http_status_code = exception.response.status_code
+            error_category = (
+                ErrorCategory.USER_ERROR
+                if _is_user_error_from_status_code(http_status_code)
+                else ErrorCategory.SYSTEM_ERROR
+            )
+        else:
+            error_category = ErrorCategory.UNKNOWN
+        activityLogger.activity_info["errorCategory"] = error_category
+        if exception.inner_exception:
+            activityLogger.activity_info["innerException"] = type(exception.inner_exception).__name__
+    elif isinstance(exception, MlException):
+        # If exception is MlException, it will have error_category, message and target attributes and will log those
+        # information in log_activity, no need more actions here.
+        pass
+    elif isinstance(exception, ValidationError):
+        # Validation error should be user error
+        activityLogger.activity_info["errorMessage"] = str(exception)
+        activityLogger.activity_info["errorCategory"] = ErrorCategory.USER_ERROR
+    else:
+        activityLogger.activity_info["errorMessage"] = "Got error {0}: '{1}' while calling {2}".format(
+            exception.__class__.__name__,
+            exception,
+            activityLogger.activity_info["activity_name"],
+        )
+        if _is_user_error_from_exception_type(exception) or _is_user_error_from_exception_type(exception.__cause__):
+            activityLogger.activity_info["errorCategory"] = ErrorCategory.USER_ERROR
+        else:
+            # Todo: should check KeyError, TypeError, ValueError caused by user before request or raise in code directly
+            activityLogger.activity_info["errorCategory"] = ErrorCategory.UNKNOWN
+    return exception
+
+
+@contextlib.contextmanager
+def log_activity(
+    logger,
+    activity_name,
+    activity_type=ActivityType.INTERNALCALL,
+    custom_dimensions=None,
+):
+    """Log an activity.
+
+    An activity is a logical block of code that consumers want to monitor.
+    To monitor, wrap the logical block of code with the ``log_activity()`` method. As an alternative, you can
+    also use the ``@monitor_with_activity`` decorator.
+
+    :param logger: The logger adapter.
+    :type logger: logging.LoggerAdapter
+    :param activity_name: The name of the activity. The name should be unique per the wrapped logical code block.
+    :type activity_name: str
+    :param activity_type: One of PUBLICAPI, INTERNALCALL, or CLIENTPROXY which represent an incoming API call,
+        an internal (function) call, or an outgoing API call. If not specified, INTERNALCALL is used.
+    :type activity_type: str
+    :param custom_dimensions: The custom properties of the activity.
+    :type custom_dimensions: dict
+    :return: An activity logger
+    :rtype: Iterable[ActivityLoggerAdapter]
+    """
+    activity_info = {
+        "activity_id": str(uuid.uuid4()),
+        "activity_name": activity_name,
+        "activity_type": activity_type,
+    }
+    custom_dimensions = custom_dimensions or {}
+    custom_dimensions.update({"client_request_id": str(uuid4())})
+    activity_info.update(custom_dimensions)
+
+    start_time = datetime.utcnow()
+    completion_status = ActivityCompletionStatus.SUCCESS
+
+    message = "ActivityStarted, {}".format(activity_name)
+    activityLogger = ActivityLoggerAdapter(logger, activity_info)  # type: ignore[arg-type]
+    activityLogger.info(message)
+    exception = None
+
+    try:
+        yield activityLogger
+    except BaseException as e:
+        exception = error_preprocess(activityLogger, e)
+        completion_status = ActivityCompletionStatus.FAILURE
+        # All the system and unknown errors except for NotImplementedError will be wrapped with a new exception.
+        if IS_IN_CI_PIPELINE and not isinstance(e, NotImplementedError):
+            if (
+                isinstance(exception, MlException)
+                and exception.error_category  # pylint: disable=no-member
+                in [ErrorCategory.SYSTEM_ERROR, ErrorCategory.UNKNOWN]
+            ) or (
+                "errorCategory" in activityLogger.activity_info
+                and activityLogger.activity_info["errorCategory"]  # type: ignore[index]
+                in [ErrorCategory.SYSTEM_ERROR, ErrorCategory.UNKNOWN]
+            ):
+                # pylint: disable=W0719
+                raise Exception("Got InternalSDKError", e) from e
+            raise
+        raise
+    finally:
+        try:
+            end_time = datetime.utcnow()
+            duration_ms = round((end_time - start_time).total_seconds() * 1000, 2)
+
+            activityLogger.activity_info["completionStatus"] = completion_status  # type: ignore[index]
+            activityLogger.activity_info["durationMs"] = duration_ms  # type: ignore[index]
+            message = "ActivityCompleted: Activity={}, HowEnded={}, Duration={} [ms]".format(
+                activity_name, completion_status, duration_ms
+            )
+            if exception:
+                activityLogger.activity_info["exception"] = type(exception).__name__  # type: ignore[index]
+                if isinstance(exception, MlException):
+                    activityLogger.activity_info[  # type: ignore[index]
+                        "errorMessage"
+                    ] = exception.no_personal_data_message  # pylint: disable=no-member
+                    # pylint: disable=no-member
+                    activityLogger.activity_info["errorTarget"] = exception.target  # type: ignore[index]
+                    activityLogger.activity_info[  # type: ignore[index]
+                        "errorCategory"
+                    ] = exception.error_category  # pylint: disable=no-member
+                    if exception.inner_exception:  # pylint: disable=no-member
+                        # pylint: disable=no-member
+                        activityLogger.activity_info["innerException"] = type(  # type: ignore[index]
+                            exception.inner_exception
+                        ).__name__
+                message += ", Exception={}".format(activityLogger.activity_info["exception"])
+                message += ", ErrorCategory={}".format(activityLogger.activity_info["errorCategory"])
+                message += ", ErrorMessage={}".format(activityLogger.activity_info["errorMessage"])
+
+                activityLogger.error(message)
+            else:
+                activityLogger.info(message)
+        except Exception:  # pylint: disable=W0718
+            pass
+
+
+# pylint: disable-next=docstring-missing-rtype
+def monitor_with_activity(
+    logger,
+    activity_name,
+    activity_type=ActivityType.INTERNALCALL,
+    custom_dimensions=None,
+):
+    """Add a wrapper for monitoring an activity (code).
+
+    An activity is a logical block of code that consumers want to monitor.
+    To monitor, use the ``@monitor_with_activity`` decorator. As an alternative, you can also wrap the
+    logical block of code with the ``log_activity()`` method.
+
+    :param logger: The operations logging class, containing loggers and tracer for the package and module
+    :type logger: ~azure.ai.ml._utils._logger_utils.OpsLogger
+    :param activity_name: The name of the activity. The name should be unique per the wrapped logical code block.
+    :type activity_name: str
+    :param activity_type: One of PUBLICAPI, INTERNALCALL, or CLIENTPROXY which represent an incoming API call,
+        an internal (function) call, or an outgoing API call. If not specified, INTERNALCALL is used.
+    :type activity_type: str
+    :param custom_dimensions: The custom properties of the activity.
+    :type custom_dimensions: dict
+    :return:
+    """
+
+    def monitor(f):
+        @functools.wraps(f)
+        def wrapper(*args, **kwargs):
+            tracer = logger.package_tracer if isinstance(logger, OpsLogger) else None
+            if tracer:
+                with tracer.start_as_current_span(ACTIVITY_SPAN):
+                    with log_activity(
+                        logger.package_logger, activity_name or f.__name__, activity_type, custom_dimensions
+                    ):
+                        return f(*args, **kwargs)
+            elif hasattr(logger, "package_logger"):
+                with log_activity(logger.package_logger, activity_name or f.__name__, activity_type, custom_dimensions):
+                    return f(*args, **kwargs)
+            else:
+                with log_activity(logger, activity_name or f.__name__, activity_type, custom_dimensions):
+                    return f(*args, **kwargs)
+
+        return wrapper
+
+    return monitor
+
+
+# pylint: disable-next=docstring-missing-rtype
+def monitor_with_telemetry_mixin(
+    logger,
+    activity_name,
+    activity_type=ActivityType.INTERNALCALL,
+    custom_dimensions=None,
+    extra_keys=None,
+):
+    """Add a wrapper for monitoring an activity (code) with telemetry mixin.
+
+    An activity is a logical block of code that consumers want to monitor.
+    Object telemetry values will be added into custom dimensions if activity parameter or return object is
+    an instance of TelemetryMixin, which means log dimension will include: 1. telemetry collected from
+    parameter object. 2. custom dimensions passed in. 3.(optional) if no telemetry found in parameter,
+    will collect from return value.
+    To monitor, use the ``@monitor_with_telemetry_mixin`` decorator.
+
+    :param logger: The operations logging class, containing loggers and tracer for the package and module
+    :type logger: logging.LoggerAdapter
+    :param activity_name: The name of the activity. The name should be unique per the wrapped logical code block.
+    :type activity_name: str
+    :param activity_type: One of PUBLICAPI, INTERNALCALL, or CLIENTPROXY which represent an incoming API call,
+        an internal (function) call, or an outgoing API call. If not specified, INTERNALCALL is used.
+    :type activity_type: str
+    :param custom_dimensions: The custom properties of the activity.
+    :type custom_dimensions: dict
+    :param extra_keys: Extra keys from parameter to be tracked. The key in extra_keys will always be added into
+        dimensions, the value of it will be str(obj) if it exists in parameters, or 'None' if not.
+    :type extra_keys: list[str]
+    :return:
+    """
+
+    logger = logger.package_logger if isinstance(logger, OpsLogger) else logger
+
+    def monitor(f):
+        def _collect_from_parameters(f, args, kwargs, extra_keys):
+            dimensions = {}
+            named_args = dict(zip(inspect.signature(f).parameters.keys(), args))
+            parameters = {**named_args, **kwargs}
+            from azure.ai.ml.entities._mixins import TelemetryMixin
+
+            # extract mixin dimensions from arguments
+            for key, obj in parameters.items():
+                try:
+                    if isinstance(obj, TelemetryMixin):
+                        dimensions.update(obj._get_telemetry_values())
+                    elif extra_keys and key in extra_keys:
+                        dimensions[key] = str(obj)
+                except Exception:  # pylint: disable=W0718
+                    pass
+            # add left keys with None
+            if extra_keys:
+                for key in extra_keys:
+                    if key not in parameters:
+                        dimensions[key] = "None"
+            return dimensions
+
+        def _collect_from_return_value(value):
+            from azure.ai.ml.entities._mixins import TelemetryMixin
+
+            try:
+                return value._get_telemetry_values() if isinstance(value, TelemetryMixin) else {}
+            except Exception:  # pylint: disable=W0718
+                return {}
+
+        @functools.wraps(f)
+        def wrapper(*args, **kwargs):
+            parameter_dimensions = _collect_from_parameters(f, args, kwargs, extra_keys)
+            dimensions = {**parameter_dimensions, **(custom_dimensions or {})}
+            with log_activity(logger, activity_name or f.__name__, activity_type, dimensions) as activityLogger:
+                return_value = f(*args, **kwargs)
+                if not parameter_dimensions:
+                    # collect from return if no dimensions from parameter
+                    activityLogger.activity_info.update(_collect_from_return_value(return_value))
+                return return_value
+
+        return wrapper
+
+    return monitor
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/logging_handler.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/logging_handler.py
new file mode 100644
index 00000000..cbbbb481
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/_telemetry/logging_handler.py
@@ -0,0 +1,150 @@
+# ---------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# ---------------------------------------------------------
+
+"""Contains functionality for sending telemetry to Application Insights via OpenCensus Azure Monitor Exporter."""
+
+import logging
+import platform
+from typing import Any
+
+from azure.ai.ml._user_agent import USER_AGENT
+
+# Disable internal azure monitor openTelemetry logs
+AZURE_MONITOR_OPENTELEMETRY_LOGGER_NAMESPACE = "azure.monitor.opentelemetry"
+logging.getLogger(AZURE_MONITOR_OPENTELEMETRY_LOGGER_NAMESPACE).addHandler(logging.NullHandler())
+
+AML_INTERNAL_LOGGER_NAMESPACE = "azure.ai.ml._telemetry"
+CONNECTION_STRING = (
+    "InstrumentationKey=71b954a8-6b7d-43f5-986c-3d3a6605d803;"
+    "IngestionEndpoint=https://westus2-0.in.applicationinsights.azure.com/;"
+    "LiveEndpoint=https://westus2.livediagnostics.monitor.azure.com/;"
+    "ApplicationId=82daf08e-6a78-455f-9ce1-9396a8b5128b"
+)
+
+test_subscriptions = [
+    "b17253fa-f327-42d6-9686-f3e553e24763",
+    "test_subscription",
+    "6560575d-fa06-4e7d-95fb-f962e74efd7a",
+    "b17253fa-f327-42d6-9686-f3e553e2452",
+    "74eccef0-4b8d-4f83-b5f9-fa100d155b22",
+    "4faaaf21-663f-4391-96fd-47197c630979",
+    "00000000-0000-0000-0000-000000000",
+]
+
+
+class CustomDimensionsFilter(logging.Filter):
+    """Add application-wide properties to log record"""
+
+    def __init__(self, custom_dimensions=None):  # pylint: disable=super-init-not-called
+        self.custom_dimensions = custom_dimensions or {}
+
+    def filter(self, record: dict) -> bool:  # type: ignore[override]
+        """Adds the default custom_dimensions into the current log record. Does not
+        otherwise filter any records
+
+        :param record: The record
+        :type record: dict
+        :return: True
+        :rtype: bool
+        """
+
+        custom_dimensions = self.custom_dimensions.copy()
+        if isinstance(custom_dimensions, dict):
+            record.__dict__.update(custom_dimensions)
+        return True
+
+
+def in_jupyter_notebook() -> bool:
+    """Checks if user is using a Jupyter Notebook. This is necessary because logging is not allowed in
+    non-Jupyter contexts.
+
+    Adapted from https://stackoverflow.com/a/22424821
+
+    :return: Whether is running in a Jupyter Notebook
+    :rtype: bool
+    """
+    try:  # cspell:ignore ipython
+        from IPython import get_ipython
+
+        if "IPKernelApp" not in get_ipython().config:
+            return False
+    except ImportError:
+        return False
+    except AttributeError:
+        return False
+    return True
+
+
+def setup_azure_monitor(connection_string=None) -> None:
+    """
+    Set up Azure Monitor distro.
+
+    This function sets up Azure Monitor using the provided connection string and specified logger name.
+
+    :param connection_string: The Application Insights connection string.
+    :type connection_string: str
+    :return: None
+    """
+    # Dynamically import the azure.monitor.opentelemetry module to avoid dependency issues later on CLI
+    from azure.monitor.opentelemetry import configure_azure_monitor
+
+    configure_azure_monitor(
+        connection_string=connection_string,
+        logger_name=AML_INTERNAL_LOGGER_NAMESPACE,
+    )
+
+
+# cspell:ignore overriden
+def configure_appinsights_logging(
+    user_agent,
+    connection_string=None,
+    enable_telemetry=True,
+    **kwargs: Any,
+) -> None:
+    """Set the Opentelemetry logging distro for specified logger and connection string to send info to AppInsights.
+
+    :param user_agent: Information about the user's browser.
+    :type user_agent: Dict[str, str]
+    :param connection_string: The Application Insights connection string.
+    :type connection_string: str
+    :param enable_telemetry: Whether to enable telemetry. Will be overriden to False if not in a Jupyter Notebook.
+    :type enable_telemetry: bool
+    :return: None
+    """
+    try:
+        if connection_string is None:
+            connection_string = CONNECTION_STRING
+
+        logger = logging.getLogger(AML_INTERNAL_LOGGER_NAMESPACE)
+        logger.setLevel(logging.INFO)
+        logger.propagate = False
+
+        if (
+            not in_jupyter_notebook()
+            or not enable_telemetry
+            or not user_agent
+            or not user_agent.lower() == USER_AGENT.lower()
+        ):
+            # Disable logging for this logger, all the child loggers will inherit this setting
+            logger.addHandler(logging.NullHandler())
+            return
+
+        if kwargs:
+            if "properties" in kwargs and "subscription_id" in kwargs.get("properties"):  # type: ignore[operator]
+                if kwargs.get("properties")["subscription_id"] in test_subscriptions:  # type: ignore[index]
+                    logger.addHandler(logging.NullHandler())
+                    return
+
+        custom_properties = {"PythonVersion": platform.python_version()}
+        custom_properties.update({"user_agent": user_agent})
+        if "properties" in kwargs:
+            custom_properties.update(kwargs.pop("properties"))
+
+        logger.addFilter(CustomDimensionsFilter(custom_properties))
+
+        setup_azure_monitor(connection_string)
+
+    except Exception:  # pylint: disable=W0718
+        # ignore any exceptions, telemetry collection errors shouldn't block an operation
+        return