""" Represents Events and their values. """ import logging import threading from datetime import datetime from enum import Enum from typing import List, Optional from uuid import UUID, uuid4 from samcli.cli.context import Context from samcli.lib.build.workflows import ALL_CONFIGS from samcli.lib.config.file_manager import FILE_MANAGER_MAPPER from samcli.lib.remote_invoke.remote_invoke_executors import RemoteInvokeEventType from samcli.lib.telemetry.telemetry import Telemetry from samcli.local.common.runtime_template import INIT_RUNTIMES LOG = logging.getLogger(__name__) class EventName(Enum): """Enum for the names of available events to track.""" USED_FEATURE = "UsedFeature" BUILD_FUNCTION_RUNTIME = "BuildFunctionRuntime" SYNC_USED = "SyncUsed" SYNC_FLOW_START = "SyncFlowStart" SYNC_FLOW_END = "SyncFlowEnd" BUILD_WORKFLOW_USED = "BuildWorkflowUsed" CONFIG_FILE_EXTENSION = "SamConfigFileExtension" HOOK_CONFIGURATIONS_USED = "HookConfigurationsUsed" REMOTE_INVOKE_EVENT_TYPE = "RemoteInvokeEventType" class UsedFeature(Enum): """Enum for the names of event values of UsedFeature""" ACCELERATE = "Accelerate" CDK = "CDK" INIT_WITH_APPLICATION_INSIGHTS = "InitWithApplicationInsights" CFNLint = "CFNLint" INVOKED_CUSTOM_LAMBDA_AUTHORIZERS = "InvokedLambdaAuthorizers" BUILD_IN_SOURCE = "BuildInSource" class EventType: """Class for Events and the types of values they may have.""" _SYNC_FLOWS = [ "AliasVersionSyncFlow", "AutoDependencyLayerSyncFlow", "AutoDependencyLayerParentSyncFlow", "FunctionSyncFlow", "FunctionLayerReferenceSync", "GenericApiSyncFlow", "HttpApiSyncFlow", "ImageFunctionSyncFlow", "LayerSyncFlow", "RestApiSyncFlow", "StepFunctionsSyncFlow", "ZipFunctionSyncFlow", "InfraSyncExecute", "SkipInfraSyncExecute", ] _WORKFLOWS = [f"{config.language}-{config.dependency_manager}" for config in ALL_CONFIGS] _HOOK_CONFIGURATIONS = [ "TerraformPlanFile", ] _event_values = { # Contains allowable values for Events EventName.USED_FEATURE: [event.value for event in UsedFeature], EventName.BUILD_FUNCTION_RUNTIME: INIT_RUNTIMES, EventName.SYNC_USED: [ "Start", "End", ], EventName.SYNC_FLOW_START: _SYNC_FLOWS, EventName.SYNC_FLOW_END: _SYNC_FLOWS, EventName.BUILD_WORKFLOW_USED: _WORKFLOWS, EventName.CONFIG_FILE_EXTENSION: list(FILE_MANAGER_MAPPER.keys()), EventName.HOOK_CONFIGURATIONS_USED: _HOOK_CONFIGURATIONS, EventName.REMOTE_INVOKE_EVENT_TYPE: list(RemoteInvokeEventType.get_possible_values()), } @staticmethod def get_accepted_values(event_name: EventName) -> List[str]: """Get all acceptable values for a given Event name.""" if event_name not in EventType._event_values: return [] return EventType._event_values[event_name] class Event: """Class to represent Events that occur in SAM CLI.""" event_name: EventName event_value: str # Validated by EventType.get_accepted_values to never be an arbitrary string thread_id: Optional[UUID] # The thread ID; used to group Events from the same command run time_stamp: str exception_name: Optional[str] def __init__( self, event_name: str, event_value: str, thread_id: Optional[UUID] = None, exception_name: Optional[str] = None ): Event._verify_event(event_name, event_value) self.event_name = EventName(event_name) self.event_value = event_value if not thread_id: thread_id = uuid4() self.thread_id = thread_id self.time_stamp = str(datetime.utcnow())[:-3] # format microseconds from 6 -> 3 figures to allow SQL casting self.exception_name = exception_name def __eq__(self, other): return ( self.event_name == other.event_name and self.event_value == other.event_value and self.exception_name == other.exception_name ) def __repr__(self): return ( f"Event(event_name={self.event_name.value}, " f"event_value={self.event_value}, " f"thread_id={self.thread_id.hex}, " f"time_stamp={self.time_stamp})", f"exception_name={self.exception_name})", ) def to_json(self): return { "event_name": self.event_name.value, "event_value": self.event_value, "thread_id": self.thread_id.hex, "time_stamp": self.time_stamp, "exception_name": self.exception_name, } @staticmethod def _verify_event(event_name: str, event_value: str) -> None: """Raise an EventCreationError if either the event name or value is not valid.""" if event_name not in Event._get_event_names(): raise EventCreationError(f"Event '{event_name}' does not exist.") if event_value not in EventType.get_accepted_values(EventName(event_name)): raise EventCreationError(f"Event '{event_name}' does not accept value '{event_value}'.") @staticmethod def _get_event_names() -> List[str]: """Retrieves a list of all valid event names.""" return [event.value for event in EventName] class EventTracker: """Class to track and recreate Events as they occur.""" _events: List[Event] = [] _event_lock = threading.Lock() _session_id: Optional[str] = None _command_name: Optional[str] = None MAX_EVENTS: int = 50 # Maximum number of events to store before sending @staticmethod def track_event( event_name: str, event_value: str, session_id: Optional[str] = None, thread_id: Optional[UUID] = None, exception_name: Optional[str] = None, ): """Method to track an event where and when it occurs. Place this method in the codepath of the event that you would like to track. For instance, if you would like to track when FeatureX is used, append this method to the end of that function. Parameters ---------- event_name: str The name of the Event. Must be a valid EventName value, or an EventCreationError will be thrown. event_value: str The value of the Event. Must be a valid EventType value for the passed event_name, or an EventCreationError will be thrown. session_id: Optional[str] The session ID to set to link back to the original command run thread_id: Optional[UUID] The thread ID of the Event to track, as a UUID. exception_name: Optional[str] The name of the exception that this event encountered when tracking a feature Examples -------- >>> def feature_x(...): # do things EventTracker.track_event("UsedFeature", "FeatureX") >>> def feature_y(...) -> Any: # do things EventTracker.track_event("UsedFeature", "FeatureY") return some_value """ if session_id: EventTracker._session_id = session_id try: should_send: bool = False # Validate the thread ID if not thread_id: # Passed value is not a UUID or None thread_id = uuid4() with EventTracker._event_lock: EventTracker._events.append( Event(event_name, event_value, thread_id=thread_id, exception_name=exception_name) ) # Get properties from the click context (needed for multithreading sending) EventTracker._set_context_property("_session_id", "session_id") EventTracker._set_context_property("_command_name", "command_path") if len(EventTracker._events) >= EventTracker.MAX_EVENTS: should_send = True if should_send: EventTracker.send_events() except EventCreationError as e: LOG.debug("Error occurred while trying to track an event: %s", e) @staticmethod def get_tracked_events() -> List[Event]: """Retrieve a list of all currently tracked Events.""" with EventTracker._event_lock: return EventTracker._events @staticmethod def clear_trackers(): """Clear the current list of tracked Events before the next session.""" with EventTracker._event_lock: EventTracker._events = [] @staticmethod def send_events() -> threading.Thread: """Call a thread to send the current list of Events via Telemetry.""" send_thread = threading.Thread(target=EventTracker._send_events_in_thread) send_thread.start() return send_thread @staticmethod def _set_context_property(event_prop: str, context_prop: str) -> None: """ Set a click context property in the event so that it is emitted when the metric is sent. This is required since the event is sent in a separate thread and no longer has access to the click context that the command was initially called with. As a workaround, we set the property here first so that it's available when calling the metrics endpoint. Parameters ---------- event_prop: str Property name to be stored in the event and consumed when emitting the metric context_prop: str Property name for the target property from the context object """ if not getattr(EventTracker, event_prop): try: ctx = Context.get_current_context() if ctx: setattr(EventTracker, event_prop, getattr(ctx, context_prop)) except RuntimeError: LOG.debug("EventTracker: Unable to obtain %s", context_prop) @staticmethod def _send_events_in_thread(): """Send the current list of Events via Telemetry.""" from samcli.lib.telemetry.metric import Metric # pylint: disable=cyclic-import msa = {} with EventTracker._event_lock: if not EventTracker._events: # Don't do anything if there are no events to send return msa["events"] = [e.to_json() for e in EventTracker._events] EventTracker._events = [] # Manual clear_trackers() since we're within the lock telemetry = Telemetry() metric = Metric("events") metric.add_data("sessionId", EventTracker._session_id) metric.add_data("commandName", EventTracker._command_name) metric.add_data("metricSpecificAttributes", msa) telemetry.emit(metric) def track_long_event( start_event_name: str, start_event_value: str, end_event_name: str, end_event_value: str, thread_id: Optional[UUID] = None, ): """Decorator for tracking events that occur at start and end of a function. The decorator tracks two Events total, where the first Event occurs at the start of the decorated function's execution (prior to its first line) and the second Event occurs after the function has ended (after the final line of the function has executed). If this decorator is being placed in a function that also contains the `track_command` decorator, ensure that this decorator is placed BELOW `track_command`. Otherwise, the current list of Events will be sent before the end_event will be added, resulting in an additional 'events' metric with only that single Event. Parameters ---------- start_event_name: str The name of the Event that is executed at the start of the decorated function's execution. Must be a valid EventName value or the decorator will not run. start_event_value: str The value of the Event that is executed at the start of the decorated function's execution. Must be a valid EventType value for the passed `start_event_name` or the decorator will not run. end_event_name: str The name of the Event that is executed at the end of the decorated function's execution. Must be a valid EventName value or the decorator will not run. end_event_value: str The value of the Event that is executed at the end of the decorated function's execution. Must be a valid EventType value for the passed `end_event_name` or the decorator will not run. thread_id: Optional[UUID] The thread ID of the Events to track, as a UUID. Examples -------- >>> @track_long_event("FuncStart", "Func1", "FuncEnd", "Func1") def func1(...): # do things >>> @track_long_event("FuncStart", "Func2", "FuncEnd", "Func2") def func2(...): # do things """ should_track = True try: # Check that passed values are valid Events Event(start_event_name, start_event_value) Event(end_event_name, end_event_value) # Validate the thread ID if not thread_id: # Passed value is not a UUID or None thread_id = uuid4() except EventCreationError as e: LOG.debug("Error occurred while trying to track an event: %s\nDecorator not run.", e) should_track = False def decorator_for_events(func): """The actual decorator""" def wrapped(*args, **kwargs): # Track starting event if should_track: EventTracker.track_event(start_event_name, start_event_value, thread_id=thread_id) exception = None # Run the function try: return_value = func(*args, **kwargs) except Exception as e: exception = e # Track ending event if should_track: EventTracker.track_event(end_event_name, end_event_value, thread_id=thread_id) EventTracker.send_events() # Ensure Events are sent at the end of execution if exception: raise exception return return_value return wrapped return decorator_for_events class EventCreationError(Exception): """Exception called when an Event is not properly created."""