-
Notifications
You must be signed in to change notification settings - Fork 17
feat: surface experiment metadata on flags and add event tracking #94
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Draft
Zaimwa9
wants to merge
4
commits into
main
Choose a base branch
from
feat/experimentation
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Draft
Changes from all commits
Commits
Show all changes
4 commits
Select commit
Hold shift + click to select a range
896260c
feat: surface experiment metadata on flags and add event tracking
Zaimwa9 a6c4caf
fix: include experiment id in exposure dedupe key
Zaimwa9 a1cdd64
fix: fetch flags for an explicitly supplied identity in getExperiment…
Zaimwa9 502a0be
fix: make flushEvents await uploads already in flight
Zaimwa9 File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,188 @@ | ||
| import 'dart:async'; | ||
| import 'dart:convert'; | ||
|
|
||
| import 'package:dio/dio.dart'; | ||
|
|
||
| import '../../version.dart'; | ||
|
|
||
| /// Batches experimentation events to `{eventsURI}v1/events`. | ||
| /// | ||
| /// Flushes on a timer, at [maxBuffer], on [flush] and best-effort on [stop]. | ||
| /// Exposures dedupe within a flush window; a failed POST retries once then drops. | ||
| class EventProcessor { | ||
| static const String flagExposureEvent = r'$flag_exposure'; | ||
| static const String eventsPath = 'v1/events'; | ||
| static const String sdkUserAgentHeader = 'Flagsmith-SDK-User-Agent'; | ||
| static const String environmentKeyHeader = 'X-Environment-Key'; | ||
| static const String contentType = 'application/json; charset=utf-8'; | ||
|
|
||
| final Dio _api; | ||
| final String _apiKey; | ||
| final String endpoint; | ||
| final int flushInterval; | ||
| final int maxBuffer; | ||
| final int retryBackoff; | ||
| final void Function(String message) _log; | ||
|
|
||
| final List<Map<String, dynamic>> _buffer = []; | ||
| final Set<String> _dedupeKeys = {}; | ||
| final Set<Future<void>> _inFlight = {}; | ||
| Timer? _timer; | ||
|
|
||
| EventProcessor({ | ||
| required Dio api, | ||
| required String apiKey, | ||
| required String eventsURI, | ||
| this.flushInterval = 10000, | ||
| this.maxBuffer = 1000, | ||
| this.retryBackoff = 1000, | ||
| void Function(String message)? log, | ||
| }) : _api = api, | ||
| _apiKey = apiKey, | ||
| _log = log ?? _noopLog, | ||
| endpoint = | ||
| '${eventsURI.endsWith('/') ? eventsURI : '$eventsURI/'}$eventsPath'; | ||
|
|
||
| static void _noopLog(String _) {} | ||
|
|
||
| List<Map<String, dynamic>> get buffer => List.unmodifiable(_buffer); | ||
|
|
||
| void trackEvent({ | ||
| required String event, | ||
| String? identifier, | ||
| Object? value, | ||
| Map<String, dynamic>? traits, | ||
| Map<String, dynamic>? metadata, | ||
| }) { | ||
| _bufferEvent( | ||
| event: event, | ||
| featureName: null, | ||
| identifier: identifier, | ||
| value: value, | ||
| traits: traits, | ||
| metadata: metadata, | ||
| dedupe: false, | ||
| ); | ||
| } | ||
|
|
||
| void trackExposureEvent({ | ||
| required String featureName, | ||
| required String identifier, | ||
| Object? value, | ||
| Map<String, dynamic>? traits, | ||
| Map<String, dynamic>? metadata, | ||
| }) { | ||
| _bufferEvent( | ||
| event: flagExposureEvent, | ||
| featureName: featureName, | ||
| identifier: identifier, | ||
| value: value, | ||
| traits: traits, | ||
| metadata: metadata, | ||
| dedupe: true, | ||
| ); | ||
| } | ||
|
|
||
| void _bufferEvent({ | ||
| required String event, | ||
| required String? featureName, | ||
| required String? identifier, | ||
| required Object? value, | ||
| required Map<String, dynamic>? traits, | ||
| required Map<String, dynamic>? metadata, | ||
| required bool dedupe, | ||
| }) { | ||
| final stringValue = value == null ? null : '$value'; | ||
| if (dedupe) { | ||
| // Experiment id is part of the key so a new experiment on the same flag | ||
| // and variant within one flush window still records its own exposure. | ||
| final key = jsonEncode([ | ||
| event, | ||
| featureName, | ||
| identifier, | ||
| stringValue, | ||
| metadata?['experiment_id'] | ||
| ]); | ||
| if (_dedupeKeys.contains(key)) { | ||
| return; | ||
| } | ||
| _dedupeKeys.add(key); | ||
| } | ||
| _buffer.add(<String, dynamic>{ | ||
| 'event': event, | ||
| 'feature_name': featureName, | ||
| 'identifier': identifier, | ||
| 'value': stringValue, | ||
| 'traits': traits, | ||
| 'metadata': <String, dynamic>{ | ||
| ...?metadata, | ||
| 'sdk_version': sdkVersion, | ||
| }, | ||
| 'timestamp': DateTime.now().millisecondsSinceEpoch, | ||
| }); | ||
| if (_buffer.length >= maxBuffer) { | ||
| unawaited(flush()); | ||
| } | ||
| } | ||
|
|
||
| /// Posts the buffered events and waits for every upload still in flight, | ||
| /// including ones started by the timer or the max-buffer trigger, so that | ||
| /// awaiting it at teardown means the POSTs have completed. Never throws. | ||
| Future<void> flush() async { | ||
| if (_buffer.isNotEmpty) { | ||
| final events = List<Map<String, dynamic>>.from(_buffer); | ||
| _buffer.clear(); | ||
| _dedupeKeys.clear(); | ||
| late final Future<void> upload; | ||
| upload = | ||
| _postBatch(events, 0).whenComplete(() => _inFlight.remove(upload)); | ||
| _inFlight.add(upload); | ||
| } | ||
| await Future.wait(_inFlight.toList()); | ||
| } | ||
|
|
||
| void start() { | ||
| _timer?.cancel(); | ||
| _timer = null; | ||
| if (flushInterval > 0) { | ||
| _timer = Timer.periodic( | ||
| Duration(milliseconds: flushInterval), (_) => unawaited(flush())); | ||
| } | ||
| } | ||
|
|
||
| /// Cancels the timer and flushes without awaiting; await [flush] for teardown. | ||
| void stop() { | ||
| _timer?.cancel(); | ||
| _timer = null; | ||
| unawaited(flush()); | ||
| } | ||
|
|
||
| Future<void> _postBatch( | ||
| List<Map<String, dynamic>> events, int attempt) async { | ||
| try { | ||
| final response = await _api.post<dynamic>( | ||
| endpoint, | ||
| data: <String, dynamic>{'events': events}, | ||
| options: Options( | ||
| contentType: contentType, | ||
| headers: <String, dynamic>{ | ||
| environmentKeyHeader: _apiKey, | ||
| sdkUserAgentHeader: getUserAgent(), | ||
| }, | ||
| ), | ||
| ); | ||
| final status = response.statusCode ?? 0; | ||
| if (status < 200 || status >= 300) { | ||
| throw StateError('unexpected status $status'); | ||
| } | ||
| _log('Events: flush successful (${events.length} events)'); | ||
| } catch (e) { | ||
| if (attempt < 1) { | ||
| _log('Events: flush failed, retrying: $e'); | ||
| await Future<void>.delayed(Duration(milliseconds: retryBackoff)); | ||
| return _postBatch(events, attempt + 1); | ||
| } | ||
| _log('Events: flush failed, dropping ${events.length} events: $e'); | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| import 'package:json_annotation/json_annotation.dart'; | ||
|
|
||
| part 'experiment.g.dart'; | ||
|
|
||
| /// The running experiment a flag was evaluated under; identity evaluations only. | ||
| @JsonSerializable() | ||
| class Experiment { | ||
| final int id; | ||
| final String name; | ||
|
|
||
| /// Whether the identity is enrolled. `variant` alone cannot tell. | ||
| @JsonKey(name: 'in_experiment', defaultValue: false) | ||
| final bool inExperiment; | ||
|
|
||
| const Experiment({ | ||
| required this.id, | ||
| required this.name, | ||
| this.inExperiment = false, | ||
| }); | ||
|
|
||
| factory Experiment.fromJson(Map<String, dynamic> json) => | ||
| _$ExperimentFromJson(json); | ||
|
|
||
| Map<String, dynamic> toJson() => _$ExperimentToJson(this); | ||
|
|
||
| @override | ||
| String toString() => 'Experiment($id:$name, inExperiment=$inExperiment)'; | ||
| } | ||
|
|
||
| /// `metadata.experiment` -> [Experiment]; null when absent or malformed. | ||
| Experiment? experimentFromMetadata(Object? metadata) { | ||
| if (metadata is! Map) { | ||
| return null; | ||
| } | ||
| final experiment = metadata['experiment']; | ||
| if (experiment is! Map) { | ||
| return null; | ||
| } | ||
| try { | ||
| return Experiment.fromJson(Map<String, dynamic>.from(experiment)); | ||
| } catch (_) { | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
| Map<String, dynamic>? experimentToMetadata(Experiment? experiment) { | ||
| if (experiment == null) { | ||
| return null; | ||
| } | ||
| return <String, dynamic>{'experiment': experiment.toJson()}; | ||
| } |
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟠 Major · ⚡ Quick win
flushEvents()can complete before an automatic upload finishes.Observed: A max-buffer or timer flush clears
_bufferbefore awaiting_postBatch; a followingflushEvents()therefore sees an empty buffer and returns while that POST is still in flight.Predicted: Code that awaits
flushEvents()during teardown after an automatic flush could terminate before the event request completes, despite the documented completion guarantee. Track in-flight flush futures and makeflush()await them as well as any newly buffered batch; add a delayed-request regression test for both max-buffer and timer flushes.