diff --git a/Cargo.lock b/Cargo.lock
index 94b2396..1dbe74e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -148,6 +148,12 @@ dependencies = [
  "generic-array",
 ]
 
+[[package]]
+name = "bumpalo"
+version = "3.20.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
+
 [[package]]
 name = "bytes"
 version = "1.11.1"
@@ -197,16 +203,21 @@ name = "code0-flow"
 version = "0.0.0"
 dependencies = [
  "async-nats",
- "async-trait",
  "dotenv",
  "futures-core",
  "log",
- "regex",
+ "opentelemetry",
+ "opentelemetry-appender-tracing",
+ "opentelemetry-otlp",
+ "opentelemetry_sdk",
  "serde",
  "serde_json",
  "tokio",
  "tonic",
  "tonic-health",
+ "tracing",
+ "tracing-opentelemetry",
+ "tracing-subscriber",
  "tucana",
  "walkdir",
 ]
@@ -431,6 +442,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10"
 dependencies = [
  "futures-core",
+ "futures-sink",
 ]
 
 [[package]]
@@ -439,6 +451,34 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
 
+[[package]]
+name = "futures-executor"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-io"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
+
+[[package]]
+name = "futures-macro"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.31"
@@ -458,8 +498,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
 dependencies = [
  "futures-core",
+ "futures-io",
+ "futures-macro",
  "futures-sink",
  "futures-task",
+ "memchr",
  "pin-project-lite",
  "pin-utils",
  "slab",
@@ -632,6 +675,7 @@ version = "0.1.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8d9b05277c7e8da2c93a568989bb6207bef0112e8d17df7a6eda4a3cf143bc5e"
 dependencies = [
+ "base64",
  "bytes",
  "futures-channel",
  "futures-core",
@@ -639,7 +683,9 @@ dependencies = [
  "http",
  "http-body",
  "hyper",
+ "ipnet",
  "libc",
+ "percent-encoding",
  "pin-project-lite",
  "socket2",
  "tokio",
@@ -771,6 +817,12 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "ipnet"
+version = "2.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2"
+
 [[package]]
 name = "itertools"
 version = "0.14.0"
@@ -786,6 +838,23 @@ version = "1.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c"
 
+[[package]]
+name = "js-sys"
+version = "0.3.103"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102"
+dependencies = [
+ "cfg-if",
+ "futures-util",
+ "wasm-bindgen",
+]
+
+[[package]]
+name = "lazy_static"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
+
 [[package]]
 name = "leb128fmt"
 version = "0.1.0"
@@ -816,6 +885,15 @@ version = "0.4.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"
 
+[[package]]
+name = "matchers"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"
+dependencies = [
+ "regex-automata",
+]
+
 [[package]]
 name = "matchit"
 version = "0.8.4"
@@ -866,6 +944,15 @@ dependencies = [
  "signatory",
 ]
 
+[[package]]
+name = "nu-ansi-term"
+version = "0.50.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
+dependencies = [
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "nuid"
 version = "0.5.0"
@@ -902,6 +989,95 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"
 
+[[package]]
+name = "opentelemetry"
+version = "0.32.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "js-sys",
+ "pin-project-lite",
+ "thiserror",
+ "tracing",
+]
+
+[[package]]
+name = "opentelemetry-appender-tracing"
+version = "0.32.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c0080f0dc1d7c786f467cd85a4e395fcab11ee852004f39a29a18ab7c25d837"
+dependencies = [
+ "opentelemetry",
+ "tracing",
+ "tracing-core",
+ "tracing-subscriber",
+]
+
+[[package]]
+name = "opentelemetry-http"
+version = "0.32.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "http",
+ "opentelemetry",
+ "reqwest",
+]
+
+[[package]]
+name = "opentelemetry-otlp"
+version = "0.32.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35"
+dependencies = [
+ "http",
+ "opentelemetry",
+ "opentelemetry-http",
+ "opentelemetry-proto",
+ "opentelemetry_sdk",
+ "prost",
+ "reqwest",
+ "thiserror",
+ "tokio",
+ "tonic",
+ "tonic-types",
+]
+
+[[package]]
+name = "opentelemetry-proto"
+version = "0.32.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638"
+dependencies = [
+ "opentelemetry",
+ "opentelemetry_sdk",
+ "prost",
+ "tonic",
+ "tonic-prost",
+]
+
+[[package]]
+name = "opentelemetry_sdk"
+version = "0.32.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9"
+dependencies = [
+ "futures-channel",
+ "futures-executor",
+ "futures-util",
+ "opentelemetry",
+ "percent-encoding",
+ "portable-atomic",
+ "rand 0.9.4",
+ "thiserror",
+ "tokio",
+ "tokio-stream",
+]
+
 [[package]]
 name = "pbjson"
 version = "0.9.0"
@@ -1157,10 +1333,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404"
 dependencies = [
  "libc",
- "rand_chacha",
+ "rand_chacha 0.3.1",
  "rand_core 0.6.4",
 ]
 
+[[package]]
+name = "rand"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea"
+dependencies = [
+ "rand_chacha 0.9.0",
+ "rand_core 0.9.5",
+]
+
 [[package]]
 name = "rand"
 version = "0.10.1"
@@ -1182,6 +1368,16 @@ dependencies = [
  "rand_core 0.6.4",
 ]
 
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.9.5",
+]
+
 [[package]]
 name = "rand_core"
 version = "0.6.4"
@@ -1191,6 +1387,15 @@ dependencies = [
  "getrandom 0.2.16",
 ]
 
+[[package]]
+name = "rand_core"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
+dependencies = [
+ "getrandom 0.3.3",
+]
+
 [[package]]
 name = "rand_core"
 version = "0.10.1"
@@ -1226,6 +1431,37 @@ version = "0.8.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4"
 
+[[package]]
+name = "reqwest"
+version = "0.13.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3"
+dependencies = [
+ "base64",
+ "bytes",
+ "futures-channel",
+ "futures-core",
+ "futures-util",
+ "http",
+ "http-body",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
+ "js-sys",
+ "log",
+ "percent-encoding",
+ "pin-project-lite",
+ "sync_wrapper",
+ "tokio",
+ "tower",
+ "tower-http",
+ "tower-service",
+ "url",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+]
+
 [[package]]
 name = "ring"
 version = "0.17.14"
@@ -1435,6 +1671,15 @@ dependencies = [
  "digest",
 ]
 
+[[package]]
+name = "sharded-slab"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6"
+dependencies = [
+ "lazy_static",
+]
+
 [[package]]
 name = "shlex"
 version = "1.3.0"
@@ -1523,6 +1768,9 @@ name = "sync_wrapper"
 version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263"
+dependencies = [
+ "futures-core",
+]
 
 [[package]]
 name = "synstructure"
@@ -1568,6 +1816,15 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "thread_local"
+version = "1.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185"
+dependencies = [
+ "cfg-if",
+]
+
 [[package]]
 name = "time"
 version = "0.3.41"
@@ -1772,6 +2029,17 @@ dependencies = [
  "tonic-build",
 ]
 
+[[package]]
+name = "tonic-types"
+version = "0.14.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "73ab1b02061f83d519bba3caa167f88f261ef05720ab8ebc954ade70de3348e8"
+dependencies = [
+ "prost",
+ "prost-types",
+ "tonic",
+]
+
 [[package]]
 name = "tower"
 version = "0.5.2"
@@ -1791,6 +2059,24 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "tower-http"
+version = "0.6.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840"
+dependencies = [
+ "bitflags",
+ "bytes",
+ "futures-util",
+ "http",
+ "http-body",
+ "pin-project-lite",
+ "tower",
+ "tower-layer",
+ "tower-service",
+ "url",
+]
+
 [[package]]
 name = "tower-layer"
 version = "0.3.3"
@@ -1805,10 +2091,11 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3"
 
 [[package]]
 name = "tracing"
-version = "0.1.41"
+version = "0.1.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0"
+checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
 dependencies = [
+ "log",
  "pin-project-lite",
  "tracing-attributes",
  "tracing-core",
@@ -1816,9 +2103,9 @@ dependencies = [
 
 [[package]]
 name = "tracing-attributes"
-version = "0.1.30"
+version = "0.1.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903"
+checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1827,11 +2114,57 @@ dependencies = [
 
 [[package]]
 name = "tracing-core"
-version = "0.1.34"
+version = "0.1.36"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a"
+dependencies = [
+ "once_cell",
+ "valuable",
+]
+
+[[package]]
+name = "tracing-log"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3"
+dependencies = [
+ "log",
+ "once_cell",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-opentelemetry"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26"
+dependencies = [
+ "js-sys",
+ "opentelemetry",
+ "smallvec",
+ "tracing",
+ "tracing-core",
+ "tracing-log",
+ "tracing-subscriber",
+ "web-time",
+]
+
+[[package]]
+name = "tracing-subscriber"
+version = "0.3.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678"
+checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319"
 dependencies = [
+ "matchers",
+ "nu-ansi-term",
  "once_cell",
+ "regex-automata",
+ "sharded-slab",
+ "smallvec",
+ "thread_local",
+ "tracing",
+ "tracing-core",
+ "tracing-log",
 ]
 
 [[package]]
@@ -1917,6 +2250,12 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
 
+[[package]]
+name = "valuable"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
+
 [[package]]
 name = "version_check"
 version = "0.9.5"
@@ -1975,6 +2314,61 @@ dependencies = [
  "wit-bindgen 0.51.0",
 ]
 
+[[package]]
+name = "wasm-bindgen"
+version = "0.2.126"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+ "rustversion",
+ "wasm-bindgen-macro",
+ "wasm-bindgen-shared",
+]
+
+[[package]]
+name = "wasm-bindgen-futures"
+version = "0.4.76"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d"
+dependencies = [
+ "js-sys",
+ "wasm-bindgen",
+]
+
+[[package]]
+name = "wasm-bindgen-macro"
+version = "0.2.126"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1"
+dependencies = [
+ "quote",
+ "wasm-bindgen-macro-support",
+]
+
+[[package]]
+name = "wasm-bindgen-macro-support"
+version = "0.2.126"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e"
+dependencies = [
+ "bumpalo",
+ "proc-macro2",
+ "quote",
+ "syn",
+ "wasm-bindgen-shared",
+]
+
+[[package]]
+name = "wasm-bindgen-shared"
+version = "0.2.126"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24"
+dependencies = [
+ "unicode-ident",
+]
+
 [[package]]
 name = "wasm-encoder"
 version = "0.244.0"
@@ -2009,6 +2403,26 @@ dependencies = [
  "semver",
 ]
 
+[[package]]
+name = "web-sys"
+version = "0.3.103"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141"
+dependencies = [
+ "js-sys",
+ "wasm-bindgen",
+]
+
+[[package]]
+name = "web-time"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb"
+dependencies = [
+ "js-sys",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "webpki-roots"
 version = "0.26.11"
diff --git a/Cargo.toml b/Cargo.toml
index 33b0be2..fcf4e63 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -8,27 +8,62 @@ homepage = "https://code0.tech"
 license = "Apache-2.0"
 
 [dependencies]
-tucana = { version = "0.0.75", features = ["aquila"] }
-async-trait = "0.1.85"
-log = "0.4.24"
-tonic = "0.14.1"
-dotenv = "0.15.0"
-tonic-health = "0.14.1"
-async-nats = "0.49.0"
-futures-core = "0.3.31"
-regex = "1.11.2"
-serde_json = "1.0.143"
-walkdir = "2.5.0"
-serde = "1.0.228"
-tokio = "1.49.0"
+tucana = { version = "0.0.75", features = ["aquila"], optional = true }
+log = { version = "0.4.24", optional = true }
+tonic = { version = "0.14.1", optional = true }
+dotenv = { version = "0.15.0", optional = true }
+tonic-health = { version = "0.14.1", optional = true }
+async-nats = { version = "0.49.0", optional = true }
+futures-core = { version = "0.3.31", optional = true }
+serde_json = { version = "1.0.143", optional = true }
+walkdir = { version = "2.5.0", optional = true }
+serde = { version = "1.0.228", features = ["derive"], optional = true }
+tokio = { version = "1.49.0", optional = true }
+opentelemetry = { version = "0.32.0", features = ["logs", "metrics", "trace"], optional = true }
+opentelemetry-appender-tracing = { version = "0.32.0", optional = true }
+opentelemetry-otlp = { version = "0.32.0", features = ["grpc-tonic", "logs", "metrics", "trace"], optional = true }
+opentelemetry_sdk = { version = "0.32.1", features = ["logs", "metrics", "rt-tokio", "trace"], optional = true }
+tracing = { version = "0.1.41", features = ["log"], optional = true }
+tracing-opentelemetry = { version = "0.33.0", optional = true }
+tracing-subscriber = { version = "0.3.20", features = ["env-filter", "fmt"], optional = true }
 
 [lib]
 doctest = true
 
 [features]
-default = ["all"]
-flow_definition = []
-flow_service = ["flow_definition"]
-flow_config = []
-flow_health = []
-all = ["flow_definition", "flow_config", "flow_health", "flow_service"]
+default = []
+flow_definition = [
+    "dep:log",
+    "dep:serde",
+    "dep:serde_json",
+    "dep:tucana",
+    "dep:walkdir",
+]
+flow_service = [
+    "flow_definition",
+    "dep:log",
+    "dep:tokio",
+    "dep:tonic",
+    "dep:tucana",
+]
+flow_config = [
+    "dep:dotenv",
+    "dep:log",
+]
+flow_health = [
+    "dep:async-nats",
+    "dep:futures-core",
+    "dep:tonic",
+    "dep:tonic-health",
+]
+flow_telemetry = [
+    "dep:opentelemetry",
+    "dep:opentelemetry-appender-tracing",
+    "dep:opentelemetry-otlp",
+    "dep:opentelemetry_sdk",
+    "dep:serde",
+    "dep:tracing",
+    "dep:tracing-opentelemetry",
+    "dep:tracing-subscriber",
+]
+all = ["flow_definition", "flow_config", "flow_health", "flow_service", "flow_telemetry"]
diff --git a/src/flow_telemetry/errors.rs b/src/flow_telemetry/errors.rs
new file mode 100644
index 0000000..fad85b6
--- /dev/null
+++ b/src/flow_telemetry/errors.rs
@@ -0,0 +1,145 @@
+use std::{
+    backtrace::Backtrace,
+    error::Error,
+    sync::atomic::{AtomicBool, Ordering},
+};
+
+static CAPTURE_BACKTRACES: AtomicBool = AtomicBool::new(false);
+
+pub const EXCEPTION_TARGET: &str = "telemetry::exception";
+pub const SUMMARY_TARGET: &str = "telemetry::error_summary";
+
+pub fn enable_backtraces() {
+    CAPTURE_BACKTRACES.store(true, Ordering::Relaxed);
+}
+
+pub fn record(
+    category: &'static str,
+    operation: &'static str,
+    error: &(dyn Error + 'static),
+    context: impl AsRef<str>,
+) {
+    let message = error.to_string();
+    let chain = error_chain(error);
+
+    tracing::error!(
+        target: SUMMARY_TARGET,
+        error_category = category,
+        operation,
+        error_message = message.as_str(),
+        error_context = context.as_ref(),
+        "operation failed"
+    );
+
+    record_exception(category, operation, &message, &chain, context.as_ref());
+}
+
+pub fn record_message(
+    category: &'static str,
+    operation: &'static str,
+    message: impl AsRef<str>,
+    context: impl AsRef<str>,
+) {
+    let message = message.as_ref();
+    tracing::error!(
+        target: SUMMARY_TARGET,
+        error_category = category,
+        operation,
+        error_message = message,
+        error_context = context.as_ref(),
+        "operation failed"
+    );
+
+    record_exception(category, operation, message, "", context.as_ref());
+}
+
+pub fn panic(message: &str, location: &str) {
+    tracing::error!(
+        target: SUMMARY_TARGET,
+        error_category = "panic",
+        operation = "process",
+        error_message = message,
+        error_context = location,
+        "process panicked"
+    );
+
+    record_exception("panic", "process", message, "", location);
+}
+
+fn error_chain(error: &(dyn Error + 'static)) -> String {
+    let mut messages = Vec::new();
+    let mut current = error.source();
+
+    while let Some(source) = current {
+        messages.push(source.to_string());
+        current = source.source();
+    }
+
+    messages.join(": ")
+}
+
+fn record_exception(
+    category: &'static str,
+    operation: &'static str,
+    message: &str,
+    chain: &str,
+    context: &str,
+) {
+    if !CAPTURE_BACKTRACES.load(Ordering::Relaxed) {
+        return;
+    }
+
+    tracing::error!(
+        target: EXCEPTION_TARGET,
+        error_category = category,
+        operation,
+        error_message = message,
+        error_chain = chain,
+        error_context = context,
+        exception_stacktrace = %Backtrace::force_capture(),
+        "exception"
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{error::Error, fmt};
+
+    use super::error_chain;
+
+    #[derive(Debug)]
+    struct TestError {
+        message: &'static str,
+        source: Option<Box<TestError>>,
+    }
+
+    impl fmt::Display for TestError {
+        fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+            formatter.write_str(self.message)
+        }
+    }
+
+    impl Error for TestError {
+        fn source(&self) -> Option<&(dyn Error + 'static)> {
+            self.source
+                .as_deref()
+                .map(|source| source as &(dyn Error + 'static))
+        }
+    }
+
+    #[test]
+    fn formats_source_chain_from_nearest_to_root() {
+        let error = TestError {
+            message: "outer",
+            source: Some(Box::new(TestError {
+                message: "middle",
+                source: Some(Box::new(TestError {
+                    message: "root",
+                    source: None,
+                })),
+            })),
+        };
+
+        assert_eq!(error_chain(&error), "middle: root");
+    }
+}
diff --git a/src/flow_telemetry/mod.rs b/src/flow_telemetry/mod.rs
new file mode 100644
index 0000000..5499a74
--- /dev/null
+++ b/src/flow_telemetry/mod.rs
@@ -0,0 +1,209 @@
+pub mod errors;
+use std::error::Error;
+
+use opentelemetry::{KeyValue, global, trace::TracerProvider as _};
+use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
+use opentelemetry_otlp::WithExportConfig;
+use opentelemetry_sdk::{
+    Resource,
+    logs::SdkLoggerProvider,
+    metrics::{PeriodicReader, SdkMeterProvider},
+    propagation::TraceContextPropagator,
+    trace::SdkTracerProvider,
+};
+use serde::{Deserialize, Serialize};
+use tracing_subscriber::{
+    EnvFilter, Layer, filter::filter_fn, layer::SubscriberExt, util::SubscriberInitExt,
+};
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(default)]
+pub struct OpenTelemetry {
+    pub enabled: bool,
+    pub service_name: String,
+    pub logs_endpoint: Option<String>,
+    pub metrics_endpoint: Option<String>,
+    pub traces_endpoint: Option<String>,
+}
+
+impl OpenTelemetry {
+    pub fn logs_endpoint(&self) -> Option<&str> {
+        non_empty_url(&self.logs_endpoint)
+    }
+
+    pub fn metrics_endpoint(&self) -> Option<&str> {
+        non_empty_url(&self.metrics_endpoint)
+    }
+
+    pub fn traces_endpoint(&self) -> Option<&str> {
+        non_empty_url(&self.traces_endpoint)
+    }
+
+    pub fn has_enabled_exporter(&self) -> bool {
+        self.logs_endpoint().is_some()
+            || self.metrics_endpoint().is_some()
+            || self.traces_endpoint().is_some()
+    }
+}
+
+impl Default for OpenTelemetry {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            service_name: env!("CARGO_PKG_NAME").into(),
+            logs_endpoint: None,
+            metrics_endpoint: None,
+            traces_endpoint: None,
+        }
+    }
+}
+
+pub struct TelemetrySettings<'a> {
+    pub environment: &'a str,
+    pub default_log_level: &'a str,
+    pub service_version: &'a str,
+    pub instrumentation_name: &'a str,
+    pub initialize_metrics: Option<fn()>,
+}
+
+pub struct Telemetry {
+    logger_provider: Option<SdkLoggerProvider>,
+    meter_provider: Option<SdkMeterProvider>,
+    tracer_provider: Option<SdkTracerProvider>,
+}
+
+impl Telemetry {
+    pub fn initialize(
+        config: &OpenTelemetry,
+        settings: TelemetrySettings<'_>,
+    ) -> Result<Self, Box<dyn Error + Send + Sync>> {
+        let filter = EnvFilter::try_from_default_env()
+            .or_else(|_| EnvFilter::try_new(settings.default_log_level))?;
+        let fmt_layer = tracing_subscriber::fmt::layer()
+            .compact()
+            .with_target(false)
+            .with_filter(filter_fn(|metadata| {
+                metadata.target() != errors::EXCEPTION_TARGET
+            }));
+
+        if !config.enabled || !config.has_enabled_exporter() {
+            tracing_subscriber::registry()
+                .with(filter)
+                .with(fmt_layer)
+                .init();
+            return Ok(Self {
+                logger_provider: None,
+                meter_provider: None,
+                tracer_provider: None,
+            });
+        }
+
+        let resource = Resource::builder()
+            .with_service_name(config.service_name.clone())
+            .with_attributes([
+                KeyValue::new("service.version", settings.service_version.to_owned()),
+                KeyValue::new(
+                    "deployment.environment.name",
+                    settings.environment.to_owned(),
+                ),
+            ])
+            .build();
+
+        let tracer_provider = if let Some(endpoint) = config.traces_endpoint() {
+            let exporter = opentelemetry_otlp::SpanExporter::builder()
+                .with_tonic()
+                .with_endpoint(endpoint.to_owned())
+                .build()?;
+            let provider = SdkTracerProvider::builder()
+                .with_resource(resource.clone())
+                .with_batch_exporter(exporter)
+                .build();
+            global::set_text_map_propagator(TraceContextPropagator::new());
+            Some(provider)
+        } else {
+            None
+        };
+
+        let logger_provider = if let Some(endpoint) = config.logs_endpoint() {
+            let exporter = opentelemetry_otlp::LogExporter::builder()
+                .with_tonic()
+                .with_endpoint(endpoint.to_owned())
+                .build()?;
+            Some(
+                SdkLoggerProvider::builder()
+                    .with_resource(resource.clone())
+                    .with_batch_exporter(exporter)
+                    .build(),
+            )
+        } else {
+            None
+        };
+
+        let meter_provider = if let Some(endpoint) = config.metrics_endpoint() {
+            let exporter = opentelemetry_otlp::MetricExporter::builder()
+                .with_tonic()
+                .with_endpoint(endpoint.to_owned())
+                .build()?;
+            let reader = PeriodicReader::builder(exporter).build();
+            let provider = SdkMeterProvider::builder()
+                .with_resource(resource)
+                .with_reader(reader)
+                .build();
+            global::set_meter_provider(provider.clone());
+            if let Some(initialize_metrics) = settings.initialize_metrics {
+                initialize_metrics();
+            }
+            Some(provider)
+        } else {
+            None
+        };
+
+        if config.logs_endpoint().is_some() || config.traces_endpoint().is_some() {
+            errors::enable_backtraces();
+        }
+
+        let trace_layer = tracer_provider.as_ref().map(|provider| {
+            tracing_opentelemetry::layer()
+                .with_tracer(provider.tracer(settings.instrumentation_name.to_owned()))
+                .with_error_records_to_exceptions(true)
+                .with_error_events_to_status(true)
+                .with_filter(filter_fn(|metadata| {
+                    metadata.target() != errors::SUMMARY_TARGET
+                }))
+        });
+        let log_layer = logger_provider.as_ref().map(|provider| {
+            OpenTelemetryTracingBridge::new(provider).with_filter(filter_fn(|metadata| {
+                metadata.target() != errors::SUMMARY_TARGET
+            }))
+        });
+
+        tracing_subscriber::registry()
+            .with(filter)
+            .with(fmt_layer)
+            .with(trace_layer)
+            .with(log_layer)
+            .init();
+
+        Ok(Self {
+            logger_provider,
+            meter_provider,
+            tracer_provider,
+        })
+    }
+
+    pub fn shutdown(self) {
+        if let Some(provider) = self.logger_provider {
+            let _ = provider.shutdown();
+        }
+        if let Some(provider) = self.meter_provider {
+            let _ = provider.shutdown();
+        }
+        if let Some(provider) = self.tracer_provider {
+            let _ = provider.shutdown();
+        }
+    }
+}
+
+fn non_empty_url(url: &Option<String>) -> Option<&str> {
+    url.as_deref().filter(|value| !value.trim().is_empty())
+}
diff --git a/src/lib.rs b/src/lib.rs
index dd792e0..94c1975 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -9,3 +9,6 @@ pub mod flow_health;
 
 #[cfg(feature = "flow_service")]
 pub mod flow_service;
+
+#[cfg(feature = "flow_telemetry")]
+pub mod flow_telemetry;