Repository: dbt-labs/dbt-core
Pull request: Convert messages to struct logs
resolves #5957 

### Description

Convert msg to be built in struct logs

### Checklist

- [x] I have read [the contributing guide](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md) and understand what's expected of me
- [x] I have signed the [CLA](https://docs.getdbt.com/docs/contributor-license-agreements)
- [x] I have run this code in development and it appears to resolve the stated issue
- [x] This PR includes tests, or tests are not required/relevant for this PR
- [x] I have [opened an issue to add/update docs](https://github.com/dbt-labs/docs.getdbt.com/issues/new/choose), or docs changes are not required/relevant for this PR
- [x] I have run `changie new` to [create a changelog entry](https://github.com/dbt-labs/dbt-core/blob/main/CONTRIBUTING.md#adding-a-changelog-entry)


Problem statement:
[CT-1267] Convert non-structured messages generated in exceptions.py to structured events
We have a number of logging events in which the string message is generated by external utilities, mostly in various functions in exceptions.py.  Because of this, the actual information used to structure the message is not included in the logging event. These events can be identified by the fact that they contain a "msg" field instead of various structured fields.

Refactor these messages so that more of the formatting code is moved into the logging event code, and so that the events capture the strings used to construct the message.

Hints from issue discussion:

Commit message: fix expected test message

Patch to review:
diff --git a/core/dbt/adapters/base/impl.py b/core/dbt/adapters/base/impl.py
index 3c301c2e7f4..33b7c45a3c4 100644
--- a/core/dbt/adapters/base/impl.py
+++ b/core/dbt/adapters/base/impl.py
@@ -41,13 +41,13 @@
 from dbt.contracts.graph.compiled import CompileResultNode, CompiledSeedNode
 from dbt.contracts.graph.manifest import Manifest, MacroManifest
 from dbt.contracts.graph.parsed import ParsedSeedNode
-from dbt.exceptions import warn_or_error
-from dbt.events.functions import fire_event
+from dbt.events.functions import fire_event, warn_or_error
 from dbt.events.types import (
     CacheMiss,
     ListRelations,
     CodeExecution,
     CodeExecutionStatus,
+    CatalogGenerationError,
 )
 from dbt.utils import filter_null_values, executor, cast_to_str
 
@@ -1327,7 +1327,7 @@ def catch_as_completed(
         elif isinstance(exc, KeyboardInterrupt) or not isinstance(exc, Exception):
             raise exc
         else:
-            warn_or_error(f"Encountered an error while generating catalog: {str(exc)}")
+            warn_or_error(CatalogGenerationError(exc=str(exc)))
             # exc is not None, derives from Exception, and isn't ctrl+c
             exceptions.append(exc)
     return merge_tables(tables), exceptions
diff --git a/core/dbt/config/runtime.py b/core/dbt/config/runtime.py
index a8edb9b096a..4049e4d95c4 100644
--- a/core/dbt/config/runtime.py
+++ b/core/dbt/config/runtime.py
@@ -3,7 +3,7 @@
 from copy import deepcopy
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Dict, Any, Optional, Mapping, Iterator, Iterable, Tuple, List, MutableSet, Type
+from typing import Dict, Any, Optional, Mapping, Iterator, Iterable, Tuple, MutableSet, Type
 
 from .profile import Profile
 from .project import Project
@@ -16,17 +16,16 @@
 from dbt.contracts.connection import AdapterRequiredConfig, Credentials
 from dbt.contracts.graph.manifest import ManifestMetadata
 from dbt.contracts.relation import ComponentName
-from dbt.ui import warning_tag
 
 from dbt.contracts.project import Configuration, UserConfig
 from dbt.exceptions import (
     RuntimeException,
     DbtProjectError,
     validator_error_message,
-    warn_or_error,
     raise_compiler_error,
 )
-
+from dbt.events.functions import warn_or_error
+from dbt.events.types import UnsedResourceConfigPath
 from dbt.dataclass_schema import ValidationError
 
 
@@ -280,11 +279,11 @@ def get_resource_config_paths(self) -> Dict[str, PathSet]:
             "exposures": self._get_config_paths(self.exposures),
         }
 
-    def get_unused_resource_config_paths(
+    def warn_for_unused_resource_config_paths(
         self,
         resource_fqns: Mapping[str, PathSet],
         disabled: PathSet,
-    ) -> List[FQNPath]:
+    ) -> None:
         """Return a list of lists of strings, where each inner list of strings
         represents a type + FQN path of a resource configuration that is not
         used.
@@ -298,23 +297,13 @@ def get_unused_resource_config_paths(
 
             for config_path in config_paths:
                 if not _is_config_used(config_path, fqns):
-                    unused_resource_config_paths.append((resource_type,) + config_path)
-        return unused_resource_config_paths
+                    resource_path = ".".join(i for i in ((resource_type,) + config_path))
+                    unused_resource_config_paths.append(resource_path)
 
-    def warn_for_unused_resource_config_paths(
-        self,
-        resource_fqns: Mapping[str, PathSet],
-        disabled: PathSet,
-    ) -> None:
-        unused = self.get_unused_resource_config_paths(resource_fqns, disabled)
-        if len(unused) == 0:
+        if len(unused_resource_config_paths) == 0:
             return
 
-        msg = UNUSED_RESOURCE_CONFIGURATION_PATH_MESSAGE.format(
-            len(unused), "\n".join("- {}".format(".".join(u)) for u in unused)
-        )
-
-        warn_or_error(msg, log_fmt=warning_tag("{}"))
+        warn_or_error(UnsedResourceConfigPath(unused_config_paths=unused_resource_config_paths))
 
     def load_dependencies(self, base_only=False) -> Mapping[str, "RuntimeConfig"]:
         if self.dependencies is None:
@@ -591,14 +580,6 @@ def from_args(cls: Type[RuntimeConfig], args: Any) -> "RuntimeConfig":
         return cls.from_parts(project=project, profile=profile, args=args)
 
 
-UNUSED_RESOURCE_CONFIGURATION_PATH_MESSAGE = """\
-Configuration paths exist in your dbt_project.yml file which do not \
-apply to any resources.
-There are {} unused configuration paths:
-{}
-"""
-
-
 def _is_config_used(path, fqns):
     if fqns:
         for fqn in fqns:
diff --git a/core/dbt/constants.py b/core/dbt/constants.py
index 1599df3e335..63213476e54 100644
--- a/core/dbt/constants.py
+++ b/core/dbt/constants.py
@@ -1,3 +1,10 @@
 SECRET_ENV_PREFIX = "DBT_ENV_SECRET_"
 DEFAULT_ENV_PLACEHOLDER = "DBT_DEFAULT_PLACEHOLDER"
 METADATA_ENV_PREFIX = "DBT_ENV_CUSTOM_ENV_"
+
+MAXIMUM_SEED_SIZE = 1 * 1024 * 1024
+MAXIMUM_SEED_SIZE_NAME = "1MB"
+
+PIN_PACKAGE_URL = (
+    "https://docs.getdbt.com/docs/package-management#section-specifying-package-versions"
+)
diff --git a/core/dbt/context/providers.py b/core/dbt/context/providers.py
index 597b526e384..280b272d553 100644
--- a/core/dbt/context/providers.py
+++ b/core/dbt/context/providers.py
@@ -53,7 +53,6 @@
     raise_compiler_error,
     ref_invalid_args,
     metric_invalid_args,
-    ref_target_not_found,
     target_not_found,
     ref_bad_context,
     wrapped_exports,
@@ -476,10 +475,11 @@ def resolve(self, target_name: str, target_package: Optional[str] = None) -> Rel
         )
 
         if target_model is None or isinstance(target_model, Disabled):
-            ref_target_not_found(
-                self.model,
-                target_name,
-                target_package,
+            target_not_found(
+                node=self.model,
+                target_name=target_name,
+                target_kind="node",
+                target_package=target_package,
                 disabled=isinstance(target_model, Disabled),
             )
         self.validate(target_model, target_name, target_package)
diff --git a/core/dbt/contracts/files.py b/core/dbt/contracts/files.py
index b915a0d1197..93f12a1411e 100644
--- a/core/dbt/contracts/files.py
+++ b/core/dbt/contracts/files.py
@@ -1,18 +1,16 @@
 import hashlib
 import os
 from dataclasses import dataclass, field
+
 from mashumaro.types import SerializableType
 from typing import List, Optional, Union, Dict, Any
 
+from dbt.constants import MAXIMUM_SEED_SIZE
 from dbt.dataclass_schema import dbtClassMixin, StrEnum
 
 from .util import SourceKey
 
 
-MAXIMUM_SEED_SIZE = 1 * 1024 * 1024
-MAXIMUM_SEED_SIZE_NAME = "1MB"
-
-
 class ParseFileType(StrEnum):
     Macro = "macro"
     Model = "model"
diff --git a/core/dbt/contracts/graph/parsed.py b/core/dbt/contracts/graph/parsed.py
index 860f3fdf662..f4de6e6155d 100644
--- a/core/dbt/contracts/graph/parsed.py
+++ b/core/dbt/contracts/graph/parsed.py
@@ -18,7 +18,7 @@
 from dbt.dataclass_schema import dbtClassMixin, ExtensibleDbtClassMixin
 
 from dbt.clients.system import write_file
-from dbt.contracts.files import FileHash, MAXIMUM_SEED_SIZE_NAME
+from dbt.contracts.files import FileHash
 from dbt.contracts.graph.unparsed import (
     UnparsedNode,
     UnparsedDocumentation,
@@ -41,7 +41,13 @@
 )
 from dbt.contracts.util import Replaceable, AdditionalPropertiesMixin
 from dbt.events.proto_types import NodeInfo
-from dbt.exceptions import warn_or_error
+from dbt.events.functions import warn_or_error
+from dbt.events.types import (
+    SeedIncreased,
+    SeedExceedsLimitSamePath,
+    SeedExceedsLimitAndPathChanged,
+    SeedExceedsLimitChecksumChanged,
+)
 from dbt import flags
 from dbt.node_types import ModelLanguage, NodeType
 
@@ -375,30 +381,28 @@ def same_seeds(first: ParsedNode, second: ParsedNode) -> bool:
     if first.checksum.name == "path":
         msg: str
         if second.checksum.name != "path":
-            msg = (
-                f"Found a seed ({first.package_name}.{first.name}) "
-                f">{MAXIMUM_SEED_SIZE_NAME} in size. The previous file was "
-                f"<={MAXIMUM_SEED_SIZE_NAME}, so it has changed"
+            warn_or_error(
+                SeedIncreased(package_name=first.package_name, name=first.name), node=first
             )
         elif result:
-            msg = (
-                f"Found a seed ({first.package_name}.{first.name}) "
-                f">{MAXIMUM_SEED_SIZE_NAME} in size at the same path, dbt "
-                f"cannot tell if it has changed: assuming they are the same"
+            warn_or_error(
+                SeedExceedsLimitSamePath(package_name=first.package_name, name=first.name),
+                node=first,
             )
         elif not result:
-            msg = (
-                f"Found a seed ({first.package_name}.{first.name}) "
-                f">{MAXIMUM_SEED_SIZE_NAME} in size. The previous file was in "
-                f"a different location, assuming it has changed"
+            warn_or_error(
+                SeedExceedsLimitAndPathChanged(package_name=first.package_name, name=first.name),
+                node=first,
             )
         else:
-            msg = (
-                f"Found a seed ({first.package_name}.{first.name}) "
-                f">{MAXIMUM_SEED_SIZE_NAME} in size. The previous file had a "
-                f"checksum type of {second.checksum.name}, so it has changed"
+            warn_or_error(
+                SeedExceedsLimitChecksumChanged(
+                    package_name=first.package_name,
+                    name=first.name,
+                    checksum_name=second.checksum.name,
+                ),
+                node=first,
             )
-        warn_or_error(msg, node=first)
 
     return result
 
diff --git a/core/dbt/contracts/project.py b/core/dbt/contracts/project.py
index b56aeddaf17..17523a40bdb 100644
--- a/core/dbt/contracts/project.py
+++ b/core/dbt/contracts/project.py
@@ -12,9 +12,7 @@
 from typing import Optional, List, Dict, Union, Any
 from mashumaro.types import SerializableType
 
-PIN_PACKAGE_URL = (
-    "https://docs.getdbt.com/docs/package-management#section-specifying-package-versions"  # noqa
-)
+
 DEFAULT_SEND_ANONYMOUS_USAGE_STATS = True
 
 
diff --git a/core/dbt/deprecations.py b/core/dbt/deprecations.py
index 223091dea60..f7cee59df5a 100644
--- a/core/dbt/deprecations.py
+++ b/core/dbt/deprecations.py
@@ -1,14 +1,14 @@
+import abc
 from typing import Optional, Set, List, Dict, ClassVar
 
 import dbt.exceptions
-from dbt import ui
 
 import dbt.tracking
 
 
 class DBTDeprecation:
     _name: ClassVar[Optional[str]] = None
-    _description: ClassVar[Optional[str]] = None
+    _event: ClassVar[Optional[str]] = None
 
     @property
     def name(self) -> str:
@@ -21,66 +21,50 @@ def track_deprecation_warn(self) -> None:
             dbt.tracking.track_deprecation_warn({"deprecation_name": self.name})
 
     @property
-    def description(self) -> str:
-        if self._description is not None:
-            return self._description
-        raise NotImplementedError("description not implemented for {}".format(self))
+    def event(self) -> abc.ABCMeta:
+        if self._event is not None:
+            module_path = dbt.events.types
+            class_name = self._event
+
+            try:
+                return getattr(module_path, class_name)
+            except AttributeError:
+                msg = f"Event Class `{class_name}` is not defined in `{module_path}`"
+                raise NameError(msg)
+        raise NotImplementedError("event not implemented for {}".format(self._event))
 
     def show(self, *args, **kwargs) -> None:
         if self.name not in active_deprecations:
-            desc = self.description.format(**kwargs)
-            msg = ui.line_wrap_message(desc, prefix="Deprecated functionality\n\n")
-            dbt.exceptions.warn_or_error(msg, log_fmt=ui.warning_tag("{}"))
+            event = self.event(**kwargs)
+            dbt.events.functions.warn_or_error(event)
             self.track_deprecation_warn()
             active_deprecations.add(self.name)
 
 
 class PackageRedirectDeprecation(DBTDeprecation):
     _name = "package-redirect"
-    _description = """\
-    The `{old_name}` package is deprecated in favor of `{new_name}`. Please update
-    your `packages.yml` configuration to use `{new_name}` instead.
-    """
+    _event = "PackageRedirectDeprecation"
 
 
 class PackageInstallPathDeprecation(DBTDeprecation):
     _name = "install-packages-path"
-    _description = """\
-    The default package install path has changed from `dbt_modules` to `dbt_packages`.
-    Please update `clean-targets` in `dbt_project.yml` and check `.gitignore` as well.
-    Or, set `packages-install-path: dbt_modules` if you'd like to keep the current value.
-    """
+    _event = "PackageInstallPathDeprecation"
 
 
-class ConfigPathDeprecation(DBTDeprecation):
-    _description = """\
-    The `{deprecated_path}` config has been renamed to `{exp_path}`.
-    Please update your `dbt_project.yml` configuration to reflect this change.
-    """
-
-
-class ConfigSourcePathDeprecation(ConfigPathDeprecation):
+class ConfigSourcePathDeprecation(DBTDeprecation):
     _name = "project-config-source-paths"
+    _event = "ConfigSourcePathDeprecation"
 
 
-class ConfigDataPathDeprecation(ConfigPathDeprecation):
+class ConfigDataPathDeprecation(DBTDeprecation):
     _name = "project-config-data-paths"
-
-
-_adapter_renamed_description = """\
-The adapter function `adapter.{old_name}` is deprecated and will be removed in
-a future release of dbt. Please use `adapter.{new_name}` instead.
-
-Documentation for {new_name} can be found here:
-
-    https://docs.getdbt.com/docs/adapter
-"""
+    _event = "ConfigDataPathDeprecation"
 
 
 def renamed_method(old_name: str, new_name: str):
     class AdapterDeprecationWarning(DBTDeprecation):
         _name = "adapter:{}".format(old_name)
-        _description = _adapter_renamed_description.format(old_name=old_name, new_name=new_name)
+        _event = "AdapterDeprecationWarning"
 
     dep = AdapterDeprecationWarning()
     deprecations_list.append(dep)
@@ -89,26 +73,12 @@ class AdapterDeprecationWarning(DBTDeprecation):
 
 class MetricAttributesRenamed(DBTDeprecation):
     _name = "metric-attr-renamed"
-    _description = """\
-dbt-core v1.3 renamed attributes for metrics:
-\n  'sql'              -> 'expression'
-\n  'type'             -> 'calculation_method'
-\n  'type: expression' -> 'calculation_method: derived'
-\nThe old metric parameter names will be fully deprecated in v1.4.
-\nPlease remove them from the metric definition of metric '{metric_name}'
-\nRelevant issue here: https://github.com/dbt-labs/dbt-core/issues/5849
-"""
+    _event = "MetricAttributesRenamed"
 
 
 class ExposureNameDeprecation(DBTDeprecation):
     _name = "exposure-name"
-    _description = """\
-    Starting in v1.3, the 'name' of an exposure should contain only letters, numbers, and underscores.
-    Exposures support a new property, 'label', which may contain spaces, capital letters, and special characters.
-    {exposure} does not follow this pattern.
-    Please update the 'name', and use the 'label' property for a human-friendly title.
-    This will raise an error in a future version of dbt-core.
-    """
+    _event = "ExposureNameDeprecation"
 
 
 def warn(name, *args, **kwargs):
@@ -125,12 +95,12 @@ def warn(name, *args, **kwargs):
 active_deprecations: Set[str] = set()
 
 deprecations_list: List[DBTDeprecation] = [
-    ExposureNameDeprecation(),
+    PackageRedirectDeprecation(),
+    PackageInstallPathDeprecation(),
     ConfigSourcePathDeprecation(),
     ConfigDataPathDeprecation(),
-    PackageInstallPathDeprecation(),
-    PackageRedirectDeprecation(),
     MetricAttributesRenamed(),
+    ExposureNameDeprecation(),
 ]
 
 deprecations: Dict[str, DBTDeprecation] = {d.name: d for d in deprecations_list}
diff --git a/core/dbt/deps/git.py b/core/dbt/deps/git.py
index 9e86367acc4..e6dcc479a80 100644
--- a/core/dbt/deps/git.py
+++ b/core/dbt/deps/git.py
@@ -9,14 +9,9 @@
     GitPackage,
 )
 from dbt.deps.base import PinnedPackage, UnpinnedPackage, get_downloads_path
-from dbt.exceptions import ExecutableError, warn_or_error, raise_dependency_error
-from dbt.events.functions import fire_event
-from dbt.events.types import EnsureGitInstalled
-from dbt import ui
-
-PIN_PACKAGE_URL = (
-    "https://docs.getdbt.com/docs/package-management#section-specifying-package-versions"  # noqa
-)
+from dbt.exceptions import ExecutableError, raise_dependency_error
+from dbt.events.functions import fire_event, warn_or_error
+from dbt.events.types import EnsureGitInstalled, DepsUnpinned
 
 
 def md5sum(s: str):
@@ -62,14 +57,6 @@ def nice_version_name(self):
         else:
             return "revision {}".format(self.revision)
 
-    def unpinned_msg(self):
-        if self.revision == "HEAD":
-            return "not pinned, using HEAD (default branch)"
-        elif self.revision in ("main", "master"):
-            return f'pinned to the "{self.revision}" branch'
-        else:
-            return None
-
     def _checkout(self):
         """Performs a shallow clone of the repository into the downloads
         directory. This function can be called repeatedly. If the project has
@@ -92,14 +79,8 @@ def _checkout(self):
     def _fetch_metadata(self, project, renderer) -> ProjectPackageMetadata:
         path = self._checkout()
 
-        if self.unpinned_msg() and self.warn_unpinned:
-            warn_or_error(
-                'The git package "{}" \n\tis {}.\n\tThis can introduce '
-                "breaking changes into your project without warning!\n\nSee {}".format(
-                    self.git, self.unpinned_msg(), PIN_PACKAGE_URL
-                ),
-                log_fmt=ui.yellow("WARNING: {}"),
-            )
+        if (self.revision == "HEAD" or self.revision in ("main", "master")) and self.warn_unpinned:
+            warn_or_error(DepsUnpinned(git=self.git))
         loaded = Project.from_project_root(path, renderer)
         return ProjectPackageMetadata.from_project(loaded)
 
diff --git a/core/dbt/events/base_types.py b/core/dbt/events/base_types.py
index cd3275c02a9..b2b9b97869b 100644
--- a/core/dbt/events/base_types.py
+++ b/core/dbt/events/base_types.py
@@ -3,6 +3,7 @@
 import threading
 from datetime import datetime
 
+
 # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
 # These base types define the _required structure_ for the concrete event #
 # types defined in types.py                                               #
diff --git a/core/dbt/events/functions.py b/core/dbt/events/functions.py
index 122171bc8bf..2425f0abd7f 100644
--- a/core/dbt/events/functions.py
+++ b/core/dbt/events/functions.py
@@ -1,9 +1,16 @@
 import betterproto
 from colorama import Style
+
 from dbt.events.base_types import NoStdOut, BaseEvent, NoFile, Cache
-from dbt.events.types import EventBufferFull, MainReportVersion, EmptyLine
+from dbt.events.helpers import env_secrets, scrub_secrets
+from dbt.events.types import (
+    EventBufferFull,
+    MainReportVersion,
+    EmptyLine,
+)
 import dbt.flags as flags
-from dbt.constants import SECRET_ENV_PREFIX, METADATA_ENV_PREFIX
+
+from dbt.constants import METADATA_ENV_PREFIX
 
 from dbt.logger import make_log_dir_if_missing, GLOBAL_LOGGER
 from datetime import datetime
@@ -18,7 +25,8 @@
 import os
 import uuid
 import threading
-from typing import List, Optional, Union, Callable, Dict
+from typing import Optional, Union, Callable, Dict
+
 from collections import deque
 
 LOG_VERSION = 3
@@ -108,19 +116,6 @@ def stop_capture_stdout_logs() -> None:
     ]
 
 
-def env_secrets() -> List[str]:
-    return [v for k, v in os.environ.items() if k.startswith(SECRET_ENV_PREFIX) and v.strip()]
-
-
-def scrub_secrets(msg: str, secrets: List[str]) -> str:
-    scrubbed = msg
-
-    for secret in secrets:
-        scrubbed = scrubbed.replace(secret, "*****")
-
-    return scrubbed
-
-
 # returns a dictionary representation of the event fields.
 # the message may contain secrets which must be scrubbed at the usage site.
 def event_to_json(
@@ -220,6 +215,15 @@ def send_to_logger(l: Union[Logger, logbook.Logger], level_tag: str, log_line: s
         )
 
 
+def warn_or_error(event, node=None):
+    if flags.WARN_ERROR:
+        from dbt.exceptions import raise_compiler_error
+
+        raise_compiler_error(scrub_secrets(event.info.msg, env_secrets()), node)
+    else:
+        fire_event(event)
+
+
 # an alternative to fire_event which only creates and logs the event value
 # if the condition is met. Does nothing otherwise.
 def fire_event_if(conditional: bool, lazy_e: Callable[[], BaseEvent]) -> None:
diff --git a/core/dbt/events/helpers.py b/core/dbt/events/helpers.py
new file mode 100644
index 00000000000..2570c8653c9
--- /dev/null
+++ b/core/dbt/events/helpers.py
@@ -0,0 +1,16 @@
+import os
+from typing import List
+from dbt.constants import SECRET_ENV_PREFIX
+
+
+def env_secrets() -> List[str]:
+    return [v for k, v in os.environ.items() if k.startswith(SECRET_ENV_PREFIX) and v.strip()]
+
+
+def scrub_secrets(msg: str, secrets: List[str]) -> str:
+    scrubbed = msg
+
+    for secret in secrets:
+        scrubbed = scrubbed.replace(secret, "*****")
+
+    return scrubbed
diff --git a/core/dbt/events/proto_types.py b/core/dbt/events/proto_types.py
index d75713285db..9a2cf242971 100644
--- a/core/dbt/events/proto_types.py
+++ b/core/dbt/events/proto_types.py
@@ -281,6 +281,65 @@ class ProjectCreated(betterproto.Message):
     slack_url: str = betterproto.string_field(4)
 
 
+@dataclass
+class PackageRedirectDeprecation(betterproto.Message):
+    """D001"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    old_name: str = betterproto.string_field(2)
+    new_name: str = betterproto.string_field(3)
+
+
+@dataclass
+class PackageInstallPathDeprecation(betterproto.Message):
+    """D002"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+
+
+@dataclass
+class ConfigSourcePathDeprecation(betterproto.Message):
+    """D003"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    deprecated_path: str = betterproto.string_field(2)
+    exp_path: str = betterproto.string_field(3)
+
+
+@dataclass
+class ConfigDataPathDeprecation(betterproto.Message):
+    """D004"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    deprecated_path: str = betterproto.string_field(2)
+    exp_path: str = betterproto.string_field(3)
+
+
+@dataclass
+class AdapterDeprecationWarning(betterproto.Message):
+    """D005"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    old_name: str = betterproto.string_field(2)
+    new_name: str = betterproto.string_field(3)
+
+
+@dataclass
+class MetricAttributesRenamed(betterproto.Message):
+    """D006"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    metric_name: str = betterproto.string_field(2)
+
+
+@dataclass
+class ExposureNameDeprecation(betterproto.Message):
+    """D007"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    exposure: str = betterproto.string_field(2)
+
+
 @dataclass
 class AdapterEventDebug(betterproto.Message):
     """E001"""
@@ -629,6 +688,14 @@ class CodeExecutionStatus(betterproto.Message):
     elapsed: float = betterproto.float_field(3)
 
 
+@dataclass
+class CatalogGenerationError(betterproto.Message):
+    """E040"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    exc: str = betterproto.string_field(2)
+
+
 @dataclass
 class WriteCatalogFailure(betterproto.Message):
     """E041"""
@@ -1066,17 +1133,119 @@ class PartialParsingDeletedExposure(betterproto.Message):
 
 
 @dataclass
-class InvalidDisabledSourceInTestNode(betterproto.Message):
+class InvalidDisabledTargetInTestNode(betterproto.Message):
     """I050"""
 
     info: "EventInfo" = betterproto.message_field(1)
-    msg: str = betterproto.string_field(2)
+    resource_type_title: str = betterproto.string_field(2)
+    unique_id: str = betterproto.string_field(3)
+    original_file_path: str = betterproto.string_field(4)
+    target_kind: str = betterproto.string_field(5)
+    target_name: str = betterproto.string_field(6)
+    target_package: str = betterproto.string_field(7)
 
 
 @dataclass
-class InvalidRefInTestNode(betterproto.Message):
+class UnsedResourceConfigPath(betterproto.Message):
     """I051"""
 
+    info: "EventInfo" = betterproto.message_field(1)
+    unused_config_paths: List[str] = betterproto.string_field(2)
+
+
+@dataclass
+class SeedIncreased(betterproto.Message):
+    """I052"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    package_name: str = betterproto.string_field(2)
+    name: str = betterproto.string_field(3)
+
+
+@dataclass
+class SeedExceedsLimitSamePath(betterproto.Message):
+    """I053"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    package_name: str = betterproto.string_field(2)
+    name: str = betterproto.string_field(3)
+
+
+@dataclass
+class SeedExceedsLimitAndPathChanged(betterproto.Message):
+    """I054"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    package_name: str = betterproto.string_field(2)
+    name: str = betterproto.string_field(3)
+
+
+@dataclass
+class SeedExceedsLimitChecksumChanged(betterproto.Message):
+    """I055"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    package_name: str = betterproto.string_field(2)
+    name: str = betterproto.string_field(3)
+    checksum_name: str = betterproto.string_field(4)
+
+
+@dataclass
+class UnusedTables(betterproto.Message):
+    """I056"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    unused_tables: List[str] = betterproto.string_field(2)
+
+
+@dataclass
+class WrongResourceSchemaFile(betterproto.Message):
+    """I057"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    patch_name: str = betterproto.string_field(2)
+    resource_type: str = betterproto.string_field(3)
+    plural_resource_type: str = betterproto.string_field(4)
+    yaml_key: str = betterproto.string_field(5)
+    file_path: str = betterproto.string_field(6)
+
+
+@dataclass
+class NoNodeForYamlKey(betterproto.Message):
+    """I058"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    patch_name: str = betterproto.string_field(2)
+    yaml_key: str = betterproto.string_field(3)
+    file_path: str = betterproto.string_field(4)
+
+
+@dataclass
+class MacroPatchNotFound(betterproto.Message):
+    """I059"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    patch_name: str = betterproto.string_field(2)
+
+
+@dataclass
+class NodeNotFoundOrDisabled(betterproto.Message):
+    """I060"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    original_file_path: str = betterproto.string_field(2)
+    unique_id: str = betterproto.string_field(3)
+    resource_type_title: str = betterproto.string_field(4)
+    target_name: str = betterproto.string_field(5)
+    target_kind: str = betterproto.string_field(6)
+    target_package: str = betterproto.string_field(7)
+    disabled: str = betterproto.string_field(8)
+
+
+@dataclass
+class GeneralMacroWarning(betterproto.Message):
+    """I061"""
+
     info: "EventInfo" = betterproto.message_field(1)
     msg: str = betterproto.string_field(2)
 
@@ -1309,6 +1478,23 @@ class DepsSetDownloadDirectory(betterproto.Message):
     path: str = betterproto.string_field(2)
 
 
+@dataclass
+class DepsUnpinned(betterproto.Message):
+    """M029"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    revision: str = betterproto.string_field(2)
+    git: str = betterproto.string_field(3)
+
+
+@dataclass
+class NoNodesForSelectionCriteria(betterproto.Message):
+    """M030"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+    spec_raw: str = betterproto.string_field(2)
+
+
 @dataclass
 class RunningOperationCaughtError(betterproto.Message):
     """Q001"""
@@ -1678,6 +1864,13 @@ class SkippingDetails(betterproto.Message):
     total: int = betterproto.int32_field(7)
 
 
+@dataclass
+class NothingToDo(betterproto.Message):
+    """Q035"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+
+
 @dataclass
 class RunningOperationUncaughtError(betterproto.Message):
     """Q036"""
@@ -1697,6 +1890,13 @@ class EndRunResult(betterproto.Message):
     success: bool = betterproto.bool_field(5)
 
 
+@dataclass
+class NoNodesSelected(betterproto.Message):
+    """Q038"""
+
+    info: "EventInfo" = betterproto.message_field(1)
+
+
 @dataclass
 class CatchableExceptionOnRun(betterproto.Message):
     """W002"""
@@ -2066,34 +2266,16 @@ class TrackingInitializeFailure(betterproto.Message):
     exc_info: str = betterproto.string_field(2)
 
 
-@dataclass
-class GeneralWarningMsg(betterproto.Message):
-    """Z046"""
-
-    info: "EventInfo" = betterproto.message_field(1)
-    msg: str = betterproto.string_field(2)
-    log_fmt: str = betterproto.string_field(3)
-
-
-@dataclass
-class GeneralWarningException(betterproto.Message):
-    """Z047"""
-
-    info: "EventInfo" = betterproto.message_field(1)
-    exc: str = betterproto.string_field(2)
-    log_fmt: str = betterproto.string_field(3)
-
-
 @dataclass
 class EventBufferFull(betterproto.Message):
-    """Z048"""
+    """Z045"""
 
     info: "EventInfo" = betterproto.message_field(1)
 
 
 @dataclass
 class RunResultWarningMessage(betterproto.Message):
-    """Z049"""
+    """Z046"""
 
     info: "EventInfo" = betterproto.message_field(1)
     msg: str = betterproto.string_field(2)
diff --git a/core/dbt/events/types.proto b/core/dbt/events/types.proto
index eaa05b4f93d..29baec9de52 100644
--- a/core/dbt/events/types.proto
+++ b/core/dbt/events/types.proto
@@ -213,6 +213,53 @@ message ProjectCreated {
     string slack_url = 4;
 }
 
+// D - Deprecation
+
+// D001
+message PackageRedirectDeprecation {
+    EventInfo info = 1;
+    string old_name = 2;
+    string new_name = 3;
+}
+
+// D002
+message PackageInstallPathDeprecation {
+    EventInfo info = 1;
+}
+
+// D003
+message ConfigSourcePathDeprecation {
+    EventInfo info = 1;
+    string deprecated_path = 2;
+    string exp_path = 3;
+}
+
+// D004
+message ConfigDataPathDeprecation {
+    EventInfo info = 1;
+    string deprecated_path = 2;
+    string exp_path = 3;
+}
+
+//D005
+message AdapterDeprecationWarning {
+    EventInfo info = 1;
+    string old_name = 2;
+    string new_name = 3;
+}
+
+//D006
+message MetricAttributesRenamed {
+    EventInfo info = 1;
+    string metric_name = 2;
+}
+
+//D007
+message ExposureNameDeprecation {
+    EventInfo info = 1;
+    string exposure = 2;
+}
+
 // E - DB Adapter
 
 // E001
@@ -455,7 +502,6 @@ message AdapterImportError {
 message PluginLoadError {
     EventInfo info = 1;
     string exc_info = 2;
-
 }
 
 // E037
@@ -478,7 +524,11 @@ message CodeExecutionStatus {
     float elapsed = 3;
 }
 
-// Skipped E040
+// E040
+message CatalogGenerationError {
+    EventInfo info = 1;
+    string exc = 2;
+}
 
 // E041
 message WriteCatalogFailure {
@@ -806,17 +856,98 @@ message PartialParsingDeletedExposure {
 }
 
 // I050
-message InvalidDisabledSourceInTestNode {
+message InvalidDisabledTargetInTestNode {
     EventInfo info = 1;
-    string msg = 2;
+    string resource_type_title = 2;
+    string unique_id = 3;
+    string original_file_path = 4;
+    string target_kind = 5;
+    string target_name = 6;
+    string target_package = 7;
 }
 
 // I051
-message InvalidRefInTestNode {
+message UnsedResourceConfigPath {
     EventInfo info = 1;
-    string msg = 2;
+    repeated string unused_config_paths = 2;
+}
+
+// I052
+message SeedIncreased {
+    EventInfo info = 1;
+    string package_name = 2;
+    string name = 3;
+}
+
+// I053
+message SeedExceedsLimitSamePath {
+    EventInfo info = 1;
+    string package_name = 2;
+    string name = 3;
+}
+
+// I054
+message SeedExceedsLimitAndPathChanged {
+    EventInfo info = 1;
+    string package_name = 2;
+    string name = 3;
+}
+
+// I055
+message SeedExceedsLimitChecksumChanged {
+    EventInfo info = 1;
+    string package_name = 2;
+    string name = 3;
+    string checksum_name = 4;
+}
+
+// I056
+message UnusedTables {
+    EventInfo info = 1;
+    repeated string unused_tables = 2;
 }
 
+// I057
+message WrongResourceSchemaFile {
+    EventInfo info = 1;
+    string patch_name = 2;
+    string resource_type = 3;
+    string plural_resource_type = 4;
+    string yaml_key = 5;
+    string file_path = 6;
+}
+
+// I058
+message NoNodeForYamlKey {
+    EventInfo info = 1;
+    string patch_name = 2;
+    string yaml_key = 3;
+    string file_path = 4;
+}
+
+// I059
+message MacroPatchNotFound {
+    EventInfo info = 1;
+    string patch_name = 2;
+}
+
+// I060
+message NodeNotFoundOrDisabled {
+    EventInfo info = 1;
+    string original_file_path = 2;
+    string unique_id = 3;
+    string resource_type_title = 4;
+    string target_name = 5;
+    string target_kind = 6;
+    string target_package = 7;
+    string disabled = 8;
+}
+
+// I061
+message GeneralMacroWarning {
+    EventInfo info = 1;
+    string msg = 2;
+}
 
 // M - Deps generation
 
@@ -992,6 +1123,19 @@ message DepsSetDownloadDirectory {
     string path = 2;
 }
 
+// M029
+message DepsUnpinned {
+    EventInfo info = 1;
+    string revision = 2;
+    string git = 3;
+}
+
+// M030
+message NoNodesForSelectionCriteria {
+    EventInfo info = 1;
+    string spec_raw = 2;
+}
+
 // Q - Node execution
 
 // Q001
@@ -1291,7 +1435,10 @@ message SkippingDetails {
     int32 total = 7;
 }
 
-// Skipped Q035
+// Q035
+message NothingToDo {
+    EventInfo info = 1;
+}
 
 // Q036
 message RunningOperationUncaughtError {
@@ -1308,6 +1455,11 @@ message EndRunResult {
     bool success = 5;
 }
 
+// Q038
+message NoNodesSelected {
+    EventInfo info = 1;
+}
+
 // W - Node testing
 
 // Skipped W001
@@ -1593,28 +1745,12 @@ message TrackingInitializeFailure {
     string exc_info = 2;
 }
 
-// Skipped Z045
-
-// Z046
-message GeneralWarningMsg {
-    EventInfo info = 1;
-    string msg = 2;
-    string log_fmt = 3;
-}
-
-// Z047
-message GeneralWarningException {
-    EventInfo info = 1;
-    string exc = 2;
-    string log_fmt = 3;
-}
-
-// Z048
+// Z045
 message EventBufferFull {
     EventInfo info = 1;
 }
 
-// Z049
+// Z046
 message RunResultWarningMessage {
     EventInfo info = 1;
     string msg = 2;
diff --git a/core/dbt/events/types.py b/core/dbt/events/types.py
index f6e66f941d2..28ce629effc 100644
--- a/core/dbt/events/types.py
+++ b/core/dbt/events/types.py
@@ -1,5 +1,6 @@
 from dataclasses import dataclass
-from dbt import ui
+from dbt.ui import line_wrap_message, warning_tag, red, green, yellow
+from dbt.constants import MAXIMUM_SEED_SIZE_NAME, PIN_PACKAGE_URL
 from dbt.events.base_types import (
     NoFile,
     DebugLevel,
@@ -32,10 +33,11 @@
 # | Code |     Description     |
 # |:----:|:-------------------:|
 # | A    | Pre-project loading |
+# | D    | Deprecations        |
 # | E    | DB adapter          |
 # | I    | Project parsing     |
 # | M    | Deps generation     |
-# | Q    | Node execution     |
+# | Q    | Node execution      |
 # | W    | Node testing        |
 # | Z    | Misc                |
 # | T    | Test only           |
@@ -305,6 +307,114 @@ def message(self) -> str:
 """
 
 
+# =======================================================
+# D - Deprecations
+# =======================================================
+
+
+@dataclass
+class PackageRedirectDeprecation(WarnLevel, pt.PackageRedirectDeprecation):  # noqa
+    def code(self):
+        return "D001"
+
+    def message(self):
+        description = (
+            f"The `{self.old_name}` package is deprecated in favor of `{self.new_name}`. Please "
+            f"update your `packages.yml` configuration to use `{self.new_name}` instead."
+        )
+        return line_wrap_message(warning_tag(f"Deprecated functionality\n\n{description}"))
+
+
+@dataclass
+class PackageInstallPathDeprecation(WarnLevel, pt.PackageInstallPathDeprecation):  # noqa
+    def code(self):
+        return "D002"
+
+    def message(self):
+        description = """\
+        The default package install path has changed from `dbt_modules` to `dbt_packages`.
+        Please update `clean-targets` in `dbt_project.yml` and check `.gitignore` as well.
+        Or, set `packages-install-path: dbt_modules` if you'd like to keep the current value.
+        """
+        return line_wrap_message(warning_tag(f"Deprecated functionality\n\n{description}"))
+
+
+@dataclass
+class ConfigSourcePathDeprecation(WarnLevel, pt.ConfigSourcePathDeprecation):  # noqa
+    def code(self):
+        return "D003"
+
+    def message(self):
+        description = (
+            f"The `{self.deprecated_path}` config has been renamed to `{self.exp_path}`."
+            "Please update your `dbt_project.yml` configuration to reflect this change."
+        )
+        return line_wrap_message(warning_tag(f"Deprecated functionality\n\n{description}"))
+
+
+@dataclass
+class ConfigDataPathDeprecation(WarnLevel, pt.ConfigDataPathDeprecation):  # noqa
+    def code(self):
+        return "D004"
+
+    def message(self):
+        description = (
+            f"The `{self.deprecated_path}` config has been renamed to `{self.exp_path}`."
+            "Please update your `dbt_project.yml` configuration to reflect this change."
+        )
+        return line_wrap_message(warning_tag(f"Deprecated functionality\n\n{description}"))
+
+
+@dataclass
+class AdapterDeprecationWarning(WarnLevel, pt.AdapterDeprecationWarning):  # noqa
+    def code(self):
+        return "D005"
+
+    def message(self):
+        description = (
+            f"The adapter function `adapter.{self.old_name}` is deprecated and will be removed in "
+            f"a future release of dbt. Please use `adapter.{self.new_name}` instead. "
+            f"\n\nDocumentation for {self.new_name} can be found here:"
+            f"\n\nhttps://docs.getdbt.com/docs/adapter"
+        )
+        return line_wrap_message(warning_tag(f"Deprecated functionality\n\n{description}"))
+
+
+@dataclass
+class MetricAttributesRenamed(WarnLevel, pt.MetricAttributesRenamed):  # noqa
+    def code(self):
+        return "D006"
+
+    def message(self):
+        description = (
+            "dbt-core v1.3 renamed attributes for metrics:"
+            "\n  'sql'              -> 'expression'"
+            "\n  'type'             -> 'calculation_method'"
+            "\n  'type: expression' -> 'calculation_method: derived'"
+            "\nThe old metric parameter names will be fully deprecated in v1.4."
+            f"\nPlease remove them from the metric definition of metric '{self.metric_name}'"
+            "\nRelevant issue here: https://github.com/dbt-labs/dbt-core/issues/5849"
+        )
+
+        return warning_tag(f"Deprecated functionality\n\n{description}")
+
+
+@dataclass
+class ExposureNameDeprecation(WarnLevel, pt.ExposureNameDeprecation):  # noqa
+    def code(self):
+        return "D007"
+
+    def message(self):
+        description = (
+            "Starting in v1.3, the 'name' of an exposure should contain only letters, "
+            "numbers, and underscores. Exposures support a new property, 'label', which may "
+            f"contain spaces, capital letters, and special characters. {self.exposure} does not "
+            "follow this pattern. Please update the 'name', and use the 'label' property for a "
+            "human-friendly title. This will raise an error in a future version of dbt-core."
+        )
+        return line_wrap_message(warning_tag(f"Deprecated functionality\n\n{description}"))
+
+
 # =======================================================
 # E - DB Adapter
 # =======================================================
@@ -675,7 +785,13 @@ def message(self) -> str:
         return f"Execution status: {self.status} in {self.elapsed} seconds"
 
 
-# Skipped E040
+@dataclass
+class CatalogGenerationError(WarnLevel, pt.CatalogGenerationError):
+    def code(self):
+        return "E040"
+
+    def message(self) -> str:
+        return f"Encountered an error while generating catalog: {self.exc}"
 
 
 @dataclass
@@ -1218,23 +1334,194 @@ def message(self) -> str:
         return f"Partial parsing: deleted exposure {self.unique_id}"
 
 
-# TODO: switch to storing structured info and calling get_target_failure_msg
 @dataclass
-class InvalidDisabledSourceInTestNode(
-    WarnLevel, EventStringFunctor, pt.InvalidDisabledSourceInTestNode
-):
+class InvalidDisabledTargetInTestNode(WarnLevel, pt.InvalidDisabledTargetInTestNode):
     def code(self):
         return "I050"
 
     def message(self) -> str:
-        return ui.warning_tag(self.msg)
+
+        target_package_string = ""
+        if self.target_package != target_package_string:
+            target_package_string = "in package '{}' ".format(self.target_package)
+
+        msg = "{} '{}' ({}) depends on a {} named '{}' {}which is disabled".format(
+            self.resource_type_title,
+            self.unique_id,
+            self.original_file_path,
+            self.target_kind,
+            self.target_name,
+            target_package_string,
+        )
+
+        return warning_tag(msg)
 
 
 @dataclass
-class InvalidRefInTestNode(DebugLevel, EventStringFunctor, pt.InvalidRefInTestNode):
+class UnsedResourceConfigPath(WarnLevel, pt.UnsedResourceConfigPath):
     def code(self):
         return "I051"
 
+    def message(self) -> str:
+        path_list = "\n".join(f"- {u}" for u in self.unused_config_paths)
+        msg = (
+            "Configuration paths exist in your dbt_project.yml file which do not "
+            "apply to any resources.\n"
+            f"There are {len(self.unused_config_paths)} unused configuration paths:\n{path_list}"
+        )
+        return warning_tag(msg)
+
+
+@dataclass
+class SeedIncreased(WarnLevel, pt.SeedIncreased):
+    def code(self):
+        return "I052"
+
+    def message(self) -> str:
+        msg = (
+            f"Found a seed ({self.package_name}.{self.name}) "
+            f">{MAXIMUM_SEED_SIZE_NAME} in size. The previous file was "
+            f"<={MAXIMUM_SEED_SIZE_NAME}, so it has changed"
+        )
+        return msg
+
+
+@dataclass
+class SeedExceedsLimitSamePath(WarnLevel, pt.SeedExceedsLimitSamePath):
+    def code(self):
+        return "I053"
+
+    def message(self) -> str:
+        msg = (
+            f"Found a seed ({self.package_name}.{self.name}) "
+            f">{MAXIMUM_SEED_SIZE_NAME} in size at the same path, dbt "
+            f"cannot tell if it has changed: assuming they are the same"
+        )
+        return msg
+
+
+@dataclass
+class SeedExceedsLimitAndPathChanged(WarnLevel, pt.SeedExceedsLimitAndPathChanged):
+    def code(self):
+        return "I054"
+
+    def message(self) -> str:
+        msg = (
+            f"Found a seed ({self.package_name}.{self.name}) "
+            f">{MAXIMUM_SEED_SIZE_NAME} in size. The previous file was in "
+            f"a different location, assuming it has changed"
+        )
+        return msg
+
+
+@dataclass
+class SeedExceedsLimitChecksumChanged(WarnLevel, pt.SeedExceedsLimitChecksumChanged):
+    def code(self):
+        return "I055"
+
+    def message(self) -> str:
+        msg = (
+            f"Found a seed ({self.package_name}.{self.name}) "
+            f">{MAXIMUM_SEED_SIZE_NAME} in size. The previous file had a "
+            f"checksum type of {self.checksum_name}, so it has changed"
+        )
+        return msg
+
+
+@dataclass
+class UnusedTables(WarnLevel, pt.UnusedTables):
+    def code(self):
+        return "I056"
+
+    def message(self) -> str:
+        msg = [
+            "During parsing, dbt encountered source overrides that had no target:",
+        ]
+        msg += self.unused_tables
+        msg.append("")
+        return warning_tag("\n".join(msg))
+
+
+@dataclass
+class WrongResourceSchemaFile(WarnLevel, pt.WrongResourceSchemaFile):
+    def code(self):
+        return "I057"
+
+    def message(self) -> str:
+        msg = line_wrap_message(
+            f"""\
+            '{self.patch_name}' is a {self.resource_type} node, but it is
+            specified in the {self.yaml_key} section of
+            {self.file_path}.
+            To fix this error, place the `{self.patch_name}`
+            specification under the {self.plural_resource_type} key instead.
+            """
+        )
+        return warning_tag(msg)
+
+
+@dataclass
+class NoNodeForYamlKey(WarnLevel, pt.NoNodeForYamlKey):
+    def code(self):
+        return "I058"
+
+    def message(self) -> str:
+        msg = (
+            f"Did not find matching node for patch with name '{self.patch_name}' "
+            f"in the '{self.yaml_key}' section of "
+            f"file '{self.file_path}'"
+        )
+        return warning_tag(msg)
+
+
+@dataclass
+class MacroPatchNotFound(WarnLevel, pt.MacroPatchNotFound):
+    def code(self):
+        return "I059"
+
+    def message(self) -> str:
+        msg = f'Found patch for macro "{self.patch_name}" which was not found'
+        return warning_tag(msg)
+
+
+@dataclass
+class NodeNotFoundOrDisabled(WarnLevel, pt.NodeNotFoundOrDisabled):
+    def code(self):
+        return "I060"
+
+    def message(self) -> str:
+        # this is duplicated logic from exceptions.get_not_found_or_disabled_msg
+        # when we convert exceptions to be stuctured maybe it can be combined?
+        # convverting the bool to a string since None is also valid
+        if self.disabled == "None":
+            reason = "was not found or is disabled"
+        elif self.disabled == "True":
+            reason = "is disabled"
+        else:
+            reason = "was not found"
+
+        target_package_string = ""
+        if self.target_package is not None:
+            target_package_string = "in package '{}' ".format(self.target_package)
+
+        msg = "{} '{}' ({}) depends on a {} named '{}' {}which {}".format(
+            self.resource_type_title,
+            self.unique_id,
+            self.original_file_path,
+            self.target_kind,
+            self.target_name,
+            target_package_string,
+            reason,
+        )
+
+        return warning_tag(msg)
+
+
+@dataclass
+class GeneralMacroWarning(WarnLevel, pt.GeneralMacroWarning):
+    def code(self):
+        return "I061"
+
     def message(self) -> str:
         return self.msg
 
@@ -1343,6 +1630,7 @@ def code(self):
         return "M011"
 
     def message(self) -> str:
+        # This is for the log method used in macros so msg cannot be built here
         return self.msg
 
 
@@ -1352,6 +1640,7 @@ def code(self):
         return "M012"
 
     def message(self) -> str:
+        # This is for the log method used in macros so msg cannot be built here
         return self.msg
 
 
@@ -1505,6 +1794,35 @@ def message(self) -> str:
         return f"Set downloads directory='{self.path}'"
 
 
+@dataclass
+class DepsUnpinned(WarnLevel, pt.DepsUnpinned):
+    def code(self):
+        return "M029"
+
+    def message(self) -> str:
+        if self.revision == "HEAD":
+            unpinned_msg = "not pinned, using HEAD (default branch)"
+        elif self.revision in ("main", "master"):
+            unpinned_msg = f'pinned to the "{self.revision}" branch'
+        else:
+            unpinned_msg = None
+
+        msg = (
+            f'The git package "{self.git}" \n\tis {unpinned_msg}.\n\tThis can introduce '
+            f"breaking changes into your project without warning!\n\nSee {PIN_PACKAGE_URL}"
+        )
+        return yellow(f"WARNING: {msg}")
+
+
+@dataclass
+class NoNodesForSelectionCriteria(WarnLevel, pt.NoNodesForSelectionCriteria):
+    def code(self):
+        return "M030"
+
+    def message(self) -> str:
+        return f"The selection criterion '{self.spec_raw}' does not match any nodes"
+
+
 # =======================================================
 # Q - Node execution
 # =======================================================
@@ -1575,7 +1893,7 @@ def message(self) -> str:
         msg = f"{info} {self.name}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.red(info),
+            status=red(info),
             index=self.index,
             total=self.num_models,
             execution_time=self.execution_time,
@@ -1592,7 +1910,7 @@ def message(self) -> str:
         msg = f"{info} {self.name}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.green(info),
+            status=green(info),
             index=self.index,
             total=self.num_models,
             execution_time=self.execution_time,
@@ -1609,7 +1927,7 @@ def message(self) -> str:
         msg = f"{info} {self.name}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.yellow(info),
+            status=yellow(info),
             index=self.index,
             total=self.num_models,
             execution_time=self.execution_time,
@@ -1626,7 +1944,7 @@ def message(self) -> str:
         msg = f"{info} {self.name}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.red(info),
+            status=red(info),
             index=self.index,
             total=self.num_models,
             execution_time=self.execution_time,
@@ -1653,7 +1971,7 @@ def message(self) -> str:
         msg = f"{info} {self.description}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.green(self.status),
+            status=green(self.status),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1670,7 +1988,7 @@ def message(self) -> str:
         msg = f"{info} {self.description}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.red(self.status.upper()),
+            status=red(self.status.upper()),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1687,7 +2005,7 @@ def message(self) -> str:
         msg = "{info} {description}".format(info=info, description=self.description, **self.cfg)
         return format_fancy_output_line(
             msg=msg,
-            status=ui.red(self.status.upper()),
+            status=red(self.status.upper()),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1704,7 +2022,7 @@ def message(self) -> str:
         msg = "{info} {description}".format(info=info, description=self.description, **self.cfg)
         return format_fancy_output_line(
             msg=msg,
-            status=ui.green(self.status),
+            status=green(self.status),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1721,7 +2039,7 @@ def message(self) -> str:
         msg = f"{info} seed file {self.schema}.{self.relation}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.red(self.status.upper()),
+            status=red(self.status.upper()),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1738,7 +2056,7 @@ def message(self) -> str:
         msg = f"{info} seed file {self.schema}.{self.relation}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.green(self.status),
+            status=green(self.status),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1755,7 +2073,7 @@ def message(self) -> str:
         msg = f"{info} freshness of {self.source_name}.{self.table_name}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.red(info),
+            status=red(info),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1772,7 +2090,7 @@ def message(self) -> str:
         msg = f"{info} freshness of {self.source_name}.{self.table_name}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.red(info),
+            status=red(info),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1789,7 +2107,7 @@ def message(self) -> str:
         msg = f"{info} freshness of {self.source_name}.{self.table_name}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.yellow(info),
+            status=yellow(info),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1806,7 +2124,7 @@ def message(self) -> str:
         msg = f"{info} freshness of {self.source_name}.{self.table_name}"
         return format_fancy_output_line(
             msg=msg,
-            status=ui.green(info),
+            status=green(info),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1820,7 +2138,7 @@ def code(self):
 
     def message(self) -> str:
         msg = "CANCEL query {}".format(self.conn_name)
-        return format_fancy_output_line(msg=msg, status=ui.red("CANCEL"), index=None, total=None)
+        return format_fancy_output_line(msg=msg, status=red("CANCEL"), index=None, total=None)
 
 
 @dataclass
@@ -1861,7 +2179,7 @@ def message(self) -> str:
             "cancellation. Some queries may still be "
             "running!"
         )
-        return ui.yellow(msg)
+        return yellow(msg)
 
 
 @dataclass
@@ -1930,7 +2248,7 @@ def message(self) -> str:
         msg = "OK hook: {}".format(self.statement)
         return format_fancy_output_line(
             msg=msg,
-            status=ui.green(self.status),
+            status=green(self.status),
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,
@@ -1949,11 +2267,17 @@ def message(self) -> str:
         else:
             msg = f"SKIP {self.resource_type} {self.node_name}"
         return format_fancy_output_line(
-            msg=msg, status=ui.yellow("SKIP"), index=self.index, total=self.total
+            msg=msg, status=yellow("SKIP"), index=self.index, total=self.total
         )
 
 
-# Skipped Q035
+@dataclass
+class NothingToDo(WarnLevel, pt.NothingToDo):
+    def code(self):
+        return "Q035"
+
+    def message(self) -> str:
+        return "Nothing to do. Try checking your model configs and model specification args"
 
 
 @dataclass
@@ -1974,6 +2298,15 @@ def message(self) -> str:
         return "Command end result"
 
 
+@dataclass
+class NoNodesSelected(WarnLevel, pt.NoNodesSelected):
+    def code(self):
+        return "Q038"
+
+    def message(self) -> str:
+        return "No nodes selected!"
+
+
 # =======================================================
 # W - Node testing
 # =======================================================
@@ -2003,7 +2336,7 @@ def message(self) -> str:
 """.strip()
 
         return "{prefix}\n{error}\n\n{note}".format(
-            prefix=ui.red(prefix), error=str(self.exc).strip(), note=internal_error_string
+            prefix=red(prefix), error=str(self.exc).strip(), note=internal_error_string
         )
 
 
@@ -2017,7 +2350,7 @@ def message(self) -> str:
         if node_description is None:
             node_description = self.unique_id
         prefix = "Unhandled error while executing {}".format(node_description)
-        return "{prefix}\n{error}".format(prefix=ui.red(prefix), error=str(self.exc).strip())
+        return "{prefix}\n{error}".format(prefix=red(prefix), error=str(self.exc).strip())
 
 
 @dataclass
@@ -2241,7 +2574,7 @@ def code(self):
 
     def message(self) -> str:
         info = "Warning"
-        return ui.yellow(f"{info} in {self.resource_type} {self.node_name} ({self.path})")
+        return yellow(f"{info} in {self.resource_type} {self.node_name} ({self.path})")
 
 
 @dataclass
@@ -2251,7 +2584,7 @@ def code(self):
 
     def message(self) -> str:
         info = "Failure"
-        return ui.red(f"{info} in {self.resource_type} {self.node_name} ({self.path})")
+        return red(f"{info} in {self.resource_type} {self.node_name} ({self.path})")
 
 
 @dataclass
@@ -2270,6 +2603,7 @@ def code(self):
         return "Z024"
 
     def message(self) -> str:
+        # This is the message on the result object, cannot be built here
         return f"  {self.msg}"
 
 
@@ -2302,13 +2636,16 @@ def message(self) -> str:
         return f"  See test failures:\n  {border}\n  {msg}\n  {border}"
 
 
+# FirstRunResultError and AfterFirstRunResultError are just splitting the message from the result
+#  object into multiple log lines
+# TODO: is this reallly needed?  See printer.py
 @dataclass
 class FirstRunResultError(ErrorLevel, EventStringFunctor, pt.FirstRunResultError):
     def code(self):
         return "Z028"
 
     def message(self) -> str:
-        return ui.yellow(self.msg)
+        return yellow(self.msg)
 
 
 @dataclass
@@ -2329,13 +2666,13 @@ def message(self) -> str:
         error_plural = pluralize(self.num_errors, "error")
         warn_plural = pluralize(self.num_warnings, "warning")
         if self.keyboard_interrupt:
-            message = ui.yellow("Exited because of keyboard interrupt.")
+            message = yellow("Exited because of keyboard interrupt.")
         elif self.num_errors > 0:
-            message = ui.red("Completed with {} and {}:".format(error_plural, warn_plural))
+            message = red("Completed with {} and {}:".format(error_plural, warn_plural))
         elif self.num_warnings > 0:
-            message = ui.yellow("Completed with {}:".format(warn_plural))
+            message = yellow("Completed with {}:".format(warn_plural))
         else:
-            message = ui.green("Completed successfully")
+            message = green("Completed successfully")
         return message
 
 
@@ -2350,7 +2687,7 @@ def code(self):
     def message(self) -> str:
         msg = f"SKIP relation {self.schema}.{self.relation} due to ephemeral model error"
         return format_fancy_output_line(
-            msg=msg, status=ui.red("ERROR SKIP"), index=self.index, total=self.total
+            msg=msg, status=red("ERROR SKIP"), index=self.index, total=self.total
         )
 
 
@@ -2446,31 +2783,10 @@ def message(self) -> str:
         return "Got an exception trying to initialize tracking"
 
 
-# Skipped Z045
-
-
-@dataclass
-class GeneralWarningMsg(WarnLevel, EventStringFunctor, pt.GeneralWarningMsg):
-    def code(self):
-        return "Z046"
-
-    def message(self) -> str:
-        return self.log_fmt.format(self.msg) if self.log_fmt is not None else self.msg
-
-
-@dataclass
-class GeneralWarningException(WarnLevel, pt.GeneralWarningException):
-    def code(self):
-        return "Z047"
-
-    def message(self) -> str:
-        return self.log_fmt.format(str(self.exc)) if self.log_fmt is not None else str(self.exc)
-
-
 @dataclass
 class EventBufferFull(WarnLevel, pt.EventBufferFull):
     def code(self):
-        return "Z048"
+        return "Z045"
 
     def message(self) -> str:
         return (
@@ -2479,12 +2795,14 @@ def message(self) -> str:
         )
 
 
+# this is the message from the result object
 @dataclass
 class RunResultWarningMessage(WarnLevel, EventStringFunctor, pt.RunResultWarningMessage):
     def code(self):
-        return "Z049"
+        return "Z046"
 
     def message(self) -> str:
+        # This is the message on the result object, cannot be formatted in event
         return self.msg
 
 
@@ -2522,6 +2840,15 @@ def message(self) -> str:
     ProjectNameAlreadyExists(name="")
     ProjectCreated(project_name="")
 
+    # D - Deprecations ======================
+    PackageRedirectDeprecation(old_name="", new_name="")
+    PackageInstallPathDeprecation()
+    ConfigSourcePathDeprecation(deprecated_path="", exp_path="")
+    ConfigDataPathDeprecation(deprecated_path="", exp_path="")
+    AdapterDeprecationWarning(old_name="", new_name="")
+    MetricAttributesRenamed(metric_name="")
+    ExposureNameDeprecation(exposure="")
+
     # E - DB Adapter ======================
     AdapterEventDebug()
     AdapterEventInfo()
@@ -2580,6 +2907,7 @@ def message(self) -> str:
     NewConnectionOpening(connection_state="")
     CodeExecution(conn_name="", code_content="")
     CodeExecutionStatus(status="", elapsed=0.1)
+    CatalogGenerationError(exc="")
     WriteCatalogFailure(num_exceptions=0)
     CatalogWritten(path="")
     CannotGenerateDocs()
@@ -2638,8 +2966,32 @@ def message(self) -> str:
     PartialParsingUpdateSchemaFile(file_id="")
     PartialParsingDeletedSource(unique_id="")
     PartialParsingDeletedExposure(unique_id="")
-    InvalidDisabledSourceInTestNode(msg="")
-    InvalidRefInTestNode(msg="")
+    InvalidDisabledTargetInTestNode(
+        resource_type_title="",
+        unique_id="",
+        original_file_path="",
+        target_kind="",
+        target_name="",
+        target_package="",
+    )
+    UnsedResourceConfigPath(unused_config_paths=[])
+    SeedIncreased(package_name="", name="")
+    SeedExceedsLimitSamePath(package_name="", name="")
+    SeedExceedsLimitAndPathChanged(package_name="", name="")
+    SeedExceedsLimitChecksumChanged(package_name="", name="", checksum_name="")
+    UnusedTables(unused_tables=[])
+    WrongResourceSchemaFile(patch_name="", resource_type="", file_path="", plural_resource_type="")
+    NoNodeForYamlKey(patch_name="", yaml_key="", file_path="")
+    MacroPatchNotFound(patch_name="")
+    NodeNotFoundOrDisabled(
+        original_file_path="",
+        unique_id="",
+        resource_type_title="",
+        target_name="",
+        target_kind="",
+        target_package="",
+        disabled="",
+    )
 
     # M - Deps generation ======================
 
@@ -2810,8 +3162,12 @@ def message(self) -> str:
         index=0,
         total=0,
     )
+    NothingToDo()
     RunningOperationUncaughtError(exc="")
     EndRunResult()
+    NoNodesSelected()
+    DepsUnpinned(revision="", git="")
+    NoNodesForSelectionCriteria(spec_raw="")
 
     # W - Node testing ======================
 
@@ -2863,6 +3219,4 @@ def message(self) -> str:
     FlushEvents()
     FlushEventsFailure()
     TrackingInitializeFailure()
-    GeneralWarningMsg(msg="", log_fmt="")
-    GeneralWarningException(exc="", log_fmt="")
     EventBufferFull()
diff --git a/core/dbt/exceptions.py b/core/dbt/exceptions.py
index db824e19bf1..f0eeb4f6d4f 100644
--- a/core/dbt/exceptions.py
+++ b/core/dbt/exceptions.py
@@ -2,11 +2,9 @@
 import functools
 from typing import NoReturn, Optional, Mapping, Any
 
-from dbt.events.functions import fire_event, scrub_secrets, env_secrets
-from dbt.events.types import GeneralWarningMsg, GeneralWarningException
+from dbt.events.helpers import env_secrets, scrub_secrets
+from dbt.events.types import GeneralMacroWarning
 from dbt.node_types import NodeType
-from dbt import flags
-from dbt.ui import line_wrap_message, warning_tag
 
 import dbt.dataclass_schema
 
@@ -570,39 +568,13 @@ def doc_target_not_found(
     raise_compiler_error(msg, model)
 
 
-def _get_target_failure_msg(
+def get_not_found_or_disabled_msg(
     original_file_path,
     unique_id,
     resource_type_title,
     target_name: str,
-    target_model_package: Optional[str],
-    include_path: bool,
-    reason: str,
     target_kind: str,
-) -> str:
-    target_package_string = ""
-    if target_model_package is not None:
-        target_package_string = "in package '{}' ".format(target_model_package)
-
-    source_path_string = ""
-    if include_path:
-        source_path_string = " ({})".format(original_file_path)
-
-    return "{} '{}'{} depends on a {} named '{}' {}which {}".format(
-        resource_type_title,
-        unique_id,
-        source_path_string,
-        target_kind,
-        target_name,
-        target_package_string,
-        reason,
-    )
-
-
-def get_target_not_found_or_disabled_msg(
-    node,
-    target_name: str,
-    target_package: Optional[str],
+    target_package: Optional[str] = None,
     disabled: Optional[bool] = None,
 ) -> str:
     if disabled is None:
@@ -611,52 +583,19 @@ def get_target_not_found_or_disabled_msg(
         reason = "is disabled"
     else:
         reason = "was not found"
-    return _get_target_failure_msg(
-        node.original_file_path,
-        node.unique_id,
-        node.resource_type.title(),
-        target_name,
-        target_package,
-        include_path=True,
-        reason=reason,
-        target_kind="node",
-    )
-
-
-def ref_target_not_found(
-    model,
-    target_model_name: str,
-    target_model_package: Optional[str],
-    disabled: Optional[bool] = None,
-) -> NoReturn:
-    msg = get_target_not_found_or_disabled_msg(
-        model, target_model_name, target_model_package, disabled
-    )
-    raise_compiler_error(msg, model)
 
+    target_package_string = ""
+    if target_package is not None:
+        target_package_string = "in package '{}' ".format(target_package)
 
-def get_not_found_or_disabled_msg(
-    node,
-    target_name: str,
-    target_kind: str,
-    target_package: Optional[str] = None,
-    disabled: Optional[bool] = None,
-) -> str:
-    if disabled is None:
-        reason = "was not found or is disabled"
-    elif disabled is True:
-        reason = "is disabled"
-    else:
-        reason = "was not found"
-    return _get_target_failure_msg(
-        node.original_file_path,
-        node.unique_id,
-        node.resource_type.title(),
+    return "{} '{}' ({}) depends on a {} named '{}' {}which {}".format(
+        resource_type_title,
+        unique_id,
+        original_file_path,
+        target_kind,
         target_name,
-        target_package,
-        include_path=True,
-        reason=reason,
-        target_kind=target_kind,
+        target_package_string,
+        reason,
     )
 
 
@@ -668,7 +607,9 @@ def target_not_found(
     disabled: Optional[bool] = None,
 ) -> NoReturn:
     msg = get_not_found_or_disabled_msg(
-        node=node,
+        original_file_path=node.original_file_path,
+        unique_id=node.unique_id,
+        resource_type_title=node.resource_type.title(),
         target_name=target_name,
         target_kind=target_kind,
         target_package=target_package,
@@ -1041,19 +982,6 @@ def raise_unrecognized_credentials_type(typename, supported_types):
     )
 
 
-def warn_invalid_patch(patch, resource_type):
-    msg = line_wrap_message(
-        f"""\
-        '{patch.name}' is a {resource_type} node, but it is
-        specified in the {patch.yaml_key} section of
-        {patch.original_file_path}.
-        To fix this error, place the `{patch.name}`
-        specification under the {resource_type.pluralize()} key instead.
-        """
-    )
-    warn_or_error(msg, log_fmt=warning_tag("{}"))
-
-
 def raise_not_implemented(msg):
     raise NotImplementedException("ERROR: {}".format(msg))
 
@@ -1067,24 +995,8 @@ def raise_duplicate_alias(
     raise AliasException(f'Got duplicate keys: ({key_names}) all map to "{canonical_key}"')
 
 
-def warn_or_error(msg, node=None, log_fmt=None):
-    if flags.WARN_ERROR:
-        raise_compiler_error(scrub_secrets(msg, env_secrets()), node)
-    else:
-        fire_event(GeneralWarningMsg(msg=msg, log_fmt=log_fmt))
-
-
-def warn_or_raise(exc, log_fmt=None):
-    if flags.WARN_ERROR:
-        raise exc
-    else:
-        fire_event(GeneralWarningException(exc=str(exc), log_fmt=log_fmt))
-
-
 def warn(msg, node=None):
-    # there's no reason to expose log_fmt to macros - it's only useful for
-    # handling colors
-    warn_or_error(msg, node=node)
+    dbt.events.functions.warn_or_error(GeneralMacroWarning(msg=msg), node=node)
     return ""
 
 
diff --git a/core/dbt/graph/selector.py b/core/dbt/graph/selector.py
index 49b73fc71c4..89de27b3697 100644
--- a/core/dbt/graph/selector.py
+++ b/core/dbt/graph/selector.py
@@ -5,13 +5,12 @@
 from .selector_methods import MethodManager
 from .selector_spec import SelectionCriteria, SelectionSpec, IndirectSelection
 
-from dbt.events.functions import fire_event
-from dbt.events.types import SelectorReportInvalidSelector
+from dbt.events.functions import fire_event, warn_or_error
+from dbt.events.types import SelectorReportInvalidSelector, NoNodesForSelectionCriteria
 from dbt.node_types import NodeType
 from dbt.exceptions import (
     InternalException,
     InvalidSelectorException,
-    warn_or_error,
 )
 from dbt.contracts.graph.compiled import GraphMemberNode
 from dbt.contracts.graph.manifest import Manifest
@@ -24,11 +23,6 @@ def get_package_names(nodes):
     return set([node.split(".")[1] for node in nodes])
 
 
-def alert_non_existence(raw_spec, nodes):
-    if len(nodes) == 0:
-        warn_or_error(f"The selection criterion '{str(raw_spec)}' does not match any nodes")
-
-
 def can_select_indirectly(node):
     """If a node is not selected itself, but its parent(s) are, it may qualify
     for indirect selection.
@@ -143,7 +137,7 @@ def select_nodes_recursively(self, spec: SelectionSpec) -> Tuple[Set[UniqueId],
             direct_nodes = self.incorporate_indirect_nodes(initial_direct, indirect_nodes)
 
             if spec.expect_exists:
-                alert_non_existence(spec.raw, direct_nodes)
+                warn_or_error(NoNodesForSelectionCriteria(spec_raw=str(spec.raw)))
 
         return direct_nodes, indirect_nodes
 
diff --git a/core/dbt/parser/manifest.py b/core/dbt/parser/manifest.py
index 29f93b5bae2..80fac715178 100644
--- a/core/dbt/parser/manifest.py
+++ b/core/dbt/parser/manifest.py
@@ -18,7 +18,7 @@
     get_adapter_package_names,
 )
 from dbt.helper_types import PathSet
-from dbt.events.functions import fire_event, get_invocation_id
+from dbt.events.functions import fire_event, get_invocation_id, warn_or_error
 from dbt.events.types import (
     PartialParsingFullReparseBecauseOfError,
     PartialParsingExceptionFile,
@@ -35,10 +35,10 @@
     PartialParsingNotEnabled,
     ParsedFileLoadFailed,
     PartialParseSaveFileNotFound,
-    InvalidDisabledSourceInTestNode,
-    InvalidRefInTestNode,
+    InvalidDisabledTargetInTestNode,
     PartialParsingProjectEnvVarsChanged,
     PartialParsingProfileEnvVarsChanged,
+    NodeNotFoundOrDisabled,
 )
 from dbt.logger import DbtProcessState
 from dbt.node_types import NodeType
@@ -71,11 +71,7 @@
 )
 from dbt.contracts.util import Writable
 from dbt.exceptions import (
-    ref_target_not_found,
-    get_target_not_found_or_disabled_msg,
     target_not_found,
-    get_not_found_or_disabled_msg,
-    warn_or_error,
 )
 from dbt.parser.base import Parser
 from dbt.parser.analysis import AnalysisParser
@@ -90,7 +86,6 @@
 from dbt.parser.seeds import SeedParser
 from dbt.parser.snapshots import SnapshotParser
 from dbt.parser.sources import SourcePatcher
-from dbt.ui import warning_tag
 from dbt.version import __version__
 
 from dbt.dataclass_schema import StrEnum, dbtClassMixin
@@ -955,65 +950,43 @@ def process_nodes(self):
         self.manifest.rebuild_ref_lookup()
 
 
-def invalid_ref_fail_unless_test(node, target_model_name, target_model_package, disabled):
-
-    if node.resource_type == NodeType.Test:
-        msg = get_target_not_found_or_disabled_msg(
-            node=node,
-            target_name=target_model_name,
-            target_package=target_model_package,
-            disabled=disabled,
-        )
-        if disabled:
-            fire_event(InvalidRefInTestNode(msg=msg))
-        else:
-            warn_or_error(msg, log_fmt=warning_tag("{}"))
-    else:
-        ref_target_not_found(
-            node,
-            target_model_name,
-            target_model_package,
-            disabled=disabled,
-        )
-
-
-def invalid_source_fail_unless_test(node, target_name, target_table_name, disabled):
+def invalid_target_fail_unless_test(
+    node,
+    target_name: str,
+    target_kind: str,
+    target_package: Optional[str] = None,
+    disabled: Optional[bool] = None,
+):
     if node.resource_type == NodeType.Test:
-        msg = get_not_found_or_disabled_msg(
-            node=node,
-            target_name=f"{target_name}.{target_table_name}",
-            target_kind="source",
-            disabled=disabled,
-        )
         if disabled:
-            fire_event(InvalidDisabledSourceInTestNode(msg=msg))
+            fire_event(
+                InvalidDisabledTargetInTestNode(
+                    resource_type_title=node.resource_type.title(),
+                    unique_id=node.unique_id,
+                    original_file_path=node.original_file_path,
+                    target_kind=target_kind,
+                    target_name=target_name,
+                    target_package=target_package if target_package else "",
+                )
+            )
         else:
-            warn_or_error(msg, log_fmt=warning_tag("{}"))
-    else:
-        target_not_found(
-            node=node,
-            target_name=f"{target_name}.{target_table_name}",
-            target_kind="source",
-            disabled=disabled,
-        )
-
-
-def invalid_metric_fail_unless_test(node, target_metric_name, target_metric_package, disabled):
-
-    if node.resource_type == NodeType.Test:
-        msg = get_target_not_found_or_disabled_msg(
-            node=node,
-            target_name=target_metric_name,
-            target_package=target_metric_package,
-            disabled=disabled,
-        )
-        warn_or_error(msg, log_fmt=warning_tag("{}"))
+            warn_or_error(
+                NodeNotFoundOrDisabled(
+                    original_file_path=node.original_file_path,
+                    unique_id=node.unique_id,
+                    resource_type_title=node.resource_type.title(),
+                    target_name=target_name,
+                    target_kind=target_kind,
+                    target_package=target_package if target_package else "",
+                    disabled=str(disabled),
+                )
+            )
     else:
         target_not_found(
             node=node,
-            target_name=target_metric_name,
-            target_kind="metric",
-            target_package=target_metric_package,
+            target_name=target_name,
+            target_kind=target_kind,
+            target_package=target_package,
             disabled=disabled,
         )
 
@@ -1121,11 +1094,6 @@ def _process_docs_for_metrics(context: Dict[str, Any], metric: ParsedMetric) ->
     metric.description = get_rendered(metric.description, context)
 
 
-# TODO: this isn't actually referenced anywhere?
-def _process_derived_metrics(context: Dict[str, Any], metric: ParsedMetric) -> None:
-    metric.description = get_rendered(metric.description, context)
-
-
 def _process_refs_for_exposure(manifest: Manifest, current_project: str, exposure: ParsedExposure):
     """Given a manifest and exposure in that manifest, process its refs"""
     for ref in exposure.refs:
@@ -1153,10 +1121,11 @@ def _process_refs_for_exposure(manifest: Manifest, current_project: str, exposur
             # This may raise. Even if it doesn't, we don't want to add
             # this exposure to the graph b/c there is no destination exposure
             exposure.config.enabled = False
-            invalid_ref_fail_unless_test(
-                exposure,
-                target_model_name,
-                target_model_package,
+            invalid_target_fail_unless_test(
+                node=exposure,
+                target_name=target_model_name,
+                target_kind="node",
+                target_package=target_model_package,
                 disabled=(isinstance(target_model, Disabled)),
             )
 
@@ -1195,13 +1164,13 @@ def _process_refs_for_metric(manifest: Manifest, current_project: str, metric: P
             # This may raise. Even if it doesn't, we don't want to add
             # this metric to the graph b/c there is no destination metric
             metric.config.enabled = False
-            invalid_ref_fail_unless_test(
-                metric,
-                target_model_name,
-                target_model_package,
+            invalid_target_fail_unless_test(
+                node=metric,
+                target_name=target_model_name,
+                target_kind="node",
+                target_package=target_model_package,
                 disabled=(isinstance(target_model, Disabled)),
             )
-
             continue
 
         target_model_id = target_model.unique_id
@@ -1239,13 +1208,13 @@ def _process_metrics_for_node(
             # This may raise. Even if it doesn't, we don't want to add
             # this node to the graph b/c there is no destination node
             node.config.enabled = False
-            invalid_metric_fail_unless_test(
-                node,
-                target_metric_name,
-                target_metric_package,
+            invalid_target_fail_unless_test(
+                node=node,
+                target_name=target_metric_name,
+                target_kind="source",
+                target_package=target_metric_package,
                 disabled=(isinstance(target_metric, Disabled)),
             )
-
             continue
 
         target_metric_id = target_metric.unique_id
@@ -1280,13 +1249,13 @@ def _process_refs_for_node(manifest: Manifest, current_project: str, node: Manif
             # This may raise. Even if it doesn't, we don't want to add
             # this node to the graph b/c there is no destination node
             node.config.enabled = False
-            invalid_ref_fail_unless_test(
-                node,
-                target_model_name,
-                target_model_package,
+            invalid_target_fail_unless_test(
+                node=node,
+                target_name=target_model_name,
+                target_kind="node",
+                target_package=target_model_package,
                 disabled=(isinstance(target_model, Disabled)),
             )
-
             continue
 
         target_model_id = target_model.unique_id
@@ -1312,8 +1281,11 @@ def _process_sources_for_exposure(
         )
         if target_source is None or isinstance(target_source, Disabled):
             exposure.config.enabled = False
-            invalid_source_fail_unless_test(
-                exposure, source_name, table_name, disabled=(isinstance(target_source, Disabled))
+            invalid_target_fail_unless_test(
+                node=exposure,
+                target_name=f"{source_name}.{table_name}",
+                target_kind="source",
+                disabled=(isinstance(target_source, Disabled)),
             )
             continue
         target_source_id = target_source.unique_id
@@ -1332,8 +1304,11 @@ def _process_sources_for_metric(manifest: Manifest, current_project: str, metric
         )
         if target_source is None or isinstance(target_source, Disabled):
             metric.config.enabled = False
-            invalid_source_fail_unless_test(
-                metric, source_name, table_name, disabled=(isinstance(target_source, Disabled))
+            invalid_target_fail_unless_test(
+                node=metric,
+                target_name=f"{source_name}.{table_name}",
+                target_kind="source",
+                disabled=(isinstance(target_source, Disabled)),
             )
             continue
         target_source_id = target_source.unique_id
@@ -1354,8 +1329,11 @@ def _process_sources_for_node(manifest: Manifest, current_project: str, node: Ma
         if target_source is None or isinstance(target_source, Disabled):
             # this folows the same pattern as refs
             node.config.enabled = False
-            invalid_source_fail_unless_test(
-                node, source_name, table_name, disabled=(isinstance(target_source, Disabled))
+            invalid_target_fail_unless_test(
+                node=node,
+                target_name=f"{source_name}.{table_name}",
+                target_kind="source",
+                disabled=(isinstance(target_source, Disabled)),
             )
             continue
         target_source_id = target_source.unique_id
diff --git a/core/dbt/parser/schemas.py b/core/dbt/parser/schemas.py
index 8b22427cb39..d47c2a29684 100644
--- a/core/dbt/parser/schemas.py
+++ b/core/dbt/parser/schemas.py
@@ -50,7 +50,6 @@
     UnparsedSourceDefinition,
 )
 from dbt.exceptions import (
-    warn_invalid_patch,
     validator_error_message,
     JSONValidationException,
     raise_invalid_property_yml_version,
@@ -60,9 +59,10 @@
     raise_duplicate_macro_patch_name,
     InternalException,
     raise_duplicate_source_patch_name,
-    warn_or_error,
     CompilationException,
 )
+from dbt.events.functions import warn_or_error
+from dbt.events.types import WrongResourceSchemaFile, NoNodeForYamlKey, MacroPatchNotFound
 from dbt.node_types import NodeType
 from dbt.parser.base import SimpleParser
 from dbt.parser.search import FileBlock
@@ -74,7 +74,6 @@
     TestBlock,
     Testable,
 )
-from dbt.ui import warning_tag
 from dbt.utils import get_pseudo_test_path, coerce_dict_str
 
 
@@ -873,7 +872,15 @@ def parse_patch(self, block: TargetBlock[NodeTarget], refs: ParserRef) -> None:
             if unique_id:
                 resource_type = NodeType(unique_id.split(".")[0])
                 if resource_type.pluralize() != patch.yaml_key:
-                    warn_invalid_patch(patch, resource_type)
+                    warn_or_error(
+                        WrongResourceSchemaFile(
+                            patch_name=patch.name,
+                            resource_type=resource_type,
+                            plural_resource_type=resource_type.pluralize(),
+                            yaml_key=patch.yaml_key,
+                            file_path=patch.original_file_path,
+                        )
+                    )
                     return
 
         elif patch.yaml_key == "analyses":
@@ -912,12 +919,13 @@ def parse_patch(self, block: TargetBlock[NodeTarget], refs: ParserRef) -> None:
 
                     node.patch(patch)
             else:
-                msg = (
-                    f"Did not find matching node for patch with name '{patch.name}' "
-                    f"in the '{patch.yaml_key}' section of "
-                    f"file '{source_file.path.original_file_path}'"
+                warn_or_error(
+                    NoNodeForYamlKey(
+                        patch_name=patch.name,
+                        yaml_key=patch.yaml_key,
+                        file_path=source_file.path.original_file_path,
+                    )
                 )
-                warn_or_error(msg, log_fmt=warning_tag("{}"))
                 return
 
         # patches can't be overwritten
@@ -977,8 +985,7 @@ def parse_patch(self, block: TargetBlock[UnparsedMacroUpdate], refs: ParserRef)
         unique_id = f"macro.{patch.package_name}.{patch.name}"
         macro = self.manifest.macros.get(unique_id)
         if not macro:
-            msg = f'Found patch for macro "{patch.name}" ' f"which was not found"
-            warn_or_error(msg, log_fmt=warning_tag("{}"))
+            warn_or_error(MacroPatchNotFound(patch_name=patch.name))
             return
         if macro.patch_path:
             package_name, existing_file_path = macro.patch_path.split("://")
diff --git a/core/dbt/parser/sources.py b/core/dbt/parser/sources.py
index 1c55281db56..30440076440 100644
--- a/core/dbt/parser/sources.py
+++ b/core/dbt/parser/sources.py
@@ -1,6 +1,6 @@
 import itertools
 from pathlib import Path
-from typing import Iterable, Dict, Optional, Set, Any
+from typing import Iterable, Dict, Optional, Set, Any, List
 from dbt.adapters.factory import get_adapter
 from dbt.config import RuntimeConfig
 from dbt.context.context_config import (
@@ -24,11 +24,12 @@
     UnparsedColumn,
     Time,
 )
-from dbt.exceptions import warn_or_error, InternalException
+from dbt.events.functions import warn_or_error
+from dbt.events.types import UnusedTables
+from dbt.exceptions import InternalException
 from dbt.node_types import NodeType
 
 from dbt.parser.schemas import SchemaParser, ParserRef
-from dbt import ui
 
 
 # An UnparsedSourceDefinition is taken directly from the yaml
@@ -307,28 +308,27 @@ def warn_unused(self) -> None:
                     unused_tables[key] = unused
 
         if unused_tables:
-            msg = self.get_unused_msg(unused_tables)
-            warn_or_error(msg, log_fmt=ui.warning_tag("{}"))
+            unused_tables_formatted = self.get_unused_msg(unused_tables)
+            warn_or_error(UnusedTables(unused_tables=unused_tables_formatted))
 
         self.manifest.source_patches = {}
 
     def get_unused_msg(
         self,
         unused_tables: Dict[SourceKey, Optional[Set[str]]],
-    ) -> str:
-        msg = [
-            "During parsing, dbt encountered source overrides that had no target:",
-        ]
+    ) -> List:
+        unused_tables_formatted = []
         for key, table_names in unused_tables.items():
             patch = self.manifest.source_patches[key]
             patch_name = f"{patch.overrides}.{patch.name}"
             if table_names is None:
-                msg.append(f"  - Source {patch_name} (in {patch.path})")
+                unused_tables_formatted.append(f"  - Source {patch_name} (in {patch.path})")
             else:
                 for table_name in sorted(table_names):
-                    msg.append(f"  - Source table {patch_name}.{table_name} " f"(in {patch.path})")
-        msg.append("")
-        return "\n".join(msg)
+                    unused_tables_formatted.append(
+                        f"  - Source table {patch_name}.{table_name} " f"(in {patch.path})"
+                    )
+        return unused_tables_formatted
 
 
 def merge_freshness_time_thresholds(
diff --git a/core/dbt/task/list.py b/core/dbt/task/list.py
index df0a181ba5c..43cd8e3f8fe 100644
--- a/core/dbt/task/list.py
+++ b/core/dbt/task/list.py
@@ -5,7 +5,9 @@
 from dbt.task.runnable import GraphRunnableTask, ManifestTask
 from dbt.task.test import TestSelector
 from dbt.node_types import NodeType
-from dbt.exceptions import RuntimeException, InternalException, warn_or_error
+from dbt.events.functions import warn_or_error
+from dbt.events.types import NoNodesSelected
+from dbt.exceptions import RuntimeException, InternalException
 from dbt.logger import log_manager
 import logging
 import dbt.events.functions as event_logger
@@ -69,7 +71,7 @@ def _iterate_selected_nodes(self):
         spec = self.get_selection_spec()
         nodes = sorted(selector.get_selected(spec))
         if not nodes:
-            warn_or_error("No nodes selected!")
+            warn_or_error(NoNodesSelected())
             return
         if self.manifest is None:
             raise InternalException("manifest is None in _iterate_selected_nodes")
diff --git a/core/dbt/task/printer.py b/core/dbt/task/printer.py
index 3861b41bef2..edb2592d194 100644
--- a/core/dbt/task/printer.py
+++ b/core/dbt/task/printer.py
@@ -120,6 +120,8 @@ def print_run_result_error(result, newline: bool = True, is_warning: bool = Fals
     elif result.message is not None:
         first = True
         for line in result.message.split("\n"):
+            # TODO: why do we format like this?  Is there a reason this needs to
+            # be split instead of sending it as a single log line?
             if first:
                 fire_event(FirstRunResultError(msg=line))
                 first = False
diff --git a/core/dbt/task/runnable.py b/core/dbt/task/runnable.py
index af0de610c98..f12ce94f830 100644
--- a/core/dbt/task/runnable.py
+++ b/core/dbt/task/runnable.py
@@ -26,7 +26,7 @@
     ModelMetadata,
     NodeCount,
 )
-from dbt.events.functions import fire_event
+from dbt.events.functions import fire_event, warn_or_error
 from dbt.events.types import (
     EmptyLine,
     PrintCancelLine,
@@ -36,6 +36,7 @@
     QueryCancelationUnsupported,
     ConcurrencyLine,
     EndRunResult,
+    NothingToDo,
 )
 from dbt.contracts.graph.compiled import CompileResultNode
 from dbt.contracts.graph.manifest import Manifest
@@ -47,7 +48,6 @@
     NotImplementedException,
     RuntimeException,
     FailFastException,
-    warn_or_error,
 )
 
 from dbt.graph import GraphQueue, NodeSelector, SelectionSpec, parse_difference, Graph
@@ -57,7 +57,6 @@
 import dbt.exceptions
 from dbt import flags
 import dbt.utils
-from dbt.ui import warning_tag
 
 RESULT_FILE_NAME = "run_results.json"
 MANIFEST_FILE_NAME = "manifest.json"
@@ -459,8 +458,7 @@ def run(self):
         if len(self._flattened_nodes) == 0:
             with TextOnly():
                 fire_event(EmptyLine())
-            msg = "Nothing to do. Try checking your model configs and model specification args"
-            warn_or_error(msg, log_fmt=warning_tag("{}"))
+            warn_or_error(NothingToDo())
             result = self.get_result(
                 results=[],
                 generated_at=datetime.utcnow(),
diff --git a/test/unit/test_config.py b/test/unit/test_config.py
index 8ca8238a7d0..867ed40d38c 100644
--- a/test/unit/test_config.py
+++ b/test/unit/test_config.py
@@ -1084,35 +1084,6 @@ def test_archive_not_allowed(self):
         with self.assertRaises(dbt.exceptions.DbtProjectError):
             self.get_project()
 
-    def test__no_unused_resource_config_paths(self):
-        self.default_project_data.update({
-            'models': model_config,
-            'seeds': {},
-        })
-        project = self.from_parts()
-
-        resource_fqns = {'models': model_fqns}
-        unused = project.get_unused_resource_config_paths(resource_fqns, [])
-        self.assertEqual(len(unused), 0)
-
-    def test__unused_resource_config_paths(self):
-        self.default_project_data.update({
-            'models': model_config['my_package_name'],
-            'seeds': {},
-        })
-        project = self.from_parts()
-
-        resource_fqns = {'models': model_fqns}
-        unused = project.get_unused_resource_config_paths(resource_fqns, [])
-        self.assertEqual(len(unused), 3)
-
-    def test__get_unused_resource_config_paths_empty(self):
-        project = self.from_parts()
-        unused = project.get_unused_resource_config_paths({'models': frozenset((
-            ('my_test_project', 'foo', 'bar'),
-            ('my_test_project', 'foo', 'baz'),
-        ))}, [])
-        self.assertEqual(len(unused), 0)
 
     def test__warn_for_unused_resource_config_paths_empty(self):
         project = self.from_parts()
@@ -1172,26 +1143,17 @@ def from_parts(self, exc=None):
         else:
             return err
 
-    def test__get_unused_resource_config_paths(self):
-        project = self.from_parts()
-        unused = project.get_unused_resource_config_paths(self.used, [])
-        self.assertEqual(len(unused), 1)
-        self.assertEqual(unused[0], ('models', 'my_test_project', 'baz'))
 
-    @mock.patch.object(dbt.config.runtime, 'warn_or_error')
-    def test__warn_for_unused_resource_config_paths(self, warn_or_error):
+    def test__warn_for_unused_resource_config_paths(self):
         project = self.from_parts()
-        project.warn_for_unused_resource_config_paths(self.used, [])
-        warn_or_error.assert_called_once()
-
-    def test__warn_for_unused_resource_config_paths_disabled(self):
-        project = self.from_parts()
-        unused = project.get_unused_resource_config_paths(
-            self.used,
-            frozenset([('my_test_project', 'baz')])
-        )
-
-        self.assertEqual(len(unused), 0)
+        with mock.patch('dbt.config.runtime.warn_or_error') as warn_or_error_patch:
+            project.warn_for_unused_resource_config_paths(self.used, [])
+            warn_or_error_patch.assert_called_once()
+            event = warn_or_error_patch.call_args[0][0]
+            assert event.info.name == 'UnsedResourceConfigPath'
+            msg = event.info.msg
+            expected_msg = "- models.my_test_project.baz"
+            assert expected_msg in msg
 
 
 class TestRuntimeConfigFiles(BaseFileTest):
diff --git a/test/unit/test_graph_selector_methods.py b/test/unit/test_graph_selector_methods.py
index e32267e2d6f..55559b13e17 100644
--- a/test/unit/test_graph_selector_methods.py
+++ b/test/unit/test_graph_selector_methods.py
@@ -981,7 +981,9 @@ def test_select_state_changed_seed_checksum_path_to_path(manifest, previous_stat
     with mock.patch('dbt.contracts.graph.parsed.warn_or_error') as warn_or_error_patch:
         assert not search_manifest_using_method(manifest, method, 'modified')
         warn_or_error_patch.assert_called_once()
-        msg = warn_or_error_patch.call_args[0][0]
+        event = warn_or_error_patch.call_args[0][0]
+        assert event.info.name == 'SeedExceedsLimitSamePath'
+        msg = event.info.msg
         assert msg.startswith('Found a seed (pkg.seed) >1MB in size')
     with mock.patch('dbt.contracts.graph.parsed.warn_or_error') as warn_or_error_patch:
         assert not search_manifest_using_method(manifest, method, 'new')
@@ -996,7 +998,9 @@ def test_select_state_changed_seed_checksum_sha_to_path(manifest, previous_state
         assert search_manifest_using_method(
             manifest, method, 'modified') == {'seed'}
         warn_or_error_patch.assert_called_once()
-        msg = warn_or_error_patch.call_args[0][0]
+        event = warn_or_error_patch.call_args[0][0]
+        assert event.info.name == 'SeedIncreased'
+        msg = event.info.msg
         assert msg.startswith('Found a seed (pkg.seed) >1MB in size')
     with mock.patch('dbt.contracts.graph.parsed.warn_or_error') as warn_or_error_patch:
         assert not search_manifest_using_method(manifest, method, 'new')
diff --git a/tests/unit/test_events.py b/tests/unit/test_events.py
index c2064b84c1a..6ba1b1ba69c 100644
--- a/tests/unit/test_events.py
+++ b/tests/unit/test_events.py
@@ -289,8 +289,14 @@ def MockNode():
     PartialParsingUpdateSchemaFile(file_id=""),
     PartialParsingDeletedSource(unique_id=""),
     PartialParsingDeletedExposure(unique_id=""),
-    InvalidDisabledSourceInTestNode(msg=""),
-    InvalidRefInTestNode(msg=""),
+    InvalidDisabledTargetInTestNode(
+        resource_type_title="",
+        unique_id="",
+        original_file_path="",
+        target_kind="",
+        target_name="",
+        target_package="",
+    ),
     RunningOperationCaughtError(exc=""),
     RunningOperationUncaughtError(exc=""),
     DbtProjectError(),
@@ -420,8 +426,6 @@ def MockNode():
     FlushEventsFailure(),
     TrackingInitializeFailure(),
     RetryExternalCall(attempt=0, max=0),
-    GeneralWarningMsg(msg="", log_fmt=""),
-    GeneralWarningException(exc="", log_fmt=""),
     PartialParsingProfileEnvVarsChanged(),
     AdapterEventDebug(name="", base_msg="", args=()),
     AdapterEventInfo(name="", base_msg="", args=()),


Code review comments:
diff_hunk: @@ -16,17 +16,16 @@
 from dbt.contracts.connection import AdapterRequiredConfig, Credentials
 from dbt.contracts.graph.manifest import ManifestMetadata
 from dbt.contracts.relation import ComponentName
-from dbt.ui import warning_tag
 
 from dbt.contracts.project import Configuration, UserConfig
 from dbt.exceptions import (
     RuntimeException,
     DbtProjectError,
     validator_error_message,
-    warn_or_error,
     raise_compiler_error,
 )
-
+from dbt.events.functions import warn_or_error
+from dbt.events.types import UnsedResourceConfigPath
path: core/dbt/config/runtime.py
text: @user1:
very cool PR! caught one typo while reading through
```suggestion
from dbt.events.types import UnusedResourceConfigPath
```

@author:
Good eye!