Skip to content

Make add connection summary as background operations. #405

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Zhidong Peng (ZhidongPeng) merged 4 commits into from
Sep 9, 2026
Merged

Make add connection summary as background operations. #405

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
34774b4
Make add connection summary as background operations.
Sep 8, 2026
3dea170
Do not redact the secrets in the command line at proxying time
Sep 8, 2026
0c56546
fix tyop
Sep 9, 2026
76c48ad
Merge branch 'dev' into connection
ZhidongPeng Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions proxy_agent/src/proxy.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,8 +232,10 @@ impl Process {
cmd = process_info.1;
}

// redact the secrets in the command line
let cmd = proxy_agent_shared::secrets_redactor::redact_secrets_string(cmd);
// Do not redact the secrets in the command line at proxying time,
// because the command line is used for authorization and redacting it may break the authorization.
// Instead, redact the secrets in the command line in background before logging or send out.
// let cmd = proxy_agent_shared::secrets_redactor::redact_secrets_string(cmd);

let process_name = process_full_path
.file_name()
Expand Down
52 changes: 10 additions & 42 deletions proxy_agent/src/proxy/proxy_server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -922,58 +922,26 @@ impl ProxyServer {
);

if log_authorize_failed {
match self
if let Err(e) = self
.connection_summary_shared_state
.add_one_failed_connection_summary(summary.clone())
.await
{
Ok(is_new_bucket) => {
if is_new_bucket {
// if it's a new bucket, we don't need to add to failed connection summary again
if let Ok(json) = serde_json::to_string(&summary) {
event_logger::write_event(
LoggerLevel::Info,
json,
"log_connection_summary",
"proxy_server",
ConnectionLogger::CONNECTION_LOGGER_KEY,
);
};
}
}
Err(e) => {
http_connection_context.log(
LoggerLevel::Warn,
format!("Failed to add failed connection summary: {e}"),
);
}
http_connection_context.log(
LoggerLevel::Warn,
format!("Failed to add failed connection summary: {e}"),
);
}
} else {
match self
if let Err(e) = self
.connection_summary_shared_state
.add_one_connection_summary(summary.clone())
.await
{
Ok(is_new_bucket) => {
if is_new_bucket {
// if it's a new bucket, we log it to event logger
if let Ok(json) = serde_json::to_string(&summary) {
event_logger::write_event(
LoggerLevel::Info,
json,
"log_connection_summary",
"proxy_server",
ConnectionLogger::CONNECTION_LOGGER_KEY,
);
};
}
}
Err(e) => {
http_connection_context.log(
LoggerLevel::Warn,
format!("Failed to add connection summary: {e}"),
);
}
http_connection_context.log(
LoggerLevel::Warn,
format!("Failed to add connection summary: {e}"),
);
}
}

Expand Down
75 changes: 42 additions & 33 deletions proxy_agent/src/shared_state/connection_summary_wrapper.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,12 @@

use crate::common::logger;
use crate::common::result::Result;
use crate::proxy::proxy_connection::ConnectionLogger;
use crate::{common::error::Error, proxy::proxy_summary::ProxySummary};
use proxy_agent_shared::logger::LoggerLevel;
use proxy_agent_shared::proxy_agent_aggregate_status::ProxyConnectionSummary;
use proxy_agent_shared::secrets_redactor;
use proxy_agent_shared::telemetry::event_logger;
use proxy_agent_shared::time_buckets::TimeBucketedItem;
use std::collections::{hash_map, HashMap};
use tokio::sync::{mpsc, oneshot};
Expand All @@ -19,11 +23,9 @@ const MAX_AGE_SECS: u64 = 4 * 3600; // 4 hours
enum ConnectionSummaryAction {
AddOneConnection {
summary: ProxySummary,
response: oneshot::Sender<bool>,
},
AddOneFailedConnection {
summary: ProxySummary,
response: oneshot::Sender<bool>,
},
GetAllConnection {
response: oneshot::Sender<Vec<ProxyConnectionSummary>>,
Expand Down Expand Up @@ -56,30 +58,45 @@ impl ConnectionSummarySharedState {

while let Some(action) = rx.recv().await {
match action {
ConnectionSummaryAction::AddOneConnection { summary, response } => {
ConnectionSummaryAction::AddOneConnection { mut summary } => {
// redact possible secrets from the process command line in background before storing it in the time-bucketed item
summary.processCmdLine =
secrets_redactor::redact_secrets_string(summary.processCmdLine.clone());
let mut is_new_bucket = true;
let key = summary.to_key_string();
if let hash_map::Entry::Vacant(e) = proxy_summary.entry(key.clone()) {
e.insert(TimeBucketedItem::new(
summary.into(),
summary.clone().into(),
bucket_duration,
max_age_duration,
));
} else if let Some(connection_summary) = proxy_summary.get_mut(&key) {
is_new_bucket = connection_summary.add_one();
}
if response.send(is_new_bucket).is_err() {
logger::write_warning("Failed to send response to ConnectionSummaryAction::AddOneConnection".to_string());
if is_new_bucket {
// if it's a new bucket, we log it to event logger
if let Ok(json) = serde_json::to_string(&summary) {
event_logger::write_event(
LoggerLevel::Info,
json,
"log_connection_summary",
"proxy_server",
ConnectionLogger::CONNECTION_LOGGER_KEY,
);
};
}
}
ConnectionSummaryAction::AddOneFailedConnection { summary, response } => {
ConnectionSummaryAction::AddOneFailedConnection { mut summary } => {
// redact possible secrets from the process command line in background before storing it in the time-bucketed item
summary.processCmdLine =
secrets_redactor::redact_secrets_string(summary.processCmdLine.clone());
let mut is_new_bucket = true;
let key = summary.to_key_string();
if let hash_map::Entry::Vacant(e) =
failed_authenticate_summary.entry(key.clone())
{
e.insert(TimeBucketedItem::new(
summary.into(),
summary.clone().into(),
bucket_duration,
max_age_duration,
));
Expand All @@ -88,8 +105,17 @@ impl ConnectionSummarySharedState {
{
is_new_bucket = connection_summary.add_one();
}
if response.send(is_new_bucket).is_err() {
logger::write_warning("Failed to send response to ConnectionSummaryAction::AddOneFailedConnection".to_string());
if is_new_bucket {
// if it's a new bucket, we log it to event logger
if let Ok(json) = serde_json::to_string(&summary) {
event_logger::write_event(
LoggerLevel::Info,
json,
"log_connection_summary",
"proxy_server",
ConnectionLogger::CONNECTION_LOGGER_KEY,
);
};
}
}
ConnectionSummaryAction::GetAllConnection { response } => {
Expand Down Expand Up @@ -137,50 +163,33 @@ impl ConnectionSummarySharedState {
}

/// Add one connection summary
/// Returns true if a new time-bucketed item was created.
/// It does implicitly removes expired time-bucketed items
pub async fn add_one_connection_summary(&self, summary: ProxySummary) -> Result<bool> {
let (response_tx, response_rx) = oneshot::channel();
pub async fn add_one_connection_summary(&self, summary: ProxySummary) -> Result<()> {
self.0
.send(ConnectionSummaryAction::AddOneConnection {
summary,
response: response_tx,
})
.send(ConnectionSummaryAction::AddOneConnection { summary })
.await
.map_err(|e| {
Error::SendError(
"ConnectionSummaryAction::AddOneConnection".to_string(),
e.to_string(),
)
})?;
response_rx.await.map_err(|e| {
Error::RecvError("ConnectionSummaryAction::AddOneConnection".to_string(), e)
})
Ok(())
}

/// Add one failed connection summary
/// Returns true if a new time bucket is created for this summary, false otherwise
/// It does implicitly removes expired time-bucketed items
pub async fn add_one_failed_connection_summary(&self, summary: ProxySummary) -> Result<bool> {
let (response_tx, response_rx) = oneshot::channel();
pub async fn add_one_failed_connection_summary(&self, summary: ProxySummary) -> Result<()> {
self.0
.send(ConnectionSummaryAction::AddOneFailedConnection {
summary,
response: response_tx,
})
.send(ConnectionSummaryAction::AddOneFailedConnection { summary })
.await
.map_err(|e| {
Error::SendError(
"ConnectionSummaryAction::AddOneFailedConnection".to_string(),
e.to_string(),
)
})?;
response_rx.await.map_err(|e| {
Error::RecvError(
"ConnectionSummaryAction::AddOneFailedConnection".to_string(),
e,
)
})
Ok(())
}

/// Clear both connection summaries explicitly
Expand Down
Loading