feat: grafana loki backend

airflow-core/src/airflow/config_templates/airflow_local_settings.py

@@ -281,6 +281,36 @@ def _default_conn_name_from(mod_path, hook_name):
             )
         )
         remote_task_handler_kwargs = {}
+    elif remote_base_log_folder.startswith("loki://"):
+        from airflow.providers.grafana.loki.log.loki_task_handler import LokiRemoteLogIO
+
+        url_parts = urlsplit(remote_base_log_folder)
+        loki_host = f"http://{url_parts.netloc}"
+        if url_parts.port:
+            pass # netloc already includes port
+
+        REMOTE_TASK_LOG = LokiRemoteLogIO(
+            **(
+                {
+                    "base_log_folder": BASE_LOG_FOLDER,
+                    "host": loki_host,
+                    "delete_local_copy": delete_local_copy,
+                }
+                | remote_task_handler_kwargs
+            )
+        )
+        # Configure logging dictionary to use LokiTaskHandler for the webserver reads
+        LOKI_REMOTE_HANDLERS: dict[str, dict[str, str | bool | None]] = {
+            "task": {
+                "class": "airflow.providers.grafana.loki.log.loki_task_handler.LokiTaskHandler",
+                "formatter": "airflow",
+                "base_log_folder": BASE_LOG_FOLDER,
+                "host": loki_host,
+                "frontend": conf.get("logging", "loki_frontend_url", fallback=""),
+            },
+        }
+        DEFAULT_LOGGING_CONFIG["handlers"].update(LOKI_REMOTE_HANDLERS)
+        remote_task_handler_kwargs = {}
     elif ELASTICSEARCH_HOST:
         from airflow.providers.elasticsearch.log.es_task_handler import ElasticsearchRemoteLogIO
 

providers/grafana/README.rst

@@ -0,0 +1,58 @@
+====================================================
+apache-airflow-providers-grafana
+====================================================
+
+Content
+-------
+
+This package provides integrations with Grafana ecosystem, specifically Loki for remote task logging in Airflow.
+
+Architecture and Design Decisions
+---------------------------------
+
+Loki Logging Backend
+~~~~~~~~~~~~~~~~~~~~
+
+The Loki logging backend for Apache Airflow is engineered to scale massively without degrading the performance of your Grafana Loki cluster. It achieves this by carefully sidestepping **cardinality explosions** and leveraging modern TSDB (Time Series Database) features like **Bloom filters**.
+
+The Cardinality Trap
+^^^^^^^^^^^^^^^^^^^^
+
+Unlike Elasticsearch, which indexes every field by default, Grafana Loki relies on a minimalistic index. In Loki, an *Index Stream* is created for every unique combination of labels. 
+
+- **Good Labels**: ``{job="airflow_tasks", dag_id="my_dag"}`` (Yields a small, stable number of streams—typically one per DAG).
+- **Bad Labels**: ``{job="airflow_tasks", dag_id="my_dag", task_id="extract", run_id="scheduled__2023...", try_number="1"}``
+
+If high-cardinality metadata like ``run_id``, ``task_id``, and ``try_number`` are used as Loki labels, Airflow will generate an infinitely growing number of unique index streams. This triggers a cardinality explosion: Loki's ingesters run out of memory tracking millions of short-lived streams, the global index inflates, and search performance collapses.
+
+The Solution: JSON payloads and TSDB Bloom Filters
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+To keep the index small and performant, the Airflow Loki Provider drops high-cardinality identifiers from the labels entirely. Instead, it embeds them directly into the JSON log payload prior to uploading chunks to the Loki Push API:
+
+.. code-block:: json
+
+    {"event": "Fetching data...", "task_id": "extract", "run_id": "scheduled__...", "try_number": "1", "map_index": "-1"}
+
+When the Airflow UI requests logs for a specific task run, the ``LokiTaskHandler`` constructs an optimized LogQL query utilizing the ``| json`` parser:
+
+.. code-block:: text
+
+    {job="airflow_tasks", dag_id="my_dag"} | json | task_id="extract" | try_number="1"
+
+Behind the scenes, Loki's modern TSDB engine automatically builds mathematical **Bloom filters** for the structured data inside the log chunks during ingestion. 
+
+When executing the query, Loki instantly resolves the ``{job="airflow_tasks", dag_id="my_dag"}`` stream. Then, before downloading or decompressing any log chunk from object storage, Loki consults the chunk's Bloom filter:
+
+1. *Are the strings ``"extract"`` and ``"1"`` anywhere within this block?*
+2. The Bloom filter answers "No" with 100% certainty in microseconds, allowing Loki to skip the entire block of data.
+3. Loki only decompresses and evaluates the JSON parser on the specific chunks that mathematically *must* contain the target task's logs.
+
+**Operating Philosophy**: By embedding dynamic metadata into the JSON payload rather than stream labels, this provider guarantees full-text indexing performance while keeping Grafana Loki infrastructure costs near zero, ensuring stability regardless of how many millions of tasks Airflow executes.
+
+Upload Chunking Strategy
+^^^^^^^^^^^^^^^^^^^^^^^^
+
+Instead of streaming logs line-by-line via HTTP (which would bottleneck the Airflow worker) or sending gigantic multi-gigabyte files in a single request (which causes worker Out-Of-Memory crashes and reverse-proxy ``413 Entity Too Large`` HTTP rejections), the provider natively chunks log payloads by **Byte Size**.
+
+Every time a task finishes, the ``LokiRemoteLogIO`` handler iterates over the local structured JSON file and calculates the string payload length incrementally. Once reaching the standard Promtail sweet spot of ``1 MiB`` per-batch, the payload is immediately POSTed to the Loki API. This guarantees optimal network throughput and 100% compliance with default ingestion limits without sacrificing worker stability.

providers/grafana/provider.yaml

@@ -0,0 +1,22 @@
+package-name: apache-airflow-providers-grafana
+name: Grafana
+description: |
+    `Grafana <https://grafana.com/>`__
+state: ready
+source-date-epoch: 1714241088
+versions:
+  - 1.0.0
+
+dependencies:
+  - apache-airflow>=2.11.0
+  - requests>=2.27.0
+  - tenacity>=8.0.0
+
+integrations:
+  - integration-name: Grafana Loki
+    external-doc-url: https://grafana.com/oss/loki/
+    logo: /integration-logos/grafana/Loki.png
+    tags: [logging]
+
+logging:
+  - airflow.providers.grafana.loki.log.loki_task_handler.LokiTaskHandler

providers/grafana/pyproject.toml

@@ -0,0 +1,62 @@
+[build-system]
+requires = ["flit_core==3.12.0"]
+build-backend = "flit_core.buildapi"
+
+[project]
+name = "apache-airflow-providers-grafana"
+version = "1.0.0"
+description = "Provider package apache-airflow-providers-grafana for Apache Airflow"
+readme = "README.rst"
+license = "Apache-2.0"
+license-files = ['LICENSE', 'NOTICE']
+authors = [
+    {name="Apache Software Foundation", email="dev@airflow.apache.org"},
+]
+maintainers = [
+    {name="Apache Software Foundation", email="dev@airflow.apache.org"},
+]
+keywords = [ "airflow-provider", "grafana", "loki", "airflow", "integration" ]
+classifiers = [
+    "Development Status :: 5 - Production/Stable",
+    "Environment :: Console",
+    "Environment :: Web Environment",
+    "Intended Audience :: Developers",
+    "Intended Audience :: System Administrators",
+    "Framework :: Apache Airflow",
+    "Framework :: Apache Airflow :: Provider",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+]
+requires-python = ">=3.10"
+
+dependencies = [
+    "apache-airflow>=2.11.0",
+    "apache-airflow-providers-common-compat>=1.12.0",
+    "requests>=2.27.0",
+    "tenacity>=8.0.0",
+]
+
+[dependency-groups]
+dev = [
+    "apache-airflow",
+    "apache-airflow-task-sdk",
+    "apache-airflow-devel-common",
+    "apache-airflow-providers-common-compat",
+]
+
+docs = [
+    "apache-airflow-devel-common[docs]"
+]
+
+[tool.uv.sources]
+apache-airflow = {workspace = true}
+apache-airflow-devel-common = {workspace = true}
+apache-airflow-task-sdk = {workspace = true}
+
+[project.entry-points."apache_airflow_provider"]
+provider_info = "airflow.providers.grafana.get_provider_info:get_provider_info"
+
+[tool.flit.module]
+name = "airflow.providers.grafana"

providers/grafana/src/airflow/providers/grafana/__init__.py

@@ -0,0 +1 @@
+# Licensed to the Apache Software Foundation (ASF) under one

providers/grafana/src/airflow/providers/grafana/get_provider_info.py

@@ -0,0 +1,42 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# NOTE! THIS FILE IS AUTOMATICALLY GENERATED AND WILL BE OVERWRITTEN!
+
+# IF YOU WANT TO MODIFY THIS FILE EXCEPT DEPENDENCIES, YOU SHOULD MODIFY THE TEMPLATE
+# `get_provider_info_TEMPLATE.py.jinja2` IN the `dev/breeze/src/airflow_breeze/templates` DIRECTORY
+
+
+def get_provider_info():
+    return {
+        "package-name": "apache-airflow-providers-grafana",
+        "name": "Grafana",
+        "description": "`Grafana <https://grafana.com/>`__\n",
+        "state": "ready",
+        "source-date-epoch": 1714241088,
+        "versions": ["1.0.0"],
+        "integrations": [
+            {
+                "integration-name": "Grafana Loki",
+                "external-doc-url": "https://grafana.com/oss/loki/",
+                "logo": "/integration-logos/grafana/Loki.png",
+                "tags": ["logging"],
+            }
+        ],
+        "logging": ["airflow.providers.grafana.loki.log.loki_task_handler.LokiTaskHandler"],
+        "dependencies": ["apache-airflow>=2.11.0", "requests>=2.27.0", "tenacity>=8.0.0"],
+    }

providers/grafana/src/airflow/providers/grafana/loki/__init__.py

@@ -0,0 +1 @@
+# Licensed to the Apache Software Foundation (ASF) under one

providers/grafana/src/airflow/providers/grafana/loki/log/__init__.py

@@ -0,0 +1 @@
+# Licensed to the Apache Software Foundation (ASF) under one