def _finish_msg_batch(self, batch, results):
if not self._config.REPORT_ENDPOINT_STATUS:
_log.warning("StatusReporter called even though status reporting "
"disabled. Ignoring.")
self._endpoint_status[IPV4].clear()
self._endpoint_status[IPV6].clear()
self._newer_dirty_endpoints.clear()
self._older_dirty_endpoints.clear()
return
if self._cleanup_pending:
try:
self._attempt_cleanup()
except EtcdException as e:
_log.error("Cleanup failed: %r", e)
_stats.increment("Status report cleanup failed")
else:
_stats.increment("Status report cleanup done")
self._cleanup_pending = False
if self._reporting_allowed:
# We're not rate limited, go ahead and do a write to etcd.
_log.debug("Status reporting is allowed by rate limit.")
if not self._older_dirty_endpoints and self._newer_dirty_endpoints:
_log.debug("_older_dirty_endpoints empty, promoting"
"_newer_dirty_endpoints")
self._older_dirty_endpoints = self._newer_dirty_endpoints
self._newer_dirty_endpoints = set()
if self._older_dirty_endpoints:
ep_id = self._older_dirty_endpoints.pop()
status_v4 = self._endpoint_status[IPV4].get(ep_id)
status_v6 = self._endpoint_status[IPV6].get(ep_id)
status = combine_statuses(status_v4, status_v6)
try:
self._write_endpoint_status_to_etcd(ep_id, status)
except EtcdException:
_log.exception("Failed to report status for %s, will "
"retry", ep_id)
# Add it into the next dirty set. Retrying in the next
# batch ensures that we try to update all of the dirty
# endpoints before we do any retries, ensuring fairness.
self._newer_dirty_endpoints.add(ep_id)
# Reset the rate limit flag.
self._reporting_allowed = False
if not self._timer_scheduled and ((not self._reporting_allowed) or
self._cleanup_pending):
# Schedule a timer to stop our rate limiting or retry cleanup.
timeout = self._config.ENDPOINT_REPORT_DELAY
timeout *= (0.9 + (random.random() * 0.2)) # Jitter by +/- 10%.
gevent.spawn_later(timeout,
self._on_timer_pop,
async=True)
self._timer_scheduled = True
评论列表
文章目录
