Skip to content

Commit bf71fbd

Browse files
looooown2006toddbaert
looooown2006
and
toddbaert
authored
fix(flagd): back off before stream reconnect (#391)
* fix(flagd): back off before stream reconnect Signed-off-by: looooown2006 <102905927+looooown2006@users.noreply.github.com> * fixup: clarify comment Signed-off-by: Todd Baert <todd.baert@dynatrace.com> * fixup: update e2e expect Signed-off-by: Todd Baert <todd.baert@dynatrace.com> --------- Signed-off-by: looooown2006 <102905927+looooown2006@users.noreply.github.com> Signed-off-by: Todd Baert <todd.baert@dynatrace.com> Co-authored-by: Todd Baert <todd.baert@dynatrace.com>
1 parent 62e251d commit bf71fbd

6 files changed

Lines changed: 174 additions & 37 deletions

File tree

‎providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/grpc.py‎

Lines changed: 44 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -67,10 +67,12 @@ def __init__(
6767
logger.debug(self.config.fatal_status_codes)
6868

6969
self.retry_grace_period = config.retry_grace_period
70+
self.retry_backoff_max_seconds = config.retry_backoff_max_ms * 0.001
7071
self.streamline_deadline_seconds = config.stream_deadline_ms * 0.001
7172
self.deadline = config.deadline_ms * 0.001
7273
self.connected = False
7374
self._is_fatal = False
75+
self._shutdown_event = threading.Event()
7476
self.channel = self._generate_channel(config)
7577
self.stub = evaluation_pb2_grpc.ServiceStub(self.channel)
7678
self.selector_metadata: tuple[tuple[str, str], ...] | None = (
@@ -140,6 +142,7 @@ def initialize(self, evaluation_context: EvaluationContext) -> None:
140142

141143
def shutdown(self) -> None:
142144
self.active = False
145+
self._shutdown_event.set()
143146
self.channel.unsubscribe(self._state_change_callback)
144147
self.channel.close()
145148
if self.timer and self.timer.is_alive():
@@ -150,6 +153,7 @@ def shutdown(self) -> None:
150153

151154
def connect(self) -> None:
152155
self.active = True
156+
self._shutdown_event.clear()
153157

154158
# Run monitoring in a separate thread
155159
self.monitor_thread = threading.Thread(
@@ -228,6 +232,41 @@ def _stream_call_args(self) -> GrpcMultiCallableArgs:
228232
call_args["metadata"] = self.selector_metadata
229233
return call_args
230234

235+
def _wait_before_reconnect(self) -> None:
236+
self._shutdown_event.wait(self.retry_backoff_max_seconds)
237+
238+
def _handle_rpc_error(self, e: grpc.RpcError) -> bool:
239+
# code() blocks until final status, finalizing the dead RPC before reconnect; keep it, do not inline into logging only
240+
# https://grpc.github.io/grpc/python/grpc.html#grpc.Call.code
241+
code = e.code()
242+
if code.name in self.config.fatal_status_codes:
243+
logger.error(f"EventStream fatal error, {code=} {e.details()=}")
244+
self._is_fatal = True
245+
self.active = False
246+
self.emit_provider_error(
247+
ProviderEventDetails(
248+
message=f"Fatal gRPC status code: {code}",
249+
error_code=ErrorCode.PROVIDER_FATAL,
250+
)
251+
)
252+
return True
253+
# non-fatal errors just reconnect; real loss surfaces as a STALE, and eventually, ERROR event
254+
logger.debug(f"EventStream error, reconnecting, {code=} {e.details()=}")
255+
return False
256+
257+
def _handle_event_stream_message(
258+
self, message: evaluation_pb2.EventStreamResponse
259+
) -> None:
260+
if message.type == "provider_ready":
261+
self.emit_provider_ready(
262+
ProviderEventDetails(message="gRPC sync connection established")
263+
)
264+
self.connected = True
265+
elif message.type == "configuration_change":
266+
msg_dict = MessageToDict(message)
267+
data = msg_dict.get("data", {})
268+
self.handle_changed_flags(data)
269+
231270
def listen(self) -> None:
232271
logger.debug("gRPC starting listener thread")
233272
call_args = self._stream_call_args()
@@ -238,38 +277,20 @@ def listen(self) -> None:
238277
try:
239278
logger.debug("Setting up gRPC sync flags connection")
240279
for message in self.stub.EventStream(request, **call_args):
241-
if message.type == "provider_ready":
242-
self.emit_provider_ready(
243-
ProviderEventDetails(
244-
message="gRPC sync connection established"
245-
)
246-
)
247-
self.connected = True
248-
elif message.type == "configuration_change":
249-
msg_dict = MessageToDict(message)
250-
data = msg_dict.get("data", {})
251-
self.handle_changed_flags(data)
280+
self._handle_event_stream_message(message)
252281

253282
if not self.active:
254283
logger.info("Terminating gRPC sync thread")
255284
return
256-
except grpc.RpcError as e: # noqa: PERF203
257-
# although it seems like this error log is not interesting, without it, the retry is not working as expected
258-
logger.debug(f"SyncFlags stream error, {e.code()=} {e.details()=}")
259-
if e.code().name in self.config.fatal_status_codes:
260-
self._is_fatal = True
261-
self.active = False
262-
self.emit_provider_error(
263-
ProviderEventDetails(
264-
message=f"Fatal gRPC status code: {e.code()}",
265-
error_code=ErrorCode.PROVIDER_FATAL,
266-
)
267-
)
285+
except grpc.RpcError as e:
286+
if self._handle_rpc_error(e):
268287
return
269288
except ParseError:
270289
logger.exception(
271290
f"Could not parse flag data using flagd syntax: {message=}"
272291
)
292+
if self.active:
293+
self._wait_before_reconnect()
273294

274295
def handle_changed_flags(self, data: typing.Any) -> None:
275296
changed_flags = list(data.get("flags", {}).keys())

‎providers/openfeature-provider-flagd/src/openfeature/contrib/provider/flagd/resolvers/process/connector/grpc_watcher.py‎

Lines changed: 18 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -43,8 +43,7 @@ def __init__(
4343

4444
self.channel = self._generate_channel(config)
4545
self.stub = sync_pb2_grpc.FlagSyncServiceStub(self.channel)
46-
self.retry_backoff_seconds = config.retry_backoff_ms * 0.001
47-
self.retry_backoff_max_seconds = config.retry_backoff_ms * 0.001
46+
self.retry_backoff_max_seconds = config.retry_backoff_max_ms * 0.001
4847
self.retry_grace_period = config.retry_grace_period
4948
self.streamline_deadline_seconds = config.stream_deadline_ms * 0.001
5049
self.deadline = config.deadline_ms * 0.001
@@ -56,6 +55,7 @@ def __init__(
5655

5756
self.connected = False
5857
self._is_fatal = False
58+
self._shutdown_event = threading.Event()
5959
self.thread: threading.Thread | None = None
6060
self.timer: threading.Timer | None = None
6161

@@ -129,6 +129,7 @@ def initialize(self, context: EvaluationContext) -> None:
129129

130130
def connect(self) -> None:
131131
self.active = True
132+
self._shutdown_event.clear()
132133

133134
# Run monitoring in a separate thread
134135
self.monitor_thread = threading.Thread(
@@ -199,6 +200,7 @@ def emit_error(self) -> None:
199200

200201
def shutdown(self) -> None:
201202
self.active = False
203+
self._shutdown_event.set()
202204
self.channel.close()
203205

204206
def _create_request_args(self) -> dict:
@@ -270,23 +272,27 @@ def _handle_flag_response(
270272

271273
def _handle_rpc_error(self, e: grpc.RpcError) -> bool:
272274
"""Handle a gRPC RpcError. Returns True if the stream loop should stop."""
273-
if e.code().name in self.config.fatal_status_codes:
274-
logger.error(f"SyncFlags stream fatal error, {e.code()=} {e.details()=}")
275+
# code() blocks until final status, finalizing the dead RPC before reconnect; keep it, do not inline into logging only
276+
# https://grpc.github.io/grpc/python/grpc.html#grpc.Call.code
277+
code = e.code()
278+
if code.name in self.config.fatal_status_codes:
279+
logger.error(f"SyncFlags stream fatal error, {code=} {e.details()=}")
275280
self._is_fatal = True
276281
self.active = False
277282
self.emit_provider_error(
278283
ProviderEventDetails(
279-
message=f"Fatal gRPC status code: {e.code()}",
284+
message=f"Fatal gRPC status code: {code}",
280285
error_code=ErrorCode.PROVIDER_FATAL,
281286
)
282287
)
283288
return True
284-
# non-fatal errors just reconnect; real loss surfaces as a STALE event
285-
logger.debug(
286-
f"SyncFlags stream error, reconnecting, {e.code()=} {e.details()=}"
287-
)
289+
# non-fatal errors just reconnect; real loss surfaces as a STALE, and eventually, ERROR event
290+
logger.debug(f"SyncFlags stream error, reconnecting, {code=} {e.details()=}")
288291
return False
289292

293+
def _wait_before_reconnect(self) -> None:
294+
self._shutdown_event.wait(self.retry_backoff_max_seconds)
295+
290296
def listen(self) -> None:
291297
call_args = self.generate_grpc_call_args()
292298
request_args = self._create_request_args()
@@ -299,7 +305,7 @@ def listen(self) -> None:
299305
for flag_rsp in self.stub.SyncFlags(request, **call_args):
300306
if self._handle_flag_response(flag_rsp, context_values_response):
301307
return
302-
except grpc.RpcError as e: # noqa: PERF203
308+
except grpc.RpcError as e:
303309
if self._handle_rpc_error(e):
304310
return
305311
except json.JSONDecodeError:
@@ -308,6 +314,8 @@ def listen(self) -> None:
308314
)
309315
except ParseError:
310316
logger.exception("Could not parse flag data using flagd syntax")
317+
if self.active:
318+
self._wait_before_reconnect()
311319

312320
def generate_grpc_call_args(self) -> GrpcMultiCallableArgs:
313321
call_args: GrpcMultiCallableArgs = {"wait_for_ready": True}

‎providers/openfeature-provider-flagd/tests/e2e/step/provider_steps.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,8 @@ def get_default_options_for_provider(
5656
"deadline_ms": 1000,
5757
"stream_deadline_ms": 0,
5858
"retry_backoff_ms": 1000,
59-
"retry_grace_period": 3,
59+
"retry_backoff_max_ms": 1000,
60+
"retry_grace_period": 2,
6061
"port": container.get_port(resolver_type),
6162
}
6263
wait: bool = True

‎providers/openfeature-provider-flagd/tests/test_grpc_resolver.py‎

Lines changed: 72 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,11 @@
1+
import unittest
12
from contextlib import suppress
2-
from unittest.mock import MagicMock, Mock
3+
from unittest.mock import MagicMock, Mock, patch
34

4-
from openfeature.contrib.provider.flagd.config import Config
5+
import grpc
6+
from grpc import Channel
7+
8+
from openfeature.contrib.provider.flagd.config import CacheType, Config
59
from openfeature.contrib.provider.flagd.flag_type import FlagType
610
from openfeature.contrib.provider.flagd.resolvers.grpc import (
711
FLAGD_SELECTOR_HEADER,
@@ -10,6 +14,14 @@
1014
from openfeature.schemas.protobuf.flagd.evaluation.v2 import evaluation_pb2
1115

1216

17+
class FakeRpcError(grpc.RpcError):
18+
def code(self):
19+
return grpc.StatusCode.UNAVAILABLE
20+
21+
def details(self):
22+
return "stream unavailable"
23+
24+
1325
def _make_resolver(selector):
1426
config = Config(host="localhost", port=8013, tls=False)
1527
config.selector = selector
@@ -77,3 +89,61 @@ def test_event_stream_omits_metadata_when_no_selector():
7789

7890
kwargs = mock_stub.EventStream.call_args.kwargs
7991
assert "metadata" not in kwargs
92+
93+
94+
class TestGrpcResolver(unittest.TestCase):
95+
def setUp(self):
96+
config = Config(
97+
cache=CacheType.DISABLED,
98+
deadline_ms=100,
99+
retry_backoff_ms=1000,
100+
retry_backoff_max_ms=5000,
101+
stream_deadline_ms=1000,
102+
)
103+
channel = Mock(spec=Channel)
104+
105+
with patch(
106+
"openfeature.contrib.provider.flagd.resolvers.grpc.GrpcResolver._generate_channel",
107+
return_value=channel,
108+
):
109+
self.grpc_resolver = GrpcResolver(
110+
config=config,
111+
emit_provider_ready=Mock(),
112+
emit_provider_error=Mock(),
113+
emit_provider_stale=Mock(),
114+
emit_provider_configuration_changed=Mock(),
115+
)
116+
117+
self.grpc_resolver.stub = MagicMock()
118+
self.grpc_resolver.active = True
119+
120+
def test_uses_max_retry_backoff_for_application_level_reconnect_delay(self):
121+
self.assertEqual(self.grpc_resolver.retry_backoff_max_seconds, 5)
122+
123+
def test_listen_backs_off_after_rpc_stream_error(self):
124+
self.grpc_resolver.stub.EventStream = Mock(side_effect=FakeRpcError())
125+
126+
with patch.object(
127+
self.grpc_resolver,
128+
"_wait_before_reconnect",
129+
side_effect=lambda: setattr(self.grpc_resolver, "active", False),
130+
) as wait_before_reconnect:
131+
self.grpc_resolver.listen()
132+
133+
wait_before_reconnect.assert_called_once()
134+
135+
def test_listen_backs_off_after_stream_completion(self):
136+
self.grpc_resolver.stub.EventStream = Mock(return_value=iter([]))
137+
138+
with patch.object(
139+
self.grpc_resolver,
140+
"_wait_before_reconnect",
141+
side_effect=lambda: setattr(self.grpc_resolver, "active", False),
142+
) as wait_before_reconnect:
143+
self.grpc_resolver.listen()
144+
145+
wait_before_reconnect.assert_called_once()
146+
147+
148+
if __name__ == "__main__":
149+
unittest.main()

‎providers/openfeature-provider-flagd/tests/test_grpc_watcher.py‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import unittest
44
from unittest.mock import MagicMock, Mock, patch
55

6+
import grpc
67
from google.protobuf.json_format import MessageToDict
78
from google.protobuf.struct_pb2 import Struct
89
from grpc import Channel
@@ -20,6 +21,14 @@
2021
from openfeature.schemas.protobuf.flagd.sync.v1.sync_pb2_grpc import FlagSyncServiceStub
2122

2223

24+
class FakeRpcError(grpc.RpcError):
25+
def code(self):
26+
return grpc.StatusCode.UNAVAILABLE
27+
28+
def details(self):
29+
return "stream unavailable"
30+
31+
2332
class TestGrpcWatcher(unittest.TestCase):
2433
def setUp(self):
2534
config = Mock(spec=Config)
@@ -36,6 +45,7 @@ def setUp(self):
3645
config.host = "localhost"
3746
config.port = 5000
3847
config.sync_metadata_disabled = False
48+
config.fatal_status_codes = []
3949

4050
flag_store = Mock(spec=FlagStore)
4151
flag_store.update.return_value = None
@@ -133,6 +143,33 @@ def test_listen_with_sync_metadata_disabled_in_config(self):
133143
)
134144
self.assertEqual(self.context, {})
135145

146+
def test_uses_max_retry_backoff_for_application_level_reconnect_delay(self):
147+
self.assertEqual(self.grpc_watcher.retry_backoff_max_seconds, 5)
148+
149+
def test_listen_backs_off_after_rpc_stream_error(self):
150+
self.mock_stub.SyncFlags = Mock(side_effect=FakeRpcError())
151+
152+
with patch.object(
153+
self.grpc_watcher,
154+
"_wait_before_reconnect",
155+
side_effect=lambda: setattr(self.grpc_watcher, "active", False),
156+
) as wait_before_reconnect:
157+
self.grpc_watcher.listen()
158+
159+
wait_before_reconnect.assert_called_once()
160+
161+
def test_listen_backs_off_after_stream_completion(self):
162+
self.mock_stub.SyncFlags = Mock(return_value=iter([]))
163+
164+
with patch.object(
165+
self.grpc_watcher,
166+
"_wait_before_reconnect",
167+
side_effect=lambda: setattr(self.grpc_watcher, "active", False),
168+
) as wait_before_reconnect:
169+
self.grpc_watcher.listen()
170+
171+
wait_before_reconnect.assert_called_once()
172+
136173
def test_selector_passed_via_both_metadata_and_body(self):
137174
"""Test that selector is passed via both gRPC metadata header and request body for backward compatibility"""
138175
self.grpc_watcher.selector = "test-selector"

‎uv.lock‎

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)