### 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>
"""Returns WasbHook."""
        remote_conn_id = conf.get("logging", "REMOTE_LOG_CONN_ID")
        try:
            from airflow.providers.microsoft.azure.hooks.wasb import WasbHook
            return WasbHook(remote_conn_id)
        except Exception:
            self.log.exception(
                "Could not create a WasbHook with connection id '%s'. "
                "Do you have apache-airflow[azure] installed? "
                "Does connection the connection exist, and is it "
                "configured properly?",
                remote_conn_id,
            )
            return None
    def set_context(self, ti) -> None:
        super().set_context(ti)
        
        
        if TYPE_CHECKING:
            assert self.handler is not None
        full_path = self.handler.baseFilename
        self.log_relative_path = Path(full_path).relative_to(self.local_base).as_posix()
        is_trigger_log_context = getattr(ti, "is_trigger_log_context", False)
        self.upload_on_close = is_trigger_log_context or not ti.raw
    def close(self) -> None: