feat: getting xray logs

marznode/service/service.proto

@@ -44,12 +44,16 @@ message XrayConfig {
   string configuration = 1;
 }
 
+message XrayLogsRequest {
+  bool include_buffer = 1;
+}
+
 service MarzService {
   rpc SyncUsers(stream UserData) returns (Empty);
   rpc RepopulateUsers(UsersData) returns (Empty);
   rpc FetchInbounds(Empty) returns (InboundsResponse);
   rpc FetchUsersStats(Empty) returns (UsersStats);
   rpc FetchXrayConfig(Empty) returns (XrayConfig);
   rpc RestartXray(XrayConfig) returns (InboundsResponse);
-  rpc StreamXrayLogs(Empty) returns(stream LogLine);
+  rpc StreamXrayLogs(XrayLogsRequest) returns(stream LogLine);
 }

marznode/service/service.py

@@ -13,7 +13,7 @@
 from marznode.xray_api.exceptions import EmailExistsError, EmailNotFoundError
 from marznode.xray_api.types.account import accounts_map
 from .service_grpc import MarzServiceBase
-from .service_pb2 import UserData, Empty, InboundsResponse, Inbound, UsersStats
+from .service_pb2 import UserData, Empty, InboundsResponse, Inbound, UsersStats, LogLine
 from .service_pb2 import XrayConfig as XrayConfig_pb2
 from .. import config
 from ..xray.base import XrayCore
@@ -130,12 +130,17 @@ async def FetchUsersStats(self,
         await stream.send_message(UsersStats(users_stats=user_stats))
 
     async def StreamXrayLogs(self,
-                             stream: 'grpclib.server.Stream[marznode.service.service_pb2.Empty,'
-                                     'marznode.service.service_pb2.LogLine]') -> None:
-        pass
-
-    async def FetchXrayConfig(self, stream: 'grpclib.server.Stream[marznode.service.service_pb2.Empty,'
-                                            'marznode.service.service_pb2.XrayConfig]') -> None:
+                             stream: Stream[Empty, LogLine]) -> None:
+        req = await stream.recv_message()
+        if req.include_buffer:
+            for line in self.xray.get_buffer():
+                await stream.send_message(LogLine(line=line))
+        log_stm = await self.xray.get_logs_stm()
+        async with log_stm:
+            async for line in log_stm:
+                await stream.send_message(LogLine(line=line))
+
+    async def FetchXrayConfig(self, stream: Stream[Empty, XrayConfig_pb2]) -> None:
         await stream.recv_message()
         with open(config.XRAY_CONFIG_PATH, 'r') as f:
             content = f.read()

marznode/service/service_grpc.py

@@ -39,7 +39,7 @@ async def RestartXray(self, stream: 'grpclib.server.Stream[marznode.service.serv
         pass
 
     @abc.abstractmethod
-    async def StreamXrayLogs(self, stream: 'grpclib.server.Stream[marznode.service.service_pb2.Empty, marznode.service.service_pb2.LogLine]') -> None:
+    async def StreamXrayLogs(self, stream: 'grpclib.server.Stream[marznode.service.service_pb2.XrayLogsRequest, marznode.service.service_pb2.LogLine]') -> None:
         pass
 
     def __mapping__(self) -> typing.Dict[str, grpclib.const.Handler]:
@@ -83,7 +83,7 @@ def __mapping__(self) -> typing.Dict[str, grpclib.const.Handler]:
             '/marznode.MarzService/StreamXrayLogs': grpclib.const.Handler(
                 self.StreamXrayLogs,
                 grpclib.const.Cardinality.UNARY_STREAM,
-                marznode.service.service_pb2.Empty,
+                marznode.service.service_pb2.XrayLogsRequest,
                 marznode.service.service_pb2.LogLine,
             ),
         }
@@ -131,6 +131,6 @@ def __init__(self, channel: grpclib.client.Channel) -> None:
         self.StreamXrayLogs = grpclib.client.UnaryStreamMethod(
             channel,
             '/marznode.MarzService/StreamXrayLogs',
-            marznode.service.service_pb2.Empty,
+            marznode.service.service_pb2.XrayLogsRequest,
             marznode.service.service_pb2.LogLine,
         )