Skip to content

deep.api.tracepoint.tracepoint_config

Internal type for configured tracepoints.

LabelExpression

A metric label expression.

Source code in deep/api/tracepoint/tracepoint_config.py 
 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
class LabelExpression:
    """A metric label expression."""

    def __init__(self, key: str, static: Optional[any] = None, expression: Optional[str] = None):
        """
        Create a new label expression.

        :param key: the label key
        :param static: the label static value
        :param expression: the label expression
        """
        self.__key = key
        self.__static = static
        self.__expression = expression

    @property
    def key(self):
        """The label key."""
        return self.__key

    @property
    def static(self):
        """The label static value."""
        return self.__static

    @property
    def expression(self):
        """The label expression."""
        return self.__expression

    def __str__(self) -> str:
        """Represent this object as a string."""
        return str(self.__dict__)

    def __repr__(self) -> str:
        """Represent this object as a string."""
        return self.__str__()

    def __eq__(self, other):
        """Check if other object is equals to this one."""
        if not isinstance(other, LabelExpression):
            return False
        return (
                self.__key == other.__key
                and self.__static == other.__static
                and self.__expression == other.__expression
        )

expression property

The label expression.

key property

The label key.

static property

The label static value.

__eq__(other)

Check if other object is equals to this one.

Source code in deep/api/tracepoint/tracepoint_config.py 
117
118
119
120
121
122
123
124
125
def __eq__(self, other):
    """Check if other object is equals to this one."""
    if not isinstance(other, LabelExpression):
        return False
    return (
            self.__key == other.__key
            and self.__static == other.__static
            and self.__expression == other.__expression
    )

__init__(key, static=None, expression=None)

Create a new label expression.

:param key: the label key :param static: the label static value :param expression: the label expression

Source code in deep/api/tracepoint/tracepoint_config.py 
82
83
84
85
86
87
88
89
90
91
92
def __init__(self, key: str, static: Optional[any] = None, expression: Optional[str] = None):
    """
    Create a new label expression.

    :param key: the label key
    :param static: the label static value
    :param expression: the label expression
    """
    self.__key = key
    self.__static = static
    self.__expression = expression

__repr__()

Represent this object as a string.

Source code in deep/api/tracepoint/tracepoint_config.py 
113
114
115
def __repr__(self) -> str:
    """Represent this object as a string."""
    return self.__str__()

__str__()

Represent this object as a string.

Source code in deep/api/tracepoint/tracepoint_config.py 
109
110
111
def __str__(self) -> str:
    """Represent this object as a string."""
    return str(self.__dict__)

MetricDefinition

The definition of a metric to collect.

Source code in deep/api/tracepoint/tracepoint_config.py 
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
class MetricDefinition:
    """The definition of a metric to collect."""

    def __init__(self, name: str, metric_type: str, labels: List[LabelExpression] = None,
                 expression: Optional[str] = None,
                 namespace: Optional[str] = None, help_str: Optional[str] = None, unit: Optional[str] = None):
        """
        Create a new metric definition.

        :param name: the metric name
        :param labels: the metric labels
        :param metric_type: the metrics type
        :param expression: the metrics expression
        :param namespace: the metric namespace
        :param help_str: the metric help into
        :param unit: the metric unit
        """
        if labels is None:
            labels = []

        self.name = name
        self.labels = labels
        self.type = metric_type
        self.expression = expression
        self.namespace = namespace
        self.help = help_str
        self.unit = unit

    def __str__(self) -> str:
        """Represent this object as a string."""
        return str(self.__dict__)

    def __repr__(self) -> str:
        """Represent this object as a string."""
        return self.__str__()

    def __eq__(self, other):
        """Check if other object is equals to this one."""
        if not isinstance(other, MetricDefinition):
            return False
        return (
                self.name == other.name
                and self.labels == other.labels
                and self.type == other.type
                and self.expression == other.expression
                and self.namespace == other.namespace
                and self.help == other.help
                and self.unit == other.unit
        )

__eq__(other)

Check if other object is equals to this one.

Source code in deep/api/tracepoint/tracepoint_config.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
def __eq__(self, other):
    """Check if other object is equals to this one."""
    if not isinstance(other, MetricDefinition):
        return False
    return (
            self.name == other.name
            and self.labels == other.labels
            and self.type == other.type
            and self.expression == other.expression
            and self.namespace == other.namespace
            and self.help == other.help
            and self.unit == other.unit
    )

__init__(name, metric_type, labels=None, expression=None, namespace=None, help_str=None, unit=None)

Create a new metric definition.

:param name: the metric name :param labels: the metric labels :param metric_type: the metrics type :param expression: the metrics expression :param namespace: the metric namespace :param help_str: the metric help into :param unit: the metric unit

Source code in deep/api/tracepoint/tracepoint_config.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def __init__(self, name: str, metric_type: str, labels: List[LabelExpression] = None,
             expression: Optional[str] = None,
             namespace: Optional[str] = None, help_str: Optional[str] = None, unit: Optional[str] = None):
    """
    Create a new metric definition.

    :param name: the metric name
    :param labels: the metric labels
    :param metric_type: the metrics type
    :param expression: the metrics expression
    :param namespace: the metric namespace
    :param help_str: the metric help into
    :param unit: the metric unit
    """
    if labels is None:
        labels = []

    self.name = name
    self.labels = labels
    self.type = metric_type
    self.expression = expression
    self.namespace = namespace
    self.help = help_str
    self.unit = unit

__repr__()

Represent this object as a string.

Source code in deep/api/tracepoint/tracepoint_config.py 
160
161
162
def __repr__(self) -> str:
    """Represent this object as a string."""
    return self.__str__()

__str__()

Represent this object as a string.

Source code in deep/api/tracepoint/tracepoint_config.py 
156
157
158
def __str__(self) -> str:
    """Represent this object as a string."""
    return str(self.__dict__)

TracePointConfig

This represents the configuration of a single tracepoint.

This is a python version of the GRPC data collected from the LongPoll.

Source code in deep/api/tracepoint/tracepoint_config.py 
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
class TracePointConfig:
    """
    This represents the configuration of a single tracepoint.

    This is a python version of the GRPC data collected from the LongPoll.
    """

    def __init__(self, tp_id: str, path: str, line_no: int, args: dict, watches: List[str],
                 metrics: List[MetricDefinition]):
        """
        Create a new tracepoint config.

        :param tp_id: the tracepoint id
        :param path: the tracepoint source file
        :param line_no: the tracepoint line number
        :param args: the tracepoint args
        :param watches: the tracepoint watches
        """
        self._id = tp_id
        self._path = path
        self._line_no = line_no
        self._args = args
        self._watches = watches

    @property
    def id(self):
        """The tracepoint id."""
        return self._id

    @property
    def path(self):
        """The tracepoint source file."""
        return self._path

    @property
    def line_no(self):
        """The tracepoint line number."""
        # todo need to support missing line number in grpc
        if self._line_no < 0:
            return 0
        return self._line_no

    @property
    def args(self):
        """The tracepoint args."""
        return self._args

    @property
    def watches(self):
        """The tracepoint watches."""
        return self._watches

    @property
    def frame_type(self):
        """The tracepoint frame type."""
        return self.get_arg(FRAME_TYPE, SINGLE_FRAME_TYPE)

    @property
    def stack_type(self):
        """The tracepoint stack type."""
        return self.get_arg(STACK_TYPE, STACK)

    @property
    def fire_count(self):
        """
        Get the allowed number of triggers.

        :return: the configured number of triggers, or -1 for unlimited triggers
        """
        return self.get_arg_int(FIRE_COUNT, 1)

    @property
    def condition(self):
        """The tracepoint condition."""
        return self.get_arg(CONDITION, None)

    def get_arg(self, name: str, default_value: any):
        """
        Get an arg from tracepoint args.

        :param name: the argument name
        :param default_value: the default value
        :return: the value, or the default value
        """
        if name in self._args:
            return self._args[name]
        return default_value

    def get_arg_int(self, name: str, default_value: int):
        """
        Get an argument from the args as an int.

        :param name: the argument name
        :param default_value: the default value to use.
        :return: the value as an int, or the default value
        """
        try:
            return int(self.get_arg(name, default_value))
        except ValueError:
            return default_value

    def __str__(self) -> str:
        """Represent this object as a string."""
        return str({'id': self._id, 'path': self._path, 'line_no': self._line_no, 'args': self._args,
                    'watches': self._watches})

    def __repr__(self) -> str:
        """Represent this object as a string."""
        return self.__str__()

args property

The tracepoint args.

condition property

The tracepoint condition.

fire_count property

Get the allowed number of triggers.

:return: the configured number of triggers, or -1 for unlimited triggers

frame_type property

The tracepoint frame type.

id property

The tracepoint id.

line_no property

The tracepoint line number.

path property

The tracepoint source file.

stack_type property

The tracepoint stack type.

watches property

The tracepoint watches.

__init__(tp_id, path, line_no, args, watches, metrics)

Create a new tracepoint config.

:param tp_id: the tracepoint id :param path: the tracepoint source file :param line_no: the tracepoint line number :param args: the tracepoint args :param watches: the tracepoint watches

Source code in deep/api/tracepoint/tracepoint_config.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def __init__(self, tp_id: str, path: str, line_no: int, args: dict, watches: List[str],
             metrics: List[MetricDefinition]):
    """
    Create a new tracepoint config.

    :param tp_id: the tracepoint id
    :param path: the tracepoint source file
    :param line_no: the tracepoint line number
    :param args: the tracepoint args
    :param watches: the tracepoint watches
    """
    self._id = tp_id
    self._path = path
    self._line_no = line_no
    self._args = args
    self._watches = watches

__repr__()

Represent this object as a string.

Source code in deep/api/tracepoint/tracepoint_config.py 
285
286
287
def __repr__(self) -> str:
    """Represent this object as a string."""
    return self.__str__()

__str__()

Represent this object as a string.

Source code in deep/api/tracepoint/tracepoint_config.py 
280
281
282
283
def __str__(self) -> str:
    """Represent this object as a string."""
    return str({'id': self._id, 'path': self._path, 'line_no': self._line_no, 'args': self._args,
                'watches': self._watches})

get_arg(name, default_value)

Get an arg from tracepoint args.

:param name: the argument name :param default_value: the default value :return: the value, or the default value

Source code in deep/api/tracepoint/tracepoint_config.py 
255
256
257
258
259
260
261
262
263
264
265
def get_arg(self, name: str, default_value: any):
    """
    Get an arg from tracepoint args.

    :param name: the argument name
    :param default_value: the default value
    :return: the value, or the default value
    """
    if name in self._args:
        return self._args[name]
    return default_value

get_arg_int(name, default_value)

Get an argument from the args as an int.

:param name: the argument name :param default_value: the default value to use. :return: the value as an int, or the default value

Source code in deep/api/tracepoint/tracepoint_config.py 
267
268
269
270
271
272
273
274
275
276
277
278
def get_arg_int(self, name: str, default_value: int):
    """
    Get an argument from the args as an int.

    :param name: the argument name
    :param default_value: the default value to use.
    :return: the value as an int, or the default value
    """
    try:
        return int(self.get_arg(name, default_value))
    except ValueError:
        return default_value

TracepointExecutionStats

This keeps track of the tracepoint stats, so we can check fire counts etc.

Source code in deep/api/tracepoint/tracepoint_config.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
class TracepointExecutionStats:
    """This keeps track of the tracepoint stats, so we can check fire counts etc."""

    def __init__(self):
        """Create a new stats object."""
        self._fire_count = 0
        self._last_fire = 0

    def fire(self, ts: int):
        """
        Record a fire.

        Call this to record this tracepoint being triggered.

        :param ts: the time in nanoseconds
        """
        self._fire_count += 1
        self._last_fire = ts

    @property
    def fire_count(self):
        """
        The number of times this tracepoint has fired.

        :return: the number of times this has fired.
        """
        return self._fire_count

    @property
    def last_fire(self):
        """
        The time this tracepoint last fired.

        :return: the time in nanoseconds.
        """
        return self._last_fire

fire_count property

The number of times this tracepoint has fired.

:return: the number of times this has fired.

last_fire property

The time this tracepoint last fired.

:return: the time in nanoseconds.

__init__()

Create a new stats object.

Source code in deep/api/tracepoint/tracepoint_config.py 
293
294
295
296
def __init__(self):
    """Create a new stats object."""
    self._fire_count = 0
    self._last_fire = 0

fire(ts)

Record a fire.

Call this to record this tracepoint being triggered.

:param ts: the time in nanoseconds

Source code in deep/api/tracepoint/tracepoint_config.py 
298
299
300
301
302
303
304
305
306
307
def fire(self, ts: int):
    """
    Record a fire.

    Call this to record this tracepoint being triggered.

    :param ts: the time in nanoseconds
    """
    self._fire_count += 1
    self._last_fire = ts

TracepointWindow

This is used to handle validating the time frame for the tracepoint.

Source code in deep/api/tracepoint/tracepoint_config.py 
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
class TracepointWindow:
    """This is used to handle validating the time frame for the tracepoint."""

    def __init__(self, start: int, end: int):
        """
        Create a new tracepoint window.

        :param start: the window start time
        :param end: the window end time
        """
        self._start = start
        self._end = end

    def in_window(self, ts):
        """
        Is the provided time in the configured window.

        :param ts: time in ms
        :return: true, if the time is within the configured window, else false
        """
        # no window configured - return True
        if self._start == 0 and self._end == 0:
            return True

        # only end configured - return if now is less than end
        if self._start == 0 and self._end > 0:
            return ts <= self._end

        # only start configured - return if now is more than start
        if self._start > 0 and self._end == 0:
            return self._start <= ts

        # if both then check both
        return self._start <= ts <= self._end

__init__(start, end)

Create a new tracepoint window.

:param start: the window start time :param end: the window end time

Source code in deep/api/tracepoint/tracepoint_config.py 
46
47
48
49
50
51
52
53
54
def __init__(self, start: int, end: int):
    """
    Create a new tracepoint window.

    :param start: the window start time
    :param end: the window end time
    """
    self._start = start
    self._end = end

in_window(ts)

Is the provided time in the configured window.

:param ts: time in ms :return: true, if the time is within the configured window, else false

Source code in deep/api/tracepoint/tracepoint_config.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def in_window(self, ts):
    """
    Is the provided time in the configured window.

    :param ts: time in ms
    :return: true, if the time is within the configured window, else false
    """
    # no window configured - return True
    if self._start == 0 and self._end == 0:
        return True

    # only end configured - return if now is less than end
    if self._start == 0 and self._end > 0:
        return ts <= self._end

    # only start configured - return if now is more than start
    if self._start > 0 and self._end == 0:
        return self._start <= ts

    # if both then check both
    return self._start <= ts <= self._end

frame_type_ordinal(frame_type)

Convert a frame type to an ordinal (essentially making it an enum).

This is useful for ordering.

:param frame_type: the frame type :return: the ordinal of the type

Source code in deep/api/tracepoint/tracepoint_config.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def frame_type_ordinal(frame_type) -> int:
    """
    Convert a frame type to an ordinal (essentially making it an enum).

     This is useful for ordering.

    :param frame_type: the frame type
    :return: the ordinal of the type
    """
    if frame_type == SINGLE_FRAME_TYPE:
        return 1
    if frame_type == ALL_FRAME_TYPE:
        return 2
    if frame_type == NO_FRAME_TYPE:
        return 0
    # default to single frame if we do not know the type
    return 1