Revert "Add distributed locking mechanism using Kazoo in Relay class, update configuration for Zookeeper connection, and modify main.py to integrate locking during data seeding operations. Update requirements to include Kazoo library."

This reverts commit a2977b2.

Author: robin-rpr

main.py

@@ -2,7 +2,6 @@
 from src.service import RIS, RouteViews
 from src.utils.config import Config
 from src.utils.logger import Logger
-from src.utils.lock import Lock
 from src.libs.bmp import BMPv3
 from src.seeder import Seeder
 from src.sender import Sender
@@ -30,7 +29,6 @@ class Relay():
     def __init__(self):
         self.queue = queueio.Queue(maxsize=10000000)
         self.executor = ThreadPoolExecutor(max_workers=4)
-        self.lock = Lock(Config.ZOOKEEPER_CONNECT, Config.RELAY_NAME)
         self.db = rocksdbpy.open_default("/var/lib/relay/" + Config.RELAY_NAME)
         self.loop = asyncio.get_event_loop()
         self.events = {
@@ -77,7 +75,6 @@ def _logger(self):
     def stop(self):
         self.events['shutdown'].set()
         self.executor.shutdown(wait=False, cancel_futures=True)
-        self.lock.stop()
         time.sleep(10)
         os._exit(1)
 
@@ -88,10 +85,6 @@ def start(self):
             if not self.is_sane():
                 raise RuntimeError("Database not sane")
 
-            # Start the lock.
-            self.lock.start()
-
-            # Add signal handlers.
             self.loop.add_signal_handler(signal.SIGTERM, self.stop)
             self.loop.add_signal_handler(signal.SIGINT, self.stop)
 
@@ -111,7 +104,6 @@ def start(self):
                     # Seed the collector.
                     # This is a one-off operation.
                     if self.db.get(b'seeded') != b'\x01':
-                        self.lock.request() # Acquire the lock.
                         index = f"{Config.RELAY_INDEX}/{datetime.now().year}.{datetime.now().month:02d}"
                         filename = f"bview.{datetime.now().year}{datetime.now().month:02d}{datetime.now().day:02d}.0000.gz"
                         self.loop.run_in_executor(self.executor,
@@ -120,7 +112,7 @@ def start(self):
                                 url=f"{index}/{filename}",
                                 events=self.events,
                                 memory=self.memory,
-                                db=self.db
+                                db=self.db,
                             ).start())
                     else:
                         self.events['seeded'].set()
@@ -160,7 +152,6 @@ def start(self):
 
                     # Seed the collector.
                     # This is a recurring operation.
-                    self.lock.request() # Acquire the lock.
                     index = f"{Config.RELAY_INDEX}/{datetime.now().year}.{datetime.now().month:02d}/RIBS"
                     response = requests.get(index, timeout=300)
                     response.raise_for_status()
@@ -174,7 +165,7 @@ def start(self):
                             url=f"{index}/{filename}",
                             events=self.events,
                             memory=self.memory,
-                            db=self.db
+                            db=self.db,
                         ).start())
 
                     # RouteViews (University of Oregon)

requirements.txt

@@ -3,5 +3,4 @@ requests==2.32.3
 beautifulsoup4==4.13.3
 confluent-kafka==2.5.3
 rocksdb-py==0.0.5
-scapy==2.6.1
-kazoo==2.10.0
+scapy==2.6.1

src/utils/config.py

@@ -8,7 +8,6 @@ class Config():
     RELAY_NAME = os.getenv('RELAY_NAME')
     RELAY_INDEX = os.getenv('RELAY_INDEX')
     COLLECTOR_CONNECT = os.getenv('COLLECTOR_CONNECT')
-    ZOOKEEPER_CONNECT = os.getenv('ZOOKEEPER_CONNECT')
 
     @staticmethod
     def validate():
@@ -18,7 +17,6 @@ def validate():
             'RELAY_NAME': Config.RELAY_NAME,
             'RELAY_INDEX': Config.RELAY_INDEX,
             'COLLECTOR_CONNECT': Config.COLLECTOR_CONNECT,
-            'ZOOKEEPER_CONNECT': Config.ZOOKEEPER_CONNECT,
         }
         for key, val in required.items():
             if not val: