Skip to content

deep.grpc

Collection of functions to convert to protobuf version of types.

We do not use the protobuf types throughout the project as they do not autocomplete or have type definitions that work in IDE. It also makes it easier to deal with agent functionality by having local types we can modify.

convert_label_expressions(label_expressions)

Convert a label expression.

:param label_expressions: the expression to convert. :return: the converted expression

Source code in deep/grpc/__init__.py 
 94
 95
 96
 97
 98
 99
100
101
102
def convert_label_expressions(label_expressions) -> List[LabelExpression]:
    """
    Convert a label expression.

    :param label_expressions: the expression to convert.
    :return: the converted expression
    """
    return [LabelExpression(label.key, __convert_static_value(label), label.expression) for
            label in label_expressions]

convert_resource(resource)

Convert an internal resource to GRPC type.

:param resource: the resource to convert :return: the converted type as GRPC.

Source code in deep/grpc/__init__.py 
71
72
73
74
75
76
77
78
def convert_resource(resource):
    """
    Convert an internal resource to GRPC type.

    :param resource: the resource to convert
    :return: the converted type as GRPC.
    """
    return __convert_attributes(resource.attributes)

convert_response(response)

Convert a response from GRPC to internal types.

This function should create a list of Triggers from the incoming configuration. The Trigger should be a location with one or more actions to perform at that location.

:param response: the response from the poll request :return: a list of trigger locations with the appropriate actions

Source code in deep/grpc/__init__.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def convert_response(response) -> List[Trigger]:
    """
    Convert a response from GRPC to internal types.

    This function should create a list of Triggers from the incoming configuration. The Trigger should be a
    location with one or more actions to perform at that location.

    :param response: the response from the poll request
    :return: a list of trigger locations with the appropriate actions
    """
    all_triggers: Dict[str, Trigger] = {}
    for r in response:
        # from the incoming tracepoints create a Trigger with actions
        trigger = build_trigger(r.ID, r.path, r.line_number, dict(r.args), [w for w in r.watches],
                                __convert_metric_definition(r.metrics))
        location_id = trigger.id
        # if we already have a trigger for this location then merge the new actions into it
        if location_id in all_triggers:
            all_triggers[location_id].merge_actions(trigger.actions)
        else:
            all_triggers[location_id] = trigger

    return list(all_triggers.values())

convert_value(value)

Convert a value from the python type.

:param value: the value to convert :return: the value wrapped in the appropriate AnyValue type.

Source code in deep/grpc/__init__.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def convert_value(value):
    """
    Convert a value from the python type.

    :param value: the value to convert
    :return: the value wrapped in the appropriate AnyValue type.
    """
    """Convert the attributes to jaeger tags."""
    if isinstance(value, bool):
        return AnyValue(bool_value=value)
    if isinstance(value, str):
        return AnyValue(string_value=value)
    if isinstance(value, int):
        return AnyValue(int_value=value)
    if isinstance(value, float):
        return AnyValue(double_value=value)
    if isinstance(value, bytes):
        return AnyValue(bytes_value=value)
    if isinstance(value, dict):
        return AnyValue(kvlist_value=__value_as_dict(value))
    if isinstance(value, list):
        return AnyValue(array_value=__value_as_list(value))

    return None