[RFC proxmox{,-backup} 0/3] fix #6373: HTTP level keepalive for http2 backup reader connection

[RFC proxmox{,-backup} 0/3] fix #6373: HTTP level keepalive for http2 backup reader connection

[Christian Ebner]

Patches implement a HTTP level keepalive mechanism to avoid of active
but idle http2 connections established by the backup reader being
dropped by reverse proxies.
Regular heartbeat requests are send to the Proxmox Backup Server, the
frequency based on a given timeout. Heartbeat traffic is only send if
no other requests have been send during the timeout period. The
timeout given in seconds is read from the
PBS_READER_HEARTBEAT_TIMEOUT environment variable, if not set no
heartbeat traffic is being send.

Since each HTTP request is being logged to the task log for http2
connections, implement a filtering logic allowing to exclude these
request logs to avoid log flodding.

If the chosen approach is acceptable, the timeout value might also
be passed as optional parameter on command invocation as well and
exposed in PVE as advanced parameter, although only important if
proxies are being used.

Example configurations for testing can be found on the individual
patch fixing the issue.

Link to the bugtracker issue:
https://bugzilla.proxmox.com/show_bug.cgi?id=6373

proxmox:

Christian Ebner (1):
  rest-server: add request logfilter by method and path in h2 service

 proxmox-rest-server/src/h2service.rs | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)


proxmox-backup:

Christian Ebner (2):
  fix #6373: HTTP level reader heartbeat for proxy connection keepalive
  api: h2service: avoid logging heartbeat requests to task log

 pbs-client/src/backup_reader.rs | 69 ++++++++++++++++++++++++++++++---
 src/api2/backup/mod.rs          |  2 +-
 src/api2/reader/mod.rs          | 22 +++++++++--
 3 files changed, 84 insertions(+), 9 deletions(-)


Summary over all repositories:
  4 files changed, 100 insertions(+), 11 deletions(-)

-- 
Generated by git-murpp 0.8.1

[PATCH proxmox 1/1] rest-server: add request logfilter by method and path in h2 service

[Christian Ebner]

Allows to optionally filter incoming requests to be excluded from the
task log. The filter method is provided on instantiation and passes
the request method and path.

A first usecase for this is to filter out HTTP level heartbeat
requests send by the client's backup reader implementation in order
to avoid log flodding.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 proxmox-rest-server/src/h2service.rs | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)

diff --git a/proxmox-rest-server/src/h2service.rs b/proxmox-rest-server/src/h2service.rs
index 09843477..2645f25e 100644
--- a/proxmox-rest-server/src/h2service.rs
+++ b/proxmox-rest-server/src/h2service.rs
@@ -27,15 +27,23 @@ pub struct H2Service<E> {
     rpcenv: E,
     worker: Arc<WorkerTask>,
     debug: bool,
+    log_filter: Option<fn(&hyper::Method, &str) -> bool>,
 }
 
 impl<E: RpcEnvironment + Clone> H2Service<E> {
-    pub fn new(rpcenv: E, worker: Arc<WorkerTask>, router: &'static Router, debug: bool) -> Self {
+    pub fn new(
+        rpcenv: E,
+        worker: Arc<WorkerTask>,
+        router: &'static Router,
+        debug: bool,
+        log_filter: Option<fn(&hyper::Method, &str) -> bool>,
+    ) -> Self {
         Self {
             rpcenv,
             worker,
             router,
             debug,
+            log_filter,
         }
     }
 
@@ -55,7 +63,13 @@ impl<E: RpcEnvironment + Clone> H2Service<E> {
             Err(err) => return future::err(http_err!(BAD_REQUEST, "{}", err)).boxed(),
         };
 
-        self.debug(format!("{method} {path}"));
+        if !self
+            .log_filter
+            .map(|filter| filter(&method, &path))
+            .unwrap_or_default()
+        {
+            self.debug(format!("{method} {path}"));
+        }
 
         let mut uri_param = HashMap::new();
 
-- 
2.47.3

[PATCH proxmox-backup 1/2] fix #6373: HTTP level reader heartbeat for proxy connection keepalive

[Christian Ebner]

Backup readers can have long periods of idle connections, e.g. if a
backup snapshot has been mounted and all relevant chunks are locally
cached or a backup session with previous metadata archive not needing
to fetch new contents while the backup is ongoing.

Proxies like e.g. HAProxy might however close idle connections for
better resource handling [0,1], even multiplexed HTTP/2 connections as
are being used for the Proxmox Backup Sever backup/reader protocol.

This mainly affects the backup reader, while the backup writer will
do indexing and chunk uploads anyways.

Therefore, perform heartbeat traffic in the backup reader http2 client
when no other requests are being send. To do so, an new `heartbeat`
API endpoint is introduced as part of the backup reader protocol,
returning empty on GET requests. By making this part of the backup
reader protocol, this is limited to active reader HTTP/2 sessions.

On the client side the heartbeat is send out periodically whenever a
timeout is being reached, the timeout however being reset if other
requests are being performed via the http2 client.

Since older servers do not provide the new API path, ignore errors
as the response is not strictly necessary for the connection to
remain established.

The timeout is currently only being performed if the timeout value
in seconds is given via the PBS_READER_HEARTBEAT_TIMEOUT.

Testing was performed using HAProxy with the Proxmox Backup Server
as backend using the following 5 second connection idle timeouts
as configuration parameters in haproxy.cfg:
```
...
defaults
    ...
    timeout connect 5000
    timeout client  5000
    timeout server  5000
    ..

frontend http-in
    bind *:8007
    mode tcp
    default_backend pbs

backend pbs
    mode tcp
    http-reuse always
    server pbs <PBS-IP>:8007 verify none
```

As command invocation:
```
PBS_READER_HEARTBEAT_TIMEOUT=1 proxmox-backup-client mount <snapshot> <archive> \
<mountpoint> --repository <user-and-realm>@<PROXY-IP>:<datastore> --verbose
```

[0] https://www.haproxy.com/documentation/haproxy-configuration-manual/latest/#4-timeout%20client
[1] https://www.haproxy.com/documentation/haproxy-configuration-manual/latest/#4.2-timeout%20http-keep-alive

Fixes: https://bugzilla.proxmox.com/show_bug.cgi?id=6373
Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 pbs-client/src/backup_reader.rs | 69 ++++++++++++++++++++++++++++++---
 src/api2/reader/mod.rs          |  9 ++++-
 2 files changed, 72 insertions(+), 6 deletions(-)

diff --git a/pbs-client/src/backup_reader.rs b/pbs-client/src/backup_reader.rs
index 88cba599b..95236bef2 100644
--- a/pbs-client/src/backup_reader.rs
+++ b/pbs-client/src/backup_reader.rs
@@ -1,10 +1,13 @@
 use anyhow::{format_err, Error};
 use std::fs::File;
 use std::io::{Seek, SeekFrom, Write};
+use std::str::FromStr;
 use std::sync::Arc;
+use std::time::Duration;
 
 use futures::future::AbortHandle;
 use serde_json::{json, Value};
+use tokio::sync::mpsc;
 
 use pbs_api_types::{BackupArchiveName, BackupDir, BackupNamespace, MANIFEST_BLOB_NAME};
 use pbs_datastore::data_blob::DataBlob;
@@ -18,26 +21,65 @@ use pbs_tools::sha::sha256;
 
 use super::{H2Client, HttpClient};
 
+enum HeartBeatMsg {
+    Reset,
+    Abort,
+}
+
 /// Backup Reader
 pub struct BackupReader {
     h2: H2Client,
     abort: AbortHandle,
     crypt_config: Option<Arc<CryptConfig>>,
+    heartbeat: Option<mpsc::Sender<HeartBeatMsg>>,
 }
 
 impl Drop for BackupReader {
     fn drop(&mut self) {
+        self.send_msg_heartbeat(HeartBeatMsg::Abort);
         self.abort.abort();
     }
 }
 
 impl BackupReader {
     fn new(h2: H2Client, abort: AbortHandle, crypt_config: Option<Arc<CryptConfig>>) -> Arc<Self> {
-        Arc::new(Self {
-            h2,
-            abort,
-            crypt_config,
-        })
+        let timeout = match std::env::var("PBS_READER_HEARTBEAT_TIMEOUT") {
+            Ok(val) => u64::from_str(&val).map(Duration::from_secs).ok(),
+            Err(_err) => None,
+        };
+
+        if let Some(timeout) = timeout {
+            let (send, mut recv) = mpsc::channel(1);
+            let backup_reader = Arc::new(Self {
+                h2,
+                abort,
+                crypt_config,
+                heartbeat: Some(send),
+            });
+            let reader_cloned = Arc::clone(&backup_reader);
+
+            tokio::spawn(async move {
+                loop {
+                    match tokio::time::timeout(timeout, recv.recv()).await {
+                        Ok(Some(HeartBeatMsg::Reset)) => (),
+                        Ok(Some(HeartBeatMsg::Abort)) | Ok(None) => break,
+                        Err(_elapsed) => {
+                            // connection idle timeout reached, send heatbeat
+                            let _ = reader_cloned.h2.get("heartbeat", None).await;
+                        }
+                    }
+                }
+            });
+
+            backup_reader
+        } else {
+            Arc::new(Self {
+                h2,
+                abort,
+                crypt_config,
+                heartbeat: None,
+            })
+        }
     }
 
     /// Create a new instance by upgrading the connection at '/api2/json/reader'
@@ -79,21 +121,25 @@ impl BackupReader {
 
     /// Execute a GET request
     pub async fn get(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
+        self.send_msg_heartbeat(HeartBeatMsg::Reset);
         self.h2.get(path, param).await
     }
 
     /// Execute a PUT request
     pub async fn put(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
+        self.send_msg_heartbeat(HeartBeatMsg::Reset);
         self.h2.put(path, param).await
     }
 
     /// Execute a POST request
     pub async fn post(&self, path: &str, param: Option<Value>) -> Result<Value, Error> {
+        self.send_msg_heartbeat(HeartBeatMsg::Reset);
         self.h2.post(path, param).await
     }
 
     /// Execute a GET request and send output to a writer
     pub async fn download<W: Write + Send>(&self, file_name: &str, output: W) -> Result<(), Error> {
+        self.send_msg_heartbeat(HeartBeatMsg::Reset);
         let path = "download";
         let param = json!({ "file-name": file_name });
         self.h2.download(path, Some(param), output).await
@@ -103,6 +149,7 @@ impl BackupReader {
     ///
     /// This writes random data, and is only useful to test download speed.
     pub async fn speedtest<W: Write + Send>(&self, output: W) -> Result<(), Error> {
+        self.send_msg_heartbeat(HeartBeatMsg::Reset);
         self.h2.download("speedtest", None, output).await
     }
 
@@ -112,12 +159,14 @@ impl BackupReader {
         digest: &[u8; 32],
         output: W,
     ) -> Result<(), Error> {
+        self.send_msg_heartbeat(HeartBeatMsg::Reset);
         let path = "chunk";
         let param = json!({ "digest": hex::encode(digest) });
         self.h2.download(path, Some(param), output).await
     }
 
     pub fn force_close(self) {
+        self.send_msg_heartbeat(HeartBeatMsg::Abort);
         self.abort.abort();
     }
 
@@ -205,4 +254,14 @@ impl BackupReader {
 
         Ok(index)
     }
+
+    /// Send given message to the heartbeat closure.
+    ///
+    /// All errors are being ignored, since they cannot be handled anyways.
+    fn send_msg_heartbeat(&self, msg: HeartBeatMsg) {
+        let _ = self
+            .heartbeat
+            .as_ref()
+            .map(|heartbeat| heartbeat.try_send(msg));
+    }
 }
diff --git a/src/api2/reader/mod.rs b/src/api2/reader/mod.rs
index f7adc366f..1c673d7a6 100644
--- a/src/api2/reader/mod.rs
+++ b/src/api2/reader/mod.rs
@@ -18,7 +18,7 @@ use proxmox_router::{
     http_err, list_subdirs_api_method, ApiHandler, ApiMethod, ApiResponseFuture, Permission,
     Router, RpcEnvironment, SubdirMap,
 };
-use proxmox_schema::{BooleanSchema, ObjectSchema};
+use proxmox_schema::{api, BooleanSchema, ObjectSchema};
 use proxmox_sortable_macro::sortable;
 
 use pbs_api_types::{
@@ -227,6 +227,7 @@ const READER_API_SUBDIRS: SubdirMap = &[
         "download",
         &Router::new().download(&API_METHOD_DOWNLOAD_FILE),
     ),
+    ("heartbeat", &Router::new().download(&API_METHOD_HEARTBEAT)),
     ("speedtest", &Router::new().download(&API_METHOD_SPEEDTEST)),
 ];
 
@@ -433,3 +434,9 @@ fn speedtest(
 
     future::ok(response).boxed()
 }
+
+#[api()]
+/// HTTP level heartbeat to avoid proxies closing long running idle backup reader connections.
+pub async fn heartbeat(_rpcenv: &mut dyn RpcEnvironment) -> Result<(), Error> {
+    Ok(())
+}
-- 
2.47.3

[PATCH proxmox-backup 2/2] api: h2service: avoid logging heartbeat requests to task log

[Christian Ebner]

Heartbeat requests can happen with a frequency as high as 1/s, adding
individual lines to the task log.

Use the http2 service filter function to exclude these request from
being logged by matching method and path accordingly.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
 src/api2/backup/mod.rs |  2 +-
 src/api2/reader/mod.rs | 13 +++++++++++--
 2 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/src/api2/backup/mod.rs b/src/api2/backup/mod.rs
index 3e6b7a950..3f32759ad 100644
--- a/src/api2/backup/mod.rs
+++ b/src/api2/backup/mod.rs
@@ -240,7 +240,7 @@ fn upgrade_to_backup_protocol(
                 ));
 
                 let service =
-                    H2Service::new(env.clone(), worker.clone(), &BACKUP_API_ROUTER, debug);
+                    H2Service::new(env.clone(), worker.clone(), &BACKUP_API_ROUTER, debug, None);
 
                 let abort_future = worker.abort_future();
 
diff --git a/src/api2/reader/mod.rs b/src/api2/reader/mod.rs
index 1c673d7a6..562e1f3f6 100644
--- a/src/api2/reader/mod.rs
+++ b/src/api2/reader/mod.rs
@@ -172,8 +172,13 @@ fn upgrade_to_backup_reader_protocol(
                     "starting new backup reader datastore '{store}': {path:?}"
                 ));
 
-                let service =
-                    H2Service::new(env.clone(), worker.clone(), &READER_API_ROUTER, debug);
+                let service = H2Service::new(
+                    env.clone(),
+                    worker.clone(),
+                    &READER_API_ROUTER,
+                    debug,
+                    Some(heartbeat_request_filter),
+                );
 
                 let mut abort_future = worker
                     .abort_future()
@@ -440,3 +445,7 @@ fn speedtest(
 pub async fn heartbeat(_rpcenv: &mut dyn RpcEnvironment) -> Result<(), Error> {
     Ok(())
 }
+
+fn heartbeat_request_filter(method: &hyper::Method, path: &str) -> bool {
+    method == hyper::Method::GET && path == "/heartbeat"
+}
-- 
2.47.3