Local auth renew proxy

.github/ansible/prod.us-west-2.hosts.yaml

@@ -41,14 +41,6 @@ storage:
           ansible_host: i-051642d372c0a4f32
         pageserver-3.us-west-2.aws.neon.tech:
           ansible_host: i-00c3844beb9ad1c6b
-        pageserver-4.us-west-2.aws.neon.tech:
-          ansible_host: i-013263dd1c239adcc
-        pageserver-5.us-west-2.aws.neon.tech:
-          ansible_host: i-00ca6417c7bf96820
-        pageserver-6.us-west-2.aws.neon.tech:
-          ansible_host: i-01cdf7d2bc1433b6a
-        pageserver-7.us-west-2.aws.neon.tech:
-          ansible_host: i-02eec9b40617db5bc
 
     safekeepers:
       hosts:
@@ -58,15 +50,4 @@ storage:
           ansible_host: i-074682f9d3c712e7c
         safekeeper-2.us-west-2.aws.neon.tech:
           ansible_host: i-042b7efb1729d7966
-        safekeeper-3.us-west-2.aws.neon.tech:
-          ansible_host: i-089f6b9ef426dff76
-        safekeeper-4.us-west-2.aws.neon.tech:
-          ansible_host: i-0fe6bf912c4710c82
-        safekeeper-5.us-west-2.aws.neon.tech:
-          ansible_host: i-0a83c1c46d2b4e409
-        safekeeper-6.us-west-2.aws.neon.tech:
-          ansible_host: i-0fef5317b8fdc9f8d
-        safekeeper-7.us-west-2.aws.neon.tech:
-          ansible_host: i-0be739190d4289bf9
-        safekeeper-8.us-west-2.aws.neon.tech:
-          ansible_host: i-00e851803669e5cfe                    
+

.github/ansible/staging.eu-west-1.hosts.yaml

@@ -35,8 +35,6 @@ storage:
       hosts:
         pageserver-0.eu-west-1.aws.neon.build:
           ansible_host: i-01d496c5041c7f34c
-        pageserver-1.eu-west-1.aws.neon.build:
-          ansible_host: i-0e8013e239ce3928c
 
     safekeepers:
       hosts:
@@ -46,15 +44,3 @@ storage:
           ansible_host: i-06969ee1bf2958bfc
         safekeeper-2.eu-west-1.aws.neon.build:
           ansible_host: i-087892e9625984a0b
-        safekeeper-3.eu-west-1.aws.neon.build:
-          ansible_host: i-0a6f91660e99e8891
-        safekeeper-4.eu-west-1.aws.neon.build:
-          ansible_host: i-0012e309e28e7c249
-        safekeeper-5.eu-west-1.aws.neon.build:
-          ansible_host: i-085a2b1193287b32e
-        safekeeper-6.eu-west-1.aws.neon.build:
-          ansible_host: i-0c713248465ed0fbd
-        safekeeper-7.eu-west-1.aws.neon.build:
-          ansible_host: i-02ad231aed2a80b7a
-        safekeeper-8.eu-west-1.aws.neon.build:
-          ansible_host: i-0dbbd8ffef66efda8

.github/helm-values/dev-eu-central-1-alpha.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.alpha.eu-central-1.internal.aws.neon.build"
-
-settings:
-  domain: "*.snirouter.alpha.eu-central-1.internal.aws.neon.build"
-  sentryEnvironment: "staging"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/dev-eu-west-1-zeta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.zeta.eu-west-1.internal.aws.neon.build"
-
-settings:
-  domain: "*.snirouter.zeta.eu-west-1.internal.aws.neon.build"
-  sentryEnvironment: "staging"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/dev-us-east-2-beta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.beta.us-east-2.internal.aws.neon.build"
-
-settings:
-  domain: "*.snirouter.beta.us-east-2.internal.aws.neon.build"
-  sentryEnvironment: "staging"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-ap-southeast-1-epsilon.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.epsilon.ap-southeast-1.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.epsilon.ap-southeast-1.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-eu-central-1-gamma.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.gamma.eu-central-1.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.gamma.eu-central-1.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-us-east-1-theta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.theta.us-east-1.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.theta.us-east-1.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-us-east-2-delta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.delta.us-east-2.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.delta.us-east-2.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/helm-values/prod-us-west-2-eta.pg-sni-router.yaml

@@ -1,19 +0,0 @@
-useCertManager: true
-
-replicaCount: 3
-
-exposedService:
-  # exposedService.port -- Exposed Service proxy port
-  port: 4432
-  annotations:
-    external-dns.alpha.kubernetes.io/hostname: "*.snirouter.eta.us-west-2.internal.aws.neon.tech"
-
-settings:
-  domain: "*.snirouter.eta.us-west-2.internal.aws.neon.tech"
-  sentryEnvironment: "production"
-
-imagePullSecrets:
-  - name: docker-hub-neon
-
-metrics:
-  enabled: false

.github/workflows/deploy-dev.yml

@@ -27,11 +27,6 @@ on:
         required: true
         type: boolean
         default: true
-      deployPgSniRouter:
-        description: 'Deploy pg-sni-router'
-        required: true
-        type: boolean
-        default: true
 
 env:
   AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_DEV }}
@@ -232,49 +227,3 @@ jobs:
   
       - name: Cleanup helm folder
         run: rm -rf ~/.cache
-
-  deploy-pg-sni-router:
-    runs-on: [ self-hosted, gen3, small ]
-    container: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/ansible:pinned
-    if: inputs.deployPgSniRouter
-    defaults:
-      run:
-        shell: bash
-    strategy:
-      matrix:
-        include:
-          - target_region:  us-east-2
-            target_cluster: dev-us-east-2-beta
-          - target_region:  eu-west-1
-            target_cluster: dev-eu-west-1-zeta
-          - target_region:  eu-central-1
-            target_cluster: dev-eu-central-1-alpha
-    environment:
-      name: dev-${{ matrix.target_region }}
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          submodules: true
-          fetch-depth: 0
-          ref: ${{ inputs.branch }}
-  
-      - name: Configure AWS Credentials
-        uses: aws-actions/configure-aws-credentials@v1-node16
-        with:
-          role-to-assume: arn:aws:iam::369495373322:role/github-runner
-          aws-region: eu-central-1
-          role-skip-session-tagging: true
-          role-duration-seconds: 1800
-  
-      - name: Configure environment
-        run: |
-          helm repo add neondatabase https://neondatabase.github.io/helm-charts
-          aws --region ${{ matrix.target_region }} eks update-kubeconfig --name  ${{ matrix.target_cluster }}
-  
-      - name: Deploy pg-sni-router
-        run:
-          helm upgrade neon-pg-sni-router neondatabase/neon-pg-sni-router --namespace neon-pg-sni-router --create-namespace --install --atomic -f .github/helm-values/${{ matrix.target_cluster }}.pg-sni-router.yaml --set image.tag=${{ inputs.dockerTag }} --set settings.sentryUrl=${{ secrets.SENTRY_URL_BROKER }} --wait --timeout 15m0s
-  
-      - name: Cleanup helm folder
-        run: rm -rf ~/.cache

.github/workflows/deploy-prod.yml

@@ -27,11 +27,6 @@ on:
         required: true
         type: boolean
         default: true
-      deployPgSniRouter:
-        description: 'Deploy pg-sni-router'
-        required: true
-        type: boolean
-        default: true
       disclamerAcknowledged:
         description: 'I confirm that there is an emergency and I can not use regular release workflow'
         required: true
@@ -176,42 +171,3 @@ jobs:
       - name: Deploy storage-broker
         run:
           helm upgrade neon-storage-broker-lb neondatabase/neon-storage-broker --namespace neon-storage-broker-lb --create-namespace --install --atomic -f .github/helm-values/${{ matrix.target_cluster }}.neon-storage-broker.yaml --set image.tag=${{ inputs.dockerTag }} --set settings.sentryUrl=${{ secrets.SENTRY_URL_BROKER }} --wait --timeout 5m0s
-
-  deploy-pg-sni-router:
-    runs-on: prod
-    container: 093970136003.dkr.ecr.eu-central-1.amazonaws.com/ansible:latest
-    if: inputs.deployPgSniRouter && inputs.disclamerAcknowledged
-    defaults:
-      run:
-        shell: bash
-    strategy:
-      matrix:
-        include:
-          - target_region:  us-east-2
-            target_cluster: prod-us-east-2-delta
-          - target_region:  us-west-2
-            target_cluster: prod-us-west-2-eta
-          - target_region: eu-central-1
-            target_cluster: prod-eu-central-1-gamma
-          - target_region: ap-southeast-1
-            target_cluster: prod-ap-southeast-1-epsilon
-          - target_region: us-east-1
-            target_cluster: prod-us-east-1-theta
-    environment:
-      name: prod-${{ matrix.target_region }}
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          submodules: true
-          fetch-depth: 0
-          ref: ${{ inputs.branch }}
-  
-      - name: Configure environment
-        run: |
-          helm repo add neondatabase https://neondatabase.github.io/helm-charts
-          aws --region ${{ matrix.target_region }} eks update-kubeconfig --name  ${{ matrix.target_cluster }}
-  
-      - name: Deploy pg-sni-router
-        run:
-          helm upgrade neon-pg-sni-router neondatabase/neon-pg-sni-router --namespace neon-pg-sni-router --create-namespace --install --atomic -f .github/helm-values/${{ matrix.target_cluster }}.pg-sni-router.yaml --set image.tag=${{ inputs.dockerTag }} --set settings.sentryUrl=${{ secrets.SENTRY_URL_BROKER }} --wait --timeout 15m0s

Cargo.lock

@@ -3960,7 +3960,8 @@ dependencies = [
 [[package]]
 name = "sharded-slab"
 version = "0.1.4"
-source = "git+https://github.com/neondatabase/sharded-slab.git?rev=98d16753ab01c61f0a028de44167307a00efea00#98d16753ab01c61f0a028de44167307a00efea00"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "900fba806f70c630b0a382d0d825e17a0f19fcd059a2ade1ff237bcddf446b31"
 dependencies = [
  "lazy_static",
 ]

Cargo.toml

@@ -161,17 +161,11 @@ rstest = "0.17"
 tempfile = "3.4"
 tonic-build = "0.9"
 
-[patch.crates-io]
-
 # This is only needed for proxy's tests.
 # TODO: we should probably fork `tokio-postgres-rustls` instead.
+[patch.crates-io]
 tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
 
-# Changes the MAX_THREADS limit from 4096 to 32768.
-# This is a temporary workaround for using tracing from many threads in safekeepers code,
-# until async safekeepers patch is merged to the main.
-sharded-slab = { git = "https://github.com/neondatabase/sharded-slab.git", rev="98d16753ab01c61f0a028de44167307a00efea00" }
-
 ################# Binary contents sections
 
 [profile.release]

Dockerfile

@@ -44,15 +44,7 @@ COPY --chown=nonroot . .
 # Show build caching stats to check if it was used in the end.
 # Has to be the part of the same RUN since cachepot daemon is killed in the end of this RUN, losing the compilation stats.
 RUN set -e \
-    && mold -run cargo build  \
-      --bin pg_sni_router  \
-      --bin pageserver  \
-      --bin pageserver_binutils  \
-      --bin draw_timeline_dir \
-      --bin safekeeper  \
-      --bin storage_broker  \
-      --bin proxy  \
-      --locked --release \
+&& mold -run cargo build --bin pageserver --bin pageserver_binutils --bin draw_timeline_dir --bin safekeeper --bin storage_broker --bin proxy --locked --release \
     && cachepot -s
 
 # Build final image
@@ -71,7 +63,6 @@ RUN set -e \
     && useradd -d /data neon \
     && chown -R neon:neon /data
 
-COPY --from=build --chown=neon:neon /home/nonroot/target/release/pg_sni_router       /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/pageserver          /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/pageserver_binutils /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/draw_timeline_dir   /usr/local/bin

pageserver/src/metrics.rs

@@ -287,33 +287,14 @@ impl EvictionsWithLowResidenceDuration {
         let Some(_counter) = self.counter.take() else {
             return;
         };
-
-        let threshold = Self::threshold_label_value(self.threshold);
-
-        let removed = EVICTIONS_WITH_LOW_RESIDENCE_DURATION.remove_label_values(&[
-            tenant_id,
-            timeline_id,
-            self.data_source,
-            &threshold,
-        ]);
-
-        match removed {
-            Err(e) => {
-                // this has been hit in staging as
-                // <https://neondatabase.sentry.io/issues/4142396994/>, but we don't know how.
-                // because we can be in the drop path already, don't risk:
-                // - "double-panic => illegal instruction" or
-                // - future "drop panick => abort"
-                //
-                // so just nag: (the error has the labels)
-                tracing::warn!("failed to remove EvictionsWithLowResidenceDuration, it was already removed? {e:#?}");
-            }
-            Ok(()) => {
-                // to help identify cases where we double-remove the same values, let's log all
-                // deletions?
-                tracing::info!("removed EvictionsWithLowResidenceDuration with {tenant_id}, {timeline_id}, {}, {threshold}", self.data_source);
-            }
-        }
+        EVICTIONS_WITH_LOW_RESIDENCE_DURATION
+            .remove_label_values(&[
+                tenant_id,
+                timeline_id,
+                self.data_source,
+                &Self::threshold_label_value(self.threshold),
+            ])
+            .expect("we own the metric, no-one else should remove it");
     }
 }
 

pageserver/src/page_service.rs

@@ -352,7 +352,7 @@ impl PageServerHandler {
         tenant_id: TenantId,
         timeline_id: TimelineId,
         ctx: RequestContext,
-    ) -> Result<(), QueryError>
+    ) -> anyhow::Result<()>
     where
         IO: AsyncRead + AsyncWrite + Send + Sync + Unpin,
     {
@@ -398,9 +398,7 @@ impl PageServerHandler {
                 Some(FeMessage::CopyData(bytes)) => bytes,
                 Some(FeMessage::Terminate) => break,
                 Some(m) => {
-                    return Err(QueryError::Other(anyhow::anyhow!(
-                        "unexpected message: {m:?} during COPY"
-                    )));
+                    anyhow::bail!("unexpected message: {m:?} during COPY");
                 }
                 None => break, // client disconnected
             };

proxy/src/bin/auth_renew_router.rs

@@ -0,0 +1,183 @@
+use std::{net::SocketAddr, sync::Arc};
+use tokio::{io::AsyncWriteExt, net::TcpListener};
+
+use anyhow::Context;
+use clap::{self, Arg};
+use futures::TryFutureExt;
+use proxy::{
+    auth::{self, AuthFlow},
+    cancellation::CancelMap,
+    compute::ConnCfg,
+    console::messages::MetricsAuxInfo,
+};
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio_postgres::config::SslMode;
+use tokio_util::sync::CancellationToken;
+use utils::project_git_version;
+
+use tracing::{error, info, warn};
+
+project_git_version!(GIT_VERSION);
+
+fn cli() -> clap::Command {
+    clap::Command::new("Auth renew proxy")
+        .disable_help_flag(true)
+        .version(GIT_VERSION)
+        .arg(
+            Arg::new("listen")
+                .short('l')
+                .long("listen")
+                .help("listen for incoming client connections on ip:port")
+                .default_value("127.0.0.1:4432"),
+        )
+        .arg(
+            Arg::new("dest-host")
+                .long("dest-host")
+                .help("destination hosts")
+                .required(true),
+        )
+        .arg(
+            Arg::new("dest-port")
+                .long("dest-port")
+                .help("destination port")
+                .default_value("5432"),
+        )
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let _logging_guard = proxy::logging::init().await?;
+    let _panic_hook_guard = utils::logging::replace_panic_hook_with_tracing_panic_hook();
+
+    let args = cli().get_matches();
+
+    let dest_host: String = args.get_one::<String>("dest-host").unwrap().parse()?;
+    let dest_port: u16 = args.get_one::<String>("dest-port").unwrap().parse()?;
+    let listen_address: SocketAddr = args.get_one::<String>("listen").unwrap().parse()?;
+
+    // Start listening for incoming client connections
+    info!("Starting proxy on {listen_address}");
+    let proxy_listener = TcpListener::bind(listen_address).await?;
+
+    let cancellation_token = CancellationToken::new();
+
+    let main = proxy::flatten_err(tokio::spawn(task_main(
+        Arc::new(dest_host),
+        dest_port,
+        proxy_listener,
+        cancellation_token.clone(),
+    )));
+    let signals_task = proxy::flatten_err(tokio::spawn(proxy::handle_signals(cancellation_token)));
+
+    tokio::select! {
+        res = main => { res?; },
+        res = signals_task => { res?; },
+    }
+
+    Ok(())
+}
+
+async fn task_main(
+    dest_host: Arc<String>,
+    dest_port: u16,
+    listener: tokio::net::TcpListener,
+    cancellation_token: CancellationToken,
+) -> anyhow::Result<()> {
+    scopeguard::defer! {
+        info!("proxy has shut down");
+    }
+
+    // When set for the server socket, the keepalive setting
+    // will be inherited by all accepted client sockets.
+    socket2::SockRef::from(&listener).set_keepalive(true)?;
+
+    let mut connections = tokio::task::JoinSet::new();
+    let cancel_map = Arc::new(CancelMap::default());
+
+    loop {
+        tokio::select! {
+            accept_result = listener.accept() => {
+                let (socket, peer_addr) = accept_result?;
+                info!("accepted postgres client connection from {peer_addr}");
+
+                let cancel_map = Arc::clone(&cancel_map);
+                let dest_host = Arc::clone(&dest_host);
+
+                connections.spawn(
+                    async move {
+                        info!("spawned a task for {peer_addr}");
+
+                        socket
+                            .set_nodelay(true)
+                            .context("failed to set socket option")?;
+
+                        handle_client(dest_host, dest_port, &cancel_map, socket).await
+                    }
+                    .unwrap_or_else(|e| {
+                        // Acknowledge that the task has finished with an error.
+                        error!("per-client task finished with an error: {e:#}");
+                    }),
+                );
+            }
+            _ = cancellation_token.cancelled() => {
+                drop(listener);
+                break;
+            }
+        }
+    }
+    // Drain connections
+    while let Some(res) = connections.join_next().await {
+        if let Err(e) = res {
+            if !e.is_panic() && !e.is_cancelled() {
+                warn!("unexpected error from joined connection task: {e:?}");
+            }
+        }
+    }
+    Ok(())
+}
+
+async fn handle_client(
+    dest_host: Arc<String>,
+    dest_port: u16,
+    cancel_map: &CancelMap,
+    stream: impl AsyncRead + AsyncWrite + Unpin,
+) -> anyhow::Result<()> {
+    let do_handshake = proxy::proxy::handshake(stream, None, cancel_map);
+    let (mut stream, params) = match do_handshake.await? {
+        Some(x) => x,
+        None => return Ok(()), // it's a cancellation request
+    };
+
+    // Here we force plain test auth for the client and using received password to authenticate
+    // to the destination server. Instead we can always trust the client and take the password / JWT
+    // each time we get a connection.
+    let password = AuthFlow::new(&mut stream)
+        .begin(auth::CleartextPassword)
+        .await?
+        .authenticate()
+        .await?;
+
+    let mut conn_cfg = ConnCfg::new();
+    conn_cfg.set_startup_params(&params);
+    conn_cfg.password(password);
+    conn_cfg.host(dest_host.as_str());
+    conn_cfg.port(dest_port);
+    conn_cfg.ssl_mode(SslMode::Require);
+
+    info!("destination: {:?}:{}", dest_host, dest_port);
+
+    let mut conn = conn_cfg
+        .connect(false)
+        .or_else(|e| stream.throw_error(e))
+        .await?;
+
+    cancel_map
+        .with_session(|session| async {
+            proxy::proxy::prepare_client_connection(&conn, false, session, &mut stream).await?;
+            let (stream, read_buf) = stream.into_inner();
+            conn.stream.write_all(&read_buf).await?;
+            let metrics_aux: MetricsAuxInfo = Default::default();
+            proxy::proxy::proxy_pass(stream, conn.stream, &metrics_aux).await
+        })
+        .await
+}

proxy/src/logging.rs

@@ -1,4 +1,3 @@
-use tracing_opentelemetry::OpenTelemetryLayer;
 use tracing_subscriber::{
     filter::{EnvFilter, LevelFilter},
     prelude::*,
@@ -22,13 +21,8 @@ pub async fn init() -> anyhow::Result<LoggingGuard> {
         .with_writer(std::io::stderr)
         .with_target(false);
 
-    let otlp_layer = tracing_utils::init_tracing("proxy")
-        .await
-        .map(OpenTelemetryLayer::new);
-
     tracing_subscriber::registry()
         .with(env_filter)
-        .with(otlp_layer)
         .with(fmt_layer)
         .try_init()?;
 

proxy/src/proxy.rs

@@ -213,7 +213,7 @@ async fn handle_client(
 /// It's easier to work with owned `stream` here as we need to upgrade it to TLS;
 /// we also take an extra care of propagating only the select handshake errors to client.
 #[tracing::instrument(skip_all)]
-async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
+pub async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
     stream: S,
     mut tls: Option<&TlsConfig>,
     cancel_map: &CancelMap,
@@ -350,7 +350,7 @@ async fn connect_to_compute(
 
 /// Finish client connection initialization: confirm auth success, send params, etc.
 #[tracing::instrument(skip_all)]
-async fn prepare_client_connection(
+pub async fn prepare_client_connection(
     node: &compute::PostgresConnection,
     reported_auth_ok: bool,
     session: cancellation::Session<'_>,