Mirrored_Repos/veloren
1
0
Fork 0
mirror of https://gitlab.com/veloren/veloren.git synced 2024-08-30 18:12:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

move whole webserver to server-cli

Browse Source
This commit is contained in:
Marcel Märtens 2023-10-15 20:17:45 +02:00
parent 38f4b8b644
commit adeab73876
11 changed files with 304 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Cargo.lock generated
View File

@ -7217,7 +7217,6 @@ version = "0.15.0"
dependencies = [
"atomicwrites",
"authc",
"axum",
"bincode",
"censor",
"chrono",
@ -7228,7 +7227,6 @@ dependencies = [
"futures-util",
"hashbrown 0.13.2",
"humantime",
"hyper",
"itertools",
"lazy_static",
"noise",
@ -7283,12 +7281,16 @@ dependencies = [
name = "veloren-server-cli"
version = "0.15.0"
dependencies = [
"axum",
"cansi",
"chrono",
"clap",
"crossterm 0.26.1",
"hyper",
"lazy_static",
"mimalloc",
"num_cpus",
"prometheus",
"ron 0.8.1",
"serde",
"shell-words",

6
server-cli/Cargo.toml
View File

@ -47,6 +47,12 @@ tracing = { workspace = true }
ron = { workspace = true }
serde = { workspace = true, features = [ "rc", "derive" ]}
#HTTP
axum = { version = "0.6.20" }
hyper = "0.14.26"
prometheus = { workspace = true }
chrono = { workspace = true }
[target.'cfg(windows)'.dependencies]
mimalloc = "0.1.29"

26
server-cli/src/main.rs
View File

@ -16,6 +16,7 @@ mod settings;
mod shutdown_coordinator;
mod tui_runner;
mod tuilog;
mod web;
use crate::{
cli::{Admin, ArgvApp, ArgvCommand, Message, SharedCommand, Shutdown},
shutdown_coordinator::ShutdownCoordinator,
@ -31,6 +32,7 @@ use std::{
sync::{atomic::AtomicBool, mpsc, Arc},
time::{Duration, Instant},
};
use tokio::sync::Notify;
use tracing::{info, trace};
lazy_static::lazy_static! {
@ -185,7 +187,7 @@ fn main() -> io::Result<()> {
info!("Starting server...");
let protocols_and_addresses = server_settings.gameserver_protocols.clone();
let metrics_port = &server_settings.metrics_address.port();
let web_port = &settings.web_address.port();
// Create server
let mut server = Server::new(
server_settings,
@ -193,10 +195,27 @@ fn main() -> io::Result<()> {
database_settings,
&server_data_dir,
&|_| {},
runtime,
Arc::clone(&runtime),
)
.expect("Failed to create server instance!");
let registry = Arc::clone(server.metrics_registry());
let chat = server.chat_cache().clone();
let metrics_shutdown = Arc::new(Notify::new());
let metrics_shutdown_clone = Arc::clone(&metrics_shutdown);
let web_chat_secret = settings.web_chat_secret.clone();
runtime.spawn(async move {
web::run(
registry,
chat,
web_chat_secret,
settings.web_address,
metrics_shutdown_clone.notified(),
)
.await
});
// Collect addresses that the server is listening to log.
let gameserver_addresses = protocols_and_addresses
.into_iter()
@ -210,7 +229,7 @@ fn main() -> io::Result<()> {
});
info!(
?metrics_port,
?web_port,
?gameserver_addresses,
"Server is ready to accept connections."
);
@ -243,6 +262,7 @@ fn main() -> io::Result<()> {
tick_no += 1;
// Terminate the server if instructed to do so by the shutdown coordinator
if shutdown_coordinator.check(&mut server, &settings) {
metrics_shutdown.notify_one();
break;
}

12
server-cli/src/settings.rs
View File

@ -1,5 +1,9 @@
use serde::{Deserialize, Serialize};
use std::{fs, path::PathBuf};
use std::{
fs,
net::{Ipv4Addr, SocketAddr},
path::PathBuf,
};
use tracing::warn;
#[derive(Clone, Debug, Serialize, Deserialize)]
@ -7,6 +11,10 @@ use tracing::warn;
pub struct Settings {
pub update_shutdown_grace_period_secs: u32,
pub update_shutdown_message: String,
pub web_address: SocketAddr,
/// SECRET API HEADER used to access the chat api, if disabled the API is
/// unreachable
pub web_chat_secret: Option<String>,
}
impl Default for Settings {
@ -14,6 +22,8 @@ impl Default for Settings {
Self {
update_shutdown_grace_period_secs: 120,
update_shutdown_message: "The server is restarting for an update".to_owned(),
web_address: SocketAddr::from((Ipv4Addr::LOCALHOST, 14005)),
web_chat_secret: None,
}
}
}

72
server/src/web/chat.rs → server-cli/src/web/chat.rs
View File

@ -5,63 +5,16 @@ use axum::{
routing::get,
Json, Router,
};
use chrono::{DateTime, Utc};
use common::comp::ChatMsg;
use chrono::DateTime;
use hyper::{Request, StatusCode};
use serde::{Deserialize, Deserializer, Serialize};
use std::{
borrow::Cow, collections::VecDeque, mem::size_of, ops::Sub, str::FromStr, sync::Arc,
time::Duration,
};
use tokio::sync::Mutex;
/// The chat cache gets it data from the gameserver and will keep it for some
/// time It will be made available for its consumers, the REST Api
#[derive(Clone)]
pub struct ChatCache {
messages: Arc<Mutex<VecDeque<ChatMessage>>>,
}
use serde::{Deserialize, Deserializer};
use server::chat::ChatCache;
use std::{borrow::Cow, str::FromStr};
/// Keep Size small, so we dont have to Clone much for each request.
#[derive(Clone)]
struct ChatToken {
secret_token: Cow<'static, str>,
}
pub struct ChatExporter {
messages: Arc<Mutex<VecDeque<ChatMessage>>>,
keep_duration: chrono::Duration,
}
impl ChatExporter {
pub fn send(&self, msg: ChatMsg) {
let time = Utc::now();
let drop_older_than = time.sub(self.keep_duration);
let mut messages = self.messages.blocking_lock();
while let Some(msg) = messages.front() && msg.time < drop_older_than {
messages.pop_front();
}
messages.push_back(ChatMessage { time, msg });
const MAX_CACHE_BYTES: usize = 10_000_000; // approx. because HashMap allocates on Heap
if messages.len() * size_of::<ChatMessage>() > MAX_CACHE_BYTES {
let msg_count = messages.len();
tracing::debug!(?msg_count, "shrinking cache");
messages.shrink_to_fit();
}
}
}
impl ChatCache {
pub fn new(keep_duration: Duration) -> (Self, ChatExporter) {
let messages: Arc<Mutex<VecDeque<ChatMessage>>> = Default::default();
let messages_clone = Arc::clone(&messages);
let keep_duration = chrono::Duration::from_std(keep_duration).unwrap();
(Self { messages }, ChatExporter {
messages: messages_clone,
keep_duration,
})
}
secret_token: Option<Cow<'static, str>>,
}
async fn validate_secret<B>(
@ -69,22 +22,25 @@ async fn validate_secret<B>(
req: Request<B>,
next: Next<B>,
) -> Result<Response, StatusCode> {
// check if this endpoint is disabled
let secret_token = token.secret_token.ok_or(StatusCode::METHOD_NOT_ALLOWED)?;
pub const X_SECRET_TOKEN: &str = "X-Secret-Token";
let session_cookie = req
.headers()
.get(X_SECRET_TOKEN)
.ok_or(StatusCode::UNAUTHORIZED)?;
if session_cookie.as_bytes() != token.secret_token.as_bytes() {
if session_cookie.as_bytes() != secret_token.as_bytes() {
return Err(StatusCode::UNAUTHORIZED);
}
Ok(next.run(req).await)
}
pub fn router(cache: ChatCache, secret_token: String) -> Router {
pub fn router(cache: ChatCache, secret_token: Option<String>) -> Router {
let token = ChatToken {
secret_token: Cow::Owned(secret_token),
secret_token: secret_token.map(Cow::Owned),
};
Router::new()
.route("/history", get(history))
@ -92,12 +48,6 @@ pub fn router(cache: ChatCache, secret_token: String) -> Router {
.with_state(cache)
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ChatMessage {
time: DateTime<Utc>,
msg: ChatMsg,
}
#[derive(Debug, Deserialize)]
struct Params {
#[serde(default, deserialize_with = "empty_string_as_none")]

6
server/src/web/mod.rs → server-cli/src/web/mod.rs
View File

@ -2,15 +2,15 @@ use axum::{extract::State, response::IntoResponse, routing::get, Router};
use core::{future::Future, ops::Deref};
use hyper::{header, http, Body, StatusCode};
use prometheus::{Registry, TextEncoder};
use server::chat::ChatCache;
use std::net::SocketAddr;
mod chat;
pub use chat::{ChatCache, ChatExporter};
pub async fn run<S, F, R>(
registry: R,
cache: ChatCache,
chat_token: String,
chat_secret: Option<String>,
addr: S,
shutdown: F,
) -> Result<(), hyper::Error>
@ -24,7 +24,7 @@ where
.with_state(registry.deref().clone());
let app = Router::new()
.nest("/chat/v1", chat::router(cache, chat_token))
.nest("/chat/v1", chat::router(cache, chat_secret))
.nest("/metrics", metrics)
.route("/health", get(|| async {}));

4
server/Cargo.toml
View File

@ -64,10 +64,6 @@ enum-map = { workspace = true }
noise = { version = "0.7", default-features = false }
censor = "0.3"
#HTTP
axum = { version = "0.6.20" }
hyper = "0.14.26"
rusqlite = { version = "0.28.0", features = ["array", "vtab", "bundled", "trace"] }
refinery = { version = "0.8.8", features = ["rusqlite"] }

228
server/src/chat.rs Normal file
View File

@ -0,0 +1,228 @@
use chrono::{DateTime, Utc};
use common::{
comp,
comp::{chat::KillType, ChatType, Content, Group, Player, UnresolvedChatMsg},
uid::IdMaps,
uuid::Uuid,
};
use serde::{Deserialize, Serialize};
use specs::{Join, World, WorldExt};
use std::{collections::VecDeque, ops::Sub, sync::Arc, time::Duration};
use tokio::sync::Mutex;
#[derive(Clone, Serialize, Deserialize)]
pub struct PlayerInfo {
uuid: Uuid,
alias: String,
}
#[derive(Clone, Serialize, Deserialize)]
pub enum KillSource {
Player(PlayerInfo, KillType),
NonPlayer(String, KillType),
NonExistent(KillType),
Environment(String),
FallDamage,
Suicide,
Other,
}
#[derive(Clone, Serialize, Deserialize)]
/// partially mapped to common::comp::ChatMsg
pub enum ChatParties {
Online(PlayerInfo),
Offline(PlayerInfo),
CommandInfo(PlayerInfo),
CommandError(PlayerInfo),
Kill(KillSource, PlayerInfo),
GroupMeta(Vec<PlayerInfo>),
Group(PlayerInfo, Vec<PlayerInfo>),
Tell(PlayerInfo, PlayerInfo),
Say(PlayerInfo),
FactionMeta(String),
Faction(PlayerInfo, String),
Region(PlayerInfo),
World(PlayerInfo),
}
#[derive(Clone, Serialize, Deserialize)]
pub struct ChatMessage {
pub time: DateTime<Utc>,
pub parties: ChatParties,
pub content: Content,
}
type MessagesStore = Arc<Mutex<VecDeque<ChatMessage>>>;
/// The chat cache gets it data from the gameserver and will keep it for some
/// time It will be made available for its consumers, the REST Api
#[derive(Clone)]
pub struct ChatCache {
pub messages: MessagesStore,
}
pub struct ChatExporter {
messages: MessagesStore,
keep_duration: chrono::Duration,
}
impl ChatMessage {
fn new(chatmsg: &UnresolvedChatMsg, parties: ChatParties) -> Self {
ChatMessage {
time: Utc::now(),
content: chatmsg.content().clone(),
parties,
}
}
}
impl ChatExporter {
pub fn generate(chatmsg: &UnresolvedChatMsg, ecs: &World) -> Option<ChatMessage> {
let id_maps = ecs.read_resource::<IdMaps>();
let players = ecs.read_storage::<Player>();
let player_info_from_uid = |uid| {
id_maps
.uid_entity(uid)
.and_then(|entry| players.get(entry))
.map(|player| PlayerInfo {
alias: player.alias.clone(),
uuid: player.uuid(),
})
};
let group_members_from_group = |g| -> Vec<_> {
let groups = ecs.read_storage::<Group>();
(&players, &groups)
.join()
.filter_map(|(player, group)| {
if g == group {
Some(PlayerInfo {
alias: player.alias.clone(),
uuid: player.uuid(),
})
} else {
None
}
})
.collect()
};
match &chatmsg.chat_type {
ChatType::Offline(from) => {
if let Some(player_info) = player_info_from_uid(*from) {
return Some(ChatMessage::new(chatmsg, ChatParties::Offline(player_info)));
}
},
ChatType::Online(from) => {
if let Some(player_info) = player_info_from_uid(*from) {
return Some(ChatMessage::new(chatmsg, ChatParties::Online(player_info)));
}
},
ChatType::Region(from) => {
if let Some(player_info) = player_info_from_uid(*from) {
return Some(ChatMessage::new(chatmsg, ChatParties::Region(player_info)));
}
},
ChatType::World(from) => {
if let Some(player_info) = player_info_from_uid(*from) {
return Some(ChatMessage::new(chatmsg, ChatParties::World(player_info)));
}
},
ChatType::Say(from) => {
if let Some(player_info) = player_info_from_uid(*from) {
return Some(ChatMessage::new(chatmsg, ChatParties::Say(player_info)));
}
},
ChatType::Tell(from, to) => {
if let (Some(from_player_info), Some(to_player_info)) =
(player_info_from_uid(*from), player_info_from_uid(*to))
{
return Some(ChatMessage::new(
chatmsg,
ChatParties::Tell(from_player_info, to_player_info),
));
}
},
ChatType::Kill(kill_source, from) => {
let kill_source = match kill_source.clone() {
comp::chat::KillSource::Player(uid, t) => {
if let Some(player_info) = player_info_from_uid(uid) {
KillSource::Player(player_info, t)
} else {
return None;
}
},
comp::chat::KillSource::NonPlayer(str, t) => KillSource::NonPlayer(str, t),
comp::chat::KillSource::NonExistent(t) => KillSource::NonExistent(t),
comp::chat::KillSource::Environment(str) => KillSource::Environment(str),
comp::chat::KillSource::FallDamage => KillSource::FallDamage,
comp::chat::KillSource::Suicide => KillSource::Suicide,
comp::chat::KillSource::Other => KillSource::Other,
};
if let Some(player_info) = player_info_from_uid(*from) {
return Some(ChatMessage::new(
chatmsg,
ChatParties::Kill(kill_source, player_info),
));
}
},
ChatType::FactionMeta(s) => {
return Some(ChatMessage::new(
chatmsg,
ChatParties::FactionMeta(s.clone()),
));
},
ChatType::Faction(from, s) => {
if let Some(player_info) = player_info_from_uid(*from) {
return Some(ChatMessage::new(
chatmsg,
ChatParties::Faction(player_info, s.clone()),
));
}
},
ChatType::GroupMeta(g) => {
let members = group_members_from_group(g);
return Some(ChatMessage::new(chatmsg, ChatParties::GroupMeta(members)));
},
ChatType::Group(from, g) => {
let members = group_members_from_group(g);
if let Some(player_info) = player_info_from_uid(*from) {
return Some(ChatMessage::new(
chatmsg,
ChatParties::Group(player_info, members),
));
}
},
_ => (),
};
None
}
pub fn send(&self, msg: ChatMessage) {
let drop_older_than = msg.time.sub(self.keep_duration);
let mut messages = self.messages.blocking_lock();
while let Some(msg) = messages.front() && msg.time < drop_older_than {
messages.pop_front();
}
messages.push_back(msg);
const MAX_CACHE_MESSAGES: usize = 10_000; // in case we have a short spam of many many messages, we dont want to keep the capacity forever
if messages.capacity() > messages.len() + MAX_CACHE_MESSAGES {
let msg_count = messages.len();
tracing::debug!(?msg_count, "shrinking cache");
messages.shrink_to_fit();
}
}
}
impl ChatCache {
pub fn new(keep_duration: Duration) -> (Self, ChatExporter) {
let messages: Arc<Mutex<VecDeque<ChatMessage>>> = Default::default();
let messages_clone = Arc::clone(&messages);
let keep_duration = chrono::Duration::from_std(keep_duration).unwrap();
(Self { messages }, ChatExporter {
messages: messages_clone,
keep_duration,
})
}
}

33
server/src/lib.rs
View File

@ -8,6 +8,7 @@
pub mod automod;
mod character_creator;
pub mod chat;
pub mod chunk_generator;
mod chunk_serialize;
pub mod client;
@ -31,7 +32,6 @@ pub mod sys;
#[cfg(feature = "persistent_world")]
pub mod terrain_persistence;
#[cfg(not(feature = "worldgen"))] mod test_world;
mod web;
mod weather;
@ -104,7 +104,7 @@ use std::{
};
#[cfg(not(feature = "worldgen"))]
use test_world::{IndexOwned, World};
use tokio::{runtime::Runtime, sync::Notify};
use tokio::runtime::Runtime;
use tracing::{debug, error, info, trace, warn};
use vek::*;
pub use world::{civ::WorldCivStage, sim::WorldSimStage, WorldGenerateStage};
@ -124,7 +124,7 @@ use {
common_state::plugin::{memory_manager::EcsWorld, PluginMgr},
};
use crate::{persistence::character_loader::CharacterScreenResponseKind, web::ChatCache};
use crate::{chat::ChatCache, persistence::character_loader::CharacterScreenResponseKind};
use common::comp::Anchor;
#[cfg(feature = "worldgen")]
pub use world::{
@ -211,7 +211,8 @@ pub struct Server {
runtime: Arc<Runtime>,
metrics_shutdown: Arc<Notify>,
metrics_registry: Arc<Registry>,
chat_cache: ChatCache,
database_settings: Arc<RwLock<DatabaseSettings>>,
disconnect_all_clients_requested: bool,
@ -483,21 +484,8 @@ impl Server {
state.ecs_mut().insert(DeletedEntities::default());
let network = Network::new_with_registry(Pid::new(), &runtime, &registry);
let metrics_shutdown = Arc::new(Notify::new());
let metrics_shutdown_clone = Arc::clone(&metrics_shutdown);
let addr = settings.metrics_address;
let (chat_cache, chat_tracker) = ChatCache::new(Duration::from_secs(60));
state.ecs_mut().insert(chat_tracker);
runtime.spawn(async move {
web::run(
Arc::clone(&registry),
chat_cache,
"secretpassword".to_string(),
addr,
metrics_shutdown_clone.notified(),
)
.await
});
let mut printed_quic_warning = false;
for protocol in &settings.gameserver_protocols {
@ -603,7 +591,8 @@ impl Server {
connection_handler,
runtime,
metrics_shutdown,
metrics_registry: registry,
chat_cache,
database_settings,
disconnect_all_clients_requested: false,
@ -668,6 +657,12 @@ impl Server {
/// Get a reference to the server's world.
pub fn world(&self) -> &World { &self.world }
/// Get a reference to the Metrics Registry
pub fn metrics_registry(&self) -> &Arc<Registry> { &self.metrics_registry }
/// Get a reference to the Chat Cache
pub fn chat_cache(&self) -> &ChatCache { &self.chat_cache }
fn parse_locations(&self, character_list_data: &mut [CharacterItem]) {
character_list_data.iter_mut().for_each(|c| {
let name = c
@ -1497,8 +1492,6 @@ impl Server {
impl Drop for Server {
fn drop(&mut self) {
self.metrics_shutdown.notify_one();
self.state
.notify_players(ServerGeneral::Disconnect(DisconnectReason::Shutdown));

6
server/src/settings.rs
View File

@ -160,7 +160,6 @@ impl CalendarMode {
#[serde(default)]
pub struct Settings {
pub gameserver_protocols: Vec<Protocol>,
pub metrics_address: SocketAddr,
pub auth_server_address: Option<String>,
pub max_players: u16,
pub world_seed: u32,
@ -202,7 +201,6 @@ impl Default for Settings {
address: SocketAddr::from((Ipv4Addr::UNSPECIFIED, 14004)),
},
],
metrics_address: SocketAddr::from((Ipv4Addr::LOCALHOST, 14005)),
auth_server_address: Some("https://auth.veloren.net".into()),
world_seed: DEFAULT_WORLD_SEED,
server_name: "Veloren Server".into(),
@ -285,10 +283,6 @@ impl Settings {
pick_unused_port().expect("Failed to find unused port!"),
)),
}],
metrics_address: SocketAddr::from((
Ipv4Addr::LOCALHOST,
pick_unused_port().expect("Failed to find unused port!"),
)),
auth_server_address: None,
// If loading the default map file, make sure the seed is also default.
world_seed: if load.map_file.is_some() {

14
server/src/state_ext.rs
View File

@ -1,5 +1,6 @@
use crate::{
automod::AutoMod,
chat::ChatExporter,
client::Client,
events::{self, update_map_markers},
persistence::PersistedComponents,
@ -8,7 +9,6 @@ use crate::{
rtsim::RtSim,
settings::Settings,
sys::sentinel::DeletedEntities,
web::ChatExporter,
wiring, BattleModeBuffer, SpawnPoint,
};
use common::{
@ -905,6 +905,10 @@ impl StateExt for State {
let group_info = msg.get_group().and_then(|g| group_manager.group_info(*g));
if let Some(exported_message) = ChatExporter::generate(&msg, ecs) {
chat_exporter.send(exported_message);
}
let resolved_msg = msg
.clone()
.map_group(|_| group_info.map_or_else(|| "???".to_string(), |i| i.name.clone()));
@ -927,7 +931,6 @@ impl StateExt for State {
| comp::ChatType::CommandError
| comp::ChatType::Meta
| comp::ChatType::World(_) => {
chat_exporter.send(resolved_msg.clone());
self.notify_players(ServerGeneral::ChatMsg(resolved_msg))
},
comp::ChatType::Online(u) => {
@ -938,7 +941,6 @@ impl StateExt for State {
client.send_fallible(ServerGeneral::ChatMsg(resolved_msg.clone()));
}
}
chat_exporter.send(resolved_msg);
},
comp::ChatType::Tell(from, to) => {
for (client, uid) in
@ -948,12 +950,10 @@ impl StateExt for State {
client.send_fallible(ServerGeneral::ChatMsg(resolved_msg.clone()));
}
}
chat_exporter.send(resolved_msg);
},
comp::ChatType::Kill(kill_source, uid) => {
let clients = ecs.read_storage::<Client>();
let clients_count = clients.count();
chat_exporter.send(resolved_msg.clone());
// Avoid chat spam, send kill message only to group or nearby players if a
// certain amount of clients are online
if clients_count
@ -1008,7 +1008,6 @@ impl StateExt for State {
}
}
}
chat_exporter.send(resolved_msg);
},
comp::ChatType::Region(uid) => {
let entity_opt = entity_from_uid(*uid);
@ -1021,7 +1020,6 @@ impl StateExt for State {
}
}
}
chat_exporter.send(resolved_msg);
},
comp::ChatType::Npc(uid) => {
let entity_opt = entity_from_uid(*uid);
@ -1067,7 +1065,6 @@ impl StateExt for State {
client.send_fallible(ServerGeneral::ChatMsg(resolved_msg.clone()));
}
}
chat_exporter.send(resolved_msg);
},
comp::ChatType::Group(from, g) => {
if group_info.is_none() {
@ -1089,7 +1086,6 @@ impl StateExt for State {
} else {
send_to_group(g, ecs, &resolved_msg);
}
chat_exporter.send(resolved_msg);
},
comp::ChatType::GroupMeta(g) => {
send_to_group(g, ecs, &resolved_msg);