Reference

def configure_logger(logger_conf_path: str):
    """Configure strefi logger.
    By default, the logger writes in .strefi.log and rotates 5 times each 10 MB.
    You can specify your own log configuration file in configparser file format.
    See also <https://docs.python.org/3/library/logging.config.html#logging-config-fileformat>

    Args:
        logger_conf_path: path of the configuration file (configparser file format).

    """
    if logger_conf_path is None:
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(name)s - %(threadName)s - %(levelname)s - %(message)s",
            datefmt="%Y-%m-%dT%H:%M:%SZ",
            handlers=[RotatingFileHandler(filename=".strefi.log", maxBytes=10 * 1024 * 1024, backupCount=5)],
        )
    else:
        config.fileConfig(logger_conf_path, disable_existing_loggers=False)

def main():
    """Entrypoint function. Launch the right method."""
    args = sys.argv[1:]
    namespace = parse_args(args)
    configure_logger(namespace.log)
    logger.debug(f"Strefi called with args {args}.")
    if namespace.command.lower() == "start":
        command.start(namespace.config)
    elif namespace.command.lower() == "stop":
        command.stop(namespace.jobid)
    elif namespace.command.lower() == "ls":
        command.ls()

def parse_args(args: list[str]) -> argparse.Namespace:
    """Read and verify program arguments.
    Exit the program if arguments are invalid.

    Args:
        args: list of program arguments.

    Returns:
        Argparse namespace.
    """
    parser = argparse.ArgumentParser(
        prog="strefi",
        description="Stream each new rows of a file and write in kafka",
        epilog="More information on GitHub",
    )
    parser.add_argument("command", help='"start" to launch stream or "stop" to kill stream')
    parser.add_argument("-c", "--config", help="configuration file path")
    parser.add_argument("-i", "--jobid", help="stream id")
    parser.add_argument("-l", "--log", help="log configuration file path (configparser file format)")

    namespace = parser.parse_args(args)

    if namespace.command.lower() not in ["start", "stop", "ls"]:
        parser.error(f"unknown command {namespace.command}")

    if namespace.command.lower() == "start":
        if namespace.config is None:
            parser.error("missing configuration file path")
    elif namespace.command.lower() == "stop":
        if namespace.jobid is None:
            parser.error("missing configuration job id")

    return namespace

def ls():
    """Display jobs with their status."""
    logger.debug("ls command is called.")
    jobs = supervisor.get_job_status()
    for job in jobs:
        status = "\033[92m RUNNING \033[00m" if job["status"] else "\033[91m FAILED \033[00m"
        print(f"{job['job_id']} \t {job['file']} \t {job['topic']} \t {status}")

def start(config_path: str):
    """Read configuration file and launch all streams.
    This function is the entrypoint of stefi start command.
    One thread with his own producer will be launched for each file.

    Args:
        config_path: Configuration file path.
    """
    logger.debug(f"start command is called with configuration {config_path}.")
    with open(config_path, "r") as f:
        config = json.loads(f.read())

    headers = kafka_utils.format_record_headers(config["headers"])

    for file in config["files"].keys():
        producer = kafka_utils.create_producer(config["producer"])
        topic = config["files"][file]
        running_path = supervisor.write_running_file(file, config["files"][file])
        thread = threading.Thread(
            target=parser.file_rows_to_topic,
            args=(file, topic, producer, config["defaults"], headers, running_path),
            name=running_path,
        )
        print(f"{re.findall(r'strefi_([a-zA-Z0-9]*)_', running_path)[0]}: {file} --> {topic}")
        thread.start()
        logger.debug(f"Thread {thread.name} started for file {file} in topic {topic}.")

    supervisor.update_running_file()

def stop(jobid: str):
    """Stop strefi threads.
    This function is the entrypoint of strefi stop command.

    Args:
        jobid: ID of the stream to kill, 'all' to kill all streams.
    """
    logger.debug(f"stop command is called for {jobid}.")
    supervisor.remove_running_file(jobid)

def file_rows_to_topic(
    file_path: str,
    topic: str,
    producer: KafkaProducer,
    defaults: dict[str, object],
    headers: dict[str, object],
    running_path: str,
):
    """Stream file and write last row in a kafka topic.
    This function is the streamed file thread entrypoint.

    Args:
        file_path: File path to stream.
        topic: Name of the target topic.
        producer: Instance of KafkaProducer.
        defaults: Configured dictionary to add in the record value.
        headers: Configured headers dictionary.
        running_path: Running file path
    """
    try:
        for line in stream_file(file_path, running_path, from_beginning=False):
            producer.send(topic, kafka_utils.format_record_value(file_path, line, defaults).encode(), headers=headers)
    except Exception as e:  # pragma: no cover
        logger.error(e)
        raise e

def stream_file(file_path: str, running_path: str, from_beginning: bool = True) -> Iterator[str]:
    """Wait file creation before calling yield_last_lines.
        Choose this method if you want stream log file which doesn't exist yet.

    Args:
        file_path: File path to stream.
        running_path: Path of running file. The function terminate when it's removed.
        from_beginning: If true, whole file will be streamed, if false, just new rows will be streamed.

    Returns:
        Yield the last line. Ignore empty line.
    """
    while os.path.exists(running_path):
        if os.path.exists(file_path):
            with open(file_path, "r") as file:
                if not from_beginning:
                    file.seek(0, os.SEEK_END)
                yield from yield_last_lines(file, running_path)
        from_beginning = True

def yield_last_lines(file: TextIO, running_path: str) -> Iterator[str]:
    """Yield last line of a file.
    This function terminates in 3 cases:
    The running file is removed, in this case all program stop.
    The streamed file is removed, in this case the program still running and waits for the file creation.
    One or more bytes were deleted from the streamed file,
    in this case the function is terminated, but it's called back just after.

    Args:
        file: Read-open file to stream.
        running_path: Path of running file. The function terminate when it's removed.

    Returns:
        Yield the last line. Ignore empty line.
    """
    logger.info(f"Starting stream {file.name}.")
    file_size = 0
    while os.path.exists(running_path):
        try:
            new_file_size = os.path.getsize(file.name)
            if new_file_size < file_size:
                logger.info(f"Some bytes were removed from {file.name}. Streaming finished.")
                break
            else:
                file_size = new_file_size
            lines = file.readlines()
            for line in lines:
                if line and line not in ["\n", ""]:
                    yield line
            time.sleep(0.5)
        except FileNotFoundError:
            logger.info(f"{file.name} was removed. Streaming finished.")
            break

def get_job_status() -> list[dict[str, object]]:
    """Return a list of all strefi threads with current status.
    If status == True, the thread is active, if status == False the thread is dead.

    Examples:
        >>> get_job_status()
        [{'job_id': '163892105422928641', 'file': 'file_a', 'topic': 'topic_a', 'status': False}]

    Returns:
        list metadata dictionaries
    """
    job_status = []
    running_file_names = [file for file in os.listdir(tempfile.gettempdir()) if "strefi" in file]
    for running_file_name in running_file_names:
        with open(os.path.join(tempfile.gettempdir(), running_file_name), "r") as f:
            running_info = json.loads(f.read())
            job_status.append(
                {
                    "job_id": re.findall(r"strefi_([a-zA-Z0-9]*)_", running_file_name)[0],
                    "file": running_info["file"],
                    "topic": running_info["topic"],
                    "status": True if time.time() - running_info["heartbeat"] < 16 else False,
                }
            )
    return job_status

def remove_running_file(job_id: str):
    """Remove a running file from a job id.

    Args:
        job_id: job id of the running file to delete. 'all' to delete all strefi running file.
    """
    job_id = "" if job_id == "all" else job_id
    running_file_names = [file for file in os.listdir(tempfile.gettempdir()) if f"strefi_{str(job_id)}" in file]
    for running_file_name in running_file_names:
        os.remove(os.path.join(tempfile.gettempdir(), running_file_name))
        logger.debug(f"{os.path.join(tempfile.gettempdir(), running_file_name)} was removed.")

def update_running_file():
    """Update the heartbeat timestamp in whole running file of active threads.
    The heartbeat is updated every 15 seconds. If the heartbeat is older than 15 seconds, the strefi thread is dead.
    """
    active_thread_running_paths = [thread.name for thread in threading.enumerate() if "strefi_" in thread.name]
    if active_thread_running_paths:
        for running_path in active_thread_running_paths:
            with open(running_path, "r") as f:
                running_info = json.loads(f.read())
            with open(running_path, "w") as f:
                f.write(RUNNING_FILE_PATTERN.format(running_info["file"], running_info["topic"], time.time()))
            logger.debug(f"{running_path} heartbeat was updated.")
        time.sleep(15)
        update_running_file()

def write_running_file(streamed_file_path: str, topic: str) -> str:
    """Create a running file in /tmp directory.
    The file name is composed with an ID generated.
    This id identify the stream and is used to delete running file.

    Returns:
        Absolute path of the running file.
    """
    job_id = abs(hash(time.time()))
    running_file = tempfile.NamedTemporaryFile(prefix=f"strefi_{job_id}_", delete=False)
    with open(running_file.name, "w") as f:
        f.write(RUNNING_FILE_PATTERN.format(streamed_file_path, topic, time.time()))
        logger.debug(f"Running file was created {running_file.name}.")
    return running_file.name

def create_producer(producer_config: dict[str, object]) -> KafkaProducer:
    """Initialize a KafkaProducer instance from dictionary.

    Examples:
        >>> create_producer({"bootstrap_servers":"localhost:9092", "client_id":"strefi-client", "acks": 0})

    Args:
        producer_config: Configuration dictionary.

    Returns:
        KafkaProducer instance.
    """
    return KafkaProducer(**producer_config)

def format_record_headers(headers_config: dict[str, str]) -> list[tuple[str, bytes]]:
    """Create record headers from dictionary.

    Examples:
        >>> format_record_headers({"version": "0.1", "type": "json"})
        [('version', b'0.1'), ('type', b'json')]

    Args:
        headers_config: Configuration dictionary.

    Returns:
        List of record headers. Values are encoded.
    """
    return [(k, headers_config[k].encode()) for k in headers_config.keys()]

def format_record_value(file_path: str, row: str, defaults_config: dict[str, object]) -> str:
    """Serialize record value.

    Examples:
        >>> format_record_value("/path/to/file", "foo", {"hostname":"lpt01", "system":"Linux"})
        "{'file': '/path/to/file', 'row': 'foo', 'hostname': 'lpt01', 'system': 'Linux'}"

    Args:
        file_path: Path of the stream file.
        row: Last row writen in the streamed file.
        defaults_config: Configured dictionary to add in the record value.

    Returns:
        Formatted string record value.
    """
    record: dict[str, object] = {"file": file_path, "row": row}
    record.update(dict(defaults_config))
    return json.dumps(record)