### 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>
"""
        Provide task_instance context to airflow task handler.
        Generally speaking returns None.  But if attr `maintain_propagate` has
        been set to propagate, then returns sentinel MAINTAIN_PROPAGATE. This
        has the effect of overriding the default behavior to set `propagate`
        to False whenever set_context is called.  At time of writing, this
        functionality is only used in unit testing.
        :param ti: task instance object
        """
        local_loc = self._init_file(ti)
        self.handler = NonCachingFileHandler(local_loc, encoding="utf-8")
        if self.formatter:
            self.handler.setFormatter(self.formatter)
        self.handler.setLevel(self.level)
        return SetContextPropagate.MAINTAIN_PROPAGATE if self.maintain_propagate else None
    @staticmethod
    def add_triggerer_suffix(full_path, job_id=None):
        """
        Helper for deriving trigger log filename from task log filename.
        E.g. given /path/to/file.log returns /path/to/file.log.trigger.123.log, where 123
        is the triggerer id.  We use the triggerer ID instead of trigger ID to distinguish
        the files because, rarely, the same trigger could get picked up by two different
        triggerer instances.
        """
        full_path = Path(full_path).as_posix()
        full_path += f".{LogType.TRIGGER.value}"
        if job_id:
            full_path += f".{job_id}.log"
        return full_path
    def emit(self, record):