Add feature flag tracking

sentry_sdk/_types.py

@@ -128,6 +128,9 @@ class SDKInfo(TypedDict):
         tuple[None, None, None],
     ]
 
+    Flag = TypedDict("Flag", {"flag": str, "result": bool})
+    Flags = list[Flag]
+
     Hint = Dict[str, Any]
 
     Breadcrumb = Dict[str, Any]

sentry_sdk/api.py

@@ -313,6 +313,12 @@ def set_extra(key, value):
     return get_isolation_scope().set_extra(key, value)
 
 
+@scopemethod
+def set_flag(flag, result):
+    # type: (str, bool) -> None
+    return get_isolation_scope().set_flag(flag, result)
+
+
 @scopemethod
 def set_user(value):
     # type: (Optional[Dict[str, Any]]) -> None

sentry_sdk/client.py

@@ -482,6 +482,9 @@ def _prepare_event(
                     "event_processor", data_category="span", quantity=spans_delta
                 )
 
+        # TODO: Where should I put this?
+        event["contexts"]["flags"] = scope._flag_manager.serialize()
+
         if (
             self.options["attach_stacktrace"]
             and "exception" not in event

sentry_sdk/flag.py

@@ -0,0 +1,55 @@
+from typing import TYPE_CHECKING
+from sentry_sdk._lru_cache import LRUCache, KEY, NEXT, VALUE
+
+
+if TYPE_CHECKING:
+    from sentry_sdk._types import Flag as FType, Flags
+
+
+class Flag:
+
+    __slots__ = ("name", "result")
+
+    def __init__(self, name, result):
+        # type: (str, bool) -> None
+        self.result = result
+        self.name = name
+
+    @property
+    def as_dict(self):
+        # type: () -> FType
+        return {
+            "flag": self.name,
+            "result": self.result,
+        }
+
+
+class FlagManager:
+
+    def __init__(self, capacity):
+        # type: (int) -> None
+        self._cache = LRUCache(max_size=capacity)
+
+    def add(self, name, result):
+        # type: (str, bool) -> None
+        # NOTE: Should we log null names?
+        if name is not None:
+            self._cache.set(name, Flag(name, result))
+
+    def serialize(self):
+        # type: () -> Flags
+        """Serialize flags.
+
+        Flags are serialized in order of first-evaluation.
+        """
+
+        def iter_flags():
+            # This only works if you know the root node is the only
+            # item with a null key. The list is circularly linked so we
+            # need a termination condition.
+            node = self._cache.root[NEXT]
+            while node[KEY] is not None:
+                yield node[VALUE].as_dict
+                node = node[NEXT]
+
+        return [f for f in iter_flags()]

sentry_sdk/scope.py

@@ -10,6 +10,7 @@
 
 from sentry_sdk.attachments import Attachment
 from sentry_sdk.consts import DEFAULT_MAX_BREADCRUMBS, FALSE_VALUES, INSTRUMENTER
+from sentry_sdk.flag import FlagManager
 from sentry_sdk.profiler.continuous_profiler import try_autostart_continuous_profiler
 from sentry_sdk.profiler.transaction_profiler import Profile
 from sentry_sdk.session import Session
@@ -190,6 +191,7 @@ class Scope:
         "client",
         "_type",
         "_last_event_id",
+        "_flag_manager",
     )
 
     def __init__(self, ty=None, client=None):
@@ -198,6 +200,7 @@ def __init__(self, ty=None, client=None):
 
         self._event_processors = []  # type: List[EventProcessor]
         self._error_processors = []  # type: List[ErrorProcessor]
+        self._flag_manager = FlagManager(capacity=50)
 
         self._name = None  # type: Optional[str]
         self._propagation_context = None  # type: Optional[PropagationContext]
@@ -786,6 +789,10 @@ def set_user(self, value):
         if session is not None:
             session.update(user=value)
 
+    def set_flag(self, flag, result):
+        # type: (str, bool) -> None
+        self._flag_manager.add(flag, result)
+
     @property
     def span(self):
         # type: () -> Optional[Span]

tests/test_flag.py

@@ -0,0 +1,30 @@
+from sentry_sdk.flag import Flag, FlagManager
+
+
+def test_serializing_flag():
+    flag = Flag("k", True)
+    assert flag.as_dict == {"flag": "k", "result": True}
+
+
+def test_flag_manager():
+    manager = FlagManager(capacity=2)
+
+    # Partially filled buffer serializes in the correct order.
+    manager.add("a", True)
+    flags = manager.serialize()
+    assert flags == [{"flag": "a", "result": True}]
+
+    # Filled buffer serializes in the correct order.
+    manager.add("b", False)
+    flags = manager.serialize()
+    assert flags == [{"flag": "a", "result": True}, {"flag": "b", "result": False}]
+
+    # Over-filled buffer serializes in the correct order.
+    manager.add("c", True)
+    flags = manager.serialize()
+    assert flags == [{"flag": "b", "result": False}, {"flag": "c", "result": True}]
+
+    # Twice-filled buffer serializes in the correct order.
+    manager.add("d", True)
+    flags = manager.serialize()
+    assert flags == [{"flag": "c", "result": True}, {"flag": "d", "result": True}]