forked from aws-powertools/powertools-lambda-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_metrics.py
More file actions
1173 lines (872 loc) · 43.4 KB
/
test_metrics.py
File metadata and controls
1173 lines (872 loc) · 43.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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
77
78
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
126
127
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
177
178
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
288
289
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
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import json
import warnings
from collections import namedtuple
from typing import Any, Dict, List, Union
import pytest
from aws_lambda_powertools import Metrics, single_metric
from aws_lambda_powertools.metrics import (
EphemeralMetrics,
MetricResolution,
MetricResolutionError,
MetricUnit,
MetricUnitError,
MetricValueError,
SchemaValidationError,
)
from aws_lambda_powertools.metrics.base import (
MAX_DIMENSIONS,
MetricManager,
reset_cold_start_flag,
)
@pytest.fixture(scope="function", autouse=True)
def reset_metric_set():
metrics = Metrics()
metrics.clear_metrics()
metrics.clear_default_dimensions()
reset_cold_start_flag() # ensure each test has cold start
yield
@pytest.fixture
def metric_with_resolution() -> Dict[str, Union[str, int]]:
return {"name": "single_metric", "unit": MetricUnit.Count, "value": 1, "resolution": MetricResolution.High}
@pytest.fixture
def metric() -> Dict[str, str]:
return {"name": "single_metric", "unit": MetricUnit.Count, "value": 1}
@pytest.fixture
def metrics() -> List[Dict[str, str]]:
return [
{"name": "metric_one", "unit": MetricUnit.Count, "value": 1},
{"name": "metric_two", "unit": MetricUnit.Count, "value": 1},
]
@pytest.fixture
def metrics_same_name() -> List[Dict[str, str]]:
return [
{"name": "metric_one", "unit": MetricUnit.Count, "value": 1},
{"name": "metric_one", "unit": MetricUnit.Count, "value": 5},
]
@pytest.fixture
def dimension() -> Dict[str, str]:
return {"name": "test_dimension", "value": "test"}
@pytest.fixture
def dimensions() -> List[Dict[str, str]]:
return [
{"name": "test_dimension", "value": "test"},
{"name": "test_dimension_2", "value": "test"},
]
@pytest.fixture
def non_str_dimensions() -> List[Dict[str, Any]]:
return [
{"name": "test_dimension", "value": True},
{"name": "test_dimension_2", "value": 3},
]
@pytest.fixture
def namespace() -> str:
return "test_namespace"
@pytest.fixture
def service() -> str:
return "test_service"
@pytest.fixture
def metadata() -> Dict[str, str]:
return {"key": "username", "value": "test"}
@pytest.fixture
def a_hundred_metrics() -> List[Dict[str, str]]:
return [{"name": f"metric_{i}", "unit": "Count", "value": 1} for i in range(100)]
@pytest.fixture
def a_hundred_metric_values() -> List[Dict[str, str]]:
return [{"name": "metric", "unit": "Count", "value": i} for i in range(100)]
def serialize_metrics(
metrics: List[Dict],
dimensions: List[Dict],
namespace: str,
metadatas: List[Dict] = None,
) -> Dict:
"""Helper function to build EMF object from a list of metrics, dimensions"""
my_metrics = MetricManager(namespace=namespace)
for dimension in dimensions:
my_metrics.add_dimension(**dimension)
for metric in metrics:
my_metrics.add_metric(**metric)
if metadatas is not None:
for metadata in metadatas:
my_metrics.add_metadata(**metadata)
if len(metrics) != 100:
return my_metrics.serialize_metric_set()
def serialize_single_metric(metric: Dict, dimension: Dict, namespace: str, metadata: Dict = None) -> Dict:
"""Helper function to build EMF object from a given metric, dimension and namespace"""
my_metrics = MetricManager(namespace=namespace)
my_metrics.add_metric(**metric)
my_metrics.add_dimension(**dimension)
if metadata is not None:
my_metrics.add_metadata(**metadata)
return my_metrics.serialize_metric_set()
def remove_timestamp(metrics: List):
"""Helper function to remove Timestamp key from EMF objects as they're built at serialization"""
for metric in metrics:
del metric["_aws"]["Timestamp"]
def capture_metrics_output(capsys):
return json.loads(capsys.readouterr().out.strip())
def capture_metrics_output_multiple_emf_objects(capsys):
return [json.loads(line.strip()) for line in capsys.readouterr().out.split("\n") if line]
def test_single_metric_logs_with_high_resolution_enum(capsys, metric_with_resolution, dimension, namespace):
# GIVEN we have a metric with high resolution as enum
# WHEN using single_metric context manager
with single_metric(namespace=namespace, **metric_with_resolution) as my_metric:
my_metric.add_dimension(**dimension)
# THEN we should only have the first metric added
output = capture_metrics_output(capsys)
expected = serialize_single_metric(metric=metric_with_resolution, dimension=dimension, namespace=namespace)
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_single_metric_logs_with_high_resolution_integer(capsys, metric_with_resolution, dimension, namespace):
# GIVEN we have a metric with high resolution as integer
metric_with_resolution["resolution"] = MetricResolution.High.value
# WHEN using single_metric context manager
with single_metric(namespace=namespace, **metric_with_resolution) as my_metric:
my_metric.add_dimension(**dimension)
# THEN we should only have the first metric added
output = capture_metrics_output(capsys)
expected = serialize_single_metric(metric=metric_with_resolution, dimension=dimension, namespace=namespace)
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_single_metric_logs_one_metric_only(capsys, metric, dimension, namespace):
# GIVEN we try adding more than one metric
# WHEN using single_metric context manager
with single_metric(namespace=namespace, **metric) as my_metric:
my_metric.add_metric(name="second_metric", unit="Count", value=1)
my_metric.add_dimension(**dimension)
output = capture_metrics_output(capsys)
expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)
# THEN we should only have the first metric added
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_single_metric_default_dimensions(capsys, metric, dimension, namespace):
# GIVEN we provide default dimensions
# WHEN using single_metric context manager
default_dimensions = {dimension["name"]: dimension["value"]}
with single_metric(namespace=namespace, default_dimensions=default_dimensions, **metric) as my_metric:
my_metric.add_metric(name="second_metric", unit="Count", value=1)
output = capture_metrics_output(capsys)
expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)
# THEN we should have default dimension added to the metric
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_single_metric_default_dimensions_inherit(capsys, metric, dimension, namespace):
# GIVEN we provide Metrics default dimensions
# WHEN using single_metric context manager
metrics = Metrics()
default_dimensions = {dimension["name"]: dimension["value"]}
metrics.set_default_dimensions(**default_dimensions)
with single_metric(namespace=namespace, default_dimensions=metrics.default_dimensions, **metric) as my_metric:
my_metric.add_metric(name="second_metric", unit="Count", value=1)
output = capture_metrics_output(capsys)
expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)
# THEN we should have default dimension added to the metric
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_log_metrics(capsys, metrics, dimensions, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(namespace=namespace)
for metric in metrics:
my_metrics.add_metric(**metric)
for dimension in dimensions:
my_metrics.add_dimension(**dimension)
# WHEN we utilize log_metrics to serialize
# and flush all metrics at the end of a function execution
@my_metrics.log_metrics
def lambda_handler(evt, ctx):
pass
lambda_handler({}, {})
output = capture_metrics_output(capsys)
expected = serialize_metrics(metrics=metrics, dimensions=dimensions, namespace=namespace)
# THEN we should have no exceptions
# and a valid EMF object should be flushed correctly
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_log_metrics_manual_flush(capsys, metrics, dimensions, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(namespace=namespace)
for metric in metrics:
my_metrics.add_metric(**metric)
for dimension in dimensions:
my_metrics.add_dimension(**dimension)
# WHEN we manually the metrics
my_metrics.flush_metrics()
output = capture_metrics_output(capsys)
expected = serialize_metrics(metrics=metrics, dimensions=dimensions, namespace=namespace)
# THEN we should have no exceptions
# and a valid EMF object should be flushed correctly
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_namespace_env_var(monkeypatch, capsys, metric, dimension, namespace):
# GIVEN POWERTOOLS_METRICS_NAMESPACE is set
monkeypatch.setenv("POWERTOOLS_METRICS_NAMESPACE", namespace)
# WHEN creating a metric without explicitly adding a namespace
with single_metric(**metric) as my_metric:
my_metric.add_dimension(**dimension)
output = capture_metrics_output(capsys)
expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)
# THEN we should add a namespace using POWERTOOLS_METRICS_NAMESPACE env var value
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_service_env_var(monkeypatch, capsys, metric, namespace):
# GIVEN we use POWERTOOLS_SERVICE_NAME
monkeypatch.setenv("POWERTOOLS_SERVICE_NAME", "test_service")
# WHEN creating a metric without explicitly adding a dimension
with single_metric(**metric, namespace=namespace):
pass
output = capture_metrics_output(capsys)
expected_dimension = {"name": "service", "value": "test_service"}
expected = serialize_single_metric(metric=metric, dimension=expected_dimension, namespace=namespace)
# THEN a metric should be logged using the implicitly created "service" dimension
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_service_env_var_with_metrics_instance(monkeypatch, capsys, metric, namespace, service):
# GIVEN we use POWERTOOLS_SERVICE_NAME
monkeypatch.setenv("POWERTOOLS_SERVICE_NAME", service)
# WHEN initializing Metrics without an explicit service name
metrics = Metrics(namespace=namespace)
metrics.add_metric(**metric)
@metrics.log_metrics
def lambda_handler(_, __):
pass
lambda_handler({}, {})
output = capture_metrics_output(capsys)
expected_dimension = {"name": "service", "value": service}
expected = serialize_single_metric(metric=metric, dimension=expected_dimension, namespace=namespace)
# THEN a metric should be logged using the implicitly created "service" dimension
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_metrics_spillover(monkeypatch, capsys, metric, dimension, namespace, a_hundred_metrics):
# GIVEN Metrics is initialized and we have over a hundred metrics to add
my_metrics = Metrics(namespace=namespace)
my_metrics.add_dimension(**dimension)
# WHEN we add more than 100 metrics
for _metric in a_hundred_metrics:
my_metrics.add_metric(**_metric)
# THEN it should serialize and flush all metrics at the 100th
# and clear all metrics and dimensions from memory
output = capture_metrics_output(capsys)
spillover_metrics = output["_aws"]["CloudWatchMetrics"][0]["Metrics"]
assert my_metrics.metric_set == {}
assert len(spillover_metrics) == 100
# GIVEN we add the 101th metric
# WHEN we already had a Metric class instance
# with an existing dimension set from the previous 100th metric batch
my_metrics.add_metric(**metric)
# THEN serializing the 101th metric should
# create a new EMF object with a single metric in it (101th)
# and contain the same dimension we previously added
serialized_101th_metric = my_metrics.serialize_metric_set()
expected_101th_metric = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)
remove_timestamp(metrics=[serialized_101th_metric, expected_101th_metric])
assert serialized_101th_metric == expected_101th_metric
def test_metric_values_spillover(monkeypatch, capsys, dimension, namespace, a_hundred_metric_values):
# GIVEN Metrics is initialized and we have over a hundred metric values to add
my_metrics = Metrics(namespace=namespace)
my_metrics.add_dimension(**dimension)
metric = a_hundred_metric_values[0]
# WHEN we add 100 metric values
for _metric in a_hundred_metric_values:
my_metrics.add_metric(**_metric)
# THEN it should serialize and flush the metric at the 100th value
# and clear all metrics and dimensions from memory
output = capture_metrics_output(capsys)
spillover_values = output[metric["name"]]
assert my_metrics.metric_set == {}
assert len(spillover_values) == 100
# GIVEN we add the 101st metric
# WHEN we already had a Metric class instance
# with an existing dimension set from the previous 100th metric batch
my_metrics.add_metric(**metric)
# THEN serializing the 101st value should
# create a new EMF object with a single value in it (101st)
# and contain the same dimension we previously added
serialized_101st_metric = my_metrics.serialize_metric_set()
expected_101st_metric = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)
remove_timestamp(metrics=[serialized_101st_metric, expected_101st_metric])
assert serialized_101st_metric == expected_101st_metric
def test_log_metrics_decorator_call_decorated_function(metric, namespace, service):
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
# WHEN log_metrics is used to serialize metrics
@my_metrics.log_metrics
def lambda_handler(evt, context):
return True
# THEN log_metrics should invoke the function it decorates
# and return no error if we have a namespace and dimension
assert lambda_handler({}, {}) is True
def test_schema_validation_incorrect_metric_resolution(metric, dimension):
# GIVEN we pass a metric resolution that is not supported by CloudWatch
metric["resolution"] = 10 # metric resolution must be 1 (High) or 60 (Standard)
# WHEN we try adding a new metric
# THEN it should fail metric unit validation
with pytest.raises(MetricResolutionError, match="Invalid metric resolution.*60"):
with single_metric(**metric) as my_metric:
my_metric.add_dimension(**dimension)
@pytest.mark.parametrize("resolution", ["sixty", False, [], {}, object])
def test_schema_validation_incorrect_metric_resolution_non_integer_enum(metric, dimension, resolution, namespace):
# GIVEN we pass a metric resolution that is not supported by CloudWatch
metric["resolution"] = resolution # metric resolution must be 1 (High) or 60 (Standard)
# WHEN we try adding a new metric
# THEN it should fail metric unit validation
with pytest.raises(MetricResolutionError, match="Invalid metric resolution.*60"):
with single_metric(namespace=namespace, **metric) as my_metric:
my_metric.add_dimension(**dimension)
def test_schema_validation_incorrect_metric_unit(metric, dimension, namespace):
# GIVEN we pass a metric unit that is not supported by CloudWatch
metric["unit"] = "incorrect_unit"
# WHEN we try adding a new metric
# THEN it should fail metric unit validation
with pytest.raises(MetricUnitError):
with single_metric(**metric) as my_metric:
my_metric.add_dimension(**dimension)
def test_schema_validation_no_namespace(metric, dimension):
# GIVEN we don't add any namespace
# WHEN we attempt to serialize a valid EMF object
# THEN it should fail namespace validation
with pytest.raises(SchemaValidationError, match="Must contain a metric namespace."):
with single_metric(**metric) as my_metric:
my_metric.add_dimension(**dimension)
def test_schema_validation_incorrect_metric_value(metric, dimension, namespace):
# GIVEN we pass an incorrect metric value (non-numeric)
metric["value"] = "some_value"
# WHEN we attempt to serialize a valid EMF object
# THEN it should fail validation and raise SchemaValidationError
with pytest.raises(MetricValueError, match=".*is not a valid number"):
with single_metric(**metric):
pass
def test_schema_no_metrics(service, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
# THEN it should fail validation and raise SchemaValidationError
with pytest.raises(SchemaValidationError, match="Must contain at least one metric."):
my_metrics.serialize_metric_set()
def test_exceed_number_of_dimensions(metric, namespace):
# GIVEN we have more dimensions than CloudWatch supports (N+1)
dimensions = [{"name": f"test_{i}", "value": "test"} for i in range(MAX_DIMENSIONS + 1)]
# WHEN we attempt to serialize them into a valid EMF object
# THEN it should fail validation and raise SchemaValidationError
with pytest.raises(SchemaValidationError, match="Maximum number of dimensions exceeded.*"):
with single_metric(**metric, namespace=namespace) as my_metric:
for dimension in dimensions:
my_metric.add_dimension(**dimension)
def test_exceed_number_of_dimensions_with_service(metric, namespace, monkeypatch):
# GIVEN we have service set and add more dimensions than CloudWatch supports (N-1)
monkeypatch.setenv("POWERTOOLS_SERVICE_NAME", "test_service")
dimensions = [{"name": f"test_{i}", "value": "test"} for i in range(MAX_DIMENSIONS)]
# WHEN we attempt to serialize them into a valid EMF object
# THEN it should fail validation and raise SchemaValidationError
with pytest.raises(SchemaValidationError, match="Maximum number of dimensions exceeded.*"):
with single_metric(**metric, namespace=namespace) as my_metric:
for dimension in dimensions:
my_metric.add_dimension(**dimension)
def test_log_metrics_during_exception(capsys, metric, dimension, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(namespace=namespace)
my_metrics.add_dimension(**dimension)
my_metrics.add_metric(**metric)
# WHEN log_metrics is used to serialize metrics
# but an error has been raised during handler execution
@my_metrics.log_metrics
def lambda_handler(evt, context):
raise ValueError("Bubble up")
with pytest.raises(ValueError):
lambda_handler({}, {})
output = capture_metrics_output(capsys)
expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace)
# THEN we should log metrics either way
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_log_metrics_raise_on_empty_metrics(capsys, metric, dimension, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(service="test_service", namespace=namespace)
# WHEN log_metrics is used with raise_on_empty_metrics param and has no metrics
@my_metrics.log_metrics(raise_on_empty_metrics=True)
def lambda_handler(evt, context):
pass
# THEN the raised exception should be SchemaValidationError
# and specifically about the lack of Metrics
with pytest.raises(SchemaValidationError, match="Must contain at least one metric."):
lambda_handler({}, {})
def test_all_possible_metric_units(metric, dimension, namespace):
# GIVEN we add a metric for each metric unit supported by CloudWatch
# where metric unit as MetricUnit key e.g. "Seconds", "BytesPerSecond"
for unit in MetricUnit:
metric["unit"] = unit.name
# WHEN we iterate over all available metric unit keys from MetricUnit enum
# THEN we raise no MetricUnitError nor SchemaValidationError
with single_metric(namespace=namespace, **metric) as my_metric:
my_metric.add_dimension(**dimension)
# WHEN we iterate over all available metric unit keys from MetricUnit enum
all_metric_units = [unit.value for unit in MetricUnit]
for unit in all_metric_units:
metric["unit"] = unit # e.g. "Seconds", "Bytes/Second"
# THEN we raise no MetricUnitError nor SchemaValidationError
with single_metric(namespace=namespace, **metric) as my_metric:
my_metric.add_dimension(**dimension)
def test_metrics_reuse_metric_set(metric, dimension, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(namespace=namespace)
my_metrics.add_metric(**metric)
# WHEN Metrics is initialized one more time
my_metrics_2 = Metrics(namespace=namespace)
# THEN Both class instances should have the same metric set
assert my_metrics_2.metric_set == my_metrics.metric_set
def test_log_metrics_clear_metrics_after_invocation(metric, service, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
my_metrics.add_metric(**metric)
# WHEN log_metrics is used to flush metrics from memory
@my_metrics.log_metrics
def lambda_handler(evt, context):
pass
lambda_handler({}, {})
# THEN metric set should be empty after function has been run
assert my_metrics.metric_set == {}
def test_log_metrics_non_string_dimension_values(capsys, service, metric, non_str_dimensions, namespace):
# GIVEN Metrics is initialized and dimensions with non-string values are added
my_metrics = Metrics(service=service, namespace=namespace)
my_metrics.add_metric(**metric)
for dimension in non_str_dimensions:
my_metrics.add_dimension(**dimension)
# WHEN we utilize log_metrics to serialize
# and flush all metrics at the end of a function execution
@my_metrics.log_metrics
def lambda_handler(evt, ctx):
pass
lambda_handler({}, {})
output = capture_metrics_output(capsys)
# THEN we should have no exceptions
# and dimension values should be serialized as strings
for dimension in non_str_dimensions:
assert isinstance(output[dimension["name"]], str)
def test_log_metrics_with_explicit_namespace(capsys, metric, service, namespace):
# GIVEN Metrics is initialized with explicit namespace
my_metrics = Metrics(service=service, namespace=namespace)
my_metrics.add_metric(**metric)
# WHEN we utilize log_metrics to serialize
# and flush all metrics at the end of a function execution
@my_metrics.log_metrics
def lambda_handler(evt, ctx):
pass
lambda_handler({}, {})
output = capture_metrics_output(capsys)
# THEN we should have no exceptions and the namespace should be set
# using the service value passed to Metrics constructor
assert namespace == output["_aws"]["CloudWatchMetrics"][0]["Namespace"]
def test_log_metrics_with_implicit_dimensions(capsys, metric, namespace, service):
# GIVEN Metrics is initialized with service specified
my_metrics = Metrics(service=service, namespace=namespace)
my_metrics.add_metric(**metric)
# WHEN we utilize log_metrics to serialize and don't explicitly add any dimensions
@my_metrics.log_metrics
def lambda_handler(evt, ctx):
pass
lambda_handler({}, {})
output = capture_metrics_output(capsys)
# THEN we should have no exceptions and the dimensions should be set to the name provided in the
# service passed to Metrics constructor
assert service == output["service"]
def test_log_metrics_with_renamed_service(capsys, metric, service):
# GIVEN Metrics is initialized with service specified
my_metrics = Metrics(service=service, namespace="test_application")
another_service_dimension = {"name": "service", "value": "another_test_service"}
@my_metrics.log_metrics
def lambda_handler(evt, ctx):
# WHEN we manually call add_dimension to change the value of the service dimension
my_metrics.add_dimension(**another_service_dimension)
my_metrics.add_metric(**metric)
lambda_handler({}, {})
output = capture_metrics_output(capsys)
lambda_handler({}, {})
second_output = capture_metrics_output(capsys)
# THEN we should have no exceptions and the dimensions should be set to the name provided in the
# add_dimension call
assert output["service"] == another_service_dimension["value"]
assert second_output["service"] == another_service_dimension["value"]
def test_namespace_var_precedence(monkeypatch, capsys, metric, dimension, namespace):
# GIVEN we use POWERTOOLS_METRICS_NAMESPACE
monkeypatch.setenv("POWERTOOLS_METRICS_NAMESPACE", "a_namespace")
# WHEN creating a metric and explicitly set a namespace
with single_metric(namespace=namespace, **metric) as my_metrics:
my_metrics.add_dimension(**dimension)
output = capture_metrics_output(capsys)
# THEN namespace should match the explicitly passed variable and not the env var
assert namespace == output["_aws"]["CloudWatchMetrics"][0]["Namespace"]
def test_log_metrics_capture_cold_start_metric(capsys, namespace, service):
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
# WHEN log_metrics is used with capture_cold_start_metric
@my_metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(evt, context):
pass
LambdaContext = namedtuple("LambdaContext", "function_name")
lambda_handler({}, LambdaContext("example_fn"))
output = capture_metrics_output(capsys)
# THEN ColdStart metric and function_name and service dimension should be logged
assert output["ColdStart"] == [1.0]
assert output["function_name"] == "example_fn"
assert output["service"] == service
def test_log_metrics_capture_cold_start_metric_no_service(capsys, namespace):
# GIVEN Metrics is initialized without service
my_metrics = Metrics(namespace=namespace)
# WHEN log_metrics is used with capture_cold_start_metric
@my_metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(evt, context):
pass
LambdaContext = namedtuple("LambdaContext", "function_name")
lambda_handler({}, LambdaContext("example_fn"))
output = capture_metrics_output(capsys)
# THEN ColdStart metric and function_name dimension should be logged
assert output["ColdStart"] == [1.0]
assert output["function_name"] == "example_fn"
assert output.get("service") is None
def test_emit_cold_start_metric_only_once(capsys, namespace, service, metric):
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
# WHEN log_metrics is used with capture_cold_start_metric
# and handler is called more than once
@my_metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(evt, context):
my_metrics.add_metric(**metric)
LambdaContext = namedtuple("LambdaContext", "function_name")
lambda_handler({}, LambdaContext("example_fn"))
_, _ = capture_metrics_output_multiple_emf_objects(capsys) # ignore first stdout captured
# THEN ColdStart metric and function_name dimension should be logged once
lambda_handler({}, LambdaContext("example_fn"))
output = capture_metrics_output(capsys)
assert "ColdStart" not in output
assert "function_name" not in output
def test_log_metrics_decorator_no_metrics_warning(dimensions, namespace, service):
# GIVEN Metrics is initialized
my_metrics = Metrics(namespace=namespace, service=service)
# WHEN using the log_metrics decorator and no metrics have been added
@my_metrics.log_metrics
def lambda_handler(evt, context):
pass
# THEN it should raise a warning instead of throwing an exception
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("default")
lambda_handler({}, {})
assert len(w) == 1
assert str(w[-1].message) == (
"No application metrics to publish. The cold-start metric may be published if enabled. "
"If application metrics should never be empty, consider using 'raise_on_empty_metrics'"
)
def test_log_metrics_with_implicit_dimensions_called_twice(capsys, metric, namespace, service):
# GIVEN Metrics is initialized with service specified
my_metrics = Metrics(service=service, namespace=namespace)
# WHEN we utilize log_metrics to serialize and don't explicitly add any dimensions,
# and the lambda function is called more than once
@my_metrics.log_metrics
def lambda_handler(evt, ctx):
my_metrics.add_metric(**metric)
return True
lambda_handler({}, {})
output = capture_metrics_output(capsys)
lambda_handler({}, {})
second_output = capture_metrics_output(capsys)
# THEN we should have no exceptions and the dimensions should be set to the name provided in the
# service passed to Metrics constructor
assert output["service"] == "test_service"
assert second_output["service"] == "test_service"
for metric_record in output["_aws"]["CloudWatchMetrics"]:
assert ["service"] in metric_record["Dimensions"]
for metric_record in second_output["_aws"]["CloudWatchMetrics"]:
assert ["service"] in metric_record["Dimensions"]
def test_add_metadata_non_string_dimension_keys(service, metric, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
my_metrics.add_metric(**metric)
# WHEN we utilize add_metadata with non-string keys
my_metrics.add_metadata(key=10, value="number_ten")
# THEN we should have no exceptions
# and dimension values should be serialized as strings
expected_metadata = {"10": "number_ten"}
assert my_metrics.metadata_set == expected_metadata
def test_add_metadata(service, metric, namespace, metadata):
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
my_metrics.add_metric(**metric)
# WHEN we utilize add_metadata with non-string keys
my_metrics.add_metadata(**metadata)
# THEN we should have no exceptions
# and dimension values should be serialized as strings
assert my_metrics.metadata_set == {metadata["key"]: metadata["value"]}
def test_log_metrics_with_metadata(capsys, metric, dimension, namespace, service, metadata):
# GIVEN Metrics is initialized
my_metrics = Metrics(namespace=namespace)
my_metrics.add_metric(**metric)
my_metrics.add_dimension(**dimension)
# WHEN we utilize log_metrics to serialize and add metadata
@my_metrics.log_metrics
def lambda_handler(evt, ctx):
my_metrics.add_metadata(**metadata)
pass
lambda_handler({}, {})
output = capture_metrics_output(capsys)
expected = serialize_single_metric(metric=metric, dimension=dimension, namespace=namespace, metadata=metadata)
# THEN we should have no exceptions and metadata
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_serialize_high_resolution_metric_set_metric_definition(
metric_with_resolution,
dimension,
namespace,
service,
metadata,
):
expected_metric_definition = {
"single_metric": [1.0],
"_aws": {
"Timestamp": 1592237875494,
"CloudWatchMetrics": [
{
"Namespace": "test_namespace",
"Dimensions": [["test_dimension", "service"]],
"Metrics": [{"Name": "single_metric", "Unit": "Count", "StorageResolution": 1}],
},
],
},
"service": "test_service",
"username": "test",
"test_dimension": "test",
}
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
my_metrics.add_metric(**metric_with_resolution)
my_metrics.add_dimension(**dimension)
my_metrics.add_metadata(**metadata)
# WHEN metrics are serialized manually
metric_definition_output = my_metrics.serialize_metric_set()
# THEN we should emit a valid embedded metric definition object
assert "Timestamp" in metric_definition_output["_aws"]
remove_timestamp(metrics=[metric_definition_output, expected_metric_definition])
assert metric_definition_output == expected_metric_definition
def test_serialize_metric_set_metric_definition(metric, dimension, namespace, service, metadata):
expected_metric_definition = {
"single_metric": [1.0],
"_aws": {
"Timestamp": 1592237875494,
"CloudWatchMetrics": [
{
"Namespace": "test_namespace",
"Dimensions": [["test_dimension", "service"]],
"Metrics": [{"Name": "single_metric", "Unit": "Count"}],
},
],
},
"service": "test_service",
"username": "test",
"test_dimension": "test",
}
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
my_metrics.add_metric(**metric)
my_metrics.add_dimension(**dimension)
my_metrics.add_metadata(**metadata)
# WHEN metrics are serialized manually
metric_definition_output = my_metrics.serialize_metric_set()
# THEN we should emit a valid embedded metric definition object
assert "Timestamp" in metric_definition_output["_aws"]
remove_timestamp(metrics=[metric_definition_output, expected_metric_definition])
assert metric_definition_output == expected_metric_definition
def test_log_metrics_capture_cold_start_metric_separately(capsys, namespace, service, metric, dimension):
# GIVEN Metrics is initialized
my_metrics = Metrics(service=service, namespace=namespace)
# WHEN log_metrics is used with capture_cold_start_metric
@my_metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(evt, context):
my_metrics.add_metric(**metric)
my_metrics.add_dimension(**dimension)
LambdaContext = namedtuple("LambdaContext", "function_name")
lambda_handler({}, LambdaContext("example_fn"))
cold_start_blob, custom_metrics_blob = capture_metrics_output_multiple_emf_objects(capsys)
# THEN ColdStart metric and function_name dimension should be logged
# in a separate EMF blob than the application metrics
assert cold_start_blob["ColdStart"] == [1.0]
assert cold_start_blob["function_name"] == "example_fn"
assert cold_start_blob["service"] == service
# and that application metrics dimensions are not part of ColdStart EMF blob
assert "test_dimension" not in cold_start_blob
# THEN application metrics EMF blob should not have
# ColdStart metric nor function_name dimension
assert "function_name" not in custom_metrics_blob
assert "ColdStart" not in custom_metrics_blob
# and that application metrics are recorded as normal
assert custom_metrics_blob["service"] == service
assert custom_metrics_blob["single_metric"] == [float(metric["value"])]
assert custom_metrics_blob["test_dimension"] == dimension["value"]
def test_log_multiple_metrics(capsys, metrics_same_name, dimensions, namespace):
# GIVEN Metrics is initialized
my_metrics = Metrics(namespace=namespace)
for dimension in dimensions:
my_metrics.add_dimension(**dimension)
# WHEN we utilize log_metrics to serialize
# and flush multiple metrics with the same name at the end of a function execution
@my_metrics.log_metrics
def lambda_handler(evt, ctx):
for metric in metrics_same_name:
my_metrics.add_metric(**metric)
lambda_handler({}, {})
output = capture_metrics_output(capsys)
expected = serialize_metrics(metrics=metrics_same_name, dimensions=dimensions, namespace=namespace)
# THEN we should have no exceptions
# and a valid EMF object should be flushed correctly
remove_timestamp(metrics=[output, expected])
assert expected == output
def test_serialize_metric_set_metric_definition_multiple_values(
metrics_same_name,
dimension,
namespace,
service,
metadata,
):
expected_metric_definition = {
"metric_one": [1.0, 5.0],
"_aws": {
"Timestamp": 1592237875494,
"CloudWatchMetrics": [
{
"Namespace": "test_namespace",
"Dimensions": [["test_dimension", "service"]],
"Metrics": [{"Name": "metric_one", "Unit": "Count"}],
},
],
},
"service": "test_service",
"username": "test",
"test_dimension": "test",
}
# GIVEN Metrics is initialized and multiple metrics are added with the same name
my_metrics = Metrics(service=service, namespace=namespace)
for metric in metrics_same_name:
my_metrics.add_metric(**metric)
my_metrics.add_dimension(**dimension)
my_metrics.add_metadata(**metadata)
# WHEN metrics are serialized manually