Stash attempt at tracing exporting

.github/workflows/build.yml

@@ -1,41 +1,102 @@
-name: Build and Publish Docker Container
+name: Build and Publish
 
 on:
   push:
-    branches:
+    branches: [ main ]
-      - main
+
+env:
+  CARGO_TERM_COLOR: always
+  RUST_BINARY_NAME: monzo-ingestion
 
 jobs:
   build:
+    name: Build and Test
     runs-on: ubuntu-latest
-    container:
+    container: catthehacker/ubuntu:act-latest
-      image: catthehacker/ubuntu:act-latest
-    env:
-      RUNNER_TOOL_CACHE: /toolcache
 
     steps:
-      - name: Checkout code
+      - uses: actions/checkout@v3
-        uses: actions/checkout@v2
 
-# Note to self: maybe look at this if we want to build outside docker?
+      - name: Install Rust
-#      - name: Build with cache
+        uses: actions-rs/toolchain@v1
-#        uses: Swatinem/rust-cache@v2
+        with:
+          toolchain: stable
+          profile: minimal
+          override: true
+          components: rustfmt, clippy
+
+      - name: Add ARM64 target
+        run: rustup target add aarch64-unknown-linux-musl
+
+      - name: Install ARM64 toolchain
+        run: |
+          apt-get update
+          apt-get install -y gcc-aarch64-linux-gnu build-essential musl-tools
+
+      - name: Cache dependencies
+        uses: actions/cache@v3
+        with:
+          path: |
+            ~/.cargo
+            target/
+          key: "${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}"
+          restore-keys: |
+            ${{ runner.os }}-cargo-            
+
+      - name: Build (x86_64)
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --release --all-features
+
+      - name: Build (ARM64)
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --release --all-features --target aarch64-unknown-linux-musl
+        env:
+          CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_LINKER: aarch64-linux-gnu-gcc
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v2
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v1
+        uses: docker/setup-buildx-action@v2
 
-      - name: Login to Docker
+      - name: Login to DockerHub
-        uses: docker/login-action@v1
+        uses: docker/login-action@v2
         with:
           registry: git.joshuacoles.me
           username: ${{ secrets.DOCKER_USERNAME }}
           password: ${{ secrets.DOCKER_PASSWORD }}
 
-      - name: Build and Push Docker image
+      - name: Upload artifacts
-        uses: docker/build-push-action@master
+        uses: actions/upload-artifact@v3
+        with:
+          name: binaries
+          path: |
+            target/release/${{ env.RUST_BINARY_NAME }}
+            target/aarch64-unknown-linux-musl/release/${{ env.RUST_BINARY_NAME }}            
+
+      # The target directory is kept in the .dockerignore file, so allow these to be copied in we need to move them to a
+      # new directory.
+      - run: mv target docker-binaries
+
+      - name: Build and push multi-arch Docker image
+        uses: docker/build-push-action@v4
         with:
           context: .
+          file: ./Dockerfile
+          platforms: linux/amd64,linux/arm64
           push: true
-          tags: git.joshuacoles.me/personal/monzo-ingestion:latest
+          tags: git.joshuacoles.me/${{ github.repository }}:${{ github.sha }},git.joshuacoles.me/${{ github.repository }}:latest
-          cache-from: type=registry,ref=user/app:latest
+          build-args: |
-          cache-to: type=inline
+            BINARY_NAME=${{ env.RUST_BINARY_NAME }}
+
+      - uses: robiningelbrecht/ntfy-action@v1.0.0
+        name: Notify via ntfy.sh
+        if: always()
+        with:
+          url: ${{ secrets.NTFY_URL }}
+          topic: ${{ secrets.NTFY_TOPIC }}
+          job_status: ${{ job.status }}

Cargo.toml

@@ -9,28 +9,37 @@ migration = { path = "migration" }
 
 axum = { version = "0.7.5", features = ["multipart"] }
 tokio = { version = "1.37.0", features = ["full"] }
-sea-orm = { version = "1.0.0-rc.4", features = [
+sea-orm = { version = "1.0.0", features = [
     "sqlx-postgres",
     "runtime-tokio-rustls",
     "macros"
 ] }
 
-serde = { version = "1.0.203", features = ["derive"] }
+serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0.117"
+serde_json = "1.0"
 tracing-subscriber = "0.3.18"
 tracing = "0.1.40"
-anyhow = { version = "1.0.86", features = ["backtrace"] }
+anyhow = { version = "1.0", features = ["backtrace"] }
-thiserror = "1.0.61"
+thiserror = "1.0"
-http = "1.1.0"
+http = "1.1"
-chrono = { version = "0.4.38", features = ["serde"] }
+chrono = { version = "0.4", features = ["serde"] }
-num-traits = "0.2.19"
+num-traits = "0.2"
 csv = "1.3.0"
-clap = "4.5.4"
+clap = "4.5"
-testcontainers = "0.17.0"
+testcontainers = "0.21"
-testcontainers-modules = { version = "0.5.0", features = ["postgres"] }
+testcontainers-modules = { version = "0.9", features = ["postgres"] }
-sqlx = { version = "0.7.4", features = ["postgres"] }
+sqlx = { version = "0.7", features = ["postgres"] }
-tower-http = { version = "0.5.2", features = ["trace"] }
+tower-http = { version = "0.5", features = ["trace"] }
-bytes = "1.6.0"
+bytes = "1.7"
+once_cell = "1.19"
+
+tracing-opentelemetry = "0.25.0"
+opentelemetry = "0.24.0"
+opentelemetry_sdk = { version = "0.24.1", features = ["rt-tokio"] }
+opentelemetry-http = { version = "0.13.0", features = ["reqwest"] }
+opentelemetry-otlp = { version = "0.17.0", features = ["grpc-tonic", "http-json", "tokio"] }
+opentelemetry-semantic-conventions = "0.16.0"
+reqwest = "0.12.7"
 
 [workspace]
 members = [".", "migration", "entity"]

Dockerfile

@@ -1,78 +1,16 @@
-# syntax=docker/dockerfile:1
+FROM --platform=$BUILDPLATFORM debian:bullseye-slim AS builder
-
+ARG TARGETPLATFORM
-# Comments are provided throughout this file to help you get started.
+ARG BINARY_NAME
-# If you need more help, visit the Dockerfile reference guide at
-# https://docs.docker.com/engine/reference/builder/
-
-################################################################################
-# Create a stage for building the application.
-
-ARG RUST_VERSION=1.76.0
-ARG APP_NAME=monzo-ingestion
-FROM rust:${RUST_VERSION}-slim-bullseye AS build
-ARG APP_NAME
 WORKDIR /app
+COPY . .
+RUN case "$TARGETPLATFORM" in \
+      "linux/amd64") BINARY_PATH="target/release/${BINARY_NAME}" ;; \
+      "linux/arm64") BINARY_PATH="target/aarch64-unknown-linux-gnu/release/${BINARY_NAME}" ;; \
+      *) exit 1 ;; \
+    esac && \
+    mv "$BINARY_PATH" /usr/local/bin/${BINARY_NAME}
 
-# Build the application.
+FROM --platform=$TARGETPLATFORM debian:bullseye-slim
-# Leverage a cache mount to /usr/local/cargo/registry/
+ARG BINARY_NAME
-# for downloaded dependencies and a cache mount to /app/target/ for 
+COPY --from=builder /usr/local/bin/${BINARY_NAME} /usr/local/bin/
-# compiled dependencies which will speed up subsequent builds.
+CMD ["${BINARY_NAME}"]
-# Leverage a bind mount to the src directory to avoid having to copy the
-# source code into the container. Once built, copy the executable to an
-# output directory before the cache mounted /app/target is unmounted.
-RUN --mount=type=bind,source=src,target=src \
-    --mount=type=bind,source=entity,target=entity \
-    --mount=type=bind,source=migration,target=migration \
-    --mount=type=bind,source=Cargo.toml,target=Cargo.toml \
-    --mount=type=bind,source=Cargo.lock,target=Cargo.lock \
-    --mount=type=cache,target=/app/target/ \
-    --mount=type=cache,target=/usr/local/cargo/registry/ \
-    <<EOF
-set -e
-cargo build --locked --release
-cp ./target/release/$APP_NAME /bin/server
-EOF
-
-################################################################################
-# Create a new stage for running the application that contains the minimal
-# runtime dependencies for the application. This often uses a different base
-# image from the build stage where the necessary files are copied from the build
-# stage.
-#
-# The example below uses the debian bullseye image as the foundation for running the app.
-# By specifying the "bullseye-slim" tag, it will also use whatever happens to be the
-# most recent version of that tag when you build your Dockerfile. If
-# reproducability is important, consider using a digest
-# (e.g., debian@sha256:ac707220fbd7b67fc19b112cee8170b41a9e97f703f588b2cdbbcdcecdd8af57).
-FROM debian:bullseye-slim AS final
-
-RUN set -ex; \
-    apt-get update && \
-    apt-get -y install --no-install-recommends \
-        ca-certificates curl && \
-    rm -rf /var/lib/apt/lists/*
-
-# Create a non-privileged user that the app will run under.
-# See https://docs.docker.com/develop/develop-images/dockerfile_best-practices/#user
-ARG UID=10001
-RUN adduser \
-    --disabled-password \
-    --gecos "" \
-    --home "/nonexistent" \
-    --shell "/sbin/nologin" \
-    --no-create-home \
-    --uid "${UID}" \
-    appuser
-USER appuser
-
-# Copy the executable from the "build" stage.
-COPY --from=build /bin/server /bin/
-
-# Expose the port that the application listens on.
-EXPOSE 3000
-
-HEALTHCHECK --interval=5s --timeout=3s --retries=3 \
-    CMD curl -f http://localhost:3000/health || exit 1
-
-# What the container should run when it is started.
-CMD ["/bin/server"]

src/error.rs

@@ -29,6 +29,7 @@ impl AppError {
 impl IntoResponse for AppError {
     fn into_response(self) -> Response {
         let status_code = match self {
+            AppError::BadRequest(_) => StatusCode::BAD_REQUEST,
             _ => StatusCode::INTERNAL_SERVER_ERROR,
         };
 

src/ingestion/db.rs

@@ -29,7 +29,7 @@ pub async fn insert(
 
     for insertions in insertions.chunks(400) {
         let (new_or_updated_insertions, inserted_transaction_ids) =
-            whittle_insertions(insertions, &db).await?;
+            whittle_insertions(insertions, db).await?;
 
         if new_or_updated_insertions.is_empty() {
             continue;
@@ -69,7 +69,7 @@ async fn update_expenditures(
 
     expenditure::Entity::insert_many(
         new_or_updated_insertions
-            .into_iter()
+            .iter()
             .flat_map(|i| &i.contained_expenditures)
             .cloned(),
     )
@@ -118,19 +118,21 @@ async fn whittle_insertions<'a>(
         .select_only()
         .columns([transaction::Column::IdentityHash])
         .filter(transaction::Column::IdentityHash.is_not_null())
-        .into_tuple::<(i64,)>()
+        .into_tuple::<(i64, )>()
         .all(tx)
         .await?;
 
+    tracing::debug!("Found existing entries: {existing_hashes:?}");
+
     // We will only update those where the hash is different to avoid unnecessary updates and
     // notifications.
     let new_or_updated_insertions = insertions
-        .into_iter()
+        .iter()
         .filter(|i| {
             let hash = i.identity_hash;
             !existing_hashes
                 .iter()
-                .any(|(existing_hash,)| *existing_hash == hash)
+                .any(|(existing_hash, )| *existing_hash == hash)
         })
         .collect::<Vec<_>>();
 
@@ -155,23 +157,28 @@ async fn notify_new_transactions(
 }
 
 mod tests {
-    use super::{notify_new_transactions, update_expenditures, update_transactions};
+    use super::{insert, notify_new_transactions, update_expenditures, update_transactions};
     use anyhow::Error;
+    use tokio::sync::OnceCell;
     use migration::MigratorTrait;
     use sea_orm::{DatabaseConnection, TransactionTrait};
+    use serde_json::Value;
     use sqlx::postgres::PgListener;
     use sqlx::PgPool;
     use testcontainers::runners::AsyncRunner;
     use testcontainers::ContainerAsync;
+    use crate::ingestion::ingestion_logic::from_json_row;
 
-    async fn initialise() -> Result<
+    #[derive(Debug)]
-        (
+    struct DatabaseInstance {
-            ContainerAsync<testcontainers_modules::postgres::Postgres>,
+        container: ContainerAsync<testcontainers_modules::postgres::Postgres>,
-            DatabaseConnection,
+        db: DatabaseConnection,
-            PgPool,
+        pool: PgPool,
-        ),
+    }
-        Error,
+
-    > {
+    static INSTANCE: OnceCell<DatabaseInstance> = OnceCell::const_new();
+
+    async fn initialise_db() -> Result<DatabaseInstance, Error> {
         let container = testcontainers_modules::postgres::Postgres::default()
             .start()
             .await?;
@@ -186,15 +193,29 @@ mod tests {
         migration::Migrator::up(&db, None).await?;
 
         let pool = PgPool::connect(connection_string).await?;
+        let instance = DatabaseInstance {
+            container,
+            db,
+            pool,
+        };
 
-        Ok((container, db, pool))
+        Ok(instance)
+    }
+
+    async fn get_or_initialize_db_instance() -> Result<
+        &'static DatabaseInstance,
+        Error,
+    > {
+        Ok(INSTANCE.get_or_init(|| async {
+            initialise_db().await.unwrap()
+        }).await)
     }
 
     #[tokio::test]
-    async fn test_no_new_insertions() -> Result<(), Error> {
+    async fn test_empty_insertion_list() -> Result<(), Error> {
-        let (_container, db, _pool) = initialise().await?;
+        let db = get_or_initialize_db_instance().await?;
         let insertions = vec![];
-        let tx = db.begin().await?;
+        let tx = db.db.begin().await?;
         update_transactions(&tx, &insertions).await?;
         update_expenditures(&tx, &insertions, &vec![]).await?;
         tx.commit().await?;
@@ -204,8 +225,8 @@ mod tests {
 
     #[tokio::test]
     async fn test_notify() -> Result<(), Error> {
-        let (_container, db, pool) = initialise().await?;
+        let dbi = get_or_initialize_db_instance().await?;
-        let mut listener = PgListener::connect_with(&pool).await?;
+        let mut listener = PgListener::connect_with(&dbi.pool).await?;
         listener.listen("monzo_new_transactions").await?;
 
         let ids = vec![
@@ -214,7 +235,7 @@ mod tests {
             "test3".to_string(),
         ];
 
-        notify_new_transactions(&db, &ids).await?;
+        notify_new_transactions(&dbi.db, &ids).await?;
 
         let notification = listener.recv().await?;
         let payload = notification.payload();
@@ -227,4 +248,52 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_notify_on_insert() -> Result<(), Error> {
+        let dbi = get_or_initialize_db_instance().await?;
+        let mut listener = PgListener::connect_with(&dbi.pool).await?;
+        listener.listen("monzo_new_transactions").await?;
+
+        let json = include_str!("../../fixtures/transactions.json");
+        let json: Vec<Vec<Value>> = serde_json::from_str(json)?;
+        let data = json
+            .iter()
+            .map(|row| from_json_row(row.clone()))
+            .collect::<Result<Vec<_>, anyhow::Error>>()
+            .unwrap();
+
+        insert(&dbi.db, data.clone()).await?;
+        let notification = listener.recv().await?;
+        let payload = notification.payload();
+        let mut payload = serde_json::from_str::<Vec<String>>(&payload)?;
+        payload.sort();
+
+        let mut ids = data.iter()
+            .map(|row| row.transaction_id.clone())
+            .collect::<Vec<_>>();
+
+        ids.sort();
+
+        assert_eq!(payload, ids, "Inserted IDs do not match");
+
+        insert(&dbi.db, data.clone()).await?;
+        let notification = listener.recv().await?;
+        let payload = notification.payload();
+        let payload = serde_json::from_str::<Vec<String>>(&payload)?;
+        assert_eq!(payload, Vec::<String>::new(), "Re-inserting identical rows triggered double notification");
+
+        let mut altered_data = data.clone();
+        altered_data[0].description = Some("New description".to_string());
+
+        assert_ne!(altered_data[0].compute_hash(), data[0].compute_hash(), "Alterations have the same hash");
+
+        insert(&dbi.db, altered_data.clone()).await?;
+        let notification = listener.recv().await?;
+        let payload = notification.payload();
+        let payload = serde_json::from_str::<Vec<String>>(&payload)?;
+        assert_eq!(payload, vec![altered_data[0].transaction_id.clone()], "Re-inserting altered row failed to re-trigger notification");
+
+        Ok(())
+    }
 }

src/ingestion/flex.rs

@@ -0,0 +1,9 @@
+#[allow(dead_code)]
+mod headings {
+    #[allow(unused_imports)]
+    pub use super::super::ingestion_logic::headings::*;
+
+    // Additional FLex headings
+    pub const MONEY_OUT: usize = 16;
+    pub const MONEY_IN: usize = 17;
+}

src/ingestion/ingestion_logic.rs

@@ -11,7 +11,7 @@ use serde_json::Value;
 use std::hash::Hash;
 
 #[allow(dead_code)]
-mod headings {
+pub(crate) mod headings {
     pub const TRANSACTION_ID: usize = 0;
     pub const DATE: usize = 1;
     pub const TIME: usize = 2;
@@ -30,19 +30,19 @@ mod headings {
     pub const CATEGORY_SPLIT: usize = 15;
 }
 
-#[derive(Debug, Eq, PartialEq, Hash)]
+#[derive(Debug, Eq, PartialEq, Hash, Clone)]
 pub struct MonzoRow {
-    category_split: Option<String>,
+    pub category_split: Option<String>,
-    primary_category: String,
+    pub primary_category: String,
-    total_amount: Decimal,
+    pub total_amount: Decimal,
-    receipt: Option<String>,
+    pub receipt: Option<String>,
-    notes: Option<String>,
+    pub notes: Option<String>,
-    emoji: Option<String>,
+    pub emoji: Option<String>,
-    description: Option<String>,
+    pub description: Option<String>,
-    transaction_type: String,
+    pub transaction_type: String,
-    title: Option<String>,
+    pub title: Option<String>,
-    timestamp: NaiveDateTime,
+    pub timestamp: NaiveDateTime,
-    transaction_id: String,
+    pub transaction_id: String,
 }
 
 impl MonzoRow {

src/ingestion/mod.rs

@@ -1,3 +1,4 @@
 pub mod db;
 pub mod ingestion_logic;
+pub mod flex;
 pub mod routes;

src/ingestion/routes.rs

@@ -43,10 +43,12 @@ pub async fn monzo_batched_csv(
     Extension(db): Extension<DatabaseConnection>,
     multipart: Multipart,
 ) -> Result<&'static str, AppError> {
+    static CSV_MISSING_ERR_MSG: &str = "No CSV file provided. Expected a multipart request with a `csv` field containing the contents of the CSV.";
+
     let csv = extract_csv(multipart)
         .await
         .map_err(|e| AppError::BadRequest(anyhow!(e)))
-        .and_then(|csv| csv.ok_or(AppError::BadRequest(anyhow!("No CSV file provided"))))?;
+        .and_then(|csv| csv.ok_or(AppError::BadRequest(anyhow!(CSV_MISSING_ERR_MSG))))?;
 
     let csv = Cursor::new(csv);
     let mut csv = csv::Reader::from_reader(csv);

src/main.rs

@@ -2,29 +2,70 @@ mod error;
 mod ingestion;
 
 use crate::error::AppError;
+use crate::ingestion::db;
+use crate::ingestion::ingestion_logic::from_csv_row;
 use crate::ingestion::routes::{monzo_batched_csv, monzo_batched_json};
 use axum::routing::{get, post};
 use axum::{Extension, Router};
-use clap::Parser;
+use clap::{Parser, Subcommand};
 use migration::{Migrator, MigratorTrait};
 use sea_orm::{ConnectionTrait, DatabaseConnection};
+use std::fs::File;
 use std::net::SocketAddr;
+use std::path::PathBuf;
 use tower_http::trace::TraceLayer;
 use tracing::log::LevelFilter;
 
-#[derive(Debug, clap::Parser)]
+use opentelemetry::trace::TracerProvider as _;
-struct Config {
+use opentelemetry_sdk::trace::TracerProvider;
-    /// If we should perform migration on startup.
+use opentelemetry_otlp;
+use tracing::{error, span};
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::Registry;
+use tracin_support::init_tracing_subscriber;
+
+mod tracin_support;
+
+#[derive(Debug, Subcommand)]
+enum Commands {
+    /// Manually run database migrations.
+    Migrate {
+        /// Number of migration steps to perform. If not provided, all migrations will be run.
+        #[arg(long)]
+        steps: Option<u32>,
+
+        /// If we should perform migration down.
+        #[arg(long)]
+        down: bool,
+    },
+
+    /// Start web app for to the google-sheets app script
+    Serve {
+        /// If we should perform migration at startup.
         #[clap(short, long, env, default_value_t = true)]
         migrate: bool,
 
         /// The server address to bind to.
         #[clap(short, long, env, default_value = "0.0.0.0:3000")]
         addr: SocketAddr,
+    },
 
+    /// Ingest a google-sheets CSV export into the database.
+    Csv {
+        /// The path of the CSV file to ingest.
+        csv_file: PathBuf,
+    },
+}
+
+/// Ingest and manage monzo transactions from the Monzo Plus auto-export feature
+#[derive(Debug, clap::Parser)]
+struct Cli {
     /// URL to PostgreSQL database.
     #[clap(short, long = "db", env)]
     database_url: String,
+
+    #[command(subcommand)]
+    command: Commands,
 }
 
 async fn health_check(
@@ -37,18 +78,48 @@ async fn health_check(
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let config: Config = Config::parse();
+    let _guard = init_tracing_subscriber();
-    let connection = sea_orm::ConnectOptions::new(&config.database_url)
+
+    let cli: Cli = Cli::parse();
+    let connection = sea_orm::ConnectOptions::new(&cli.database_url)
         .sqlx_logging_level(LevelFilter::Debug)
         .to_owned();
 
     let connection = sea_orm::Database::connect(connection).await?;
 
-    if config.migrate {
+    match cli.command {
+        Commands::Migrate { steps, down } => {
+            if down {
+                Migrator::down(&connection, steps).await?;
+            } else {
+                Migrator::up(&connection, steps).await?
+            }
+        }
+
+        Commands::Serve { migrate, addr } => {
+            if migrate {
                 Migrator::up(&connection, None).await?;
             }
 
-    tracing_subscriber::fmt::init();
+            serve_web(addr, connection).await?;
+        }
+
+        Commands::Csv { csv_file } => {
+            let mut csv = csv::Reader::from_reader(File::open(csv_file)?);
+            let data = csv.records();
+            let data = data
+                .filter_map(|f| f.ok())
+                .map(from_csv_row)
+                .collect::<Result<_, _>>()?;
+
+            db::insert(&connection, data).await?;
+        }
+    }
+
+    Ok(())
+}
+
+async fn serve_web(address: SocketAddr, connection: DatabaseConnection) -> anyhow::Result<()> {
     let app = Router::new()
         .route("/health", get(health_check))
         .route("/monzo-batch-export", post(monzo_batched_json))
@@ -56,9 +127,9 @@ async fn main() -> anyhow::Result<()> {
         .layer(Extension(connection.clone()))
         .layer(TraceLayer::new_for_http());
 
-    tracing::debug!("listening on {}", &config.addr);
+    tracing::info!("listening on {}", &address);
-    let listener = tokio::net::TcpListener::bind(&config.addr).await.unwrap();
+    let listener = tokio::net::TcpListener::bind(&address).await?;
-    axum::serve(listener, app).await.unwrap();
+    axum::serve(listener, app).await?;
 
     Ok(())
 }

src/tracin_support.rs

@@ -0,0 +1,108 @@
+use opentelemetry::{global, trace::TracerProvider, Key, KeyValue};
+use opentelemetry_sdk::{
+    metrics::{
+        reader::{DefaultAggregationSelector, DefaultTemporalitySelector},
+        Aggregation, Instrument, MeterProviderBuilder, PeriodicReader, SdkMeterProvider, Stream,
+    },
+    runtime,
+    trace::{BatchConfig, RandomIdGenerator, Sampler, Tracer},
+    Resource,
+};
+use opentelemetry_semantic_conventions::{
+    resource::{DEPLOYMENT_ENVIRONMENT, SERVICE_NAME, SERVICE_VERSION},
+    SCHEMA_URL,
+};
+use tracing::Level;
+use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
+
+// Create a Resource that captures information about the entity for which telemetry is recorded.
+fn resource() -> Resource {
+    Resource::from_schema_url(
+        [
+            KeyValue::new(SERVICE_NAME, env!("CARGO_PKG_NAME")),
+            KeyValue::new(SERVICE_VERSION, env!("CARGO_PKG_VERSION")),
+            KeyValue::new(DEPLOYMENT_ENVIRONMENT, "develop"),
+        ],
+        SCHEMA_URL,
+    )
+}
+
+// Construct MeterProvider for MetricsLayer
+fn init_meter_provider() -> SdkMeterProvider {
+    let exporter = opentelemetry_otlp::new_exporter()
+        .http()
+        .with_http_client(reqwest::Client::new())
+        .build_metrics_exporter(
+            Box::new(DefaultAggregationSelector::new()),
+            Box::new(DefaultTemporalitySelector::new()),
+        )
+        .unwrap();
+
+    let reader = PeriodicReader::builder(exporter, runtime::Tokio)
+        .with_interval(std::time::Duration::from_secs(30))
+        .build();
+
+    let meter_provider = MeterProviderBuilder::default()
+        .with_resource(resource())
+        .with_reader(reader)
+        .build();
+
+    global::set_meter_provider(meter_provider.clone());
+
+    meter_provider
+}
+
+// Construct Tracer for OpenTelemetryLayer
+fn init_tracer() -> Tracer {
+    let provider = opentelemetry_otlp::new_pipeline()
+        .tracing()
+        .with_trace_config(
+            opentelemetry_sdk::trace::Config::default()
+                // Customize sampling strategy
+                .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
+                    1.0,
+                ))))
+                // If export trace to AWS X-Ray, you can use XrayIdGenerator
+                .with_id_generator(RandomIdGenerator::default())
+                .with_resource(resource()),
+        )
+        .with_batch_config(BatchConfig::default())
+        .with_exporter(opentelemetry_otlp::new_exporter().tonic())
+        .install_batch(runtime::Tokio)
+        .unwrap();
+
+    global::set_tracer_provider(provider.clone());
+    provider.tracer("tracing-otel-subscriber")
+}
+
+// Initialize tracing-subscriber and return OtelGuard for opentelemetry-related termination processing
+pub(crate) fn init_tracing_subscriber() -> OtelGuard {
+    let meter_provider = init_meter_provider();
+    let tracer = init_tracer();
+
+    tracing_subscriber::registry()
+        .with(tracing_subscriber::filter::LevelFilter::from_level(
+            Level::DEBUG,
+        ))
+        .with(tracing_subscriber::fmt::layer())
+        .with(MetricsLayer::new(meter_provider.clone()))
+        .with(OpenTelemetryLayer::new(tracer))
+        .init();
+
+    OtelGuard { meter_provider }
+}
+
+pub(crate) struct OtelGuard {
+    meter_provider: SdkMeterProvider,
+}
+
+impl Drop for OtelGuard {
+    fn drop(&mut self) {
+        if let Err(err) = self.meter_provider.shutdown() {
+            eprintln!("{err:?}");
+        }
+
+        global::shutdown_tracer_provider();
+    }
+}