Add Strict Metaschema Validation (#1101)

Improve schema validation.

This adds strict validation of config module definitions at testing
time, with plumbing included for future runtime validation. This
eliminates a class of bugs resulting from schemas that have definitions
that are incorrect, but get interpreted by jsonschema as
"additionalProperties" that are therefore ignored.

- Add strict meta-schema for jsonschema unit test validation
- Separate schema from module metadata structure
- Improve type annotations for various functions and data types

Cleanup:
- Remove unused jsonschema "required" elements 
- Eliminate manual memoization in schema.py:get_schema(),
        reference module.__doc__ directly

25 files changed, 701 insertions, 291 deletions

diff --git a/cloudinit/cmd/clean.py b/cloudinit/cmd/clean.py
index 928a8eea..3502dd56 100644
--- a/cloudinit/cmd/clean.py
+++ b/cloudinit/cmd/clean.py
@@ -11,11 +11,9 @@ import sys
 
 from cloudinit.stages import Init
 from cloudinit.subp import (ProcessExecutionError, subp)
-from cloudinit.util import (del_dir, del_file, get_config_logfiles, is_link)
-
-
-def error(msg):
-    sys.stderr.write("ERROR: " + msg + "\n")
+from cloudinit.util import (
+    del_dir, del_file, get_config_logfiles, is_link, error
+)
 
 
 def get_parser(parser=None):
diff --git a/cloudinit/cmd/cloud_id.py b/cloudinit/cmd/cloud_id.py
index 97608921..0cdc9675 100755
--- a/cloudinit/cmd/cloud_id.py
+++ b/cloudinit/cmd/cloud_id.py
@@ -6,6 +6,7 @@ import argparse
 import json
 import sys
 
+from cloudinit.util import error
 from cloudinit.sources import (
     INSTANCE_JSON_FILE, METADATA_UNKNOWN, canonical_cloud_id)
 
@@ -40,11 +41,6 @@ def get_parser(parser=None):
     return parser
 
 
-def error(msg):
-    sys.stderr.write('ERROR: %s\n' % msg)
-    return 1
-
-
 def handle_args(name, args):
     """Handle calls to 'cloud-id' cli.
 
diff --git a/cloudinit/config/cc_apk_configure.py b/cloudinit/config/cc_apk_configure.py
index 84d7a0b6..d227a58d 100644
--- a/cloudinit/config/cc_apk_configure.py
+++ b/cloudinit/config/cc_apk_configure.py
@@ -12,8 +12,7 @@ from cloudinit import log as logging
 from cloudinit import temp_utils
 from cloudinit import templater
 from cloudinit import util
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit.settings import PER_INSTANCE
 
 LOG = logging.getLogger(__name__)
@@ -56,7 +55,7 @@ REPOSITORIES_TEMPLATE = """\
 
 frequency = PER_INSTANCE
 distros = ['alpine']
-schema = {
+meta = {
     'id': 'cc_apk_configure',
     'name': 'APK Configure',
     'title': 'Configure apk repositories file',
@@ -95,6 +94,9 @@ schema = {
         """),
     ],
     'frequency': frequency,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'apk_repos': {
@@ -171,14 +173,13 @@ schema = {
                     """)
                 }
             },
-            'required': [],
             'minProperties': 1,  # Either preserve_repositories or alpine_repo
             'additionalProperties': False,
         }
     }
 }
 
-__doc__ = get_schema_doc(schema)
+__doc__ = get_meta_doc(meta, schema)
 
 
 def handle(name, cfg, cloud, log, _args):
diff --git a/cloudinit/config/cc_apt_configure.py b/cloudinit/config/cc_apt_configure.py
index 86d0feae..2e844c2c 100644
--- a/cloudinit/config/cc_apt_configure.py
+++ b/cloudinit/config/cc_apt_configure.py
@@ -14,8 +14,7 @@ import re
 import pathlib
 from textwrap import dedent
 
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit import gpg
 from cloudinit import log as logging
 from cloudinit import subp
@@ -75,7 +74,8 @@ mirror_property = {
         }
     }
 }
-schema = {
+
+meta = {
     'id': 'cc_apt_configure',
     'name': 'Apt Configure',
     'title': 'Configure apt for the user',
@@ -155,6 +155,9 @@ schema = {
                       <key data>
                       ------END PGP PUBLIC KEY BLOCK-------""")],
     'frequency': frequency,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'apt': {
@@ -398,7 +401,7 @@ schema = {
     }
 }
 
-__doc__ = get_schema_doc(schema)
+__doc__ = get_meta_doc(meta, schema)
 
 
 # place where apt stores cached repository data
diff --git a/cloudinit/config/cc_bootcmd.py b/cloudinit/config/cc_bootcmd.py
index 246e4497..06f7a26e 100644
--- a/cloudinit/config/cc_bootcmd.py
+++ b/cloudinit/config/cc_bootcmd.py
@@ -12,8 +12,7 @@
 import os
 from textwrap import dedent
 
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit.settings import PER_ALWAYS
 from cloudinit import temp_utils
 from cloudinit import subp
@@ -29,7 +28,7 @@ frequency = PER_ALWAYS
 
 distros = ['all']
 
-schema = {
+meta = {
     'id': 'cc_bootcmd',
     'name': 'Bootcmd',
     'title': 'Run arbitrary commands early in the boot process',
@@ -57,6 +56,9 @@ schema = {
             - [ cloud-init-per, once, mymkfs, mkfs, /dev/vdb ]
     """)],
     'frequency': PER_ALWAYS,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'bootcmd': {
@@ -69,12 +71,11 @@ schema = {
             'additionalItems': False,  # Reject items of non-string non-list
             'additionalProperties': False,
             'minItems': 1,
-            'required': [],
         }
     }
 }
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 
 def handle(name, cfg, cloud, log, _args):
diff --git a/cloudinit/config/cc_chef.py b/cloudinit/config/cc_chef.py
index 7b20222e..ed734d1c 100644
--- a/cloudinit/config/cc_chef.py
+++ b/cloudinit/config/cc_chef.py
@@ -14,8 +14,7 @@ import os
 from textwrap import dedent
 
 from cloudinit import subp
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit import templater
 from cloudinit import temp_utils
 from cloudinit import url_helper
@@ -89,7 +88,8 @@ CHEF_EXEC_DEF_ARGS = tuple(['-d', '-i', '1800', '-s', '20'])
 
 frequency = PER_ALWAYS
 distros = ["all"]
-schema = {
+
+meta = {
     'id': 'cc_chef',
     'name': 'Chef',
     'title': 'module that configures, starts and installs chef',
@@ -126,6 +126,9 @@ schema = {
           ssl_verify_mode: :verify_peer
           validation_name: yourorg-validator""")],
     'frequency': frequency,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'chef': {
@@ -357,7 +360,7 @@ schema = {
     }
 }
 
-__doc__ = get_schema_doc(schema)
+__doc__ = get_meta_doc(meta, schema)
 
 
 def post_run_chef(chef_cfg, log):
diff --git a/cloudinit/config/cc_install_hotplug.py b/cloudinit/config/cc_install_hotplug.py
index da98c409..9b4075cc 100644
--- a/cloudinit/config/cc_install_hotplug.py
+++ b/cloudinit/config/cc_install_hotplug.py
@@ -6,7 +6,7 @@ from textwrap import dedent
 from cloudinit import util
 from cloudinit import subp
 from cloudinit import stages
-from cloudinit.config.schema import get_schema_doc, validate_cloudconfig_schema
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit.distros import ALL_DISTROS
 from cloudinit.event import EventType, EventScope
 from cloudinit.settings import PER_INSTANCE
@@ -15,7 +15,7 @@ from cloudinit.settings import PER_INSTANCE
 frequency = PER_INSTANCE
 distros = [ALL_DISTROS]
 
-schema = {
+meta = {
     "id": "cc_install_hotplug",
     "name": "Install Hotplug",
     "title": "Install hotplug if supported and enabled",
@@ -49,6 +49,9 @@ schema = {
         """),
     ],
     "frequency": frequency,
+}
+
+schema = {
     "type": "object",
     "properties": {
         "updates": {
@@ -81,7 +84,7 @@ schema = {
     }
 }
 
-__doc__ = get_schema_doc(schema)
+__doc__ = get_meta_doc(meta, schema)
 
 
 HOTPLUG_UDEV_PATH = "/etc/udev/rules.d/10-cloud-init-hook-hotplug.rules"
diff --git a/cloudinit/config/cc_locale.py b/cloudinit/config/cc_locale.py
index 4f8b7bf6..7fed9abd 100644
--- a/cloudinit/config/cc_locale.py
+++ b/cloudinit/config/cc_locale.py
@@ -11,13 +11,13 @@
 from textwrap import dedent
 
 from cloudinit import util
-from cloudinit.config.schema import get_schema_doc, validate_cloudconfig_schema
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit.settings import PER_INSTANCE
 
 
 frequency = PER_INSTANCE
 distros = ['all']
-schema = {
+meta = {
     'id': 'cc_locale',
     'name': 'Locale',
     'title': 'Set system locale',
@@ -39,6 +39,9 @@ schema = {
             """),
     ],
     'frequency': frequency,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'locale': {
@@ -57,7 +60,7 @@ schema = {
     },
 }
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 
 def handle(name, cfg, cloud, log, args):
diff --git a/cloudinit/config/cc_ntp.py b/cloudinit/config/cc_ntp.py
index c3aee798..9c085a04 100644
--- a/cloudinit/config/cc_ntp.py
+++ b/cloudinit/config/cc_ntp.py
@@ -16,7 +16,7 @@ from cloudinit import templater
 from cloudinit import type_utils
 from cloudinit import subp
 from cloudinit import util
-from cloudinit.config.schema import get_schema_doc, validate_cloudconfig_schema
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit.settings import PER_INSTANCE
 
 LOG = logging.getLogger(__name__)
@@ -140,7 +140,7 @@ DISTRO_CLIENT_CONFIG = {
 # configuration options before actually attempting to deploy with said
 # configuration.
 
-schema = {
+meta = {
     'id': 'cc_ntp',
     'name': 'NTP',
     'title': 'enable and configure ntp',
@@ -190,6 +190,9 @@ schema = {
             - ntp.ubuntu.com
             - 192.168.23.2""")],
     'frequency': PER_INSTANCE,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'ntp': {
@@ -289,12 +292,10 @@ schema = {
                     },
                     # Don't use REQUIRED_NTP_CONFIG_KEYS to allow for override
                     # of builtin client values.
-                    'required': [],
                     'minProperties': 1,  # If we have config, define something
                     'additionalProperties': False
                 },
             },
-            'required': [],
             'additionalProperties': False
         }
     }
@@ -303,7 +304,7 @@ REQUIRED_NTP_CONFIG_KEYS = frozenset([
     'check_exe', 'confpath', 'packages', 'service_name'])
 
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 
 def distro_ntp_client_configs(distro):
diff --git a/cloudinit/config/cc_resizefs.py b/cloudinit/config/cc_resizefs.py
index 990a6939..00bb7ae7 100644
--- a/cloudinit/config/cc_resizefs.py
+++ b/cloudinit/config/cc_resizefs.py
@@ -13,8 +13,7 @@ import os
 import stat
 from textwrap import dedent
 
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit.settings import PER_ALWAYS
 from cloudinit import subp
 from cloudinit import util
@@ -24,7 +23,7 @@ NOBLOCK = "noblock"
 frequency = PER_ALWAYS
 distros = ['all']
 
-schema = {
+meta = {
     'id': 'cc_resizefs',
     'name': 'Resizefs',
     'title': 'Resize filesystem',
@@ -42,6 +41,9 @@ schema = {
     'examples': [
         'resize_rootfs: false  # disable root filesystem resize operation'],
     'frequency': PER_ALWAYS,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'resize_rootfs': {
@@ -52,7 +54,7 @@ schema = {
     }
 }
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 
 def _resize_btrfs(mount_point, devpth):
diff --git a/cloudinit/config/cc_runcmd.py b/cloudinit/config/cc_runcmd.py
index 15960c7d..2f5e02cb 100644
--- a/cloudinit/config/cc_runcmd.py
+++ b/cloudinit/config/cc_runcmd.py
@@ -8,8 +8,7 @@
 
 """Runcmd: run arbitrary commands at rc.local with output to the console"""
 
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit.distros import ALL_DISTROS
 from cloudinit.settings import PER_INSTANCE
 from cloudinit import util
@@ -26,7 +25,7 @@ from textwrap import dedent
 
 distros = [ALL_DISTROS]
 
-schema = {
+meta = {
     'id': 'cc_runcmd',
     'name': 'Runcmd',
     'title': 'Run arbitrary commands',
@@ -58,6 +57,9 @@ schema = {
             - [ wget, "http://example.org", -O, /tmp/index.html ]
     """)],
     'frequency': PER_INSTANCE,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'runcmd': {
@@ -71,12 +73,11 @@ schema = {
             'additionalItems': False,  # Reject items of non-string non-list
             'additionalProperties': False,
             'minItems': 1,
-            'required': [],
         }
     }
 }
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 
 def handle(name, cfg, cloud, log, _args):
diff --git a/cloudinit/config/cc_snap.py b/cloudinit/config/cc_snap.py
index 20ed7d2f..21f30b57 100644
--- a/cloudinit/config/cc_snap.py
+++ b/cloudinit/config/cc_snap.py
@@ -8,8 +8,7 @@ import sys
 from textwrap import dedent
 
 from cloudinit import log as logging
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit.settings import PER_INSTANCE
 from cloudinit.subp import prepend_base_command
 from cloudinit import subp
@@ -21,7 +20,7 @@ frequency = PER_INSTANCE
 
 LOG = logging.getLogger(__name__)
 
-schema = {
+meta = {
     'id': 'cc_snap',
     'name': 'Snap',
     'title': 'Install, configure and manage snapd and snap packages',
@@ -103,6 +102,9 @@ schema = {
                     signed_assertion_blob_here
     """)],
     'frequency': PER_INSTANCE,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'snap': {
@@ -139,13 +141,12 @@ schema = {
                 }
             },
             'additionalProperties': False,  # Reject keys not in schema
-            'required': [],
             'minProperties': 1
         }
     }
 }
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 SNAP_CMD = "snap"
 ASSERTIONS_FILE = "/var/lib/cloud/instance/snapd.assertions"
diff --git a/cloudinit/config/cc_ubuntu_advantage.py b/cloudinit/config/cc_ubuntu_advantage.py
index d61dc655..831a92a2 100644
--- a/cloudinit/config/cc_ubuntu_advantage.py
+++ b/cloudinit/config/cc_ubuntu_advantage.py
@@ -4,8 +4,7 @@
 
 from textwrap import dedent
 
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit import log as logging
 from cloudinit.settings import PER_INSTANCE
 from cloudinit import subp
@@ -16,7 +15,7 @@ UA_URL = 'https://ubuntu.com/advantage'
 
 distros = ['ubuntu']
 
-schema = {
+meta = {
     'id': 'cc_ubuntu_advantage',
     'name': 'Ubuntu Advantage',
     'title': 'Configure Ubuntu Advantage support services',
@@ -61,6 +60,9 @@ schema = {
           - fips
         """)],
     'frequency': PER_INSTANCE,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'ubuntu_advantage': {
@@ -82,7 +84,7 @@ schema = {
     }
 }
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 LOG = logging.getLogger(__name__)
 
diff --git a/cloudinit/config/cc_ubuntu_drivers.py b/cloudinit/config/cc_ubuntu_drivers.py
index 2d1d2b32..7f617efe 100644
--- a/cloudinit/config/cc_ubuntu_drivers.py
+++ b/cloudinit/config/cc_ubuntu_drivers.py
@@ -5,8 +5,7 @@
 import os
 from textwrap import dedent
 
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit import log as logging
 from cloudinit.settings import PER_INSTANCE
 from cloudinit import subp
@@ -18,7 +17,7 @@ LOG = logging.getLogger(__name__)
 
 frequency = PER_INSTANCE
 distros = ['ubuntu']
-schema = {
+meta = {
     'id': 'cc_ubuntu_drivers',
     'name': 'Ubuntu Drivers',
     'title': 'Interact with third party drivers in Ubuntu.',
@@ -32,6 +31,9 @@ schema = {
             license-accepted: true
         """)],
     'frequency': frequency,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'drivers': {
@@ -64,7 +66,7 @@ schema = {
 OLD_UBUNTU_DRIVERS_STDERR_NEEDLE = (
     "ubuntu-drivers: error: argument <command>: invalid choice: 'install'")
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 
 # Use a debconf template to configure a global debconf variable
diff --git a/cloudinit/config/cc_write_files.py b/cloudinit/config/cc_write_files.py
index 41c75fa2..55f8c684 100644
--- a/cloudinit/config/cc_write_files.py
+++ b/cloudinit/config/cc_write_files.py
@@ -10,8 +10,7 @@ import base64
 import os
 from textwrap import dedent
 
-from cloudinit.config.schema import (
-    get_schema_doc, validate_cloudconfig_schema)
+from cloudinit.config.schema import get_meta_doc, validate_cloudconfig_schema
 from cloudinit import log as logging
 from cloudinit.settings import PER_INSTANCE
 from cloudinit import util
@@ -38,7 +37,7 @@ supported_encoding_types = [
     'gz', 'gzip', 'gz+base64', 'gzip+base64', 'gz+b64', 'gzip+b64', 'b64',
     'base64']
 
-schema = {
+meta = {
     'id': 'cc_write_files',
     'name': 'Write Files',
     'title': 'write arbitrary files',
@@ -111,6 +110,9 @@ schema = {
           defer: true
         """)],
     'frequency': frequency,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'write_files': {
@@ -187,7 +189,7 @@ schema = {
     }
 }
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 
 def handle(name, cfg, _cloud, log, _args):
diff --git a/cloudinit/config/cc_write_files_deferred.py b/cloudinit/config/cc_write_files_deferred.py
index 0c75aa22..4fc8659c 100644
--- a/cloudinit/config/cc_write_files_deferred.py
+++ b/cloudinit/config/cc_write_files_deferred.py
@@ -4,34 +4,31 @@
 
 """Defer writing certain files"""
 
-from textwrap import dedent
-
 from cloudinit.config.schema import validate_cloudconfig_schema
 from cloudinit import util
 from cloudinit.config.cc_write_files import (
     schema as write_files_schema, write_files, DEFAULT_DEFER)
 
+# meta is not used in this module, but it remains as code documentation
+#
+# id: cc_write_files_deferred'
+# name: 'Write Deferred Files
+# distros: ['all'],
+# frequency: PER_INSTANCE,
+# title:
+#    write certain files, whose creation as been deferred, during
+#    final stage
+# description:
+#    This module is based on `'Write Files' <write-files>`__, and
+#    will handle all files from the write_files list, that have been
+#    marked as deferred and thus are not being processed by the
+#    write-files module.
+#
+#    *Please note that his module is not exposed to the user through
+#    its own dedicated top-level directive.*
+
+schema = write_files_schema
 
-schema = util.mergemanydict([
-    {
-        'id': 'cc_write_files_deferred',
-        'name': 'Write Deferred Files',
-        'title': dedent("""\
-            write certain files, whose creation as been deferred, during
-            final stage
-        """),
-        'description': dedent("""\
-            This module is based on `'Write Files' <write-files>`__, and
-            will handle all files from the write_files list, that have been
-            marked as deferred and thus are not being processed by the
-            write-files module.
-
-            *Please note that his module is not exposed to the user through
-            its own dedicated top-level directive.*
-        """)
-    },
-    write_files_schema
-])
 
 # Not exposed, because related modules should document this behaviour
 __doc__ = None
diff --git a/cloudinit/config/cc_zypper_add_repo.py b/cloudinit/config/cc_zypper_add_repo.py
index 05855b0c..bf1638fb 100644
--- a/cloudinit/config/cc_zypper_add_repo.py
+++ b/cloudinit/config/cc_zypper_add_repo.py
@@ -9,14 +9,14 @@ import configobj
 import os
 from textwrap import dedent
 
-from cloudinit.config.schema import get_schema_doc
+from cloudinit.config.schema import get_meta_doc
 from cloudinit import log as logging
 from cloudinit.settings import PER_ALWAYS
 from cloudinit import util
 
 distros = ['opensuse', 'sles']
 
-schema = {
+meta = {
     'id': 'cc_zypper_add_repo',
     'name': 'ZypperAddRepo',
     'title': 'Configure zypper behavior and add zypper repositories',
@@ -51,6 +51,9 @@ schema = {
             # any setting in /etc/zypp/zypp.conf
     """)],
     'frequency': PER_ALWAYS,
+}
+
+schema = {
     'type': 'object',
     'properties': {
         'zypper': {
@@ -86,14 +89,13 @@ schema = {
                         /etc/zypp/zypp.conf'""")
                 }
             },
-            'required': [],
             'minProperties': 1,  # Either config or repo must be provided
             'additionalProperties': False,  # only repos and config allowed
         }
     }
 }
 
-__doc__ = get_schema_doc(schema)  # Supplement python help()
+__doc__ = get_meta_doc(meta, schema)  # Supplement python help()
 
 LOG = logging.getLogger(__name__)
 
diff --git a/cloudinit/config/schema.py b/cloudinit/config/schema.py
index 456bab2c..d32b7c01 100644
--- a/cloudinit/config/schema.py
+++ b/cloudinit/config/schema.py
@@ -3,19 +3,22 @@
 
 from cloudinit.cmd.devel import read_cfg_paths
 from cloudinit import importer
-from cloudinit.util import find_modules, load_file
+from cloudinit.importer import MetaSchema
+from cloudinit.util import find_modules, load_file, error
 
 import argparse
 from collections import defaultdict
 from copy import deepcopy
+from functools import partial
 import logging
 import os
 import re
 import sys
 import yaml
 
+error = partial(error, sys_exit=True)
+
 _YAML_MAP = {True: 'true', False: 'false', None: 'null'}
-SCHEMA_UNDEFINED = b'UNDEFINED'
 CLOUD_CONFIG_HEADER = b'#cloud-config'
 SCHEMA_DOC_TMPL = """
 {name}
@@ -34,7 +37,7 @@ SCHEMA_DOC_TMPL = """
 {property_doc}
 {examples}
 """
-SCHEMA_PROPERTY_TMPL = '{prefix}**{prop_name}:** ({type}) {description}'
+SCHEMA_PROPERTY_TMPL = "{prefix}**{prop_name}:** ({prop_type}) {description}"
 SCHEMA_LIST_ITEM_TMPL = (
     '{prefix}Each item in **{prop_name}** list supports the following keys:')
 SCHEMA_EXAMPLES_HEADER = '\n**Examples**::\n\n'
@@ -72,45 +75,102 @@ def is_schema_byte_string(checker, instance):
             isinstance(instance, (bytes,)))
 
 
-def validate_cloudconfig_schema(config, schema, strict=False):
-    """Validate provided config meets the schema definition.
+def get_jsonschema_validator():
+    """Get metaschema validator and format checker
 
-    @param config: Dict of cloud configuration settings validated against
-        schema.
-    @param schema: jsonschema dict describing the supported schema definition
-       for the cloud config module (config.cc_*).
-    @param strict: Boolean, when True raise SchemaValidationErrors instead of
-       logging warnings.
+    Older versions of jsonschema require some compatibility changes.
 
-    @raises: SchemaValidationError when provided config does not validate
-        against the provided schema.
+    @returns: Tuple: (jsonschema.Validator, FormatChecker)
+    @raises: ImportError when jsonschema is not present
     """
-    try:
-        from jsonschema import Draft4Validator, FormatChecker
-        from jsonschema.validators import create, extend
-    except ImportError:
-        logging.debug(
-            'Ignoring schema validation. python-jsonschema is not present')
-        return
+    from jsonschema import Draft4Validator, FormatChecker
+    from jsonschema.validators import create
 
     # Allow for bytes to be presented as an acceptable valid value for string
     # type jsonschema attributes in cloud-init's schema.
     # This allows #cloud-config to provide valid yaml "content: !!binary | ..."
+
+    strict_metaschema = deepcopy(Draft4Validator.META_SCHEMA)
+    strict_metaschema['additionalProperties'] = False
     if hasattr(Draft4Validator, 'TYPE_CHECKER'):  # jsonschema 3.0+
         type_checker = Draft4Validator.TYPE_CHECKER.redefine(
             'string', is_schema_byte_string)
-        cloudinitValidator = extend(Draft4Validator, type_checker=type_checker)
+        cloudinitValidator = create(
+            meta_schema=strict_metaschema,
+            validators=Draft4Validator.VALIDATORS,
+            version="draft4",
+            type_checker=type_checker)
     else:  # jsonschema 2.6 workaround
         types = Draft4Validator.DEFAULT_TYPES
-        # Allow bytes as well as string (and disable a spurious
-        # unsupported-assignment-operation pylint warning which appears because
-        # this code path isn't written against the latest jsonschema).
+        # Allow bytes as well as string (and disable a spurious unsupported
+        # assignment-operation pylint warning which appears because this
+        # code path isn't written against the latest jsonschema).
         types['string'] = (str, bytes)  # pylint: disable=E1137
         cloudinitValidator = create(
-            meta_schema=Draft4Validator.META_SCHEMA,
+            meta_schema=strict_metaschema,
             validators=Draft4Validator.VALIDATORS,
             version="draft4",
             default_types=types)
+    return (cloudinitValidator, FormatChecker)
+
+
+def validate_cloudconfig_metaschema(validator, schema: dict, throw=True):
+    """Validate provided schema meets the metaschema definition. Return strict
+    Validator and FormatChecker for use in validation
+    @param validator: Draft4Validator instance used to validate the schema
+    @param schema: schema to validate
+    @param throw: Sometimes the validator and checker are required, even if
+        the schema is invalid. Toggle for whether to raise
+        SchemaValidationError or log warnings.
+
+    @raises: ImportError when jsonschema is not present
+    @raises: SchemaValidationError when the schema is invalid
+    """
+
+    from jsonschema.exceptions import SchemaError
+
+    try:
+        validator.check_schema(schema)
+    except SchemaError as err:
+        # Raise SchemaValidationError to avoid jsonschema imports at call
+        # sites
+        if throw:
+            raise SchemaValidationError(
+                schema_errors=(
+                    ('.'.join([str(p) for p in err.path]), err.message),
+                )
+            ) from err
+        logging.warning(
+            "Meta-schema validation failed, attempting to validate config "
+            "anyway: %s", err)
+
+
+def validate_cloudconfig_schema(
+    config: dict, schema: dict, strict=False, strict_metaschema=False
+):
+    """Validate provided config meets the schema definition.
+
+    @param config: Dict of cloud configuration settings validated against
+        schema. Ignored if strict_metaschema=True
+    @param schema: jsonschema dict describing the supported schema definition
+       for the cloud config module (config.cc_*).
+    @param strict: Boolean, when True raise SchemaValidationErrors instead of
+       logging warnings.
+    @param strict_metaschema: Boolean, when True validates schema using strict
+       metaschema definition at runtime (currently unused)
+
+    @raises: SchemaValidationError when provided config does not validate
+        against the provided schema.
+    """
+    try:
+        (cloudinitValidator, FormatChecker) = get_jsonschema_validator()
+        if strict_metaschema:
+            validate_cloudconfig_metaschema(
+                cloudinitValidator, schema, throw=False)
+    except ImportError:
+        logging.debug("Ignoring schema validation. jsonschema is not present")
+        return
+
     validator = cloudinitValidator(schema, format_checker=FormatChecker())
     errors = ()
     for error in sorted(validator.iter_errors(config), key=lambda e: e.path):
@@ -301,12 +361,15 @@ def _schemapath_for_cloudconfig(config, original_content):
     return schema_line_numbers
 
 
-def _get_property_type(property_dict):
-    """Return a string representing a property type from a given jsonschema."""
-    property_type = property_dict.get('type', SCHEMA_UNDEFINED)
-    if property_type == SCHEMA_UNDEFINED and property_dict.get('enum'):
+def _get_property_type(property_dict: dict) -> str:
+    """Return a string representing a property type from a given
+    jsonschema.
+    """
+    property_type = property_dict.get("type")
+    if property_type is None and property_dict.get("enum"):
         property_type = [
-            str(_YAML_MAP.get(k, k)) for k in property_dict['enum']]
+            str(_YAML_MAP.get(k, k)) for k in property_dict["enum"]
+        ]
     if isinstance(property_type, list):
         property_type = '/'.join(property_type)
     items = property_dict.get('items', {})
@@ -317,12 +380,12 @@ def _get_property_type(property_dict):
             sub_property_type += '/'
         sub_property_type += '(' + _get_property_type(sub_item) + ')'
     if sub_property_type:
-        return '{0} of {1}'.format(property_type, sub_property_type)
-    return property_type
+        return "{0} of {1}".format(property_type, sub_property_type)
+    return property_type or "UNDEFINED"
 
 
-def _parse_description(description, prefix):
-    """Parse description from the schema in a format that we can better
+def _parse_description(description, prefix) -> str:
+    """Parse description from the meta in a format that we can better
     display in our docs. This parser does three things:
 
     - Guarantee that a paragraph will be in a single line
@@ -330,7 +393,7 @@ def _parse_description(description, prefix):
       the first paragraph
     - Proper align lists of items
 
-    @param description: The original description in the schema.
+    @param description: The original description in the meta.
     @param prefix: The number of spaces used to align the current description
     """
     list_paragraph = prefix * 3
@@ -343,20 +406,24 @@ def _parse_description(description, prefix):
     return description
 
 
-def _get_property_doc(schema, prefix='    '):
+def _get_property_doc(schema: dict, prefix="    ") -> str:
     """Return restructured text describing the supported schema properties."""
     new_prefix = prefix + '    '
     properties = []
     for prop_key, prop_config in schema.get('properties', {}).items():
-        # Define prop_name and dscription for SCHEMA_PROPERTY_TMPL
+        # Define prop_name and description for SCHEMA_PROPERTY_TMPL
         description = prop_config.get('description', '')
 
-        properties.append(SCHEMA_PROPERTY_TMPL.format(
-            prefix=prefix,
-            prop_name=prop_key,
-            type=_get_property_type(prop_config),
-            description=_parse_description(description, prefix)))
-        items = prop_config.get('items')
+        # Define prop_name and description for SCHEMA_PROPERTY_TMPL
+        properties.append(
+            SCHEMA_PROPERTY_TMPL.format(
+                prefix=prefix,
+                prop_name=prop_key,
+                description=_parse_description(description, prefix),
+                prop_type=_get_property_type(prop_config),
+            )
+        )
+        items = prop_config.get("items")
         if items:
             if isinstance(items, list):
                 for item in items:
@@ -373,9 +440,9 @@ def _get_property_doc(schema, prefix='    '):
     return '\n\n'.join(properties)
 
 
-def _get_schema_examples(schema, prefix=''):
-    """Return restructured text describing the schema examples if present."""
-    examples = schema.get('examples')
+def _get_examples(meta: MetaSchema) -> str:
+    """Return restructured text describing the meta examples if present."""
+    examples = meta.get("examples")
     if not examples:
         return ''
     rst_content = SCHEMA_EXAMPLES_HEADER
@@ -390,48 +457,111 @@ def _get_schema_examples(schema, prefix=''):
     return rst_content
 
 
-def get_schema_doc(schema):
-    """Return reStructured text rendering the provided jsonschema.
+def get_meta_doc(meta: MetaSchema, schema: dict) -> str:
+    """Return reStructured text rendering the provided metadata.
 
-    @param schema: Dict of jsonschema to render.
-    @raise KeyError: If schema lacks an expected key.
+    @param meta: Dict of metadata to render.
+    @raise KeyError: If metadata lacks an expected key.
     """
-    schema_copy = deepcopy(schema)
-    schema_copy['property_doc'] = _get_property_doc(schema)
-    schema_copy['examples'] = _get_schema_examples(schema)
-    schema_copy['distros'] = ', '.join(schema['distros'])
+
+    if not meta or not schema:
+        raise ValueError("Expected meta and schema")
+    keys = set(meta.keys())
+    expected = set(
+        {
+            "id",
+            "title",
+            "examples",
+            "frequency",
+            "distros",
+            "description",
+            "name",
+        }
+    )
+    error_message = ""
+    if expected - keys:
+        error_message = "Missing expected keys in module meta: {}".format(
+            expected - keys
+        )
+    elif keys - expected:
+        error_message = (
+            "Additional unexpected keys found in module meta: {}".format(
+                keys - expected
+            )
+        )
+    if error_message:
+        raise KeyError(error_message)
+
+    # cast away type annotation
+    meta_copy = dict(deepcopy(meta))
+    meta_copy["property_doc"] = _get_property_doc(schema)
+    meta_copy["examples"] = _get_examples(meta)
+    meta_copy["distros"] = ", ".join(meta["distros"])
     # Need an underbar of the same length as the name
-    schema_copy['title_underbar'] = re.sub(r'.', '-', schema['name'])
-    return SCHEMA_DOC_TMPL.format(**schema_copy)
+    meta_copy["title_underbar"] = re.sub(r".", "-", meta["name"])
+    template = SCHEMA_DOC_TMPL.format(**meta_copy)
+    return template
+
+
+def get_modules() -> dict:
+    configs_dir = os.path.dirname(os.path.abspath(__file__))
+    return find_modules(configs_dir)
+
 
+def load_doc(requested_modules: list) -> str:
+    """Load module docstrings
 
-FULL_SCHEMA = None
+    Docstrings are generated on module load. Reduce, reuse, recycle.
+    """
+    docs = ""
+    all_modules = list(get_modules().values()) + ["all"]
+    invalid_docs = set(requested_modules).difference(set(all_modules))
+    if invalid_docs:
+        error(
+            "Invalid --docs value {}. Must be one of: {}".format(
+                list(invalid_docs), ", ".join(all_modules),
+            )
+        )
+    for mod_name in all_modules:
+        if "all" in requested_modules or mod_name in requested_modules:
+            (mod_locs, _) = importer.find_module(
+                mod_name, ["cloudinit.config"], ["schema"]
+            )
+            if mod_locs:
+                mod = importer.import_module(mod_locs[0])
+                docs += mod.__doc__ or ""
+    return docs
 
 
-def get_schema():
+def get_schema() -> dict:
     """Return jsonschema coalesced from all cc_* cloud-config module."""
-    global FULL_SCHEMA
-    if FULL_SCHEMA:
-        return FULL_SCHEMA
     full_schema = {
-        '$schema': 'http://json-schema.org/draft-04/schema#',
-        'id': 'cloud-config-schema', 'allOf': []}
-
-    configs_dir = os.path.dirname(os.path.abspath(__file__))
-    potential_handlers = find_modules(configs_dir)
-    for (_fname, mod_name) in potential_handlers.items():
-        mod_locs, _looked_locs = importer.find_module(
-            mod_name, ['cloudinit.config'], ['schema'])
+        "$schema": "http://json-schema.org/draft-04/schema#",
+        "id": "cloud-config-schema",
+        "allOf": [],
+    }
+
+    for (_, mod_name) in get_modules().items():
+        (mod_locs, _) = importer.find_module(
+            mod_name, ["cloudinit.config"], ["schema"]
+        )
         if mod_locs:
             mod = importer.import_module(mod_locs[0])
-            full_schema['allOf'].append(mod.schema)
-    FULL_SCHEMA = full_schema
+            full_schema["allOf"].append(mod.schema)
     return full_schema
 
 
-def error(message):
-    print(message, file=sys.stderr)
-    sys.exit(1)
+def get_meta() -> dict:
+    """Return metadata coalesced from all cc_* cloud-config module."""
+    full_meta = dict()
+    for (_, mod_name) in get_modules().items():
+        mod_locs, _ = importer.find_module(
+            mod_name, ["cloudinit.config"], ["meta"]
+        )
+        if mod_locs:
+            mod = importer.import_module(mod_locs[0])
+            full_meta[mod.meta["id"]] = mod.meta
+    return full_meta
 
 
 def get_parser(parser=None):
@@ -474,15 +604,7 @@ def handle_schema_args(name, args):
                 cfg_name = args.config_file
             print("Valid cloud-config:", cfg_name)
     elif args.docs:
-        schema_ids = [subschema['id'] for subschema in full_schema['allOf']]
-        schema_ids += ['all']
-        invalid_docs = set(args.docs).difference(set(schema_ids))
-        if invalid_docs:
-            error('Invalid --docs value {0}. Must be one of: {1}'.format(
-                  list(invalid_docs), ', '.join(schema_ids)))
-        for subschema in full_schema['allOf']:
-            if 'all' in args.docs or subschema['id'] in args.docs:
-                print(get_schema_doc(subschema))
+        print(load_doc(args.docs))
 
 
 def main():
diff --git a/cloudinit/importer.py b/cloudinit/importer.py
index f1194fbe..4e677af3 100644
--- a/cloudinit/importer.py
+++ b/cloudinit/importer.py
@@ -9,6 +9,27 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
 import sys
+import typing
+
+# annotations add value for development, but don't break old versions
+# pyver: 3.5 -> 3.8
+# pylint: disable=E1101
+if sys.version_info >= (3, 8) and hasattr(typing, "TypeDict"):
+    MetaSchema = typing.TypedDict(
+        "MetaSchema",
+        {
+            "name": str,
+            "id": str,
+            "title": str,
+            "description": str,
+            "distros": typing.List[str],
+            "examples": typing.List[str],
+            "frequency": str,
+        },
+    )
+else:
+    MetaSchema = dict
+# pylint: enable=E1101
 
 
 def import_module(module_name):
@@ -16,7 +37,8 @@ def import_module(module_name):
     return sys.modules[module_name]
 
 
-def find_module(base_name, search_paths, required_attrs=None):
+def find_module(base_name: str, search_paths, required_attrs=None) -> tuple:
+    """Finds and imports specified modules"""
     if not required_attrs:
         required_attrs = []
     # NOTE(harlowja): translate the search paths to include the base name.
diff --git a/cloudinit/util.py b/cloudinit/util.py
index 2045a6ab..1b462a38 100644
--- a/cloudinit/util.py
+++ b/cloudinit/util.py
@@ -347,7 +347,7 @@ def extract_usergroup(ug_pair):
     return (u, g)
 
 
-def find_modules(root_dir):
+def find_modules(root_dir) -> dict:
     entries = dict()
     for fname in glob.glob(os.path.join(root_dir, "*.py")):
         if not os.path.isfile(fname):
@@ -2751,4 +2751,19 @@ def get_proc_ppid(pid):
         ppid = int(parts[3])
     return ppid
 
+
+def error(msg, rc=1, fmt='Error:\n{}', sys_exit=False):
+    """
+    Print error to stderr and return or exit
+
+    @param msg: message to print
+    @param rc: return code (default: 1)
+    @param fmt: format string for putting message in (default: 'Error:\n {}')
+    @param sys_exit: exit when called (default: false)
+    """
+    print(fmt.format(msg), file=sys.stderr)
+    if sys_exit:
+        sys.exit(rc)
+    return rc
+
 # vi: ts=4 expandtab
diff --git a/doc/rtd/conf.py b/doc/rtd/conf.py
index 684822c2..4316b5d9 100644
--- a/doc/rtd/conf.py
+++ b/doc/rtd/conf.py
@@ -1,6 +1,8 @@
 import os
 import sys
 
+from cloudinit import version
+
 # If extensions (or modules to document with autodoc) are in another directory,
 # add these directories to sys.path here. If the directory is relative to the
 # documentation root, use os.path.abspath to make it absolute, like shown here.
@@ -9,8 +11,6 @@ sys.path.insert(0, os.path.abspath('../'))
 sys.path.insert(0, os.path.abspath('./'))
 sys.path.insert(0, os.path.abspath('.'))
 
-from cloudinit import version
-from cloudinit.config.schema import get_schema_doc
 
 # Supress warnings for docs that aren't used yet
 # unused_docs = [
@@ -66,12 +66,3 @@ html_theme = 'sphinx_rtd_theme'
 # The name of an image file (relative to this directory) to place at the top
 # of the sidebar.
 html_logo = 'static/logo.png'
-
-def generate_docstring_from_schema(app, what, name, obj, options, lines):
-    """Override module docs from schema when present."""
-    if what == 'module' and hasattr(obj, "schema"):
-        del lines[:]
-        lines.extend(get_schema_doc(obj.schema).split('\n'))
-
-def setup(app):
-    app.connect('autodoc-process-docstring', generate_docstring_from_schema)
diff --git a/tests/unittests/cmd/test_clean.py b/tests/unittests/cmd/test_clean.py
index 81fc930e..3bb0ee9b 100644
--- a/tests/unittests/cmd/test_clean.py
+++ b/tests/unittests/cmd/test_clean.py
@@ -137,7 +137,7 @@ class TestClean(CiTestCase):
                 clean.remove_artifacts, remove_logs=False)
         self.assertEqual(1, retcode)
         self.assertEqual(
-            'ERROR: Could not remove %s/dir1: oops\n' % self.artifact_dir,
+            'Error:\nCould not remove %s/dir1: oops\n' % self.artifact_dir,
             m_stderr.getvalue())
 
     def test_handle_clean_args_reboots(self):
diff --git a/tests/unittests/cmd/test_cloud_id.py b/tests/unittests/cmd/test_cloud_id.py
index 12fc80e8..9a010402 100644
--- a/tests/unittests/cmd/test_cloud_id.py
+++ b/tests/unittests/cmd/test_cloud_id.py
@@ -51,7 +51,7 @@ class TestCloudId(CiTestCase):
                     cloud_id.main()
         self.assertEqual(1, context_manager.exception.code)
         self.assertIn(
-            "ERROR: File not found '%s'" % self.instance_data,
+            "Error:\nFile not found '%s'" % self.instance_data,
             m_stderr.getvalue())
 
     def test_cloud_id_non_json_instance_data(self):
@@ -64,7 +64,7 @@ class TestCloudId(CiTestCase):
                     cloud_id.main()
         self.assertEqual(1, context_manager.exception.code)
         self.assertIn(
-            "ERROR: File '%s' is not valid json." % self.instance_data,
+            "Error:\nFile '%s' is not valid json." % self.instance_data,
             m_stderr.getvalue())
 
     def test_cloud_id_from_cloud_name_in_instance_data(self):
diff --git a/tests/unittests/config/test_schema.py b/tests/unittests/config/test_schema.py
index b01f5eea..f90e0f62 100644
--- a/tests/unittests/config/test_schema.py
+++ b/tests/unittests/config/test_schema.py
@@ -1,13 +1,10 @@
 # This file is part of cloud-init. See LICENSE file for license information.
-import cloudinit
-from cloudinit.config.schema import (
-    CLOUD_CONFIG_HEADER, SchemaValidationError, annotated_cloudconfig_file,
-    get_schema_doc, get_schema, validate_cloudconfig_file,
-    validate_cloudconfig_schema, main)
-from cloudinit.util import write_file
 
-from tests.unittests.helpers import CiTestCase, mock, skipUnlessJsonSchema
 
+import importlib
+import sys
+import inspect
+import logging
 from copy import copy
 import itertools
 import pytest
@@ -15,6 +12,63 @@ from pathlib import Path
 from textwrap import dedent
 from yaml import safe_load
 
+import cloudinit
+from cloudinit.config.schema import (
+    CLOUD_CONFIG_HEADER,
+    SchemaValidationError,
+    annotated_cloudconfig_file,
+    get_meta_doc,
+    get_schema,
+    get_jsonschema_validator,
+    validate_cloudconfig_file,
+    validate_cloudconfig_metaschema,
+    validate_cloudconfig_schema,
+    main,
+    MetaSchema,
+)
+from cloudinit.util import write_file
+from tests.unittests.helpers import CiTestCase, mock, skipUnlessJsonSchema
+
+
+def get_schemas() -> dict:
+    """Return all module schemas
+
+    Assumes that module schemas have the variable name "schema"
+    """
+    return get_module_variable("schema")
+
+
+def get_metas() -> dict:
+    """Return all module metas
+
+    Assumes that module schemas have the variable name "schema"
+    """
+    return get_module_variable("meta")
+
+
+def get_module_variable(var_name) -> dict:
+    """Inspect modules and get variable from module matching var_name"""
+    schemas = {}
+
+    files = list(Path("../../cloudinit/config/").glob("cc_*.py"))
+    modules = [mod.stem for mod in files]
+
+    for module in modules:
+        importlib.import_module("cloudinit.config.{}".format(module))
+
+    for k, v in sys.modules.items():
+        path = Path(k)
+
+        if "cloudinit.config" == path.stem and path.suffix[1:4] == "cc_":
+            module_name = path.suffix[1:]
+            members = inspect.getmembers(v)
+            schemas[module_name] = None
+            for name, value in members:
+                if name == var_name:
+                    schemas[module_name] = value
+                    break
+    return schemas
+
 
 class GetSchemaTest(CiTestCase):
 
@@ -34,25 +88,17 @@ class GetSchemaTest(CiTestCase):
                 'cc_ubuntu_advantage',
                 'cc_ubuntu_drivers',
                 'cc_write_files',
-                'cc_write_files_deferred',
                 'cc_zypper_add_repo',
                 'cc_chef',
                 'cc_install_hotplug',
             ],
-            [subschema['id'] for subschema in schema['allOf']])
-        self.assertEqual('cloud-config-schema', schema['id'])
+            [meta["id"] for meta in get_metas().values() if meta is not None],
+        )
+        self.assertEqual("cloud-config-schema", schema["id"])
         self.assertEqual(
-            'http://json-schema.org/draft-04/schema#',
-            schema['$schema'])
-        # FULL_SCHEMA is updated by the get_schema call
-        from cloudinit.config.schema import FULL_SCHEMA
-        self.assertCountEqual(['id', '$schema', 'allOf'], FULL_SCHEMA.keys())
-
-    def test_get_schema_returns_global_when_set(self):
-        """When FULL_SCHEMA global is already set, get_schema returns it."""
-        m_schema_path = 'cloudinit.config.schema.FULL_SCHEMA'
-        with mock.patch(m_schema_path, {'here': 'iam'}):
-            self.assertEqual({'here': 'iam'}, get_schema())
+            "http://json-schema.org/draft-04/schema#", schema["$schema"]
+        )
+        self.assertCountEqual(["id", "$schema", "allOf"], get_schema().keys())
 
 
 class SchemaValidationErrorTest(CiTestCase):
@@ -93,8 +139,9 @@ class ValidateCloudConfigSchemaTest(CiTestCase):
         with mock.patch.dict('sys.modules', **{'jsonschema': ImportError()}):
             validate_cloudconfig_schema({'p1': -1}, schema, strict=True)
         self.assertIn(
-            'Ignoring schema validation. python-jsonschema is not present',
-            self.logs.getvalue())
+            "Ignoring schema validation. jsonschema is not present",
+            self.logs.getvalue(),
+        )
 
     @skipUnlessJsonSchema()
     def test_validateconfig_schema_strict_raises_errors(self):
@@ -117,14 +164,48 @@ class ValidateCloudConfigSchemaTest(CiTestCase):
             "Cloud config schema errors: p1: '-1' is not a 'email'",
             str(context_mgr.exception))
 
+    @skipUnlessJsonSchema()
+    def test_validateconfig_schema_honors_formats_strict_metaschema(self):
+        """With strict True and strict_metascheam True, ensure errors on format
+        """
+        schema = {"properties": {"p1": {"type": "string", "format": "email"}}}
+        with self.assertRaises(SchemaValidationError) as context_mgr:
+            validate_cloudconfig_schema(
+                {"p1": "-1"}, schema, strict=True, strict_metaschema=True
+            )
+        self.assertEqual(
+            "Cloud config schema errors: p1: '-1' is not a 'email'",
+            str(context_mgr.exception),
+        )
+
+    @skipUnlessJsonSchema()
+    def test_validateconfig_strict_metaschema_do_not_raise_exception(self):
+        """With strict_metaschema=True, do not raise exceptions.
+
+        This flag is currently unused, but is intended for run-time validation.
+        This should warn, but not raise.
+        """
+        schema = {"properties": {"p1": {"types": "string", "format": "email"}}}
+        validate_cloudconfig_schema(
+            {"p1": "-1"}, schema, strict_metaschema=True
+        )
+        assert (
+            "Meta-schema validation failed, attempting to validate config"
+            in self.logs.getvalue()
+        )
+
 
 class TestCloudConfigExamples:
-    schema = get_schema()
+    schema = get_schemas()
+    metas = get_metas()
     params = [
-        (schema["id"], example)
-        for schema in schema["allOf"] for example in schema["examples"]]
+        (meta["id"], example)
+        for meta in metas.values()
+        if meta and meta.get("examples")
+        for example in meta.get("examples")
+    ]
 
-    @pytest.mark.parametrize("schema_id,example", params)
+    @pytest.mark.parametrize("schema_id, example", params)
     @skipUnlessJsonSchema()
     def test_validateconfig_schema_of_example(self, schema_id, example):
         """ For a given example in a config module we test if it is valid
@@ -201,22 +282,42 @@ class ValidateCloudConfigFileTest(CiTestCase):
 
 
 class GetSchemaDocTest(CiTestCase):
-    """Tests for get_schema_doc."""
+    """Tests for get_meta_doc."""
 
     def setUp(self):
         super(GetSchemaDocTest, self).setUp()
         self.required_schema = {
-            'title': 'title', 'description': 'description', 'id': 'id',
-            'name': 'name', 'frequency': 'frequency',
-            'distros': ['debian', 'rhel']}
+            "title": "title",
+            "description": "description",
+            "id": "id",
+            "name": "name",
+            "frequency": "frequency",
+            "distros": ["debian", "rhel"],
+        }
+        self.meta = MetaSchema(
+            {
+                "title": "title",
+                "description": "description",
+                "id": "id",
+                "name": "name",
+                "frequency": "frequency",
+                "distros": ["debian", "rhel"],
+                "examples": [
+                    'ex1:\n    [don\'t, expand, "this"]',
+                    "ex2: true",
+                ],
+            }
+        )
 
-    def test_get_schema_doc_returns_restructured_text(self):
-        """get_schema_doc returns restructured text for a cloudinit schema."""
+    def test_get_meta_doc_returns_restructured_text(self):
+        """get_meta_doc returns restructured text for a cloudinit schema."""
         full_schema = copy(self.required_schema)
         full_schema.update(
             {'properties': {
                 'prop1': {'type': 'array', 'description': 'prop-description',
                           'items': {'type': 'integer'}}}})
+
+        doc = get_meta_doc(self.meta, full_schema)
         self.assertEqual(
             dedent("""
                 name
@@ -232,47 +333,51 @@ class GetSchemaDocTest(CiTestCase):
                 **Supported distros:** debian, rhel
 
                 **Config schema**:
-                    **prop1:** (array of integer) prop-description\n\n"""),
-            get_schema_doc(full_schema))
+                    **prop1:** (array of integer) prop-description
 
-    def test_get_schema_doc_handles_multiple_types(self):
-        """get_schema_doc delimits multiple property types with a '/'."""
-        full_schema = copy(self.required_schema)
-        full_schema.update(
-            {'properties': {
-                'prop1': {'type': ['string', 'integer'],
-                          'description': 'prop-description'}}})
+                **Examples**::
+
+                    ex1:
+                        [don't, expand, "this"]
+                    # --- Example2 ---
+                    ex2: true
+            """),
+            doc,
+        )
+
+    def test_get_meta_doc_handles_multiple_types(self):
+        """get_meta_doc delimits multiple property types with a '/'."""
+        schema = {"properties": {"prop1": {"type": ["string", "integer"]}}}
         self.assertIn(
-            '**prop1:** (string/integer) prop-description',
-            get_schema_doc(full_schema))
+            "**prop1:** (string/integer)", get_meta_doc(self.meta, schema)
+        )
 
-    def test_get_schema_doc_handles_enum_types(self):
-        """get_schema_doc converts enum types to yaml and delimits with '/'."""
-        full_schema = copy(self.required_schema)
-        full_schema.update(
-            {'properties': {
-                'prop1': {'enum': [True, False, 'stuff'],
-                          'description': 'prop-description'}}})
+    def test_get_meta_doc_handles_enum_types(self):
+        """get_meta_doc converts enum types to yaml and delimits with '/'."""
+        schema = {"properties": {"prop1": {"enum": [True, False, "stuff"]}}}
         self.assertIn(
-            '**prop1:** (true/false/stuff) prop-description',
-            get_schema_doc(full_schema))
+            "**prop1:** (true/false/stuff)", get_meta_doc(self.meta, schema)
+        )
 
-    def test_get_schema_doc_handles_nested_oneof_property_types(self):
-        """get_schema_doc describes array items oneOf declarations in type."""
-        full_schema = copy(self.required_schema)
-        full_schema.update(
-            {'properties': {
-                'prop1': {'type': 'array',
-                          'items': {
-                              'oneOf': [{'type': 'string'},
-                                        {'type': 'integer'}]},
-                          'description': 'prop-description'}}})
+    def test_get_meta_doc_handles_nested_oneof_property_types(self):
+        """get_meta_doc describes array items oneOf declarations in type."""
+        schema = {
+            "properties": {
+                "prop1": {
+                    "type": "array",
+                    "items": {
+                        "oneOf": [{"type": "string"}, {"type": "integer"}]
+                    },
+                }
+            }
+        }
         self.assertIn(
-            '**prop1:** (array of (string)/(integer)) prop-description',
-            get_schema_doc(full_schema))
+            "**prop1:** (array of (string)/(integer))",
+            get_meta_doc(self.meta, schema),
+        )
 
-    def test_get_schema_doc_handles_string_examples(self):
-        """get_schema_doc properly indented examples as a list of strings."""
+    def test_get_meta_doc_handles_string_examples(self):
+        """get_meta_doc properly indented examples as a list of strings."""
         full_schema = copy(self.required_schema)
         full_schema.update(
             {'examples': ['ex1:\n    [don\'t, expand, "this"]', 'ex2: true'],
@@ -291,16 +396,17 @@ class GetSchemaDocTest(CiTestCase):
                     # --- Example2 ---
                     ex2: true
             """),
-            get_schema_doc(full_schema))
+            get_meta_doc(self.meta, full_schema),
+        )
 
-    def test_get_schema_doc_properly_parse_description(self):
-        """get_schema_doc description properly formatted"""
-        full_schema = copy(self.required_schema)
-        full_schema.update(
-            {'properties': {
-                'p1': {
-                    'type': 'string',
-                    'description': dedent("""\
+    def test_get_meta_doc_properly_parse_description(self):
+        """get_meta_doc description properly formatted"""
+        schema = {
+            "properties": {
+                "p1": {
+                    "type": "string",
+                    "description": dedent(
+                        """\
                         This item
                         has the
                         following options:
@@ -312,8 +418,8 @@ class GetSchemaDocTest(CiTestCase):
                         The default value is
                         option1""")
                 }
-            }}
-        )
+            }
+        }
 
         self.assertIn(
             dedent("""
@@ -325,16 +431,28 @@ class GetSchemaDocTest(CiTestCase):
                             - option3
 
                     The default value is option1
+
             """),
-            get_schema_doc(full_schema))
+            get_meta_doc(self.meta, schema),
+        )
 
-    def test_get_schema_doc_raises_key_errors(self):
-        """get_schema_doc raises KeyErrors on missing keys."""
-        for key in self.required_schema:
-            invalid_schema = copy(self.required_schema)
-            invalid_schema.pop(key)
+    def test_get_meta_doc_raises_key_errors(self):
+        """get_meta_doc raises KeyErrors on missing keys."""
+        schema = {
+            "properties": {
+                "prop1": {
+                    "type": "array",
+                    "items": {
+                        "oneOf": [{"type": "string"}, {"type": "integer"}]
+                    },
+                }
+            }
+        }
+        for key in self.meta:
+            invalid_meta = copy(self.meta)
+            invalid_meta.pop(key)
             with self.assertRaises(KeyError) as context_mgr:
-                get_schema_doc(invalid_schema)
+                get_meta_doc(invalid_meta, schema)
             self.assertIn(key, str(context_mgr.exception))
 
 
@@ -418,6 +536,7 @@ class TestMain:
 
         _out, err = capsys.readouterr()
         expected = (
+            'Error:\n'
             'Expected one of --config-file, --system or --docs arguments\n'
         )
         assert expected == err
@@ -431,6 +550,7 @@ class TestMain:
 
         _out, err = capsys.readouterr()
         expected = (
+            'Error:\n'
             'Expected one of --config-file, --system or --docs arguments\n'
         )
         assert expected == err
@@ -443,7 +563,7 @@ class TestMain:
                 main()
         assert 1 == context_manager.value.code
         _out, err = capsys.readouterr()
-        assert 'Configfile NOT_A_FILE does not exist\n' == err
+        assert 'Error:\nConfigfile NOT_A_FILE does not exist\n' == err
 
     def test_main_prints_docs(self, capsys):
         """When --docs parameter is provided, main generates documentation."""
@@ -489,12 +609,13 @@ class TestMain:
         assert 1 == context_manager.value.code
         _out, err = capsys.readouterr()
         expected = (
-            'Unable to read system userdata as non-root user. Try using sudo\n'
+            'Error:\nUnable to read system userdata as non-root user. '
+            'Try using sudo\n'
         )
         assert expected == err
 
 
-def _get_schema_doc_examples():
+def _get_meta_doc_examples():
     examples_dir = Path(
         cloudinit.__file__).parent.parent / 'doc' / 'examples'
     assert examples_dir.is_dir()
@@ -507,9 +628,49 @@ def _get_schema_doc_examples():
 class TestSchemaDocExamples:
     schema = get_schema()
 
-    @pytest.mark.parametrize("example_path", _get_schema_doc_examples())
+    @pytest.mark.parametrize("example_path", _get_meta_doc_examples())
     @skipUnlessJsonSchema()
     def test_schema_doc_examples(self, example_path):
         validate_cloudconfig_file(str(example_path), self.schema)
 
+
+class TestStrictMetaschema:
+    """Validate that schemas follow a stricter metaschema definition than
+    the default. This disallows arbitrary key/value pairs.
+    """
+
+    @skipUnlessJsonSchema()
+    def test_modules(self):
+        """Validate all modules with a stricter metaschema"""
+        (validator, _) = get_jsonschema_validator()
+        for (name, value) in get_schemas().items():
+            if value:
+                validate_cloudconfig_metaschema(validator, value)
+            else:
+                logging.warning("module %s has no schema definition", name)
+
+    @skipUnlessJsonSchema()
+    def test_validate_bad_module(self):
+        """Throw exception by default, don't throw if throw=False
+
+        item should be 'items' and is therefore interpreted as an additional
+        property which is invalid with a strict metaschema
+        """
+        (validator, _) = get_jsonschema_validator()
+        schema = {
+            "type": "array",
+            "item": {
+                "type": "object",
+            },
+        }
+        with pytest.raises(
+            SchemaValidationError,
+            match=(r"Additional properties are not allowed.*")
+        ):
+
+            validate_cloudconfig_metaschema(validator, schema)
+
+        validate_cloudconfig_metaschema(validator, schema, throw=False)
+
+
 # vi: ts=4 expandtab syntax=python
diff --git a/tests/unittests/test_cli.py b/tests/unittests/test_cli.py
index fd717f34..d0162673 100644
--- a/tests/unittests/test_cli.py
+++ b/tests/unittests/test_cli.py
@@ -1,6 +1,7 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
 import os
+import contextlib
 import io
 from collections import namedtuple
 
@@ -214,26 +215,106 @@ class TestCLI(test_helpers.FilesystemMockingTestCase):
         self.assertEqual(1, exit_code)
         # Known whitebox output from schema subcommand
         self.assertEqual(
+            'Error:\n'
             'Expected one of --config-file, --system or --docs arguments\n',
             self.stderr.getvalue())
 
-    def test_wb_devel_schema_subcommand_doc_content(self):
-        """Validate that doc content is sane from known examples."""
+    def test_wb_devel_schema_subcommand_doc_all_spot_check(self):
+        """Validate that doc content has correct values from known examples.
+
+        Ensure that schema doc is returned
+        """
+
+        # Note: patchStdoutAndStderr() is convenient for reducing boilerplate,
+        # but inspecting the code for debugging is not ideal
+        # contextlib.redirect_stdout() provides similar behavior as a context
+        # manager
         stdout = io.StringIO()
-        self.patchStdoutAndStderr(stdout=stdout)
-        self._call_main(['cloud-init', 'devel', 'schema', '--docs', 'all'])
-        expected_doc_sections = [
-            '**Supported distros:** all',
-            ('**Supported distros:** almalinux, alpine, centos, cloudlinux, '
-             'debian, eurolinux, fedora, openEuler, opensuse, photon, rhel, '
-             'rocky, sles, ubuntu, virtuozzo'),
-            '**Config schema**:\n    **resize_rootfs:** (true/false/noblock)',
-            '**Examples**::\n\n    runcmd:\n        - [ ls, -l, / ]\n'
-        ]
+        with contextlib.redirect_stdout(stdout):
+            self._call_main(["cloud-init", "devel", "schema", "--docs", "all"])
+            expected_doc_sections = [
+                "**Supported distros:** all",
+                (
+                    "**Supported distros:** almalinux, alpine, centos, "
+                    "cloudlinux, debian, eurolinux, fedora, openEuler, "
+                    "opensuse, photon, rhel, rocky, sles, ubuntu, virtuozzo"
+                ),
+                "**Config schema**:\n    **resize_rootfs:** "
+                "(true/false/noblock)",
+                "**Examples**::\n\n    runcmd:\n        - [ ls, -l, / ]\n",
+            ]
+        stdout = stdout.getvalue()
+        for expected in expected_doc_sections:
+            self.assertIn(expected, stdout)
+
+    def test_wb_devel_schema_subcommand_single_spot_check(self):
+        """Validate that doc content has correct values from known example.
+
+        Validate 'all' arg
+        """
+
+        # Note: patchStdoutAndStderr() is convenient for reducing boilerplate,
+        # but inspecting the code for debugging is not ideal
+        # contextlib.redirect_stdout() provides similar behavior as a context
+        # manager
+        stdout = io.StringIO()
+        with contextlib.redirect_stdout(stdout):
+            self._call_main(
+                ["cloud-init", "devel", "schema", "--docs", "cc_runcmd"]
+            )
+            expected_doc_sections = [
+                "Runcmd\n------\n**Summary:** Run arbitrary commands"
+            ]
         stdout = stdout.getvalue()
         for expected in expected_doc_sections:
             self.assertIn(expected, stdout)
 
+    def test_wb_devel_schema_subcommand_multiple_spot_check(self):
+        """Validate that doc content has correct values from known example.
+
+        Validate single arg
+        """
+
+        stdout = io.StringIO()
+        with contextlib.redirect_stdout(stdout):
+            self._call_main(
+                [
+                    "cloud-init",
+                    "devel",
+                    "schema",
+                    "--docs",
+                    "cc_runcmd",
+                    "cc_resizefs",
+                ]
+            )
+            expected_doc_sections = [
+                "Runcmd\n------\n**Summary:** Run arbitrary commands",
+                "Resizefs\n--------\n**Summary:** Resize filesystem",
+            ]
+        stdout = stdout.getvalue()
+        for expected in expected_doc_sections:
+            self.assertIn(expected, stdout)
+
+    def test_wb_devel_schema_subcommand_bad_arg_fails(self):
+        """Validate that doc content has correct values from known example.
+
+        Validate multiple args
+        """
+
+        # Note: patchStdoutAndStderr() is convenient for reducing boilerplate,
+        # but inspecting the code for debugging is not ideal
+        # contextlib.redirect_stdout() provides similar behavior as a context
+        # manager
+        stderr = io.StringIO()
+        with contextlib.redirect_stderr(stderr):
+            self._call_main(
+                ["cloud-init", "devel", "schema", "--docs", "garbage_value"]
+            )
+            expected_doc_sections = ["Invalid --docs value"]
+        stderr = stderr.getvalue()
+        for expected in expected_doc_sections:
+            self.assertIn(expected, stderr)
+
     @mock.patch('cloudinit.cmd.main.main_single')
     def test_single_subcommand(self, m_main_single):
         """The subcommand 'single' calls main_single with valid args."""