Skip to content

Commit

Permalink
Fix for outdated sensors after rebalancing (#72)
Browse files Browse the repository at this point in the history
* Rename test classes to start with `Test`

Previously these classes started with `test_`, such classes are not
collected by pytest so these tests were not run.

* Failing test for clearing sensors on rebalance

* Clear sensors related to topics after rebalance

* Refactor prometheus metrics

* Failing test for prometheus metrics with topic labels after rebalance

* Clear prometheus topic related metrics on rebalance

* Failing test for IndexError in `on_stream_event_out`

* Fix IndexError in `on_stream_event_out`
  • Loading branch information
krzysieksulejczak committed Jan 14, 2021
1 parent fcb6b18 commit 0a4d059
Show file tree
Hide file tree
Showing 7 changed files with 774 additions and 347 deletions.
9 changes: 9 additions & 0 deletions faust/sensors/monitor.py
Expand Up @@ -571,6 +571,7 @@ def on_assignment_completed(
def on_rebalance_start(self, app: AppT) -> Dict:
"""Cluster rebalance in progress."""
self.rebalances = app.rebalancing_count
self._clear_topic_related_sensors()
return {"time_start": self.time()}

def on_rebalance_return(self, app: AppT, state: Dict) -> None:
Expand All @@ -596,6 +597,7 @@ def on_rebalance_end(self, app: AppT, state: Dict) -> None:
latency_end=latency_end,
)
deque_pushpopmax(self.rebalance_end_latency, latency_end, self.max_avg_history)
self._clear_topic_related_sensors()

def on_web_request_start(
self, app: AppT, request: web.Request, *, view: web.View = None
Expand Down Expand Up @@ -633,3 +635,10 @@ def _normalize(
substitution: str = RE_NORMALIZE_SUBSTITUTION
) -> str:
return pattern.sub(substitution, name)

def _clear_topic_related_sensors(self) -> None:
self.tp_committed_offsets.clear()
self.tp_read_offsets.clear()
self.tp_end_offsets.clear()
self.topic_buffer_full.clear()
self.stream_inbound_time.clear()

0 comments on commit 0a4d059

Please sign in to comment.