Compare commits

..

No commits in common. "main" and "recursive-crawl" have entirely different histories.

6 changed files with 734 additions and 898 deletions

1340
Cargo.lock generated

File diff suppressed because it is too large Load diff

View file

@ -5,18 +5,17 @@ authors = ["Felix Ableitner"]
edition = "2018" edition = "2018"
[dependencies] [dependencies]
reqwest = { version = "0.11.13", default-features = false, features = ["json", "rustls-tls"] } reqwest = { version = "0.11.10", default-features = false, features = ["json", "rustls-tls"] }
serde = { version = "1.0.149", features = ["derive"] } serde = { version = "1.0.137", features = ["derive"] }
anyhow = "1.0.66" anyhow = "1.0.57"
tokio = { version = "1.22.0", features = ["macros", "rt-multi-thread"] } tokio = { version = "1.18.1", features = ["macros", "rt-multi-thread"] }
futures = "0.3.25" futures = "0.3.21"
serde_json = "1.0.89" serde_json = "1.0.81"
semver = "1.0.14" semver = "1.0.9"
once_cell = "1.16.0" once_cell = "1.10.0"
lemmy_api_common = "=0.16.0" lemmy_api_common = "0.16.0"
lemmy_db_schema = "=0.16.0"
async-recursion = "1.0.0" async-recursion = "1.0.0"
log = "0.4.17" log = "0.4.17"
derive-new = "0.5.9" derive-new = "0.5.9"
stderrlog = "0.5.4" stderrlog = "0.5.1"
structopt = "0.3.26" structopt = "0.3.26"

View file

@ -1,12 +1,12 @@
use crate::node_info::{NodeInfo, NodeInfoWellKnown};
use crate::CLIENT; use crate::CLIENT;
use anyhow::{anyhow, Error}; use crate::REQUEST_TIMEOUT;
use anyhow::Error;
use async_recursion::async_recursion; use async_recursion::async_recursion;
use futures::future::join_all; use futures::future::join_all;
use lemmy_api_common::site::GetSiteResponse; use lemmy_api_common::site::GetSiteResponse;
use log::debug; use log::debug;
use reqwest::Url;
use semver::Version; use semver::Version;
use serde::Serialize;
use std::collections::HashSet; use std::collections::HashSet;
use std::ops::Deref; use std::ops::Deref;
use std::sync::Arc; use std::sync::Arc;
@ -27,16 +27,15 @@ pub struct CrawlParams {
crawled_instances: Arc<Mutex<HashSet<String>>>, crawled_instances: Arc<Mutex<HashSet<String>>>,
} }
#[derive(Debug)] #[derive(Serialize, Debug)]
pub struct CrawlResult { pub struct InstanceDetails {
pub domain: String, pub domain: String,
pub node_info: NodeInfo, pub site_info: GetSiteResponse,
pub site_info: Option<GetSiteResponse>,
} }
impl CrawlJob { impl CrawlJob {
#[async_recursion] #[async_recursion]
pub async fn crawl(self) -> Vec<Result<CrawlResult, Error>> { pub async fn crawl(self) -> Vec<Result<InstanceDetails, Error>> {
// need to acquire and release mutex before recursing, otherwise it will deadlock // need to acquire and release mutex before recursing, otherwise it will deadlock
{ {
let mut crawled_instances = self.params.crawled_instances.deref().lock().await; let mut crawled_instances = self.params.crawled_instances.deref().lock().await;
@ -53,83 +52,46 @@ impl CrawlJob {
return vec![]; return vec![];
} }
debug!( debug!("Starting crawl for {}, distance {}", &self.domain, &self.current_distance);
"Starting crawl for {}, distance {}", let site_info = match self.fetch_instance_details().await {
&self.domain, &self.current_distance
);
let (node_info, site_info) = match self.fetch_instance_details().await {
Ok(o) => o, Ok(o) => o,
Err(e) => return vec![Err(e)], Err(e) => return vec![Err(e)],
}; };
let mut crawl_result = CrawlResult {
domain: self.domain.clone(),
node_info,
site_info: None,
};
if let Some(site_info) = site_info { if site_info.1 < self.params.min_lemmy_version {
match Version::parse(&site_info.version) { return vec![];
Ok(version) => {
if version < self.params.min_lemmy_version {
return vec![Ok(crawl_result)];
}
}
Err(e) => return vec![Err(e.into())],
} }
let mut result = vec![]; let mut result = vec![];
if let Some(federated) = &site_info.federated_instances { if let Some(federated) = &site_info.0.federated_instances {
for domain in federated.linked.iter() { for domain in federated.linked.iter() {
let crawl_job = CrawlJob::new( let crawl_job =
domain.clone(), CrawlJob::new(domain.clone(), self.current_distance + 1, self.params.clone());
self.current_distance + 1,
self.params.clone(),
);
result.push(crawl_job.crawl()); result.push(crawl_job.crawl());
} }
} }
let mut result2: Vec<Result<CrawlResult, Error>> = let mut result2: Vec<Result<InstanceDetails, Error>> =
join_all(result).await.into_iter().flatten().collect(); join_all(result).await.into_iter().flatten().collect();
debug!("Successfully finished crawl for {}", &self.domain); debug!("Successfully finished crawl for {}", &self.domain);
crawl_result.site_info = Some(site_info); result2.push(Ok(InstanceDetails {
result2.push(Ok(crawl_result)); domain: self.domain,
site_info: site_info.0,
}));
result2 result2
} else {
vec![Ok(crawl_result)]
}
} }
async fn fetch_instance_details(&self) -> Result<(NodeInfo, Option<GetSiteResponse>), Error> { async fn fetch_instance_details(&self) -> Result<(GetSiteResponse, Version), Error> {
let rel_node_info: Url = Url::parse("http://nodeinfo.diaspora.software/ns/schema/2.0") let site_info_url = format!("https://{}/api/v3/site", &self.domain);
.expect("parse nodeinfo relation url");
let node_info_well_known = CLIENT
.get(&format!("https://{}/.well-known/nodeinfo", &self.domain))
.send()
.await?
.json::<NodeInfoWellKnown>()
.await?;
let node_info_url = node_info_well_known
.links
.into_iter()
.find(|l| l.rel == rel_node_info)
.ok_or_else(|| anyhow!("failed to find nodeinfo link for {}", &self.domain))?
.href;
let node_info = CLIENT
.get(node_info_url)
.send()
.await?
.json::<NodeInfo>()
.await?;
let site_info = CLIENT let site_info = CLIENT
.get(&format!("https://{}/api/v3/site", &self.domain)) .get(&site_info_url)
.timeout(REQUEST_TIMEOUT)
.send() .send()
.await? .await?
.json::<GetSiteResponse>() .json::<GetSiteResponse>()
.await .await?;
.ok(); let version = Version::parse(&site_info.version)?;
Ok((node_info, site_info)) Ok((site_info, version))
} }
} }

View file

@ -1,46 +1,28 @@
#[macro_use] #[macro_use]
extern crate derive_new; extern crate derive_new;
use crate::crawl::{CrawlJob, CrawlParams, CrawlResult}; use crate::crawl::{CrawlJob, CrawlParams, InstanceDetails};
use crate::node_info::{NodeInfo, NodeInfoUsage, NodeInfoUsers};
use anyhow::Error; use anyhow::Error;
use futures::future::join_all; use futures::future::join_all;
use lemmy_api_common::site::GetSiteResponse;
use log::warn;
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
use reqwest::{Client, ClientBuilder}; use reqwest::Client;
use semver::Version; use semver::Version;
use serde::Serialize;
use std::collections::HashSet; use std::collections::HashSet;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::sync::Mutex; use tokio::sync::Mutex;
pub mod crawl; pub mod crawl;
mod node_info;
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
static CLIENT: Lazy<Client> = Lazy::new(|| { static CLIENT: Lazy<Client> = Lazy::new(Client::default);
ClientBuilder::new()
.timeout(REQUEST_TIMEOUT)
.user_agent("lemmy-stats-crawler")
.build()
.expect("build reqwest client")
});
#[derive(Serialize, Debug)]
pub struct CrawlResult2 {
pub domain: String,
pub site_info: GetSiteResponse,
pub federated_counts: Option<NodeInfoUsage>,
}
pub async fn start_crawl( pub async fn start_crawl(
start_instances: Vec<String>, start_instances: Vec<String>,
exclude_domains: Vec<String>, exclude_domains: Vec<String>,
max_distance: i32, max_distance: i32,
) -> Result<Vec<CrawlResult2>, Error> { ) -> Result<Vec<InstanceDetails>, Error> {
let params = Arc::new(CrawlParams::new( let params = Arc::new(CrawlParams::new(
min_lemmy_version().await?, min_lemmy_version().await?,
exclude_domains, exclude_domains,
@ -53,29 +35,25 @@ pub async fn start_crawl(
jobs.push(job.crawl()); jobs.push(job.crawl());
} }
let crawl_results: Vec<CrawlResult> = join_all(jobs) // TODO: log the errors
let mut instance_details: Vec<InstanceDetails> = join_all(jobs)
.await .await
.into_iter() .into_iter()
.flatten() .flatten()
.inspect(|r| { .filter_map(|r| r.ok())
if let Err(e) = r {
warn!("{}", e)
}
})
.filter_map(Result::ok)
.collect(); .collect();
let mut crawl_results = calculate_federated_site_aggregates(crawl_results)?;
// Sort by active monthly users descending // Sort by active monthly users descending
crawl_results.sort_unstable_by_key(|i| { instance_details.sort_unstable_by_key(|i| {
i.site_info i.site_info
.site_view .site_view
.as_ref() .as_ref()
.map(|s| s.counts.users_active_month) .map(|s| s.counts.users_active_month)
.unwrap_or(0) .unwrap_or(0)
}); });
crawl_results.reverse(); instance_details.reverse();
Ok(crawl_results)
Ok(instance_details)
} }
/// calculate minimum allowed lemmy version based on current version. in case of current version /// calculate minimum allowed lemmy version based on current version. in case of current version
@ -92,47 +70,3 @@ async fn min_lemmy_version() -> Result<Version, Error> {
version.minor -= 1; version.minor -= 1;
Ok(version) Ok(version)
} }
fn calculate_federated_site_aggregates(
crawl_results: Vec<CrawlResult>,
) -> Result<Vec<CrawlResult2>, Error> {
let node_info: Vec<(String, NodeInfo)> = crawl_results
.iter()
.map(|c| (c.domain.clone(), c.node_info.clone()))
.collect();
let lemmy_instances: Vec<(String, GetSiteResponse)> = crawl_results
.into_iter()
.filter_map(|c| {
let domain = c.domain;
c.site_info.map(|c2| (domain, c2))
})
.collect();
let mut ret = vec![];
for instance in &lemmy_instances {
let federated_counts = if let Some(federated_instances) = &instance.1.federated_instances {
node_info
.iter()
.filter(|i| federated_instances.linked.contains(&i.0) || i.0 == instance.0)
.map(|i| i.1.usage.clone())
.reduce(|a, b| NodeInfoUsage {
users: NodeInfoUsers {
total: a.users.total + b.users.total,
active_halfyear: a.users.active_halfyear + b.users.active_halfyear,
active_month: a.users.active_month + b.users.active_month,
},
posts: a.posts + b.posts,
comments: a.comments + b.comments,
})
} else {
None
};
// TODO: workaround because GetSiteResponse doesnt implement clone
let site_info = serde_json::from_str(&serde_json::to_string(&instance.1)?)?;
ret.push(CrawlResult2 {
domain: instance.0.clone(),
site_info,
federated_counts,
});
}
Ok(ret)
}

View file

@ -1,25 +1,25 @@
use anyhow::Error; use anyhow::Error;
use lemmy_stats_crawler::{start_crawl, CrawlResult2}; use lemmy_stats_crawler::crawl::InstanceDetails;
use lemmy_stats_crawler::start_crawl;
use serde::Serialize; use serde::Serialize;
use structopt::StructOpt; use structopt::StructOpt;
#[derive(StructOpt, Debug)] #[derive(StructOpt, Debug)]
#[structopt()] #[structopt()]
struct Parameters { struct Parameters {
#[structopt(short, long, use_delimiter = true, default_value = "lemmy.ml")] #[structopt(short, long, default_value = "lemmy.ml")]
start_instances: Vec<String>, start_instances: Vec<String>,
#[structopt( #[structopt(short, long, default_value = "ds9.lemmy.ml, enterprise.lemmy.ml, voyager.lemmy.ml, test.lemmy.ml")]
short,
long,
use_delimiter = true,
default_value = "ds9.lemmy.ml,enterprise.lemmy.ml,voyager.lemmy.ml,test.lemmy.ml"
)]
exclude_instances: Vec<String>, exclude_instances: Vec<String>,
#[structopt(short, long, default_value = "20")] #[structopt(short, long, default_value = "20")]
max_crawl_distance: i32, max_crawl_distance: i32,
/// Silence all output /// Silence all output
#[structopt(short, long)] #[structopt(short, long)]
quiet: bool, quiet: bool,
/// Verbose mode (-v, -vv, -vvv, etc)
#[structopt(short = "v", long = "verbose", parse(from_occurrences))]
verbose: usize,
} }
#[tokio::main] #[tokio::main]
@ -29,7 +29,7 @@ pub async fn main() -> Result<(), Error> {
stderrlog::new() stderrlog::new()
.module(module_path!()) .module(module_path!())
.quiet(params.quiet) .quiet(params.quiet)
.verbosity(1) .verbosity(params.verbose)
.init()?; .init()?;
eprintln!("Crawling..."); eprintln!("Crawling...");
@ -56,10 +56,10 @@ struct TotalStats {
users_active_week: i64, users_active_week: i64,
users_active_month: i64, users_active_month: i64,
users_active_halfyear: i64, users_active_halfyear: i64,
instance_details: Vec<CrawlResult2>, instance_details: Vec<InstanceDetails>,
} }
fn aggregate(instance_details: Vec<CrawlResult2>) -> TotalStats { fn aggregate(instance_details: Vec<InstanceDetails>) -> TotalStats {
let mut online_users = 0; let mut online_users = 0;
let mut total_users = 0; let mut total_users = 0;
let mut users_active_day = 0; let mut users_active_day = 0;

View file

@ -1,47 +0,0 @@
use reqwest::Url;
use serde::{Deserialize, Serialize};
#[derive(Deserialize, Debug)]
pub struct NodeInfoWellKnown {
pub links: Vec<NodeInfoWellKnownLinks>,
}
#[derive(Deserialize, Debug)]
pub struct NodeInfoWellKnownLinks {
pub rel: Url,
pub href: Url,
}
#[derive(Deserialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct NodeInfo {
pub version: String,
pub software: NodeInfoSoftware,
pub protocols: Vec<String>,
pub usage: NodeInfoUsage,
pub open_registrations: bool,
}
#[derive(Deserialize, Debug, Clone)]
pub struct NodeInfoSoftware {
pub name: String,
pub version: String,
}
#[derive(Deserialize, Serialize, Debug, Clone, Default)]
#[serde(rename_all = "camelCase", default)]
pub struct NodeInfoUsage {
pub users: NodeInfoUsers,
#[serde(rename(deserialize = "localPosts"))]
pub posts: i64,
#[serde(rename(deserialize = "localComments"))]
pub comments: i64,
}
#[derive(Deserialize, Serialize, Debug, Clone, Default)]
#[serde(rename_all = "camelCase", default)]
pub struct NodeInfoUsers {
pub total: i64,
pub active_halfyear: i64,
pub active_month: i64,
}