57 lines
1.8 KiB
Python
Executable File
57 lines
1.8 KiB
Python
Executable File
#!/usr/bin/env python
|
|
import asyncio
|
|
import msgpack
|
|
import os
|
|
import struct
|
|
import socket
|
|
from motor.motor_asyncio import AsyncIOMotorClient
|
|
from datetime import datetime
|
|
|
|
MONGO_URI = os.getenv("MONGO_URI", "mongodb://127.0.0.1:27017/default")
|
|
FQDN = socket.getfqdn()
|
|
EXPIRE_AFTER_SECONDS = int(os.getenv("EXPIRE_AFTER_SECONDS", "1209600"))
|
|
|
|
db = AsyncIOMotorClient(MONGO_URI).get_default_database()
|
|
db.log.create_index("time", expireAfterSeconds=EXPIRE_AFTER_SECONDS)
|
|
|
|
class Server(object):
|
|
def __init__(self):
|
|
self._loop = asyncio.get_event_loop()
|
|
|
|
async def __call__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
|
|
unpacker = msgpack.Unpacker(raw=False)
|
|
try:
|
|
while True:
|
|
data = await reader.read(n=4096)
|
|
if not data:
|
|
raise ConnectionError("Client connection closed")
|
|
unpacker.feed(data)
|
|
for obj in unpacker:
|
|
self._loop.create_task(self._handle_request(obj, writer))
|
|
except ConnectionError:
|
|
pass
|
|
|
|
async def _handle_request(self, obj, writer: asyncio.StreamWriter) -> None:
|
|
tag, time, record, options = obj
|
|
seconds, nanoseconds = struct.unpack(">II", time.data)
|
|
assert not options, "Can't handle options"
|
|
record["time"] = datetime.fromtimestamp(seconds).replace(microsecond=int(nanoseconds/1000))
|
|
record.pop("container_id", None)
|
|
record["host"] = FQDN
|
|
result = await db.log.insert_one(record)
|
|
|
|
|
|
async def main():
|
|
try:
|
|
server = await asyncio.start_server(Server(), host="127.0.0.1", port=24224)
|
|
while True:
|
|
await asyncio.sleep(0.1)
|
|
finally:
|
|
server.close()
|
|
|
|
|
|
try:
|
|
asyncio.get_event_loop().run_until_complete(main())
|
|
except KeyboardInterrupt:
|
|
pass
|