### 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>
if ti.state in (TaskInstanceState.RUNNING, TaskInstanceState.DEFERRED) and not executor_messages:
            served_messages, served_logs = self._read_from_logs_server(ti, worker_log_rel_path)
            messages_list.extend(served_messages)
        elif ti.state not in State.unfinished and not (local_logs or remote_logs):
            
            
            
            served_messages, served_logs = self._read_from_logs_server(ti, worker_log_rel_path)
            messages_list.extend(served_messages)
        logs = "\n".join(
            _interleave_logs(
                *local_logs,
                *remote_logs,
                *(executor_logs or []),
                *served_logs,
            )
        )
        log_pos = len(logs)
        messages = "".join([f"*** {x}\n" for x in messages_list])
        end_of_log = ti.try_number != try_number or ti.state not in (
            TaskInstanceState.RUNNING,
            TaskInstanceState.DEFERRED,
        )
        if metadata and "log_pos" in metadata:
            previous_chars = metadata["log_pos"]
            logs = logs[previous_chars:]  
        out_message = logs if "log_pos" in (metadata or {}) else messages + logs
        return out_message, {"end_of_log": end_of_log, "log_pos": log_pos}
    @staticmethod
    def _get_pod_namespace(ti: TaskInstance):