Set clients host

client_go/client.go

@@ -12,7 +12,7 @@ import (
 	"github.com/kysre/TurtleMQ/client_go/queue"
 )
 
-const HOST = "localhost"
+const HOST = "64.226.122.208"
 
 type SubscribeFunction func(key string, value []byte)
 

client_py/client.py

@@ -13,7 +13,7 @@
 class QueueClient:
     stub = None
     replica_stub = None
-    HOST = "localhost"
+    HOST = "64.226.122.208"
     PORT, REPLICA_PORT = "8000", "8001"
     SUBSCRIBE_WORKERS = 3
     SUBSCRIBE_SLEEP_TIMEOUT = 2

datanode/src/datanode_server.py

@@ -13,6 +13,7 @@
 from prometheus_client import Counter, Gauge, Summary, Histogram, generate_latest, REGISTRY, start_http_server
 import os
 import time
+from threading import Thread
 
 DISK_TOTAL_SIZE = Gauge('disk_total_size', 'Total size of disk', labelnames=["provider"])
 DISK_USED_SIZE = Gauge('disk_used_size', 'Used size of disk', labelnames=["provider"])
@@ -176,6 +177,20 @@ def push_to_partition(partition_index: int,
     shared_partition.push(partition_message, partition_index)
 
 
+def notify_leader_task():
+    datanode_name, port = ConfigManager.get_prop('datanode_name'), ConfigManager.get_prop('server_port')
+    leader_host, leader_port = ConfigManager.get_prop('leader_host'), ConfigManager.get_prop('leader_port')
+    while True:
+        try:
+            channel = grpc.insecure_channel(f"{leader_host}:{leader_port}")
+            stub = leader_pb2_grpc.LeaderStub(channel)
+            add_request = leader_pb2.AddDataNodeRequest(address=f'{datanode_name}:{port}')
+            stub.AddDataNode(add_request)
+        except grpc.RpcError as e:
+            logger.exception(f"Error in notifying leader: {e}.")
+        time.sleep(5)
+
+
 def serve():
     # Start metrics server
     start_http_server(9000)
@@ -198,14 +213,8 @@ def serve():
     logger.info('Server started, listening on ' + port)
 
     # notify leader
-    try:
-        leader_host, leader_port = ConfigManager.get_prop('leader_host'), ConfigManager.get_prop('leader_port')
-        channel = grpc.insecure_channel(f"{leader_host}:{leader_port}")
-        stub = leader_pb2_grpc.LeaderStub(channel)
-        add_request = leader_pb2.AddDataNodeRequest(address=f'{datanode_name}:{port}')
-        stub.AddDataNode(add_request)
-    except grpc.RpcError as e:
-        logger.exception(f"Error in notifying leader: {e}.")
+    notify_leader_task_thread = Thread(target=notify_leader_task)
+    notify_leader_task_thread.start()
 
     server.wait_for_termination()
 

docker-compose.yaml

@@ -9,6 +9,19 @@ services:
     volumes:
       - ./prometheus:/etc/prometheus
       - prom_data:/prometheus
+    deploy:
+      resources:
+        limits:
+          cpus: '0.2'
+          memory: 300M
+
+  node_exporter:
+    image: prom/node-exporter:latest
+    deploy:
+      resources:
+        limits:
+          cpus: '0.1'
+          memory: 100M
 
   grafana:
     image: grafana/grafana
@@ -23,6 +36,11 @@ services:
     volumes:
       - ./grafana:/etc/grafana/provisioning/datasources
       - grafana_data:/var/lib/grafana
+    deploy:
+      resources:
+        limits:
+          cpus: '0.4'
+          memory: 400M
 
   leader_0:
     image: kysre/turtlemq:leader-${LEADER_IMAGE_TAG}
@@ -36,6 +54,11 @@ services:
       - LEADER_LEADER_REPLICAHOST=leader_1
     ports:
       - '8000:8888'
+    deploy:
+      resources:
+        limits:
+          cpus: '0.2'
+          memory: 200M
 
   leader_1:
     image: kysre/turtlemq:leader-${LEADER_IMAGE_TAG}
@@ -50,6 +73,11 @@ services:
       - LEADER_LEADER_REPLICAHOST=leader_0
     ports:
       - '8001:8888'
+    deploy:
+      resources:
+        limits:
+          cpus: '0.2'
+          memory: 200M
 
   datanode_0:
     image: kysre/turtlemq:datanode-${DATANODE_IMAGE_TAG}
@@ -69,6 +97,11 @@ services:
       - PARTITIONS_COUNT=100
     volumes:
       - datanode_0_vol:/var/lib/turtlemq/data/
+    deploy:
+      resources:
+        limits:
+          cpus: '0.15'
+          memory: 300M
 
   datanode_1:
     image: kysre/turtlemq:datanode-${DATANODE_IMAGE_TAG}
@@ -88,6 +121,11 @@ services:
       - PARTITIONS_COUNT=100
     volumes:
       - datanode_1_vol:/var/lib/turtlemq/data/
+    deploy:
+      resources:
+        limits:
+          cpus: '0.15'
+          memory: 300M
 
   datanode_2:
     image: kysre/turtlemq:datanode-${DATANODE_IMAGE_TAG}
@@ -107,6 +145,11 @@ services:
       - PARTITIONS_COUNT=100
     volumes:
       - datanode_2_vol:/var/lib/turtlemq/data/
+    deploy:
+      resources:
+        limits:
+          cpus: '0.15'
+          memory: 300M
 
   datanode_3:
     image: kysre/turtlemq:datanode-${DATANODE_IMAGE_TAG}
@@ -126,6 +169,11 @@ services:
       - PARTITIONS_COUNT=100
     volumes:
       - datanode_3_vol:/var/lib/turtlemq/data/
+    deploy:
+      resources:
+        limits:
+          cpus: '0.15'
+          memory: 300M
 
 volumes:
   prom_data:
@@ -134,9 +182,17 @@ volumes:
     driver: local
   datanode_0_vol:
     driver: local
+    driver_opts:
+      o: "size=5GB"
   datanode_1_vol:
     driver: local
+    driver_opts:
+      o: "size=5GB"
   datanode_2_vol:
     driver: local
+    driver_opts:
+      o: "size=5GB"
   datanode_3_vol:
     driver: local
+    driver_opts:
+      o: "size=5GB"

functional_tests.py

@@ -0,0 +1,26 @@
+import time
+
+from client_py.client import QueueClient
+
+
+def test_subscribe_function_1(key: str, value: bytes):
+    print(f'{key}, {value}')
+
+
+def test_subscribe_function_2(key: str, value: bytes):
+    print(f'{key}, {value}')
+
+
+if __name__ == '__main__':
+    test_client_1 = QueueClient()
+    test_client_2 = QueueClient()
+    test_client_3 = QueueClient()
+
+    test_client_2.subscribe(test_subscribe_function_1)
+    test_client_3.subscribe(test_subscribe_function_2)
+
+    for i in range(200):
+        test_client_1.push(f'{i}', b'value')
+        time.sleep(0.1)
+
+    time.sleep(500)