feat(meter): ignore late events

Author: hekike

openmeter/billing/httpdriver/invoicecost.go

@@ -130,6 +130,8 @@ func (h *handler) GetInvoiceLineCost() GetInvoiceLineCostHandler {
 				To:             &line.Period.End,
 				FilterGroupBy:  meterGroupByFilters,
 				FilterCustomer: []streaming.Customer{*customer},
+				// We ignore late events because the data is ingested after the invoice is collected
+				IgnoreLateEvents: invoice.CollectionAt,
 			}
 
 			if request.Params.GroupBy != nil {

openmeter/streaming/clickhouse/connector.go

@@ -142,18 +142,19 @@ func (c *Connector) QueryMeter(ctx context.Context, namespace string, meter mete
 	sort.Strings(groupBy)
 
 	query := queryMeter{
-		Database:        c.config.Database,
-		EventsTableName: c.config.EventsTableName,
-		Namespace:       namespace,
-		Meter:           meter,
-		From:            params.From,
-		To:              params.To,
-		FilterCustomer:  params.FilterCustomer,
-		FilterSubject:   params.FilterSubject,
-		FilterGroupBy:   params.FilterGroupBy,
-		GroupBy:         groupBy,
-		WindowSize:      params.WindowSize,
-		WindowTimeZone:  params.WindowTimeZone,
+		Database:         c.config.Database,
+		EventsTableName:  c.config.EventsTableName,
+		Namespace:        namespace,
+		Meter:            meter,
+		From:             params.From,
+		To:               params.To,
+		IgnoreLateEvents: params.IgnoreLateEvents,
+		FilterCustomer:   params.FilterCustomer,
+		FilterSubject:    params.FilterSubject,
+		FilterGroupBy:    params.FilterGroupBy,
+		GroupBy:          groupBy,
+		WindowSize:       params.WindowSize,
+		WindowTimeZone:   params.WindowTimeZone,
 	}
 
 	// Load cached rows if any

openmeter/streaming/clickhouse/meter_query.go

@@ -18,18 +18,19 @@ import (
 )
 
 type queryMeter struct {
-	Database        string
-	EventsTableName string
-	Namespace       string
-	Meter           meterpkg.Meter
-	FilterCustomer  []streaming.Customer
-	FilterSubject   []string
-	FilterGroupBy   map[string][]string
-	From            *time.Time
-	To              *time.Time
-	GroupBy         []string
-	WindowSize      *meterpkg.WindowSize
-	WindowTimeZone  *time.Location
+	Database         string
+	EventsTableName  string
+	Namespace        string
+	Meter            meterpkg.Meter
+	FilterCustomer   []streaming.Customer
+	FilterSubject    []string
+	FilterGroupBy    map[string][]string
+	From             *time.Time
+	To               *time.Time
+	IgnoreLateEvents *time.Time
+	GroupBy          []string
+	WindowSize       *meterpkg.WindowSize
+	WindowTimeZone   *time.Location
 }
 
 // from returns the from time for the query.
@@ -286,6 +287,13 @@ func (d *queryMeter) timeWhere(query *sqlbuilder.SelectBuilder) *sqlbuilder.Sele
 		query = query.Where(query.LessThan(timeColumn, d.To.Unix()))
 	}
 
+	// If ignore late events is set, we filter out data that was ingested after the query period
+	// Note: This filter runs on a non sorthed column
+	if d.IgnoreLateEvents != nil {
+		ingestedAtColumn := getColumn("ingested_at")
+		query = query.Where(query.LessThan(ingestedAtColumn, d.IgnoreLateEvents.Unix()))
+	}
+
 	return query
 }
 

openmeter/streaming/query_params.go

@@ -19,6 +19,9 @@ type QueryParams struct {
 	GroupBy        []string
 	WindowSize     *meter.WindowSize
 	WindowTimeZone *time.Location
+	// IgnoreLateEvents is used to filter out data that was ingested after the query period.
+	// This is useful for queries that are not real-time, such as billing queries.
+	IgnoreLateEvents *time.Time
 }
 
 // Validate validates query params focusing on `from` and `to` being aligned with query and meter window sizes
@@ -41,6 +44,11 @@ func (p *QueryParams) Validate() error {
 		}
 	}
 
+	// Check that ignore late events is after to
+	if p.IgnoreLateEvents != nil && p.To != nil && p.To.After(*p.IgnoreLateEvents) {
+		errs = append(errs, errors.New("ignore late events must be after to"))
+	}
+
 	// This is required because otherwise the response would be ambiguous
 	if len(p.FilterSubject) > 1 && !slices.Contains(p.GroupBy, "subject") {
 		errs = append(errs, errors.New("multiple subject filters are only allowed with subject group by"))