Updated tier watcher (#24)

Author: paulobressan

.gitignore

@@ -5,4 +5,4 @@ cert
 docker-compose.yml
 .env*
 env-crd.yaml
-tiers.toml
+tiers*

proxy/src/config.rs

@@ -1,10 +1,11 @@
-use std::{env, path::PathBuf};
+use std::{env, path::PathBuf, time::Duration};
 
 #[derive(Debug, Clone)]
 pub struct Config {
     pub proxy_addr: String,
     pub proxy_namespace: String,
     pub proxy_tiers_path: PathBuf,
+    pub proxy_tiers_poll_interval: Duration,
     pub prometheus_addr: String,
     pub ssl_crt_path: String,
     pub ssl_key_path: String,
@@ -17,8 +18,16 @@ impl Config {
             proxy_addr: env::var("PROXY_ADDR").expect("PROXY_ADDR must be set"),
             proxy_namespace: env::var("PROXY_NAMESPACE").expect("PROXY_NAMESPACE must be set"),
             proxy_tiers_path: env::var("PROXY_TIERS_PATH")
-            .map(|v| v.into())
-            .expect("PROXY_TIERS_PATH must be set"),
+                .map(|v| v.into())
+                .expect("PROXY_TIERS_PATH must be set"),
+            proxy_tiers_poll_interval: env::var("PROXY_TIERS_POLL_INTERVAL")
+                .map(|v| {
+                    Duration::from_secs(
+                        v.parse::<u64>()
+                            .expect("PROXY_TIERS_POLL_INTERVAL must be a number in seconds. eg: 2"),
+                    )
+                })
+                .unwrap_or(Duration::from_secs(2)),
             prometheus_addr: env::var("PROMETHEUS_ADDR").expect("PROMETHEUS_ADDR must be set"),
             ssl_crt_path: env::var("SSL_CRT_PATH").expect("SSL_CRT_PATH must be set"),
             ssl_key_path: env::var("SSL_KEY_PATH").expect("SSL_KEY_PATH must be set"),

proxy/src/tiers.rs

@@ -1,11 +1,9 @@
-use std::error::Error;
-use std::{fs, sync::Arc};
-
 use async_trait::async_trait;
-use notify::{Event, RecommendedWatcher, RecursiveMode, Watcher};
+use notify::{PollWatcher, RecursiveMode, Watcher};
 use pingora::{server::ShutdownWatch, services::background::BackgroundService};
 use serde_json::Value;
-use tokio::runtime::{Handle, Runtime};
+use std::error::Error;
+use std::{fs, sync::Arc};
 use tracing::{error, info, warn};
 
 use crate::{config::Config, State, Tier};
@@ -42,16 +40,6 @@ impl TierBackgroundService {
     }
 }
 
-fn runtime_handle() -> Handle {
-    match Handle::try_current() {
-        Ok(h) => h,
-        Err(_) => {
-            let rt = Runtime::new().unwrap();
-            rt.handle().clone()
-        }
-    }
-}
-
 #[async_trait]
 impl BackgroundService for TierBackgroundService {
     async fn start(&self, mut _shutdown: ShutdownWatch) {
@@ -60,19 +48,13 @@ impl BackgroundService for TierBackgroundService {
             return;
         }
 
-        let (tx, mut rx) = tokio::sync::mpsc::channel::<Event>(1);
+        let (tx, rx) = std::sync::mpsc::channel();
 
-        let watcher_result = RecommendedWatcher::new(
-            move |result: Result<Event, notify::Error>| {
-                let event = result.unwrap();
-                if event.kind.is_modify() {
-                    runtime_handle().block_on(async {
-                        tx.send(event).await.unwrap();
-                    });
-                }
-            },
-            notify::Config::default(),
-        );
+        let watcher_config = notify::Config::default()
+            .with_compare_contents(true)
+            .with_poll_interval(self.config.proxy_tiers_poll_interval);
+
+        let watcher_result = PollWatcher::new(tx, watcher_config);
         if let Err(err) = watcher_result {
             error!(error = err.to_string(), "error to watcher tier");
             return;
@@ -85,14 +67,17 @@ impl BackgroundService for TierBackgroundService {
             return;
         }
 
-        loop {
-            if rx.recv().await.is_some() {
-                if let Err(err) = self.update_tiers().await {
-                    error!(error = err.to_string(), "error to update tiers");
-                    continue;
-                }
+        for result in rx {
+            match result {
+                Ok(_event) => {
+                    if let Err(err) = self.update_tiers().await {
+                        error!(error = err.to_string(), "error to update tiers");
+                        continue;
+                    }
 
-                info!("tiers modified");
+                    info!("tiers modified");
+                }
+                Err(err) => error!(error = err.to_string(), "watch error"),
             }
         }
     }