mirror of
https://github.com/sadoyan/aralez.git
synced 2026-10-23 06:01:35 +08:00
file delete
This commit is contained in:
Ara Sadoyan
@@ -1,91 +0,0 @@
|
|||||||
use crate::utils::httpclient;
|
|
||||||
use crate::utils::kuberconsul::{list_to_upstreams, ServiceDiscovery};
|
|
||||||
use crate::utils::parceyaml::build_headers;
|
|
||||||
use crate::utils::structs::{Configuration, GlobalServiceMapping, UpstreamsDashMap};
|
|
||||||
use async_trait::async_trait;
|
|
||||||
use dashmap::DashMap;
|
|
||||||
use pingora::prelude::sleep;
|
|
||||||
use rand::RngExt;
|
|
||||||
use serde::Deserialize;
|
|
||||||
use std::env;
|
|
||||||
use std::sync::Arc;
|
|
||||||
use std::time::Duration;
|
|
||||||
use tokio::sync::mpsc::Sender;
|
|
||||||
|
|
||||||
#[derive(Debug, Deserialize)]
|
|
||||||
pub struct ConsulService {
|
|
||||||
#[serde(rename = "Address")]
|
|
||||||
pub address: String,
|
|
||||||
#[serde(rename = "ServicePort")]
|
|
||||||
pub port: u16,
|
|
||||||
}
|
|
||||||
pub struct ConsulDiscovery;
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl ServiceDiscovery for ConsulDiscovery {
|
|
||||||
async fn fetch_upstreams(&self, config: Arc<Configuration>, toreturn: Sender<Configuration>) {
|
|
||||||
loop {
|
|
||||||
let upstreams = UpstreamsDashMap::new();
|
|
||||||
|
|
||||||
if let Some(consul) = config.consul.clone() {
|
|
||||||
let servers = consul.servers.unwrap_or_else(|| {
|
|
||||||
vec![format!(
|
|
||||||
"{}:{}",
|
|
||||||
env::var("CONSUL_SERVICE_HOST").unwrap_or_else(|_| "0.0.0.0".to_string()),
|
|
||||||
env::var("CONSUL_SERVICE_PORT").unwrap_or_else(|_| "0".to_string())
|
|
||||||
)]
|
|
||||||
});
|
|
||||||
let end = servers.len().saturating_sub(1);
|
|
||||||
let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 };
|
|
||||||
let consul_data = servers.get(num).unwrap().to_string();
|
|
||||||
let ss = format!("{}/v1/catalog/service/", consul_data);
|
|
||||||
let dc = format!("{}/v1/catalog/services?filter=%22aralez.service=yes%22+in+ServiceTags", consul_data);
|
|
||||||
|
|
||||||
if let Some(infolist) = httpclient::for_consul_list(dc.as_str(), consul.token.clone()).await {
|
|
||||||
for (k, v) in infolist {
|
|
||||||
let client_header_map: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
|
|
||||||
let mut client_header_list = Vec::new();
|
|
||||||
build_headers(&v.client_headers, config.as_ref(), &mut client_header_list);
|
|
||||||
|
|
||||||
let sever_header_map: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
|
|
||||||
let mut server_header_list = Vec::new();
|
|
||||||
build_headers(&v.server_headers, config.as_ref(), &mut server_header_list);
|
|
||||||
|
|
||||||
if !client_header_list.is_empty() {
|
|
||||||
let path_key = v.path.clone();
|
|
||||||
client_header_map.insert(Arc::from(path_key), client_header_list);
|
|
||||||
config.client_headers.insert(Arc::from(v.host.clone()), client_header_map);
|
|
||||||
}
|
|
||||||
if !server_header_list.is_empty() {
|
|
||||||
let path_key = v.path.clone();
|
|
||||||
sever_header_map.insert(Arc::from(path_key), server_header_list);
|
|
||||||
config.server_headers.insert(Arc::from(v.host.clone()), sever_header_map);
|
|
||||||
}
|
|
||||||
let pref = format!("{}{}", ss, k);
|
|
||||||
let gsm = &GlobalServiceMapping {
|
|
||||||
upstream: k.clone(),
|
|
||||||
hostname: v.host.clone(),
|
|
||||||
path: Some(v.path),
|
|
||||||
to_https: v.to_https,
|
|
||||||
redirect_to: v.redirect,
|
|
||||||
authorization: v.auth,
|
|
||||||
rate_limit: v.rate,
|
|
||||||
x4xx_limit: v.xrate,
|
|
||||||
client_headers: v.client_headers,
|
|
||||||
server_headers: v.server_headers,
|
|
||||||
};
|
|
||||||
|
|
||||||
let list = httpclient::for_consul(pref.as_str(), consul.token.clone(), gsm).await;
|
|
||||||
list_to_upstreams(list, &upstreams, gsm);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some(lt) = crate::utils::kuberconsul::clone_compare(&upstreams, &config).await {
|
|
||||||
let _ = toreturn.send(lt).await;
|
|
||||||
}
|
|
||||||
|
|
||||||
sleep(Duration::from_secs(5)).await;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,67 +0,0 @@
|
|||||||
use crate::utils::consul::ConsulDiscovery;
|
|
||||||
use crate::utils::filewatch;
|
|
||||||
use crate::utils::kuberconsul::ServiceDiscovery;
|
|
||||||
use crate::utils::kubernetes::KubernetesDiscovery;
|
|
||||||
use crate::utils::structs::{Configuration, UpstreamsDashMap};
|
|
||||||
use crate::web::webserver;
|
|
||||||
use async_trait::async_trait;
|
|
||||||
use std::sync::Arc;
|
|
||||||
use tokio::sync::mpsc::Sender;
|
|
||||||
|
|
||||||
pub struct APIUpstreamProvider {
|
|
||||||
pub config_api_enabled: bool,
|
|
||||||
pub address: String,
|
|
||||||
pub masterkey: Option<String>,
|
|
||||||
pub certs_dir: String,
|
|
||||||
pub config_dir: String,
|
|
||||||
pub upstreams_file: String,
|
|
||||||
pub file_server_address: Option<String>,
|
|
||||||
pub file_server_folder: Option<String>,
|
|
||||||
pub current_upstreams: Arc<UpstreamsDashMap>,
|
|
||||||
pub full_upstreams: Arc<UpstreamsDashMap>,
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct FromFileProvider {
|
|
||||||
pub path: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct ConsulProvider {
|
|
||||||
pub config: Arc<Configuration>,
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct KubernetesProvider {
|
|
||||||
pub config: Arc<Configuration>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
pub trait Discovery {
|
|
||||||
async fn start(&self, tx: Sender<Configuration>);
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Discovery for APIUpstreamProvider {
|
|
||||||
async fn start(&self, toreturn: Sender<Configuration>) {
|
|
||||||
webserver::run_server(self, toreturn, self.current_upstreams.clone(), self.full_upstreams.clone()).await;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Discovery for FromFileProvider {
|
|
||||||
async fn start(&self, tx: Sender<Configuration>) {
|
|
||||||
tokio::spawn(filewatch::start(self.path.clone(), tx));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Discovery for ConsulProvider {
|
|
||||||
async fn start(&self, tx: Sender<Configuration>) {
|
|
||||||
tokio::spawn(ConsulDiscovery.fetch_upstreams(self.config.clone(), tx));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Discovery for KubernetesProvider {
|
|
||||||
async fn start(&self, tx: Sender<Configuration>) {
|
|
||||||
tokio::spawn(KubernetesDiscovery.fetch_upstreams(self.config.clone(), tx));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,63 +0,0 @@
|
|||||||
use crate::utils::structs::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap};
|
|
||||||
use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams};
|
|
||||||
use async_trait::async_trait;
|
|
||||||
use dashmap::DashMap;
|
|
||||||
use std::sync::atomic::AtomicUsize;
|
|
||||||
use std::sync::Arc;
|
|
||||||
use tokio::fs::File;
|
|
||||||
use tokio::io::AsyncReadExt;
|
|
||||||
use tokio::sync::mpsc::Sender;
|
|
||||||
|
|
||||||
#[allow(clippy::type_complexity)]
|
|
||||||
pub fn list_to_upstreams(lt: Option<DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>>, upstreams: &UpstreamsDashMap, i: &GlobalServiceMapping) {
|
|
||||||
if let Some(list) = lt {
|
|
||||||
let key: Arc<str> = Arc::from(i.hostname.as_str());
|
|
||||||
match upstreams.get(&key) {
|
|
||||||
Some(upstr) => {
|
|
||||||
for (k, v) in list {
|
|
||||||
upstr.value().insert(k, v);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
None => {
|
|
||||||
upstreams.insert(key, list);
|
|
||||||
}
|
|
||||||
};
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn match_path(conf: &GlobalServiceMapping, upstreams: &DashMap<Arc<str>, (Vec<Arc<InnerMap>>, AtomicUsize)>, values: Vec<Arc<InnerMap>>) {
|
|
||||||
let path = conf.path.as_deref().unwrap_or("/");
|
|
||||||
upstreams.insert(Arc::from(path), (values, AtomicUsize::new(0)));
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn read_token(path: &str) -> String {
|
|
||||||
let mut file = File::open(path).await.unwrap();
|
|
||||||
let mut contents = String::new();
|
|
||||||
file.read_to_string(&mut contents).await.unwrap();
|
|
||||||
contents.trim().to_string()
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
pub trait ServiceDiscovery {
|
|
||||||
async fn fetch_upstreams(&self, config: Arc<Configuration>, toreturn: Sender<Configuration>);
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn clone_compare(upstreams: &UpstreamsDashMap, config: &Arc<Configuration>) -> Option<Configuration> {
|
|
||||||
if !compare_dashmaps(upstreams, &config.upstreams) {
|
|
||||||
let tosend = Configuration {
|
|
||||||
upstreams: Default::default(),
|
|
||||||
client_headers: config.client_headers.clone(),
|
|
||||||
server_headers: config.server_headers.clone(),
|
|
||||||
consul: config.consul.clone(),
|
|
||||||
kubernetes: config.kubernetes.clone(),
|
|
||||||
typecfg: config.typecfg.clone(),
|
|
||||||
extraparams: config.extraparams.clone(),
|
|
||||||
};
|
|
||||||
clone_dashmap_into(upstreams, &config.upstreams);
|
|
||||||
clone_dashmap_into(upstreams, &tosend.upstreams);
|
|
||||||
|
|
||||||
print_upstreams(&tosend.upstreams, &tosend.extraparams);
|
|
||||||
return Some(tosend);
|
|
||||||
}
|
|
||||||
None
|
|
||||||
}
|
|
||||||
@@ -1,163 +0,0 @@
|
|||||||
use crate::utils::httpclient;
|
|
||||||
use crate::utils::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate};
|
|
||||||
use crate::utils::parceyaml::build_headers;
|
|
||||||
use crate::utils::structs::{Configuration, GlobalServiceMapping, UpstreamsDashMap};
|
|
||||||
use async_trait::async_trait;
|
|
||||||
use dashmap::DashMap;
|
|
||||||
use log::error;
|
|
||||||
use rand::RngExt;
|
|
||||||
use std::env;
|
|
||||||
use std::sync::Arc;
|
|
||||||
|
|
||||||
use crate::utils::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery};
|
|
||||||
use rustls::crypto::ring::default_provider;
|
|
||||||
use rustls::crypto::CryptoProvider;
|
|
||||||
use tokio::sync::mpsc::Sender;
|
|
||||||
|
|
||||||
#[derive(Debug, serde::Deserialize)]
|
|
||||||
#[serde(rename_all = "camelCase")]
|
|
||||||
pub struct KubeEndpointSliceList {
|
|
||||||
#[serde(default)]
|
|
||||||
pub items: Vec<KubeEndpointSlice>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, serde::Deserialize)]
|
|
||||||
#[serde(rename_all = "camelCase")]
|
|
||||||
pub struct KubeEndpointSlice {
|
|
||||||
pub ports: Option<Vec<KubePort>>,
|
|
||||||
#[serde(default)]
|
|
||||||
pub endpoints: Vec<KubeEndpoint>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, serde::Deserialize)]
|
|
||||||
#[serde(rename_all = "camelCase")]
|
|
||||||
pub struct KubeEndpoint {
|
|
||||||
#[serde(default)]
|
|
||||||
pub addresses: Vec<String>,
|
|
||||||
pub conditions: Option<KubeConditions>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, serde::Deserialize)]
|
|
||||||
#[serde(rename_all = "camelCase")]
|
|
||||||
pub struct KubeConditions {
|
|
||||||
pub ready: Option<bool>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, serde::Deserialize)]
|
|
||||||
#[serde(rename_all = "camelCase")]
|
|
||||||
pub struct KubePort {
|
|
||||||
pub port: Option<u16>,
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct KubernetesDiscovery;
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl ServiceDiscovery for KubernetesDiscovery {
|
|
||||||
async fn fetch_upstreams(&self, config: Arc<Configuration>, toreturn: Sender<Configuration>) {
|
|
||||||
if let Some(kuber) = config.kubernetes.clone() {
|
|
||||||
let servers = kuber.servers.unwrap_or_else(|| {
|
|
||||||
vec![format!(
|
|
||||||
"{}:{}",
|
|
||||||
env::var("KUBERNETES_SERVICE_HOST").unwrap_or_else(|_| "0.0.0.0".to_string()),
|
|
||||||
env::var("KUBERNETES_SERVICE_PORT_HTTPS").unwrap_or_else(|_| "0".to_string())
|
|
||||||
)]
|
|
||||||
});
|
|
||||||
|
|
||||||
let end = servers.len().saturating_sub(1);
|
|
||||||
let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 };
|
|
||||||
let server = servers.get(num).unwrap().to_string();
|
|
||||||
let path = kuber.tokenpath.unwrap_or_else(|| "/var/run/secrets/kubernetes.io/serviceaccount/token".to_string());
|
|
||||||
let token = crate::utils::kuberconsul::read_token(path.as_str()).await;
|
|
||||||
|
|
||||||
if CryptoProvider::get_default().is_none() {
|
|
||||||
default_provider().install_default().expect("Failed to install rustls crypto provider");
|
|
||||||
}
|
|
||||||
|
|
||||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<AralezRouteUpdate>(100);
|
|
||||||
let srv = server.clone();
|
|
||||||
let tok = token.clone();
|
|
||||||
tokio::spawn(async move {
|
|
||||||
if let Err(e) = start_ingress_watcher_with_config(srv.as_str(), tok.as_str(), tx, "aralez").await {
|
|
||||||
error!("Fatal watcher error: {:?}", e);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
let upstreams = UpstreamsDashMap::new();
|
|
||||||
|
|
||||||
while let Some(update) = rx.recv().await {
|
|
||||||
let host_key: Arc<str> = Arc::from(update.host.as_str());
|
|
||||||
|
|
||||||
if update.is_deleted {
|
|
||||||
upstreams.remove(&host_key);
|
|
||||||
config.client_headers.remove(&host_key);
|
|
||||||
} else {
|
|
||||||
let service = GlobalServiceMapping {
|
|
||||||
upstream: update.service_name.clone(),
|
|
||||||
hostname: update.host.clone(),
|
|
||||||
path: Some(update.path.clone()),
|
|
||||||
to_https: None,
|
|
||||||
redirect_to: None,
|
|
||||||
authorization: None,
|
|
||||||
rate_limit: update.rate_limit,
|
|
||||||
x4xx_limit: update.x4xx_limit,
|
|
||||||
client_headers: None,
|
|
||||||
server_headers: None,
|
|
||||||
};
|
|
||||||
|
|
||||||
let cheader_list: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
|
|
||||||
let sheader_list: DashMap<Arc<str>, Vec<(String, Arc<str>)>> = DashMap::new();
|
|
||||||
|
|
||||||
let mut client_headers_list = Vec::new();
|
|
||||||
let mut server_headers_list = Vec::new();
|
|
||||||
|
|
||||||
let mut chl: Option<Vec<String>> = None;
|
|
||||||
let mut shl: Option<Vec<String>> = None;
|
|
||||||
|
|
||||||
if let Some(hdr) = service.client_headers.as_ref() {
|
|
||||||
chl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned());
|
|
||||||
}
|
|
||||||
if let Some(hdr) = service.server_headers.as_ref() {
|
|
||||||
shl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned());
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some(ch) = update.client_headers {
|
|
||||||
chl.get_or_insert_with(Vec::new).extend(ch);
|
|
||||||
}
|
|
||||||
if let Some(hdr) = update.server_headers.as_ref() {
|
|
||||||
shl.get_or_insert_with(Vec::new).extend(hdr.iter().cloned());
|
|
||||||
}
|
|
||||||
|
|
||||||
build_headers(&chl, config.as_ref(), &mut client_headers_list);
|
|
||||||
if !client_headers_list.is_empty() {
|
|
||||||
let path_key = service.path.as_deref().unwrap_or("/");
|
|
||||||
cheader_list.insert(Arc::from(path_key), client_headers_list);
|
|
||||||
config.client_headers.insert(host_key.clone(), cheader_list);
|
|
||||||
}
|
|
||||||
|
|
||||||
build_headers(&shl, config.as_ref(), &mut server_headers_list);
|
|
||||||
if !server_headers_list.is_empty() {
|
|
||||||
let path_key = service.path.as_deref().unwrap_or("/");
|
|
||||||
sheader_list.insert(Arc::from(path_key), server_headers_list);
|
|
||||||
config.server_headers.insert(host_key.clone(), sheader_list);
|
|
||||||
}
|
|
||||||
|
|
||||||
let url = format!(
|
|
||||||
"https://{}/apis/discovery.k8s.io/v1/namespaces/{}/endpointslices?labelSelector=kubernetes.io%2Fservice-name%3D{}",
|
|
||||||
server, update.namespace, update.service_name
|
|
||||||
);
|
|
||||||
|
|
||||||
let list = httpclient::for_kuber(&url, &token, &service).await;
|
|
||||||
|
|
||||||
if list.is_none() {
|
|
||||||
error!("Endpoint URL: {} returned empty/invalid response", url);
|
|
||||||
}
|
|
||||||
|
|
||||||
list_to_upstreams(list, &upstreams, &service);
|
|
||||||
}
|
|
||||||
if let Some(lt) = clone_compare(&upstreams, &config).await {
|
|
||||||
let _ = toreturn.send(lt).await;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,279 +0,0 @@
|
|||||||
use dashmap::DashMap;
|
|
||||||
use futures::StreamExt;
|
|
||||||
use hyper::Uri;
|
|
||||||
use k8s_openapi::api::core::v1::{Node, Service};
|
|
||||||
use k8s_openapi::api::networking::v1::Ingress;
|
|
||||||
use kube::api::{Patch, PatchParams};
|
|
||||||
use kube::{
|
|
||||||
api::Api,
|
|
||||||
client::Client,
|
|
||||||
config::{AuthInfo, Config},
|
|
||||||
runtime::watcher::{self, Event},
|
|
||||||
};
|
|
||||||
use log::{error, info};
|
|
||||||
use serde_json::json;
|
|
||||||
use std::sync::Arc;
|
|
||||||
use std::time::Duration;
|
|
||||||
use tokio::sync::mpsc;
|
|
||||||
|
|
||||||
pub struct AralezRouteUpdate {
|
|
||||||
pub host: String,
|
|
||||||
pub path: String,
|
|
||||||
pub namespace: String,
|
|
||||||
pub service_name: String,
|
|
||||||
pub is_deleted: bool,
|
|
||||||
pub rate_limit: Option<isize>,
|
|
||||||
pub x4xx_limit: Option<u32>,
|
|
||||||
pub client_headers: Option<Vec<String>>,
|
|
||||||
pub server_headers: Option<Vec<String>>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
|
||||||
struct AralezAnnotation {
|
|
||||||
rate_limit: Option<isize>,
|
|
||||||
x4xx_limit: Option<u32>,
|
|
||||||
namespace: String,
|
|
||||||
client_headers: Option<Vec<String>>,
|
|
||||||
server_headers: Option<Vec<String>>,
|
|
||||||
}
|
|
||||||
|
|
||||||
lazy_static::lazy_static! {
|
|
||||||
static ref PATCHED_INGRESSES: Arc<DashMap<String, String>> = Arc::new(DashMap::new());
|
|
||||||
}
|
|
||||||
pub async fn start_ingress_watcher_with_config(
|
|
||||||
api_server: &str,
|
|
||||||
token: &str,
|
|
||||||
tx: mpsc::Sender<AralezRouteUpdate>,
|
|
||||||
target_class: &'static str,
|
|
||||||
) -> Result<(), Box<dyn std::error::Error>> {
|
|
||||||
let client = create_custom_k8s_client(api_server, token, true).await?;
|
|
||||||
let ingresses: Api<Ingress> = Api::all(client.clone());
|
|
||||||
|
|
||||||
let watcher_config = watcher::Config::default().any_semantic();
|
|
||||||
let mut stream = watcher::watcher(ingresses, watcher_config).boxed();
|
|
||||||
|
|
||||||
while let Some(event) = stream.next().await {
|
|
||||||
match event {
|
|
||||||
Ok(Event::Apply(ing)) => {
|
|
||||||
process_ingress_change(&client, &ing, target_class, false, &tx).await;
|
|
||||||
}
|
|
||||||
Ok(Event::Delete(ing)) => {
|
|
||||||
process_ingress_change(&client, &ing, target_class, true, &tx).await;
|
|
||||||
}
|
|
||||||
Ok(Event::InitApply(ing)) => {
|
|
||||||
process_ingress_change(&client, &ing, target_class, false, &tx).await;
|
|
||||||
}
|
|
||||||
Ok(Event::InitDone) => {
|
|
||||||
info!("Ingress watcher initial sync complete");
|
|
||||||
}
|
|
||||||
Err(err) => {
|
|
||||||
error!("Kubernetes Ingress watch error: {:?}", err);
|
|
||||||
}
|
|
||||||
_ => {}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn process_ingress_change(client: &Client, ing: &Ingress, target_class: &str, is_deleted: bool, tx: &mpsc::Sender<AralezRouteUpdate>) {
|
|
||||||
let spec = match &ing.spec {
|
|
||||||
Some(s) => s,
|
|
||||||
None => return,
|
|
||||||
};
|
|
||||||
let class_name = spec.ingress_class_name.as_deref().unwrap_or_default();
|
|
||||||
if class_name != target_class {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
if !is_deleted {
|
|
||||||
if let (Some(name), Some(ns)) = (ing.metadata.name.as_deref(), ing.metadata.namespace.as_deref()) {
|
|
||||||
let key = format!("{}/{}", ns, name);
|
|
||||||
let external_ip = resolve_external_ip(client).await;
|
|
||||||
let status_already_correct = ing
|
|
||||||
.status
|
|
||||||
.as_ref()
|
|
||||||
.and_then(|s| s.load_balancer.as_ref())
|
|
||||||
.and_then(|lb| lb.ingress.as_ref())
|
|
||||||
.map(|list| list.iter().any(|i| i.ip.as_deref() == Some(&external_ip)))
|
|
||||||
.unwrap_or(false);
|
|
||||||
|
|
||||||
if status_already_correct {
|
|
||||||
PATCHED_INGRESSES.insert(key, external_ip);
|
|
||||||
} else {
|
|
||||||
PATCHED_INGRESSES.insert(key.clone(), external_ip.clone());
|
|
||||||
let client_clone = client.clone();
|
|
||||||
let name_clone = name.to_string();
|
|
||||||
let ns_clone = ns.to_string();
|
|
||||||
let key_clone = key.clone();
|
|
||||||
tokio::spawn(async move {
|
|
||||||
if let Err(e) = patch_ingress_status_address(&client_clone, &ns_clone, &name_clone, &external_ip).await {
|
|
||||||
error!("Failed to patch ingress status address {}/{}: {}", &ns_clone, &name_clone, e);
|
|
||||||
PATCHED_INGRESSES.remove(&key_clone);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
if let (Some(name), Some(ns)) = (ing.metadata.name.as_deref(), ing.metadata.namespace.as_deref()) {
|
|
||||||
PATCHED_INGRESSES.remove(&format!("{}/{}", ns, name));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if let Some(rules) = &spec.rules {
|
|
||||||
for rule in rules {
|
|
||||||
let host = rule.host.clone().unwrap_or_else(|| "*".to_string());
|
|
||||||
|
|
||||||
if let Some(http) = &rule.http {
|
|
||||||
for path_entry in &http.paths {
|
|
||||||
let path = path_entry.path.clone().unwrap_or_else(|| "/".to_string());
|
|
||||||
if let Some(service) = &path_entry.backend.service {
|
|
||||||
let service_name = service.name.clone();
|
|
||||||
let annot = parse_ingress_config(ing);
|
|
||||||
let update = AralezRouteUpdate {
|
|
||||||
host: host.clone(),
|
|
||||||
path,
|
|
||||||
service_name,
|
|
||||||
is_deleted,
|
|
||||||
rate_limit: annot.rate_limit,
|
|
||||||
x4xx_limit: annot.x4xx_limit,
|
|
||||||
client_headers: annot.client_headers,
|
|
||||||
server_headers: annot.server_headers,
|
|
||||||
namespace: annot.namespace,
|
|
||||||
};
|
|
||||||
if let Err(e) = tx.send(update).await {
|
|
||||||
error!("Failed to send route update to channel: {:?}", e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn create_custom_k8s_client(api_server_url: &str, bearer_token: &str, insecure_tls: bool) -> Result<Client, Box<dyn std::error::Error>> {
|
|
||||||
let full_url = if !api_server_url.starts_with("http://") && !api_server_url.starts_with("https://") {
|
|
||||||
format!("https://{}", api_server_url)
|
|
||||||
} else {
|
|
||||||
api_server_url.to_string()
|
|
||||||
};
|
|
||||||
let uri: Uri = full_url.parse()?;
|
|
||||||
let mut config = Config::new(uri);
|
|
||||||
config.auth_info = AuthInfo {
|
|
||||||
token: Some(bearer_token.trim().to_string().into()),
|
|
||||||
..Default::default()
|
|
||||||
};
|
|
||||||
config.connect_timeout = Some(Duration::from_secs(10));
|
|
||||||
config.read_timeout = None;
|
|
||||||
if insecure_tls {
|
|
||||||
config.accept_invalid_certs = true;
|
|
||||||
}
|
|
||||||
let client = Client::try_from(config)?;
|
|
||||||
Ok(client)
|
|
||||||
}
|
|
||||||
|
|
||||||
fn parse_ingress_config(ing: &Ingress) -> AralezAnnotation {
|
|
||||||
let namespace = ing.metadata.namespace.clone().unwrap_or_else(|| "default".to_string());
|
|
||||||
let annotations = ing.metadata.annotations.as_ref();
|
|
||||||
|
|
||||||
let client_headers = annotations.and_then(|a| a.get("aralez.rs/client_headers")).and_then(|ch| serde_json::from_str(ch).ok());
|
|
||||||
|
|
||||||
let server_headers = annotations.and_then(|a| a.get("aralez.rs/server_headers")).and_then(|sh| serde_json::from_str(sh).ok());
|
|
||||||
|
|
||||||
AralezAnnotation {
|
|
||||||
rate_limit: annotations.and_then(|a| a.get("aralez.rs/rate_limit")).and_then(|v| v.parse().ok()),
|
|
||||||
x4xx_limit: annotations.and_then(|a| a.get("aralez.rs/x4xx_limit")).and_then(|v| v.parse().ok()),
|
|
||||||
client_headers,
|
|
||||||
server_headers,
|
|
||||||
namespace,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn patch_ingress_status_address(client: &Client, namespace: &str, name: &str, external_ip: &str) -> Result<(), kube::Error> {
|
|
||||||
let ingresses: Api<Ingress> = Api::namespaced(client.clone(), namespace);
|
|
||||||
let patch = json!({
|
|
||||||
"status": {
|
|
||||||
"loadBalancer": {
|
|
||||||
"ingress": [
|
|
||||||
{
|
|
||||||
"ip": external_ip
|
|
||||||
}
|
|
||||||
]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
|
||||||
let params = PatchParams::default();
|
|
||||||
|
|
||||||
match ingresses.patch_status(name, ¶ms, &Patch::Strategic(patch)).await {
|
|
||||||
Ok(_) => {
|
|
||||||
info!("Successfully patched Ingress {}/{} ADDRESS to {}", namespace, name, external_ip);
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
Err(e) => {
|
|
||||||
error!("Failed to patch status for Ingress {}/{}: {:?}", namespace, name, e);
|
|
||||||
Err(e)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn discover_node_ip(client: &Client) -> Option<String> {
|
|
||||||
let nodes: Api<Node> = Api::all(client.clone());
|
|
||||||
if let Ok(node_list) = nodes.list(&Default::default()).await {
|
|
||||||
for node in node_list {
|
|
||||||
if let Some(status) = node.status {
|
|
||||||
if let Some(addresses) = status.addresses {
|
|
||||||
for addr in &addresses {
|
|
||||||
if addr.type_ == "ExternalIP" {
|
|
||||||
return Some(addr.address.clone());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
for addr in &addresses {
|
|
||||||
if addr.type_ == "InternalIP" {
|
|
||||||
return Some(addr.address.clone());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
None
|
|
||||||
}
|
|
||||||
pub async fn discover_controller_ip(client: &Client, namespace: &str, service_name: &str) -> Option<String> {
|
|
||||||
let services: Api<Service> = Api::namespaced(client.clone(), namespace);
|
|
||||||
if let Ok(svc) = services.get(service_name).await {
|
|
||||||
if let Some(status) = svc.status {
|
|
||||||
if let Some(lb) = status.load_balancer {
|
|
||||||
if let Some(ingress_list) = lb.ingress {
|
|
||||||
for ing in ingress_list {
|
|
||||||
if let Some(ip) = ing.ip {
|
|
||||||
return Some(ip);
|
|
||||||
}
|
|
||||||
if let Some(hostname) = ing.hostname {
|
|
||||||
return Some(hostname);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
None
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn resolve_external_ip(client: &Client) -> String {
|
|
||||||
if let Ok(ip) = std::env::var("ARALEZ_EXTERNAL_IP") {
|
|
||||||
let trimmed = ip.trim();
|
|
||||||
if !trimmed.is_empty() {
|
|
||||||
return trimmed.to_string();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
let svc_name = std::env::var("ARALEZ_SERVICE_NAME").unwrap_or_else(|_| "aralez-service".to_string());
|
|
||||||
let svc_ns = std::env::var("ARALEZ_SERVICE_NAMESPACE").unwrap_or_else(|_| "aralez".to_string());
|
|
||||||
|
|
||||||
if let Some(ip) = discover_controller_ip(client, &svc_ns, &svc_name).await {
|
|
||||||
return ip;
|
|
||||||
}
|
|
||||||
if let Some(ip) = discover_node_ip(client).await {
|
|
||||||
return ip;
|
|
||||||
}
|
|
||||||
|
|
||||||
"127.0.0.1".to_string()
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user