Oliver/internet_mapper
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

close #24, make program aware of the files already on disk

Browse Source
This commit is contained in:
Oliver
2025-10-09 21:52:41 -06:00
parent b0fe7f4761
commit 5b728bacd6
2 changed files with 190 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/filesystem.rs
View File

@@ -34,10 +34,35 @@ pub fn as_path(url: &Url, content_type: &HeaderValue) -> PathBuf {
url_path url_path
} }
pub async fn check_file_length(file: &PathBuf) -> Option<u64> {
match tokio::fs::OpenOptions::new()
.write(false)
.read(true)
.create(false)
.open(file).await
{
Ok(file) => {
match file.metadata().await {
Ok(meta) => {
return Some(meta.len())
},
Err(err) => {
error!("Failed to get metadata. {}", err)
},
}
},
Err(err) => {
error!("Failed to open file for testing... {}", err);
},
}
None
}
pub async fn init(filename: &PathBuf) -> Option<fs::File> { pub async fn init(filename: &PathBuf) -> Option<fs::File> {
let file = async || tokio::fs::OpenOptions::new() let file = async || tokio::fs::OpenOptions::new()
.write(true) .write(true)
.append(true) .append(false)
.create(true) .create(true)
.open(&filename).await; .open(&filename).await;

154
src/main.rs
View File

@@ -5,17 +5,27 @@
extern crate html5ever; extern crate html5ever;
use std::{ use std::{
collections::HashSet, fs::File, io::Read, sync::{Arc, LazyLock} collections::HashSet,
fs::File,
io::Read,
sync::{Arc, LazyLock},
}; };
use futures_util::StreamExt;
use opentelemetry::{global::{self}, metrics::{Counter, Meter, UpDownCounter}};
use opentelemetry_otlp::{Protocol, WithExportConfig};
use db::{connect, Website}; use db::{connect, Website};
use futures_util::StreamExt;
use opentelemetry::{
global::{self},
metrics::{Counter, Meter, UpDownCounter},
};
use opentelemetry_otlp::{Protocol, WithExportConfig};
use opentelemetry_sdk::{metrics::SdkMeterProvider, trace::SdkTracerProvider}; use opentelemetry_sdk::{metrics::SdkMeterProvider, trace::SdkTracerProvider};
use serde::Deserialize; use serde::Deserialize;
use surrealdb::{engine::remote::ws::Client, Surreal}; use surrealdb::{engine::remote::ws::Client, Surreal};
use tokio::{io::{AsyncWriteExt, BufWriter}, sync::RwLock, task::JoinSet}; use tokio::{
io::{AsyncWriteExt, BufWriter},
sync::RwLock,
task::JoinSet,
};
use tracing::{debug, error, info, instrument, level_filters::LevelFilter, trace, warn}; use tracing::{debug, error, info, instrument, level_filters::LevelFilter, trace, warn};
use tracing_subscriber::{fmt, layer::SubscriberExt, EnvFilter, Layer, Registry}; use tracing_subscriber::{fmt, layer::SubscriberExt, EnvFilter, Layer, Registry};
@@ -26,41 +36,29 @@ mod filesystem;
mod parser; mod parser;
static METER: LazyLock<Meter> = LazyLock::new(|| global::meter("Internet_Mapper")); static METER: LazyLock<Meter> = LazyLock::new(|| global::meter("Internet_Mapper"));
static BATCH_SIZE: LazyLock<Counter<u64>> = LazyLock::new(|| static BATCH_SIZE: LazyLock<Counter<u64>> =
METER LazyLock::new(|| METER.u64_counter("crawler_batch_size").build());
.u64_counter("crawler_batch_size") static BEING_PROCESSED: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| {
.build()
);
static BEING_PROCESSED: LazyLock<UpDownCounter<i64>> = LazyLock::new(||
METER METER
.i64_up_down_counter("crawler_pages_being_processed") .i64_up_down_counter("crawler_pages_being_processed")
.build() .build()
); });
static BEING_PARSED: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| static BEING_PARSED: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| {
METER METER
.i64_up_down_counter("crawler_pages_being_parsed") .i64_up_down_counter("crawler_pages_being_parsed")
.build() .build()
); });
static BEING_STREAMED: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| static BEING_STREAMED: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| {
METER METER
.i64_up_down_counter("crawler_pages_being_streamed") .i64_up_down_counter("crawler_pages_being_streamed")
.build() .build()
); });
static GET_IN_FLIGHT: LazyLock<UpDownCounter<i64>> = LazyLock::new(|| static GET_IN_FLIGHT: LazyLock<UpDownCounter<i64>> =
METER LazyLock::new(|| METER.i64_up_down_counter("crawler_gets_in_flight").build());
.i64_up_down_counter("crawler_gets_in_flight") static TOTAL_BYTES_DOWN: LazyLock<Counter<u64>> =
.build() LazyLock::new(|| METER.u64_counter("crawler_total_bytes_down").build());
); static SITES_CRAWLED: LazyLock<Counter<u64>> =
static TOTAL_BYTES_DOWN: LazyLock<Counter<u64>> = LazyLock::new(|| LazyLock::new(|| METER.u64_counter("crawler_total_sites_crawled").build());
METER
.u64_counter("crawler_total_bytes_down")
.build()
);
static SITES_CRAWLED: LazyLock<Counter<u64>> = LazyLock::new(||
METER
.u64_counter("crawler_total_sites_crawled")
.build()
);
static CONFIG: LazyLock<Config> = LazyLock::new(|| { static CONFIG: LazyLock<Config> = LazyLock::new(|| {
let mut file = File::open("./Crawler.toml").expect("Failed to read Crawler.toml"); let mut file = File::open("./Crawler.toml").expect("Failed to read Crawler.toml");
@@ -130,15 +128,25 @@ async fn main() {
// let mut main_loop_span= TRACER.start("Main-Loop"); // let mut main_loop_span= TRACER.start("Main-Loop");
let mut futures = JoinSet::new(); let mut futures = JoinSet::new();
for _ in 0..CONFIG.batch_size { for _ in 0..CONFIG.batch_size {
futures.spawn(process_single_thread(&CONFIG, db.clone(), reqwest.clone(), crawled.clone())); futures.spawn(process_single_thread(
&CONFIG,
db.clone(),
reqwest.clone(),
crawled.clone(),
));
} }
while let Some(_) = futures.join_next().await { while let Some(_) = futures.join_next().await {
// Budget - Threads - This thread (1) // Budget - Threads - This thread (1)
// Would roughly be the acceptable amount at which a thread should exit // Would roughly be the acceptable amount at which a thread should exit
if *(crawled.read().await) < CONFIG.budget - CONFIG.batch_size -1 { if *(crawled.read().await) < CONFIG.budget - CONFIG.batch_size - 1 {
warn!("Thread terminated early, restarting"); warn!("Thread terminated early, restarting");
futures.spawn(process_single_thread(&CONFIG, db.clone(), reqwest.clone(), crawled.clone())); futures.spawn(process_single_thread(
&CONFIG,
db.clone(),
reqwest.clone(),
crawled.clone(),
));
} }
} }
@@ -148,7 +156,12 @@ async fn main() {
info!("Done"); info!("Done");
} }
async fn process_single_thread(config: &Config, db: Surreal<Client>, reqwest: reqwest::Client, crawled: Arc<RwLock<usize>>) { async fn process_single_thread(
config: &Config,
db: Surreal<Client>,
reqwest: reqwest::Client,
crawled: Arc<RwLock<usize>>,
) {
while *(crawled.read().await) < config.budget { while *(crawled.read().await) < config.budget {
let uncrawled = get_next(&db.clone(), &config).await; let uncrawled = get_next(&db.clone(), &config).await;
match uncrawled { match uncrawled {
@@ -158,11 +171,11 @@ async fn process_single_thread(config: &Config, db: Surreal<Client>, reqwest: re
// Somehow this write doesn't hang on the while's read? // Somehow this write doesn't hang on the while's read?
let mut c = crawled.write().await; let mut c = crawled.write().await;
*c += 1; *c += 1;
}, }
None => { None => {
warn!("fn::get_next() returned None"); warn!("fn::get_next() returned None");
return; return;
}, }
} }
} }
} }
@@ -171,7 +184,6 @@ async fn process_single_thread(config: &Config, db: Surreal<Client>, reqwest: re
/// Downloads and crawls and stores a webpage. /// Downloads and crawls and stores a webpage.
/// It is acceptable to clone `db`, `reqwest`, and `s3` because they all use `Arc`s internally. - Noted by Oliver /// It is acceptable to clone `db`, `reqwest`, and `s3` because they all use `Arc`s internally. - Noted by Oliver
async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Client) { async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Client) {
// METRICS // METRICS
debug!(url = &site.site.as_str(), "Process: {}", &site.site); debug!(url = &site.site.as_str(), "Process: {}", &site.site);
BEING_PROCESSED.add(1, &[]); BEING_PROCESSED.add(1, &[]);
@@ -183,6 +195,8 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
// Send the http request (get) // Send the http request (get)
GET_IN_FLIGHT.add(1, &[]); GET_IN_FLIGHT.add(1, &[]);
if let Ok(response) = request_builder.send().await { if let Ok(response) = request_builder.send().await {
let mut skip_download = false;
GET_IN_FLIGHT.add(-1, &[]); GET_IN_FLIGHT.add(-1, &[]);
let headers = response.headers(); let headers = response.headers();
@@ -195,32 +209,59 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
let CT = headers.get("Content-Type"); let CT = headers.get("Content-Type");
let ct = headers.get("content-type"); let ct = headers.get("content-type");
let ct = match (CT,ct) { let ct = match (CT, ct) {
(None, None) => { (None, None) => {
warn!("Server did not respond with Content-Type header. Url: {} Headers: ({:?})", site.site.to_string(), headers); warn!(
return "Server did not respond with Content-Type header. Url: {} Headers: ({:?})",
}, site.site.to_string(),
headers
);
return;
}
(None, Some(a)) => a, (None, Some(a)) => a,
(Some(a), None) => a, (Some(a), None) => a,
(Some(a), Some(_)) => a, (Some(a), Some(_)) => a,
}; };
// create filepath (handles / -> /index.html) // create filepath (handles / -> /index.html)
let path = filesystem::as_path(&site.site, ct); let real_path = filesystem::as_path(&site.site, ct);
let mut tmp_path= path.clone(); let mut tmp_path = real_path.clone();
if !(tmp_path.add_extension("crawl_temp")) { if !(tmp_path.add_extension("crawl_temp")) {
warn!("Failed to add extension to file"); warn!("Failed to add extension to file");
// fallback ig // fallback ig
tmp_path = tmp_path.with_extension("crawl_temp"); tmp_path = tmp_path.with_extension("crawl_temp");
} }
// CODE FOR UPDATING DOWNLOADED CONTENT:
// Check the Content-Length header (we assume the server is telling the truth) (I don't see
// a reason for it to lie in this case).
// And see if the file on the disk is the same length.
// Yes, technically this isn't perfect, but the other option is storing ETags, which I
// don't want to do right now.
if let Some(len) = headers.get("Content-Length") {
if let Ok(s) = len.to_str() {
// length is in bytes
if let Ok(len) = s.parse::<u64>() {
if let Some(disk_len) = filesystem::check_file_length(&real_path).await {
if disk_len == len {
skip_download = true;
}
}
}
}
}
if skip_download {
trace!("Skipping download...");
} else {
// make sure that the file is good to go // make sure that the file is good to go
if let Some(file) = filesystem::init(&tmp_path).await { if let Some(file) = filesystem::init(&tmp_path).await {
// Get body from response // Get body from response
// stream the response onto the disk // stream the response onto the disk
let mut stream = response.bytes_stream(); let mut stream = response.bytes_stream();
let should_parse = path.to_string_lossy().ends_with(".html"); let should_parse = real_path.to_string_lossy().ends_with(".html");
let mut writer = BufWriter::new(file); let mut writer = BufWriter::new(file);
let mut buf: Vec<u8> = Vec::new(); let mut buf: Vec<u8> = Vec::new();
@@ -239,18 +280,18 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
if should_parse { if should_parse {
data.iter().for_each(|f| buf.push(*f)); data.iter().for_each(|f| buf.push(*f));
} }
}, }
Err(err) => { Err(err) => {
error!("{err}") error!("{err}")
}, }
} }
} }
let _ = writer.flush().await; let _ = writer.flush().await;
// rename the temp file into the real file name // rename the temp file into the real file name
if let Err(err) = tokio::fs::rename(&tmp_path, &path).await { if let Err(err) = tokio::fs::rename(&tmp_path, &real_path).await {
error!( error!(
from = &*tmp_path.to_string_lossy(), from = &*tmp_path.to_string_lossy(),
to = &*path.to_string_lossy(), to = &*real_path.to_string_lossy(),
"Error renaming file: {}", "Error renaming file: {}",
err err
); );
@@ -289,6 +330,7 @@ async fn process(mut site: Website, db: Surreal<Client>, reqwest: reqwest::Clien
site.status_code = code.as_u16(); site.status_code = code.as_u16();
Website::store_all(vec![site.clone()], &db).await; Website::store_all(vec![site.clone()], &db).await;
} }
}
} else { } else {
error!(url = site.site.as_str(), "Failed to get: {}", &site.site); error!(url = site.site.as_str(), "Failed to get: {}", &site.site);
} }
@@ -311,11 +353,11 @@ fn load_tracing(config: &Config) -> SdkTracerProvider {
} }
fn load_logging(config: &Config) { fn load_logging(config: &Config) {
// let otlp_log = opentelemetry_otlp::LogExporter::builder() // let otlp_log = opentelemetry_otlp::LogExporter::builder()
// .with_tonic() // .with_tonic()
// .with_endpoint(endpoint) // .with_endpoint(endpoint)
// .build() // .build()
// .unwrap(); // .unwrap();
// let logger_provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder() // let logger_provider = opentelemetry_sdk::logs::SdkLoggerProvider::builder()
// .with_simple_exporter(otlp_log) // .with_simple_exporter(otlp_log)
// .build(); // .build();
@@ -336,7 +378,7 @@ fn load_logging(config: &Config) {
.with_file(true) .with_file(true)
.json() .json()
.with_writer(writer) .with_writer(writer)
.with_filter(filter) .with_filter(filter),
); );
tracing::subscriber::set_global_default(registry).expect("Failed to set default subscriber"); tracing::subscriber::set_global_default(registry).expect("Failed to set default subscriber");
@@ -355,5 +397,3 @@ fn load_metrics(config: &Config) -> SdkMeterProvider {
.build(); .build();
metrics_provider metrics_provider
} }