Align timeseries buckets

snuba/web/rpc/v1/endpoint_time_series.py

@@ -55,6 +55,11 @@
     ]
 )
 
+
+def _rewind(start_timestamp: int, granularity: int) -> int:
+    return (start_timestamp // granularity) * granularity
+
+
 _MAX_BUCKETS_IN_REQUEST = 1000
 
 
@@ -121,11 +126,14 @@ def _convert_result_timeseries(
         tuple[str, str], dict[int, Dict[str, Any]]
     ] = defaultdict(dict)
 
+    start_timestamp_seconds = _rewind(
+        request.meta.start_timestamp.seconds, request.granularity_secs
+    )
     query_duration = (
         request.meta.end_timestamp.seconds - request.meta.start_timestamp.seconds
     )
     time_buckets = [
-        Timestamp(seconds=(request.meta.start_timestamp.seconds) + secs)
+        Timestamp(seconds=(start_timestamp_seconds) + secs)
         for secs in range(0, query_duration, request.granularity_secs)
     ]
 

tests/web/rpc/v1/test_endpoint_time_series.py

@@ -226,6 +226,52 @@ def test_sum(self) -> None:
             ),
         ]
 
+    def test_start_time_not_divisible_by_time_buckets_returns_valid_data(self) -> None:
+        # store a a test metric with a value of 1, every second of one hour
+        granularity_secs = 300
+        query_duration = 60 * 30
+        store_timeseries(
+            BASE_TIME,
+            1,
+            3600,
+            metrics=[DummyMetric("test_metric", get_value=lambda x: 1)],
+        )
+
+        message = TimeSeriesRequest(
+            meta=RequestMeta(
+                project_ids=[1, 2, 3],
+                organization_id=1,
+                cogs_category="something",
+                referrer="something",
+                start_timestamp=Timestamp(seconds=int(BASE_TIME.timestamp() + 1)),
+                end_timestamp=Timestamp(
+                    seconds=int(BASE_TIME.timestamp() + query_duration)
+                ),
+            ),
+            aggregations=[
+                AttributeAggregation(
+                    aggregate=Function.FUNCTION_SUM,
+                    key=AttributeKey(type=AttributeKey.TYPE_FLOAT, name="test_metric"),
+                    label="sum",
+                    extrapolation_mode=ExtrapolationMode.EXTRAPOLATION_MODE_NONE,
+                ),
+                AttributeAggregation(
+                    aggregate=Function.FUNCTION_AVG,
+                    key=AttributeKey(type=AttributeKey.TYPE_FLOAT, name="test_metric"),
+                    label="avg",
+                    extrapolation_mode=ExtrapolationMode.EXTRAPOLATION_MODE_NONE,
+                ),
+            ],
+            granularity_secs=granularity_secs,
+        )
+        response = EndpointTimeSeries().execute(message)
+
+        print(response)
+        for ts in response.result_timeseries:
+            # expect ts.data_points to look like this: [, , , , , ]
+            for datapoint in ts.data_points:
+                assert datapoint != DataPoint()
+
     def test_with_group_by(self) -> None:
         store_timeseries(
             BASE_TIME,