feat：服务注册发现

unit-executor/main.py

@@ -1,17 +1,54 @@
+import socket
+import sys
+import uuid
 import grpc
+from typing import Tuple
+import settings
+from register.consul import ConsulRegister
 from services.executor import ApiRunServer
 from protos import executor_pb2_grpc
 import asyncio
 from utils.logger import logger
 
 
+def get_ip_port() -> Tuple[str, int]:
+    sock_ip = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock_ip.connect(('8.8.8.8', 80))
+    ip = sock_ip.getsockname()[0]
+    sock_ip.close()
+
+    sock_port = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock_port.bind(("", 0))
+    _, port = sock_port.getsockname()
+    sock_port.close()
+    return ip, port
+
+
 async def main():
+    ip, port = get_ip_port()
     server = grpc.aio.server()
     executor_pb2_grpc.add_ExecutorServiceServicer_to_server(ApiRunServer(), server)
-    server.add_insecure_port("0.0.0.0:5001")
+    server.add_insecure_port(f"{ip}:{port}")
+
+    service_id = str(uuid.uuid1())
+
+    consul = ConsulRegister(settings.CONSUL_HOST, settings.CONSUL_PORT)
+    if not consul.register_consul(
+            name=settings.SERVICE_NAME,
+            service_id=service_id,
+            address=ip, port=port,
+            tags=settings.SERVICE_TAGS
+    ):
+        logger.info(f"{settings.SERVICE_NAME} 服务注册失败")
+
+        sys.exit(0)
+
     await server.start()
-    logger.info("启动服务: 0.0.0.0:5001")
-    await server.wait_for_termination()
+    logger.info(f"gRPC服务已经启动：{ip}:{port}")
+    try:
+        await server.wait_for_termination()
+    finally:
+        consul.deregister(service_id)
 
 
 if __name__ == '__main__':

unit-executor/register/consul.py

@@ -3,65 +3,32 @@
 import consul
 import requests
 import random
-from register import base
 
 
-class ConsulRegister(base.Register, ABC):
+class ConsulRegister(object):
     def __init__(self, host, port):
         self.host = host
         self.port = port
         self.consul = consul.Consul(host=host, port=port)
 
     def register_consul(
             self,
-            name: str,
-            address: str,
-            port: int,
-            tags: str,
-            check: dict
+            name,
+            service_id,
+            address,
+            port,
+            tags
     ):
 
-        if check is None:
-            check = {
-                "GRPC": f"{address}:{port}",
-                "GRPCUseTLS": False,
-                "Timeout": "5s",
-                "Interval": "5s",
-                "DeregisterCriticalServiceAfter": "15s"
-            }
-        else:
-            check = check
-
-        service_id = uuid.uuid4().hex
-        return self.consul.agent.service.register(
+        register = self.consul.agent.service.register(
             name=name,
             service_id=service_id,
             address=address,
             port=port,
             tags=tags,
-            check=check
+            check=consul.Check.tcp(host=address, port=port, interval='10s')
         )
+        return register
 
     def deregister(self, service_id):
         return self.consul.agent.service.deregister(service_id)
-
-    def get_all_service(self):
-        return self.consul.agent.services()
-
-    def filter_service(self, filter):
-        url = f"http://{self.host}:{self.port}/v1/agent/services"
-        params = {
-            "filter": filter
-        }
-        return requests.get(url, params=params).json()
-
-    def get_host_port(self, filter):
-        url = f"http://{self.host}:{self.port}/v1/agent/services"
-        params = {
-            "filter": filter
-        }
-        data = requests.get(url, params=params).json()
-        if data:
-            service_info = random.choice(list(data.values()))
-            return service_info["Address"], service_info["Port"]
-        return None, None

unit-executor/requirements.txt

@@ -0,0 +1,16 @@
+atomic_bomb_engine==0.41.3
+Faker==8.11.0
+grpcio==1.64.1
+jmespath==1.0.1
+jsonpath==0.82
+loguru==0.6.0
+MySQL-python==1.2.5
+numpy==1.24.2
+protobuf==5.28.0
+pydantic==2.8.2
+python_dateutil==2.8.2
+Requests==2.32.3
+requests_toolbelt==1.0.0
+rsa==4.7.2
+simplejson==3.17.0
+urllib3==2.2.2

unit-executor/settings/__init__.py

@@ -3,3 +3,12 @@
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
 BASE_LOG_DIR = os.path.join(BASE_DIR, "logs")
+
+
+# consul的配置
+CONSUL_HOST = "127.0.0.1"
+CONSUL_PORT = "8500"
+
+# 服务相关的配置
+SERVICE_NAME = "unit_executor"
+SERVICE_TAGS = ['unit_executor', 'grpc']