diff --git a/octodns/record.py b/octodns/record.py
index f9812a6..514c2e6 100644
--- a/octodns/record.py
+++ b/octodns/record.py
@@ -128,6 +128,8 @@ class Record(object):
 
         octodns = data.get('octodns', {})
         self.ignored = octodns.get('ignored', False)
+        self.excluded = octodns.get('excluded', [])
+        self.included = octodns.get('included', [])
 
     def _data(self):
         return {'ttl': self.ttl}
diff --git a/octodns/zone.py b/octodns/zone.py
index 74e5d9e..bbc38d0 100644
--- a/octodns/zone.py
+++ b/octodns/zone.py
@@ -110,10 +110,29 @@ class Zone(object):
         for record in filter(_is_eligible, self.records):
             if record.ignored:
                 continue
+            elif len(record.included) > 0 and \
+                    target.id not in record.included:
+                self.log.debug('changes:  skipping record=%s %s - %s not'
+                               ' included ', record.fqdn, record._type,
+                               target.id)
+                continue
+            elif target.id in record.excluded:
+                self.log.debug('changes:  skipping record=%s %s - %s '
+                               'excluded ', record.fqdn, record._type,
+                               target.id)
+                continue
             try:
                 desired_record = desired_records[record]
                 if desired_record.ignored:
                     continue
+                elif len(desired_record.included) > 0 and \
+                        target.id not in desired_record.included:
+                    self.log.debug('changes:  skipping record=%s %s - %s'
+                                   'not included ', record.fqdn, record._type,
+                                   target.id)
+                    continue
+                elif target.id in desired_record.excluded:
+                    continue
             except KeyError:
                 if not target.supports(record):
                     self.log.debug('changes:  skipping record=%s %s - %s does '
@@ -141,6 +160,18 @@ class Zone(object):
         for record in filter(_is_eligible, desired.records - self.records):
             if record.ignored:
                 continue
+            elif len(record.included) > 0 and \
+                    target.id not in record.included:
+                self.log.debug('changes:  skipping record=%s %s - %s not'
+                               ' included ', record.fqdn, record._type,
+                               target.id)
+                continue
+            elif target.id in record.excluded:
+                self.log.debug('changes:  skipping record=%s %s - %s '
+                               'excluded ', record.fqdn, record._type,
+                               target.id)
+                continue
+
             if not target.supports(record):
                 self.log.debug('changes:  skipping record=%s %s - %s does not '
                                'support it', record.fqdn, record._type,
diff --git a/tests/config/unit.tests.yaml b/tests/config/unit.tests.yaml
index 5241406..1da2465 100644
--- a/tests/config/unit.tests.yaml
+++ b/tests/config/unit.tests.yaml
@@ -56,11 +56,23 @@ cname:
   ttl: 300
   type: CNAME
   value: unit.tests.
+excluded:
+  octodns:
+    excluded:
+    - test
+  type: CNAME
+  value: unit.tests.
 ignored:
   octodns:
     ignored: true
   type: A
   value: 9.9.9.9
+included:
+  octodns:
+    included:
+    - test
+  type: CNAME
+  value: unit.tests.
 mx:
   ttl: 300
   type: MX
diff --git a/tests/fixtures/cloudflare-dns_records-page-2.json b/tests/fixtures/cloudflare-dns_records-page-2.json
index 195d6de..150951b 100644
--- a/tests/fixtures/cloudflare-dns_records-page-2.json
+++ b/tests/fixtures/cloudflare-dns_records-page-2.json
@@ -139,14 +139,31 @@
       "meta": {
         "auto_added": false
       }
+    },
+    {
+      "id": "fc12ab34cd5611334422ab3322997656",
+      "type": "CNAME",
+      "name": "included.unit.tests",
+      "content": "unit.tests",
+      "proxiable": true,
+      "proxied": false,
+      "ttl": 3600,
+      "locked": false,
+      "zone_id": "ff12ab34cd5611334422ab3322997650",
+      "zone_name": "unit.tests",
+      "modified_on": "2017-03-11T18:01:43.940682Z",
+      "created_on": "2017-03-11T18:01:43.940682Z",
+      "meta": {
+        "auto_added": false
+      }
     }
   ],
   "result_info": {
     "page": 2,
     "per_page": 10,
     "total_pages": 2,
-    "count": 8,
-    "total_count": 19
+    "count": 9,
+    "total_count": 20
   },
   "success": true,
   "errors": [],
diff --git a/tests/fixtures/dnsimple-page-2.json b/tests/fixtures/dnsimple-page-2.json
index 40aaa48..a42c393 100644
--- a/tests/fixtures/dnsimple-page-2.json
+++ b/tests/fixtures/dnsimple-page-2.json
@@ -175,12 +175,28 @@
       "system_record": false,
       "created_at": "2017-03-09T15:55:09Z",
       "updated_at": "2017-03-09T15:55:09Z"
+    },
+    {
+      "id": 12188805,
+      "zone_id": "unit.tests",
+      "parent_id": null,
+      "name": "included",
+      "content": "unit.tests",
+      "ttl": 3600,
+      "priority": null,
+      "type": "CNAME",
+      "regions": [
+        "global"
+      ],
+      "system_record": false,
+      "created_at": "2017-03-09T15:55:09Z",
+      "updated_at": "2017-03-09T15:55:09Z"
     }
   ],
   "pagination": {
     "current_page": 2,
     "per_page": 20,
-    "total_entries": 30,
+    "total_entries": 32,
     "total_pages": 2
   }
 }
diff --git a/tests/fixtures/powerdns-full-data.json b/tests/fixtures/powerdns-full-data.json
index b8f8bf3..3d445d4 100644
--- a/tests/fixtures/powerdns-full-data.json
+++ b/tests/fixtures/powerdns-full-data.json
@@ -242,6 +242,18 @@
             ],
             "ttl": 3600,
             "type": "CAA"
+        },
+	{
+            "comments": [],
+            "name": "included.unit.tests.",
+            "records": [
+                {
+                    "content": "unit.tests.",
+                    "disabled": false
+                }
+            ],
+            "ttl": 3600,
+            "type": "CNAME"
         }
     ],
     "serial": 2017012803,
diff --git a/tests/helpers.py b/tests/helpers.py
index adac81d..632f258 100644
--- a/tests/helpers.py
+++ b/tests/helpers.py
@@ -18,6 +18,7 @@ class SimpleSource(object):
 class SimpleProvider(object):
     SUPPORTS_GEO = False
     SUPPORTS = set(('A',))
+    id = 'test'
 
     def __init__(self, id='test'):
         pass
@@ -34,6 +35,7 @@ class SimpleProvider(object):
 
 class GeoProvider(object):
     SUPPORTS_GEO = True
+    id = 'test'
 
     def __init__(self, id='test'):
         pass
diff --git a/tests/test_octodns_manager.py b/tests/test_octodns_manager.py
index a5f2022..36b1f4c 100644
--- a/tests/test_octodns_manager.py
+++ b/tests/test_octodns_manager.py
@@ -102,12 +102,12 @@ class TestManager(TestCase):
             environ['YAML_TMP_DIR'] = tmpdir.dirname
             tc = Manager(get_config_filename('simple.yaml')) \
                 .sync(dry_run=False)
-            self.assertEquals(19, tc)
+            self.assertEquals(20, tc)
 
             # try with just one of the zones
             tc = Manager(get_config_filename('simple.yaml')) \
                 .sync(dry_run=False, eligible_zones=['unit.tests.'])
-            self.assertEquals(13, tc)
+            self.assertEquals(14, tc)
 
             # the subzone, with 2 targets
             tc = Manager(get_config_filename('simple.yaml')) \
@@ -122,18 +122,18 @@ class TestManager(TestCase):
             # Again with force
             tc = Manager(get_config_filename('simple.yaml')) \
                 .sync(dry_run=False, force=True)
-            self.assertEquals(19, tc)
+            self.assertEquals(20, tc)
 
             # Again with max_workers = 1
             tc = Manager(get_config_filename('simple.yaml'), max_workers=1) \
                 .sync(dry_run=False, force=True)
-            self.assertEquals(19, tc)
+            self.assertEquals(20, tc)
 
             # Include meta
             tc = Manager(get_config_filename('simple.yaml'), max_workers=1,
                          include_meta=True) \
                 .sync(dry_run=False, force=True)
-            self.assertEquals(23, tc)
+            self.assertEquals(24, tc)
 
     def test_eligible_targets(self):
         with TemporaryDirectory() as tmpdir:
@@ -159,13 +159,13 @@ class TestManager(TestCase):
                 fh.write('---\n{}')
 
             changes = manager.compare(['in'], ['dump'], 'unit.tests.')
-            self.assertEquals(13, len(changes))
+            self.assertEquals(14, len(changes))
 
             # Compound sources with varying support
             changes = manager.compare(['in', 'nosshfp'],
                                       ['dump'],
                                       'unit.tests.')
-            self.assertEquals(12, len(changes))
+            self.assertEquals(13, len(changes))
 
             with self.assertRaises(Exception) as ctx:
                 manager.compare(['nope'], ['dump'], 'unit.tests.')
diff --git a/tests/test_octodns_provider_base.py b/tests/test_octodns_provider_base.py
index fde1396..1bf3fd7 100644
--- a/tests/test_octodns_provider_base.py
+++ b/tests/test_octodns_provider_base.py
@@ -17,6 +17,7 @@ class HelperProvider(BaseProvider):
     log = getLogger('HelperProvider')
 
     SUPPORTS = set(('A',))
+    id = 'test'
 
     def __init__(self, extra_changes, apply_disabled=False,
                  include_change_callback=None):
diff --git a/tests/test_octodns_provider_cloudflare.py b/tests/test_octodns_provider_cloudflare.py
index 04a46e0..ef8a51c 100644
--- a/tests/test_octodns_provider_cloudflare.py
+++ b/tests/test_octodns_provider_cloudflare.py
@@ -118,7 +118,7 @@ class TestCloudflareProvider(TestCase):
 
             zone = Zone('unit.tests.', [])
             provider.populate(zone)
-            self.assertEquals(10, len(zone.records))
+            self.assertEquals(11, len(zone.records))
 
             changes = self.expected.changes(zone, provider)
             self.assertEquals(0, len(changes))
@@ -126,7 +126,7 @@ class TestCloudflareProvider(TestCase):
         # re-populating the same zone/records comes out of cache, no calls
         again = Zone('unit.tests.', [])
         provider.populate(again)
-        self.assertEquals(10, len(again.records))
+        self.assertEquals(11, len(again.records))
 
     def test_apply(self):
         provider = CloudflareProvider('test', 'email', 'token')
@@ -140,12 +140,12 @@ class TestCloudflareProvider(TestCase):
                     'id': 42,
                 }
             },  # zone create
-        ] + [None] * 17  # individual record creates
+        ] + [None] * 18  # individual record creates
 
         # non-existant zone, create everything
         plan = provider.plan(self.expected)
-        self.assertEquals(10, len(plan.changes))
-        self.assertEquals(10, provider.apply(plan))
+        self.assertEquals(11, len(plan.changes))
+        self.assertEquals(11, provider.apply(plan))
 
         provider._request.assert_has_calls([
             # created the domain
@@ -170,7 +170,7 @@ class TestCloudflareProvider(TestCase):
             }),
         ], True)
         # expected number of total calls
-        self.assertEquals(19, provider._request.call_count)
+        self.assertEquals(20, provider._request.call_count)
 
         provider._request.reset_mock()
 
diff --git a/tests/test_octodns_provider_dnsimple.py b/tests/test_octodns_provider_dnsimple.py
index b44d6ee..60dacf1 100644
--- a/tests/test_octodns_provider_dnsimple.py
+++ b/tests/test_octodns_provider_dnsimple.py
@@ -78,14 +78,14 @@ class TestDnsimpleProvider(TestCase):
 
             zone = Zone('unit.tests.', [])
             provider.populate(zone)
-            self.assertEquals(15, len(zone.records))
+            self.assertEquals(16, len(zone.records))
             changes = self.expected.changes(zone, provider)
             self.assertEquals(0, len(changes))
 
         # 2nd populate makes no network calls/all from cache
         again = Zone('unit.tests.', [])
         provider.populate(again)
-        self.assertEquals(15, len(again.records))
+        self.assertEquals(16, len(again.records))
 
         # bust the cache
         del provider._zone_records[zone.name]
@@ -129,8 +129,8 @@ class TestDnsimpleProvider(TestCase):
         ]
         plan = provider.plan(self.expected)
 
-        # No root NS, no ignored
-        n = len(self.expected.records) - 2
+        # No root NS, no ignored, no excluded
+        n = len(self.expected.records) - 3
         self.assertEquals(n, len(plan.changes))
         self.assertEquals(n, provider.apply(plan))
 
@@ -147,7 +147,7 @@ class TestDnsimpleProvider(TestCase):
             }),
         ])
         # expected number of total calls
-        self.assertEquals(27, provider._client._request.call_count)
+        self.assertEquals(28, provider._client._request.call_count)
 
         provider._client._request.reset_mock()
 
diff --git a/tests/test_octodns_provider_powerdns.py b/tests/test_octodns_provider_powerdns.py
index b6e02ff..22ccdd6 100644
--- a/tests/test_octodns_provider_powerdns.py
+++ b/tests/test_octodns_provider_powerdns.py
@@ -78,8 +78,8 @@ class TestPowerDnsProvider(TestCase):
         expected = Zone('unit.tests.', [])
         source = YamlProvider('test', join(dirname(__file__), 'config'))
         source.populate(expected)
-        expected_n = len(expected.records) - 1
-        self.assertEquals(15, expected_n)
+        expected_n = len(expected.records) - 2
+        self.assertEquals(16, expected_n)
 
         # No diffs == no changes
         with requests_mock() as mock:
@@ -87,7 +87,7 @@ class TestPowerDnsProvider(TestCase):
 
             zone = Zone('unit.tests.', [])
             provider.populate(zone)
-            self.assertEquals(15, len(zone.records))
+            self.assertEquals(16, len(zone.records))
             changes = expected.changes(zone, provider)
             self.assertEquals(0, len(changes))
 
@@ -167,7 +167,7 @@ class TestPowerDnsProvider(TestCase):
         expected = Zone('unit.tests.', [])
         source = YamlProvider('test', join(dirname(__file__), 'config'))
         source.populate(expected)
-        self.assertEquals(16, len(expected.records))
+        self.assertEquals(18, len(expected.records))
 
         # A small change to a single record
         with requests_mock() as mock:
diff --git a/tests/test_octodns_provider_yaml.py b/tests/test_octodns_provider_yaml.py
index 36cd8d6..a199355 100644
--- a/tests/test_octodns_provider_yaml.py
+++ b/tests/test_octodns_provider_yaml.py
@@ -30,7 +30,7 @@ class TestYamlProvider(TestCase):
 
         # without it we see everything
         source.populate(zone)
-        self.assertEquals(16, len(zone.records))
+        self.assertEquals(18, len(zone.records))
 
         # Assumption here is that a clean round-trip means that everything
         # worked as expected, data that went in came back out and could be
@@ -49,12 +49,12 @@ class TestYamlProvider(TestCase):
 
             # We add everything
             plan = target.plan(zone)
-            self.assertEquals(13, len(filter(lambda c: isinstance(c, Create),
+            self.assertEquals(14, len(filter(lambda c: isinstance(c, Create),
                                              plan.changes)))
             self.assertFalse(isfile(yaml_file))
 
             # Now actually do it
-            self.assertEquals(13, target.apply(plan))
+            self.assertEquals(14, target.apply(plan))
             self.assertTrue(isfile(yaml_file))
 
             # There should be no changes after the round trip
@@ -64,7 +64,7 @@ class TestYamlProvider(TestCase):
 
             # A 2nd sync should still create everything
             plan = target.plan(zone)
-            self.assertEquals(13, len(filter(lambda c: isinstance(c, Create),
+            self.assertEquals(14, len(filter(lambda c: isinstance(c, Create),
                                              plan.changes)))
 
             with open(yaml_file) as fh:
diff --git a/tests/test_octodns_zone.py b/tests/test_octodns_zone.py
index 8d75100..94faef3 100644
--- a/tests/test_octodns_zone.py
+++ b/tests/test_octodns_zone.py
@@ -236,3 +236,102 @@ class TestZone(TestCase):
         zone.add_record(cname)
         with self.assertRaises(InvalidNodeException):
             zone.add_record(a)
+
+    def test_excluded_records(self):
+        zone_normal = Zone('unit.tests.', [])
+        zone_excluded = Zone('unit.tests.', [])
+        zone_missing = Zone('unit.tests.', [])
+
+        normal = Record.new(zone_normal, 'www', {
+            'ttl': 60,
+            'type': 'A',
+            'value': '9.9.9.9',
+        })
+        zone_normal.add_record(normal)
+
+        excluded = Record.new(zone_excluded, 'www', {
+            'octodns': {
+                'excluded': ['test']
+            },
+            'ttl': 60,
+            'type': 'A',
+            'value': '9.9.9.9',
+        })
+        zone_excluded.add_record(excluded)
+
+        provider = SimpleProvider()
+
+        self.assertFalse(zone_normal.changes(zone_excluded, provider))
+        self.assertTrue(zone_normal.changes(zone_missing, provider))
+
+        self.assertFalse(zone_excluded.changes(zone_normal, provider))
+        self.assertFalse(zone_excluded.changes(zone_missing, provider))
+
+        self.assertTrue(zone_missing.changes(zone_normal, provider))
+        self.assertFalse(zone_missing.changes(zone_excluded, provider))
+
+    def test_included_records(self):
+        zone_normal = Zone('unit.tests.', [])
+        zone_included = Zone('unit.tests.', [])
+        zone_missing = Zone('unit.tests.', [])
+
+        normal = Record.new(zone_normal, 'www', {
+            'ttl': 60,
+            'type': 'A',
+            'value': '9.9.9.9',
+        })
+        zone_normal.add_record(normal)
+
+        included = Record.new(zone_included, 'www', {
+            'octodns': {
+                'included': ['test']
+            },
+            'ttl': 60,
+            'type': 'A',
+            'value': '9.9.9.9',
+        })
+        zone_included.add_record(included)
+
+        provider = SimpleProvider()
+
+        self.assertFalse(zone_normal.changes(zone_included, provider))
+        self.assertTrue(zone_normal.changes(zone_missing, provider))
+
+        self.assertFalse(zone_included.changes(zone_normal, provider))
+        self.assertTrue(zone_included.changes(zone_missing, provider))
+
+        self.assertTrue(zone_missing.changes(zone_normal, provider))
+        self.assertTrue(zone_missing.changes(zone_included, provider))
+
+    def test_not_included_records(self):
+        zone_normal = Zone('unit.tests.', [])
+        zone_included = Zone('unit.tests.', [])
+        zone_missing = Zone('unit.tests.', [])
+
+        normal = Record.new(zone_normal, 'www', {
+            'ttl': 60,
+            'type': 'A',
+            'value': '9.9.9.9',
+        })
+        zone_normal.add_record(normal)
+
+        included = Record.new(zone_included, 'www', {
+            'octodns': {
+                'included': ['not-here']
+            },
+            'ttl': 60,
+            'type': 'A',
+            'value': '9.9.9.9',
+        })
+        zone_included.add_record(included)
+
+        provider = SimpleProvider()
+
+        self.assertFalse(zone_normal.changes(zone_included, provider))
+        self.assertTrue(zone_normal.changes(zone_missing, provider))
+
+        self.assertFalse(zone_included.changes(zone_normal, provider))
+        self.assertFalse(zone_included.changes(zone_missing, provider))
+
+        self.assertTrue(zone_missing.changes(zone_normal, provider))
+        self.assertFalse(zone_missing.changes(zone_included, provider))