Broker changes

.github/workflows/ci.yml

@@ -57,6 +57,7 @@ jobs:
         uses: docker/build-push-action@v5
         with:
           context: ./kafka_server/coordinator/
+          platforms: linux/amd64,linux/arm64
           push: true
           tags: mahdigheidi/sad-coordinator:latest
 
@@ -66,18 +67,18 @@ jobs:
       matrix:
         python-version: ["3.8", "3.9", "3.10"]
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v3
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install flake8 
+        pip install flake8
     - name: Analysing the code with flake8
       run: |
-        flake8 --ignore E501 kafka_server/
+        flake8 --ignore=E401,E402,E501 kafka_server/
     # - name: Test with pytest
     #   run: |
     #     pytest
@@ -88,10 +89,10 @@ jobs:
 
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
 
       - name: Set up Python
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v5
         with:
           python-version: 3.x
 
@@ -101,7 +102,7 @@ jobs:
 
       - name: Run bandit
         run: |
-          bandit -r .
+          bandit --skip B104 -r .
 
 
   deploy-prod-broker:
@@ -111,7 +112,7 @@ jobs:
 
     steps:
     - name: checkout repo
-      uses: actions/checkout@v2
+      uses: actions/checkout@v4
 
     - name: set up ssh keys
       uses: shimataro/ssh-key-action@v2
@@ -120,18 +121,41 @@ jobs:
         known_hosts: ${{ secrets.SSH_HOST }}
     - name: connect and pull
       run: |
-        ssh -i ~/.ssh/id_rsa [email protected] "docker compose up -d broker-1 broker-2 broker-3 && exit"
-    - name: cleanup
-      run: rm -rf ~/.ssh
+        ssh -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no ${{ secrets.SSH_USER }}@${{ secrets.SSH_HOST }} "docker-compose up -d broker-1 broker-2 broker-3 && exit"
 
   deploy-prod-coordinator:
     name: Deploy Coordinator to Production
-    runs-on: ubuntu-latest
+    runs-on: self-hosted
     needs: build-push-coordinator
 
     steps:
-      - name: Deploy to production
-        run: |
-          # Here you can add your deployment commands/scripts
-          # For example:
-          ssh user@your-production-server 'docker pull ghcr.io/your-github-username/your-repo-name:latest && docker-compose up -d'
+    - name: checkout repo
+      uses: actions/checkout@v4
+
+    - name: set up ssh keys
+      uses: shimataro/ssh-key-action@v2
+      with:
+        key: ${{ secrets.SSH_PRIVATE_KEY }}
+        known_hosts: ${{ secrets.SSH_HOST }}
+    - name: connect and pull
+      run: |
+        ls
+        ssh -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no ${{ secrets.SSH_USER }}@${{ secrets.SSH_HOST }} "docker-compose up -d coordinator-1 coordinator-2 && exit"
+
+  deploy-monitoring:
+    name: Deploy Monitoring to Production
+    runs-on: self-hosted
+
+    steps:
+    - name: checkout repo
+      uses: actions/checkout@v4
+
+    - name: set up ssh keys
+      uses: shimataro/ssh-key-action@v2
+      with:
+        key: ${{ secrets.SSH_PRIVATE_KEY }}
+        known_hosts: ${{ secrets.SSH_HOST }}
+    - name: connect and pull
+      run: |
+        ls
+        ssh -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no ${{ secrets.SSH_USER }}@${{ secrets.SSH_HOST }} "docker-compose up -d node_exporter prometheus grafana && exit"

docker-compose.yml

@@ -45,36 +45,45 @@ services:
 
   prometheus:
     image: prom/prometheus:latest
-    command:
-      - --storage.tsdb.retention.time=7d
-      - --config.file=/etc/prometheus/prometheus.yml
-      # - --storage.tsdb.retention.size=10GB
+    container_name: prometheus
+    restart: unless-stopped
+    volumes:
+      - ./monitoring/prometheus:/etc/prometheus/prometheus.yml
+      - prometheus-data:/prometheus
     ports:
       - "9090:9090"
-    volumes:
-      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
+    networks:
+      - broker
 
   grafana:
     image: grafana/grafana:latest
     ports:
       - "3000:3000"
     volumes:
-      - ./monitoring/grafana-provisioning/dashboards:/etc/grafana/provisioning/dashboards
-      - ./monitoring/grafana-provisioning/datasources:/etc/grafana/provisioning/datasources
-
-  node_exporter_1:
-    image: prom/node-exporter:latest
-    ports:
-      - "9100:9100"
+      - grafana-data:/var/lib/grafana
     networks:
       - broker
 
-  node_exporter_2:
-    image: prom/node-exporter:latest
+  node_exporter:
+    image: quay.io/prometheus/node-exporter:latest
+    container_name: node_exporter
+    command:
+      - '--path.rootfs=/host'
+    pid: host
     ports:
-      - "9101:9100"
+      - "9100:9100"
+    restart: unless-stopped
+    volumes:
+      - '/:/host:ro,rslave'
     networks:
       - broker
 
+volumes:
+  prometheus-data:
+    driver: local
+  grafana-data:
+    driver: local
+
 networks:
   broker:
+    driver: bridge

kafka_server/broker/Dockerfile

@@ -7,4 +7,5 @@ COPY . .
 RUN pip install -r requirements.txt
 EXPOSE 5003
 ENV FLASK_APP=app
-CMD ["python", "-u", "controller/produce.py"]
+CMD ["python", "-u", "controller/produce.py"]
+# CMD ["python", "controller/produce.py"]

kafka_server/broker/controller/coordinator.py

@@ -42,7 +42,7 @@ def _post(data, url: str):
         print(f"master coordinator {_master_coordinator_url()} is not alive")
 
     try:
-        response = requests.post(coordinator, json=data, data=json.dumps(data))
+        response = requests.post(coordinator, json=data, data=json.dumps(data), timeout=2)
         return response.status_code == 200
     except requests.RequestException as e:
         print(f"Error on post {e}")
@@ -55,7 +55,7 @@ def _get(data, url: str):
             coordinator = _url(master_not_replica=master_alive, url=url)
             if not master_alive:
                 print(f"master coordinator {_master_coordinator_url()} is not alive")
-            response = requests.get(coordinator, json=data, data=json.dumps(data))
+            response = requests.get(coordinator, json=data, data=json.dumps(data), timeout=2)
             if response.status_code == 200:
                 return json.loads(response.content.decode("utf-8"))
         except requests.RequestException as e:
@@ -78,12 +78,7 @@ def heartbeat():
             'ip': os.getenv("IP"),
             'port': os.getenv("PORT"),
         }
-
-        if _post(payload, heartbeat_url):
-            print(f"Heartbeat {payload}")
-        else:
-            print("Heartbeat Failed")
-
+        _post(payload, heartbeat_url)
         time.sleep(3)  # 3 Seconds wait for another heartbeat
 
 

kafka_server/broker/controller/produce.py

@@ -2,19 +2,21 @@
 import os
 import sys
 import threading
-from main import init
+
+
+BROKER_PROJECT_PATH = os.getenv("BROKER_PROJECT_PATH", "/app/")
+sys.path.append(os.path.abspath(BROKER_PROJECT_PATH))
+
 from file.indexer import Indexer
 from file.read import Read
 from file.write import Write
+from main import init
 from flask import Flask, request, jsonify
 from manager.env import get_primary_partition, get_replica_url
 from metrics import coordinator_write_requests, coordinator_replicate_index_requests
 from prometheus_client import make_wsgi_app
 from werkzeug.middleware.dispatcher import DispatcherMiddleware
 
-BROKER_PROJECT_PATH = os.getenv("BROKER_PROJECT_PATH", "/app/")
-sys.path.append(os.path.abspath(BROKER_PROJECT_PATH))
-
 
 app = Flask(__name__)
 app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {
@@ -34,7 +36,7 @@ def write():
         # Assuming the request body is in JSON format with 'key' and 'value' fields
         data = request.get_json()
         key = data.get('key')
-        value = data.get('value').encode('utf-8')
+        value = data.get('value')
         print(key, value)
 
         write_instance = Write(get_primary_partition(), get_replica_url())
@@ -98,7 +100,7 @@ def subscription():
         data = json.loads(request.data.decode("utf-8"))
         brokers = data['brokers']
 
-        brokers_file_path = os.path.join(os.getcwd(), '../data', 'subscriptions', 'brokers.json')
+        brokers_file_path = os.path.join(os.getcwd(), 'data', 'subscriptions', 'brokers.json')
 
         with open(brokers_file_path, "w") as file:
             json.dump(brokers, file)
@@ -120,7 +122,7 @@ def subscribers():
         data = json.loads(request.data.decode("utf-8"))
         brokers = data['subscribers']
 
-        subscribers_file_path = os.path.join(os.getcwd(), '../data', 'subscriptions', 'subscribers.json')
+        subscribers_file_path = os.path.join(os.getcwd(), 'data', 'subscriptions', 'subscribers.json')
 
         with open(subscribers_file_path, "w+") as file:
             json.dump(brokers, file)
@@ -137,6 +139,7 @@ def broker_down():
         data = json.loads(request.data.decode("utf-8"))
         partition = data['partition']
         os.environ['REPLICA_MIRROR_DOWN'] = str(partition)
+        print(os.environ['REPLICA_MIRROR_DOWN'], "replica mirror down")
         return jsonify({'status': 'Data written successfully.'}), 200
     except Exception as e:
         print(e)
@@ -182,4 +185,5 @@ def ack():
 crun.daemon = True
 crun.start()
 
-app.run('0.0.0.0', port=5003, debug=True)
+broker_listening_addr = '0.0.0.0'
+app.run(broker_listening_addr, port=5003)

kafka_server/broker/file/hash.py

@@ -2,4 +2,4 @@
 
 
 def hash_md5(key: str):
-    return hashlib.md5(key.encode()).hexdigest()
+    return hashlib.md5(key.encode(), usedforsecurity=False).hexdigest()

kafka_server/broker/file/indexer.py

@@ -13,17 +13,15 @@ class Indexer:
 
     def __new__(cls, partition: str, replica: str = None):
         with cls._lock:
-            if partition not in cls._instances:
-                cls._instances[partition] = super().__new__(cls)
-                cls._instances[partition].partition = partition
-                cls._instances[partition].replica = replica
-                cls._instances[partition]._write = 0
-                cls._instances[partition]._read = 0
-                cls._instances[partition]._sync = 0
-                cls._instances[partition].load()
-                path = cls._instances[partition].__dir_path()
+            if f"{partition}-{replica}" not in cls._instances:
+                cls._instances[f"{partition}-{replica}"] = super().__new__(cls)
+                cls._instances[f"{partition}-{replica}"].partition = partition
+                cls._instances[f"{partition}-{replica}"].replica = replica
+                cls._instances[f"{partition}-{replica}"].load()
+                path = cls._instances[f"{partition}-{replica}"].__dir_path()
                 os.makedirs(path, exist_ok=True)
-            return cls._instances[partition]
+
+            return cls._instances[f"{partition}-{replica}"]
 
     def load(self):
         self._write = self._load_variable('write')
@@ -106,9 +104,12 @@ def __dir_path(self) -> str:
 
     def send_to_replica(self):
         if self.replica is None:
+            print("No replica found /n/n/n")
             return
+
         url = f'{self.replica}/replica/index'
         data = {'partition': self.partition, 'read': self._read, 'sync': self._sync}
+        print(data, "to Replica")
         response = requests.post(url, json=data, timeout=2)
         if response.status_code != 200:
             raise Exception(f'indexed not yet updated {response}')