Merge pull request #114 from smileisak/freeze-all-operators

Freeze all operators with same priority and warn

.gitignore

@@ -25,3 +25,9 @@ junit.xml
 # Documentation
 docs/_build
 docs/packages
+
+# VirtualEnv
+env
+
+# VSCode
+.vscode

docs/peering.rst

@@ -14,7 +14,8 @@ notices that other operators start with a higher priority, it freezes
 its operation until those operators stop working.
 
 This is done to prevent collisions of multiple operators handling
-the same objects.
+the same objects. If two operators runs with the same priority  all operators 
+issue a warning and freeze, so that the cluster becomes not served anymore.
 
 To set the operator's priority, use :option:`--priority`:
 

kopf/engines/peering.py

@@ -36,14 +36,13 @@
 import os
 import random
 import socket
-from typing import Optional, Mapping, Iterable, Union
+from typing import Iterable, Mapping, Optional, Union
 
 import iso8601
 
 from kopf.clients import fetching
 from kopf.clients import patching
 from kopf.reactor import registries
-
 logger = logging.getLogger(__name__)
 
 # The CRD info on the special sync-object.
@@ -231,12 +230,14 @@ async def peers_handler(
         if not freeze.is_set():
             logger.info(f"Freezing operations in favour of {prio_peers}.")
             freeze.set()
-    else:
-        if same_peers:
-            logger.warning(f"Possibly conflicting operators with the same priority: {same_peers}.")
-        if freeze.is_set():
-            logger.info(f"Resuming operations after the freeze.")
-            freeze.clear()
+
+    elif same_peers:
+        logger.warning(f"Possibly conflicting operators with the same priority: {same_peers}.")
+        logger.warning(f"Freezed all Operators: {peers}")
+        freeze.set()
+    elif freeze.is_set():
+        logger.info(f"Resuming operations after the freeze. Conflicting operators with the same priority are gone")
+        freeze.clear()
 
 
 async def peers_keepalive(
@@ -256,9 +257,12 @@ async def peers_keepalive(
             await asyncio.sleep(max(1, int(ourselves.lifetime.total_seconds() - 10)))
     finally:
         try:
-            await ourselves.disappear()
-        except:
+            await asyncio.shield(ourselves.disappear())
+        except asyncio.CancelledError:
+            # It is the cancellation of `keepalive()`, not of the shielded `disappear()`.
             pass
+        except Exception:
+            logger.exception(f"Couldn't remove self from the peering. Ignoring.")
 
 
 def detect_own_id() -> str: