Directly use the unix timestamp instead of time

output/cloud/expv2/collect.go

@@ -9,9 +9,7 @@ import (
 )
 
 type timeBucket struct {
-	// TODO: for performance reasons, use directly a unix time here
-	// so we can avoid time->unix->time
-	Time  time.Time
+	Time  int64
 	Sinks map[metrics.TimeSeries]metricValue
 }
 
@@ -68,12 +66,18 @@ func newCollector(aggrPeriod, waitPeriod time.Duration) (*collector, error) {
 	if aggrPeriod == 0 {
 		return nil, errors.New("aggregation period is not allowed to be zero")
 	}
+	if aggrPeriod != aggrPeriod.Truncate(time.Second) {
+		return nil, errors.New("aggregation period is not allowed to have sub-second precision")
+	}
 	if waitPeriod == 0 {
 		// TODO: we could simplify the expiring logic
 		// just having an internal static logic.
 		// Like skip only not closed buckets bucketEnd > now.
 		return nil, errors.New("aggregation wait period is not allowed to be zero")
 	}
+	if waitPeriod != waitPeriod.Truncate(time.Second) {
+		return nil, errors.New("aggregation wait period is not allowed to have sub-second precision")
+	}
 	return &collector{
 		bq:                bucketQ{},
 		nowFunc:           time.Now,
@@ -152,9 +156,8 @@ func (c *collector) bucketID(t time.Time) int64 {
 	return t.UnixNano() / int64(c.aggregationPeriod)
 }
 
-func (c *collector) timeFromBucketID(id int64) time.Time {
-	return time.Unix(0, id*int64(c.aggregationPeriod)).
-		Truncate(time.Microsecond).UTC()
+func (c *collector) timeFromBucketID(id int64) int64 {
+	return id * int64(c.aggregationPeriod)
 }
 
 func (c *collector) bucketCutoffID() int64 {

output/cloud/expv2/collect_test.go

@@ -9,6 +9,17 @@ import (
 	"go.k6.io/k6/metrics"
 )
 
+func TestNewCollectorError(t *testing.T) {
+	t.Parallel()
+
+	// TODO: more cases
+	_, err := newCollector(4*time.Second+300*time.Millisecond, 1*time.Second)
+	require.ErrorContains(t, err, "sub-second precision")
+
+	_, err = newCollector(4*time.Second, 1*time.Second+300*time.Millisecond)
+	require.ErrorContains(t, err, "sub-second precision")
+}
+
 func TestCollectorCollectSample(t *testing.T) {
 	t.Parallel()
 
@@ -174,7 +185,7 @@ func TestCollectorExpiredBucketsCutoff(t *testing.T) {
 	assert.NotContains(t, c.timeBuckets, 3)
 
 	require.Len(t, expired, 1)
-	expDateTime := time.Unix(9, 0).UTC()
+	expDateTime := time.Unix(9, 0).UTC().UnixNano()
 	assert.Equal(t, expDateTime, expired[0].Time)
 }
 
@@ -205,7 +216,7 @@ func TestCollectorTimeFromBucketID(t *testing.T) {
 	c := collector{aggregationPeriod: 3 * time.Second}
 
 	// exp = TimeFromUnix(bucketID * aggregationPeriod) = Time(49 * 3s)
-	exp := time.Date(1970, time.January, 1, 0, 2, 27, 0, time.UTC)
+	exp := time.Date(1970, time.January, 1, 0, 2, 27, 0, time.UTC).UnixNano()
 	assert.Equal(t, exp, c.timeFromBucketID(49))
 }
 
@@ -228,16 +239,16 @@ func TestBucketQPush(t *testing.T) {
 	t.Parallel()
 
 	bq := bucketQ{}
-	bq.Push([]timeBucket{{Time: time.Unix(1, 0)}})
+	bq.Push([]timeBucket{{Time: int64(1 * time.Second)}})
 	require.Len(t, bq.buckets, 1)
 }
 
 func TestBucketQPopAll(t *testing.T) {
 	t.Parallel()
 	bq := bucketQ{
 		buckets: []timeBucket{
-			{Time: time.Unix(1, 0)},
-			{Time: time.Unix(2, 0)},
+			{Time: int64(1 * time.Second)},
+			{Time: int64(2 * time.Second)},
 		},
 	}
 	buckets := bq.PopAll()
@@ -273,7 +284,7 @@ func TestBucketQPushPopConcurrency(t *testing.T) {
 		}
 	}()
 
-	now := time.Now()
+	now := time.Now().Truncate(time.Second).UnixNano()
 	for {
 		select {
 		case <-stop:

output/cloud/expv2/flush_test.go

@@ -2,7 +2,6 @@ package expv2
 
 import (
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -24,7 +23,7 @@ func TestMetricSetBuilderAddTimeBucket(t *testing.T) {
 	}
 
 	tb := timeBucket{
-		Time: time.Unix(1, 0),
+		Time: 1,
 		Sinks: map[metrics.TimeSeries]metricValue{
 			timeSeries: &counter{},
 		},

output/cloud/expv2/hdr.go

@@ -3,10 +3,8 @@ package expv2
 import (
 	"math"
 	"math/bits"
-	"time"
 
 	"go.k6.io/k6/output/cloud/expv2/pbcloud"
-	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 const (
@@ -183,9 +181,9 @@ func (h *histogram) appendBuckets(index uint32) {
 }
 
 // histogramAsProto converts the histogram into the equivalent Protobuf version.
-func histogramAsProto(h *histogram, time time.Time) *pbcloud.TrendHdrValue {
+func histogramAsProto(h *histogram, time int64) *pbcloud.TrendHdrValue {
 	hval := &pbcloud.TrendHdrValue{
-		Time:              timestamppb.New(time),
+		Time:              timestampAsProto(time),
 		MinResolution:     1.0,
 		SignificantDigits: 2,
 		LowerCounterIndex: h.FirstNotZeroBucket,

output/cloud/expv2/hdr_test.go

@@ -311,6 +311,6 @@ func TestHistogramAsProto(t *testing.T) {
 		tc.exp.MinResolution = 1.0
 		tc.exp.SignificantDigits = 2
 		tc.exp.Time = &timestamppb.Timestamp{Seconds: 1}
-		assert.Equal(t, tc.exp, histogramAsProto(&h, time.Unix(1, 0)), tc.name)
+		assert.Equal(t, tc.exp, histogramAsProto(&h, time.Unix(1, 0).UnixNano()), tc.name)
 	}
 }

output/cloud/expv2/mapping.go

@@ -2,7 +2,6 @@ package expv2
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/mstoykov/atlas"
 	"go.k6.io/k6/metrics"
@@ -51,7 +50,7 @@ func mapMetricTypeProto(mt metrics.MetricType) pbcloud.MetricType {
 func addBucketToTimeSeriesProto(
 	timeSeries *pbcloud.TimeSeries,
 	mt metrics.MetricType,
-	time time.Time,
+	time int64,
 	value metricValue,
 ) {
 	if timeSeries.Samples == nil {
@@ -62,13 +61,13 @@ func addBucketToTimeSeriesProto(
 	case *counter:
 		samples := timeSeries.GetCounterSamples()
 		samples.Values = append(samples.Values, &pbcloud.CounterValue{
-			Time:  timestamppb.New(time),
+			Time:  timestampAsProto(time),
 			Value: typedMetricValue.Sum,
 		})
 	case *gauge:
 		samples := timeSeries.GetGaugeSamples()
 		samples.Values = append(samples.Values, &pbcloud.GaugeValue{
-			Time:  timestamppb.New(time),
+			Time:  timestampAsProto(time),
 			Last:  typedMetricValue.Last,
 			Min:   typedMetricValue.Max,
 			Max:   typedMetricValue.Min,
@@ -78,7 +77,7 @@ func addBucketToTimeSeriesProto(
 	case *rate:
 		samples := timeSeries.GetRateSamples()
 		samples.Values = append(samples.Values, &pbcloud.RateValue{
-			Time:         timestamppb.New(time),
+			Time:         timestampAsProto(time),
 			NonzeroCount: typedMetricValue.NonZeroCount,
 			TotalCount:   typedMetricValue.Total,
 		})
@@ -111,3 +110,16 @@ func initTimeSeriesSamples(timeSeries *pbcloud.TimeSeries, mt metrics.MetricType
 		}
 	}
 }
+
+func timestampAsProto(unixnano int64) *timestamppb.Timestamp {
+	sec := unixnano / 1e9
+	return &timestamppb.Timestamp{
+		Seconds: sec,
+		// sub-second precision for aggregation is not expected
+		// so we don't waste cpu-time computing nanos.
+		//
+		// In the case this assumption is changed then enable them
+		// by int32(unixnano - (sec * 1e9))
+		Nanos: 0,
+	}
+}

output/cloud/expv2/mapping_test.go

@@ -0,0 +1,21 @@
+package expv2
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTimestampAsProto(t *testing.T) {
+	t.Parallel()
+
+	date := time.Unix(10, 0).UTC()
+	timestamp := timestampAsProto(date.UnixNano())
+	assert.Equal(t, date, timestamp.AsTime())
+
+	// sub-second precision is not supported
+	date = time.Unix(10, 282).UTC()
+	timestamp = timestampAsProto(date.UnixNano())
+	assert.Equal(t, time.Unix(10, 0).UTC(), timestamp.AsTime())
+}

output/cloud/expv2/output_test.go

@@ -114,7 +114,7 @@ func TestOutputCollectSamples(t *testing.T) {
 	require.True(t, ok)
 	assert.Equal(t, 5.0, counter.Sum)
 
-	expTime := time.Date(2023, time.May, 1, 1, 1, 15, 0, time.UTC)
+	expTime := time.Date(2023, time.May, 1, 1, 1, 15, 0, time.UTC).UnixNano()
 	assert.Equal(t, expTime, buckets[0].Time)
 }