Add basic addition support to cloudwatch.get_metric_data expressions

moto/cloudwatch/metric_data_expression_parser.py

@@ -1,13 +1,54 @@
 from typing import Any, Dict, List, SupportsFloat, Tuple
 
+from .exceptions import ValidationError
+
 
 def parse_expression(
     expression: str, results: List[Dict[str, Any]]
 ) -> Tuple[List[SupportsFloat], List[str]]:
     values: List[SupportsFloat] = []
     timestamps: List[str] = []
+
+    if "+" in expression:
+        expression = expression.replace(" ", "")
+        plus_splits = expression.split("+")
+        if len(plus_splits) != 2:
+            raise ValidationError(
+                "Metric math expressions only support adding two values together"
+            )
+
+        by_timestamp = results_by_timestamp(results)
+        for timestamp, vals in by_timestamp.items():
+            first = vals.get(plus_splits[0], 0.0)
+            second = vals.get(plus_splits[1], 0.0)
+
+            values.append(first + second)
+            timestamps.append(timestamp)
+
     for result in results:
         if result.get("id") == expression:
             values.extend(result["vals"])
             timestamps.extend(result["timestamps"])
     return values, timestamps
+
+
+def results_by_timestamp(
+    results: List[Dict[str, Any]],
+) -> Dict[str, Dict[str, float]]:
+    out: Dict[str, Dict[str, float]] = {}
+
+    for result in results:
+        this_id = result.get("id")
+        if not this_id:
+            continue
+
+        for i in range(0, len(result["vals"])):
+            timestamp = result["timestamps"][i]
+            value = result["vals"][i]
+
+            if timestamp not in out:
+                out[timestamp] = {}
+
+            out[timestamp][this_id] = value
+
+    return out

tests/test_cloudwatch/test_cloudwatch_expression_parser.py

@@ -1,3 +1,5 @@
+import datetime
+
 from moto.cloudwatch.metric_data_expression_parser import parse_expression
 
 
@@ -38,3 +40,26 @@ def test_complex_expression():
     ]
     res = parse_expression("totalBytes/10", result_from_previous_queries)
     assert res == ([], [])
+
+
+def test_addition_expression():
+    t3 = datetime.datetime.now(datetime.timezone.utc)
+    t2 = t3 - datetime.timedelta(minutes=1)
+    t1 = t2 - datetime.timedelta(minutes=1)
+
+    results_from_previous_queries = [
+        {
+            "id": "first",
+            "label": "first",
+            "vals": [10.0, 15.0, 30.0],
+            "timestamps": [t1, t2, t3],
+        },
+        {
+            "id": "second",
+            "label": "second",
+            "vals": [25.0, 5.0, 3.0],
+            "timestamps": [t1, t2, t3],
+        },
+    ]
+    res = parse_expression("first + second", results_from_previous_queries)
+    assert res == ([35.0, 20.0, 33.0], [t1, t2, t3])

tests/test_cloudwatch/test_cloudwatch_expressions.py

@@ -75,6 +75,94 @@ def test_get_metric_data_with_simple_expression():
     assert results[0]["Values"] == [25.0]
 
 
+@mock_aws
+def test_get_metric_data_with_basic_addition():
+    t3 = datetime.now(tz=timezone.utc).replace(second=0, microsecond=0)
+    t2 = t3 - timedelta(minutes=1)
+    t1 = t2 - timedelta(minutes=1)
+
+    utc_now = datetime.now(tz=timezone.utc)
+    cloudwatch = boto3.client("cloudwatch", "eu-west-1")
+    namespace = "my_namespace/"
+    cloudwatch.put_metric_data(
+        Namespace=namespace,
+        MetricData=[
+            {
+                "MetricName": "metric1",
+                "Value": 25,
+                "Unit": "Bytes",
+                "Timestamp": t1,
+            },
+            {
+                "MetricName": "metric1",
+                "Value": 30,
+                "Unit": "Bytes",
+                "Timestamp": t2,
+            },
+            {
+                "MetricName": "metric1",
+                "Value": 10,
+                "Unit": "Bytes",
+                "Timestamp": t3,
+            },
+            {
+                "MetricName": "metric2",
+                "Value": 9,
+                "Unit": "Bytes",
+                "Timestamp": t1,
+            },
+            {
+                "MetricName": "metric2",
+                "Value": 5,
+                "Unit": "Bytes",
+                "Timestamp": t2,
+            },
+            {
+                "MetricName": "metric2",
+                "Value": 10,
+                "Unit": "Bytes",
+                "Timestamp": t3,
+            },
+        ],
+    )
+    # get_metric_data 1
+    results = cloudwatch.get_metric_data(
+        MetricDataQueries=[
+            {
+                "Id": "result1",
+                "Expression": "first + second",
+                "Label": "e1",
+            },
+            {
+                "Id": "first",
+                "MetricStat": {
+                    "Metric": {"Namespace": namespace, "MetricName": "metric1"},
+                    "Period": 60,
+                    "Stat": "Sum",
+                },
+                "ReturnData": False,
+            },
+            {
+                "Id": "second",
+                "MetricStat": {
+                    "Metric": {"Namespace": namespace, "MetricName": "metric2"},
+                    "Period": 60,
+                    "Stat": "Sum",
+                },
+                "ReturnData": False,
+            },
+        ],
+        StartTime=utc_now - timedelta(minutes=5),
+        EndTime=utc_now + timedelta(minutes=5),
+        ScanBy="TimestampAscending",
+    )["MetricDataResults"]
+    #
+    assert len(results) == 1
+    assert results[0]["Id"] == "result1"
+    assert results[0]["Label"] == "e1"
+    assert results[0]["Values"] == [34, 35, 20]
+
+
 @cloudwatch_aws_verified
 @pytest.mark.aws_verified
 def test_get_metric_data_with_expressive_expression():