Line data Source code
1 : use std::fs;
2 : use std::io::ErrorKind;
3 : use std::path::Path;
4 : use std::process::Command;
5 : use std::time::Duration;
6 : use std::{fs::OpenOptions, io::Write};
7 :
8 : use anyhow::{Context, Result, anyhow};
9 : use tracing::{error, info, instrument, warn};
10 :
11 : const POSTGRES_LOGS_CONF_PATH: &str = "/etc/rsyslog.d/postgres_logs.conf";
12 :
13 0 : fn get_rsyslog_pid() -> Option<String> {
14 0 : let output = Command::new("pgrep")
15 0 : .arg("rsyslogd")
16 0 : .output()
17 0 : .expect("Failed to execute pgrep");
18 0 :
19 0 : if !output.stdout.is_empty() {
20 0 : let pid = std::str::from_utf8(&output.stdout)
21 0 : .expect("Invalid UTF-8 in process output")
22 0 : .trim()
23 0 : .to_string();
24 0 : Some(pid)
25 : } else {
26 0 : None
27 : }
28 0 : }
29 :
30 : // Restart rsyslogd to apply the new configuration.
31 : // This is necessary, because there is no other way to reload the rsyslog configuration.
32 : //
33 : // Rsyslogd shouldn't lose any messages, because of the restart,
34 : // because it tracks the last read position in the log files
35 : // and will continue reading from that position.
36 : // TODO: test it properly
37 : //
38 0 : fn restart_rsyslog() -> Result<()> {
39 0 : let old_pid = get_rsyslog_pid().context("rsyslogd is not running")?;
40 0 : info!("rsyslogd is running with pid: {}, restart it", old_pid);
41 :
42 : // kill it to restart
43 0 : let _ = Command::new("pkill")
44 0 : .arg("rsyslogd")
45 0 : .output()
46 0 : .context("Failed to stop rsyslogd")?;
47 :
48 0 : Ok(())
49 0 : }
50 :
51 0 : pub fn configure_audit_rsyslog(
52 0 : log_directory: String,
53 0 : tag: &str,
54 0 : remote_endpoint: &str,
55 0 : ) -> Result<()> {
56 0 : let config_content: String = format!(
57 0 : include_str!("config_template/compute_audit_rsyslog_template.conf"),
58 0 : log_directory = log_directory,
59 0 : tag = tag,
60 0 : remote_endpoint = remote_endpoint
61 0 : );
62 0 :
63 0 : info!("rsyslog config_content: {}", config_content);
64 :
65 0 : let rsyslog_conf_path = "/etc/rsyslog.d/compute_audit_rsyslog.conf";
66 0 : let mut file = OpenOptions::new()
67 0 : .create(true)
68 0 : .write(true)
69 0 : .truncate(true)
70 0 : .open(rsyslog_conf_path)?;
71 :
72 0 : file.write_all(config_content.as_bytes())?;
73 :
74 0 : info!(
75 0 : "rsyslog configuration file {} added successfully. Starting rsyslogd",
76 : rsyslog_conf_path
77 : );
78 :
79 : // start the service, using the configuration
80 0 : restart_rsyslog()?;
81 :
82 0 : Ok(())
83 0 : }
84 :
85 : /// Configuration for enabling Postgres logs forwarding from rsyslogd
86 : pub struct PostgresLogsRsyslogConfig<'a> {
87 : pub host: Option<&'a str>,
88 : }
89 :
90 : impl<'a> PostgresLogsRsyslogConfig<'a> {
91 4 : pub fn new(host: Option<&'a str>) -> Self {
92 4 : Self { host }
93 4 : }
94 :
95 4 : pub fn build(&self) -> Result<String> {
96 4 : match self.host {
97 3 : Some(host) => {
98 3 : if let Some((target, port)) = host.split_once(":") {
99 2 : Ok(format!(
100 2 : include_str!(
101 2 : "config_template/compute_rsyslog_postgres_export_template.conf"
102 2 : ),
103 2 : logs_export_target = target,
104 2 : logs_export_port = port,
105 2 : ))
106 : } else {
107 1 : Err(anyhow!("Invalid host format for Postgres logs export"))
108 : }
109 : }
110 1 : None => Ok("".to_string()),
111 : }
112 4 : }
113 :
114 0 : fn current_config() -> Result<String> {
115 0 : let config_content = match std::fs::read_to_string(POSTGRES_LOGS_CONF_PATH) {
116 0 : Ok(c) => c,
117 0 : Err(err) if err.kind() == ErrorKind::NotFound => String::new(),
118 0 : Err(err) => return Err(err.into()),
119 : };
120 0 : Ok(config_content)
121 0 : }
122 :
123 : /// Returns the default host for otel collector that receives Postgres logs
124 1 : pub fn default_host(project_id: &str) -> String {
125 1 : format!(
126 1 : "config-{}-collector.neon-telemetry.svc.cluster.local:10514",
127 1 : project_id
128 1 : )
129 1 : }
130 : }
131 :
132 0 : pub fn configure_postgres_logs_export(conf: PostgresLogsRsyslogConfig) -> Result<()> {
133 0 : let new_config = conf.build()?;
134 0 : let current_config = PostgresLogsRsyslogConfig::current_config()?;
135 :
136 0 : if new_config == current_config {
137 0 : info!("postgres logs rsyslog configuration is up-to-date");
138 0 : return Ok(());
139 0 : }
140 0 :
141 0 : // When new config is empty we can simply remove the configuration file.
142 0 : if new_config.is_empty() {
143 0 : info!("removing rsyslog config file: {}", POSTGRES_LOGS_CONF_PATH);
144 0 : match std::fs::remove_file(POSTGRES_LOGS_CONF_PATH) {
145 0 : Ok(_) => {}
146 0 : Err(err) if err.kind() == ErrorKind::NotFound => {}
147 0 : Err(err) => return Err(err.into()),
148 : }
149 0 : restart_rsyslog()?;
150 0 : return Ok(());
151 0 : }
152 0 :
153 0 : info!(
154 0 : "configuring rsyslog for postgres logs export to: {:?}",
155 : conf.host
156 : );
157 :
158 0 : let mut file = OpenOptions::new()
159 0 : .create(true)
160 0 : .write(true)
161 0 : .truncate(true)
162 0 : .open(POSTGRES_LOGS_CONF_PATH)?;
163 0 : file.write_all(new_config.as_bytes())?;
164 :
165 0 : info!(
166 0 : "rsyslog configuration file {} added successfully. Starting rsyslogd",
167 : POSTGRES_LOGS_CONF_PATH
168 : );
169 :
170 0 : restart_rsyslog()?;
171 0 : Ok(())
172 0 : }
173 :
174 : #[instrument(skip_all)]
175 : async fn pgaudit_gc_main_loop(log_directory: String) -> Result<()> {
176 : info!("running pgaudit GC main loop");
177 : loop {
178 : // Check log_directory for old pgaudit logs and delete them.
179 : // New log files are checked every 5 minutes, as set in pgaudit.log_rotation_age
180 : // Find files that were not modified in the last 15 minutes and delete them.
181 : // This should be enough time for rsyslog to process the logs and for us to catch the alerts.
182 : //
183 : // In case of a very high load, we might need to adjust this value and pgaudit.log_rotation_age.
184 : //
185 : // TODO: add some smarter logic to delete the files that are fully streamed according to rsyslog
186 : // imfile-state files, but for now just do a simple GC to avoid filling up the disk.
187 : let _ = Command::new("find")
188 : .arg(&log_directory)
189 : .arg("-name")
190 : .arg("audit*.log")
191 : .arg("-mmin")
192 : .arg("+15")
193 : .arg("-delete")
194 : .output()?;
195 :
196 : // also collect the metric for the size of the log directory
197 0 : async fn get_log_files_size(path: &Path) -> Result<u64> {
198 0 : let mut total_size = 0;
199 :
200 0 : for entry in fs::read_dir(path)? {
201 0 : let entry = entry?;
202 0 : let entry_path = entry.path();
203 0 :
204 0 : if entry_path.is_file() && entry_path.to_string_lossy().ends_with("log") {
205 0 : total_size += entry.metadata()?.len();
206 0 : }
207 : }
208 :
209 0 : Ok(total_size)
210 0 : }
211 :
212 : let log_directory_size = get_log_files_size(Path::new(&log_directory))
213 : .await
214 0 : .unwrap_or_else(|e| {
215 0 : warn!("Failed to get log directory size: {}", e);
216 0 : 0
217 0 : });
218 : crate::metrics::AUDIT_LOG_DIR_SIZE.set(log_directory_size as f64);
219 : tokio::time::sleep(Duration::from_secs(60)).await;
220 : }
221 : }
222 :
223 : // launch pgaudit GC thread to clean up the old pgaudit logs stored in the log_directory
224 0 : pub fn launch_pgaudit_gc(log_directory: String) {
225 0 : tokio::spawn(async move {
226 0 : if let Err(e) = pgaudit_gc_main_loop(log_directory).await {
227 0 : error!("pgaudit GC main loop failed: {}", e);
228 0 : }
229 0 : });
230 0 : }
231 :
232 : #[cfg(test)]
233 : mod tests {
234 : use crate::rsyslog::PostgresLogsRsyslogConfig;
235 :
236 : #[test]
237 1 : fn test_postgres_logs_config() {
238 1 : {
239 1 : // Verify empty config
240 1 : let conf = PostgresLogsRsyslogConfig::new(None);
241 1 : let res = conf.build();
242 1 : assert!(res.is_ok());
243 1 : let conf_str = res.unwrap();
244 1 : assert_eq!(&conf_str, "");
245 : }
246 :
247 : {
248 : // Verify config
249 1 : let conf = PostgresLogsRsyslogConfig::new(Some("collector.cvc.local:514"));
250 1 : let res = conf.build();
251 1 : assert!(res.is_ok());
252 1 : let conf_str = res.unwrap();
253 1 : assert!(conf_str.contains("omfwd"));
254 1 : assert!(conf_str.contains(r#"target="collector.cvc.local""#));
255 1 : assert!(conf_str.contains(r#"port="514""#));
256 : }
257 :
258 : {
259 : // Verify invalid config
260 1 : let conf = PostgresLogsRsyslogConfig::new(Some("invalid"));
261 1 : let res = conf.build();
262 1 : assert!(res.is_err());
263 : }
264 :
265 : {
266 : // Verify config with default host
267 1 : let host = PostgresLogsRsyslogConfig::default_host("shy-breeze-123");
268 1 : let conf = PostgresLogsRsyslogConfig::new(Some(&host));
269 1 : let res = conf.build();
270 1 : assert!(res.is_ok());
271 1 : let conf_str = res.unwrap();
272 1 : assert!(conf_str.contains(r#"shy-breeze-123"#));
273 1 : assert!(conf_str.contains(r#"port="10514""#));
274 : }
275 1 : }
276 : }
|
