### Discussed in https://github.com/apache/airflow/discussions/31832

<div type='discussions-op-text'>

<sup>Originally posted by **michalc** June 10, 2023</sup>
Should something like the below be in the codebase? It's a simple handler for storing Airflow task logs in Redis, enforcing a max number of entries per try, and an expiry time for the logs

Happy to raise a PR (and I guessed a lot at how things should be... so suspect can be improved upon...)

```python
class RedisHandler(logging.Handler):
    def __init__(self, client, key):
        super().__init__()
        self.client = client
        self.key = key

    def emit(self, record):
        p = self.client.pipeline()
        p.rpush(self.key, self.format(record))
        p.ltrim(self.key, start=-10000, end=-1)
        p.expire(self.key, time=60 * 60 * 24 * 28)
        p.execute()


class RedisTaskHandler(FileTaskHandler, LoggingMixin):
    """
    RedisTaskHandler is a python log handler that handles and reads
    task instance logs. It extends airflow FileTaskHandler and
    uploads to and reads from Redis.
    """

    trigger_should_wrap = True

    def __init__(self, base_log_folder: str, redis_url):
        super().__init__(base_log_folder)
        self.handler = None
        self.client = redis.Redis.from_url(redis_url)

    def _read(
        self,
        ti,
        try_number,
        metadata=None,
    ):
        log_str = b"\n".join(
            self.client.lrange(self._render_filename(ti, try_number), start=0, end=-1)
        ).decode("utf-8")
        return log_str, {"end_of_log": True}

    def set_context(self, ti):
        super().set_context(ti)
        self.handler = RedisHandler(
            self.client, self._render_filename(ti, ti.try_number)
        )
        self.handler.setFormatter(self.formatter)
```</div>
as the impersonated user. However, if the user manually runs tasks
        of the SubDagOperator through the UI, then the log files are created
        by the user that runs the Airflow command. For example, the Airflow
        run command may be run by the `airflow_sudoable` user, but the Airflow
        tasks may be run by the `airflow` user. If the log files are not
        writable by both users, then it's possible that re-running a task
        via the UI (or vice versa) results in a permission error as the task
        tries to write to a log file created by the other user.
        We leave it up to the user to manage their permissions by exposing configuration for both
        new folders and new log files. Default is to make new log folders and files group-writeable
        to handle most common impersonation use cases. The requirement in this case will be to make
        sure that the same group is set as default group for both - impersonated user and main airflow
        user.
        """
        new_folder_permissions = int(
            conf.get("logging", "file_task_handler_new_folder_permissions", fallback="0o775"), 8
        )
        directory.mkdir(mode=new_folder_permissions, parents=True, exist_ok=True)
        if directory.stat().st_mode % 0o1000 != new_folder_permissions % 0o1000:
            print(f"Changing {directory} permission to {new_folder_permissions}")
            try:
                directory.chmod(new_folder_permissions)
            except PermissionError as e:
                
                
                
                
                print(f"Failed to change {directory} permission to {new_folder_permissions}: {e}")
                pass
    def _init_file(self, ti):