firecracker

1
# Copyright 2023 Amazon.com, Inc. or its affiliates. All Rights Reserved.
2
# SPDX-License-Identifier: Apache-2.0
3

4
"""Fixture to send metrics to AWS CloudWatch and validate Firecracker metrics
5

6
We use the aws-embedded-metrics library although it has some sharp corners,
7
namely:
8

9
1. It uses asyncio, which complicates the flushing a bit.
10

11
2. It has an stateful API. Setting dimensions will override previous ones.
12

13
Example:
14

15
    set_dimensions("instance")
16
    put_metric("duration", 1)
17
    set_dimensions("cpu")
18
    put_metric("duration", 1)
19

20
This will end with 2 identical metrics with dimension "cpu" (the last one). The
21
correct way of doing it is:
22

23
    set_dimensions("instance")
24
    put_metric("duration", 1)
25
    flush()
26
    set_dimensions("cpu")
27
    put_metric("duration", 1)
28

29
This is not very intuitive, but we assume all metrics within a test will have
30
the same dimensions.
31

32
# Debugging
33

34
You can override the destination of the metrics to stdout with:
35

36
    AWS_EMF_NAMESPACE=$USER-test
37
    AWS_EMF_ENVIRONMENT=local ./tools/devtest test
38

39
# References:
40

41
- https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html
42
- https://github.com/awslabs/aws-embedded-metrics-python
43
"""
44

45
import asyncio
46
import json
47
import os
48
import socket
49
from urllib.parse import urlparse
50

51
from aws_embedded_metrics.constants import DEFAULT_NAMESPACE
52
from aws_embedded_metrics.logger.metrics_logger_factory import create_metrics_logger
53

54

55
class MetricsWrapperDummy:
56
    """Send metrics to /dev/null"""
57

58
    def set_dimensions(self, *args, **kwargs):
59
        """Set dimensions"""
60

61
    def put_metric(self, *args, **kwargs):
62
        """Put a datapoint with given dimensions"""
63

64
    def set_property(self, *args, **kwargs):
65
        """Set a property"""
66

67
    def flush(self):
68
        """Flush any remaining metrics"""
69

70

71
class MetricsWrapper:
72
    """A convenient metrics logger"""
73

74
    def __init__(self, logger):
75
        self.logger = logger
76

77
    def __getattr__(self, attr):
78
        """Dispatch methods to logger instance"""
79
        if attr not in self.__dict__:
80
            return getattr(self.logger, attr)
81
        return getattr(self, attr)
82

83
    def flush(self):
84
        """Flush any remaining metrics"""
85
        asyncio.run(self.logger.flush())
86

87

88
def get_metrics_logger():
89
    """Get a new metrics logger object"""
90
    # if no metrics namespace, don't output metrics
91
    if "AWS_EMF_NAMESPACE" not in os.environ:
92
        return MetricsWrapperDummy()
93
    logger = create_metrics_logger()
94
    logger.reset_dimensions(False)
95
    return MetricsWrapper(logger)
96

97

98
def emit_raw_emf(emf_msg: dict):
99
    """Emites a raw EMF log message to the local cloudwatch agent"""
100
    if "AWS_EMF_AGENT_ENDPOINT" not in os.environ:
101
        return
102

103
    namespace = os.environ.get("AWS_EMF_NAMESPACE", DEFAULT_NAMESPACE)
104
    emf_msg["_aws"]["LogGroupName"] = os.environ.get(
105
        "AWS_EMF_LOG_GROUP_NAME", f"{namespace}-metrics"
106
    )
107
    emf_msg["_aws"]["LogStreamName"] = os.environ.get("AWS_EMF_LOG_STREAM_NAME", "")
108
    for metrics in emf_msg["_aws"]["CloudWatchMetrics"]:
109
        metrics["Namespace"] = namespace
110

111
    emf_endpoint = urlparse(os.environ["AWS_EMF_AGENT_ENDPOINT"])
112
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
113
        sock.sendto(
114
            (json.dumps(emf_msg) + "\n").encode("utf-8"),
115
            (emf_endpoint.hostname, emf_endpoint.port),
116
        )
117

118

119
UNIT_REDUCTIONS = {
120
    "Microseconds": "Milliseconds",
121
    "Milliseconds": "Seconds",
122
    "Bytes": "Kilobytes",
123
    "Kilobytes": "Megabytes",
124
    "Megabytes": "Gigabytes",
125
    "Gigabytes": "Terabytes",
126
    "Bits": "Kilobits",
127
    "Kilobits": "Megabits",
128
    "Megabits": "Gigabits",
129
    "Gigabits": "Terabit",
130
    "Bytes/Second": "Kilobytes/Second",
131
    "Kilobytes/Second": "Megabytes/Second",
132
    "Megabytes/Second": "Gigabytes/Second",
133
    "Gigabytes/Second": "Terabytes/Second",
134
    "Bits/Second": "Kilobits/Second",
135
    "Kilobits/Second": "Megabits/Second",
136
    "Megabits/Second": "Gigabits/Second",
137
    "Gigabits/Second": "Terabits/Second",
138
}
139
INV_UNIT_REDUCTIONS = {v: k for k, v in UNIT_REDUCTIONS.items()}
140

141

142
UNIT_SHORTHANDS = {
143
    "Seconds": "s",
144
    "Microseconds": "μs",
145
    "Milliseconds": "ms",
146
    "Bytes": "B",
147
    "Kilobytes": "KB",
148
    "Megabytes": "MB",
149
    "Gigabytes": "GB",
150
    "Terabytes": "TB",
151
    "Bits": "Bit",
152
    "Kilobits": "KBit",
153
    "Megabits": "MBit",
154
    "Gigabits": "GBit",
155
    "Terabits": "TBit",
156
    "Percent": "%",
157
    "Count": "",
158
    "Bytes/Second": "B/s",
159
    "Kilobytes/Second": "KB/s",
160
    "Megabytes/Second": "MB/s",
161
    "Gigabytes/Second": "GB/s",
162
    "Terabytes/Second": "TB/s",
163
    "Bits/Second": "Bit/s",
164
    "Kilobits/Second": "KBit/s",
165
    "Megabits/Second": "MBit/s",
166
    "Gigabits/Second": "GBit/s",
167
    "Terabits/Second": "TBit/s",
168
    "Count/Second": "Hz",
169
    "None": "",
170
}
171

172

173
def reduce_value(value, unit):
174
    """
175
    Utility function for expressing a value in the largest possible unit in which it would still be >= 1
176

177
    For example, `reduce_value(1_000_000, Bytes)` would return (1, Megabytes)
178
    """
179
    # Could do this recursively, but I am worried about infinite recursion
180
    # due to precision problems (e.g. infinite loop of dividing/multiplying by 1000, alternating
181
    # between values < 1 and >= 1000).
182
    while abs(value) < 1 and unit in INV_UNIT_REDUCTIONS:
183
        value *= 1000
184
        unit = INV_UNIT_REDUCTIONS[unit]
185
    while abs(value) >= 1000 and unit in UNIT_REDUCTIONS:
186
        value /= 1000
187
        unit = UNIT_REDUCTIONS[unit]
188

189
    return value, unit
190

191

192
def format_with_reduced_unit(value, unit):
193
    """
194
    Utility function for pretty printing a given value by choosing a unit as large as possible,
195
    and then outputting its shorthand.
196

197
    For example, `format_with_reduced_unit(1_000_000, Bytes)` would return "1MB".
198
    """
199
    reduced_value, reduced_unit = reduce_value(value, unit)
200
    formatted_unit = UNIT_SHORTHANDS.get(reduced_unit, reduced_unit)
201

202
    return f"{reduced_value:.2f}{formatted_unit}"
203