### 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>
messages = []
        logs = []
        try:
            log_type = LogType.TRIGGER if ti.triggerer_job else LogType.WORKER
            url, rel_path = self._get_log_retrieval_url(ti, worker_log_rel_path, log_type=log_type)
            response = _fetch_logs_from_service(url, rel_path)
            if response.status_code == 403:
                messages.append(
                    "!!!! Please make sure that all your Airflow components (e.g. "
                    "schedulers, webservers, workers and triggerer) have "
                    "the same 'secret_key' configured in 'webserver' section and "
                    "time is synchronized on all your machines (for example with ntpd)\n"
                    "See more at https://airflow.apache.org/docs/apache-airflow/"
                    "stable/configurations-ref.html#secret-key"
                )
            
            response.raise_for_status()
            if response.text:
                messages.append(f"Found logs served from host {url}")
                logs.append(response.text)
        except Exception as e:
            messages.append(f"Could not read served logs: {str(e)}")
            logger.exception("Could not read served logs")
        return messages, logs
    def _read_remote_logs(self, ti, try_number, metadata=None) -> tuple[list[str], list[str]]: