feature: Add environment variable controlling the log grooming freque…

…ncy (apache#46237)

* New feature: Add env variable controlling the log grooming frequency

* pre-commit

* fix: copy+paste errors, adjust test

* format

Dockerfile

@@ -1271,10 +1271,11 @@ set -euo pipefail
 
 readonly DIRECTORY="${AIRFLOW_HOME:-/usr/local/airflow}"
 readonly RETENTION="${AIRFLOW__LOG_RETENTION_DAYS:-15}"
+readonly FREQUENCY="${AIRFLOW__LOG_CLEANUP_FREQUENCY_MINUTES:-15}"
 
 trap "exit" INT TERM
 
-readonly EVERY=$((15*60))
+readonly EVERY=$((FREQUENCY*60))
 
 echo "Cleaning logs every $EVERY seconds"
 

chart/templates/dag-processor/dag-processor-deployment.yaml

@@ -214,6 +214,10 @@ spec:
           {{- if .Values.dagProcessor.logGroomerSidecar.retentionDays }}
             - name: AIRFLOW__LOG_RETENTION_DAYS
               value: "{{ .Values.dagProcessor.logGroomerSidecar.retentionDays }}"
+          {{- end }}
+          {{- if .Values.dagProcessor.logGroomerSidecar.frequencyMinutes }}
+            - name: AIRFLOW__LOG_CLEANUP_FREQUENCY_MINUTES
+              value: "{{ .Values.dagProcessor.logGroomerSidecar.frequencyMinutes }}"
           {{- end }}
             - name: AIRFLOW_HOME
               value: "{{ .Values.airflowHome }}"

chart/templates/scheduler/scheduler-deployment.yaml

@@ -272,6 +272,10 @@ spec:
           {{- if .Values.scheduler.logGroomerSidecar.retentionDays }}
             - name: AIRFLOW__LOG_RETENTION_DAYS
               value: "{{ .Values.scheduler.logGroomerSidecar.retentionDays }}"
+          {{- end }}
+          {{- if .Values.scheduler.logGroomerSidecar.frequencyMinutes }}
+            - name: AIRFLOW__LOG_CLEANUP_FREQUENCY_MINUTES
+              value: "{{ .Values.scheduler.logGroomerSidecar.frequencyMinutes }}"
           {{- end }}
             - name: AIRFLOW_HOME
               value: "{{ .Values.airflowHome }}"

chart/templates/triggerer/triggerer-deployment.yaml

@@ -244,6 +244,10 @@ spec:
           {{- if .Values.triggerer.logGroomerSidecar.retentionDays }}
             - name: AIRFLOW__LOG_RETENTION_DAYS
               value: "{{ .Values.triggerer.logGroomerSidecar.retentionDays }}"
+          {{- end }}
+          {{- if .Values.triggerer.logGroomerSidecar.frequencyMinutes }}
+            - name: AIRFLOW__LOG_CLEANUP_FREQUENCY_MINUTES
+              value: "{{ .Values.triggerer.logGroomerSidecar.frequencyMinutes }}"
           {{- end }}
             - name: AIRFLOW_HOME
               value: "{{ .Values.airflowHome }}"

chart/templates/workers/worker-deployment.yaml

@@ -329,6 +329,10 @@ spec:
           {{- if .Values.workers.logGroomerSidecar.retentionDays }}
             - name: AIRFLOW__LOG_RETENTION_DAYS
               value: "{{ .Values.workers.logGroomerSidecar.retentionDays }}"
+          {{- end }}
+          {{- if .Values.workers.logGroomerSidecar.frequencyMinutes }}
+            - name: AIRFLOW__LOG_CLEANUP_FREQUENCY_MINUTES
+              value: "{{ .Values.workers.logGroomerSidecar.frequencyMinutes }}"
           {{- end }}
             - name: AIRFLOW_HOME
               value: "{{ .Values.airflowHome }}"

chart/values.schema.json

@@ -12205,6 +12205,11 @@
                     "type": "integer",
                     "default": 15
                 },
+                "frequencyMinutes": {
+                    "description": "Number of minutes between attempts to groom the Airflow logs in log groomer sidecar.",
+                    "type": "integer",
+                    "default": 15
+                },
                 "env": {
                     "description": "Add additional env vars to log groomer sidecar container (templated).",
                     "items": {

chart/values.yaml

@@ -824,6 +824,8 @@ workers:
     args: ["bash", "/clean-logs"]
     # Number of days to retain logs
     retentionDays: 15
+    # frequency to attempt to groom logs, in minutes
+    frequencyMinutes: 15
     resources: {}
     #  limits:
     #   cpu: 100m
@@ -1026,6 +1028,8 @@ scheduler:
     args: ["bash", "/clean-logs"]
     # Number of days to retain logs
     retentionDays: 15
+    # frequency to attempt to groom logs, in minutes
+    frequencyMinutes: 15
     resources: {}
     #  limits:
     #   cpu: 100m
@@ -1739,6 +1743,8 @@ triggerer:
     args: ["bash", "/clean-logs"]
     # Number of days to retain logs
     retentionDays: 15
+    # frequency to attempt to groom logs, in minutes
+    frequencyMinutes: 15
     resources: {}
     #  limits:
     #   cpu: 100m
@@ -1925,6 +1931,8 @@ dagProcessor:
     args: ["bash", "/clean-logs"]
     # Number of days to retain logs
     retentionDays: 15
+    # frequency to attempt to groom logs, in minutes
+    frequencyMinutes: 15
     resources: {}
     #  limits:
     #   cpu: 100m

scripts/docker/clean-logs.sh

@@ -21,10 +21,11 @@ set -euo pipefail
 
 readonly DIRECTORY="${AIRFLOW_HOME:-/usr/local/airflow}"
 readonly RETENTION="${AIRFLOW__LOG_RETENTION_DAYS:-15}"
+readonly FREQUENCY="${AIRFLOW__LOG_CLEANUP_FREQUENCY_MINUTES:-15}"
 
 trap "exit" INT TERM
 
-readonly EVERY=$((15*60))
+readonly EVERY=$((FREQUENCY*60))
 
 echo "Cleaning logs every $EVERY seconds"
 

tests/charts/log_groomer.py

@@ -182,14 +182,42 @@ def test_log_groomer_retention_days_overrides(self, retention_days, retention_re
 
         if retention_result:
             assert (
-                jmespath.search("spec.template.spec.containers[1].env[0].name", docs[0])
-                == "AIRFLOW__LOG_RETENTION_DAYS"
+                jmespath.search(
+                    "spec.template.spec.containers[1].env[?name=='AIRFLOW__LOG_RETENTION_DAYS'].value | [0]",
+                    docs[0],
+                )
+                == retention_result
             )
-            assert retention_result == jmespath.search(
-                "spec.template.spec.containers[1].env[0].value", docs[0]
+        else:
+            assert len(jmespath.search("spec.template.spec.containers[1].env", docs[0])) == 2
+
+    @pytest.mark.parametrize("frequency_minutes, frequency_result", [(None, None), (20, "20")])
+    def test_log_groomer_frequency_minutes_overrides(self, frequency_minutes, frequency_result):
+        if self.obj_name == "dag-processor":
+            values = {
+                "dagProcessor": {
+                    "enabled": True,
+                    "logGroomerSidecar": {"frequencyMinutes": frequency_minutes},
+                }
+            }
+        else:
+            values = {f"{self.folder}": {"logGroomerSidecar": {"frequencyMinutes": frequency_minutes}}}
+
+        docs = render_chart(
+            values=values,
+            show_only=[f"templates/{self.folder}/{self.obj_name}-deployment.yaml"],
+        )
+
+        if frequency_result:
+            assert (
+                jmespath.search(
+                    "spec.template.spec.containers[1].env[?name=='AIRFLOW__LOG_CLEANUP_FREQUENCY_MINUTES'].value | [0]",
+                    docs[0],
+                )
+                == frequency_result
             )
         else:
-            assert len(jmespath.search("spec.template.spec.containers[1].env", docs[0])) == 1
+            assert len(jmespath.search("spec.template.spec.containers[1].env", docs[0])) == 2
 
     def test_log_groomer_resources(self):
         if self.obj_name == "dag-processor":
@@ -242,4 +270,7 @@ def test_log_groomer_has_airflow_home(self):
             values=values, show_only=[f"templates/{self.folder}/{self.obj_name}-deployment.yaml"]
         )
 
-        assert jmespath.search("spec.template.spec.containers[1].env[1].name", docs[0]) == "AIRFLOW_HOME"
+        assert (
+            jmespath.search("spec.template.spec.containers[1].env[?name=='AIRFLOW_HOME'].name | [0]", docs[0])
+            == "AIRFLOW_HOME"
+        )