Merge pull request #1061 from octodns/meta-processor

Meta processor

CHANGELOG.md

@@ -14,7 +14,6 @@
 * YamlProvider now supports a `shared_filename` that can be used to add a set of
   common records across all zones using the provider. It can be used stand-alone
   or in combination with zone files and/or split configs to aid in DRYing up DNS
-  configs.
 * YamlProvider now supports an `!include` directive which enables shared
   snippets of config to be reused across many records, e.g. common dynamic rules
   across a set of services with service-specific pool values or a unified SFP
@@ -23,6 +22,9 @@
   ValidationError in 2.x
 * SpfDnsLookupProcessor is formally deprcated in favor of the version relocated
   into https://github.com/octodns/octodns-spf and will be removed in 2.x
+* MetaProcessor added to enable some useful/cool options for debugging/tracking
+  DNS changes. Specifically timestamps/uuid so you can track whether changes
+  that have been pushed to providers have propogated/transferred correctly.
 
 #### Stuff
 
@@ -33,6 +35,9 @@
 * Add --all option to octodns-validate to enable showing all record validation
   errors (as warnings) rather than exiting on the first. Exit code is non-zero
   when there are any validation errors.
+* New `post_processors` manager configuration parameter to add global processors
+  that run AFTER zone-specific processors. This should allow more complete
+  control over when processors are run.
 
 ## v1.0.0 - 2023-07-30 - The One
 

README.md

@@ -331,6 +331,7 @@ Similar to providers, but can only serve to populate records into a zone, cannot
 | [AcmeMangingProcessor](/octodns/processor/acme.py) | Useful when processes external to octoDNS are managing acme challenge DNS records, e.g. LetsEncrypt |
 | [AutoArpa](/octodns/processor/arpa.py) | See [Automatic PTR generation](#automatic-ptr-generation) below |
 | [IgnoreRootNsFilter](/octodns/processor/filter.py) | Filter that INGORES root/APEX NS records and prevents octoDNS from trying to manage them (where supported.) |
+| [MetaProcessor](/octodns/processor/meta.py) | Adds a special meta record with timing, UUID, providers, and/or version to aid in debugging and monitoring. |
 | [NameAllowlistFilter](/octodns/processor/filter.py) | Filter that ONLY manages records that match specified naming patterns, all others will be ignored |
 | [NameRejectlistFilter](/octodns/processor/filter.py) | Filter that INGORES records that match specified naming patterns, all others will be managed |
 | [OwnershipProcessor](/octodns/processor/ownership.py) | Processor that implements ownership in octoDNS so that it can manage only the records in a zone in sources and will ignore all others. |

octodns/manager.py

@@ -14,10 +14,10 @@ from sys import stdout
 from . import __VERSION__
 from .idna import IdnaDict, idna_decode, idna_encode
 from .processor.arpa import AutoArpa
+from .processor.meta import MetaProcessor
 from .provider.base import BaseProvider
 from .provider.plan import Plan
 from .provider.yaml import SplitYamlProvider, YamlProvider
-from .record import Record
 from .yaml import safe_load
 from .zone import Zone
 
@@ -114,6 +114,11 @@ class Manager(object):
         self.global_processors = manager_config.get('processors', [])
         self.log.info('__init__: global_processors=%s', self.global_processors)
 
+        self.global_post_processors = manager_config.get('post_processors', [])
+        self.log.info(
+            '__init__: global_post_processors=%s', self.global_post_processors
+        )
+
         providers_config = self.config['providers']
         self.providers = self._config_providers(providers_config)
 
@@ -122,13 +127,28 @@ class Manager(object):
 
         if self.auto_arpa:
             self.log.info(
-                '__init__: adding auto-arpa to processors and providers, appending it to global_processors list'
+                '__init__: adding auto-arpa to processors and providers, prepending it to global_post_processors list'
             )
             kwargs = self.auto_arpa if isinstance(auto_arpa, dict) else {}
             auto_arpa = AutoArpa('auto-arpa', **kwargs)
             self.providers[auto_arpa.name] = auto_arpa
             self.processors[auto_arpa.name] = auto_arpa
-            self.global_processors.append(auto_arpa.name)
+            self.global_post_processors = [
+                auto_arpa.name
+            ] + self.global_post_processors
+
+        if self.include_meta:
+            self.log.info(
+                '__init__: adding meta to processors and providers, appending it to global_post_processors list'
+            )
+            meta = MetaProcessor(
+                'meta',
+                record_name='octodns-meta',
+                include_time=False,
+                include_provider=True,
+            )
+            self.processors[meta.id] = meta
+            self.global_post_processors.append(meta.id)
 
         plan_outputs_config = manager_config.get(
             'plan_outputs',
@@ -433,17 +453,6 @@ class Manager(object):
         plans = []
 
         for target in targets:
-            if self.include_meta:
-                meta = Record.new(
-                    zone,
-                    'octodns-meta',
-                    {
-                        'type': 'TXT',
-                        'ttl': 60,
-                        'value': f'provider={target.id}',
-                    },
-                )
-                zone.add_record(meta, replace=True)
             try:
                 plan = target.plan(zone, processors=processors)
             except TypeError as e:
@@ -634,7 +643,11 @@ class Manager(object):
 
             try:
                 collected = []
-                for processor in self.global_processors + processors:
+                for processor in (
+                    self.global_processors
+                    + processors
+                    + self.global_post_processors
+                ):
                     collected.append(self.processors[processor])
                 processors = collected
             except KeyError:

octodns/processor/base.py

@@ -9,7 +9,8 @@ class ProcessorException(Exception):
 
 class BaseProcessor(object):
     def __init__(self, name):
-        self.name = name
+        # TODO: name is DEPRECATED, remove in 2.0
+        self.id = self.name = name
 
     def process_source_zone(self, desired, sources):
         '''

octodns/processor/meta.py

@@ -0,0 +1,145 @@
+#
+#
+#
+
+from datetime import datetime
+from logging import getLogger
+from uuid import uuid4
+
+from .. import __VERSION__
+from ..record import Record
+from .base import BaseProcessor
+
+
+def _keys(values):
+    return set(v.split('=', 1)[0] for v in values)
+
+
+class MetaProcessor(BaseProcessor):
+    '''
+    Add a special metadata record with timestamps, UUIDs, versions, and/or
+    provider name. Will only be updated when there are other changes being made.
+    A useful tool to aid in debugging and monitoring of DNS infrastructure.
+
+    Timestamps or UUIDs can be useful in checking whether changes are
+    propagating, either from a provider's backend to their servers or via AXFRs.
+
+    Provider can be utilized to determine which DNS system responded to a query
+    when things are operating in dual authority or split horizon setups.
+
+    Creates a TXT record with the name configured with values based on processor
+    settings. Values are in the form `key=<value>`, e.g.
+    `time=2023-09-10T05:49:04.246953`
+
+    processors:
+      meta:
+        class: octodns.processor.meta.MetaProcessor
+        # The name to use for the meta record.
+        # (default: meta)
+        record_name: meta
+        # Include a timestamp with a UTC value indicating the timeframe when the
+        # last change was made.
+        # (default: true)
+        include_time: true
+        # Include a UUID that can be utilized to uniquely identify the run
+        # pushing data
+        # (default: false)
+        include_uuid: false
+        # Include the provider id for the target where data is being pushed
+        # (default: false)
+        include_provider: false
+        # Include the octoDNS version being used
+        # (default: false)
+        include_version: false
+    '''
+
+    @classmethod
+    def now(cls):
+        return datetime.utcnow().isoformat()
+
+    @classmethod
+    def uuid(cls):
+        return str(uuid4())
+
+    def __init__(
+        self,
+        id,
+        record_name='meta',
+        include_time=True,
+        include_uuid=False,
+        include_version=False,
+        include_provider=False,
+        ttl=60,
+    ):
+        self.log = getLogger(f'MetaSource[{id}]')
+        super().__init__(id)
+        self.log.info(
+            '__init__: record_name=%s, include_time=%s, include_uuid=%s, include_version=%s, include_provider=%s, ttl=%d',
+            record_name,
+            include_time,
+            include_uuid,
+            include_version,
+            include_provider,
+            ttl,
+        )
+        self.record_name = record_name
+        values = []
+        if include_time:
+            time = self.now()
+            values.append(f'time={time}')
+        if include_uuid:
+            uuid = self.uuid() if include_uuid else None
+            values.append(f'uuid={uuid}')
+        if include_version:
+            values.append(f'octodns-version={__VERSION__}')
+        self.include_provider = include_provider
+        values.sort()
+        self.values = values
+        self.ttl = ttl
+
+    def process_source_zone(self, desired, sources):
+        meta = Record.new(
+            desired,
+            self.record_name,
+            {'ttl': self.ttl, 'type': 'TXT', 'values': self.values},
+            # we may be passing in empty values here to be filled out later in
+            # process_target_zone
+            lenient=True,
+        )
+        desired.add_record(meta)
+        return desired
+
+    def process_target_zone(self, existing, target):
+        if self.include_provider:
+            # look for the meta record
+            for record in sorted(existing.records):
+                if record.name == self.record_name and record._type == 'TXT':
+                    # we've found it, make a copy we can modify
+                    record = record.copy()
+                    record.values = record.values + [f'provider={target.id}']
+                    record.values.sort()
+                    existing.add_record(record, replace=True)
+                    break
+
+        return existing
+
+    def _up_to_date(self, change):
+        # existing state, if there is one
+        existing = getattr(change, 'existing', None)
+        return existing is not None and _keys(existing.values) == _keys(
+            self.values
+        )
+
+    def process_plan(self, plan, sources, target):
+        if (
+            plan
+            and len(plan.changes) == 1
+            and self._up_to_date(plan.changes[0])
+        ):
+            # the only change is the meta record, and it's not meaningfully
+            # changing so we don't actually want to make the change
+            return None
+
+        # There's more than one thing changing so meta should update and/or meta
+        # is meaningfully changing or being created...
+        return plan

tests/test_octodns_processor_meta.py

@@ -0,0 +1,202 @@
+#
+#
+#
+
+from unittest import TestCase
+from unittest.mock import patch
+
+from octodns import __VERSION__
+from octodns.processor.meta import MetaProcessor
+from octodns.provider.plan import Plan
+from octodns.record import Create, Record, Update
+from octodns.zone import Zone
+
+
+class TestMetaProcessor(TestCase):
+    zone = Zone('unit.tests.', [])
+
+    meta_needs_update = Record.new(
+        zone,
+        'meta',
+        {
+            'type': 'TXT',
+            'ttl': 60,
+            # will always need updating
+            'values': ['uuid'],
+        },
+    )
+
+    meta_up_to_date = Record.new(
+        zone,
+        'meta',
+        {
+            'type': 'TXT',
+            'ttl': 60,
+            # only has time, value should be ignored
+            'values': ['time=xxx'],
+        },
+    )
+
+    not_meta = Record.new(
+        zone,
+        'its-not-meta',
+        {
+            'type': 'TXT',
+            'ttl': 60,
+            # has time, but name is wrong so won't matter
+            'values': ['time=xyz'],
+        },
+    )
+
+    @patch('octodns.processor.meta.MetaProcessor.now')
+    @patch('octodns.processor.meta.MetaProcessor.uuid')
+    def test_args_and_values(self, uuid_mock, now_mock):
+        # defaults, just time
+        uuid_mock.side_effect = [Exception('not used')]
+        now_mock.side_effect = ['the-time']
+        proc = MetaProcessor('test')
+        self.assertEqual(['time=the-time'], proc.values)
+
+        # just uuid
+        uuid_mock.side_effect = ['abcdef-1234567890']
+        now_mock.side_effect = [Exception('not used')]
+        proc = MetaProcessor('test', include_time=False, include_uuid=True)
+        self.assertEqual(['uuid=abcdef-1234567890'], proc.values)
+
+        # just version
+        uuid_mock.side_effect = [Exception('not used')]
+        now_mock.side_effect = [Exception('not used')]
+        proc = MetaProcessor('test', include_time=False, include_version=True)
+        self.assertEqual([f'octodns-version={__VERSION__}'], proc.values)
+
+        # just provider
+        proc = MetaProcessor('test', include_time=False, include_provider=True)
+        self.assertTrue(proc.include_provider)
+        self.assertFalse(proc.values)
+
+        # everything
+        uuid_mock.side_effect = ['abcdef-1234567890']
+        now_mock.side_effect = ['the-time']
+        proc = MetaProcessor(
+            'test',
+            include_time=True,
+            include_uuid=True,
+            include_version=True,
+            include_provider=True,
+        )
+        self.assertEqual(
+            [
+                f'octodns-version={__VERSION__}',
+                'time=the-time',
+                'uuid=abcdef-1234567890',
+            ],
+            proc.values,
+        )
+        self.assertTrue(proc.include_provider)
+
+    def test_uuid(self):
+        proc = MetaProcessor('test', include_time=False, include_uuid=True)
+        self.assertEqual(1, len(proc.values))
+        self.assertTrue(proc.values[0].startswith('uuid'))
+        # uuid's have 4 -
+        self.assertEqual(4, proc.values[0].count('-'))
+
+    def test_up_to_date(self):
+        proc = MetaProcessor('test')
+
+        # Creates always need to happen
+        self.assertFalse(proc._up_to_date(Create(self.meta_needs_update)))
+        self.assertFalse(proc._up_to_date(Create(self.meta_up_to_date)))
+
+        # Updates depend on the contents
+        self.assertFalse(proc._up_to_date(Update(self.meta_needs_update, None)))
+        self.assertTrue(proc._up_to_date(Update(self.meta_up_to_date, None)))
+
+    @patch('octodns.processor.meta.MetaProcessor.now')
+    def test_process_source_zone(self, now_mock):
+        now_mock.side_effect = ['the-time']
+        proc = MetaProcessor('test')
+
+        # meta record was added
+        desired = self.zone.copy()
+        processed = proc.process_source_zone(desired, None)
+        record = next(iter(processed.records))
+        self.assertEqual(self.meta_up_to_date, record)
+        self.assertEqual(['time=the-time'], record.values)
+
+    def test_process_target_zone(self):
+        proc = MetaProcessor('test')
+
+        # with defaults, not enabled
+        zone = self.zone.copy()
+        processed = proc.process_target_zone(zone, None)
+        self.assertFalse(processed.records)
+
+        # enable provider
+        proc = MetaProcessor('test', include_provider=True)
+
+        class DummyTarget:
+            id = 'dummy'
+
+        # enabled provider, no meta record, shouldn't happen, but also shouldn't
+        # blow up
+        processed = proc.process_target_zone(zone, DummyTarget())
+        self.assertFalse(processed.records)
+
+        # enabled provider, should now look for and update the provider value,
+        # - only record so nothing to skip over
+        # - time value in there to be skipped over
+        proc = MetaProcessor('test', include_provider=True)
+        zone = self.zone.copy()
+        meta = self.meta_up_to_date.copy()
+        zone.add_record(meta)
+        processed = proc.process_target_zone(zone, DummyTarget())
+        record = next(iter(processed.records))
+        self.assertEqual(['provider=dummy', 'time=xxx'], record.values)
+
+        # add another unrelated record that needs to be skipped
+        proc = MetaProcessor('test', include_provider=True)
+        zone = self.zone.copy()
+        meta = self.meta_up_to_date.copy()
+        zone.add_record(meta)
+        zone.add_record(self.not_meta)
+        processed = proc.process_target_zone(zone, DummyTarget())
+        self.assertEqual(2, len(processed.records))
+        record = [r for r in processed.records if r.name == proc.record_name][0]
+        self.assertEqual(['provider=dummy', 'time=xxx'], record.values)
+
+    def test_process_plan(self):
+        proc = MetaProcessor('test')
+
+        # no plan, shouldn't happen, but we shouldn't blow up
+        self.assertFalse(proc.process_plan(None, None, None))
+
+        # plan with just an up to date meta record, should kill off the plan
+        plan = Plan(
+            None,
+            None,
+            [Update(self.meta_up_to_date, self.meta_needs_update)],
+            True,
+        )
+        self.assertFalse(proc.process_plan(plan, None, None))
+
+        # plan with an out of date meta record, should leave the plan alone
+        plan = Plan(
+            None,
+            None,
+            [Update(self.meta_needs_update, self.meta_up_to_date)],
+            True,
+        )
+        self.assertEqual(plan, proc.process_plan(plan, None, None))
+
+        # plan with other changes preserved even if meta was somehow up to date
+        plan = Plan(
+            None,
+            None,
+            [
+                Update(self.meta_up_to_date, self.meta_needs_update),
+                Create(self.not_meta),
+            ],
+            True,
+        )
+        self.assertEqual(plan, proc.process_plan(plan, None, None))