logserver package

Submodules

logserver.server module

class logserver.server.LogServer

Bases: object

Main component of the Log Storage stage to enter data into the pipeline

Receives and sends single log lines. Simultaneously, listens for messages via Kafka and reads newly added lines from an input file. Sends every log line to a Kafka topic under which it is obtained by the next stage.

async fetch_from_file(file: str = '/opt/file.txt') → None

Starts a loop to continuously check for new lines at the end of the input file and sends them.

Checks are done every 0.1 seconds. If one or multiple new lines are found, any empty lines are removed and the remaining lines are sent individually. For each fetched log line, the unprocessed log line string including its timestamp (“timestamp_in”) is logged.

Parameters:

file (str) – Filename of the file to be read. Default: File configured in config.yaml (pipeline.log_storage.logserver.input_file)

async fetch_from_kafka() → None

Fetches data from the configured Kafka topic in a loop.

Starts an asynchronous loop to continuously fetch new data from the Kafka topic. When a message is consumed, the unprocessed log line string including its timestamp (“timestamp_in”) is logged.

send(message_id: UUID, message: str) → None

Sends a message using Kafka.

Logs the time of sending the message to Kafka as a “timestamp_out” event.

Parameters:

• message_id (uuid.UUID) – UUID of the message to be sent.

• message (str) – Message to be sent.

async start() → None

Starts the tasks to both fetch messages from Kafka and read them from the input file.

logserver.server.main() → None

Creates the LogServer instance and starts it.