Merge pull request #1318 from muelli/ttlclamp

processors: Add a simple TTL clamping processor

.changelog/2809d288040441ccb8e6633f514b09b0.md

@@ -0,0 +1,5 @@
+---
+type: minor
+---
+
+Add processor for clamping TTLs

octodns/processor/clamp.py

@@ -0,0 +1,73 @@
+from logging import getLogger
+
+from .base import BaseProcessor, ProcessorException
+
+
+class TTLArgumentException(ProcessorException):
+    pass
+
+
+class TtlClampProcessor(BaseProcessor):
+    """
+    Processor that clamps TTL values to a specified range.
+
+    Configuration:
+        min_ttl: Minimum TTL value (default: 300 seconds / 5 minutes)
+        max_ttl: Maximum TTL value (default: 86400 seconds / 24 hours)
+
+    Example config.yaml:
+        processors:
+          clamp:
+            class: octodns.processor.clamp.TtlClampProcessor
+            min_ttl: 300
+            max_ttl: 3600
+
+        zones:
+          example.com.:
+            sources:
+              - config
+            processors:
+              - clamp
+            targets:
+              - route53
+    """
+
+    def __init__(self, id, min_ttl=300, max_ttl=86400):
+        super().__init__(id)
+        self.log = getLogger(self.__class__.__name__)
+        if not min_ttl <= max_ttl:
+            raise TTLArgumentException(
+                f'Min TTL {min_ttl} is not lower than max TTL {max_ttl}'
+            )
+        self.min_ttl = min_ttl
+        self.max_ttl = max_ttl
+        self.log.info('__init__: min=%ds, max=%ds', self.min_ttl, self.max_ttl)
+
+    def process_source_zone(self, desired, sources):
+        """
+        Process records from source zone(s).
+
+        Args:
+            desired: Zone object containing the desired records
+            sources: List of source names
+
+        Returns:
+            The modified zone
+        """
+        self.log.debug('process_source_zone: desired=%s', desired.name)
+
+        for record in desired.records:
+            original_ttl = record.ttl
+            clamped_ttl = max(self.min_ttl, min(self.max_ttl, original_ttl))
+
+            if clamped_ttl != original_ttl:
+                self.log.info(
+                    'process_source_zone: clamping TTL for %s (%s) %s -> %s',
+                    record.fqdn,
+                    record._type,
+                    original_ttl,
+                    clamped_ttl,
+                )
+                record.ttl = clamped_ttl
+
+        return desired

tests/test_octodns_processor_clamp.py

@@ -0,0 +1,120 @@
+from unittest import TestCase
+
+from octodns.processor.clamp import TTLArgumentException, TtlClampProcessor
+from octodns.record.base import Record
+from octodns.zone import Zone
+
+
+class TestClampProcessor(TestCase):
+
+    def test_processor_min(self):
+        "Test the processor for clamping to the minimum"
+        min_ttl = 42
+        processor = TtlClampProcessor('test', min_ttl=min_ttl)
+
+        too_low_ttl = 23
+        self.assertLess(too_low_ttl, min_ttl)
+
+        zone = Zone('unit.tests.', [])
+        zone.add_record(
+            Record.new(
+                zone, '', {'type': 'TXT', 'ttl': too_low_ttl, 'value': 'foo'}
+            )
+        )
+
+        processed_zone = processor.process_source_zone(zone.copy(), None)
+        self.assertNotEqual(zone, processed_zone)
+
+        self.assertEqual(len(processed_zone.records), len(zone.records))
+        self.assertEqual(len(processed_zone.records), 1)
+        self.assertEqual(processed_zone.records.pop().ttl, min_ttl)
+
+    def test_processor_max(self):
+        "Test the processor for clamping to the maximum"
+        max_ttl = 4711
+        processor = TtlClampProcessor('test', max_ttl=max_ttl)
+
+        too_high_ttl = max_ttl + 1
+        self.assertLess(max_ttl, too_high_ttl)
+
+        zone = Zone('unit.tests.', [])
+        zone.add_record(
+            Record.new(
+                zone, '', {'type': 'TXT', 'ttl': too_high_ttl, 'value': 'foo'}
+            )
+        )
+
+        processed_zone = processor.process_source_zone(zone.copy(), None)
+        self.assertNotEqual(zone, processed_zone)
+
+        self.assertEqual(len(processed_zone.records), len(zone.records))
+        self.assertEqual(len(processed_zone.records), 1)
+        self.assertEqual(processed_zone.records.pop().ttl, max_ttl)
+
+    def test_processor_maxmin(self):
+        "Test the processor for unlogical arguments"
+        min_ttl = 42
+        max_ttl = 23
+        self.assertRaises(
+            TTLArgumentException,
+            TtlClampProcessor,
+            'test',
+            min_ttl=min_ttl,
+            max_ttl=max_ttl,
+        )
+
+    def test_processor_minmax(self):
+        "Test the processor for clamping both min and max values"
+        min_ttl = 42
+        max_ttl = 4711
+        processor = TtlClampProcessor('test', min_ttl=min_ttl, max_ttl=max_ttl)
+
+        too_low_ttl = min_ttl - 1
+        too_high_ttl = max_ttl + 1
+        self.assertLess(too_low_ttl, min_ttl)
+        self.assertLess(too_low_ttl, min_ttl)
+        self.assertLess(max_ttl, too_high_ttl)
+
+        zone = Zone('unit.tests.', [])
+        zone.add_record(
+            Record.new(
+                zone,
+                'high',
+                {'type': 'TXT', 'ttl': too_high_ttl, 'value': 'high'},
+            )
+        )
+        zone.add_record(
+            Record.new(
+                zone, 'low', {'type': 'TXT', 'ttl': too_low_ttl, 'value': 'low'}
+            )
+        )
+
+        processed_zone = processor.process_source_zone(zone.copy(), None)
+        self.assertNotEqual(zone, processed_zone)
+
+        processed_records = sorted(
+            list(processed_zone.records), key=lambda r: r.ttl
+        )
+        self.assertEqual(len(processed_records), 2)
+
+        self.assertEqual(processed_records[0].ttl, min_ttl)
+        self.assertEqual(processed_records[1].ttl, max_ttl)
+
+    def test_processor_noclamp(self):
+        "Test the processor for working with TTLs not requiring any clamping"
+        min_ttl = 23
+        max_ttl = 4711
+        processor = TtlClampProcessor('test', min_ttl=min_ttl, max_ttl=max_ttl)
+
+        ttl = 42
+
+        self.assertLess(min_ttl, ttl)
+        self.assertLess(ttl, max_ttl)
+
+        zone = Zone('unit.tests.', [])
+        zone.add_record(
+            Record.new(zone, '', {'type': 'TXT', 'ttl': ttl, 'value': 'foo'})
+        )
+
+        processed_zone = processor.process_source_zone(zone.copy(), None)
+        self.assertEqual(processed_zone.records.pop().ttl, ttl)