fix(meterexport): allow groubpys (#3689)

Author: GAlexIHU

openmeter/meterexport/service.go

@@ -28,7 +28,7 @@ type Service interface {
 	//
 	// NOTE: Currently only SUM and COUNT meters are supported.
 	// NOTE: GroupBy values are not yet supported.
-	// NOTE: Subjects and Customers are not honored in the exported data.
+	// NOTE: Customers are not honored in the exported data.
 	ExportSyntheticMeterData(ctx context.Context, config DataExportConfig, result chan<- streaming.RawEvent, err chan<- error) error
 
 	// ExportSyntheticMeterDataIter is an iterator-based wrapper around ExportSyntheticMeterData.

openmeter/meterexport/service/funnel.go

@@ -49,18 +49,10 @@ func (p funnelParams) validateUnsupportedParams() []error {
 		errs = append(errs, errors.New("filter customer is not supported"))
 	}
 
-	if len(p.queryParams.FilterSubject) > 0 {
-		errs = append(errs, errors.New("filter subject is not supported"))
-	}
-
 	if len(p.queryParams.FilterGroupBy) > 0 {
 		errs = append(errs, errors.New("filter group by is not supported"))
 	}
 
-	if len(p.queryParams.GroupBy) > 0 {
-		errs = append(errs, errors.New("group by is not supported"))
-	}
-
 	return errs
 }
 
@@ -101,6 +93,7 @@ func (s *service) funnel(ctx context.Context, params funnelParams, resultCh chan
 			To:             &queryTo,
 			WindowSize:     params.queryParams.WindowSize,
 			WindowTimeZone: params.queryParams.WindowTimeZone,
+			GroupBy:        params.queryParams.GroupBy,
 		}
 
 		rows, err := s.StreamingConnector.QueryMeter(ctx, params.meter.Namespace, params.meter, queryParams)

openmeter/meterexport/service/service.go

@@ -11,7 +11,6 @@ import (
 type Config struct {
 	// Configuration
 	EventSourceGroup string
-	ExportSubject    string
 
 	// Dependencies
 	StreamingConnector streaming.Connector
@@ -21,10 +20,6 @@ type Config struct {
 func (c Config) validate() error {
 	var errs []error
 
-	if c.ExportSubject == "" {
-		errs = append(errs, errors.New("export subject is required"))
-	}
-
 	if c.EventSourceGroup == "" {
 		errs = append(errs, errors.New("event source group is required"))
 	}

openmeter/meterexport/service/service_test.go

@@ -209,7 +209,6 @@ func TestExportSyntheticMeterData(t *testing.T) {
 
 			// Create service
 			svc, err := New(Config{
-				ExportSubject:      "test-subject",
 				EventSourceGroup:   "test-source",
 				StreamingConnector: mockStreaming,
 				MeterService:       mockMeterService,
@@ -308,7 +307,6 @@ func TestExportSyntheticMeterData_ContextCancellation(t *testing.T) {
 		}
 
 		svc, err := New(Config{
-			ExportSubject:      "test-subject",
 			EventSourceGroup:   "test-source",
 			StreamingConnector: mockStreaming,
 			MeterService:       mockMeterService,
@@ -384,7 +382,6 @@ func TestExportSyntheticMeterData_ContextCancellation(t *testing.T) {
 		mockStreaming.AddSimpleEvent("test-meter", 10.0, now.Add(-5*time.Minute))
 
 		svc, err := New(Config{
-			ExportSubject:      "test-subject",
 			EventSourceGroup:   "test-source",
 			StreamingConnector: mockStreaming,
 			MeterService:       mockMeterService,
@@ -453,15 +450,6 @@ func TestServiceNew(t *testing.T) {
 		assert.Contains(t, err.Error(), "meter service is required")
 	})
 
-	t.Run("should fail without export subject", func(t *testing.T) {
-		_, err := New(Config{
-			StreamingConnector: testutils.NewMockStreamingConnector(t),
-			MeterService:       NewMockMeterService(),
-		})
-		require.Error(t, err)
-		assert.Contains(t, err.Error(), "export subject is required")
-	})
-
 	t.Run("should fail without event source group", func(t *testing.T) {
 		_, err := New(Config{
 			StreamingConnector: testutils.NewMockStreamingConnector(t),
@@ -475,7 +463,6 @@ func TestServiceNew(t *testing.T) {
 		svc, err := New(Config{
 			StreamingConnector: testutils.NewMockStreamingConnector(t),
 			MeterService:       NewMockMeterService(),
-			ExportSubject:      "test-subject",
 			EventSourceGroup:   "test-source",
 		})
 		require.NoError(t, err)
@@ -512,7 +499,6 @@ func TestExportSyntheticMeterDataIter(t *testing.T) {
 		mockStreaming.AddSimpleEvent("test-meter", 30.0, now.Add(-2*time.Minute))
 
 		svc, err := New(Config{
-			ExportSubject:      "test-subject",
 			EventSourceGroup:   "test-source",
 			StreamingConnector: mockStreaming,
 			MeterService:       mockMeterService,
@@ -576,7 +562,6 @@ func TestExportSyntheticMeterDataIter(t *testing.T) {
 		}
 
 		svc, err := New(Config{
-			ExportSubject:      "test-subject",
 			EventSourceGroup:   "test-source",
 			StreamingConnector: mockStreaming,
 			MeterService:       mockMeterService,
@@ -621,7 +606,6 @@ func TestExportSyntheticMeterDataIter(t *testing.T) {
 		mockStreaming := testutils.NewMockStreamingConnector(t)
 
 		svc, err := New(Config{
-			ExportSubject:      "test-subject",
 			EventSourceGroup:   "test-source",
 			StreamingConnector: mockStreaming,
 			MeterService:       mockMeterService,
@@ -668,7 +652,6 @@ func TestExportSyntheticMeterDataIter(t *testing.T) {
 		mockStreaming := testutils.NewMockStreamingConnector(t)
 
 		svc, err := New(Config{
-			ExportSubject:      "test-subject",
 			EventSourceGroup:   "test-source",
 			StreamingConnector: mockStreaming,
 			MeterService:       mockMeterService,

openmeter/meterexport/service/syntheticdata.go

@@ -126,6 +126,7 @@ func (s *service) ExportSyntheticMeterData(ctx context.Context, config meterexpo
 				To:             config.Period.To,
 				WindowSize:     &config.ExportWindowSize,
 				WindowTimeZone: time.UTC,
+				GroupBy:        []string{"subject"},
 			},
 		}, meterRowCh, meterRowErrCh)
 	})
@@ -146,7 +147,7 @@ func (s *service) createEventFromMeterRow(m meter.Meter, row meter.MeterQueryRow
 		ID:         ulid.Make().String(),
 		Type:       m.EventType, // We reuse the same type as the source meter
 		Source:     fmt.Sprintf("%s:%s/%s", s.EventSourceGroup, m.Namespace, m.ID),
-		Subject:    s.ExportSubject,
+		Subject:    lo.FromPtr(row.Subject),
 		IngestedAt: clock.Now(),
 		Time:       row.WindowStart,
 		CustomerID: nil,