mirror of
https://github.com/sadoyan/aralez.git
synced 2026-10-23 06:01:35 +08:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Ara Sadoyan
|
56c41e2c19 | ||
|
Ara Sadoyan
|
6ea8b42835 | ||
|
Ara Sadoyan
|
af771287cb | ||
|
Ara Sadoyan
|
0c489b649a | ||
|
Ara Sadoyan
|
f895ba4554 |
Generated
+70
-4
@@ -173,6 +173,7 @@ dependencies = [
|
||||
"prometheus",
|
||||
"rand 0.10.2",
|
||||
"rcgen",
|
||||
"redis",
|
||||
"reqwest",
|
||||
"rustls 0.23.41",
|
||||
"rustls-pemfile",
|
||||
@@ -198,6 +199,12 @@ dependencies = [
|
||||
"rustversion",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "arcstr"
|
||||
version = "1.2.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d"
|
||||
|
||||
[[package]]
|
||||
name = "arraydeque"
|
||||
version = "0.5.1"
|
||||
@@ -285,6 +292,17 @@ dependencies = [
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-lock"
|
||||
version = "3.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
|
||||
dependencies = [
|
||||
"event-listener",
|
||||
"event-listener-strategy",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-stream"
|
||||
version = "0.3.6"
|
||||
@@ -1080,12 +1098,16 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
|
||||
|
||||
[[package]]
|
||||
name = "combine"
|
||||
version = "4.6.7"
|
||||
version = "4.6.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd"
|
||||
checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"memchr",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1481,7 +1503,7 @@ dependencies = [
|
||||
"asn1-rs",
|
||||
"displaydoc",
|
||||
"nom",
|
||||
"num-bigint",
|
||||
"num-bigint 0.4.8",
|
||||
"num-traits",
|
||||
"rusticata-macros",
|
||||
]
|
||||
@@ -3350,6 +3372,16 @@ dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num-bigint"
|
||||
version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0"
|
||||
dependencies = [
|
||||
"num-integer",
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num-bigint-dig"
|
||||
version = "0.8.6"
|
||||
@@ -4224,6 +4256,34 @@ dependencies = [
|
||||
"yasna",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redis"
|
||||
version = "1.7.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ed5b98ae99461e02895c3a96a25dbe4d852a3d5e5b815d31d5d9e4b5344d99e9"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"arcstr",
|
||||
"async-lock",
|
||||
"backon",
|
||||
"bytes",
|
||||
"cfg-if",
|
||||
"combine",
|
||||
"futures-channel",
|
||||
"futures-util",
|
||||
"itoa",
|
||||
"num-bigint 0.5.1",
|
||||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"ryu",
|
||||
"sha1_smol",
|
||||
"socket2 0.6.4",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"url",
|
||||
"xxhash-rust",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redox_syscall"
|
||||
version = "0.5.18"
|
||||
@@ -4885,6 +4945,12 @@ dependencies = [
|
||||
"digest 0.10.7",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sha1_smol"
|
||||
version = "1.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d"
|
||||
|
||||
[[package]]
|
||||
name = "sha2"
|
||||
version = "0.10.9"
|
||||
@@ -4971,7 +5037,7 @@ version = "0.6.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0d585997b0ac10be3c5ee635f1bab02d512760d14b7c468801ac8a01d9ae5f1d"
|
||||
dependencies = [
|
||||
"num-bigint",
|
||||
"num-bigint 0.4.8",
|
||||
"num-traits",
|
||||
"thiserror 2.0.18",
|
||||
"time",
|
||||
|
||||
+2
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "aralez"
|
||||
version = "0.95.1"
|
||||
version = "0.95.2"
|
||||
edition = "2024"
|
||||
license = "Apache-2.0"
|
||||
description = "Reverse proxy built on top of Cloudflare's Pingora"
|
||||
@@ -72,3 +72,4 @@ reqwest = "0.12.28"
|
||||
hickory-resolver = { version = "0.26", features = ["tokio"] }
|
||||
aws-sdk-route53 = "1.127.0"
|
||||
aws-config = { version = "1.12.0", features = ["behavior-version-latest"] }
|
||||
redis = { version = "1.7.1", features = ["tokio-comp", "connection-manager"] }
|
||||
|
||||
@@ -61,29 +61,40 @@ Built on Rust, on top of **Cloudflare’s Pingora engine**, **Aralez** delivers
|
||||
### `main.yaml`
|
||||
|
||||
| Key | Example Value | Description |
|
||||
|----------------------------------|----------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------|
|
||||
| **threads** | 12 | Number of running daemon threads. Optional, defaults to 1 |
|
||||
| **runuser** | aralez | Optional. Username for running aralez after dropping root privileges (requires launch as root) |
|
||||
| **rungroup** | aralez | Optional. Group for running aralez after dropping root privileges (requires launch as root) |
|
||||
| **daemon** | false | Run in background (boolean) |
|
||||
| **upstream_keepalive_pool_size** | 500 | Pool size for upstream keepalive connections |
|
||||
| **pid_file** | /tmp/aralez.pid | Path to PID file |
|
||||
| **config_address** | 0.0.0.0:3000 | HTTP API address for pushing upstreams.yaml from remote location |
|
||||
| **proxy_tls_grade** | high, medium, unsafe | Grade of TLS ciphers. `high` matches Qualys SSL Labs A+ (defaults to `medium`) |
|
||||
| **proxy_address_http** | 0.0.0.0:6193 | Aralez HTTP bind address |
|
||||
| **proxy_address_tls** | 0.0.0.0:6194 | Aralez HTTPS bind address (Optional) |
|
||||
| **proxy_configs** | /etc/aralez/ | Direcotry containing configuration files, must be writeable by user `aralez` |
|
||||
| **upstreams_conf** | /etc/aralez/upstreams.yaml | Location of the upstreams file |
|
||||
| **access_log** | access | Configure access logging. Values: `access, error` |
|
||||
| **log_level** | info | Log level: `info`, `warn`, `error`, `debug`, `trace`, `off` |
|
||||
| **log_file** | /tmp/aralez.log, 20, 5, compress | Optional, the location of log file, file size in megabytes, number of log files to keep, compress files .valueas are separated by comma. Defaults to None, 100, 5, No |
|
||||
| **hc_method** | HEAD | Healthcheck method: HEAD, GET, POST (UPPERCASE) |
|
||||
| **hc_interval** | 2 | Interval for health checks in seconds |
|
||||
| **file_server_folder** | /some/local/folder | Optional. Local folder to serve |
|
||||
| **file_server_address** | 127.0.0.1:3002 | Optional. Local address for file server |
|
||||
| **config_api_enabled** | true | Enable/disable remote config push capability |
|
||||
| **cache_size_mb** | 50 | Optional. Enable internal cache with size in megabytes of value. Cache is entirely disabled when this line is commented out. |
|
||||
| **cache_ttl** | 10 | Optional. Set Default TTL in seconds for cached items, works if cache control headers are not set via upstream. Defaults to 60 if cache_size_mb is set. |
|
||||
|----------------------------------|-------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
|
||||
| **threads** | `12` | Number of running daemon threads. Optional, defaults to 1 |
|
||||
| **runuser** | `aralez` | Optional. Username for running aralez after dropping root privileges (requires launch as root) |
|
||||
| **rungroup** | `aralez` | Optional. Group for running aralez after dropping root privileges (requires launch as root) |
|
||||
| **daemon** | `false` | Run in background (boolean) |
|
||||
| **upstream_keepalive_pool_size** | `500` | Pool size for upstream keepalive connections |
|
||||
| **pid_file** | `/tmp/aralez.pid` | Path to PID file |
|
||||
| **upgrade_sock** | `/tmp/aralez.sock` | Path to socket file |
|
||||
| **config_address** | `0.0.0.0:3000` | HTTP API address for pushing upstreams.yaml from remote location |
|
||||
| **proxy_tls_grade** | `high, medium, unsafe` | Grade of TLS ciphers. `high` matches Qualys SSL Labs A+ (defaults to `medium`) |
|
||||
| **proxy_address_http** | `0.0.0.0:6193` | Aralez HTTP bind address |
|
||||
| **proxy_address_tls** | `0.0.0.0:6194` | Aralez HTTPS bind address (Optional) |
|
||||
| **proxy_configs** | `/etc/aralez/` | Direcotry containing configuration files, must be writeable by user `aralez` |
|
||||
| **upstreams_conf** | `/etc/aralez/upstreams.yaml` | Location of the upstreams file |
|
||||
| **access_log** | `access` | Configure access logging. Values: `access, error` |
|
||||
| **log_level** | `info` | Log level: `info`, `warn`, `error`, `debug`, `trace`, `off` |
|
||||
| **log_file** | `/tmp/aralez.log, 20, 5, compress` | Optional, the location of log file, file size in megabytes, number of log files to keep, compress files .valueas are separated by comma. Defaults to None, 100, 5, No |
|
||||
| **log_pattern** | `{d(%Y-%m-%d %H:%M:%S)} {l} - {m}{n}` | Optional log pattern, defaults to `{d(%Y-%m-%d %H:%M:%S)} {l} - {m}{n}`. See log4rs official [documentation](https://docs.rs/log4rs/latest/log4rs/encode/pattern/struct.PatternEncoderDeserializer.html#configuration) |
|
||||
| **hc_method** | `HEAD` | Healthcheck method: HEAD, GET, POST (UPPERCASE) |
|
||||
| **hc_interval** | `2` | Interval for health checks in seconds |
|
||||
| **file_server_folder** | `/some/local/folder` | Optional. Local folder to serve |
|
||||
| **file_server_address** | `127.0.0.1:3002` | Optional. Local address for file server |
|
||||
| **config_api_enabled** | `true` | Enable/disable remote config push capability |
|
||||
| **cache_size_mb** | `50` | Optional. Enable internal cache with size in megabytes of value. Cache is entirely disabled when this line is commented out. |
|
||||
| **cache_enabled** | `bool` | Optional. Disabled by default |
|
||||
| **cache_storage** | `memory, redis://host:port` | Optional Defaults to memory works with Redis, Valkey, Dragonfly (requires redis:// prefix) |
|
||||
| **cache_ttl** | `60` | Optional. Set internal cache with size in megabytes of value. Defaults to 50Mb if cache_enabled: true |
|
||||
| **tcp_keepalive_idle** | `60` | Seconds of inactivity before the kernel starts sending keepalive probes to a downstream client. Defaults to 60 |
|
||||
| **tcp_keepalive_interval** | `10` | Seconds between individual keepalive probes if the client does not respond. Defaults to 10 |
|
||||
| **tcp_keepalive_count** | `5` | Number of unanswered probes before the kernel declares the connection dead and closes it. Defaults to 5 |
|
||||
| **cluster_member** | `master, worker` | Optional. Enables cluster mode, Sets node type to master or worker. Disabled by default |
|
||||
| **worker_nodes** | `http(s)://host:port,http(s)://host:port` | Required is if cluster_member: master. Comma separated list of worker nodes http(s)://host:port |
|
||||
| **log_structured** | `elasticsearch` | Optional. Enables structured logging and sets backend type. Only elasticsearch et the moment. |
|
||||
| **acme_dns_provider** | `cloudflare, route53, digitalocean, etc` | Optional. Enable ACME DNS-01 challenge, sets DNS provider. ONly coudflare, route53, digitalocean at the moment. |
|
||||
|
||||
---
|
||||
|
||||
@@ -297,10 +308,27 @@ openssl req -x509 -newkey rsa:4096 \
|
||||
|
||||
## Remote Config API
|
||||
|
||||
Push new `upstreams.yaml` over HTTP to `config_address` (`:3000` by default). Useful for CI/CD automation or remote config updates. URL parameter. `key=MASTERKEY` is required. `MASTERKEY` is the value of `master_key` in the `main.yaml`
|
||||
Push new `upstreams.yaml` over HTTP to `config_address` (`127.0.0.1:3000` by default).
|
||||
Useful for CI/CD automation or remote config updates.
|
||||
|
||||
```bash
|
||||
curl -XPOST --data-binary @./etc/upstreams.txt 127.0.0.1:3000/conf?key=${MASTERKEY}
|
||||
curl -XPOST --data-binary @./etc/upstreams.txt 127.0.0.1:3000/conf
|
||||
```
|
||||
|
||||
Don't run API on address 0.0.0.0 on production servers, keep it on 127.0.0.1 and configure appropriate upstream with supported authentication and TLS termination
|
||||
|
||||
Example:
|
||||
|
||||
```yaml
|
||||
config.example.com:
|
||||
paths:
|
||||
"/conf":
|
||||
authorization:
|
||||
type: "basic"
|
||||
data: "user:pass"
|
||||
healthcheck: false
|
||||
servers:
|
||||
- "127.0.0.1:3000"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
+12
-6
@@ -4,25 +4,31 @@ runuser: aralez # Username for running aralez after dropping root privileges, re
|
||||
rungroup: aralez # Group for running aralez after dropping root privileges, requires program to start as root
|
||||
upstream_keepalive_pool_size: 500 # Pool size for upstream keepalive connections
|
||||
pid_file: /tmp/aralez.pid # Path to PID file
|
||||
#error_log: /tmp/aralez_err.log # Path to error log
|
||||
upgrade_sock: /tmp/aralez.sock # Path to socket file
|
||||
cache_enabled: true # Optional. Disabled by default
|
||||
cache_storage: memory # memory | redis://host:port . Optional Defaults to memory works with Redis, Valkey, Dragonfly (requires redis:// prefix)
|
||||
cache_size_mb: 50 # Optional. Set internal cache with size in megabytes of value. Defaults to 50Mb if cache_enabled: true
|
||||
cache_ttl: 60 # Optional. Set Default TTL in seconds for cached items, works if cache control headers are not set via upstream. Defaults to 60 if cache_enabled: true
|
||||
config_api_enabled: true # Boolean to enable/disable remote config push capability.
|
||||
config_address: 127.0.0.1:3000 # HTTP API address for pushing upstreams.yaml from remote location
|
||||
proxy_address_http: 0.0.0.0:80 # Proxy HTTP bind address
|
||||
proxy_address_tls: 0.0.0.0:443 # Optional, Proxy TLS bind address
|
||||
proxy_configs: /opt/aralez/asyncweb/etc # Mandatory if proxy_address_tls set, should contain a certificate and key files strictly in a format {NAME}.crt, {NAME}.key.
|
||||
proxy_configs: /opt/aralez/etc # Mandatory if proxy_address_tls set, should contain a certificate and key files strictly in a format {NAME}.crt, {NAME}.key.
|
||||
proxy_tls_grade: high # Grade of TLS suite for proxy (high, medium, unsafe), matching grades of Qualys SSL Labs
|
||||
upstreams_conf: /opt/aralez/etc/upstreams.yaml # the location of upstreams file
|
||||
file_server_folder: /opt/aralez/public # Optional, local folder to serve
|
||||
file_server_address: 0.0.0.0:3002 # Optional, Local address for file server. Can set as upstream for public access.
|
||||
file_server_address: 127.0.0.1:3002 # Optional, Local address for file server. Can set as upstream for public access.
|
||||
log_level: info # info, warn, error, debug, trace, off
|
||||
log_file: /tmp/aralez.log, 20, 5, compress # Optional, the location of log file, file size in megabytes, number of log files to keep, compress files .valueas are separated by comma.
|
||||
log_pattern: "{d(%Y-%m-%d %H:%M:%S)} {h({l})} {t} - {m} {n}" # Optional log pattern, defaults to "{d(%Y-%m-%d %H:%M:%S)} {l} - {m}{n}"
|
||||
access_log: error # all, error, (Off if commented)
|
||||
access_log: error # all, error, (Off by default)
|
||||
hc_method: HEAD # Healthcheck method (HEAD, GET, POST are supported) UPPERCASE
|
||||
hc_interval: 2 #Interval for health checks in seconds
|
||||
tcp_keepalive_idle: 60 # Seconds of inactivity before the kernel starts sending keepalive probes to a downstream client
|
||||
tcp_keepalive_interval: 10 # Seconds between individual keepalive probes if the client does not respond
|
||||
tcp_keepalive_count: 5 # Number of unanswered probes before the kernel declares the connection dead and closes it
|
||||
cache_size_mb: 50 # Optional. Enable internal cache with size in megabytes of value. Cache is entirely disabled when this line is commented out
|
||||
cache_ttl: 120 # Optional. Set Default TTL in seconds for cached items, works if cache control headers are not set via upstream. Defaults to 60 if cache_size_mb is set
|
||||
|
||||
#cluster_member: master | worker Optional. Enables cluster mode, Sets node type to master or worker. Disabled by default
|
||||
#worker_nodes: http://127.0.0.1:8000, http://127.0.0.1:4000 Required is if cluster_member: master. Comma separated list of worker nodes http(s)://host:port
|
||||
#log_structured: elasticsearch Optional. Enables structured logging and sets backend type. Only elasticsearch et the moment.
|
||||
#acme_dns_provider: cloudflare Optional. Enable ACME DNS-01 challenge, sets DNS provider. ONly coudflare, route53, digitalocean at the moment.
|
||||
|
||||
+1
-1
@@ -130,7 +130,7 @@ upstreams:
|
||||
"/":
|
||||
healthcheck: false
|
||||
servers:
|
||||
- "127.0.0.1:3000"
|
||||
- "127.0.0.1:8000"
|
||||
"/.well-known/acme-challenge":
|
||||
healthcheck: false
|
||||
servers:
|
||||
|
||||
Vendored
+2
@@ -0,0 +1,2 @@
|
||||
pub mod redis;
|
||||
pub mod types;
|
||||
Vendored
+222
@@ -0,0 +1,222 @@
|
||||
use async_trait::async_trait;
|
||||
use bytes::Bytes;
|
||||
use log::warn;
|
||||
use pingora::Result;
|
||||
use pingora_cache::storage::{HandleHit, HandleMiss, HitHandler, MissFinishType, MissHandler, Storage};
|
||||
use pingora_cache::trace::SpanHandle;
|
||||
use pingora_cache::{CacheKey, CacheMeta, PurgeOutcome, PurgeTarget, PurgeType};
|
||||
use redis::aio::ConnectionManager;
|
||||
use redis::{AsyncCommands, Client, RedisResult};
|
||||
use std::any::Any;
|
||||
use std::borrow::Cow;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct RedisCache {
|
||||
pub client: Arc<ConnectionManager>,
|
||||
pub ttl: u64,
|
||||
}
|
||||
|
||||
pub struct RedisHitHandler {
|
||||
body: Option<Bytes>,
|
||||
}
|
||||
|
||||
pub struct RedisMissHandler {
|
||||
client: Arc<ConnectionManager>,
|
||||
body_key: String,
|
||||
body_buffer: Vec<u8>,
|
||||
ttl: u64,
|
||||
}
|
||||
|
||||
impl RedisCache {
|
||||
pub async fn new(url: &str, ttl: u64) -> RedisResult<Self> {
|
||||
let client = Client::open(url)?;
|
||||
let manager = ConnectionManager::new(client).await?;
|
||||
Ok(Self { client: Arc::new(manager), ttl })
|
||||
}
|
||||
fn make_key(key: &CacheKey) -> Cow<'_, str> {
|
||||
match key.primary_key_str() {
|
||||
Some(s) => Cow::Borrowed(s),
|
||||
None => Cow::Owned(key.primary_key().iter().map(|b| format!("{:02x}", b)).collect()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl HandleHit for RedisHitHandler {
|
||||
async fn read_body(&mut self) -> Result<Option<Bytes>> {
|
||||
Ok(self.body.take())
|
||||
}
|
||||
|
||||
async fn finish(self: Box<Self>, _storage: &'static (dyn Storage + Sync), _key: &CacheKey, _trace: &SpanHandle) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn can_seek(&self) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
fn seek(&mut self, _start: usize, _end: Option<usize>) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn should_count_access(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn get_eviction_weight(&self) -> usize {
|
||||
self.body.as_ref().map(|b| b.len()).unwrap_or(0)
|
||||
}
|
||||
|
||||
fn entry_id(&self) -> Option<pingora_cache::eviction::CacheEntryId> {
|
||||
None
|
||||
}
|
||||
|
||||
fn as_any(&self) -> &(dyn Any + Send + Sync) {
|
||||
self
|
||||
}
|
||||
|
||||
fn as_any_mut(&mut self) -> &mut (dyn Any + Send + Sync) {
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl HandleMiss for RedisMissHandler {
|
||||
async fn write_body(&mut self, data: Bytes, _eof: bool) -> Result<()> {
|
||||
self.body_buffer.extend_from_slice(&data);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn finish(self: Box<Self>) -> Result<MissFinishType> {
|
||||
let len = self.body_buffer.len();
|
||||
let mut conn = (*self.client).clone();
|
||||
if let Err(e) = conn.set_ex::<_, _, ()>(&self.body_key, &self.body_buffer, self.ttl).await {
|
||||
warn!("Failed to write cache entry to Redis: {e}");
|
||||
}
|
||||
Ok(MissFinishType::Appended(len, None))
|
||||
}
|
||||
|
||||
fn entry_id(&self) -> Option<pingora_cache::eviction::CacheEntryId> {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Storage for RedisCache {
|
||||
async fn lookup(&'static self, key: &CacheKey, _trace: &SpanHandle) -> Result<Option<(CacheMeta, HitHandler)>> {
|
||||
let base = Self::make_key(key);
|
||||
|
||||
// Pre-allocate exact capacities to prevent reallocations
|
||||
let mut meta_key = String::with_capacity(base.len() + 5);
|
||||
meta_key.push_str(&base);
|
||||
meta_key.push_str(":meta");
|
||||
|
||||
let mut body_key = String::with_capacity(base.len() + 5);
|
||||
body_key.push_str(&base);
|
||||
body_key.push_str(":body");
|
||||
|
||||
let mut internal_key = String::with_capacity(meta_key.len() + 9);
|
||||
internal_key.push_str(&meta_key);
|
||||
internal_key.push_str(":internal");
|
||||
|
||||
let mut header_key = String::with_capacity(meta_key.len() + 7);
|
||||
header_key.push_str(&meta_key);
|
||||
header_key.push_str(":header");
|
||||
|
||||
let mut conn = (*self.client).clone();
|
||||
|
||||
let internal: Option<Vec<u8>> = conn.get(&internal_key).await.ok().flatten();
|
||||
let header: Option<Vec<u8>> = conn.get(&header_key).await.ok().flatten();
|
||||
let body: Option<Vec<u8>> = conn.get(&body_key).await.ok().flatten();
|
||||
|
||||
match (internal, header, body) {
|
||||
(Some(internal_bytes), Some(header_bytes), Some(body_bytes)) => match CacheMeta::deserialize(&internal_bytes, &header_bytes) {
|
||||
Ok(meta) => {
|
||||
let handler: HitHandler = Box::new(RedisHitHandler {
|
||||
body: Some(Bytes::from(body_bytes)),
|
||||
});
|
||||
Ok(Some((meta, handler)))
|
||||
}
|
||||
Err(_) => {
|
||||
let _: RedisResult<()> = conn.del(&[&internal_key, &header_key, &body_key]).await;
|
||||
Ok(None)
|
||||
}
|
||||
},
|
||||
_ => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_miss_handler(&'static self, key: &CacheKey, meta: &CacheMeta, _trace: &SpanHandle) -> Result<MissHandler> {
|
||||
let base = Self::make_key(key);
|
||||
|
||||
let mut meta_key = String::with_capacity(base.len() + 5);
|
||||
meta_key.push_str(&base);
|
||||
meta_key.push_str(":meta");
|
||||
|
||||
let mut body_key = String::with_capacity(base.len() + 5);
|
||||
body_key.push_str(&base);
|
||||
body_key.push_str(":body");
|
||||
|
||||
let mut internal_key = String::with_capacity(meta_key.len() + 9);
|
||||
internal_key.push_str(&meta_key);
|
||||
internal_key.push_str(":internal");
|
||||
|
||||
let mut header_key = String::with_capacity(meta_key.len() + 7);
|
||||
header_key.push_str(&meta_key);
|
||||
header_key.push_str(":header");
|
||||
|
||||
let ttl = self.ttl;
|
||||
let (internal_bytes, header_bytes) = meta.serialize()?;
|
||||
let mut conn = (*self.client).clone();
|
||||
|
||||
let _: RedisResult<()> = conn.set_ex(&internal_key, &internal_bytes, ttl).await;
|
||||
let _: RedisResult<()> = conn.set_ex(&header_key, &header_bytes, ttl).await;
|
||||
|
||||
Ok(Box::new(RedisMissHandler {
|
||||
client: self.client.clone(),
|
||||
body_key,
|
||||
body_buffer: Vec::with_capacity(4096),
|
||||
ttl,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn purge(&'static self, _target: PurgeTarget<'_>, _purge_type: PurgeType, _trace: &SpanHandle) -> Result<PurgeOutcome> {
|
||||
Ok(PurgeOutcome::NotFound)
|
||||
}
|
||||
|
||||
async fn update_meta(&'static self, key: &CacheKey, meta: &CacheMeta, _trace: &SpanHandle) -> Result<bool> {
|
||||
let base = Self::make_key(key);
|
||||
|
||||
let mut meta_key = String::with_capacity(base.len() + 5);
|
||||
meta_key.push_str(&base);
|
||||
meta_key.push_str(":meta");
|
||||
|
||||
let mut internal_key = String::with_capacity(meta_key.len() + 9);
|
||||
internal_key.push_str(&meta_key);
|
||||
internal_key.push_str(":internal");
|
||||
|
||||
let mut header_key = String::with_capacity(meta_key.len() + 7);
|
||||
header_key.push_str(&meta_key);
|
||||
header_key.push_str(":header");
|
||||
|
||||
let ttl = self.ttl;
|
||||
match meta.serialize() {
|
||||
Ok((internal_bytes, header_bytes)) => {
|
||||
let mut conn = (*self.client).clone();
|
||||
let _: RedisResult<()> = conn.set_ex(&internal_key, &internal_bytes, ttl).await;
|
||||
let _: RedisResult<()> = conn.set_ex(&header_key, &header_bytes, ttl).await;
|
||||
Ok(true)
|
||||
}
|
||||
Err(_) => Ok(false),
|
||||
}
|
||||
}
|
||||
|
||||
fn support_streaming_partial_write(&self) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
fn as_any(&self) -> &(dyn Any + Send + Sync) {
|
||||
self
|
||||
}
|
||||
}
|
||||
Vendored
+41
@@ -0,0 +1,41 @@
|
||||
use crate::cache::redis::RedisCache;
|
||||
use crate::utils::lazylock::{CACHE_TTL, EVICTION};
|
||||
use crate::utils::types::AppConfig;
|
||||
use log::{info, warn};
|
||||
use pingora_cache::eviction::simple_lru::Manager;
|
||||
use pingora_cache::{MemCache, Storage};
|
||||
use std::sync::{Arc, OnceLock};
|
||||
|
||||
pub static GLOBAL_STORAGE: OnceLock<Box<dyn Storage + Send + Sync>> = OnceLock::new();
|
||||
|
||||
pub async fn init_cache(config: Arc<AppConfig>) {
|
||||
let cache_storage = config.cache_storage.clone().unwrap_or_else(|| {
|
||||
info!("No cache_storage configured, defaulting to memory");
|
||||
"memory".to_string()
|
||||
});
|
||||
let cache_type = cache_storage.splitn(2, ':').next().unwrap_or("memory");
|
||||
let ttl = config.cache_ttl.unwrap_or(60);
|
||||
|
||||
match cache_type {
|
||||
"memory" => {
|
||||
CACHE_TTL.set(ttl).ok();
|
||||
if let Some(size) = config.cache_size_mb {
|
||||
let cache_size = size * 1024 * 1024;
|
||||
drop(EVICTION.set(Manager::new(cache_size)));
|
||||
}
|
||||
GLOBAL_STORAGE.set(Box::new(MemCache::new())).ok();
|
||||
info!("Cache initialized successfully. Storage: {}", cache_storage);
|
||||
}
|
||||
"redis" => match RedisCache::new(&cache_storage, ttl).await {
|
||||
Ok(cache) => {
|
||||
CACHE_TTL.set(ttl).ok();
|
||||
GLOBAL_STORAGE.set(Box::new(cache)).ok();
|
||||
info!("Cache initialized successfully. Storage: {}", cache_storage);
|
||||
}
|
||||
Err(e) => log::error!("Cache storage initialization failed: {}", e),
|
||||
},
|
||||
_ => {
|
||||
warn!("Unsupported storage, disabling caching: {}", cache_storage);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
use crate::cache::types::init_cache;
|
||||
use crate::core::proxyhttp::LB;
|
||||
use crate::ingress::{APIUpstreamProvider, ConsulProvider, Discovery, FromFileProvider, KubernetesProvider};
|
||||
use crate::logging::core::{init_access_log, init_access_logging, log_builder};
|
||||
@@ -29,16 +30,15 @@ impl BackgroundService for LB {
|
||||
let config = load_configuration(self.config.upstreams_conf.clone().as_str(), "filepath")
|
||||
.await
|
||||
.0
|
||||
.expect("Failed to load configuration");
|
||||
.die("Failed to load configuration");
|
||||
|
||||
match config.typecfg.as_str() {
|
||||
"file" => {
|
||||
info!("Running File discovery, requested type is: {}", config.typecfg);
|
||||
tx.send(config).await.unwrap();
|
||||
tx.send(config).await.die("Running File discovery");
|
||||
let file_load = FromFileProvider {
|
||||
path: self.config.upstreams_conf.clone(),
|
||||
};
|
||||
// let _ = tokio::spawn(async move { file_load.start(tx).await });
|
||||
drop(tokio::spawn(async move { file_load.start(tx).await }));
|
||||
}
|
||||
"kubernetes" => {
|
||||
@@ -79,11 +79,12 @@ impl BackgroundService for LB {
|
||||
let ff = self.ump_full.clone();
|
||||
let im = self.ump_byid.clone();
|
||||
let (hc_method, hc_interval) = (self.config.hc_method.clone(), self.config.hc_interval);
|
||||
tokio::spawn(async move { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await });
|
||||
tokio::spawn(async move { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().die("Unable to spawn HC worker"))).await });
|
||||
tokio::spawn(async move { calc_cache_metrics().await });
|
||||
let acme_provider = self.config.acme_dns_provider.clone();
|
||||
tokio::spawn(async move { refresh_order(certdir, confdir, acme_provider).await });
|
||||
|
||||
init_cache(self.config.clone()).await;
|
||||
init_access_logging(self.config.access_log.clone()).await;
|
||||
loop {
|
||||
tokio::select! {
|
||||
@@ -91,7 +92,7 @@ impl BackgroundService for LB {
|
||||
break;
|
||||
}
|
||||
val = rx.recv() => {
|
||||
let start = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
|
||||
let start = SystemTime::now().duration_since(UNIX_EPOCH).die(format!("ON {} {}", file!(), line!()).as_str());
|
||||
if let Some(ss) = val {
|
||||
clone_dashmap_into(&ss.upstreams, &self.ump_full);
|
||||
clone_dashmap_into(&ss.upstreams, &self.ump_upst);
|
||||
@@ -138,7 +139,7 @@ impl BackgroundService for LB {
|
||||
}
|
||||
}
|
||||
}
|
||||
let end = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
|
||||
let end = SystemTime::now().duration_since(UNIX_EPOCH).die(format!("ON {} {}", file!(), line!()).as_str());
|
||||
let startup = end - start;
|
||||
info!("Loading upstreams took: {}ms", startup.as_millis());
|
||||
}
|
||||
|
||||
+9
-10
@@ -1,7 +1,8 @@
|
||||
use crate::auth::authenticate;
|
||||
use crate::cache::types::GLOBAL_STORAGE;
|
||||
use crate::core::gethosts::{GetHost, GetHostsReturHeaders};
|
||||
use crate::logging::core::access_log;
|
||||
use crate::utils::lazylock::{CACHE_LOCK, CACHE_TTL, EVICTION, LOCALHOST, MEM_CACHE, RATE_LIMITER, REQUESTS_4XX, REVERSE_STORE};
|
||||
use crate::utils::lazylock::{CACHE_LOCK, CACHE_TTL, EVICTION, LOCALHOST, RATE_LIMITER, REQUESTS_4XX, REVERSE_STORE};
|
||||
use crate::utils::metrics::*;
|
||||
use crate::utils::types::{AppConfig, Extraparams, Headers, InnerMap, UpstreamsDashMap, UpstreamsIdMap};
|
||||
use arc_swap::ArcSwap;
|
||||
@@ -67,8 +68,7 @@ impl ProxyHttp for LB {
|
||||
|
||||
fn cache_key_callback(&self, session: &Session, ctx: &mut Self::CTX) -> Result<CacheKey> {
|
||||
let host = ctx.hostname.as_deref().unwrap_or("default");
|
||||
let path = session.req_header().uri.path();
|
||||
|
||||
let path = session.req_header().uri.path_and_query().map_or("/", |pq| pq.as_str());
|
||||
let primary_key = KEY_BUF.with(|buf| {
|
||||
let mut b = buf.borrow_mut();
|
||||
b.clear();
|
||||
@@ -76,15 +76,14 @@ impl ProxyHttp for LB {
|
||||
b.push_str(path);
|
||||
b.clone()
|
||||
});
|
||||
|
||||
let secondary_key = session.req_header().uri.path_and_query().map(|pq| pq.as_str()).unwrap_or("/");
|
||||
Ok(CacheKey::new(primary_key, secondary_key))
|
||||
Ok(CacheKey::new(primary_key, host /* host is just for debugging */))
|
||||
}
|
||||
|
||||
fn request_cache_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<()> {
|
||||
if self.cache_enabled && session.req_header().method == Method::GET {
|
||||
if let Some(eviction) = EVICTION.get() {
|
||||
session.cache.enable(&*MEM_CACHE, Some(eviction), None, Some(&*CACHE_LOCK), None);
|
||||
let has_auth = _ctx.upstream_peer.as_ref().map(|m| m.authorization.is_some()).unwrap_or(false) || _ctx.extraparams.authentication.is_some();
|
||||
if self.cache_enabled && session.req_header().method == Method::GET && !has_auth {
|
||||
if let Some(storage) = GLOBAL_STORAGE.get() {
|
||||
let eviction_mgr = EVICTION.get().map(|e| e as &(dyn pingora_cache::eviction::EvictionManager + Sync));
|
||||
session.cache.enable(storage.as_ref(), eviction_mgr, None, Some(&*CACHE_LOCK), None);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
|
||||
+20
-16
@@ -29,20 +29,24 @@ use std::time::Duration;
|
||||
use std::{fs, thread};
|
||||
|
||||
pub fn run() {
|
||||
default_provider().install_default().expect("Failed to install RusTLS crypto provider");
|
||||
default_provider().install_default().die("Failed to install RusTLS crypto provider:");
|
||||
let parameters = Opt::parse_args();
|
||||
let file = parameters.conf.clone().unwrap();
|
||||
let file = parameters.conf.clone().die("Failed to clone the config:");
|
||||
let maincfg = crate::utils::parceyaml::parce_main_config(file.as_str());
|
||||
|
||||
let master_server = maincfg.cluster_member.as_deref() == Some("master");
|
||||
let cache_enabled = maincfg.cache_size_mb.is_some();
|
||||
CACHE_TTL.set(maincfg.cache_ttl.unwrap_or(60)).unwrap();
|
||||
if let Some(size) = maincfg.cache_size_mb {
|
||||
let cache_size = size * 1024 * 1024;
|
||||
drop(EVICTION.set(Manager::new(cache_size)));
|
||||
}
|
||||
|
||||
let mut server = Server::new(parameters).unwrap();
|
||||
let cache_enabled = maincfg.cache_enabled.unwrap_or(false);
|
||||
|
||||
// ========================================================================== //
|
||||
CACHE_TTL.set(maincfg.cache_ttl.unwrap_or(60)).die("Failed to set cache TTL");
|
||||
if cache_enabled {
|
||||
let cache_size = maincfg.cache_size_mb.unwrap_or(50);
|
||||
let _ = EVICTION.set(Manager::new(cache_size * 1024 * 1024));
|
||||
}
|
||||
// ========================================================================== //
|
||||
|
||||
let mut server = Server::new(parameters).die("Failed to start the server:");
|
||||
server.bootstrap();
|
||||
|
||||
let uf_config = Arc::new(DashMap::new());
|
||||
@@ -106,25 +110,25 @@ pub fn run() {
|
||||
if let Some(bind_address_tls) = bind_address_tls {
|
||||
check_priv(bind_address_tls.as_str());
|
||||
let (tx, rx): (Sender<Vec<CertificateConfig>>, Receiver<Vec<CertificateConfig>>) = channel();
|
||||
let certs_path = cfg.proxy_configs.clone().unwrap() + "/certificates";
|
||||
let certs_path = cfg.proxy_configs.clone().die("Failed certificates path") + "/certificates";
|
||||
|
||||
if fs::metadata(certs_path.clone()).is_err() {
|
||||
fs::create_dir_all(certs_path.clone()).unwrap();
|
||||
fs::create_dir_all(certs_path.clone()).die("Failed to create directories");
|
||||
}
|
||||
thread::spawn(move || {
|
||||
folder_watch(certs_path, tx).unwrap();
|
||||
folder_watch(certs_path, tx).die("Failed to spawn certificates channel");
|
||||
});
|
||||
let certificate_configs = rx.recv().unwrap();
|
||||
let certificate_configs = rx.recv().die("Failed to spawn certificates channel");
|
||||
let first_set = load::Certificates::new(&certificate_configs, grade.as_str()).unwrap_or_else(|| panic!("Unable to load initial certificate info"));
|
||||
let certificates = Arc::new(ArcSwap::from_pointee(first_set));
|
||||
let certs_for_callback = certificates.clone();
|
||||
|
||||
let certs_for_watcher = certificates.clone();
|
||||
let new_certs = load::Certificates::new(&certificate_configs, grade.as_str());
|
||||
certs_for_watcher.store(Arc::new(new_certs.unwrap()));
|
||||
certs_for_watcher.store(Arc::new(new_certs.die("Failed to build the watcher")));
|
||||
|
||||
let mut tls_settings =
|
||||
TlsSettings::intermediate(&certs_for_callback.load().default_cert_path, &certs_for_callback.load().default_key_path).expect("unable to load or parse cert/key");
|
||||
TlsSettings::intermediate(&certs_for_callback.load().default_cert_path, &certs_for_callback.load().default_key_path).die("unable to load or parse cert/key");
|
||||
|
||||
grades::set_tsl_grade(&mut tls_settings, grade.as_str());
|
||||
tls_settings.set_servername_callback(move |ssl_ref: &mut SslRef, ssl_alert: &mut SslAlert| certs_for_callback.load().server_name_callback(ssl_ref, ssl_alert));
|
||||
@@ -162,7 +166,7 @@ pub fn run() {
|
||||
if let Err(e) = write_pid_file(pf.as_str()) {
|
||||
panic!("Failed to write PID file: {} : {}", pf, e);
|
||||
}
|
||||
let mut signals = Signals::new([SIGINT, SIGTERM, SIGQUIT]).unwrap();
|
||||
let mut signals = Signals::new([SIGINT, SIGTERM, SIGQUIT]).die("Failed to create signals");
|
||||
if let Some(sig) = signals.forever().next() {
|
||||
match sig {
|
||||
SIGINT => info!("SIGINT received! Exiting..."),
|
||||
|
||||
+47
-57
@@ -3,10 +3,12 @@ use crate::ingress::APIUpstreamProvider;
|
||||
use crate::tls::acme::http01::{acme_create, acme_order, http01_challenge};
|
||||
use crate::utils::lazylock::NODE_STATUS;
|
||||
use crate::utils::metrics::{MEMORY_USAGE, OPEN_FILES, get_memory_usage, get_open_files};
|
||||
use crate::utils::tools::{upstreams_liveness_json, upstreams_to_json};
|
||||
use crate::utils::tools::{Die, upstreams_liveness_json, upstreams_to_json};
|
||||
use crate::utils::types::{Config, Configuration, UpstreamsDashMap};
|
||||
use axum::body::Body;
|
||||
use axum::extract::ConnectInfo;
|
||||
use axum::extract::{Query, State};
|
||||
use axum::http::HeaderMap;
|
||||
use axum::http::{Response, StatusCode};
|
||||
use axum::response::IntoResponse;
|
||||
use axum::routing::{any, get, post};
|
||||
@@ -86,14 +88,16 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
|
||||
|
||||
let listener = port_is_available("Config API", &config.address).await;
|
||||
info!("Starting the API server on: {}", config.address);
|
||||
// let api_server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
|
||||
let api_server = tokio::spawn(async move { axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>()).await.unwrap() });
|
||||
|
||||
let api_server = tokio::spawn(async move {
|
||||
axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
|
||||
.await
|
||||
.die("Can't start API server")
|
||||
});
|
||||
let (tx, mut rx) = mpsc::channel(1);
|
||||
std::thread::spawn(move || {
|
||||
let mut signals = Signals::new([SIGQUIT]).unwrap();
|
||||
let mut signals = Signals::new([SIGQUIT]).die("Can't initialize OS SIGNALS");
|
||||
if let Some(sig) = signals.forever().next() {
|
||||
tx.blocking_send(sig).unwrap();
|
||||
tx.blocking_send(sig).die("Can;t send signal to main thread");
|
||||
}
|
||||
});
|
||||
rx.recv().await;
|
||||
@@ -104,39 +108,50 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
|
||||
info!("Exiting...");
|
||||
}
|
||||
|
||||
async fn conf(State(st): State<AppState>, Query(params): Query<HashMap<String, String>>, content: String) -> impl IntoResponse {
|
||||
async fn conf(
|
||||
State(st): State<AppState>,
|
||||
ConnectInfo(addr): ConnectInfo<SocketAddr>,
|
||||
Query(params): Query<HashMap<String, String>>,
|
||||
headers: HeaderMap,
|
||||
content: String,
|
||||
) -> impl IntoResponse {
|
||||
if !st.config_api_enabled {
|
||||
return Response::builder().status(StatusCode::FORBIDDEN).body(Body::from("Config API is disabled !\n")).unwrap();
|
||||
return (StatusCode::FORBIDDEN, "Config API is disabled !\n").into_response();
|
||||
}
|
||||
|
||||
let api_key = headers.get("API-KEY").and_then(|v| v.to_str().ok());
|
||||
let query_key = params.get("key").map(|k| k.as_str());
|
||||
let authorized = match st.master_key.as_deref() {
|
||||
Some(master_key) if !master_key.is_empty() => api_key == Some(master_key) || query_key == Some(master_key),
|
||||
_ => false,
|
||||
};
|
||||
if !authorized {
|
||||
warn!("Unauthorised access to /conf endpoint from {}", addr.ip());
|
||||
return (StatusCode::UNAUTHORIZED, "Unauthorized: Invalid API-KEY\n").into_response();
|
||||
}
|
||||
|
||||
let strcontent = content.as_str();
|
||||
let parsed = noyalib::from_str::<Config>(strcontent);
|
||||
match parsed {
|
||||
Ok(_) => {
|
||||
if params.contains_key("save") {
|
||||
tokio::spawn(async move { apply_config(content.as_str(), st, true).await });
|
||||
} else {
|
||||
tokio::spawn(async move { apply_config(content.as_str(), st, false).await });
|
||||
}
|
||||
Response::builder().status(StatusCode::OK).body(Body::from("Accepted! Applying in background\n")).unwrap()
|
||||
let save = params.contains_key("save");
|
||||
tokio::spawn(async move {
|
||||
apply_config(&content, st, save).await;
|
||||
});
|
||||
(StatusCode::OK, "Accepted! Applying in background\n").into_response()
|
||||
}
|
||||
Err(err) => {
|
||||
error!("Failed to parse upstreams file: {}", err);
|
||||
Response::builder().status(StatusCode::BAD_GATEWAY).body(Body::from(format!("Failed: {}\n", err))).unwrap()
|
||||
(StatusCode::BAD_REQUEST, format!("Failed: {}\n", err)).into_response()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
use axum::extract::ConnectInfo;
|
||||
use axum::http::HeaderMap;
|
||||
async fn workerconf(State(st): State<AppState>, ConnectInfo(addr): ConnectInfo<SocketAddr>, headers: HeaderMap, content: String) -> impl IntoResponse {
|
||||
let api_key = headers.get("API-KEY").and_then(|v| v.to_str().ok());
|
||||
if api_key != st.master_key.as_deref() {
|
||||
warn!("Unauthorised access to workerconf endpoint from {}", addr.ip());
|
||||
return Response::builder()
|
||||
.status(StatusCode::UNAUTHORIZED)
|
||||
.body(Body::from("Unauthorized: Invalid API-KEY\n"))
|
||||
.unwrap();
|
||||
return (StatusCode::UNAUTHORIZED, "Unauthorized: Invalid API-KEY\n").into_response();
|
||||
}
|
||||
if let Some(member) = NODE_STATUS.get() {
|
||||
if let Some(mbr) = member.as_ref() {
|
||||
@@ -148,31 +163,22 @@ async fn workerconf(State(st): State<AppState>, ConnectInfo(addr): ConnectInfo<S
|
||||
match parsed {
|
||||
Ok(_) => {
|
||||
tokio::spawn(async move { apply_config(content.as_str(), st, true).await });
|
||||
return Response::builder().status(StatusCode::OK).body(Body::from("Accepted! Applying in background\n")).unwrap();
|
||||
return (StatusCode::OK, "Accepted! Applying in background\n").into_response();
|
||||
}
|
||||
Err(err) => {
|
||||
error!("Failed to parse upstreams file: {}", err);
|
||||
return Response::builder().status(StatusCode::BAD_GATEWAY).body(Body::from(format!("Failed: {}\n", err))).unwrap();
|
||||
return (StatusCode::BAD_GATEWAY, format!("Failed: {}\n", err)).into_response();
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
return Response::builder()
|
||||
.status(StatusCode::BAD_GATEWAY)
|
||||
.body(Body::from("Node is not running on worker mode !\n"))
|
||||
.unwrap();
|
||||
return (StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response();
|
||||
}
|
||||
}
|
||||
}
|
||||
return Response::builder()
|
||||
.status(StatusCode::BAD_GATEWAY)
|
||||
.body(Body::from("Node is not running on worker mode !\n"))
|
||||
.unwrap();
|
||||
return (StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response();
|
||||
}
|
||||
Response::builder()
|
||||
.status(StatusCode::BAD_GATEWAY)
|
||||
.body(Body::from("Node is not running on worker mode !\n"))
|
||||
.unwrap()
|
||||
(StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response()
|
||||
}
|
||||
|
||||
async fn apply_config(content: &str, st: AppState, save: bool) {
|
||||
@@ -236,17 +242,13 @@ async fn metrics() -> impl IntoResponse {
|
||||
let encoder = TextEncoder::new();
|
||||
let mut buffer = Vec::new();
|
||||
if let Err(e) = encoder.encode(&metric_families, &mut buffer) {
|
||||
// encoding error fallback
|
||||
return Response::builder()
|
||||
.status(StatusCode::INTERNAL_SERVER_ERROR)
|
||||
.body(Body::from(format!("Failed to encode metrics: {}", e)))
|
||||
.unwrap();
|
||||
return (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to encode metrics: {}", e)).into_response();
|
||||
}
|
||||
Response::builder()
|
||||
.status(StatusCode::OK)
|
||||
.header("Content-Type", encoder.format_type())
|
||||
.body(Body::from(buffer))
|
||||
.unwrap()
|
||||
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response())
|
||||
}
|
||||
|
||||
#[allow(clippy::needless_return)]
|
||||
@@ -257,7 +259,7 @@ async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String,
|
||||
.status(StatusCode::OK)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(Body::from(format!("{}", r)))
|
||||
.unwrap();
|
||||
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
|
||||
}
|
||||
if params.contains_key("all") {
|
||||
let resp = upstreams_to_json(&st.current_upstreams);
|
||||
@@ -267,29 +269,17 @@ async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String,
|
||||
.status(StatusCode::OK)
|
||||
.header("Content-Type", "application/json")
|
||||
.body(Body::from(j))
|
||||
.unwrap();
|
||||
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
|
||||
}
|
||||
Err(e) => {
|
||||
return Response::builder()
|
||||
.status(StatusCode::INTERNAL_SERVER_ERROR)
|
||||
.body(Body::from(format!("Failed to get status: {}", e)))
|
||||
.unwrap();
|
||||
Err(e) => return (StatusCode::BAD_GATEWAY, format!("Internal Server Error {}\n", e)).into_response(),
|
||||
}
|
||||
}
|
||||
}
|
||||
Response::builder()
|
||||
.status(StatusCode::INTERNAL_SERVER_ERROR)
|
||||
.body(Body::from("Parameter mismatch"))
|
||||
.unwrap()
|
||||
(StatusCode::INTERNAL_SERVER_ERROR, "Parameter mismatch\n").into_response()
|
||||
}
|
||||
|
||||
pub async fn port_is_available(name: &str, address: &str) -> TcpListener {
|
||||
let addr = SocketAddr::from_str(address).unwrap_or_else(|e| panic!("{}: Invalid address format: {:?}", name, e));
|
||||
let t = Duration::from_secs(2);
|
||||
|
||||
//if addr.ip() == IpAddr::V4(Ipv4Addr::UNSPECIFIED) {
|
||||
// addr.set_ip(IpAddr::V4(Ipv4Addr::LOCALHOST));
|
||||
//}
|
||||
let p = addr.port();
|
||||
loop {
|
||||
match TcpListener::bind(addr).await {
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
use crate::ingress::kuberconsul;
|
||||
use crate::ingress::kuberconsul::{list_to_upstreams, ServiceDiscovery};
|
||||
use crate::ingress::kuberconsul::{ServiceDiscovery, list_to_upstreams};
|
||||
use crate::utils::httpclient;
|
||||
use crate::utils::parceyaml::build_headers;
|
||||
use crate::utils::tools::Die;
|
||||
use crate::utils::types::{Configuration, GlobalServiceMapping, UpstreamsDashMap};
|
||||
use async_trait::async_trait;
|
||||
use dashmap::DashMap;
|
||||
@@ -38,7 +39,7 @@ impl ServiceDiscovery for ConsulDiscovery {
|
||||
});
|
||||
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 consul_data = servers.get(num).die("Reading consul data").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);
|
||||
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams};
|
||||
use crate::utils::tools::{Die, clone_dashmap_into, compare_dashmaps, print_upstreams};
|
||||
use crate::utils::types::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap};
|
||||
use async_trait::async_trait;
|
||||
use dashmap::DashMap;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use tokio::fs::File;
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::sync::mpsc::Sender;
|
||||
@@ -31,9 +31,9 @@ pub fn match_path(conf: &GlobalServiceMapping, upstreams: &DashMap<Arc<str>, (Ve
|
||||
}
|
||||
|
||||
pub async fn read_token(path: &str) -> String {
|
||||
let mut file = File::open(path).await.unwrap();
|
||||
let mut file = File::open(path).await.die("Reading kubernetes token");
|
||||
let mut contents = String::new();
|
||||
file.read_to_string(&mut contents).await.unwrap();
|
||||
file.read_to_string(&mut contents).await.die("Kubernetes token to string");
|
||||
contents.trim().to_string()
|
||||
}
|
||||
|
||||
|
||||
@@ -9,10 +9,11 @@ use std::env;
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::ingress::kuberconsul;
|
||||
use crate::ingress::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery};
|
||||
use crate::ingress::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate};
|
||||
use rustls::crypto::ring::default_provider;
|
||||
use crate::ingress::kuberconsul::{ServiceDiscovery, clone_compare, list_to_upstreams};
|
||||
use crate::ingress::kubewatcher::{AralezRouteUpdate, start_ingress_watcher_with_config};
|
||||
use crate::utils::tools::Die;
|
||||
use rustls::crypto::CryptoProvider;
|
||||
use rustls::crypto::ring::default_provider;
|
||||
use tokio::sync::mpsc::Sender;
|
||||
|
||||
#[derive(Debug, serde::Deserialize)]
|
||||
@@ -66,12 +67,12 @@ impl ServiceDiscovery for KubernetesDiscovery {
|
||||
|
||||
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 server = servers.get(num).die("Getting k8s api server").to_string();
|
||||
let path = kuber.tokenpath.unwrap_or_else(|| "/var/run/secrets/kubernetes.io/serviceaccount/token".to_string());
|
||||
let token = kuberconsul::read_token(path.as_str()).await;
|
||||
|
||||
if CryptoProvider::get_default().is_none() {
|
||||
default_provider().install_default().expect("Failed to install rustls crypto provider");
|
||||
default_provider().install_default().die("Failed to install rustls crypto provider");
|
||||
}
|
||||
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel::<AralezRouteUpdate>(100);
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
pub mod auth;
|
||||
pub mod cache;
|
||||
pub mod core;
|
||||
pub mod ingress;
|
||||
pub mod logging;
|
||||
|
||||
+9
-8
@@ -1,13 +1,14 @@
|
||||
use crate::logging::types::start_logging_backend;
|
||||
use crate::utils::metrics::LOGGING_ERRORS;
|
||||
use crate::utils::tools::Die;
|
||||
use crate::utils::types::AppConfig;
|
||||
use anyhow::Result;
|
||||
use log::{error, info, log, Level, LevelFilter, Record};
|
||||
use log::{Level, LevelFilter, Record, error, info, log};
|
||||
use log4rs::append::Append;
|
||||
use log4rs::append::rolling_file::RollingFileAppender;
|
||||
use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
|
||||
use log4rs::append::rolling_file::policy::compound::roll::fixed_window::FixedWindowRoller;
|
||||
use log4rs::append::rolling_file::policy::compound::trigger::size::SizeTrigger;
|
||||
use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
|
||||
use log4rs::append::rolling_file::RollingFileAppender;
|
||||
use log4rs::append::Append;
|
||||
use log4rs::config::Logger;
|
||||
use log4rs::{
|
||||
append::console::ConsoleAppender,
|
||||
@@ -149,7 +150,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
|
||||
.logger(Logger::builder().build("h2", LevelFilter::Off));
|
||||
}
|
||||
|
||||
let config = config_builder.build(Root::builder().appender("stdout").build(log_level)).unwrap();
|
||||
let config = config_builder.build(Root::builder().appender("stdout").build(log_level)).die("Logging config builder");
|
||||
|
||||
if let Err(e) = log4rs::init_config(config) {
|
||||
eprintln!("Warning: log4rs logger already initialized: {:?}", e);
|
||||
@@ -177,7 +178,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
|
||||
_ => format!("{}.{{}}", path),
|
||||
};
|
||||
|
||||
let roller = FixedWindowRoller::builder().build(pattern_str.as_str(), keep).unwrap();
|
||||
let roller = FixedWindowRoller::builder().build(pattern_str.as_str(), keep).die("Logging roller builder");
|
||||
let trigger = SizeTrigger::new(size_mb * 1024 * 1024);
|
||||
let policy = CompoundPolicy::new(Box::new(trigger), Box::new(roller));
|
||||
|
||||
@@ -193,7 +194,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
|
||||
.appender(Appender::builder().build("file", Box::new(file)))
|
||||
.logger(Logger::builder().build("pingora_proxy", LevelFilter::Off))
|
||||
.build(Root::builder().appender("file").build(log_level))
|
||||
.unwrap();
|
||||
.die("Parsing Lo4rsConfig");
|
||||
if let Err(e) = log4rs::init_config(config) {
|
||||
eprintln!("Warning: log4rs logger already initialized: {:?}", e);
|
||||
}
|
||||
@@ -204,7 +205,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
|
||||
.appender(Appender::builder().build("stdout", Box::new(stdout)))
|
||||
.logger(Logger::builder().build("pingora_proxy", LevelFilter::Off))
|
||||
.build(Root::builder().appender("stdout").build(log_level))
|
||||
.unwrap();
|
||||
.die("Parsing Lo4rsConfig");
|
||||
if let Err(e) = log4rs::init_config(config) {
|
||||
eprintln!("Warning: log4rs logger already initialized: {:?}", e);
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
use crate::tls::acme::lookup::lookup_wait;
|
||||
use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
|
||||
use crate::utils::tools::Die;
|
||||
use log::info;
|
||||
use reqwest::Client;
|
||||
use serde_json::json;
|
||||
@@ -12,8 +13,8 @@ pub struct CloudflareProvider {
|
||||
|
||||
impl CloudflareProvider {
|
||||
pub fn new() -> Self {
|
||||
let api_token = std::env::var("CLOUDFLARE_API_TOKEN").expect("Environment variable CLOUDFLARE_API_TOKEN is required");
|
||||
let zone_id = std::env::var("CLOUDFLARE_ZONE_ID").expect("Environment variable CLOUDFLARE_ZONE_ID is required");
|
||||
let api_token = std::env::var("CLOUDFLARE_API_TOKEN").die("Environment variable CLOUDFLARE_API_TOKEN is required");
|
||||
let zone_id = std::env::var("CLOUDFLARE_ZONE_ID").die("Environment variable CLOUDFLARE_ZONE_ID is required");
|
||||
|
||||
Self {
|
||||
client: Client::new(),
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
use crate::tls::acme::lookup::lookup_wait;
|
||||
use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
|
||||
use crate::utils::tools::Die;
|
||||
use async_trait::async_trait;
|
||||
use log::info;
|
||||
use serde_json::json;
|
||||
@@ -12,7 +13,7 @@ pub struct DigitalOceanProvider {
|
||||
|
||||
impl DigitalOceanProvider {
|
||||
pub fn new() -> Self {
|
||||
let token = env::var("DIGITALOCEAN_TOKEN").expect("Environment variable DIGITALOCEAN_TOKEN is required");
|
||||
let token = env::var("DIGITALOCEAN_TOKEN").die("Environment variable DIGITALOCEAN_TOKEN is required");
|
||||
|
||||
Self {
|
||||
token,
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
|
||||
use crate::tls::acme::lookup::lookup_wait;
|
||||
use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
|
||||
use aws_config::SdkConfig;
|
||||
use crate::utils::tools::Die;
|
||||
use aws_sdk_route53::Client;
|
||||
use aws_sdk_route53::types::{Change, ChangeAction, ChangeBatch, ResourceRecord, ResourceRecordSet, RrType};
|
||||
use log::info;
|
||||
@@ -17,13 +17,20 @@ pub struct Route53Provider {
|
||||
|
||||
impl Route53Provider {
|
||||
pub fn new() -> Self {
|
||||
let hosted_zone_id = std::env::var("AWS_HOSTED_ZONE_ID").expect("Environment variable AWS_HOSTED_ZONE_ID is required");
|
||||
// Load AWS SDK config synchronously using Tokio runtime handle to fit factory construction
|
||||
let config: SdkConfig = if let Ok(handle) = Handle::try_current() {
|
||||
tokio::task::block_in_place(|| handle.block_on(aws_config::load_from_env()))
|
||||
let hosted_zone_id = std::env::var("AWS_HOSTED_ZONE_ID").die("Environment variable AWS_HOSTED_ZONE_ID is required");
|
||||
|
||||
let config = if let Ok(handle) = Handle::try_current() {
|
||||
// If we're inside a tokio runtime, we can block on it directly
|
||||
handle.block_on(aws_config::load_from_env())
|
||||
} else {
|
||||
tokio::runtime::Runtime::new().unwrap().block_on(aws_config::load_from_env())
|
||||
// If we're completely outside tokio, spin up a temporary current-thread runtime
|
||||
tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.die("Failed to create current-thread runtime")
|
||||
.block_on(aws_config::load_from_env())
|
||||
};
|
||||
|
||||
let client = Client::new(&config);
|
||||
Self { client, hosted_zone_id }
|
||||
}
|
||||
|
||||
+6
-17
@@ -5,7 +5,6 @@ use axum::body::Body;
|
||||
use axum::extract::State;
|
||||
use axum::http::{Response, StatusCode};
|
||||
use axum::response::IntoResponse;
|
||||
|
||||
#[allow(clippy::needless_return)]
|
||||
pub async fn acme_create(State(state): State<webserver::AppState>) -> impl IntoResponse {
|
||||
match account::load_or_create(state.cert_creds.as_str()).await {
|
||||
@@ -14,13 +13,10 @@ pub async fn acme_create(State(state): State<webserver::AppState>) -> impl IntoR
|
||||
.status(StatusCode::OK)
|
||||
.header("Content-Type", "text/plain")
|
||||
.body(Body::from(txt))
|
||||
.unwrap();
|
||||
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
|
||||
}
|
||||
Err(e) => {
|
||||
return Response::builder()
|
||||
.status(StatusCode::INTERNAL_SERVER_ERROR)
|
||||
.body(Body::from(format!("Failed to create account: {}", e)))
|
||||
.unwrap();
|
||||
return (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to create account {}\n", e)).into_response();
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -33,13 +29,10 @@ pub async fn acme_order(State(state): State<webserver::AppState>, axum::extract:
|
||||
.status(StatusCode::OK)
|
||||
.header("Content-Type", "text/plain")
|
||||
.body(Body::from(txt))
|
||||
.unwrap();
|
||||
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
|
||||
}
|
||||
Err(e) => {
|
||||
return Response::builder()
|
||||
.status(StatusCode::INTERNAL_SERVER_ERROR)
|
||||
.body(Body::from(format!("Failed to order a certificate: {}", e)))
|
||||
.unwrap();
|
||||
return (StatusCode::INTERNAL_SERVER_ERROR, format!("Internal Server Error: {}\n", e)).into_response();
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -50,12 +43,8 @@ pub async fn http01_challenge(axum::extract::Path(token): axum::extract::Path<St
|
||||
.status(StatusCode::OK)
|
||||
.header("Content-Type", "text/plain")
|
||||
.body(Body::from(key_authorization.clone()))
|
||||
.unwrap();
|
||||
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
|
||||
}
|
||||
}
|
||||
Response::builder()
|
||||
.status(StatusCode::NOT_FOUND)
|
||||
.header("Content-Type", "text/plain")
|
||||
.body(Body::from("Not found"))
|
||||
.unwrap()
|
||||
(StatusCode::NOT_FOUND, "Not found\n").into_response()
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
use crate::utils::tools::Die;
|
||||
use hickory_resolver::TokioResolver;
|
||||
use hickory_resolver::config::{NameServerConfig, ResolverConfig, ResolverOpts};
|
||||
use hickory_resolver::net::runtime::TokioRuntimeProvider;
|
||||
@@ -17,7 +18,7 @@ pub async fn lookup_wait(record_name: &str, dns_server_ip: Option<&str>, max_wai
|
||||
let resolver = TokioResolver::builder_with_config(config, TokioRuntimeProvider::default())
|
||||
.with_options(opts)
|
||||
.build()
|
||||
.expect("failed to build DNS resolver");
|
||||
.die("failed to build DNS resolver");
|
||||
|
||||
let start = std::time::Instant::now();
|
||||
let poll_interval = Duration::from_secs(3);
|
||||
|
||||
@@ -2,6 +2,7 @@ use crate::tls::acme::account::get_account;
|
||||
use crate::tls::acme::types::get_provider;
|
||||
use crate::utils::lazylock::CHALLENGES;
|
||||
use crate::utils::lazylock::DOMAINS;
|
||||
use crate::utils::tools::Die;
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
|
||||
use instant_acme::{ChallengeType, Identifier, NewOrder, RetryPolicy};
|
||||
@@ -63,7 +64,7 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
|
||||
let key_auth_str = key_auth.as_str().to_string();
|
||||
|
||||
if is_dns {
|
||||
let prov = provider.as_ref().unwrap();
|
||||
let prov = provider.as_ref().die("Failed to get DNS provider");
|
||||
let dns_provider = get_provider(prov.clone());
|
||||
let dns_value = calculate_dns_value(&key_auth_str);
|
||||
let clean_domain = domain_name.trim_start_matches("*.").to_string();
|
||||
@@ -82,7 +83,7 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
|
||||
}
|
||||
} else {
|
||||
let token = key_auth_str.split('.').next().ok_or("Invalid key authorization")?.to_string();
|
||||
CHALLENGES.write().unwrap().insert(token, key_auth_str);
|
||||
CHALLENGES.write().die("Failed to write challenge").insert(token, key_auth_str);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||
challenge_handle.set_ready().await?;
|
||||
}
|
||||
@@ -98,7 +99,7 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
|
||||
let csr_der = signing_request.der();
|
||||
order.finalize_csr(csr_der).await?;
|
||||
let cert_chain_pem = order.poll_certificate(&RetryPolicy::default()).await?;
|
||||
CHALLENGES.write().unwrap().clear();
|
||||
CHALLENGES.write().die("Challenges lock poisoned").clear();
|
||||
let private_key_pem = private_key.serialize_pem();
|
||||
fs::write(crt, cert_chain_pem)?;
|
||||
fs::write(key, private_key_pem)?;
|
||||
|
||||
+4
-2
@@ -1,8 +1,9 @@
|
||||
use crate::tls::grades;
|
||||
use crate::utils::tools::Die;
|
||||
use dashmap::DashMap;
|
||||
use log::error;
|
||||
use pingora::tls::ssl::{NameType, SniError, SslAlert, SslContext, SslFiletype, SslMethod, SslRef};
|
||||
use rustls_pemfile::{read_one, Item};
|
||||
use rustls_pemfile::{Item, read_one};
|
||||
use serde::Deserialize;
|
||||
use std::collections::HashSet;
|
||||
use std::fs::File;
|
||||
@@ -10,6 +11,7 @@ use std::io::BufReader;
|
||||
use x509_parser::extensions::GeneralName;
|
||||
use x509_parser::nom::Err as NomErr;
|
||||
use x509_parser::prelude::*;
|
||||
|
||||
#[derive(Clone, Deserialize, Debug)]
|
||||
pub struct CertificateConfig {
|
||||
pub cert_path: String,
|
||||
@@ -37,7 +39,7 @@ pub struct Certificates {
|
||||
|
||||
impl Certificates {
|
||||
pub fn new(configs: &Vec<CertificateConfig>, _grade: &str) -> Option<Self> {
|
||||
let default_cert = configs.first().expect("At least one TLS certificate required");
|
||||
let default_cert = configs.first().die("At least one TLS certificate required");
|
||||
let mut cert_infos = Vec::new();
|
||||
let name_map: DashMap<String, SslContext> = DashMap::new();
|
||||
for config in configs {
|
||||
|
||||
+4
-3
@@ -1,5 +1,6 @@
|
||||
use crate::utils::hcclient::{HttpClientRequest, httpclient};
|
||||
use crate::utils::lazylock::{GLOBAL_PARAMS, MASTER_KEY, WORKERS};
|
||||
use crate::utils::tools::Die;
|
||||
use crate::utils::types::{Consul, Extraparams, Headers, Kubernetes, UpstreamsDashMap};
|
||||
use bytes::Bytes;
|
||||
use log::{error, info, warn};
|
||||
@@ -44,7 +45,7 @@ pub struct SendConfigDto {
|
||||
}
|
||||
|
||||
pub fn serialize_config(data: &UpstreamsDashMap) -> String {
|
||||
let config = GLOBAL_PARAMS.read().unwrap();
|
||||
let config = GLOBAL_PARAMS.read().die("Failed to read config");
|
||||
let mut snapshot: HashMap<String, UpstreamConfigDto> = HashMap::new();
|
||||
|
||||
for outer_entry in data.iter() {
|
||||
@@ -268,7 +269,7 @@ pub fn to_yaml(config: &SendConfigDto) -> String {
|
||||
}
|
||||
|
||||
pub async fn send_to_worker(data: String) {
|
||||
let masterkey = MASTER_KEY.get().expect("Master key not set");
|
||||
let masterkey = MASTER_KEY.get().die("Master key not set");
|
||||
if let Some(workers) = WORKERS.get() {
|
||||
for worker in workers {
|
||||
let prefix = if worker.tls { "https://" } else { "http://" };
|
||||
@@ -282,7 +283,7 @@ pub async fn send_to_worker(data: String) {
|
||||
port: worker.port,
|
||||
url: worker_url.as_str(),
|
||||
payload: Bytes::from(data.clone()),
|
||||
headers: Some(vec![("API-KEY", masterkey.clone().expect("Master key must be set in the cluster mode").deref())]),
|
||||
headers: Some(vec![("API-KEY", masterkey.clone().die("Master key must be set in the cluster mode").deref())]),
|
||||
})
|
||||
.await;
|
||||
if (200..500).contains(&status.2) {
|
||||
|
||||
+10
-3
@@ -1,17 +1,19 @@
|
||||
use bytes::Bytes;
|
||||
use http_body_util::Full;
|
||||
use hyper::Request;
|
||||
use hyper_util::client::legacy::connect::HttpConnector;
|
||||
use hyper_util::client::legacy::Client;
|
||||
use hyper_util::client::legacy::connect::HttpConnector;
|
||||
use hyper_util::rt::TokioExecutor;
|
||||
use log::error;
|
||||
|
||||
pub async fn ping_grpc(addr: &str) -> bool {
|
||||
let connector = HttpConnector::new();
|
||||
let client = Client::builder(TokioExecutor::new()).http2_only(true).build(connector);
|
||||
let uri = format!("{}{}", addr, "/aralez.Probe/Ping");
|
||||
let body = Full::new(Bytes::from(vec![0, 0, 0, 0, 0]));
|
||||
let request = Request::post(uri).header("content-type", "application/grpc").header("te", "trailers").body(body).unwrap();
|
||||
match client.request(request).await {
|
||||
|
||||
match Request::post(uri.clone()).header("content-type", "application/grpc").header("te", "trailers").body(body) {
|
||||
Ok(request) => match client.request(request).await {
|
||||
Ok(resp) => resp
|
||||
.headers()
|
||||
.get("content-type")
|
||||
@@ -19,5 +21,10 @@ pub async fn ping_grpc(addr: &str) -> bool {
|
||||
.map(|v| v.starts_with("application/grpc"))
|
||||
.unwrap_or(false),
|
||||
Err(_) => false,
|
||||
},
|
||||
Err(e) => {
|
||||
error!("Failed to create request: {}", e);
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -126,7 +126,7 @@ async fn detect_tls(ip: &str, port: &u16, sni: &str) -> bool {
|
||||
.danger_accept_invalid_certs(true)
|
||||
.danger_accept_invalid_hostnames(true)
|
||||
.build()
|
||||
.unwrap(),
|
||||
.die("Failed to create TLS connector"),
|
||||
);
|
||||
connector.connect(sni, stream).await.is_ok()
|
||||
}
|
||||
|
||||
+31
-18
@@ -1,9 +1,10 @@
|
||||
use crate::utils::lazylock::EVICTION;
|
||||
use crate::utils::tools::Die;
|
||||
use pingora_cache::eviction::EvictionManager;
|
||||
use pingora_http::Method;
|
||||
use pingora_http::StatusCode;
|
||||
use pingora_http::Version;
|
||||
use prometheus::{register_histogram, register_int_counter, register_int_counter_vec, register_int_gauge, Histogram, IntCounter, IntCounterVec, IntGauge};
|
||||
use prometheus::{Histogram, IntCounter, IntCounterVec, IntGauge, register_histogram, register_int_counter, register_int_counter_vec, register_int_gauge};
|
||||
use std::sync::Arc;
|
||||
use std::sync::LazyLock;
|
||||
use std::time::Duration;
|
||||
@@ -16,19 +17,28 @@ pub struct MetricTypes {
|
||||
pub version: Version,
|
||||
}
|
||||
|
||||
pub static OPEN_FILES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_open_files", "Number of open file descriptors").unwrap());
|
||||
pub static LOGGING_ERRORS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_logging_errors", "Number of log errors").unwrap());
|
||||
pub static MEMORY_USAGE: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_memory_bytes", "Total memory allocated in bytes").unwrap());
|
||||
pub static ACTIVE_SESSIONS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").unwrap());
|
||||
pub static REQUEST_COUNT: LazyLock<IntCounter> = LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").unwrap());
|
||||
pub static OPEN_FILES: LazyLock<IntGauge> =
|
||||
LazyLock::new(|| register_int_gauge!("aralez_open_files", "Number of open file descriptors").die("Failed to register OPEN_FILES metric"));
|
||||
pub static LOGGING_ERRORS: LazyLock<IntGauge> =
|
||||
LazyLock::new(|| register_int_gauge!("aralez_logging_errors", "Number of log errors").die("Failed to register LOGGING_ERRORS metric"));
|
||||
pub static MEMORY_USAGE: LazyLock<IntGauge> =
|
||||
LazyLock::new(|| register_int_gauge!("aralez_memory_bytes", "Total memory allocated in bytes").die("Failed to register MEMORY_USAGE metric"));
|
||||
pub static ACTIVE_SESSIONS: LazyLock<IntGauge> =
|
||||
LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").die("Failed to register ACTIVE_SESSIONS metric"));
|
||||
pub static REQUEST_COUNT: LazyLock<IntCounter> =
|
||||
LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").die("Failed to register REQUEST_COUNT metric"));
|
||||
|
||||
pub static CACHE_SIZE_BYTES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_size_bytes", "Current cache size in bytes").unwrap());
|
||||
pub static CACHE_ITEMS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_items", "Current number of cached objects").unwrap());
|
||||
pub static CACHE_EVICTED_BYTES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_bytes_total", "Total bytes evicted from cache").unwrap());
|
||||
pub static CACHE_SIZE_BYTES: LazyLock<IntGauge> =
|
||||
LazyLock::new(|| register_int_gauge!("aralez_cache_size_bytes", "Current cache size in bytes").die("Failed to register CACHE_SIZE_BYTES metric"));
|
||||
pub static CACHE_ITEMS: LazyLock<IntGauge> =
|
||||
LazyLock::new(|| register_int_gauge!("aralez_cache_items", "Current number of cached objects").die("Failed to register CACHE_ITEMS metric"));
|
||||
pub static CACHE_EVICTED_BYTES: LazyLock<IntGauge> =
|
||||
LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_bytes_total", "Total bytes evicted from cache").die("Failed to register CACHE_EVICTED_BYTES metric"));
|
||||
|
||||
pub static CACHE_EVICTED_ITEMS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_items_total", "Total cache items evicted").unwrap());
|
||||
pub static CACHE_EVICTED_ITEMS: LazyLock<IntGauge> =
|
||||
LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_items_total", "Total cache items evicted").die("Failed to register CACHE_EVICTED_ITEMS metric"));
|
||||
pub static RESPONSE_CODES: LazyLock<IntCounterVec> =
|
||||
LazyLock::new(|| register_int_counter_vec!("aralez_responses_total", "Responses grouped by status code", &["status"]).unwrap());
|
||||
LazyLock::new(|| register_int_counter_vec!("aralez_responses_total", "Responses grouped by status code", &["status"]).die("Failed to register RESPONSE_CODES metric"));
|
||||
|
||||
pub static RESPONSE_LATENCY: LazyLock<Histogram> = LazyLock::new(|| {
|
||||
register_histogram!(
|
||||
@@ -36,17 +46,20 @@ pub static RESPONSE_LATENCY: LazyLock<Histogram> = LazyLock::new(|| {
|
||||
"Response latency in seconds",
|
||||
vec![0.001, 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0]
|
||||
)
|
||||
.unwrap()
|
||||
.die("Failed to register RESPONSE_LATENCY metric")
|
||||
});
|
||||
|
||||
pub static REQUESTS_BY_METHOD: LazyLock<IntCounterVec> =
|
||||
LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_method_total", "Number of requests by HTTP method", &["method"]).unwrap());
|
||||
pub static REQUESTS_BY_METHOD: LazyLock<IntCounterVec> = LazyLock::new(|| {
|
||||
register_int_counter_vec!("aralez_requests_by_method_total", "Number of requests by HTTP method", &["method"]).die("Failed to register REQUESTS_BY_METHOD metric")
|
||||
});
|
||||
|
||||
pub static REQUESTS_BY_UPSTREAM: LazyLock<IntCounterVec> =
|
||||
LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_upstream", "Number of requests by UPSTREAM server", &["upstream"]).unwrap());
|
||||
pub static REQUESTS_BY_UPSTREAM: LazyLock<IntCounterVec> = LazyLock::new(|| {
|
||||
register_int_counter_vec!("aralez_requests_by_upstream", "Number of requests by UPSTREAM server", &["upstream"]).die("Failed to register REQUESTS_BY_UPSTREAM metric")
|
||||
});
|
||||
|
||||
pub static REQUESTS_BY_VERSION: LazyLock<IntCounterVec> =
|
||||
LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_version_total", "Number of requests by HTTP versions", &["version"]).unwrap());
|
||||
pub static REQUESTS_BY_VERSION: LazyLock<IntCounterVec> = LazyLock::new(|| {
|
||||
register_int_counter_vec!("aralez_requests_by_version_total", "Number of requests by HTTP versions", &["version"]).die("Failed to register REQUESTS_BY_VERSION metric")
|
||||
});
|
||||
pub static HTTP_11_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/1.1"]));
|
||||
pub static HTTP_20_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/2.0"]));
|
||||
pub static HTTP_30_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/3.0"]));
|
||||
|
||||
+14
-14
@@ -2,7 +2,7 @@ use crate::utils::deser::{send_to_worker, serialize_config};
|
||||
use crate::utils::healthcheck;
|
||||
use crate::utils::lazylock::{DOMAINS, GLOBAL_PARAMS, IM_A_MASTER, MASTER_KEY, NODE_STATUS, REVERSE_STORE, WORKERS};
|
||||
use crate::utils::state::{is_first_run, mark_not_first_run};
|
||||
use crate::utils::tools::clone_dashmap_into;
|
||||
use crate::utils::tools::{Die, clone_dashmap_into};
|
||||
use crate::utils::types::*;
|
||||
use dashmap::DashMap;
|
||||
use log::{error, info, warn};
|
||||
@@ -48,8 +48,8 @@ pub async fn load_configuration(d: &str, kind: &str) -> (Option<Configuration>,
|
||||
return (None, err_msg);
|
||||
}
|
||||
|
||||
let mut confdir = Path::new(d).parent().unwrap().to_path_buf();
|
||||
let mut autocfg = Path::new(d).parent().unwrap().to_path_buf();
|
||||
let mut confdir = Path::new(d).parent().die("Invalid configuration path").to_path_buf();
|
||||
let mut autocfg = Path::new(d).parent().die("Invalid configuration path").to_path_buf();
|
||||
|
||||
autocfg.push("autoconfigs");
|
||||
if fs::metadata(autocfg.clone()).is_err() {
|
||||
@@ -245,7 +245,7 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) {
|
||||
}
|
||||
|
||||
{
|
||||
let mut config_guard = GLOBAL_PARAMS.write().expect("Global config lock poisoned");
|
||||
let mut config_guard = GLOBAL_PARAMS.write().die("Global config lock poisoned");
|
||||
config_guard.client_headers = config.client_headers.clone();
|
||||
config_guard.server_headers = config.server_headers.clone();
|
||||
config_guard.extraparams = config.extraparams.clone();
|
||||
@@ -274,8 +274,8 @@ async fn populate_file_upstreams(config: &mut Configuration, parsed: &Config) {
|
||||
}
|
||||
}
|
||||
pub fn parce_main_config(path: &str) -> AppConfig {
|
||||
let data = fs::read_to_string(path).expect("Failed to read main config file");
|
||||
let mut cfo: AppConfig = noyalib::from_str(&data).expect("Failed to parse main config file");
|
||||
let data = fs::read_to_string(path).die("Failed to read main config file");
|
||||
let mut cfo: AppConfig = noyalib::from_str(&data).die("Failed to parse main config file");
|
||||
if let Ok(jwt_key) = env::var("MASTER_KEY") {
|
||||
cfo.master_key = Some(jwt_key);
|
||||
} else if let Ok(jwt_key) = env::var("JWT_KEY") {
|
||||
@@ -301,19 +301,19 @@ pub fn parce_main_config(path: &str) -> AppConfig {
|
||||
|
||||
cfo.proxy_tls_grade = parce_tls_grades(cfo.proxy_tls_grade.clone());
|
||||
if let Some(member) = cfo.cluster_member.clone() {
|
||||
NODE_STATUS.set(Some(Arc::from(member.as_str()))).expect("Config can only be initialized once");
|
||||
let mk = cfo.master_key.clone().expect("In CLuster mode master key must be set");
|
||||
MASTER_KEY.set(Some(Arc::from(mk.as_str()))).expect("Config can only be initialized once");
|
||||
NODE_STATUS.set(Some(Arc::from(member.as_str()))).die("Config can only be initialized once");
|
||||
let mk = cfo.master_key.clone().die("In CLuster mode master key must be set");
|
||||
MASTER_KEY.set(Some(Arc::from(mk.as_str()))).die("Config can only be initialized once");
|
||||
if member == "master" {
|
||||
IM_A_MASTER.set(true).expect("Master status can only be initialized once");
|
||||
let worker_str = cfo.worker_nodes.as_deref().expect("At least one worker must be set in master mode");
|
||||
IM_A_MASTER.set(true).die("Master status can only be initialized once");
|
||||
let worker_str = cfo.worker_nodes.as_deref().die("At least one worker must be set in master mode");
|
||||
let mut wks = Vec::new();
|
||||
for worker in worker_str.split(',') {
|
||||
let clean = worker.replace("//", "");
|
||||
let wk_vec = clean.trim().split(':').collect::<Vec<&str>>();
|
||||
let wk = WorkerNodes {
|
||||
address: Arc::from(wk_vec[1]),
|
||||
port: wk_vec[2].parse::<u16>().expect("Invalid worker port number"),
|
||||
port: wk_vec[2].parse::<u16>().die("Invalid worker port number"),
|
||||
tls: match wk_vec[0] {
|
||||
"https" => true,
|
||||
"http" => false,
|
||||
@@ -324,13 +324,13 @@ pub fn parce_main_config(path: &str) -> AppConfig {
|
||||
wks.push(wk);
|
||||
// println!("{:?}", wk);
|
||||
// if let Some((host, port_str)) = worker.trim().split_once(':') {
|
||||
// let port = port_str.parse::<u16>().expect("Invalid worker port number");
|
||||
// let port = port_str.parse::<u16>().die("Invalid worker port number");
|
||||
// wks.push((host.to_string(), port));
|
||||
// } else {
|
||||
// panic!("Invalid worker format, expected host:port");
|
||||
// }
|
||||
}
|
||||
WORKERS.set(wks).expect("Workers list can only be initialized once");
|
||||
WORKERS.set(wks).die("Workers list can only be initialized once");
|
||||
}
|
||||
}
|
||||
cfo
|
||||
|
||||
+3
-2
@@ -1,3 +1,4 @@
|
||||
use crate::utils::tools::Die;
|
||||
use std::sync::{LazyLock, RwLock};
|
||||
|
||||
#[derive(Debug)]
|
||||
@@ -8,11 +9,11 @@ pub struct SharedState {
|
||||
pub static GLOBAL_STATE: LazyLock<RwLock<SharedState>> = LazyLock::new(|| RwLock::new(SharedState { first_run: true }));
|
||||
|
||||
pub fn mark_not_first_run() {
|
||||
let mut state = GLOBAL_STATE.write().unwrap();
|
||||
let mut state = GLOBAL_STATE.write().die("Writing global state failed");
|
||||
state.first_run = false;
|
||||
}
|
||||
|
||||
pub fn is_first_run() -> bool {
|
||||
let state = GLOBAL_STATE.read().unwrap();
|
||||
let state = GLOBAL_STATE.read().die("Writing global state failed");
|
||||
state.first_run
|
||||
}
|
||||
|
||||
+46
-8
@@ -7,13 +7,14 @@ use privdrop::PrivDrop;
|
||||
use serde_json::{Value, json};
|
||||
use sha2::{Digest, Sha256};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::fmt::Write;
|
||||
use std::fmt::{Debug, Write};
|
||||
use std::fs::OpenOptions;
|
||||
use std::io::Write as IoWrite;
|
||||
use std::net::SocketAddr;
|
||||
use std::net::TcpListener;
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
use std::panic::Location;
|
||||
use std::process::Command;
|
||||
use std::str::FromStr;
|
||||
use std::sync::Arc;
|
||||
@@ -23,9 +24,9 @@ use std::{env, fs, process, thread, time};
|
||||
pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams) {
|
||||
let mut out = String::new();
|
||||
for host_entry in upstreams.iter() {
|
||||
writeln!(out, "Hostname: {}", host_entry.key()).unwrap();
|
||||
writeln!(out, "Hostname: {}", host_entry.key()).unwrap_or_default();
|
||||
for path_entry in host_entry.value().iter() {
|
||||
writeln!(out, " Path: {}", path_entry.key()).unwrap();
|
||||
writeln!(out, " Path: {}", path_entry.key()).unwrap_or_default();
|
||||
for f in path_entry.value().0.clone() {
|
||||
writeln!(
|
||||
out,
|
||||
@@ -38,7 +39,7 @@ pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams)
|
||||
if f.rate_limit > 0 { f.rate_limit } else { extraparams.rate_limit },
|
||||
if f.x4xx_limit > 0 { f.x4xx_limit } else { extraparams.x4xx_limit }
|
||||
)
|
||||
.unwrap();
|
||||
.unwrap_or_default();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -161,9 +162,9 @@ pub fn clone_idmap_into(original: &UpstreamsDashMap, cloned: &UpstreamsIdMap) {
|
||||
pub fn listdir(dir: String) -> Vec<load::CertificateConfig> {
|
||||
let mut f = HashMap::new();
|
||||
let mut certificate_configs: Vec<load::CertificateConfig> = vec![];
|
||||
let paths = fs::read_dir(dir).unwrap();
|
||||
let paths = fs::read_dir(dir).die("listing certificates directory failed");
|
||||
for path in paths {
|
||||
let path_str = path.unwrap().path().to_str().unwrap().to_owned();
|
||||
let path_str = path.die("getting certs path").path().to_str().die("certs pat t &str").to_owned();
|
||||
if path_str.ends_with(".crt") {
|
||||
let name = path_str.replace(".crt", "");
|
||||
let mut inner = vec![];
|
||||
@@ -212,9 +213,9 @@ pub fn drop_priv(user: String, group: String, http_addr: String, tls_addr: Optio
|
||||
}
|
||||
|
||||
pub fn check_priv(addr: &str) {
|
||||
let port = SocketAddr::from_str(addr).map(|sa| sa.port()).expect("Failed to parse address port ");
|
||||
let port = SocketAddr::from_str(addr).map(|sa| sa.port()).die("Failed to parse address port ");
|
||||
if port < 1024 {
|
||||
let meta = std::fs::metadata("/proc/self").map(|m| m.uid()).unwrap();
|
||||
let meta = std::fs::metadata("/proc/self").map(|m| m.uid()).die("Reading /proc/self");
|
||||
if meta != 0 {
|
||||
error!("Running on privileged port requires to start as ROOT");
|
||||
process::exit(1)
|
||||
@@ -364,3 +365,40 @@ pub fn get_hostname() -> String {
|
||||
.filter(|s| !s.is_empty())
|
||||
.unwrap_or_else(|| "unknown".to_string())
|
||||
}
|
||||
|
||||
// pub fn die(message: &str) -> ! {
|
||||
// log::error!("{message}");
|
||||
// process::exit(1);
|
||||
// }
|
||||
|
||||
pub trait Die<T> {
|
||||
fn die(self, message: &str) -> T;
|
||||
}
|
||||
|
||||
impl<T, E: Debug> Die<T> for Result<T, E> {
|
||||
#[track_caller]
|
||||
fn die(self, message: &str) -> T {
|
||||
match self {
|
||||
Ok(value) => value,
|
||||
Err(e) => {
|
||||
let loc = Location::caller();
|
||||
eprintln!("\x1b[1;31mERROR:\x1b[0m {message}: {e:?}: {}: {}", loc.file(), loc.line());
|
||||
process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Die<T> for Option<T> {
|
||||
#[track_caller]
|
||||
fn die(self, message: &str) -> T {
|
||||
match self {
|
||||
Some(value) => value,
|
||||
None => {
|
||||
let loc = Location::caller();
|
||||
eprintln!("\x1b[1;31mERROR:\x1b[0m {message} at {}:{}", loc.file(), loc.line());
|
||||
process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -112,6 +112,8 @@ pub struct AppConfig {
|
||||
pub hc_method: String,
|
||||
pub upstreams_conf: String,
|
||||
pub log_level: String,
|
||||
pub cache_enabled: Option<bool>,
|
||||
pub cache_storage: Option<String>,
|
||||
pub cache_size_mb: Option<usize>,
|
||||
pub cache_ttl: Option<u64>,
|
||||
pub access_log: Option<String>,
|
||||
|
||||
+5
-5
@@ -1,6 +1,6 @@
|
||||
use crate::tls::CertificateConfig;
|
||||
use crate::utils::parceyaml::load_configuration;
|
||||
use crate::utils::tools::listdir;
|
||||
use crate::utils::tools::{Die, listdir};
|
||||
use crate::utils::types::Configuration;
|
||||
use log::{error, info};
|
||||
use notify::event::ModifyKind;
|
||||
@@ -15,7 +15,7 @@ use tokio::task;
|
||||
pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
|
||||
sleep(Duration::from_millis(50)).await; // For having nice logs :-)
|
||||
let file_path = fp.as_str();
|
||||
let parent_dir = Path::new(file_path).parent().unwrap();
|
||||
let parent_dir = Path::new(file_path).parent().die("Get parent folder");
|
||||
let (local_tx, mut local_rx) = tokio::sync::mpsc::channel::<notify::Result<Event>>(1);
|
||||
|
||||
let _watcher_handle = task::spawn_blocking({
|
||||
@@ -27,8 +27,8 @@ pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
|
||||
},
|
||||
Config::default(),
|
||||
)
|
||||
.unwrap();
|
||||
watcher.watch(&parent_dir, RecursiveMode::Recursive).unwrap();
|
||||
.die("Initialize file watcher");
|
||||
watcher.watch(&parent_dir, RecursiveMode::Recursive).die("");
|
||||
let (_rtx, mut rrx) = tokio::sync::mpsc::channel::<bool>(1);
|
||||
let _ = rrx.blocking_recv();
|
||||
}
|
||||
@@ -42,7 +42,7 @@ pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
|
||||
start = Instant::now();
|
||||
let snd = load_configuration(file_path, "filepath").await.0;
|
||||
if let Some(snd) = snd {
|
||||
toreturn.send(snd).await.unwrap();
|
||||
toreturn.send(snd).await.die("Send to file watcher");
|
||||
}
|
||||
}
|
||||
_ => (),
|
||||
|
||||
Reference in New Issue
Block a user