Merge ~cgrabowski/maas:update_dns_via_nsupdate into maas:master

[Christian Grabowski]

cleanup dns updates triggers

add nsupdate key

add tsig to packaging

use nsupdate to send bulk commands

queue up diffs of dns changes and process them
queue up DNS updates

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/468/consoleText
COMMIT: c8bd1525bae3aff0a8d360bddbfdf0146e80077c

review: Needs Fixing

[Jack Lloyd-Walters (lloydwaltersj)]

some minor questions

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/482/consoleText
COMMIT: a5b053f45432fbc744aea674be38d2c31f9ad258

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/488/consoleText
COMMIT: 4b2b6ef4d224bea21a59050d17e084319f6fc3e0

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/489/consoleText
COMMIT: 9d290212c760d6c33ea320aff61c9dc8277fe86e

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/490/consoleText
COMMIT: 425c4d2f3382e79c4b1721d6cb6c5d8b18b20e50

review: Needs Fixing

[Alexsander de Souza (alexsander-souza)]

a few questions inline

review: Needs Information

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: fcd948f51f72e398ab6fa1b162b4c3381f14fd29

review: Approve

[Alexsander de Souza (alexsander-souza)]

+1

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/493/consoleText
COMMIT: 8021acff300d749f36c9724cc923bc2e14ac2837

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/514/consoleText
COMMIT: 3a9b652d51a05cb8e83fa4536d33212ea9fb5161

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/556/consoleText
COMMIT: 1defa9a5206bc2d223ed61597ec17fd836d6e11a

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/559/consoleText
COMMIT: 68c4616d4557b29c622eef4fe493e96691ee76a8

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/564/consoleText
COMMIT: 91834894c931d55620fd93e66e7a8903d5a0f63d

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/565/consoleText
COMMIT: f5f85a4eac33db34c8553e8ea6362bbb7a8c5d9a

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/566/consoleText
COMMIT: 64368c19977fd2b14d4a50d02cd002aba07dee7d

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/575/consoleText
COMMIT: 405b6868c73076743b0baa8b48bc7b332170279c

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1154/consoleText
COMMIT: 556b628e5dd110dee9c4ad5561d2a944cfc4059c

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1193/consoleText
COMMIT: 09b92908a47c3ccb9c7210424fe31963078dcadf

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1204/consoleText
COMMIT: e592b85a0b8c74f506d2e1ec448f22a3857fc9f7

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1278/consoleText
COMMIT: 210616767ee57baf8976183f0bfeebbd6ec9deca

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1281/consoleText
COMMIT: 85d59ce7f102b1b242847a2c171adfa2a2bb3c97

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1282/consoleText
COMMIT: d479aede2dd0692029202e33f03325968905426b

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1340/consoleText
COMMIT: e314a01b5cc04452f6783007a4a234037d9c2b45

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1342/consoleText
COMMIT: 668bf3a42b042fcb9019af9d1c443f299da571b4

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1343/consoleText
COMMIT: a3bb7ab6ba6724d12d3e342a46bbbcefde041e51

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1344/consoleText
COMMIT: 939d94111b35d11541eab813c15fd59b8f57fa22

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1345/consoleText
COMMIT: 6fb564b57fe58a2c34ac80330e01dc7feec909de

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: 4e44d9c55fa32e19ccc7b937aa0b75522e8b25d6

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1362/consoleText
COMMIT: 4e44d9c55fa32e19ccc7b937aa0b75522e8b25d6

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1366/consoleText
COMMIT: 8af81b7358c1c4caad6f61574410c0ef9991bafc

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1367/consoleText
COMMIT: 56138ea436744ab571f9b187f7e6fd94f54f6e71

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1368/consoleText
COMMIT: a42fc61d6c6030228b9f77957bd0bd6da608a009

review: Needs Fixing

[Christian Grabowski (cgrabowski)]

jenkins: !test

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: a42fc61d6c6030228b9f77957bd0bd6da608a009

review: Approve

[Alexsander de Souza (alexsander-souza)]

+1, a few nits inline

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1386/consoleText
COMMIT: a9b01be54adf1137008717c2b55118aa3b22e363

review: Needs Fixing

[Christian Grabowski (cgrabowski)]

jenkins: !test

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1387/consoleText
COMMIT: a9b01be54adf1137008717c2b55118aa3b22e363

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1388/consoleText
COMMIT: e94338c9ee9e34fe44662b901a4f904f0f411091

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci.internal:8080/job/maas-tester/1389/consoleText
COMMIT: b4cf0e57bb27ea4a5e7e0ed880500b90b3ce2a92

review: Needs Fixing

[MAAS Lander (maas-lander)]

UNIT TESTS
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: a1306793412e4e2391470d984523214c06ed4544

review: Approve

[MAAS Lander (maas-lander)]

LANDING
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci.internal:8080/job/maas-tester/1391/consoleText

[MAAS Lander (maas-lander)]

LANDING
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci.internal:8080/job/maas-tester/1392/consoleText

[MAAS Lander (maas-lander)]

LANDING
-b update_dns_via_nsupdate lp:~cgrabowski/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci.internal:8080/job/maas-tester/1393/consoleText

diff --git a/debian/maas-region-api.dirs b/debian/maas-region-api.dirs
index d5ba327..1e65cfa 100644
--- a/debian/maas-region-api.dirs
+++ b/debian/maas-region-api.dirs
@@ -1,2 +1,3 @@
 etc/bind/maas
+var/lib/bind/maas
 var/lib/maas/prometheus
diff --git a/debian/maas-region-api.postinst b/debian/maas-region-api.postinst
index a1fb1a1..778504d 100644
--- a/debian/maas-region-api.postinst
+++ b/debian/maas-region-api.postinst
@@ -30,6 +30,11 @@ configure_libdir() {
     if [ -f /var/lib/maas/maas_id ]; then
        chown maas:maas /var/lib/maas/maas_id
     fi
+
+    if [ -d /var/lib/maas/bind ]; then
+        chown maas:bind /var/lib/bind/maas
+        chmod 0775 /var/lib/bind/maas
+    fi
 }
 
 edit_named_options() {
diff --git a/src/maasserver/dns/config.py b/src/maasserver/dns/config.py
index 102e33d..c484570 100644
--- a/src/maasserver/dns/config.py
+++ b/src/maasserver/dns/config.py
@@ -17,7 +17,9 @@ from maasserver.dns.zonegenerator import (
 )
 from maasserver.enum import IPADDRESS_TYPE, RDNS_MODE
 from maasserver.models.config import Config
+from maasserver.models.dnsdata import DNSData
 from maasserver.models.dnspublication import DNSPublication
+from maasserver.models.dnsresource import DNSResource
 from maasserver.models.domain import Domain
 from maasserver.models.node import RackController
 from maasserver.models.subnet import Subnet
@@ -28,7 +30,9 @@ from provisioningserver.dns.actions import (
     bind_write_options,
     bind_write_zones,
 )
+from provisioningserver.dns.config import DynamicDNSUpdate
 from provisioningserver.logger import get_maas_logger
+from provisioningserver.utils.shell import ExternalProcessError
 
 maaslog = get_maas_logger("dns")
 
@@ -61,7 +65,12 @@ def forward_domains_to_forwarded_zones(forward_domains):
     ]
 
 
-def dns_update_all_zones(reload_retry=False, reload_timeout=2):
+def dns_update_all_zones(
+    reload_retry=False,
+    reload_timeout=2,
+    dynamic_updates=None,
+    requires_reload=False,
+):
     """Update all zone files for all domains.
 
     Serving these zone files means updating BIND's configuration to include
@@ -70,10 +79,23 @@ def dns_update_all_zones(reload_retry=False, reload_timeout=2):
     :param reload_retry: Should the DNS server reload be retried in case
         of failure? Defaults to `False`.
     :type reload_retry: bool
+
+    :param reload_timeout: How many seconds to wait for BIND's reload to succeed
+    :type reload_timeout: int
+
+    :param dynamic_updates: A list of updates to send via nsupdate to BIND
+    :type dynamic_updates: list[DynamicDNSUpdate]
+
+    :param requires_reload: If true, dynamic updates are ignored and a full reload will occur
+    :type requires_reload: bool
     """
     if not is_dns_enabled():
         return
 
+    if not dynamic_updates:
+        dynamic_updates = []
+
+    reloaded = True
     domains = Domain.objects.filter(authoritative=True)
     forwarded_zones = forward_domains_to_forwarded_zones(
         Domain.objects.get_forward_domains()
@@ -87,8 +109,13 @@ def dns_update_all_zones(reload_retry=False, reload_timeout=2):
         default_ttl,
         serial,
         internal_domains=[get_internal_domain()],
+        dynamic_updates=dynamic_updates,
+        force_config_write=requires_reload,
     ).as_list()
-    bind_write_zones(zones)
+    try:
+        bind_write_zones(zones)
+    except ExternalProcessError:  # dynamic update failed
+        reloaded = False
 
     # We should not be calling bind_write_options() here; call-sites should be
     # making a separate call. It's a historical legacy, where many sites now
@@ -110,14 +137,21 @@ def dns_update_all_zones(reload_retry=False, reload_timeout=2):
         forwarded_zones=forwarded_zones,
     )
 
-    # Reloading with retries may be a legacy from Celery days, or it may be
-    # necessary to recover from races during start-up. We're not sure if it is
-    # actually needed but it seems safer to maintain this behaviour until we
-    # have a better understanding.
-    if reload_retry:
-        reloaded = bind_reload_with_retries(timeout=reload_timeout)
-    else:
-        reloaded = bind_reload(timeout=reload_timeout)
+    if not requires_reload:
+        for zone in zones:
+            if zone.requires_reload:
+                requires_reload = True
+                break
+
+    if requires_reload:
+        # Reloading with retries may be a legacy from Celery days, or it may be
+        # necessary to recover from races during start-up. We're not sure if it is
+        # actually needed but it seems safer to maintain this behaviour until we
+        # have a better understanding.
+        if reload_retry:
+            reloaded = bind_reload_with_retries(timeout=reload_timeout)
+        else:
+            reloaded = bind_reload(timeout=reload_timeout)
 
     # Return the current serial and list of domain names.
     return serial, reloaded, [domain.name for domain in domains]
@@ -251,3 +285,121 @@ def get_internal_domain():
         ttl=15,
         resources=resources,
     )
+
+
+def process_dns_update_notify(message):
+    updates = []
+    update_list = message.split(" ")
+    op = update_list[0]
+    zone = None
+    name = None
+    rectype = None
+    ttl = None
+    answer = None
+    if op == "RELOAD":
+        return (updates, True)
+    elif op == "INSERT-DATA" or op == "UPDATE-DATA":
+        dns_data = DNSData.objects.get(id=int(update_list[1]))
+        zone = dns_data.dnsresource.domain.name
+        name = dns_data.dnsresource.name
+        ttl = dns_data.ttl
+        rectype = dns_data.rrtype
+        answer = dns_data.rrdata
+    else:
+        zone = update_list[1]
+        name = f"{update_list[2]}.{zone}"
+        rectype = update_list[3]
+        if op == "INSERT" or op == "UPDATE":
+            ttl = int(update_list[-2]) if update_list[-2] else None
+            answer = update_list[-1]
+
+    match op:
+        case "UPDATE":
+            updates.append(
+                DynamicDNSUpdate.create_from_trigger(
+                    operation="DELETE",
+                    zone=zone,
+                    name=name,
+                    rectype=rectype,
+                    answer=answer,
+                )
+            )
+            updates.append(
+                DynamicDNSUpdate.create_from_trigger(
+                    operation="INSERT",
+                    zone=zone,
+                    name=name,
+                    rectype=rectype,
+                    ttl=ttl,
+                    answer=answer,
+                )
+            )
+        case "INSERT":
+            updates.append(
+                DynamicDNSUpdate.create_from_trigger(
+                    operation=op,
+                    zone=zone,
+                    name=name,
+                    rectype=rectype,
+                    ttl=ttl,
+                    answer=answer,
+                )
+            )
+        case _:
+            # special case where we know an IP has been deleted but, we can't fetch the value
+            # and the rrecord may still have other answers
+            if op == "DELETE-IP":
+                updates.append(
+                    DynamicDNSUpdate.create_from_trigger(
+                        operation="DELETE",
+                        zone=zone,
+                        name=name,
+                        rectype=rectype,
+                    )
+                )
+                if rectype == "A":
+                    updates.append(
+                        DynamicDNSUpdate.create_from_trigger(
+                            operation="DELETE",
+                            zone=zone,
+                            name=name,
+                            rectype="AAAA",
+                        )
+                    )
+                resource = DNSResource.objects.get(
+                    name=update_list[2], domain__name=zone
+                )
+                updates += [
+                    DynamicDNSUpdate.create_from_trigger(
+                        operation="INSERT",
+                        zone=zone,
+                        name=name,
+                        rectype=rectype,
+                        ttl=int(resource.address_ttl)
+                        if resource.address_ttl
+                        else None,
+                        answer=ip.ip,
+                    )
+                    for ip in resource.ip_addresses.all()
+                ]
+
+            elif len(update_list) > 4:  # has an answer
+                updates.append(
+                    DynamicDNSUpdate.create_from_trigger(
+                        operation=op,
+                        zone=zone,
+                        name=name,
+                        rectype=rectype,
+                        answer=update_list[-1],
+                    )
+                )
+            else:
+                updates.append(
+                    DynamicDNSUpdate.create_from_trigger(
+                        operation=op,
+                        zone=zone,
+                        name=name,
+                        rectype=rectype,
+                    )
+                )
+    return (updates, False)
diff --git a/src/maasserver/dns/tests/test_config.py b/src/maasserver/dns/tests/test_config.py
index 1b2f15a..913c96e 100644
--- a/src/maasserver/dns/tests/test_config.py
+++ b/src/maasserver/dns/tests/test_config.py
@@ -29,6 +29,7 @@ from maasserver.dns.config import (
     get_trusted_acls,
     get_trusted_networks,
     get_upstream_dns,
+    process_dns_update_notify,
 )
 from maasserver.dns.zonegenerator import InternalDomainResourseRecord
 from maasserver.enum import IPADDRESS_TYPE, NODE_STATUS
@@ -40,10 +41,15 @@ from maasserver.testing.factory import factory
 from maasserver.testing.testcase import MAASServerTestCase
 from maastesting.matchers import MockCalledOnceWith
 from provisioningserver.dns.commands import get_named_conf, setup_dns
-from provisioningserver.dns.config import compose_config_path, DNSConfig
+from provisioningserver.dns.config import (
+    compose_config_path,
+    DNSConfig,
+    DynamicDNSUpdate,
+)
 from provisioningserver.dns.testing import (
     patch_dns_config_path,
     patch_dns_rndc_port,
+    patch_zone_file_config_path,
 )
 from provisioningserver.testing.bindfixture import allocate_ports, BINDServer
 from provisioningserver.testing.tests.test_bindfixture import dig_call
@@ -127,6 +133,7 @@ class TestDNSServer(MAASServerTestCase):
         patch_dns_config_path(self, self.bind.config.homedir)
         # Use a random port for rndc.
         patch_dns_rndc_port(self, allocate_ports("localhost")[0])
+        patch_zone_file_config_path(self, config_dir=self.bind.config.homedir)
         # This simulates what should happen when the package is
         # installed:
         # Create MAAS-specific DNS configuration files.
@@ -431,6 +438,28 @@ class TestDNSConfigModifications(TestDNSServer):
             ),
         )
 
+    def test_dns_update_all_zones_does_not_reload_if_it_does_not_need_to(self):
+        self.patch(settings, "DNS_CONNECT", True)
+        domain = factory.make_Domain()
+        # These domains should not show up. Just to test we create them.
+        for _ in range(3):
+            factory.make_Domain(authoritative=False)
+        node, static = self.create_node_with_static_ip(domain=domain)
+        fake_serial = random.randint(1, 1000)
+        self.patch(
+            dns_config_module, "current_zone_serial"
+        ).return_value = fake_serial
+        reload_call = self.patch(dns_config_module, "bind_reload")
+        serial1, reloaded, _ = dns_update_all_zones(
+            reload_timeout=RELOAD_TIMEOUT
+        )
+        self.assertTrue(reloaded)
+        factory.make_DNSResource(domain=domain)
+        _, _, _ = dns_update_all_zones(  # should be a dynamic update
+            reload_timeout=RELOAD_TIMEOUT
+        )
+        reload_call.assert_called_once()
+
 
 class TestDNSDynamicIPAddresses(TestDNSServer):
     """Allocated nodes with IP addresses in the dynamic range get a DNS
@@ -754,3 +783,122 @@ class TestGetResourceNameForSubnet(MAASServerTestCase):
     def test_returns_valid(self):
         subnet = factory.make_Subnet(cidr=self.cidr)
         self.assertEqual(self.result, get_resource_name_for_subnet(subnet))
+
+
+class TestProcessDNSUpdateNotify(MAASServerTestCase):
+    def test_insert(self):
+        domain = factory.make_Domain()
+        resource = factory.make_DNSResource(domain=domain)
+        ip = resource.ip_addresses.first().ip
+        message = f"INSERT {domain.name} {resource.name} A {resource.address_ttl if resource.address_ttl else 60} {ip}"
+        result, _ = process_dns_update_notify(message)
+        self.assertCountEqual(
+            [
+                DynamicDNSUpdate(
+                    operation="INSERT",
+                    zone=domain.name,
+                    name=f"{resource.name}.{domain.name}",
+                    ttl=resource.address_ttl if resource.address_ttl else 60,
+                    answer=ip,
+                    rectype="A" if IPAddress(ip).version == 4 else "AAAA",
+                )
+            ],
+            result,
+        )
+
+    def test_delete_without_ip(self):
+        domain = factory.make_Domain()
+        resource = factory.make_DNSResource(domain=domain)
+        message = f"DELETE {domain.name} {resource.name} A"
+        result, _ = process_dns_update_notify(message)
+        self.assertCountEqual(
+            [
+                DynamicDNSUpdate(
+                    operation="DELETE",
+                    zone=domain.name,
+                    name=f"{resource.name}.{domain.name}",
+                    rectype="A",
+                )
+            ],
+            result,
+        )
+
+    def test_delete_with_ip(self):
+        domain = factory.make_Domain()
+        resource = factory.make_DNSResource(domain=domain)
+        ip = resource.ip_addresses.first().ip
+        message = f"DELETE {domain.name} {resource.name} A {ip}"
+        result, _ = process_dns_update_notify(message)
+        self.assertCountEqual(
+            [
+                DynamicDNSUpdate(
+                    operation="DELETE",
+                    zone=domain.name,
+                    name=f"{resource.name}.{domain.name}",
+                    answer=ip,
+                    rectype="A" if IPAddress(ip).version == 4 else "AAAA",
+                )
+            ],
+            result,
+        )
+
+    def test_update(self):
+        domain = factory.make_Domain()
+        resource = factory.make_DNSResource(domain=domain)
+        ip = resource.ip_addresses.first().ip
+        message = f"UPDATE {domain.name} {resource.name} A {resource.address_ttl if resource.address_ttl else 60} {ip}"
+        result, _ = process_dns_update_notify(message)
+        self.assertCountEqual(
+            [
+                DynamicDNSUpdate(
+                    operation="DELETE",
+                    zone=domain.name,
+                    name=f"{resource.name}.{domain.name}",
+                    answer=ip,
+                    rectype="A" if IPAddress(ip).version == 4 else "AAAA",
+                ),
+                DynamicDNSUpdate(
+                    operation="INSERT",
+                    zone=domain.name,
+                    name=f"{resource.name}.{domain.name}",
+                    ttl=resource.address_ttl if resource.address_ttl else 60,
+                    answer=ip,
+                    rectype="A" if IPAddress(ip).version == 4 else "AAAA",
+                ),
+            ],
+            result,
+        )
+
+    def test_delete_ip(self):
+        domain = factory.make_Domain()
+        resource = factory.make_DNSResource(domain=domain)
+        ip = resource.ip_addresses.first().ip
+        ip2 = factory.make_StaticIPAddress()
+        resource.ip_addresses.add(ip2)
+        message = f"DELETE-IP {domain.name} {resource.name} A {resource.address_ttl if resource.address_ttl else 60} {ip}"
+        resource.ip_addresses.first().delete()
+        result, _ = process_dns_update_notify(message)
+        self.assertCountEqual(
+            [
+                DynamicDNSUpdate(
+                    operation="DELETE",
+                    zone=domain.name,
+                    name=f"{resource.name}.{domain.name}",
+                    rectype="A",
+                ),
+                DynamicDNSUpdate(
+                    operation="DELETE",
+                    zone=domain.name,
+                    name=f"{resource.name}.{domain.name}",
+                    rectype="AAAA",
+                ),
+                DynamicDNSUpdate(
+                    operation="INSERT",
+                    zone=domain.name,
+                    name=f"{resource.name}.{domain.name}",
+                    rectype="A" if IPAddress(ip2.ip).version == 4 else "AAAA",
+                    answer=ip2.ip,
+                ),
+            ],
+            result,
+        )
diff --git a/src/maasserver/dns/zonegenerator.py b/src/maasserver/dns/zonegenerator.py
index 0e4c872..ef31cfa 100644
--- a/src/maasserver/dns/zonegenerator.py
+++ b/src/maasserver/dns/zonegenerator.py
@@ -21,6 +21,7 @@ from maasserver.models.domain import Domain
 from maasserver.models.staticipaddress import StaticIPAddress
 from maasserver.models.subnet import Subnet
 from maasserver.server_address import get_maas_facing_server_addresses
+from provisioningserver.dns.config import DynamicDNSUpdate
 from provisioningserver.dns.zoneconfig import (
     DNSForwardZoneConfig,
     DNSReverseZoneConfig,
@@ -200,6 +201,8 @@ class ZoneGenerator:
         default_ttl=None,
         serial=None,
         internal_domains=None,
+        dynamic_updates=None,
+        force_config_write=False,
     ):
         """
         :param serial: A serial number to reuse when creating zones in bulk.
@@ -215,6 +218,10 @@ class ZoneGenerator:
         self.internal_domains = internal_domains
         if self.internal_domains is None:
             self.internal_domains = []
+        self._dynamic_updates = dynamic_updates
+        if self._dynamic_updates is None:
+            self._dynamic_updates = []
+        self.force_config_write = force_config_write  # some data changed that nsupdate cannot update if true
 
     @staticmethod
     def _get_mappings():
@@ -235,6 +242,8 @@ class ZoneGenerator:
         rrset_mappings,
         default_ttl,
         internal_domains,
+        dynamic_updates,
+        force_config_write,
     ):
         """Generator of forward zones, collated by domain name."""
         dns_ip_list = get_dns_server_addresses(filter_allowed_dns=False)
@@ -290,6 +299,12 @@ class ZoneGenerator:
                             (ttl, "AAAA", dns_ip.format())
                         )
 
+            domain_updates = [
+                update
+                for update in dynamic_updates
+                if update.zone == domain.name
+            ]
+
             yield DNSForwardZoneConfig(
                 domain.name,
                 serial=serial,
@@ -301,6 +316,8 @@ class ZoneGenerator:
                 ns_host_name=ns_host_name,
                 other_mapping=other_mapping,
                 dynamic_ranges=dynamic_ranges,
+                dynamic_updates=domain_updates,
+                force_config_write=force_config_write,
             )
 
         # Create the forward zone config for the internal domains.
@@ -313,6 +330,13 @@ class ZoneGenerator:
                     resource_mapping.rrset.add(
                         (internal_domain.ttl, record.rrtype, record.rrdata)
                     )
+
+            domain_updates = [
+                update
+                for update in dynamic_updates
+                if update.zone == internal_domain.name
+            ]
+
             yield DNSForwardZoneConfig(
                 internal_domain.name,
                 serial=serial,
@@ -324,11 +348,19 @@ class ZoneGenerator:
                 ns_host_name=ns_host_name,
                 other_mapping=other_mapping,
                 dynamic_ranges=[],
+                dynamic_updates=domain_updates,
+                force_config_write=force_config_write,
             )
 
     @staticmethod
     def _gen_reverse_zones(
-        subnets, serial, ns_host_name, mappings, default_ttl
+        subnets,
+        serial,
+        ns_host_name,
+        mappings,
+        default_ttl,
+        dynamic_updates,
+        force_config_write,
     ):
         """Generator of reverse zones, sorted by network."""
 
@@ -418,6 +450,15 @@ class ZoneGenerator:
                 del rfc2317_glue[network]
             else:
                 glue = set()
+
+            domain_updates = [
+                DynamicDNSUpdate.as_reverse_record_update(update, subnet)
+                for update in dynamic_updates
+                if update.answer
+                and update.answer_is_ip
+                and (IPAddress(update.answer) in IPNetwork(subnet.cidr))
+            ]
+
             yield DNSReverseZoneConfig(
                 ns_host_name,
                 serial=serial,
@@ -430,6 +471,8 @@ class ZoneGenerator:
                 exclude={
                     IPNetwork(s.cidr) for s in subnets if s is not subnet
                 },
+                dynamic_updates=domain_updates,
+                force_config_write=force_config_write,
             )
         # Now provide any remaining rfc2317 glue networks.
         for network, ranges in rfc2317_glue.items():
@@ -445,6 +488,8 @@ class ZoneGenerator:
                     for s in subnets
                     if network in IPNetwork(s.cidr)
                 },
+                dynamic_updates=domain_updates,
+                force_config_write=force_config_write,
             )
 
     def __iter__(self):
@@ -470,9 +515,17 @@ class ZoneGenerator:
                 rrset_mappings,
                 default_ttl,
                 self.internal_domains,
+                self._dynamic_updates,
+                self.force_config_write,
             ),
             self._gen_reverse_zones(
-                self.subnets, serial, ns_host_name, mappings, default_ttl
+                self.subnets,
+                serial,
+                ns_host_name,
+                mappings,
+                default_ttl,
+                self._dynamic_updates,
+                self.force_config_write,
             ),
         )
 
diff --git a/src/maasserver/region_controller.py b/src/maasserver/region_controller.py
index e795afb..69eae10 100644
--- a/src/maasserver/region_controller.py
+++ b/src/maasserver/region_controller.py
@@ -41,7 +41,10 @@ from twisted.internet.task import LoopingCall
 from twisted.names.client import Resolver
 
 from maasserver import eventloop, locks
-from maasserver.dns.config import dns_update_all_zones
+from maasserver.dns.config import (
+    dns_update_all_zones,
+    process_dns_update_notify,
+)
 from maasserver.macaroon_auth import get_auth_info
 from maasserver.models.dnspublication import DNSPublication
 from maasserver.models.rbacsync import RBAC_ACTION, RBACLastSync, RBACSync
@@ -94,6 +97,8 @@ class RegionControllerService(Service):
         self.needsDNSUpdate = False
         self.needsProxyUpdate = False
         self.needsRBACUpdate = False
+        self._dns_updates = []
+        self._dns_requires_full_reload = False
         self.postgresListener = postgresListener
         self.dnsResolver = Resolver(
             resolv=None,
@@ -110,6 +115,9 @@ class RegionControllerService(Service):
         """Start listening for messages."""
         super().startService()
         self.postgresListener.register("sys_dns", self.markDNSForUpdate)
+        self.postgresListener.register(
+            "sys_dns_updates", self.queueDynamicDNSUpdate
+        )
         self.postgresListener.register("sys_proxy", self.markProxyForUpdate)
         self.postgresListener.register("sys_rbac", self.markRBACForUpdate)
         self.postgresListener.register(
@@ -164,6 +172,21 @@ class RegionControllerService(Service):
         )
         eventloop.restart()
 
+    def queueDynamicDNSUpdate(self, channel, message):
+        """
+        Called when the `sys_dns_update` message is received
+        and queues updates for existing domains
+        """
+        if message == "":
+            return
+
+        (new_updates, need_reload) = process_dns_update_notify(message)
+
+        self._dns_requires_full_reload = (
+            self._dns_requires_full_reload or need_reload
+        )
+        self._dns_updates += new_updates
+
     def startProcessing(self):
         """Start the process looping call."""
         if not self.processing.running:
@@ -193,10 +216,22 @@ class RegionControllerService(Service):
             if delay:
                 return pause(delay)
 
+        def _clear_dynamic_dns_updates(d):
+            self._dns_updates = []
+            self._dns_requires_full_reload = False
+            return d
+
         defers = []
         if self.needsDNSUpdate:
             self.needsDNSUpdate = False
-            d = deferToDatabase(transactional(dns_update_all_zones))
+            d = deferToDatabase(
+                transactional(
+                    dns_update_all_zones,
+                ),
+                dynamic_updates=self._dns_updates,
+                requires_reload=self._dns_requires_full_reload,
+            )
+            d.addCallback(_clear_dynamic_dns_updates)
             d.addCallback(self._checkSerial)
             d.addCallback(self._logDNSReload)
             # Order here matters, first needsDNSUpdate is set then pass the
diff --git a/src/maasserver/region_script.py b/src/maasserver/region_script.py
index 0c87c61..7894260 100644
--- a/src/maasserver/region_script.py
+++ b/src/maasserver/region_script.py
@@ -42,6 +42,7 @@ def run_django(is_snap, is_devenv):
                 "MAAS_DNS_CONFIG_DIR": os.path.join(snap_data, "bind"),
                 "MAAS_PROXY_CONFIG_DIR": os.path.join(snap_data, "proxy"),
                 "MAAS_SYSLOG_CONFIG_DIR": os.path.join(snap_data, "syslog"),
+                "MAAS_ZONE_FILE_CONFIG_DIR": os.path.join(snap_data, "bind"),
                 "MAAS_IMAGES_KEYRING_FILEPATH": (
                     "/snap/maas/current/usr/share/keyrings/"
                     "ubuntu-cloudimage-keyring.gpg"
diff --git a/src/maasserver/tests/test_region_controller.py b/src/maasserver/tests/test_region_controller.py
index 0995258..4f40354 100644
--- a/src/maasserver/tests/test_region_controller.py
+++ b/src/maasserver/tests/test_region_controller.py
@@ -69,6 +69,7 @@ class TestRegionControllerService(MAASServerTestCase):
             listener.register,
             MockCallsMatch(
                 call("sys_dns", service.markDNSForUpdate),
+                call("sys_dns_updates", service.queueDynamicDNSUpdate),
                 call("sys_proxy", service.markProxyForUpdate),
                 call("sys_rbac", service.markRBACForUpdate),
                 call("sys_vault_migration", service.restartRegion),
@@ -216,7 +217,9 @@ class TestRegionControllerService(MAASServerTestCase):
         mock_msg = self.patch(region_controller.log, "msg")
         service.startProcessing()
         yield service.processingDefer
-        self.assertThat(mock_dns_update_all_zones, MockCalledOnceWith())
+        mock_dns_update_all_zones.assert_called_once_with(
+            dynamic_updates=[], requires_reload=False
+        )
         self.assertThat(mock_check_serial, MockCalledOnceWith(dns_result))
         self.assertThat(
             mock_msg,
@@ -258,7 +261,11 @@ class TestRegionControllerService(MAASServerTestCase):
         service.startProcessing()
         yield service.processingDefer
         self.assertThat(
-            mock_dns_update_all_zones, MockCallsMatch(call(), call())
+            mock_dns_update_all_zones,
+            MockCallsMatch(
+                call(dynamic_updates=[], requires_reload=False),
+                call(dynamic_updates=[], requires_reload=False),
+            ),
         )
         self.assertThat(
             mock_check_serial,
@@ -313,7 +320,9 @@ class TestRegionControllerService(MAASServerTestCase):
         mock_err = self.patch(region_controller.log, "err")
         service.startProcessing()
         yield service.processingDefer
-        self.assertThat(mock_dns_update_all_zones, MockCalledOnceWith())
+        mock_dns_update_all_zones.assert_called_once_with(
+            dynamic_updates=[], requires_reload=False
+        )
         self.assertThat(
             mock_err, MockCalledOnceWith(ANY, "Failed configuring DNS.")
         )
@@ -400,12 +409,11 @@ class TestRegionControllerService(MAASServerTestCase):
         mock_rbacSync.return_value = None
         service.startProcessing()
         yield service.processingDefer
-        self.assertThat(mock_dns_update_all_zones, MockCalledOnceWith())
-        self.assertThat(mock_check_serial, MockCalledOnceWith(dns_result))
-        self.assertThat(
-            mock_proxy_update_config, MockCalledOnceWith(reload_proxy=True)
+        mock_dns_update_all_zones.assert_called_once_with(
+            dynamic_updates=[], requires_reload=False
         )
-        self.assertThat(mock_rbacSync, MockCalledOnceWith())
+        mock_proxy_update_config.assert_called_once_with(reload_proxy=True)
+        mock_rbacSync.assert_called_once()
 
     def make_soa_result(self, serial):
         return RRHeader(
@@ -626,7 +634,9 @@ class TestRegionControllerServiceTransactional(MAASTransactionServerTestCase):
         mock_msg = self.patch(region_controller.log, "msg")
         service.startProcessing()
         yield service.processingDefer
-        self.assertThat(mock_dns_update_all_zones, MockCalledOnceWith())
+        mock_dns_update_all_zones.assert_called_once_with(
+            dynamic_updates=[], requires_reload=False
+        )
         self.assertThat(mock_check_serial, MockCalledOnceWith(dns_result))
         self.assertThat(
             mock_msg,
@@ -670,7 +680,9 @@ class TestRegionControllerServiceTransactional(MAASTransactionServerTestCase):
             " * %s" % publication.source
             for publication in reversed(publications[1:])
         )
-        self.assertThat(mock_dns_update_all_zones, MockCalledOnceWith())
+        mock_dns_update_all_zones.assert_called_once_with(
+            dynamic_updates=[], requires_reload=False
+        )
         self.assertThat(mock_check_serial, MockCalledOnceWith(dns_result))
         self.assertThat(mock_msg, MockCalledOnceWith(expected_msg))
 
diff --git a/src/maasserver/triggers/system.py b/src/maasserver/triggers/system.py
index 0cf4e90..a18fe8a 100644
--- a/src/maasserver/triggers/system.py
+++ b/src/maasserver/triggers/system.py
@@ -1901,6 +1901,159 @@ RBAC_RPOOL_DELETE = dedent(
 )
 
 
+# handles dynamic updates when a dnsresource is created
+# or a StaticIPAddres maps to a resource or was deleted
+def render_dns_dynamic_update_dnsresource_ip_addresses_procedure(op):
+    return dedent(
+        f"""\
+        CREATE OR REPLACE FUNCTION sys_dns_updates_dns_ip_{op}()
+        RETURNS trigger as $$
+        DECLARE
+          ip_addr text;
+          rname text;
+          rdomain_id bigint;
+          domain text;
+          ttl integer;
+        BEGIN
+          ASSERT TG_WHEN = 'AFTER', 'May only run as an AFTER trigger';
+          ASSERT TG_LEVEL <> 'STATEMENT', 'Should not be used as a STATEMENT level trigger', TG_NAME;
+          IF (TG_OP = 'INSERT' AND TG_LEVEL = 'ROW') THEN
+            SELECT host(ip) INTO ip_addr FROM maasserver_staticipaddress WHERE id=NEW.staticipaddress_id;
+            SELECT name, domain_id, COALESCE(address_ttl, 0) INTO rname, rdomain_id, ttl FROM maasserver_dnsresource WHERE id=NEW.dnsresource_id;
+            SELECT name INTO domain FROM maasserver_domain WHERE id=rdomain_id;
+            PERFORM pg_notify('sys_dns_updates', 'INSERT ' || domain || ' ' || rname || ' A ' || ttl || ' ' || ip_addr);
+          ELSIF (TG_OP = 'DELETE' AND TG_LEVEl = 'ROW') THEN
+            IF EXISTS(SELECT id FROM maasserver_dnsresource WHERE id=OLD.dnsresource_id) THEN
+              IF EXISTS(SELECT id FROM maasserver_staticipaddress WHERE id=OLD.staticipaddress_id) THEN
+                SELECT host(ip) INTO ip_addr FROM maasserver_staticipaddress WHERE id=OLD.staticipaddress_id;
+                SELECT name, domain_id INTO rname, rdomain_id FROM maasserver_dnsresource WHERE id=OLD.dnsresource_id;
+                SELECT name INTO domain FROM maasserver_domain WHERE id=rdomain_id;
+                PERFORM pg_notify('sys_dns_updates', 'DELETE ' || domain || ' ' || rname || ' A ' || ip_addr);
+              ELSE
+                SELECT name, domain_id INTO rname, rdomain_id FROM maasserver_dnsresource WHERE id=NEW.dnsresource_id;
+                SELECT name INTO domain FROM maasserver_domain WHERE id=rdomain_id;
+                PERFORM pg_notify('sys_dns_updates', 'DELETE-IP ' || domain || ' ' || rname || ' A');
+                PERFORM pg_notify('sys_dns_updates', 'DELETE-IP ' || domain || ' ' || rname || ' AAAA');
+              END IF;
+            END IF;
+          END IF;
+          RETURN NULL;
+        END;
+        $$ LANGUAGE plpgsql;
+        """
+    )
+
+
+# handles when ttl or name is modified or resource is deleted,
+# DNS_DYNAMIC_UPDATE_DNSRESOURCE_STATICIPADDRESS covers the case of insert
+def render_dns_dynamic_update_dnsresource_procedure(op):
+    return dedent(
+        f"""\
+        CREATE OR REPLACE FUNCTION sys_dns_updates_maasserver_dnsresource_{op}()
+        RETURNS trigger as $$
+        DECLARE
+          ip_addr text;
+          ips text[];
+          domain text;
+        BEGIN
+          ASSERT TG_WHEN = 'AFTER', 'May only run as an AFTER trigger';
+          ASSERT TG_LEVEL <> 'STATEMENT', 'Should not be used as a STATEMENT level trigger', TG_NAME;
+          IF (TG_OP = 'UPDATE' AND TG_LEVEL = 'ROW') THEN
+            IF NEW IS DISTINCT FROM OLD THEN
+                SELECT array_agg(host(ip)) INTO ips FROM maasserver_dnsresource_ip_addresses m
+                  INNER JOIN maasserver_staticipaddress ON maasserver_staticipaddress.id=m.staticipaddress_id
+                  WHERE dnsresource_id=NEW.id;
+                SELECT name INTO domain FROM maasserver_domain WHERE id=NEW.domain_id;
+                IF array_length(ips, 1) > 0 THEN
+                    FOREACH ip_addr IN ARRAY ips
+                    LOOP
+                      IF OLD.name <> NEW.name THEN
+                        PERFORM pg_notify('sys_dns_updates', 'DELETE ' || domain || ' ' || OLD.name || ' A ' || ip_addr);
+                        PERFORM pg_notify('sys_dns_updates', 'INSERT ' || domain || ' ' || NEW.name || ' A ' || NEW.address_ttl || ' ' || ip_addr);
+                      ELSE
+                        PERFORM pg_notify('sys_dns_updates', 'UPDATE ' || domain || ' ' || NEW.name || ' A ' || NEW.address_ttl || ' ' || ip_addr);
+                      END IF;
+                    END LOOP;
+                END IF;
+            ELSE
+              RETURN NULL;
+            END IF;
+          ELSIF (TG_OP = 'DELETE' AND TG_LEVEL = 'ROW') THEN
+            SELECT name INTO domain FROM maasserver_domain WHERE id=NEW.domain_id;
+            PERFORM pg_notify('sys_dns_updates', 'DELETE ' || domain || ' ' || OLD.name || ' A');
+          END IF;
+          RETURN NULL;
+        END;
+        $$ LANGUAGE plpgsql;
+        """
+    )
+
+
+def render_dns_dynamic_update_dnsdata_procedure(op):
+    return dedent(
+        f"""\
+        CREATE OR REPLACE FUNCTION sys_dns_updates_maasserver_dnsdata_{op}()
+        RETURNS trigger as $$
+        DECLARE
+          rname text;
+          rdomain_id bigint;
+          domain text;
+          ttl int;
+        BEGIN
+          ASSERT TG_WHEN = 'AFTER', 'May only run as an AFTER trigger';
+          ASSERT TG_LEVEL <> 'STATEMENT', 'Should not be used as a STATEMENT level trigger', TG_NAME;
+          IF (TG_OP = 'UPDATE' AND TG_LEVEL = 'ROW') THEN
+            IF NEW IS DISTINCT FROM OLD THEN
+                PERFORM pg_notify('sys_dns_updates', 'UPDATE-DATA ' || NEW.id);
+            ELSE
+              RETURN NULL;
+            END IF;
+          ELSIF (TG_OP = 'INSERT' AND TG_LEVEL = 'ROW') THEN
+            PERFORM pg_notify('sys_dns_updates', 'INSERT-DATA ' || NEW.id);
+          ELSIF (TG_OP = 'DELETE' AND TG_LEVEL = 'ROW') THEN
+            SELECT name, domain_id INTO rname, rdomain_id from maasserver_dnsresource WHERE id=OLD.dnsresource_id;
+            SELECT name INTO domain FROM maasserver_domain WHERE id=rdomain_id;
+            PERFORM pg_notify('sys_dns_updates', 'DELETE ' || domain || ' ' || rname || ' ' || OLD.rrtype);
+          END IF;
+          RETURN NULL;
+        END;
+        $$ LANGUAGE plpgsql;
+        """
+    )
+
+
+def render_dns_dynamic_update_domain_procedure(op):
+    return dedent(
+        f"""\
+        CREATE OR REPLACE FUNCTION sys_dns_updates_maasserver_domain_{op}()
+        RETURNS trigger as $$
+        BEGIN
+          ASSERT TG_WHEN = 'AFTER', 'May only run as an AFTER trigger';
+          ASSERT TG_LEVEL <> 'STATEMENT', 'Should not be used as a STATEMENT level trigger', TG_NAME;
+          PERFORM pg_notify('sys_dns_updates', 'RELOAD');
+          RETURN NULL;
+        END;
+        $$ LANGUAGE plpgsql;
+        """
+    )
+
+
+def render_dns_dynamic_update_subnet_procedure(op):
+    return dedent(
+        f"""\
+        CREATE OR REPLACE FUNCTION sys_dns_updates_maasserver_subnet_{op}()
+        RETURNS trigger as $$
+        BEGIN
+          ASSERT TG_WHEN = 'AFTER', 'May only run as an AFTER trigger';
+          ASSERT TG_LEVEL <> 'STATEMENT', 'Should not be used as a STATEMENT level trigger', TG_NAME;
+          PERFORM pg_notify('sys_dns_updates', 'RELOAD');
+          RETURN NULL;
+        END;
+        $$ LANGUAGE plpgsql;
+        """
+    )
+
+
 def render_sys_proxy_procedure(proc_name, on_delete=False):
     """Render a database procedure with name `proc_name` that notifies that a
     proxy update is needed.
@@ -2163,3 +2316,90 @@ def register_system_triggers():
     register_trigger(
         "maasserver_resourcepool", "sys_rbac_rpool_delete", "delete"
     )
+
+    register_procedure(
+        render_dns_dynamic_update_dnsresource_ip_addresses_procedure("insert")
+    )
+    register_trigger(
+        "maasserver_dnsresource_ip_addresses",
+        "sys_dns_updates_dns_ip_insert",
+        "insert",
+    )
+    register_procedure(
+        render_dns_dynamic_update_dnsresource_ip_addresses_procedure("delete")
+    )
+    register_trigger(
+        "maasserver_dnsresource_ip_addresses",
+        "sys_dns_updates_dns_ip_delete",
+        "delete",
+    )
+    register_procedure(
+        render_dns_dynamic_update_dnsresource_procedure("update")
+    )
+    register_trigger(
+        "maasserver_dnsresource",
+        "sys_dns_updates_maasserver_dnsresource_update",
+        "update",
+    )
+    register_procedure(
+        render_dns_dynamic_update_dnsresource_procedure("delete")
+    )
+    register_trigger(
+        "maasserver_dnsresource",
+        "sys_dns_updates_maasserver_dnsresource_delete",
+        "delete",
+    )
+    register_procedure(render_dns_dynamic_update_dnsdata_procedure("insert"))
+    register_trigger(
+        "maasserver_dnsdata",
+        "sys_dns_updates_maasserver_dnsdata_insert",
+        "update",
+    )
+    register_procedure(render_dns_dynamic_update_dnsdata_procedure("update"))
+    register_trigger(
+        "maasserver_dnsdata",
+        "sys_dns_updates_maasserver_dnsdata_update",
+        "update",
+    )
+    register_procedure(render_dns_dynamic_update_dnsdata_procedure("delete"))
+    register_trigger(
+        "maasserver_dnsdata",
+        "sys_dns_updates_maasserver_dnsdata_delete",
+        "delete",
+    )
+    register_procedure(render_dns_dynamic_update_domain_procedure("insert"))
+    register_trigger(
+        "maasserver_domain",
+        "sys_dns_updates_maasserver_domain_insert",
+        "insert",
+    )
+    register_procedure(render_dns_dynamic_update_domain_procedure("update"))
+    register_trigger(
+        "maasserver_domain",
+        "sys_dns_updates_maasserver_domain_update",
+        "update",
+    )
+    register_procedure(render_dns_dynamic_update_domain_procedure("delete"))
+    register_trigger(
+        "maasserver_domain",
+        "sys_dns_updates_maasserver_domain_delete",
+        "delete",
+    )
+    register_procedure(render_dns_dynamic_update_subnet_procedure("insert"))
+    register_trigger(
+        "maasserver_subnet",
+        "sys_dns_updates_maasserver_subnet_insert",
+        "insert",
+    )
+    register_procedure(render_dns_dynamic_update_subnet_procedure("update"))
+    register_trigger(
+        "maasserver_subnet",
+        "sys_dns_updates_maasserver_subnet_update",
+        "update",
+    )
+    register_procedure(render_dns_dynamic_update_subnet_procedure("delete"))
+    register_trigger(
+        "maasserver_subnet",
+        "sys_dns_updates_maasserver_subnet_delete",
+        "delete",
+    )
diff --git a/src/maasserver/triggers/testing.py b/src/maasserver/triggers/testing.py
index 538fbc4..551ef4d 100644
--- a/src/maasserver/triggers/testing.py
+++ b/src/maasserver/triggers/testing.py
@@ -4,11 +4,10 @@
 """Helper class for all tests using the `PostgresListenerService` under
 `maasserver.triggers.tests`."""
 
-
 from django.contrib.auth.models import User
 from piston3.models import Token
 from testtools.matchers import GreaterThan, Is, Not
-from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.internet.defer import DeferredQueue, inlineCallbacks, returnValue
 
 from maasserver.enum import INTERFACE_TYPE, NODE_TYPE
 from maasserver.listener import PostgresListenerService
@@ -53,6 +52,7 @@ from maasserver.models.virtualblockdevice import VirtualBlockDevice
 from maasserver.models.vlan import VLAN
 from maasserver.models.zone import Zone
 from maasserver.testing.factory import factory, RANDOM
+from maasserver.triggers import register_trigger
 from maasserver.utils.orm import reload_object, transactional
 from maasserver.utils.threads import deferToDatabase
 from maastesting.crochet import wait_for
@@ -942,3 +942,44 @@ class RBACHelpersMixin:
                 GreaterThan(old.id),
                 "RBAC sync tracking has not been modified again.",
             )
+
+
+class NotifyHelperMixin:
+
+    channels = ()
+    channel_queues = {}
+    postgres_listener_service = None
+
+    @inlineCallbacks
+    def set_service(self, listener):
+        self.postgres_listener_service = listener
+        yield self.postgres_listener_service.startService()
+
+    def register_trigger(self, table, channel, ops=(), trigger=None):
+        if channel not in self.channels:
+            self.postgres_listener_service.registerChannel(channel)
+            self.postgres_listener_service.register(channel, self.listen)
+            self.channels = self.channels + (channel,)
+        for op in ops:
+            trigger = trigger or f"{channel}_{table}_{op}"
+            register_trigger(table, trigger, op)
+
+    @inlineCallbacks
+    def listen(self, channel, msg):
+        if msg:
+            yield self.channel_queues[channel].put(msg)
+
+    @inlineCallbacks
+    def get_notify(self, channel):
+        msg = yield self.channel_queues[channel].get()
+        return msg
+
+    def start_reading(self):
+        for channel in self.channels:
+            self.channel_queues[channel] = DeferredQueue()
+        self.postgres_listener_service.startReading()
+
+    def stop_reading(self):
+        for channel in self.channels:
+            del self.channel_queues[channel]
+        self.postgres_listener_service.stopReading()
diff --git a/src/maasserver/triggers/tests/test_init.py b/src/maasserver/triggers/tests/test_init.py
index 7b6e8ee..e4fb3ed 100644
--- a/src/maasserver/triggers/tests/test_init.py
+++ b/src/maasserver/triggers/tests/test_init.py
@@ -70,15 +70,25 @@ class TestTriggersUsed(MAASServerTestCase):
         "dnsdata_sys_dns_dnsdata_delete",
         "dnsdata_sys_dns_dnsdata_insert",
         "dnsdata_sys_dns_dnsdata_update",
+        "dnsdata_sys_dns_updates_maasserver_dnsdata_delete",
+        "dnsdata_sys_dns_updates_maasserver_dnsdata_insert",
+        "dnsdata_sys_dns_updates_maasserver_dnsdata_update",
         "dnspublication_sys_dns_publish",
         "dnsresource_ip_addresses_sys_dns_dnsresource_ip_link",
         "dnsresource_ip_addresses_sys_dns_dnsresource_ip_unlink",
         "dnsresource_sys_dns_dnsresource_delete",
         "dnsresource_sys_dns_dnsresource_insert",
         "dnsresource_sys_dns_dnsresource_update",
+        "dnsresource_sys_dns_updates_maasserver_dnsresource_update",
+        "dnsresource_sys_dns_updates_maasserver_dnsresource_delete",
+        "dnsresource_ip_addresses_sys_dns_updates_dns_ip_insert",
+        "dnsresource_ip_addresses_sys_dns_updates_dns_ip_delete",
         "domain_sys_dns_domain_delete",
         "domain_sys_dns_domain_insert",
         "domain_sys_dns_domain_update",
+        "domain_sys_dns_updates_maasserver_domain_delete",
+        "domain_sys_dns_updates_maasserver_domain_insert",
+        "domain_sys_dns_updates_maasserver_domain_update",
         "interface_ip_addresses_sys_dns_nic_ip_link",
         "interface_ip_addresses_sys_dns_nic_ip_unlink",
         "interface_sys_dhcp_interface_update",
@@ -99,6 +109,9 @@ class TestTriggersUsed(MAASServerTestCase):
         "staticipaddress_sys_dhcp_staticipaddress_insert",
         "staticipaddress_sys_dhcp_staticipaddress_update",
         "staticipaddress_sys_dns_staticipaddress_update",
+        "subnet_sys_dns_updates_maasserver_subnet_delete",
+        "subnet_sys_dns_updates_maasserver_subnet_insert",
+        "subnet_sys_dns_updates_maasserver_subnet_update",
         "subnet_sys_dhcp_subnet_delete",
         "subnet_sys_dhcp_subnet_update",
         "subnet_sys_dns_subnet_delete",
diff --git a/src/maasserver/triggers/tests/test_system.py b/src/maasserver/triggers/tests/test_system.py
index 7d646ae..d5222b4 100644
--- a/src/maasserver/triggers/tests/test_system.py
+++ b/src/maasserver/triggers/tests/test_system.py
@@ -5,13 +5,26 @@
 from contextlib import closing
 
 from django.db import connection
+from twisted.internet.defer import inlineCallbacks
 
 from maasserver.models.dnspublication import zone_serial
-from maasserver.testing.testcase import MAASServerTestCase
+from maasserver.testing.factory import factory
+from maasserver.testing.testcase import (
+    MAASServerTestCase,
+    MAASTransactionServerTestCase,
+)
 from maasserver.triggers.system import register_system_triggers
+from maasserver.triggers.testing import (
+    NotifyHelperMixin,
+    TransactionalHelpersMixin,
+)
 from maasserver.utils.orm import psql_array
+from maasserver.utils.threads import deferToDatabase
+from maastesting.crochet import wait_for
 from maastesting.matchers import MockCalledOnceWith
 
+wait_for_reactor = wait_for()
+
 
 class TestTriggers(MAASServerTestCase):
     def test_register_system_triggers(self):
@@ -88,3 +101,269 @@ class TestTriggers(MAASServerTestCase):
         mock_create = self.patch(zone_serial, "create_if_not_exists")
         register_system_triggers()
         self.assertThat(mock_create, MockCalledOnceWith())
+
+
+class TestSysDNSUpdates(
+    MAASTransactionServerTestCase, TransactionalHelpersMixin, NotifyHelperMixin
+):
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_dnsresource_ip_addresses_insert(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        domain = yield deferToDatabase(self.create_domain)
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_dnsresource_ip_addresses",
+            "sys_dns_updates",
+            ops=("insert",),
+            trigger="sys_dns_updates_dns_ip_insert",
+        )
+        self.start_reading()
+        try:
+            static_ip = yield deferToDatabase(self.create_staticipaddress)
+            rec = yield deferToDatabase(
+                self.create_dnsresource,
+                params={"domain": domain, "ip_addresses": [static_ip]},
+            )
+            yield self.get_notify(
+                "sys_dns_updates"
+            )  # ignore RELOAD from domain creation
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(
+                msg,
+                f"INSERT {domain.name} {rec.name} A {domain.ttl if domain.ttl else 0} {static_ip.ip}",
+            )
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_dnsresource_ip_addresses_delete(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        domain = yield deferToDatabase(self.create_domain)
+        static_ip = yield deferToDatabase(self.create_staticipaddress)
+        rec = yield deferToDatabase(
+            self.create_dnsresource,
+            params={"domain": domain, "ip_addresses": [static_ip]},
+        )
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_dnsresource_ip_addresses",
+            "sys_dns_updates",
+            ops=("delete",),
+            trigger="sys_dns_updates_dns_ip_delete",
+        )
+        self.start_reading()
+        try:
+            yield deferToDatabase(rec.delete)
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(
+                msg, f"DELETE {domain.name} {rec.name} A {static_ip.ip}"
+            )
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_dnsresource_update(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        domain = yield deferToDatabase(self.create_domain)
+        static_ip = yield deferToDatabase(self.create_staticipaddress)
+        rec = yield deferToDatabase(
+            self.create_dnsresource,
+            params={"domain": domain, "ip_addresses": [static_ip]},
+        )
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_dnsresource",
+            "sys_dns_updates",
+            ops=("update",),
+        )
+        self.start_reading()
+        try:
+            rec.address_ttl = 30
+            yield deferToDatabase(rec.save)
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(
+                msg, f"UPDATE {domain.name} {rec.name} A 30 {static_ip.ip}"
+            )
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_dnsresource_delete(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        domain = yield deferToDatabase(self.create_domain)
+        static_ip = yield deferToDatabase(self.create_staticipaddress)
+        rec = yield deferToDatabase(
+            self.create_dnsresource,
+            params={"domain": domain, "ip_addresses": [static_ip]},
+        )
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_dnsresource",
+            "sys_dns_updates",
+            ops=("delete",),
+        )
+        self.start_reading()
+        try:
+            yield deferToDatabase(rec.delete)
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(
+                msg, f"DELETE {domain.name} {rec.name} A {static_ip.ip}"
+            )
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_dnsdata_update(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        domain = yield deferToDatabase(self.create_domain)
+        rec = yield deferToDatabase(
+            self.create_dnsresource, params={"domain": domain}
+        )
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_dnsdata",
+            "sys_dns_updates",
+            ops=("update",),
+        )
+        self.start_reading()
+        try:
+            dnsdata = yield deferToDatabase(
+                self.create_dnsdata,
+                params={
+                    "dnsresource": rec,
+                    "rrtype": "TXT",
+                    "rrdata": factory.make_name(),
+                },
+            )
+            dnsdata.rrdata = factory.make_name()
+            yield deferToDatabase(dnsdata.save)
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(
+                msg,
+                f"UPDATE-DATA {dnsdata.id}",
+            )
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_dnsdata_insert(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        domain = yield deferToDatabase(self.create_domain)
+        rec = yield deferToDatabase(
+            self.create_dnsresource, {"domain": domain}
+        )
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_dnsdata",
+            "sys_dns_updates",
+            ops=("insert",),
+        )
+        self.start_reading()
+        try:
+            dnsdata = yield deferToDatabase(
+                self.create_dnsdata,
+                params={
+                    "dnsresource": rec,
+                    "rrtype": "TXT",
+                    "rrdata": factory.make_name(),
+                },
+            )
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(
+                msg,
+                f"INSERT-DATA {dnsdata.id}",
+            )
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_dnsdata_delete(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        domain = yield deferToDatabase(self.create_domain)
+        rec = yield deferToDatabase(
+            self.create_dnsresource, params={"domain": domain}
+        )
+        dnsdata = yield deferToDatabase(
+            self.create_dnsdata,
+            params={
+                "dnsresource": rec,
+                "rrtype": "TXT",
+                "rrdata": factory.make_name(),
+            },
+        )
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_dnsdata",
+            "sys_dns_updates",
+            ops=("delete",),
+        )
+        self.start_reading()
+        try:
+            yield deferToDatabase(dnsdata.delete)
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(
+                msg, f"DELETE {domain.name} {rec.name} {dnsdata.rrtype}"
+            )
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_domain_reload(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_domain",
+            "sys_dns_updates",
+            ops=("insert",),
+        )
+        self.start_reading()
+        try:
+            yield deferToDatabase(self.create_domain)
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(msg, "RELOAD")
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test_dns_dynamic_update_subnet_reload(self):
+        listener = self.make_listener_without_delay()
+        yield self.set_service(listener)
+        yield deferToDatabase(
+            self.register_trigger,
+            "maasserver_subnet",
+            "sys_dns_updates",
+            ops=("insert",),
+        )
+        self.start_reading()
+        try:
+            yield deferToDatabase(self.create_subnet)
+            msg = yield self.get_notify("sys_dns_updates")
+            self.assertEqual(msg, "RELOAD")
+        finally:
+            self.stop_reading()
+            yield self.postgres_listener_service.stopService()
diff --git a/src/provisioningserver/dns/actions.py b/src/provisioningserver/dns/actions.py
index 7b8b1ca..bafbac9 100644
--- a/src/provisioningserver/dns/actions.py
+++ b/src/provisioningserver/dns/actions.py
@@ -10,14 +10,18 @@ from time import sleep
 from provisioningserver.dns.config import (
     DNSConfig,
     execute_rndc_command,
+    get_nsupdate_key_path,
     set_up_options_conf,
 )
 from provisioningserver.logger import get_maas_logger
-from provisioningserver.utils.shell import ExternalProcessError
+from provisioningserver.utils.shell import ExternalProcessError, run_command
 
 maaslog = get_maas_logger("dns")
 
 
+MAAS_NSUPDATE_HOST = "localhost"
+
+
 def bind_reconfigure():
     """Ask BIND to reload its configuration and *new* zone files.
 
@@ -135,3 +139,50 @@ def bind_write_zones(zones):
     """
     for zone in zones:
         zone.write_config()
+
+
+class NSUpdateCommand:
+    executable = "nsupdate"
+
+    def __init__(self, zone, updates, **kwargs):
+        self._zone = zone
+        self._updates = updates
+        self._serial = kwargs.get("serial")
+        self._zone_ttl = kwargs["ttl"]
+
+    def _format_update(self, update):
+        if update.operation == "DELETE":
+            if update.answer:
+                return f"update delete {update.name} {update.rectype} {update.answer}"
+            return f"update delete {update.name} {update.rectype}"
+        ttl = update.ttl
+        if ttl is None:
+            ttl = self._zone_ttl
+        return (
+            f"update add {update.name} {ttl} {update.rectype} {update.answer}"
+        )
+
+    def update(self, server_address=MAAS_NSUPDATE_HOST):
+        stdin = [f"zone {self._zone}"] + [
+            self._format_update(update) for update in self._updates
+        ]
+        if server_address:
+            stdin = [f"server {server_address}"] + stdin
+
+        if self._serial:
+            stdin.append(
+                f"update add {self._zone} {self._zone_ttl} SOA {self._zone}. nobody.example.com. {self._serial} 600 1800 604800 {self._zone_ttl}"
+            )
+
+        stdin.append("send\n")
+
+        cmd = [self.executable, "-k", get_nsupdate_key_path()]
+        if len(self._updates) > 1:
+            cmd.append("-v")  # use TCP for bulk payloads
+
+        try:
+            run_command(*cmd, stdin="\n".join(stdin).encode("ascii"))
+        except CalledProcessError as exc:
+            maaslog.error(f"dynamic update of DNS failed: {exc}")
+            ExternalProcessError.upgrade(exc)
+            raise
diff --git a/src/provisioningserver/dns/commands/setup_dns.py b/src/provisioningserver/dns/commands/setup_dns.py
index 65b1166..cdf4e38 100644
--- a/src/provisioningserver/dns/commands/setup_dns.py
+++ b/src/provisioningserver/dns/commands/setup_dns.py
@@ -16,8 +16,10 @@ from textwrap import dedent
 
 from provisioningserver.dns.config import (
     DNSConfig,
+    set_up_nsupdate_key,
     set_up_options_conf,
     set_up_rndc,
+    set_up_zone_file_dir,
 )
 
 
@@ -49,6 +51,8 @@ def run(args, stdout=sys.stdout, stderr=sys.stderr):
     :param stdout: Standard output stream to write to.
     :param stderr: Standard error stream to write to.
     """
+    set_up_nsupdate_key()
+    set_up_zone_file_dir()
     set_up_rndc()
     set_up_options_conf(overwrite=not args.no_clobber)
     config = DNSConfig()
diff --git a/src/provisioningserver/dns/commands/tests/test_setup_dns.py b/src/provisioningserver/dns/commands/tests/test_setup_dns.py
index dc73242..e720cff 100644
--- a/src/provisioningserver/dns/commands/tests/test_setup_dns.py
+++ b/src/provisioningserver/dns/commands/tests/test_setup_dns.py
@@ -17,7 +17,10 @@ from provisioningserver.dns.config import (
     MAAS_NAMED_CONF_NAME,
     MAAS_RNDC_CONF_NAME,
 )
-from provisioningserver.dns.testing import patch_dns_config_path
+from provisioningserver.dns.testing import (
+    patch_dns_config_path,
+    patch_zone_file_config_path,
+)
 
 
 class TestSetupCommand(MAASTestCase):
@@ -35,6 +38,7 @@ class TestSetupCommand(MAASTestCase):
     def test_writes_configuration(self):
         dns_conf_dir = self.make_dir()
         patch_dns_config_path(self, dns_conf_dir)
+        patch_zone_file_config_path(self, dns_conf_dir)
         self.run_command()
         named_config = os.path.join(dns_conf_dir, MAAS_NAMED_CONF_NAME)
         rndc_conf_path = os.path.join(dns_conf_dir, MAAS_RNDC_CONF_NAME)
@@ -43,6 +47,7 @@ class TestSetupCommand(MAASTestCase):
     def test_does_not_overwrite_config(self):
         dns_conf_dir = self.make_dir()
         patch_dns_config_path(self, dns_conf_dir)
+        patch_zone_file_config_path(self, dns_conf_dir)
         random_content = factory.make_string()
         factory.make_file(
             location=dns_conf_dir,
diff --git a/src/provisioningserver/dns/config.py b/src/provisioningserver/dns/config.py
index d5409e8..5673b4a 100644
--- a/src/provisioningserver/dns/config.py
+++ b/src/provisioningserver/dns/config.py
@@ -6,12 +6,17 @@
 
 from collections import namedtuple
 from contextlib import contextmanager
+from dataclasses import dataclass
 from datetime import datetime
 import errno
+import grp
 import os
 import os.path
 import re
 import sys
+from typing import Optional
+
+from netaddr import AddrFormatError, IPAddress
 
 from provisioningserver.logger import get_maas_logger
 from provisioningserver.utils import load_template, locate_config
@@ -26,6 +31,63 @@ MAAS_NAMED_CONF_NAME = "named.conf.maas"
 MAAS_NAMED_CONF_OPTIONS_INSIDE_NAME = "named.conf.options.inside.maas"
 MAAS_NAMED_RNDC_CONF_NAME = "named.conf.rndc.maas"
 MAAS_RNDC_CONF_NAME = "rndc.conf.maas"
+MAAS_NSUPDATE_KEY_NAME = "keys.conf.maas"
+MAAS_ZONE_FILE_DIR = "/var/lib/bind/maas"
+MAAS_ZONE_FILE_GROUP = "bind"
+
+
+@dataclass
+class DynamicDNSUpdate:
+    operation: str
+    name: str
+    zone: str
+    rectype: str
+    ttl: Optional[int] = None
+    subnet: Optional[str] = None  # for reverse updates
+    answer: Optional[str] = None
+
+    @classmethod
+    def _is_ip(cls, answer):
+        if not answer:
+            return False
+        try:
+            IPAddress(answer)
+        except AddrFormatError:
+            return False
+        else:
+            return True
+
+    @classmethod
+    def create_from_trigger(cls, **kwargs):
+        answer = kwargs.get("answer")
+        rectype = kwargs.pop("rectype")
+        if answer:
+            del kwargs["answer"]
+        # the DB trigger is unable to figure out if an IP is v6, so we do it here instead
+        if cls._is_ip(answer):
+            ip = IPAddress(answer)
+            if ip.version == 6:
+                rectype = "AAAA"
+        return cls(answer=answer, rectype=rectype, **kwargs)
+
+    @classmethod
+    def as_reverse_record_update(cls, fwd_update, subnet):
+        if not fwd_update.answer_is_ip:
+            return None
+        ip = IPAddress(fwd_update.answer)
+        return cls(
+            operation=fwd_update.operation,
+            name=ip.reverse_dns,
+            zone=fwd_update.zone,
+            subnet=subnet,
+            ttl=fwd_update.ttl,
+            answer=fwd_update.name,
+            rectype="PTR",
+        )
+
+    @property
+    def answer_is_ip(self):
+        return DynamicDNSUpdate._is_ip(self.answer)
 
 
 def get_dns_config_dir():
@@ -40,6 +102,19 @@ def get_dns_config_dir():
         return setting
 
 
+def get_zone_file_config_dir():
+    """
+    Location of MAAS' zone files, separate from config files
+    so that bind can write to the location as well
+    """
+    setting = os.getenv("MAAS_ZONE_FILE_CONFIG_DIR", MAAS_ZONE_FILE_DIR)
+    if isinstance(setting, bytes):
+        fsenc = sys.getfilesystemencoding()
+        return setting.decode(fsenc)
+    else:
+        return setting
+
+
 def get_bind_config_dir():
     """Location of bind configuration files."""
     setting = os.getenv(
@@ -155,6 +230,30 @@ def get_rndc_conf_path():
     return compose_config_path(MAAS_RNDC_CONF_NAME)
 
 
+def get_nsupdate_key_path():
+    return compose_config_path(MAAS_NSUPDATE_KEY_NAME)
+
+
+def set_up_nsupdate_key():
+    tsig = call_and_check(["tsig-keygen", "-a", "HMAC-SHA512", "maas."])
+    atomic_write(tsig, get_nsupdate_key_path(), overwrite=True, mode=0o644)
+
+
+def set_up_zone_file_dir():
+    p = get_zone_file_config_dir()
+    if not os.path.exists(p):
+        os.mkdir(p)
+
+        uid = os.getuid()
+        gid = 0
+        group = grp.getgrnam(MAAS_ZONE_FILE_GROUP)
+        if group:
+            gid = group.gr_gid
+
+        os.chown(p, uid, gid)
+        os.chmod(p, 0o775)
+
+
 def set_up_rndc():
     """Writes out the two files needed to enable MAAS to use rndc commands:
     MAAS_RNDC_CONF_NAME and MAAS_NAMED_RNDC_CONF_NAME.
@@ -232,12 +331,16 @@ def set_up_options_conf(overwrite=True, **kwargs):
 
 
 def compose_config_path(filename):
-    """Return the full path for a DNS config or zone file."""
+    """Return the full path for a DNS config"""
     return os.path.join(get_dns_config_dir(), filename)
 
 
+def compose_zone_file_config_path(filename):
+    return os.path.join(get_zone_file_config_dir(), filename)
+
+
 def compose_bind_config_path(filename):
-    """Return the full path for a DNS config or zone file."""
+    """Return the full path for a DNS config"""
     return os.path.join(get_bind_config_dir(), filename)
 
 
@@ -309,6 +412,7 @@ class DNSConfig:
             "forwarded_zones": self.forwarded_zones,
             "DNS_CONFIG_DIR": get_dns_config_dir(),
             "named_rndc_conf_path": get_named_rndc_conf_path(),
+            "nsupdate_keys_conf_path": get_nsupdate_key_path(),
             "trusted_networks": trusted_networks,
             "modified": str(datetime.today()),
         }
diff --git a/src/provisioningserver/dns/testing.py b/src/provisioningserver/dns/testing.py
index 2415e0b..4892f7a 100644
--- a/src/provisioningserver/dns/testing.py
+++ b/src/provisioningserver/dns/testing.py
@@ -17,6 +17,16 @@ def patch_dns_config_path(testcase, config_dir=None):
     return config_dir
 
 
+def patch_zone_file_config_path(testcase, config_dir=None):
+    """Set the DNS config dir to a temporary directory, and return its path."""
+    if config_dir is None:
+        config_dir = testcase.make_dir()
+    testcase.useFixture(
+        EnvironmentVariable("MAAS_ZONE_FILE_CONFIG_DIR", config_dir)
+    )
+    return config_dir
+
+
 def patch_dns_rndc_port(testcase, port):
     testcase.useFixture(EnvironmentVariable("MAAS_DNS_RNDC_PORT", "%d" % port))
 
diff --git a/src/provisioningserver/dns/tests/test_actions.py b/src/provisioningserver/dns/tests/test_actions.py
index 72c5b5a..1087bc0 100644
--- a/src/provisioningserver/dns/tests/test_actions.py
+++ b/src/provisioningserver/dns/tests/test_actions.py
@@ -20,11 +20,19 @@ from maastesting.factory import factory
 from maastesting.matchers import MockCalledOnceWith, MockCallsMatch
 from maastesting.testcase import MAASTestCase
 from provisioningserver.dns import actions
+from provisioningserver.dns.actions import (
+    get_nsupdate_key_path,
+    NSUpdateCommand,
+)
 from provisioningserver.dns.config import (
+    DynamicDNSUpdate,
     MAAS_NAMED_CONF_NAME,
     MAAS_NAMED_CONF_OPTIONS_INSIDE_NAME,
 )
-from provisioningserver.dns.testing import patch_dns_config_path
+from provisioningserver.dns.testing import (
+    patch_dns_config_path,
+    patch_zone_file_config_path,
+)
 from provisioningserver.dns.tests.test_zoneconfig import HostnameIPMapping
 from provisioningserver.dns.zoneconfig import (
     DNSForwardZoneConfig,
@@ -208,6 +216,7 @@ class TestConfiguration(MAASTestCase):
         )
 
     def test_bind_write_zones_writes_file(self):
+        zone_file_dir = patch_zone_file_config_path(self)
         domain = factory.make_string()
         network = IPNetwork("192.168.0.3/24")
         dns_ip_list = [factory.pick_ip_in_network(network)]
@@ -229,8 +238,8 @@ class TestConfiguration(MAASTestCase):
         forward_file_name = "zone.%s" % domain
         reverse_file_name = "zone.0.168.192.in-addr.arpa"
         expected_files = [
-            join(self.dns_conf_dir, forward_file_name),
-            join(self.dns_conf_dir, reverse_file_name),
+            join(zone_file_dir, forward_file_name),
+            join(zone_file_dir, reverse_file_name),
         ]
         self.assertThat(expected_files, AllMatch(FileExists()))
 
@@ -270,3 +279,112 @@ class TestConfiguration(MAASTestCase):
         self.assertThat(
             expected_options_file, FileContains(expected_options_content)
         )
+
+
+class TestNSUpdateCommand(MAASTestCase):
+    def test_format_update_deletion(self):
+        domain = factory.make_name()
+        update = DynamicDNSUpdate(
+            operation="DELETE",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="A",
+        )
+        cmd = NSUpdateCommand(
+            domain,
+            [update],
+            serial=random.randint(1, 100),
+            ttl=random.randint(1, 100),
+        )
+        self.assertEqual(
+            f"update delete {update.name} A", cmd._format_update(update)
+        )
+
+    def test_format_update_addition(self):
+        domain = factory.make_name()
+        update = DynamicDNSUpdate(
+            operation="INSERT",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="A",
+            answer=factory.make_ip_address(),
+        )
+        ttl = random.randint(1, 100)
+        cmd = NSUpdateCommand(
+            domain, [update], serial=random.randint(1, 100), ttl=ttl
+        )
+        self.assertEqual(
+            f"update add {update.name} {ttl} A {update.answer}",
+            cmd._format_update(update),
+        )
+
+    def test_nsupdate_sends_a_single_update(self):
+        domain = factory.make_name()
+        update = DynamicDNSUpdate(
+            operation="INSERT",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="A",
+            answer=factory.make_ip_address(),
+        )
+        serial = random.randint(1, 100)
+        ttl = random.randint(1, 100)
+        cmd = NSUpdateCommand(domain, [update], serial=serial, ttl=ttl)
+        run_command = self.patch(actions, "run_command")
+        cmd.update()
+        run_command.assert_called_once_with(
+            "nsupdate",
+            "-k",
+            get_nsupdate_key_path(),
+            stdin="\n".join(
+                [
+                    "server localhost",
+                    f"zone {domain}",
+                    f"update add {update.name} {ttl} A {update.answer}",
+                    f"update add {domain} {ttl} SOA {domain}. nobody.example.com. {serial} 600 1800 604800 {ttl}",
+                    "send\n",
+                ]
+            ).encode("ascii"),
+        )
+
+    def test_nsupdate_sends_a_bulk_update(self):
+        domain = factory.make_name()
+        deletions = [
+            DynamicDNSUpdate(
+                operation="DELETE",
+                zone=domain,
+                name=f"{factory.make_name()}.{domain}",
+                rectype="A",
+            )
+            for _ in range(2)
+        ]
+        additions = [
+            DynamicDNSUpdate(
+                operation="INSERT",
+                zone=domain,
+                name=f"{factory.make_name()}.{domain}",
+                rectype="A",
+                answer=factory.make_ip_address(),
+            )
+            for _ in range(2)
+        ]
+        ttl = random.randint(1, 100)
+        cmd = NSUpdateCommand(domain, deletions + additions, ttl=ttl)
+        run_command = self.patch(actions, "run_command")
+        cmd.update()
+        expected_stdin = [
+            "server localhost",
+            f"zone {domain}",
+            f"update delete {deletions[0].name} {deletions[0].rectype}",
+            f"update delete {deletions[1].name} {deletions[1].rectype}",
+            f"update add {additions[0].name} {ttl} {additions[0].rectype} {additions[0].answer}",
+            f"update add {additions[1].name} {ttl} {additions[1].rectype} {additions[1].answer}",
+            "send\n",
+        ]
+        run_command.assert_called_once_with(
+            "nsupdate",
+            "-k",
+            get_nsupdate_key_path(),
+            "-v",
+            stdin="\n".join(expected_stdin).encode("ascii"),
+        )
diff --git a/src/provisioningserver/dns/tests/test_config.py b/src/provisioningserver/dns/tests/test_config.py
index c29c3ca..1e02498 100644
--- a/src/provisioningserver/dns/tests/test_config.py
+++ b/src/provisioningserver/dns/tests/test_config.py
@@ -11,7 +11,7 @@ from textwrap import dedent
 from unittest.mock import Mock, sentinel
 
 from fixtures import EnvironmentVariable
-from netaddr import IPNetwork
+from netaddr import IPAddress, IPNetwork
 from testtools.matchers import (
     AllMatch,
     Contains,
@@ -40,6 +40,7 @@ from provisioningserver.dns.config import (
     DNSConfig,
     DNSConfigDirectoryMissing,
     DNSConfigFail,
+    DynamicDNSUpdate,
     execute_rndc_command,
     extract_suggested_named_conf,
     generate_rndc,
@@ -593,3 +594,75 @@ class TestDNSConfig(MAASTestCase):
                 ),
             ),
         )
+
+
+class TestDynamicDNSUpdate(MAASTestCase):
+    def test_create_from_trigger_v4(self):
+        domain = factory.make_name()
+        update = DynamicDNSUpdate.create_from_trigger(
+            operation="INSERT",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="A",
+            answer=factory.make_ip_address(ipv6=False),
+        )
+        self.assertEqual(update.rectype, "A")
+
+    def test_create_from_trigger_v6(self):
+        domain = factory.make_name()
+        update = DynamicDNSUpdate.create_from_trigger(
+            operation="INSERT",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="A",
+            answer=factory.make_ip_address(ipv6=True),
+        )
+        self.assertEqual(update.rectype, "AAAA")
+
+    def test_answer_is_ip_returns_true_when_answer_is_an_ip(self):
+        domain = factory.make_name()
+        update = DynamicDNSUpdate(
+            operation="INSERT",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="A",
+            answer=factory.make_ip_address(),
+        )
+        self.assertTrue(update.answer_is_ip)
+
+    def test_answer_is_ip_returns_false_when_answer_is_not_an_ip(self):
+        domain = factory.make_name()
+        update = DynamicDNSUpdate(
+            operation="INSERT",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="CNAME",
+            answer=factory.make_name(),
+        )
+        self.assertFalse(update.answer_is_ip)
+
+    def test_as_reverse_record_update(self):
+        domain = factory.make_name()
+        subnet = factory.make_ip4_or_6_network()
+        fwd_update = DynamicDNSUpdate(
+            operation="INSERT",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="A",
+            answer=str(IPAddress(subnet.next())),
+        )
+        expected_rev_update = DynamicDNSUpdate(
+            operation="INSERT",
+            zone=domain,
+            name=IPAddress(fwd_update.answer).reverse_dns,
+            rectype="PTR",
+            ttl=fwd_update.ttl,
+            subnet=str(subnet),
+            answer=fwd_update.name,
+        )
+        rev_update = DynamicDNSUpdate.as_reverse_record_update(
+            fwd_update, str(subnet)
+        )
+        self.assertEqual(expected_rev_update.name, rev_update.name)
+        self.assertEqual(expected_rev_update.rectype, rev_update.rectype)
+        self.assertEqual(expected_rev_update.answer, rev_update.answer)
diff --git a/src/provisioningserver/dns/tests/test_zoneconfig.py b/src/provisioningserver/dns/tests/test_zoneconfig.py
index ba84efb..f97adfd 100644
--- a/src/provisioningserver/dns/tests/test_zoneconfig.py
+++ b/src/provisioningserver/dns/tests/test_zoneconfig.py
@@ -22,8 +22,13 @@ from twisted.python.filepath import FilePath
 from maastesting.factory import factory
 from maastesting.matchers import MockNotCalled
 from maastesting.testcase import MAASTestCase
-from provisioningserver.dns.config import get_dns_config_dir
-from provisioningserver.dns.testing import patch_dns_config_path
+from provisioningserver.dns import actions
+from provisioningserver.dns.config import (
+    DynamicDNSUpdate,
+    get_nsupdate_key_path,
+    get_zone_file_config_dir,
+)
+from provisioningserver.dns.testing import patch_zone_file_config_path
 from provisioningserver.dns.zoneconfig import (
     DNSForwardZoneConfig,
     DNSReverseZoneConfig,
@@ -100,7 +105,7 @@ class TestDNSForwardZoneConfig(MAASTestCase):
         domain = factory.make_name("zone")
         dns_zone_config = DNSForwardZoneConfig(domain)
         self.assertEqual(
-            os.path.join(get_dns_config_dir(), "zone.%s" % domain),
+            os.path.join(get_zone_file_config_dir(), "zone.%s" % domain),
             dns_zone_config.zone_info[0].target_path,
         )
 
@@ -173,7 +178,7 @@ class TestDNSForwardZoneConfig(MAASTestCase):
         )
 
     def test_handles_slash_32_dynamic_range(self):
-        target_dir = patch_dns_config_path(self)
+        target_dir = patch_zone_file_config_path(self)
         domain = factory.make_string()
         network = factory.make_ipv4_network()
         ipv4_hostname = factory.make_name("host")
@@ -222,7 +227,7 @@ class TestDNSForwardZoneConfig(MAASTestCase):
         )
 
     def test_writes_dns_zone_config(self):
-        target_dir = patch_dns_config_path(self)
+        target_dir = patch_zone_file_config_path(self)
         domain = factory.make_string()
         network = factory.make_ipv4_network()
         ipv4_hostname = factory.make_name("host")
@@ -269,7 +274,7 @@ class TestDNSForwardZoneConfig(MAASTestCase):
         )
 
     def test_writes_dns_zone_config_with_NS_record(self):
-        target_dir = patch_dns_config_path(self)
+        target_dir = patch_zone_file_config_path(self)
         addr_ttl = random.randint(10, 100)
         ns_host_name = factory.make_name("ns")
         dns_zone_config = DNSForwardZoneConfig(
@@ -286,7 +291,7 @@ class TestDNSForwardZoneConfig(MAASTestCase):
         )
 
     def test_ignores_generate_directives_for_v6_dynamic_ranges(self):
-        patch_dns_config_path(self)
+        patch_zone_file_config_path(self)
         domain = factory.make_string()
         network = factory.make_ipv4_network()
         ipv4_hostname = factory.make_name("host")
@@ -314,7 +319,7 @@ class TestDNSForwardZoneConfig(MAASTestCase):
         self.assertThat(get_generate_directives, MockNotCalled())
 
     def test_config_file_is_world_readable(self):
-        patch_dns_config_path(self)
+        patch_zone_file_config_path(self)
         dns_zone_config = DNSForwardZoneConfig(
             factory.make_string(), serial=random.randint(1, 100)
         )
@@ -322,6 +327,94 @@ class TestDNSForwardZoneConfig(MAASTestCase):
         filepath = FilePath(dns_zone_config.zone_info[0].target_path)
         self.assertTrue(filepath.getPermissions().other.read)
 
+    def test_zone_file_exists(self):
+        patch_zone_file_config_path(self)
+        domain = factory.make_string()
+        network = factory.make_ipv4_network()
+        ipv4_hostname = factory.make_name("host")
+        ipv4_ip = factory.pick_ip_in_network(network)
+        ipv6_hostname = factory.make_name("host")
+        ipv6_ip = factory.make_ipv6_address()
+        ipv6_network = factory.make_ipv6_network()
+        dynamic_range = IPRange(ipv6_network.first, ipv6_network.last)
+        ttl = random.randint(10, 300)
+        mapping = {
+            ipv4_hostname: HostnameIPMapping(None, ttl, {ipv4_ip}),
+            ipv6_hostname: HostnameIPMapping(None, ttl, {ipv6_ip}),
+        }
+        dns_zone_config = DNSForwardZoneConfig(
+            domain,
+            serial=random.randint(1, 100),
+            mapping=mapping,
+            default_ttl=ttl,
+            dynamic_ranges=[dynamic_range],
+        )
+        self.patch(dns_zone_config, "get_GENERATE_directives")
+        self.assertFalse(
+            dns_zone_config.zone_file_exists(dns_zone_config.zone_info[0])
+        )
+        dns_zone_config.write_config()
+        self.assertTrue(
+            dns_zone_config.zone_file_exists(dns_zone_config.zone_info[0])
+        )
+
+    def test_uses_dynamic_update_when_zone_has_been_configured_once(self):
+        patch_zone_file_config_path(self)
+        domain = factory.make_string()
+        network = factory.make_ipv4_network()
+        ipv4_hostname = factory.make_name("host")
+        ipv4_ip = factory.pick_ip_in_network(network)
+        ipv6_hostname = factory.make_name("host")
+        ipv6_ip = factory.make_ipv6_address()
+        ipv6_network = factory.make_ipv6_network()
+        dynamic_range = IPRange(ipv6_network.first, ipv6_network.last)
+        ttl = random.randint(10, 300)
+        mapping = {
+            ipv4_hostname: HostnameIPMapping(None, ttl, {ipv4_ip}),
+            ipv6_hostname: HostnameIPMapping(None, ttl, {ipv6_ip}),
+        }
+        dns_zone_config = DNSForwardZoneConfig(
+            domain,
+            serial=random.randint(1, 100),
+            mapping=mapping,
+            default_ttl=ttl,
+            dynamic_ranges=[dynamic_range],
+        )
+        self.patch(dns_zone_config, "get_GENERATE_directives")
+        run_command = self.patch(actions, "run_command")
+        dns_zone_config.write_config()
+        update = DynamicDNSUpdate.create_from_trigger(
+            operation="INSERT",
+            zone=domain,
+            name=f"{factory.make_name()}.{domain}",
+            rectype="A",
+            answer=factory.make_ip_address(),
+        )
+        new_dns_zone_config = DNSForwardZoneConfig(
+            domain,
+            serial=random.randint(1, 100),
+            mapping=mapping,
+            default_ttl=ttl,
+            dynamic_ranges=[dynamic_range],
+            dynamic_updates=[update],
+        )
+        new_dns_zone_config.write_config()
+        expected_stdin = "\n".join(
+            [
+                "server localhost",
+                f"zone {domain}",
+                f"update add {update.name} {ttl} {'A' if IPAddress(update.answer).version == 4 else 'AAAA'} {update.answer}",
+                f"update add {domain} {new_dns_zone_config.default_ttl} SOA {domain}. nobody.example.com. {new_dns_zone_config.serial} 600 1800 604800 {new_dns_zone_config.default_ttl}",
+                "send\n",
+            ]
+        )
+        run_command.assert_called_once_with(
+            "nsupdate",
+            "-k",
+            get_nsupdate_key_path(),
+            stdin=expected_stdin.encode("ascii"),
+        )
+
 
 class TestDNSReverseZoneConfig(MAASTestCase):
     """Tests for DNSReverseZoneConfig."""
@@ -340,7 +433,7 @@ class TestDNSReverseZoneConfig(MAASTestCase):
             ),
         )
 
-    def test_computes_dns_config_file_paths(self):
+    def test_computes_zone_file_config_file_paths(self):
         domain = factory.make_name("zone")
         reverse_file_name = [
             "zone.%d.168.192.in-addr.arpa" % i for i in range(4)
@@ -350,11 +443,11 @@ class TestDNSReverseZoneConfig(MAASTestCase):
         )
         for i in range(4):
             self.assertEqual(
-                os.path.join(get_dns_config_dir(), reverse_file_name[i]),
+                os.path.join(get_zone_file_config_dir(), reverse_file_name[i]),
                 dns_zone_config.zone_info[i].target_path,
             )
 
-    def test_computes_dns_config_file_paths_for_small_network(self):
+    def test_computes_zone_file_config_file_paths_for_small_network(self):
         domain = factory.make_name("zone")
         reverse_file_name = "zone.192-27.0.168.192.in-addr.arpa"
         dns_zone_config = DNSReverseZoneConfig(
@@ -362,7 +455,7 @@ class TestDNSReverseZoneConfig(MAASTestCase):
         )
         self.assertEqual(1, len(dns_zone_config.zone_info))
         self.assertEqual(
-            os.path.join(get_dns_config_dir(), reverse_file_name),
+            os.path.join(get_zone_file_config_dir(), reverse_file_name),
             dns_zone_config.zone_info[0].target_path,
         )
 
@@ -620,7 +713,7 @@ class TestDNSReverseZoneConfig(MAASTestCase):
         )
 
     def test_writes_dns_zone_config_with_NS_record(self):
-        target_dir = patch_dns_config_path(self)
+        target_dir = patch_zone_file_config_path(self)
         network = factory.make_ipv4_network()
         ns_host_name = factory.make_name("ns")
         dns_zone_config = DNSReverseZoneConfig(
@@ -637,7 +730,7 @@ class TestDNSReverseZoneConfig(MAASTestCase):
             )
 
     def test_writes_reverse_dns_zone_config(self):
-        target_dir = patch_dns_config_path(self)
+        target_dir = patch_zone_file_config_path(self)
         domain = factory.make_string()
         ns_host_name = factory.make_name("ns")
         network = IPNetwork("192.168.0.1/22")
@@ -676,7 +769,7 @@ class TestDNSReverseZoneConfig(MAASTestCase):
             )
 
     def test_writes_reverse_dns_zone_config_for_small_network(self):
-        target_dir = patch_dns_config_path(self)
+        target_dir = patch_zone_file_config_path(self)
         domain = factory.make_string()
         ns_host_name = factory.make_name("ns")
         network = IPNetwork("192.168.0.1/26")
@@ -710,7 +803,7 @@ class TestDNSReverseZoneConfig(MAASTestCase):
         )
 
     def test_ignores_generate_directives_for_v6_dynamic_ranges(self):
-        patch_dns_config_path(self)
+        patch_zone_file_config_path(self)
         domain = factory.make_string()
         network = IPNetwork("192.168.0.1/22")
         dynamic_network = IPNetwork("%s/64" % factory.make_ipv6_address())
@@ -729,7 +822,7 @@ class TestDNSReverseZoneConfig(MAASTestCase):
         self.assertThat(get_generate_directives, MockNotCalled())
 
     def test_reverse_config_file_is_world_readable(self):
-        patch_dns_config_path(self)
+        patch_zone_file_config_path(self)
         dns_zone_config = DNSReverseZoneConfig(
             factory.make_string(),
             serial=random.randint(1, 100),
@@ -740,6 +833,61 @@ class TestDNSReverseZoneConfig(MAASTestCase):
             filepath = FilePath(tgt)
             self.assertTrue(filepath.getPermissions().other.read)
 
+    def test_dynamic_update_when_zone_file_exists(self):
+        patch_zone_file_config_path(self)
+        domain = factory.make_string()
+        network = IPNetwork("10.0.0.0/24")
+        ip1 = factory.pick_ip_in_network(network)
+        ip2 = factory.pick_ip_in_network(network)
+        hostname1 = f"{factory.make_string()}.{domain}"
+        hostname2 = f"{factory.make_string()}.{domain}"
+        fwd_updates = [
+            DynamicDNSUpdate(
+                operation="INSERT",
+                zone=domain,
+                name=hostname1,
+                rectype="A",
+                answer=ip1,
+            ),
+            DynamicDNSUpdate(
+                operation="INSERT",
+                zone=domain,
+                name=hostname2,
+                rectype="A",
+                answer=ip2,
+            ),
+        ]
+        rev_updates = [
+            DynamicDNSUpdate.as_reverse_record_update(update, str(network))
+            for update in fwd_updates
+        ]
+        zone = DNSReverseZoneConfig(
+            domain,
+            serial=random.randint(1, 100),
+            network=network,
+            dynamic_updates=rev_updates,
+        )
+        run_command = self.patch(actions, "run_command")
+        zone.write_config()
+        zone.write_config()
+        expected_stdin = "\n".join(
+            [
+                "server localhost",
+                "zone 0.0.10.in-addr.arpa",
+                f"update add {IPAddress(ip1).reverse_dns} {zone.default_ttl} PTR {hostname1}",
+                f"update add {IPAddress(ip2).reverse_dns} {zone.default_ttl} PTR {hostname2}",
+                f"update add 0.0.10.in-addr.arpa {zone.default_ttl} SOA 0.0.10.in-addr.arpa. nobody.example.com. {zone.serial} 600 1800 604800 {zone.default_ttl}",
+                "send\n",
+            ]
+        )
+        run_command.assert_called_once_with(
+            "nsupdate",
+            "-k",
+            get_nsupdate_key_path(),
+            "-v",
+            stdin=expected_stdin.encode("ascii"),
+        )
+
 
 class TestDNSReverseZoneConfig_GetGenerateDirectives(MAASTestCase):
     """Tests for `DNSReverseZoneConfig.get_GENERATE_directives()`."""
diff --git a/src/provisioningserver/dns/zoneconfig.py b/src/provisioningserver/dns/zoneconfig.py
index d393818..eb28986 100644
--- a/src/provisioningserver/dns/zoneconfig.py
+++ b/src/provisioningserver/dns/zoneconfig.py
@@ -6,12 +6,14 @@
 
 from datetime import datetime
 from itertools import chain
+import os
 
 from netaddr import IPAddress, IPNetwork, spanning_cidr
 from netaddr.core import AddrFormatError
 
+from provisioningserver.dns.actions import NSUpdateCommand
 from provisioningserver.dns.config import (
-    compose_config_path,
+    compose_zone_file_config_path,
     render_dns_template,
     report_missing_config_dir,
 )
@@ -115,7 +117,9 @@ class DomainInfo:
         self.subnetwork = subnetwork
         self.zone_name = zone_name
         if target_path is None:
-            self.target_path = compose_config_path("zone.%s" % zone_name)
+            self.target_path = compose_zone_file_config_path(
+                "zone.%s" % zone_name
+            )
         else:
             self.target_path = target_path
 
@@ -140,9 +144,12 @@ class DomainConfigBase:
         self.ns_host_name = kwargs.pop("ns_host_name", None)
         self.serial = serial
         self.zone_info = zone_info
-        self.target_base = compose_config_path("zone")
+        self.target_base = compose_zone_file_config_path("zone")
         self.default_ttl = kwargs.pop("default_ttl", 30)
         self.ns_ttl = kwargs.pop("ns_ttl", self.default_ttl)
+        self.requires_reload = False
+        self._dynamic_updates = kwargs.pop("dynamic_updates", [])
+        self.force_config_write = kwargs.pop("force_config_write", False)
 
     def make_parameters(self):
         """Return a dict of the common template parameters."""
@@ -155,6 +162,27 @@ class DomainConfigBase:
             "ns_host_name": self.ns_host_name,
         }
 
+    def zone_file_exists(self, zone_info):
+        try:
+            os.stat(zone_info.target_path)
+        except FileNotFoundError:
+            return False
+        else:
+            return True
+
+    def dynamic_update(self, zone_info):
+        nsupdate = NSUpdateCommand(
+            zone_info.zone_name,
+            [
+                update
+                for update in self._dynamic_updates
+                if update.zone == zone_info.zone_name or update.subnet
+            ],
+            serial=self.serial,
+            ttl=self.default_ttl,
+        )
+        nsupdate.update()
+
     @classmethod
     def write_zone_file(cls, output_file, *parameters):
         """Write a zone file based on the zone file template.
@@ -292,22 +320,29 @@ class DNSForwardZoneConfig(DomainConfigBase):
                     if dynamic_range.version == 4
                 )
             )
-            self.write_zone_file(
-                zi.target_path,
-                self.make_parameters(),
-                {
-                    "mappings": {
-                        "A": self.get_A_mapping(self._mapping, self._ipv4_ttl),
-                        "AAAA": self.get_AAAA_mapping(
-                            self._mapping, self._ipv6_ttl
+            if not self.force_config_write and self.zone_file_exists(zi):
+                if len(self._dynamic_updates) > 0:
+                    self.dynamic_update(zi)
+            else:
+                self.requires_reload = True
+                self.write_zone_file(
+                    zi.target_path,
+                    self.make_parameters(),
+                    {
+                        "mappings": {
+                            "A": self.get_A_mapping(
+                                self._mapping, self._ipv4_ttl
+                            ),
+                            "AAAA": self.get_AAAA_mapping(
+                                self._mapping, self._ipv6_ttl
+                            ),
+                        },
+                        "other_mapping": enumerate_rrset_mapping(
+                            self._other_mapping
                         ),
+                        "generate_directives": {"A": generate_directives},
                     },
-                    "other_mapping": enumerate_rrset_mapping(
-                        self._other_mapping
-                    ),
-                    "generate_directives": {"A": generate_directives},
-                },
-            )
+                )
 
 
 class DNSReverseZoneConfig(DomainConfigBase):
@@ -575,21 +610,28 @@ class DNSReverseZoneConfig(DomainConfigBase):
                     if dynamic_range.version == 4
                 )
             )
-            self.write_zone_file(
-                zi.target_path,
-                self.make_parameters(),
-                {
-                    "mappings": {
-                        "PTR": self.get_PTR_mapping(
-                            self._mapping, zi.subnetwork
-                        )
-                    },
-                    "other_mapping": [],
-                    "generate_directives": {
-                        "PTR": generate_directives,
-                        "CNAME": self.get_rfc2317_GENERATE_directives(
-                            zi.subnetwork, self._rfc2317_ranges, self.domain
-                        ),
+            if not self.force_config_write and self.zone_file_exists(zi):
+                if len(self._dynamic_updates) > 0:
+                    self.dynamic_update(zi)
+            else:
+                self.requires_reload = True
+                self.write_zone_file(
+                    zi.target_path,
+                    self.make_parameters(),
+                    {
+                        "mappings": {
+                            "PTR": self.get_PTR_mapping(
+                                self._mapping, zi.subnetwork
+                            )
+                        },
+                        "other_mapping": [],
+                        "generate_directives": {
+                            "PTR": generate_directives,
+                            "CNAME": self.get_rfc2317_GENERATE_directives(
+                                zi.subnetwork,
+                                self._rfc2317_ranges,
+                                self.domain,
+                            ),
+                        },
                     },
-                },
-            )
+                )
diff --git a/src/provisioningserver/templates/dns/named.conf.template b/src/provisioningserver/templates/dns/named.conf.template
index f0f9841..7217ca8 100644
--- a/src/provisioningserver/templates/dns/named.conf.template
+++ b/src/provisioningserver/templates/dns/named.conf.template
@@ -1,4 +1,5 @@
 include "{{named_rndc_conf_path}}";
+include "{{nsupdate_keys_conf_path}}";
 
 # Authoritative Zone declarations.
 {{for zone in zones}}
@@ -6,6 +7,9 @@ include "{{named_rndc_conf_path}}";
 zone "{{zoneinfo.zone_name}}" {
     type master;
     file "{{zoneinfo.target_path}}";
+    allow-update {
+        key maas.;
+    };
 };
 {{endfor}}
 {{endfor}}