deep.processor.context.snapshot_action

Handling for snapshot actions.

DeferredSnapshotActionCallback

Bases: ActionCallback

Defer the send action to the end of the line or function.

Source code in deep/processor/context/snapshot_action.py

class DeferredSnapshotActionCallback(ActionCallback):
    """Defer the send action to the end of the line or function."""

    def __init__(self, action_context: ActionContext, snapshot: EventSnapshot):
        """
        Create a new action callback.

        :param action_context: the triggering action context
        :param snapshot: the generated snapshot
        """
        self.__action_context = action_context
        self.__snapshot = snapshot

    def process(self, ctx: 'TriggerContext', event: str, frame: FrameType, arg: any) -> bool:
        """
        Process a callback.

        :param ctx: the context for this trigger
        :param event: the event
        :param frame: the frame data
        :param arg: the arg from settrace
        :return: True, to keep this callback until next match.
        """
        if event in ['exception', 'return']:
            watch, new_vars, _ = self.__action_context.process_capture_variable(event, arg)
            self.__snapshot.add_watch_result(watch)
            self.__snapshot.merge_var_lookup(new_vars)

        ctx.push_service.push_snapshot(self.__snapshot)
        return False

__init__(action_context, snapshot)

Create a new action callback.

:param action_context: the triggering action context :param snapshot: the generated snapshot

Source code in deep/processor/context/snapshot_action.py

def __init__(self, action_context: ActionContext, snapshot: EventSnapshot):
    """
    Create a new action callback.

    :param action_context: the triggering action context
    :param snapshot: the generated snapshot
    """
    self.__action_context = action_context
    self.__snapshot = snapshot

process(ctx, event, frame, arg)

Process a callback.

:param ctx: the context for this trigger :param event: the event :param frame: the frame data :param arg: the arg from settrace :return: True, to keep this callback until next match.

Source code in deep/processor/context/snapshot_action.py

def process(self, ctx: 'TriggerContext', event: str, frame: FrameType, arg: any) -> bool:
    """
    Process a callback.

    :param ctx: the context for this trigger
    :param event: the event
    :param frame: the frame data
    :param arg: the arg from settrace
    :return: True, to keep this callback until next match.
    """
    if event in ['exception', 'return']:
        watch, new_vars, _ = self.__action_context.process_capture_variable(event, arg)
        self.__snapshot.add_watch_result(watch)
        self.__snapshot.merge_var_lookup(new_vars)

    ctx.push_service.push_snapshot(self.__snapshot)
    return False

DeferredSnapshotActionResult

Bases: ActionResult

The result of a deferred snapshot action.

Source code in deep/processor/context/snapshot_action.py

class DeferredSnapshotActionResult(ActionResult):
    """The result of a deferred snapshot action."""

    def __init__(self, action_context: ActionContext, snapshot: EventSnapshot):
        """
        Create a new snapshot action result.

        :param action_context: the action context that created this result
        :param snapshot: the snapshot result
        """
        self.action_context = action_context
        self.snapshot = snapshot

    def process(self, ctx: 'TriggerContext') -> Optional[ActionCallback]:
        """
        Process this result.

        :param ctx: the triggering context

        :return: an action callback if we need to do something at the 'end', or None
        """
        snapshot = self._decorate_snapshot(ctx)
        return DeferredSnapshotActionCallback(self.action_context, snapshot)

    def _decorate_snapshot(self, ctx):
        attributes = BoundedAttributes(
            attributes={'context': ctx.id, 'tracepoint': self.action_context.location_action.tracepoint.id},
            immutable=False)
        for decorator in ctx.config.snapshot_decorators:
            try:
                decorate = decorator.decorate(self.snapshot.id_str, self.action_context)
                if decorate is not None:
                    attributes.merge_in(decorate)
            except Exception:
                deep.logging.exception("Failed to decorate snapshot: %s ", decorator)
        self.snapshot.attributes.merge_in(attributes)
        return self.snapshot

__init__(action_context, snapshot)

Create a new snapshot action result.

:param action_context: the action context that created this result :param snapshot: the snapshot result

Source code in deep/processor/context/snapshot_action.py

def __init__(self, action_context: ActionContext, snapshot: EventSnapshot):
    """
    Create a new snapshot action result.

    :param action_context: the action context that created this result
    :param snapshot: the snapshot result
    """
    self.action_context = action_context
    self.snapshot = snapshot

process(ctx)

Process this result.

:param ctx: the triggering context

:return: an action callback if we need to do something at the 'end', or None

Source code in deep/processor/context/snapshot_action.py

def process(self, ctx: 'TriggerContext') -> Optional[ActionCallback]:
    """
    Process this result.

    :param ctx: the triggering context

    :return: an action callback if we need to do something at the 'end', or None
    """
    snapshot = self._decorate_snapshot(ctx)
    return DeferredSnapshotActionCallback(self.action_context, snapshot)

SendSnapshotActionResult

Bases: DeferredSnapshotActionResult

The result of a successful snapshot action.

Source code in deep/processor/context/snapshot_action.py

class SendSnapshotActionResult(DeferredSnapshotActionResult):
    """The result of a successful snapshot action."""

    def __init__(self, action_context: ActionContext, snapshot: EventSnapshot):
        """
        Create a new snapshot action result.

        :param action_context: the action context that created this result
        :param snapshot: the snapshot result
        """
        super().__init__(action_context, snapshot)

    def process(self, ctx: 'TriggerContext') -> Optional[ActionCallback]:
        """
        Process this result.

        :param ctx: the triggering context

        :return: an action callback if we need to do something at the 'end', or None
        """
        snapshot = self._decorate_snapshot(ctx)
        ctx.push_service.push_snapshot(snapshot)
        return None

__init__(action_context, snapshot)

Create a new snapshot action result.

:param action_context: the action context that created this result :param snapshot: the snapshot result

Source code in deep/processor/context/snapshot_action.py

def __init__(self, action_context: ActionContext, snapshot: EventSnapshot):
    """
    Create a new snapshot action result.

    :param action_context: the action context that created this result
    :param snapshot: the snapshot result
    """
    super().__init__(action_context, snapshot)

process(ctx)

Process this result.

:param ctx: the triggering context

:return: an action callback if we need to do something at the 'end', or None

Source code in deep/processor/context/snapshot_action.py

def process(self, ctx: 'TriggerContext') -> Optional[ActionCallback]:
    """
    Process this result.

    :param ctx: the triggering context

    :return: an action callback if we need to do something at the 'end', or None
    """
    snapshot = self._decorate_snapshot(ctx)
    ctx.push_service.push_snapshot(snapshot)
    return None

SnapshotActionContext

Bases: FrameCollectorContext, ActionContext

The context to use when capturing a snapshot.

Source code in deep/processor/context/snapshot_action.py

class SnapshotActionContext(FrameCollectorContext, ActionContext):
    """The context to use when capturing a snapshot."""

    @property
    def max_tp_process_time(self) -> int:
        """The max time to spend processing a tracepoint."""
        return self.location_action.config.get('MAX_TP_PROCESS_TIME', 100)

    @property
    def collection_config(self) -> VariableProcessorConfig:
        """The variable processing config."""
        config = VariableProcessorConfig()
        config.max_string_length = self.location_action.config.get('MAX_STRING_LENGTH',
                                                                   config.DEFAULT_MAX_STRING_LENGTH)
        config.max_collection_size = self.location_action.config.get('MAX_COLLECTION_SIZE',
                                                                     config.DEFAULT_MAX_COLLECTION_SIZE)
        config.max_variables = self.location_action.config.get('MAX_VARIABLES', config.DEFAULT_MAX_VARIABLES)
        config.max_var_depth = self.location_action.config.get('MAX_VAR_DEPTH', config.DEFAULT_MAX_VAR_DEPTH)
        return config

    @property
    def ts(self) -> int:
        """The timestamp in nanoseconds for this trigger."""
        return self.trigger_context.ts

    def should_collect_vars(self, current_frame_index: int) -> bool:
        """
        Check if we can collect data for a frame.

        Frame indexes start from 0 (as the current frame) and increase as we go back up the stack.

        :param (int) current_frame_index: the current frame index.
        :return (bool): if we should collect the frame vars.
        """
        config_type = self.location_action.config.get(FRAME_TYPE, SINGLE_FRAME_TYPE)
        if config_type == NO_FRAME_TYPE:
            return False
        if config_type == ALL_FRAME_TYPE:
            return True
        return current_frame_index == 0

    def is_app_frame(self, filename: str) -> Tuple[bool, str]:
        """
        Check if the current frame is a user application frame.

        :param filename: the frame file name
        :return: True if add frame, else False
        """
        return self.trigger_context.config.is_app_frame(filename)

    @property
    def watches(self):
        """The configured watches."""
        return self.location_action.config.get("watches", [])

    @property
    def log_msg(self):
        """The configured log message on the tracepoint."""
        return self.location_action.config.get(LOG_MSG, None)

    def _process_action(self):
        collector = FrameCollector(self, self.trigger_context.frame)

        frames, variables = collector.collect(self.trigger_context.vars, self.trigger_context.var_cache)

        snapshot = EventSnapshot(self.location_action.tracepoint, self.trigger_context.ts,
                                 self.trigger_context.resource, frames, variables)

        # process the snapshot watches
        for watch in self.watches:
            result, watch_lookup, _ = self.eval_watch(watch, WATCH_SOURCE_WATCH)
            snapshot.add_watch_result(result)
            snapshot.merge_var_lookup(watch_lookup)

        log_msg = self.log_msg
        if log_msg is not None:
            # create and process the log message
            context = LogActionContext(self.trigger_context, LocationAction(self.location_action.id, None, {
                LOG_MSG: log_msg,
            }, LocationAction.ActionType.Log))
            log, watches, log_vars = context.process_log(log_msg)
            snapshot.log_msg = log
            for watch in watches:
                snapshot.add_watch_result(watch)
            snapshot.merge_var_lookup(log_vars)
            self.trigger_context.attach_result(LogActionResult(context.location_action, log))

        if self.trigger_context.event in ['exception', 'return']:
            watch, new_vars, _ = self.process_capture_variable(self.trigger_context.event, self.trigger_context.arg)
            snapshot.add_watch_result(watch)
            snapshot.merge_var_lookup(new_vars)

        snapshot.complete()
        if self._is_deferred():
            self.trigger_context.attach_result(DeferredSnapshotActionResult(self, snapshot))
        else:
            self.trigger_context.attach_result(SendSnapshotActionResult(self, snapshot))

    def _is_deferred(self):
        stage = self.location_action.config.get(STAGE, None)
        if stage is None:
            return False
        return stage == LINE_CAPTURE or stage == METHOD_CAPTURE

collection_config: VariableProcessorConfig property

The variable processing config.

log_msg property

The configured log message on the tracepoint.

max_tp_process_time: int property

The max time to spend processing a tracepoint.

ts: int property

The timestamp in nanoseconds for this trigger.

watches property

The configured watches.

is_app_frame(filename)

Check if the current frame is a user application frame.

:param filename: the frame file name :return: True if add frame, else False

Source code in deep/processor/context/snapshot_action.py

def is_app_frame(self, filename: str) -> Tuple[bool, str]:
    """
    Check if the current frame is a user application frame.

    :param filename: the frame file name
    :return: True if add frame, else False
    """
    return self.trigger_context.config.is_app_frame(filename)

should_collect_vars(current_frame_index)

Check if we can collect data for a frame.

Frame indexes start from 0 (as the current frame) and increase as we go back up the stack.

:param (int) current_frame_index: the current frame index. :return (bool): if we should collect the frame vars.

Source code in deep/processor/context/snapshot_action.py

def should_collect_vars(self, current_frame_index: int) -> bool:
    """
    Check if we can collect data for a frame.

    Frame indexes start from 0 (as the current frame) and increase as we go back up the stack.

    :param (int) current_frame_index: the current frame index.
    :return (bool): if we should collect the frame vars.
    """
    config_type = self.location_action.config.get(FRAME_TYPE, SINGLE_FRAME_TYPE)
    if config_type == NO_FRAME_TYPE:
        return False
    if config_type == ALL_FRAME_TYPE:
        return True
    return current_frame_index == 0