diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml
index 44600c712e588..57c2fb7ff9110 100644
--- a/.github/workflows/integration-test.yml
+++ b/.github/workflows/integration-test.yml
@@ -55,6 +55,7 @@ jobs:
       fail-fast: false
       matrix:
         include:
+          - test: 'amqp'
           - test: 'apex'
           - test: 'aws'
           - test: 'axiom'
diff --git a/Cargo.lock b/Cargo.lock
index 03dba1f41a2a2..43553d1f4071a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -61,6 +61,54 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "amq-protocol"
+version = "7.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "acc7cad07d1b4533fcb46f0819a6126fa201fd0385469aba75e405424f3fe009"
+dependencies = [
+ "amq-protocol-tcp",
+ "amq-protocol-types",
+ "amq-protocol-uri",
+ "cookie-factory",
+ "nom",
+ "serde",
+]
+
+[[package]]
+name = "amq-protocol-tcp"
+version = "7.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d8b20aba8c35a0b885e1e978eff456ced925730a4e012e63e4ff89a1deb602b"
+dependencies = [
+ "amq-protocol-uri",
+ "tcp-stream",
+ "tracing 0.1.34",
+]
+
+[[package]]
+name = "amq-protocol-types"
+version = "7.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e245e0e9083b6a6db5f8c10013074cb382266eb9e2a37204d19c651b8d3b8114"
+dependencies = [
+ "cookie-factory",
+ "nom",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "amq-protocol-uri"
+version = "7.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "56987108bf48d2eb500cae8896cd9291564eedd8744776ecc5c3338a8b2ca5f8"
+dependencies = [
+ "amq-protocol-types",
+ "percent-encoding",
+ "url",
+]
+
 [[package]]
 name = "ansi_term"
 version = "0.12.1"
@@ -223,6 +271,33 @@ dependencies = [
  "futures-lite",
 ]
 
+[[package]]
+name = "async-global-executor"
+version = "2.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5262ed948da60dd8956c6c5aca4d4163593dddb7b32d73267c93dab7b2e98940"
+dependencies = [
+ "async-channel",
+ "async-executor",
+ "async-io",
+ "async-lock",
+ "blocking",
+ "futures-lite",
+ "num_cpus",
+ "once_cell",
+]
+
+[[package]]
+name = "async-global-executor-trait"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "33dd14c5a15affd2abcff50d84efd4009ada28a860f01c14f9d654f3e81b3f75"
+dependencies = [
+ "async-global-executor",
+ "async-trait",
+ "executor-trait",
+]
+
 [[package]]
 name = "async-graphql"
 version = "4.0.13"
@@ -362,6 +437,18 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "async-reactor-trait"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a6012d170ad00de56c9ee354aef2e358359deb1ec504254e0e5a3774771de0e"
+dependencies = [
+ "async-io",
+ "async-trait",
+ "futures-core",
+ "reactor-trait",
+]
+
 [[package]]
 name = "async-recursion"
 version = "1.0.0"
@@ -1708,6 +1795,12 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e"
 
+[[package]]
+name = "cookie-factory"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "396de984970346b0d9e93d1415082923c679e5ae5c3ee3dcbd104f5610af126b"
+
 [[package]]
 name = "core-foundation"
 version = "0.9.3"
@@ -2515,6 +2608,15 @@ version = "2.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77f3309417938f28bf8228fcff79a4a37103981e3e186d2ccd19c74b38f4eb71"
 
+[[package]]
+name = "executor-trait"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1a1052dd43212a7777ec6a69b117da52f5e52f07aec47d00c1a2b33b85d06b08"
+dependencies = [
+ "async-trait",
+]
+
 [[package]]
 name = "exitcode"
 version = "1.1.2"
@@ -2633,6 +2735,18 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b59b6469c35ab601d6487d28879bccfbe8c896c33a3fe699c4d29817e552cc58"
 
+[[package]]
+name = "flume"
+version = "0.10.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1657b4441c3403d9f7b3409e47575237dac27b1b5726df654a6ecbf92f0f7577"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "pin-project",
+ "spin 0.9.3",
+]
+
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -3879,6 +3993,28 @@ dependencies = [
  "regex",
 ]
 
+[[package]]
+name = "lapin"
+version = "2.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bd03ea5831b44775e296239a64851e2fd14a80a363d202ba147009ffc994ff0f"
+dependencies = [
+ "amq-protocol",
+ "async-global-executor-trait",
+ "async-reactor-trait",
+ "async-trait",
+ "executor-trait",
+ "flume",
+ "futures-core",
+ "futures-io",
+ "parking_lot",
+ "pinky-swear",
+ "reactor-trait",
+ "serde",
+ "tracing 0.1.34",
+ "waker-fn",
+]
+
 [[package]]
 name = "lazy_static"
 version = "1.4.0"
@@ -5316,6 +5452,18 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
+[[package]]
+name = "pinky-swear"
+version = "6.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d894b67aa7a4bf295db5e85349078c604edaa6fa5c8721e8eca3c7729a27f2ac"
+dependencies = [
+ "doc-comment",
+ "flume",
+ "parking_lot",
+ "tracing 0.1.34",
+]
+
 [[package]]
 name = "pkcs8"
 version = "0.7.6"
@@ -5997,6 +6145,17 @@ dependencies = [
  "zstd-sys",
 ]
 
+[[package]]
+name = "reactor-trait"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "438a4293e4d097556730f4711998189416232f009c137389e0f961d2bc0ddc58"
+dependencies = [
+ "async-trait",
+ "futures-core",
+ "futures-io",
+]
+
 [[package]]
 name = "redis"
 version = "0.21.6"
@@ -7094,6 +7253,9 @@ name = "spin"
 version = "0.9.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c530c2b0d0bf8b69304b39fe2001993e267461948b890cd037d8ad4293fa1a0d"
+dependencies = [
+ "lock_api",
+]
 
 [[package]]
 name = "spki"
@@ -7277,6 +7439,15 @@ version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f764005d11ee5f36500a149ace24e00e3da98b0158b3e2d53a7495660d3f4d60"
 
+[[package]]
+name = "tcp-stream"
+version = "0.24.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09a4b0a70bac0a58ca6a7659d1328e34ee462339c70b0fa49f72bad1f278910a"
+dependencies = [
+ "cfg-if 1.0.0",
+]
+
 [[package]]
 name = "temp-dir"
 version = "0.1.11"
@@ -8510,6 +8681,7 @@ dependencies = [
  "itertools 0.10.4",
  "k8s-openapi",
  "kube",
+ "lapin",
  "libc",
  "listenfd",
  "logfmt",
diff --git a/Cargo.toml b/Cargo.toml
index d2fb56096ed98..369d0e84afc99 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -198,6 +198,9 @@ prost-types = { version = "0.10.1", default-features = false, optional = true }
 goauth = { version = "0.13.1", optional = true }
 smpl_jwt = { version = "0.7.1", default-features = false, optional = true }
 
+# AMQP
+lapin = { version = "2.1.1", default-features = false, optional = true }
+
 # API
 async-graphql = { version = "4.0.13", default-features = false, optional = true, features = ["chrono"] }
 async-graphql-warp = { version = "4.0.13", default-features = false, optional = true }
@@ -442,6 +445,7 @@ enrichment-tables-geoip = ["dep:maxminddb"]
 # Sources
 sources = ["sources-logs", "sources-metrics"]
 sources-logs = [
+  "sources-amqp",
   "sources-aws_kinesis_firehose",
   "sources-aws_s3",
   "sources-aws_sqs",
@@ -484,6 +488,7 @@ sources-metrics = [
   "sources-vector",
 ]
 
+sources-amqp = ["lapin"]
 sources-apache_metrics = []
 sources-aws_ecs_metrics = []
 sources-aws_kinesis_firehose = ["dep:base64", "dep:infer"]
@@ -577,6 +582,7 @@ transforms-throttle = ["dep:governor"]
 # Sinks
 sinks = ["sinks-logs", "sinks-metrics"]
 sinks-logs = [
+  "sinks-amqp",
   "sinks-apex",
   "sinks-aws_cloudwatch_logs",
   "sinks-aws_kinesis_firehose",
@@ -631,6 +637,7 @@ sinks-metrics = [
   "sinks-splunk_hec"
 ]
 
+sinks-amqp = ["lapin"]
 sinks-apex = []
 sinks-aws_cloudwatch_logs = ["aws-core", "dep:aws-sdk-cloudwatchlogs"]
 sinks-aws_cloudwatch_metrics = ["aws-core", "dep:aws-sdk-cloudwatch"]
@@ -693,6 +700,7 @@ nightly = []
 
 # Testing-related features
 all-integration-tests = [
+  "amqp-integration-tests",
   "apex-integration-tests",
   "aws-integration-tests",
   "axiom-integration-tests",
@@ -728,6 +736,8 @@ all-integration-tests = [
   "dnstap-integration-tests",
 ]
 
+amqp-integration-tests = ["sources-amqp", "sinks-amqp"]
+
 aws-integration-tests = [
   "aws-cloudwatch-logs-integration-tests",
   "aws-cloudwatch-metrics-integration-tests",
@@ -877,4 +887,3 @@ name = "codecs"
 path = "benches/codecs/main.rs"
 harness = false
 required-features = ["codecs-benches"]
-
diff --git a/Makefile b/Makefile
index 01f6db048f217..516bf9d5d59c5 100644
--- a/Makefile
+++ b/Makefile
@@ -334,7 +334,7 @@ test-enterprise: ## Runs enterprise related behavioral tests
 
 .PHONY: test-integration
 test-integration: ## Runs all integration tests
-test-integration: test-integration-apex test-integration-aws test-integration-axiom test-integration-azure test-integration-clickhouse test-integration-docker-logs test-integration-elasticsearch
+test-integration: test-integration-amqp test-integration-apex test-integration-aws test-integration-axiom test-integration-azure test-integration-clickhouse test-integration-docker-logs test-integration-elasticsearch
 test-integration: test-integration-eventstoredb test-integration-fluent test-integration-gcp test-integration-humio test-integration-http-scrape test-integration-influxdb
 test-integration: test-integration-kafka test-integration-logstash test-integration-loki test-integration-mongodb test-integration-nats
 test-integration: test-integration-nginx test-integration-opentelemetry test-integration-postgres test-integration-prometheus test-integration-pulsar
diff --git a/scripts/integration/docker-compose.amqp.yml b/scripts/integration/docker-compose.amqp.yml
new file mode 100644
index 0000000000000..c3eca99cf70b9
--- /dev/null
+++ b/scripts/integration/docker-compose.amqp.yml
@@ -0,0 +1,36 @@
+version: "3"
+
+services:
+  rabbitmq:
+    image: docker.io/rabbitmq:3.8
+    ports:
+      - 5672:5672
+  runner:
+    build:
+      context:  ${PWD}
+      dockerfile: scripts/integration/Dockerfile
+      args:
+        - RUST_VERSION=${RUST_VERSION}
+    working_dir: /code
+    command:
+      - "cargo"
+      - "nextest"
+      - "run"
+      - "--no-fail-fast"
+      - "--no-default-features"
+      - "--features"
+      - "amqp-integration-tests"
+      - "--lib"
+      - "${FILTER:-::amqp::}"
+    depends_on:
+      - rabbitmq
+    volumes:
+      - ${PWD}:/code
+      - target:/code/target
+      - cargogit:/usr/local/cargo/git
+      - cargoregistry:/usr/local/cargo/registry
+
+volumes:
+  target: {}
+  cargogit: {}
+  cargoregistry: {}
diff --git a/src/amqp.rs b/src/amqp.rs
new file mode 100644
index 0000000000000..82059e82c97d0
--- /dev/null
+++ b/src/amqp.rs
@@ -0,0 +1,68 @@
+use lapin::tcp::{OwnedIdentity, OwnedTLSConfig};
+use vector_config::configurable_component;
+
+/// Connection options for `AMQP`.
+#[configurable_component]
+#[derive(Clone, Debug)]
+pub(crate) struct AmqpConfig {
+    /// URI for the `AMQP` server.
+    ///
+    /// Format: amqp://<user>:<password>@<host>:<port>/<vhost>?timeout=<seconds>
+    pub(crate) connection_string: String,
+
+    #[configurable(derived)]
+    pub(crate) tls: Option<crate::tls::TlsConfig>,
+}
+
+impl Default for AmqpConfig {
+    fn default() -> Self {
+        Self {
+            connection_string: "amqp://127.0.0.1/%2f".to_string(),
+            tls: None,
+        }
+    }
+}
+
+impl AmqpConfig {
+    pub(crate) async fn connect(
+        &self,
+    ) -> Result<(lapin::Connection, lapin::Channel), Box<dyn std::error::Error + Send + Sync>> {
+        debug!("Connecting to {}.", self.connection_string);
+        let addr = self.connection_string.clone();
+        let conn = match &self.tls {
+            Some(tls) => {
+                let cert_chain = if let Some(ca) = &tls.ca_file {
+                    Some(tokio::fs::read_to_string(ca.to_owned()).await?)
+                } else {
+                    None
+                };
+                let identity = if let Some(identity) = &tls.key_file {
+                    let der = tokio::fs::read(identity.to_owned()).await?;
+                    Some(OwnedIdentity {
+                        der,
+                        password: tls
+                            .key_pass
+                            .as_ref()
+                            .map(|s| s.to_string())
+                            .unwrap_or_else(String::default),
+                    })
+                } else {
+                    None
+                };
+                let tls_config = OwnedTLSConfig {
+                    identity,
+                    cert_chain,
+                };
+                lapin::Connection::connect_with_config(
+                    &addr,
+                    lapin::ConnectionProperties::default(),
+                    tls_config,
+                )
+                .await
+            }
+            None => lapin::Connection::connect(&addr, lapin::ConnectionProperties::default()).await,
+        }?;
+        let channel = conn.create_channel().await?;
+        Ok((conn, channel))
+    }
+}
diff --git a/src/app.rs b/src/app.rs
index d59b0c68a6326..a94f6e00561c9 100644
--- a/src/app.rs
+++ b/src/app.rs
@@ -88,6 +88,7 @@ impl Application {
                     "tower_limit=trace".to_owned(),
                     format!("rdkafka={}", level),
                     format!("buffers={}", level),
+                    format!("lapin={}", level),
                     format!("kube={}", level),
                 ]
                 .join(","),
diff --git a/src/internal_events/amqp.rs b/src/internal_events/amqp.rs
new file mode 100644
index 0000000000000..7034e5f2f3bfe
--- /dev/null
+++ b/src/internal_events/amqp.rs
@@ -0,0 +1,155 @@
+#[cfg(feature = "sources-amqp")]
+pub mod source {
+    use metrics::counter;
+    use vector_common::internal_event::{error_stage, error_type};
+    use vector_core::internal_event::InternalEvent;
+
+    #[derive(Debug)]
+    pub struct AmqpBytesReceived {
+        pub byte_size: usize,
+        pub protocol: &'static str,
+    }
+
+    impl InternalEvent for AmqpBytesReceived {
+        fn emit(self) {
+            trace!(
+                message = "Bytes received.",
+                byte_size = %self.byte_size,
+                protocol = %self.protocol,
+            );
+            counter!(
+                "component_received_bytes_total",
+                self.byte_size as u64,
+                "protocol" => self.protocol,
+            );
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct AmqpEventError {
+        pub error: lapin::Error,
+    }
+
+    impl InternalEvent for AmqpEventError {
+        fn emit(self) {
+            error!(message = "Failed to read message.",
+                   error = ?self.error,
+                   error_type = error_type::REQUEST_FAILED,
+                   stage = error_stage::RECEIVING,
+                   internal_log_rate_secs = 10
+            );
+            counter!(
+                "component_errors_total", 1,
+                "error_type" => error_type::REQUEST_FAILED,
+                "stage" => error_stage::RECEIVING,
+            );
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct AmqpAckError {
+        pub error: lapin::Error,
+    }
+
+    impl InternalEvent for AmqpAckError {
+        fn emit(self) {
+            error!(message = "Unable to ack.",
+                   error = ?self.error,
+                   error_type = error_type::ACKNOWLEDGMENT_FAILED,
+                   stage = error_stage::RECEIVING,
+                   internal_log_rate_secs = 10
+            );
+            counter!(
+                "component_errors_total", 1,
+                "error_type" => error_type::ACKNOWLEDGMENT_FAILED,
+                "stage" => error_stage::RECEIVING,
+            );
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct AmqpRejectError {
+        pub error: lapin::Error,
+    }
+
+    impl InternalEvent for AmqpRejectError {
+        fn emit(self) {
+            error!(message = "Unable to reject.",
+                   error = ?self.error,
+                   error_type = error_type::COMMAND_FAILED,
+                   stage = error_stage::RECEIVING,
+                   internal_log_rate_secs = 10
+            );
+            counter!(
+                "component_errors_total", 1,
+                "error_type" => error_type::COMMAND_FAILED,
+                "stage" => error_stage::RECEIVING,
+            );
+        }
+    }
+}
+
+#[cfg(feature = "sinks-amqp")]
+pub mod sink {
+    use crate::{
+        emit,
+        internal_events::{ComponentEventsDropped, UNINTENTIONAL},
+    };
+    use metrics::counter;
+    use vector_common::internal_event::{error_stage, error_type};
+    use vector_core::internal_event::InternalEvent;
+
+    #[derive(Debug)]
+    pub struct AmqpDeliveryError<'a> {
+        pub error: &'a lapin::Error,
+    }
+
+    impl InternalEvent for AmqpDeliveryError<'_> {
+        fn emit(self) {
+            let deliver_reason = "Unable to deliver.";
+
+            error!(message = deliver_reason,
+                   error = ?self.error,
+                   error_type = error_type::REQUEST_FAILED,
+                   stage = error_stage::SENDING,
+                   internal_log_rate_secs = 10
+            );
+            counter!(
+                "component_errors_total", 1,
+                "error_type" => error_type::REQUEST_FAILED,
+                "stage" => error_stage::SENDING,
+            );
+            emit!(ComponentEventsDropped::<UNINTENTIONAL> {
+                count: 1,
+                reason: deliver_reason
+            });
+        }
+    }
+
+    #[derive(Debug)]
+    pub struct AmqpAcknowledgementError<'a> {
+        pub error: &'a lapin::Error,
+    }
+
+    impl InternalEvent for AmqpAcknowledgementError<'_> {
+        fn emit(self) {
+            let ack_reason = "Acknowledgement failed.";
+
+            error!(message = ack_reason,
+                   error = ?self.error,
+                   error_type = error_type::REQUEST_FAILED,
+                   stage = error_stage::SENDING,
+                   internal_log_rate_secs = 10
+            );
+            counter!(
+                "component_errors_total", 1,
+                "error_type" => error_type::REQUEST_FAILED,
+                "stage" => error_stage::SENDING,
+            );
+            emit!(ComponentEventsDropped::<UNINTENTIONAL> {
+                count: 1,
+                reason: ack_reason
+            });
+        }
+    }
+}
diff --git a/src/internal_events/mod.rs b/src/internal_events/mod.rs
index da6007836dbae..0354665402423 100644
--- a/src/internal_events/mod.rs
+++ b/src/internal_events/mod.rs
@@ -2,6 +2,8 @@ pub mod prelude;
 
 mod adaptive_concurrency;
 mod aggregate;
+#[cfg(any(feature = "sources-amqp", feature = "sinks-amqp"))]
+mod amqp;
 #[cfg(feature = "sources-apache_metrics")]
 mod apache_metrics;
 #[cfg(feature = "api")]
@@ -133,6 +135,8 @@ pub(crate) use mongodb_metrics::*;
 
 #[cfg(feature = "transforms-aggregate")]
 pub(crate) use self::aggregate::*;
+#[cfg(any(feature = "sources-amqp", feature = "sinks-amqp"))]
+pub(crate) use self::amqp::*;
 #[cfg(feature = "sources-apache_metrics")]
 pub(crate) use self::apache_metrics::*;
 #[cfg(feature = "api")]
diff --git a/src/lib.rs b/src/lib.rs
index 7c0f321bf452e..2e93c4ec78330 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -40,6 +40,8 @@ pub mod generate_schema;
 #[macro_use]
 #[allow(unreachable_pub)]
 pub mod internal_events;
+#[cfg(feature = "lapin")]
+pub mod amqp;
 #[cfg(feature = "api")]
 #[allow(unreachable_pub)]
 pub mod api;
diff --git a/src/sinks/amqp/config.rs b/src/sinks/amqp/config.rs
new file mode 100644
index 0000000000000..a131e25438f8b
--- /dev/null
+++ b/src/sinks/amqp/config.rs
@@ -0,0 +1,104 @@
+//! Configuration functionality for the `AMQP` sink.
+use crate::{
+    amqp::AmqpConfig,
+    codecs::EncodingConfig,
+    config::{DataType, GenerateConfig, Input, SinkConfig, SinkContext},
+    sinks::{Healthcheck, VectorSink},
+    template::Template,
+};
+use codecs::TextSerializerConfig;
+use futures::FutureExt;
+use std::sync::Arc;
+use vector_config::configurable_component;
+use vector_core::config::AcknowledgementsConfig;
+
+use super::sink::AmqpSink;
+
+/// Configuration for the `amqp` sink.
+///
+/// Supports AMQP version 0.9.1
+#[configurable_component(sink("amqp"))]
+#[derive(Clone, Debug)]
+pub struct AmqpSinkConfig {
+    /// The exchange to publish messages to.
+    #[configurable(metadata(templateable))]
+    pub(crate) exchange: Template,
+
+    /// Template used to generate a routing key which corresponds to a queue binding.
+    #[configurable(metadata(templateable))]
+    pub(crate) routing_key: Option<Template>,
+
+    /// Connection options for the `amqp` sink.
+    pub(crate) connection: AmqpConfig,
+
+    #[configurable(derived)]
+    pub(crate) encoding: EncodingConfig,
+
+    #[configurable(derived)]
+    #[serde(
+        default,
+        deserialize_with = "crate::serde::bool_or_struct",
+        skip_serializing_if = "crate::serde::skip_serializing_if_default"
+    )]
+    pub(crate) acknowledgements: AcknowledgementsConfig,
+}
+
+impl Default for AmqpSinkConfig {
+    fn default() -> Self {
+        Self {
+            exchange: Template::try_from("vector").unwrap(),
+            routing_key: None,
+            encoding: TextSerializerConfig::new().into(),
+            connection: AmqpConfig::default(),
+            acknowledgements: AcknowledgementsConfig::default(),
+        }
+    }
+}
+
+impl GenerateConfig for AmqpSinkConfig {
+    fn generate_config() -> toml::Value {
+        toml::from_str(
+            r#"connection.connection_string = "amqp://localhost:5672/%2f"
+            routing_key = "user_id"
+            exchange = "test"
+            encoding.codec = "json""#,
+        )
+        .unwrap()
+    }
+}
+
+#[async_trait::async_trait]
+impl SinkConfig for AmqpSinkConfig {
+    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        let sink = AmqpSink::new(self.clone()).await?;
+        let hc = healthcheck(Arc::clone(&sink.channel)).boxed();
+        Ok((VectorSink::from_event_streamsink(sink), hc))
+    }
+
+    fn input(&self) -> Input {
+        Input::new(DataType::Log)
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.acknowledgements
+    }
+}
+
+pub(super) async fn healthcheck(channel: Arc<lapin::Channel>) -> crate::Result<()> {
+    trace!("Healthcheck started.");
+
+    if !channel.status().connected() {
+        return Err(Box::new(std::io::Error::new(
+            std::io::ErrorKind::BrokenPipe,
+            "Not Connected",
+        )));
+    }
+
+    trace!("Healthcheck completed.");
+    Ok(())
+}
+
+#[test]
+pub fn generate_config() {
+    crate::test_util::test_generate_config::<AmqpSinkConfig>();
+}
diff --git a/src/sinks/amqp/encoder.rs b/src/sinks/amqp/encoder.rs
new file mode 100644
index 0000000000000..6e86828c82923
--- /dev/null
+++ b/src/sinks/amqp/encoder.rs
@@ -0,0 +1,30 @@
+//! Encoding for the `AMQP` sink.
+use crate::{
+    event::Event,
+    sinks::util::encoding::{write_all, Encoder},
+};
+use bytes::BytesMut;
+use std::io;
+use tokio_util::codec::Encoder as _;
+
+#[derive(Clone, Debug)]
+pub(super) struct AmqpEncoder {
+    pub(super) encoder: crate::codecs::Encoder<()>,
+    pub(super) transformer: crate::codecs::Transformer,
+}
+
+impl Encoder<Event> for AmqpEncoder {
+    fn encode_input(&self, mut input: Event, writer: &mut dyn io::Write) -> io::Result<usize> {
+        let mut body = BytesMut::new();
+        self.transformer.transform(&mut input);
+        let mut encoder = self.encoder.clone();
+        encoder
+            .encode(input, &mut body)
+            .map_err(|_| io::Error::new(io::ErrorKind::Other, "unable to encode"))?;
+
+        let body = body.freeze();
+        write_all(writer, 1, body.as_ref())?;
+
+        Ok(body.len())
+    }
+}
diff --git a/src/sinks/amqp/integration_tests.rs b/src/sinks/amqp/integration_tests.rs
new file mode 100644
index 0000000000000..7199b2cbaf673
--- /dev/null
+++ b/src/sinks/amqp/integration_tests.rs
@@ -0,0 +1,218 @@
+use super::*;
+use crate::{
+    config::{SinkConfig, SinkContext},
+    serde::{default_decoding, default_framing_message_based},
+    shutdown::ShutdownSignal,
+    template::Template,
+    test_util::{
+        components::{run_and_assert_sink_compliance, SINK_TAGS},
+        random_lines_with_stream, random_string,
+    },
+    SourceSender,
+};
+use futures::StreamExt;
+use std::{sync::Arc, time::Duration};
+use vector_core::config::LogNamespace;
+
+pub fn make_config() -> AmqpSinkConfig {
+    let mut config = AmqpSinkConfig {
+        exchange: Template::try_from("it").unwrap(),
+        ..Default::default()
+    };
+    let user = std::env::var("AMQP_USER").unwrap_or_else(|_| "guest".to_string());
+    let pass = std::env::var("AMQP_PASSWORD").unwrap_or_else(|_| "guest".to_string());
+    let vhost = std::env::var("AMQP_VHOST").unwrap_or_else(|_| "%2f".to_string());
+    config.connection.connection_string =
+        format!("amqp://{}:{}@rabbitmq:5672/{}", user, pass, vhost);
+    config
+}
+
+#[tokio::test]
+async fn healthcheck() {
+    crate::test_util::trace_init();
+    let exchange = format!("test-{}-exchange", random_string(10));
+
+    let mut config = make_config();
+    config.exchange = Template::try_from(exchange.as_str()).unwrap();
+    let (_conn, channel) = config.connection.connect().await.unwrap();
+    super::config::healthcheck(Arc::new(channel)).await.unwrap();
+}
+
+#[tokio::test]
+async fn amqp_happy_path_plaintext() {
+    crate::test_util::trace_init();
+
+    amqp_happy_path().await;
+}
+
+#[tokio::test]
+async fn amqp_round_trip_plaintext() {
+    crate::test_util::trace_init();
+
+    amqp_round_trip().await;
+}
+
+async fn amqp_happy_path() {
+    let mut config = make_config();
+    let exchange = format!("test-{}-exchange", random_string(10));
+    config.exchange = Template::try_from(exchange.as_str()).unwrap();
+    let queue = format!("test-{}-queue", random_string(10));
+
+    let (_conn, channel) = config.connection.connect().await.unwrap();
+    let exchange_opts = lapin::options::ExchangeDeclareOptions {
+        auto_delete: true,
+        ..Default::default()
+    };
+    channel
+        .exchange_declare(
+            &exchange,
+            lapin::ExchangeKind::Fanout,
+            exchange_opts,
+            lapin::types::FieldTable::default(),
+        )
+        .await
+        .unwrap();
+
+    let cx = SinkContext::new_test();
+    let (sink, healthcheck) = config.build(cx).await.unwrap();
+    healthcheck.await.expect("Health check failed");
+
+    // prepare consumer
+    let queue_opts = lapin::options::QueueDeclareOptions {
+        auto_delete: true,
+        ..Default::default()
+    };
+    channel
+        .queue_declare(&queue, queue_opts, lapin::types::FieldTable::default())
+        .await
+        .unwrap();
+
+    channel
+        .queue_bind(
+            &queue,
+            &exchange,
+            "",
+            lapin::options::QueueBindOptions::default(),
+            lapin::types::FieldTable::default(),
+        )
+        .await
+        .unwrap();
+
+    let consumer = format!("test-{}-consumer", random_string(10));
+    let mut consumer = channel
+        .basic_consume(
+            &queue,
+            &consumer,
+            lapin::options::BasicConsumeOptions::default(),
+            lapin::types::FieldTable::default(),
+        )
+        .await
+        .unwrap();
+
+    let num_events = 1000;
+    let (input, events) = random_lines_with_stream(100, num_events, None);
+    run_and_assert_sink_compliance(sink, events, &SINK_TAGS).await;
+
+    // loop instead of iter so we can set a timeout
+    let mut failures = 0;
+    let mut out = Vec::new();
+    while failures < 10 && out.len() < input.len() {
+        if let Ok(Some(try_msg)) =
+            tokio::time::timeout(Duration::from_secs(10), consumer.next()).await
+        {
+            let msg = try_msg.unwrap();
+            let s = String::from_utf8_lossy(msg.data.as_slice()).into_owned();
+            out.push(s);
+        } else {
+            failures += 1;
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+    }
+
+    assert_eq!(out.len(), input.len());
+    assert_eq!(out, input);
+}
+
+async fn amqp_round_trip() {
+    let mut config = make_config();
+    let exchange = format!("test-{}-exchange", random_string(10));
+    config.exchange = Template::try_from(exchange.as_str()).unwrap();
+    let queue = format!("test-{}-queue", random_string(10));
+
+    let (_conn, channel) = config.connection.connect().await.unwrap();
+    let exchange_opts = lapin::options::ExchangeDeclareOptions {
+        auto_delete: true,
+        ..Default::default()
+    };
+    channel
+        .exchange_declare(
+            &exchange,
+            lapin::ExchangeKind::Fanout,
+            exchange_opts,
+            lapin::types::FieldTable::default(),
+        )
+        .await
+        .unwrap();
+
+    let cx = SinkContext::new_test();
+    let (amqp_sink, healthcheck) = config.build(cx).await.unwrap();
+    healthcheck.await.expect("Health check failed");
+
+    let source_cfg = crate::sources::amqp::AmqpSourceConfig {
+        connection: config.connection.clone(),
+        queue: queue.clone(),
+        consumer: format!("test-{}-amqp-source", random_string(10)),
+        routing_key_field: "routing".to_string(),
+        exchange_key: "exchange".to_string(),
+        offset_key: "offset".to_string(),
+        framing: default_framing_message_based(),
+        decoding: default_decoding(),
+        log_namespace: Some(true),
+        acknowledgements: true.into(),
+    };
+    let (tx, rx) = SourceSender::new_test();
+    let amqp_source = crate::sources::amqp::amqp_source(
+        &source_cfg,
+        ShutdownSignal::noop(),
+        tx,
+        LogNamespace::Legacy,
+        true,
+    )
+    .await
+    .unwrap();
+
+    // prepare server
+    let queue_opts = lapin::options::QueueDeclareOptions {
+        auto_delete: true,
+        ..Default::default()
+    };
+    channel
+        .queue_declare(&queue, queue_opts, lapin::types::FieldTable::default())
+        .await
+        .unwrap();
+
+    channel
+        .queue_bind(
+            &queue,
+            &exchange,
+            "",
+            lapin::options::QueueBindOptions::default(),
+            lapin::types::FieldTable::default(),
+        )
+        .await
+        .unwrap();
+
+    let _source_fut = tokio::spawn(amqp_source);
+
+    // Have sink publish events
+    let events_fut = async move {
+        let num_events = 1000;
+        let (_, events) = random_lines_with_stream(100, num_events, None);
+        run_and_assert_sink_compliance(amqp_sink, events, &SINK_TAGS).await;
+        num_events
+    };
+    let nb_events_published = tokio::spawn(events_fut).await.unwrap();
+    let output = crate::test_util::collect_n(rx, 1000).await;
+
+    assert_eq!(output.len(), nb_events_published);
+}
diff --git a/src/sinks/amqp/mod.rs b/src/sinks/amqp/mod.rs
new file mode 100644
index 0000000000000..749f892f1ccd2
--- /dev/null
+++ b/src/sinks/amqp/mod.rs
@@ -0,0 +1,21 @@
+//! `AMQP` sink.
+//! Handles version AMQP 0.9.1 which is used by RabbitMQ.
+mod config;
+mod encoder;
+mod request_builder;
+mod service;
+mod sink;
+
+#[cfg(all(test, feature = "amqp-integration-tests"))]
+mod integration_tests;
+
+pub use config::AmqpSinkConfig;
+use snafu::Snafu;
+
+#[derive(Debug, Snafu)]
+enum BuildError {
+    #[snafu(display("creating amqp producer failed: {}", source))]
+    AmqpCreateFailed {
+        source: Box<dyn std::error::Error + Send + Sync>,
+    },
+}
diff --git a/src/sinks/amqp/request_builder.rs b/src/sinks/amqp/request_builder.rs
new file mode 100644
index 0000000000000..d0c8de075471f
--- /dev/null
+++ b/src/sinks/amqp/request_builder.rs
@@ -0,0 +1,66 @@
+//! Request builder for the `AMQP` sink.
+//! Responsible for taking the event (which includes rendered template values) and turning
+//! it into the raw bytes and other data needed to send the request to `AMQP`.
+use crate::{
+    event::Event,
+    sinks::util::{request_builder::EncodeResult, Compression, RequestBuilder},
+};
+use bytes::Bytes;
+use std::io;
+use vector_common::finalization::{EventFinalizers, Finalizable};
+
+use super::{encoder::AmqpEncoder, service::AmqpRequest, sink::AmqpEvent};
+
+pub(super) struct AmqpMetadata {
+    exchange: String,
+    routing_key: String,
+    finalizers: EventFinalizers,
+}
+
+/// Build the request to send to `AMQP` by using the encoder to convert it into
+/// raw bytes and pass along the resolved template fields to determine where to
+/// route the event.
+pub(super) struct AmqpRequestBuilder {
+    pub(super) encoder: AmqpEncoder,
+}
+
+impl RequestBuilder<AmqpEvent> for AmqpRequestBuilder {
+    type Metadata = AmqpMetadata;
+    type Events = Event;
+    type Encoder = AmqpEncoder;
+    type Payload = Bytes;
+    type Request = AmqpRequest;
+    type Error = io::Error;
+
+    fn compression(&self) -> Compression {
+        Compression::None
+    }
+
+    fn encoder(&self) -> &Self::Encoder {
+        &self.encoder
+    }
+
+    fn split_input(&self, mut input: AmqpEvent) -> (Self::Metadata, Self::Events) {
+        let metadata = AmqpMetadata {
+            exchange: input.exchange,
+            routing_key: input.routing_key,
+            finalizers: input.event.take_finalizers(),
+        };
+
+        (metadata, input.event)
+    }
+
+    fn build_request(
+        &self,
+        metadata: Self::Metadata,
+        payload: EncodeResult<Self::Payload>,
+    ) -> Self::Request {
+        let body = payload.into_payload();
+        AmqpRequest::new(
+            body,
+            metadata.exchange,
+            metadata.routing_key,
+            metadata.finalizers,
+        )
+    }
+}
diff --git a/src/sinks/amqp/service.rs b/src/sinks/amqp/service.rs
new file mode 100644
index 0000000000000..2a38f607930e4
--- /dev/null
+++ b/src/sinks/amqp/service.rs
@@ -0,0 +1,139 @@
+//! The main tower service that takes the request created by the request builder
+//! and sends it to `AMQP`.
+use crate::internal_events::sink::{AmqpAcknowledgementError, AmqpDeliveryError};
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use lapin::{options::BasicPublishOptions, BasicProperties};
+use snafu::Snafu;
+use std::{
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tower::Service;
+use vector_common::{
+    finalization::{EventFinalizers, EventStatus, Finalizable},
+    internal_event::EventsSent,
+};
+use vector_core::stream::DriverResponse;
+
+/// The request contains the data to send to `AMQP` together
+/// with the information need to route the message.
+pub(super) struct AmqpRequest {
+    body: Bytes,
+    exchange: String,
+    routing_key: String,
+    finalizers: EventFinalizers,
+}
+
+impl AmqpRequest {
+    pub(super) fn new(
+        body: Bytes,
+        exchange: String,
+        routing_key: String,
+        finalizers: EventFinalizers,
+    ) -> Self {
+        Self {
+            body,
+            exchange,
+            routing_key,
+            finalizers,
+        }
+    }
+}
+
+impl Finalizable for AmqpRequest {
+    fn take_finalizers(&mut self) -> EventFinalizers {
+        std::mem::take(&mut self.finalizers)
+    }
+}
+
+/// A successful response from `AMQP`.
+pub(super) struct AmqpResponse {
+    byte_size: usize,
+}
+
+impl DriverResponse for AmqpResponse {
+    fn event_status(&self) -> EventStatus {
+        EventStatus::Delivered
+    }
+
+    fn events_sent(&self) -> EventsSent {
+        EventsSent {
+            count: 1,
+            byte_size: self.byte_size,
+            output: None,
+        }
+    }
+
+    fn bytes_sent(&self) -> Option<(usize, &str)> {
+        Some((self.byte_size, "amqp_0_9_1"))
+    }
+}
+
+/// The tower service that handles the actual sending of data to `AMQP`.
+pub(super) struct AmqpService {
+    pub(super) channel: Arc<lapin::Channel>,
+}
+
+#[derive(Debug, Snafu)]
+pub(super) enum AmqpError {
+    #[snafu(display("Failed retrieving Acknowledgement: {}", error))]
+    AmqpAcknowledgementFailed { error: lapin::Error },
+
+    #[snafu(display("Failed AMQP request: {}", error))]
+    AmqpDeliveryFailed { error: lapin::Error },
+}
+
+impl Service<AmqpRequest> for AmqpService {
+    type Response = AmqpResponse;
+
+    type Error = AmqpError;
+
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: AmqpRequest) -> Self::Future {
+        let channel = Arc::clone(&self.channel);
+
+        Box::pin(async move {
+            channel
+                .confirm_select(lapin::options::ConfirmSelectOptions::default())
+                .await
+                .unwrap();
+
+            let byte_size = req.body.len();
+            let fut = channel
+                .basic_publish(
+                    &req.exchange,
+                    &req.routing_key,
+                    BasicPublishOptions::default(),
+                    req.body.as_ref(),
+                    BasicProperties::default(),
+                )
+                .await;
+
+            match fut {
+                Ok(result) => match result.await {
+                    Ok(lapin::publisher_confirm::Confirmation::Nack(_)) => {
+                        warn!("Received Negative Acknowledgement from AMQP server.");
+                        Ok(AmqpResponse { byte_size })
+                    }
+                    Err(error) => {
+                        // TODO: In due course the caller could emit these on error.
+                        emit!(AmqpAcknowledgementError { error: &error });
+                        Err(AmqpError::AmqpAcknowledgementFailed { error })
+                    }
+                    Ok(_) => Ok(AmqpResponse { byte_size }),
+                },
+                Err(error) => {
+                    // TODO: In due course the caller could emit these on error.
+                    emit!(AmqpDeliveryError { error: &error });
+                    Err(AmqpError::AmqpDeliveryFailed { error })
+                }
+            }
+        })
+    }
+}
diff --git a/src/sinks/amqp/sink.rs b/src/sinks/amqp/sink.rs
new file mode 100644
index 0000000000000..ee923cd260143
--- /dev/null
+++ b/src/sinks/amqp/sink.rs
@@ -0,0 +1,137 @@
+//! The sink for the `AMQP` sink that wires together the main stream that takes the
+//! event and sends it to `AMQP`.
+use crate::{
+    codecs::Transformer, event::Event, internal_events::TemplateRenderingError,
+    sinks::util::builder::SinkBuilderExt, template::Template,
+};
+use async_trait::async_trait;
+use futures::StreamExt;
+use futures_util::stream::BoxStream;
+use lapin::options::ConfirmSelectOptions;
+use std::sync::Arc;
+use tower::ServiceBuilder;
+use vector_core::sink::StreamSink;
+
+use super::{
+    config::AmqpSinkConfig, encoder::AmqpEncoder, request_builder::AmqpRequestBuilder,
+    service::AmqpService, BuildError,
+};
+
+/// Stores the event together with the rendered exchange and routing_key values.
+/// This is passed into the `RequestBuilder` which then splits it out into the event
+/// and metadata containing the exchange and routing_key.
+/// This event needs to be created prior to building the request so we can filter out
+/// any events that error whilst redndering the templates.
+pub(super) struct AmqpEvent {
+    pub(super) event: Event,
+    pub(super) exchange: String,
+    pub(super) routing_key: String,
+}
+
+pub(super) struct AmqpSink {
+    pub(super) channel: Arc<lapin::Channel>,
+    exchange: Template,
+    routing_key: Option<Template>,
+    transformer: Transformer,
+    encoder: crate::codecs::Encoder<()>,
+}
+
+impl AmqpSink {
+    pub(super) async fn new(config: AmqpSinkConfig) -> crate::Result<Self> {
+        let (_, channel) = config
+            .connection
+            .connect()
+            .await
+            .map_err(|e| BuildError::AmqpCreateFailed { source: e })?;
+
+        channel
+            .confirm_select(ConfirmSelectOptions::default())
+            .await
+            .map_err(|e| BuildError::AmqpCreateFailed {
+                source: Box::new(e),
+            })?;
+
+        let transformer = config.encoding.transformer();
+        let serializer = config.encoding.build()?;
+        let encoder = crate::codecs::Encoder::<()>::new(serializer);
+
+        Ok(AmqpSink {
+            channel: Arc::new(channel),
+            exchange: config.exchange,
+            routing_key: config.routing_key,
+            transformer,
+            encoder,
+        })
+    }
+
+    /// Transforms an event into an `AMQP` event by rendering the required template fields.
+    /// Returns None if there is an error whilst rendering.
+    fn make_amqp_event(&self, event: Event) -> Option<AmqpEvent> {
+        let exchange = self
+            .exchange
+            .render_string(&event)
+            .map_err(|missing_keys| {
+                emit!(TemplateRenderingError {
+                    error: missing_keys,
+                    field: Some("exchange"),
+                    drop_event: true,
+                })
+            })
+            .ok()?;
+
+        let routing_key = match &self.routing_key {
+            None => String::new(),
+            Some(key) => key
+                .render_string(&event)
+                .map_err(|missing_keys| {
+                    emit!(TemplateRenderingError {
+                        error: missing_keys,
+                        field: Some("routing_key"),
+                        drop_event: true,
+                    })
+                })
+                .ok()?,
+        };
+
+        Some(AmqpEvent {
+            event,
+            exchange,
+            routing_key,
+        })
+    }
+
+    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
+        let request_builder = AmqpRequestBuilder {
+            encoder: AmqpEncoder {
+                encoder: self.encoder.clone(),
+                transformer: self.transformer.clone(),
+            },
+        };
+        let service = ServiceBuilder::new().service(AmqpService {
+            channel: Arc::clone(&self.channel),
+        });
+
+        let sink = input
+            .filter_map(|event| std::future::ready(self.make_amqp_event(event)))
+            .request_builder(None, request_builder)
+            .filter_map(|request| async move {
+                match request {
+                    Err(e) => {
+                        error!("Failed to build AMQP request: {:?}.", e);
+                        None
+                    }
+                    Ok(req) => Some(req),
+                }
+            })
+            .into_driver(service);
+
+        sink.run().await
+    }
+}
+
+#[async_trait]
+impl StreamSink<Event> for AmqpSink {
+    async fn run(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
+        self.run_inner(input).await
+    }
+}
diff --git a/src/sinks/mod.rs b/src/sinks/mod.rs
index d21286cd377b3..e1b8bb3938a17 100644
--- a/src/sinks/mod.rs
+++ b/src/sinks/mod.rs
@@ -4,6 +4,8 @@ use snafu::Snafu;
 
 pub mod util;
 
+#[cfg(feature = "sinks-amqp")]
+pub mod amqp;
 #[cfg(feature = "sinks-apex")]
 pub mod apex;
 #[cfg(feature = "sinks-aws_cloudwatch_logs")]
@@ -130,6 +132,10 @@ pub enum HealthcheckError {
 #[serde(tag = "type", rename_all = "snake_case")]
 #[enum_dispatch(SinkConfig)]
 pub enum Sinks {
+    /// AMQP.
+    #[cfg(feature = "sinks-amqp")]
+    Amqp(#[configurable(derived)] amqp::AmqpSinkConfig),
+
     /// Apex Logs.
     #[cfg(feature = "sinks-apex")]
     Apex(#[configurable(derived)] apex::ApexSinkConfig),
@@ -364,6 +370,8 @@ impl NamedComponent for Sinks {
 
     fn get_component_name(&self) -> &'static str {
         match self {
+            #[cfg(feature = "sinks-amqp")]
+            Self::Amqp(config) => config.get_component_name(),
             #[cfg(feature = "sinks-apex")]
             Self::Apex(config) => config.get_component_name(),
             #[cfg(feature = "sinks-aws_cloudwatch_logs")]
diff --git a/src/sources/amqp.rs b/src/sources/amqp.rs
new file mode 100644
index 0000000000000..838e091442c2b
--- /dev/null
+++ b/src/sources/amqp.rs
@@ -0,0 +1,576 @@
+//! `AMQP` source.
+//! Handles version AMQP 0.9.1 which is used by RabbitMQ.
+use crate::{
+    amqp::AmqpConfig,
+    codecs::{Decoder, DecodingConfig},
+    config::{log_schema, Output, SourceConfig, SourceContext},
+    event::{BatchNotifier, BatchStatus},
+    internal_events::{
+        source::{AmqpAckError, AmqpBytesReceived, AmqpEventError, AmqpRejectError},
+        StreamClosedError,
+    },
+    serde::{bool_or_struct, default_decoding, default_framing_message_based},
+    shutdown::ShutdownSignal,
+    SourceSender,
+};
+use async_stream::stream;
+use bytes::Bytes;
+use chrono::{TimeZone, Utc};
+use codecs::decoding::{DeserializerConfig, FramingConfig};
+use futures::{FutureExt, StreamExt};
+use futures_util::Stream;
+use lapin::{acker::Acker, message::Delivery, Channel};
+use snafu::Snafu;
+use std::{io::Cursor, pin::Pin};
+use tokio_util::codec::FramedRead;
+use vector_common::{finalizer::UnorderedFinalizer, internal_event::EventsReceived};
+use vector_config::configurable_component;
+use vector_core::{
+    config::{AcknowledgementsConfig, LogNamespace},
+    event::Event,
+};
+
+#[derive(Debug, Snafu)]
+enum BuildError {
+    #[snafu(display("Could not create AMQP consumer: {}", source))]
+    AmqpCreateError {
+        source: Box<dyn std::error::Error + Send + Sync>,
+    },
+    #[snafu(display("Could not subscribe to AMQP queue: {}", source))]
+    AmqpSubscribeError { source: lapin::Error },
+}
+
+/// Configuration for the `amqp` source.
+///
+/// Supports AMQP version 0.9.1
+#[configurable_component(source("amqp"))]
+#[derive(Clone, Debug, Derivative)]
+#[derivative(Default)]
+#[serde(deny_unknown_fields)]
+pub struct AmqpSourceConfig {
+    /// The name of the queue to consume.
+    #[serde(default = "default_queue")]
+    pub(crate) queue: String,
+
+    /// The identifier for the consumer.
+    #[serde(default = "default_consumer")]
+    pub(crate) consumer: String,
+
+    /// Connection options for `AMQP` source.
+    pub(crate) connection: AmqpConfig,
+
+    /// The `AMQP` routing key.
+    #[serde(default = "default_routing_key_field")]
+    pub(crate) routing_key_field: String,
+
+    /// The `AMQP` exchange key.
+    #[serde(default = "default_exchange_key")]
+    pub(crate) exchange_key: String,
+
+    /// The `AMQP` offset key.
+    #[serde(default = "default_offset_key")]
+    pub(crate) offset_key: String,
+
+    /// The namespace to use. This overrides the global setting.
+    #[serde(default)]
+    pub log_namespace: Option<bool>,
+
+    #[configurable(derived)]
+    #[serde(default = "default_framing_message_based")]
+    #[derivative(Default(value = "default_framing_message_based()"))]
+    pub(crate) framing: FramingConfig,
+
+    #[configurable(derived)]
+    #[serde(default = "default_decoding")]
+    #[derivative(Default(value = "default_decoding()"))]
+    pub(crate) decoding: DeserializerConfig,
+
+    #[configurable(derived)]
+    #[serde(default, deserialize_with = "bool_or_struct")]
+    pub(crate) acknowledgements: AcknowledgementsConfig,
+}
+
+fn default_queue() -> String {
+    "vector".into()
+}
+
+fn default_consumer() -> String {
+    "vector".into()
+}
+
+fn default_routing_key_field() -> String {
+    "routing".into()
+}
+
+fn default_exchange_key() -> String {
+    "exchange".into()
+}
+
+fn default_offset_key() -> String {
+    "offset".into()
+}
+
+impl_generate_config_from_default!(AmqpSourceConfig);
+
+impl AmqpSourceConfig {
+    fn decoder(&self, log_namespace: LogNamespace) -> Decoder {
+        DecodingConfig::new(self.framing.clone(), self.decoding.clone(), log_namespace).build()
+    }
+}
+
+#[async_trait::async_trait]
+impl SourceConfig for AmqpSourceConfig {
+    async fn build(&self, cx: SourceContext) -> crate::Result<super::Source> {
+        let log_namespace = cx.log_namespace(self.log_namespace);
+        let acknowledgements = cx.do_acknowledgements(&self.acknowledgements);
+
+        amqp_source(self, cx.shutdown, cx.out, log_namespace, acknowledgements).await
+    }
+
+    fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<Output> {
+        let log_namespace = global_log_namespace.merge(self.log_namespace);
+        let schema_definition = self
+            .decoding
+            .schema_definition(log_namespace)
+            .with_standard_vector_source_metadata();
+
+        vec![Output::default(self.decoding.output_type()).with_schema_definition(schema_definition)]
+    }
+
+    fn can_acknowledge(&self) -> bool {
+        true
+    }
+}
+
+#[derive(Debug)]
+struct FinalizerEntry {
+    acker: Acker,
+}
+
+impl From<Delivery> for FinalizerEntry {
+    fn from(delivery: Delivery) -> Self {
+        Self {
+            acker: delivery.acker,
+        }
+    }
+}
+
+pub(crate) async fn amqp_source(
+    config: &AmqpSourceConfig,
+    shutdown: ShutdownSignal,
+    out: SourceSender,
+    log_namespace: LogNamespace,
+    acknowledgements: bool,
+) -> crate::Result<super::Source> {
+    let config = config.clone();
+    let (_conn, channel) = config
+        .connection
+        .connect()
+        .await
+        .map_err(|source| BuildError::AmqpCreateError { source })?;
+
+    Ok(Box::pin(run_amqp_source(
+        config,
+        shutdown,
+        out,
+        channel,
+        log_namespace,
+        acknowledgements,
+    )))
+}
+
+struct Keys<'a> {
+    routing_key_field: &'a str,
+    routing: &'a str,
+    exchange_key: &'a str,
+    exchange: &'a str,
+    offset_key: &'a str,
+    delivery_tag: i64,
+}
+
+/// Populates the decoded event with extra metadata.
+fn populate_event(
+    event: &mut Event,
+    timestamp: chrono::DateTime<Utc>,
+    keys: &Keys<'_>,
+    log_namespace: LogNamespace,
+) {
+    let log = event.as_mut_log();
+
+    log_namespace.insert_vector_metadata(
+        log,
+        log_schema().timestamp_key(),
+        "ingest_timestamp",
+        timestamp,
+    );
+
+    log_namespace.insert_vector_metadata(
+        log,
+        log_schema().source_type_key(),
+        "source_type",
+        "amqp",
+    );
+
+    log_namespace.insert_source_metadata(
+        "amqp",
+        log,
+        keys.routing_key_field,
+        "routing",
+        keys.routing.to_string(),
+    );
+
+    log_namespace.insert_source_metadata(
+        "amqp",
+        log,
+        keys.exchange_key,
+        "exchange",
+        keys.exchange.to_string(),
+    );
+
+    log_namespace.insert_source_metadata("amqp", log, keys.offset_key, "offset", keys.delivery_tag);
+}
+
+/// Receives an event from `AMQP` and pushes it along the pipeline.
+async fn receive_event(
+    config: &AmqpSourceConfig,
+    out: &mut SourceSender,
+    log_namespace: LogNamespace,
+    finalizer: Option<&UnorderedFinalizer<FinalizerEntry>>,
+    msg: Delivery,
+) -> Result<(), ()> {
+    let payload = Cursor::new(Bytes::copy_from_slice(&msg.data));
+    let mut stream = FramedRead::new(payload, config.decoder(log_namespace));
+
+    // Extract timestamp from amqp message
+    let timestamp = msg
+        .properties
+        .timestamp()
+        .and_then(|millis| Utc.timestamp_millis_opt(millis as _).latest())
+        .unwrap_or_else(Utc::now);
+
+    let routing = msg.routing_key.to_string();
+    let exchange = msg.exchange.to_string();
+    let keys = Keys {
+        routing_key_field: config.routing_key_field.as_str(),
+        exchange_key: config.exchange_key.as_str(),
+        offset_key: config.offset_key.as_str(),
+        routing: &routing,
+        exchange: &exchange,
+        delivery_tag: msg.delivery_tag as i64,
+    };
+
+    let stream = stream! {
+        while let Some(result) = stream.next().await {
+            match result {
+                Ok((events, byte_size)) => {
+                    emit!(AmqpBytesReceived {
+                        byte_size,
+                        protocol: "amqp_0_9_1",
+                    });
+
+                    emit!(EventsReceived {
+                        byte_size: byte_size,
+                        count: events.len(),
+                    });
+
+                    for mut event in events {
+                        populate_event(&mut event,
+                                       timestamp,
+                                       &keys,
+                                       log_namespace);
+
+                        yield event;
+                    }
+                }
+                Err(error) => {
+                    use codecs::StreamDecodingError as _;
+
+                    // Error is logged by `codecs::Decoder`, no further handling
+                    // is needed here.
+                    if !error.can_continue() {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+    .boxed();
+
+    finalize_event_stream(finalizer, out, stream, msg).await;
+
+    Ok(())
+}
+
+/// Send the event stream created by the framed read to the `out` stream.
+async fn finalize_event_stream(
+    finalizer: Option<&UnorderedFinalizer<FinalizerEntry>>,
+    out: &mut SourceSender,
+    mut stream: Pin<Box<dyn Stream<Item = Event> + Send + '_>>,
+    msg: Delivery,
+) {
+    match finalizer {
+        Some(finalizer) => {
+            let (batch, receiver) = BatchNotifier::new_with_receiver();
+            let mut stream = stream.map(|event| event.with_batch_notifier(&batch));
+
+            match out.send_event_stream(&mut stream).await {
+                Err(error) => {
+                    emit!(StreamClosedError { error, count: 1 });
+                }
+                Ok(_) => {
+                    finalizer.add(msg.into(), receiver);
+                }
+            }
+        }
+        None => match out.send_event_stream(&mut stream).await {
+            Err(error) => {
+                emit!(StreamClosedError { error, count: 1 });
+            }
+            Ok(_) => {
+                let ack_options = lapin::options::BasicAckOptions::default();
+                if let Err(error) = msg.acker.ack(ack_options).await {
+                    emit!(AmqpAckError { error });
+                }
+            }
+        },
+    }
+}
+
+/// Runs the `AMQP` source involving the main loop pulling data from the server.
+async fn run_amqp_source(
+    config: AmqpSourceConfig,
+    shutdown: ShutdownSignal,
+    mut out: SourceSender,
+    channel: Channel,
+    log_namespace: LogNamespace,
+    acknowledgements: bool,
+) -> Result<(), ()> {
+    let (finalizer, mut ack_stream) =
+        UnorderedFinalizer::<FinalizerEntry>::maybe_new(acknowledgements, shutdown.clone());
+
+    debug!("Starting amqp source, listening to queue {}.", config.queue);
+    let mut consumer = channel
+        .basic_consume(
+            &config.queue,
+            &config.consumer,
+            lapin::options::BasicConsumeOptions::default(),
+            lapin::types::FieldTable::default(),
+        )
+        .await
+        .map_err(|error| {
+            error!(message = "Failed to consume.", error = ?error, internal_log_rate_secs = 10);
+        })?
+        .fuse();
+    let mut shutdown = shutdown.fuse();
+    loop {
+        tokio::select! {
+            _ = &mut shutdown => break,
+            entry = ack_stream.next() => {
+                if let Some((status, entry)) = entry {
+                    handle_ack(status, entry).await;
+                }
+            },
+            opt_m = consumer.next() => {
+                if let Some(try_m) = opt_m {
+                    match try_m {
+                        Err(error) => {
+                            emit!(AmqpEventError { error });
+                            return Err(());
+                        }
+                        Ok(msg) => {
+                            receive_event(&config, &mut out, log_namespace, finalizer.as_ref(), msg).await?
+                        }
+                    }
+                } else {
+                    break
+                }
+            }
+        };
+    }
+
+    Ok(())
+}
+
+async fn handle_ack(status: BatchStatus, entry: FinalizerEntry) {
+    match status {
+        BatchStatus::Delivered => {
+            let ack_options = lapin::options::BasicAckOptions::default();
+            if let Err(error) = entry.acker.ack(ack_options).await {
+                emit!(AmqpAckError { error });
+            }
+        }
+        BatchStatus::Errored => {
+            let ack_options = lapin::options::BasicRejectOptions::default();
+            if let Err(error) = entry.acker.reject(ack_options).await {
+                emit!(AmqpRejectError { error });
+            }
+        }
+        BatchStatus::Rejected => {
+            let ack_options = lapin::options::BasicRejectOptions::default();
+            if let Err(error) = entry.acker.reject(ack_options).await {
+                emit!(AmqpRejectError { error });
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+pub mod test {
+    use super::*;
+
+    #[test]
+    fn generate_config() {
+        crate::test_util::test_generate_config::<AmqpSourceConfig>();
+    }
+
+    pub fn make_config() -> AmqpSourceConfig {
+        let mut config = AmqpSourceConfig {
+            queue: "it".to_string(),
+            ..Default::default()
+        };
+        let user = std::env::var("AMQP_USER").unwrap_or_else(|_| "guest".to_string());
+        let pass = std::env::var("AMQP_PASSWORD").unwrap_or_else(|_| "guest".to_string());
+        let vhost = std::env::var("AMQP_VHOST").unwrap_or_else(|_| "%2f".to_string());
+        config.connection.connection_string =
+            format!("amqp://{}:{}@rabbitmq:5672/{}", user, pass, vhost);
+        config
+    }
+}
+
+/// Integration tests use the docker compose files in `scripts/integration/docker-compose.amqp.yml`.
+#[cfg(feature = "amqp-integration-tests")]
+#[cfg(test)]
+mod integration_test {
+    use super::test::*;
+    use super::*;
+    use crate::{
+        shutdown::ShutdownSignal,
+        test_util::{
+            components::{run_and_assert_source_compliance, SOURCE_TAGS},
+            random_string,
+        },
+        SourceSender,
+    };
+    use chrono::Utc;
+    use lapin::options::*;
+    use lapin::BasicProperties;
+    use tokio::time::Duration;
+
+    #[tokio::test]
+    async fn amqp_source_create_ok() {
+        let config = make_config();
+        assert!(amqp_source(
+            &config,
+            ShutdownSignal::noop(),
+            SourceSender::new_test().0,
+            LogNamespace::Legacy,
+            false,
+        )
+        .await
+        .is_ok());
+    }
+
+    async fn send_event(
+        channel: &lapin::Channel,
+        exchange: &str,
+        routing_key: &str,
+        text: &str,
+        _timestamp: i64,
+    ) {
+        let payload = text.as_bytes();
+        let payload_len = payload.len();
+        trace!("Sending message of length {} to {}.", payload_len, exchange,);
+
+        channel
+            .basic_publish(
+                exchange,
+                routing_key,
+                BasicPublishOptions::default(),
+                payload.as_ref(),
+                BasicProperties::default(),
+            )
+            .await
+            .unwrap()
+            .await
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn amqp_source_consume_event() {
+        let exchange = format!("test-{}-exchange", random_string(10));
+        let queue = format!("test-{}-queue", random_string(10));
+        let routing_key = "my_key";
+        trace!("Test exchange name: {}.", exchange);
+        let consumer = format!("test-consumer-{}", random_string(10));
+
+        let mut config = make_config();
+        config.consumer = consumer;
+        config.queue = queue;
+        config.routing_key_field = "message_key".to_string();
+        config.exchange_key = "exchange".to_string();
+        let (_conn, channel) = config.connection.connect().await.unwrap();
+        let exchange_opts = lapin::options::ExchangeDeclareOptions {
+            auto_delete: true,
+            ..Default::default()
+        };
+
+        channel
+            .exchange_declare(
+                &exchange,
+                lapin::ExchangeKind::Fanout,
+                exchange_opts,
+                lapin::types::FieldTable::default(),
+            )
+            .await
+            .unwrap();
+
+        let queue_opts = QueueDeclareOptions {
+            auto_delete: true,
+            ..Default::default()
+        };
+        channel
+            .queue_declare(
+                &config.queue,
+                queue_opts,
+                lapin::types::FieldTable::default(),
+            )
+            .await
+            .unwrap();
+
+        channel
+            .queue_bind(
+                &config.queue,
+                &exchange,
+                "",
+                lapin::options::QueueBindOptions::default(),
+                lapin::types::FieldTable::default(),
+            )
+            .await
+            .unwrap();
+
+        trace!("Sending event...");
+        let now = Utc::now();
+        send_event(
+            &channel,
+            &exchange,
+            routing_key,
+            "my message",
+            now.timestamp_millis(),
+        )
+        .await;
+
+        trace!("Receiving event...");
+        let events =
+            run_and_assert_source_compliance(config, Duration::from_secs(1), &SOURCE_TAGS).await;
+        assert!(!events.is_empty());
+
+        let log = events[0].as_log();
+        trace!("{:?}", log);
+        assert_eq!(log[log_schema().message_key()], "my message".into());
+        assert_eq!(log["message_key"], routing_key.into());
+        assert_eq!(log[log_schema().source_type_key()], "amqp".into());
+        let log_ts = log[log_schema().timestamp_key()].as_timestamp().unwrap();
+        assert!(log_ts.signed_duration_since(now) < chrono::Duration::seconds(1));
+        assert_eq!(log["exchange"], exchange.into());
+    }
+}
diff --git a/src/sources/mod.rs b/src/sources/mod.rs
index 797cc814abd64..58ef0324fdb1e 100644
--- a/src/sources/mod.rs
+++ b/src/sources/mod.rs
@@ -1,6 +1,8 @@
 use enum_dispatch::enum_dispatch;
 use snafu::Snafu;
 
+#[cfg(feature = "sources-amqp")]
+pub mod amqp;
 #[cfg(feature = "sources-apache_metrics")]
 pub mod apache_metrics;
 #[cfg(feature = "sources-aws_ecs_metrics")]
@@ -104,6 +106,10 @@ enum BuildError {
 #[serde(tag = "type", rename_all = "snake_case")]
 #[enum_dispatch(SourceConfig)]
 pub enum Sources {
+    /// AMQP.
+    #[cfg(feature = "sources-amqp")]
+    Amqp(#[configurable(derived)] amqp::AmqpSourceConfig),
+
     /// Apache HTTP Server (HTTPD) Metrics.
     #[cfg(feature = "sources-apache_metrics")]
     ApacheMetrics(#[configurable(derived)] apache_metrics::ApacheMetricsConfig),
@@ -297,6 +303,8 @@ impl NamedComponent for Sources {
 
     fn get_component_name(&self) -> &'static str {
         match self {
+            #[cfg(feature = "sources-amqp")]
+            Self::Amqp(config) => config.get_component_name(),
             #[cfg(feature = "sources-apache_metrics")]
             Self::ApacheMetrics(config) => config.get_component_name(),
             #[cfg(feature = "sources-aws_ecs_metrics")]
diff --git a/website/cue/reference/components/amqp.cue b/website/cue/reference/components/amqp.cue
new file mode 100644
index 0000000000000..ad164ad881449
--- /dev/null
+++ b/website/cue/reference/components/amqp.cue
@@ -0,0 +1,68 @@
+package metadata
+
+components: _amqp: {
+	features: {
+		send: to: {
+			service: services.amqp
+			interface: {
+				socket: {
+					api: {
+						title: "AMQP protocol"
+						url:   urls.amqp_protocol
+					}
+					direction: "outgoing"
+					protocols: ["tcp"]
+					ssl: "optional"
+				}
+			}
+		}
+		tls: {
+			enabled:                true
+			can_enable:             true
+			can_verify_certificate: true
+			enabled_default:        false
+		}
+	}
+
+	support: {
+		targets: {
+			"aarch64-unknown-linux-gnu":      true
+			"aarch64-unknown-linux-musl":     true
+			"armv7-unknown-linux-gnueabihf":  true
+			"armv7-unknown-linux-musleabihf": true
+			"x86_64-apple-darwin":            true
+			"x86_64-pc-windows-msv":          true
+			"x86_64-unknown-linux-gnu":       true
+			"x86_64-unknown-linux-musl":      true
+		}
+		requirements: []
+		warnings: []
+		notices: []
+	}
+
+	configuration: {
+		connection_string: {
+			description: """
+				Connection string to use when connecting to an AMQP server in the format of amqp://user:password@host:port/vhost?timeout=seconds.
+				The default vhost can be represented as %2f.
+				"""
+			required: true
+			warnings: []
+			type: string: {
+				examples: ["amqp://user:password@127.0.0.1:5672/%2f?timeout=10"]
+				syntax: "literal"
+			}
+		}
+	}
+
+	how_it_works: {
+		lapin: {
+			title: "Lapin"
+			body:  """
+				The `amqp` source and sink uses [`lapin`](\(urls.lapin)) under the hood. This
+				is a reliable pure rust library that facilitates communication with AMPQ servers
+				such as RabbitMQ.
+				"""
+		}
+	}
+}
diff --git a/website/cue/reference/components/sinks/amqp.cue b/website/cue/reference/components/sinks/amqp.cue
new file mode 100644
index 0000000000000..883d302b3c5d3
--- /dev/null
+++ b/website/cue/reference/components/sinks/amqp.cue
@@ -0,0 +1,98 @@
+package metadata
+
+components: sinks: amqp: {
+	title: "AMQP"
+
+	classes: {
+		commonly_used: true
+		delivery:      "at_least_once"
+		development:   "beta"
+		egress_method: "dynamic"
+		service_providers: ["AMQP"]
+		stateful: false
+	}
+
+	features: {
+		acknowledgements: false
+		buffer: enabled:      true
+		healthcheck: enabled: true
+		send: {
+			batch: {
+				enabled:      false
+				common:       false
+				timeout_secs: null
+			}
+			compression: {
+				enabled: true
+				default: "none"
+				algorithms: ["none", "gzip", "lz4", "snappy", "zstd"]
+				levels: ["none", "fast", "default", "best", 0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
+			}
+			encoding: {
+				enabled: true
+				codec: {
+					enabled: true
+					enum: ["json", "text"]
+				}
+			}
+			request: enabled: false
+			tls: {
+				enabled:                true
+				can_verify_certificate: false
+				can_verify_hostname:    false
+				enabled_default:        false
+				enabled_by_scheme:      false
+			}
+			to: components._amqp.features.send.to
+		}
+	}
+
+	support: components._amqp.support
+
+	configuration: {
+		connection: {
+			description: "Connection options for the AMQP sink."
+			required:    true
+			warnings: []
+			type: object: {
+				examples: []
+				options: {
+					connection_string: components._amqp.configuration.connection_string
+				}
+			}
+		}
+		exchange: {
+			description: "The exchange to publish messages to."
+			required:    true
+			warnings: []
+			type: string: {
+				examples: ["message_exchange"]
+				syntax: "literal"
+			}
+		}
+		routing_key: {
+			common:      false
+			description: "Template use to generate a routing key which corresponds to a queue binding."
+			required:    false
+			warnings: []
+			type: string: {
+				examples: ["{{ field_a }}-{{ field_b }}"]
+				syntax:  "literal"
+				default: null
+			}
+		}
+	}
+
+	input: {
+		logs:    true
+		metrics: null
+		traces:  false
+	}
+
+	how_it_works: components._amqp.how_it_works
+
+	telemetry: metrics: {
+		events_discarded_total:  components.sources.internal_metrics.output.metrics.events_discarded_total
+		processing_errors_total: components.sources.internal_metrics.output.metrics.processing_errors_total
+	}
+}
diff --git a/website/cue/reference/components/sources/amqp.cue b/website/cue/reference/components/sources/amqp.cue
new file mode 100644
index 0000000000000..771391a5fcaa1
--- /dev/null
+++ b/website/cue/reference/components/sources/amqp.cue
@@ -0,0 +1,142 @@
+package metadata
+
+components: sources: amqp: {
+	title: "AMQP"
+
+	features: {
+		acknowledgements: false
+		collect: {
+			checkpoint: enabled: false
+			from: {
+				service: services.amqp
+				interface: {
+					socket: {
+						api: {
+							title: "AMQP protocol"
+							url:   urls.amqp_protocol
+						}
+						direction: "incoming"
+						port:      5672
+						protocols: ["tcp"]
+						ssl: "optional"
+					}
+				}
+			}
+		}
+		multiline: enabled: false
+	}
+
+	classes: {
+		commonly_used: true
+		deployment_roles: ["aggregator"]
+		delivery:      "at_least_once"
+		development:   "beta"
+		egress_method: "stream"
+		stateful:      false
+	}
+
+	support: components._amqp.support
+
+	installation: {
+		platform_name: null
+	}
+
+	configuration: {
+		connection: {
+			description: "Connection options for the AMQP source."
+			required:    true
+			warnings: []
+			type: object: {
+				examples: []
+				options: {
+					connection_string: components._amqp.configuration.connection_string
+				}
+			}
+		}
+		group_id: {
+			description: "The consumer group name to be used to consume events from AMQP."
+			required:    true
+			warnings: []
+			type: string: {
+				examples: ["consumer-group-name"]
+				syntax: "literal"
+			}
+		}
+		routing_key_field: {
+			common:      true
+			description: "The log field name to use for the AMQP routing key."
+			required:    false
+			warnings: []
+			type: string: {
+				default: "routing"
+				examples: ["routing"]
+				syntax: "literal"
+			}
+		}
+		exchange_key: {
+			common:      true
+			description: "The log field name to use for the AMQP exchange key."
+			required:    false
+			warnings: []
+			type: string: {
+				default: "exchange"
+				examples: ["exchange"]
+				syntax: "literal"
+			}
+		}
+		offset_key: {
+			common:      true
+			description: "The log field name to use for the AMQP offset key."
+			required:    false
+			warnings: []
+			type: string: {
+				default: "offset"
+				examples: ["offset"]
+				syntax: "literal"
+			}
+		}
+	}
+
+	output: logs: record: {
+		description: "An individual AMQP record."
+		fields: {
+			message: {
+				description: "The raw line from the AMQP record."
+				required:    true
+				type: string: {
+					examples: ["53.126.150.246 - - [01/Oct/2020:11:25:58 -0400] \"GET /disintermediate HTTP/2.0\" 401 20308"]
+					syntax: "literal"
+				}
+			}
+			offset: {
+				description: "The AMQP offset at the time the record was retrieved."
+				required:    true
+				type: uint: {
+					examples: [100]
+					unit: null
+				}
+			}
+			timestamp: fields._current_timestamp & {
+				description: "The timestamp encoded in the AMQP message or the current time if it cannot be fetched."
+			}
+			exchange: {
+				description: "The AMQP exchange that the record came from."
+				required:    true
+				type: string: {
+					examples: ["topic"]
+					syntax: "literal"
+				}
+			}
+		}
+	}
+
+	telemetry: metrics: {
+		events_in_total:                      components.sources.internal_metrics.output.metrics.events_in_total
+		consumer_offset_updates_failed_total: components.sources.internal_metrics.output.metrics.consumer_offset_updates_failed_total
+		events_failed_total:                  components.sources.internal_metrics.output.metrics.events_failed_total
+		processed_bytes_total:                components.sources.internal_metrics.output.metrics.processed_bytes_total
+		processed_events_total:               components.sources.internal_metrics.output.metrics.processed_events_total
+	}
+
+	how_it_works: components._amqp.how_it_works
+}
diff --git a/website/cue/reference/services/amqp.cue b/website/cue/reference/services/amqp.cue
new file mode 100644
index 0000000000000..a8a005b868e63
--- /dev/null
+++ b/website/cue/reference/services/amqp.cue
@@ -0,0 +1,10 @@
+package metadata
+
+services: amqp: {
+	name:     "AMQP"
+	thing:    "\(name) topics"
+	url:      urls.amqp_protocol
+	versions: "= 0.9"
+
+	description: "The Advanced Message Queuing Protocol (AMQP) is an open standard for passing business messages between applications or organizations.  It connects systems, feeds business processes with the information they need and reliably transmits onward the instructions that achieve their goals."
+}
diff --git a/website/cue/reference/urls.cue b/website/cue/reference/urls.cue
index 986f2bc8f829f..3fa9a7b0de7f8 100644
--- a/website/cue/reference/urls.cue
+++ b/website/cue/reference/urls.cue
@@ -6,6 +6,7 @@ urls: {
 	affine_type_system:                         "\(wikipedia)/wiki/Substructural_type_system#Affine_type_systems"
 	adaptive_request_concurrency_post:          "/blog/adaptive-request-concurrency/"
 	amazon_linux:                               "https://aws.amazon.com/amazon-linux-ami/"
+	amqp_protocol:                              "https://www.amqp.org/about/what"
 	ansi_escape_codes:                          "\(wikipedia)/wiki/ANSI_escape_code"
 	apache:                                     "https://httpd.apache.org"
 	apache_common:                              "\(apache)/docs/current/logs.html#common"
@@ -289,6 +290,7 @@ urls: {
 	kubernetes_rbac:                            "\(kubernetes)/docs/reference/access-authn-authz/rbac/"
 	kubernetes_request_verbs:                   "\(kubernetes)/docs/reference/access-authn-authz/authorization/#determine-the-request-verb"
 	kubernetes_watch_api:                       "\(kubernetes)/docs/reference/generated/kubernetes-api/v1.10/#watch-30"
+	lapin:                                      "https://crates.io/crates/lapin"
 	leveldb:                                    "\(github)/google/leveldb"
 	leveldb_sys_2:                              "https://crates.io/crates/leveldb-sys"
 	leveldb_sys_3:                              "\(github)/vectordotdev/leveldb-sys/tree/v3.0.0"