Cargo.toml

@@ -2,15 +2,20 @@
 name = "log2db"
 version = "0.1.0"
 authors = ["Yannik"]
-edition = "2021"
+edition = "2018"
 
 [dependencies]
 structopt = "*"
-tokio = {version="*", features=["rt", "macros"]}
+tokio = "0.2.0-alpha.6"
-tokio-util = {version="*", features=["codec"]}
-tokio-postgres = {version = "*", features =["with-chrono-0_4"]}
 chrono = "*"
-futures-util = "*"
+
+[dependencies.tokio-postgres]
+git = "https://github.com/sfackler/rust-postgres.git"
+features = ["with-chrono-0_4"]
+
+[dependencies.futures-preview]
+version = "=0.3.0-alpha.19"
+features = [ "async-await" ]
 
 [profile.release]
 lto = "fat"

readme.md

@@ -1,4 +0,0 @@
-nimmt syslog von antennen und tuts in postgress.
-für weniger glibc-versionsabhängingkeit bauen mit:
-
-    cargo build --release --target=x86_64-unknown-linux-musl

src/main.rs

@@ -1,8 +1,9 @@
 use std::str::FromStr;
 
-use tokio_postgres::{Client, Config, NoTls, Statement};
+use tokio_postgres::{Config, NoTls, Client, Statement};
 
 use tokio::net::{TcpListener, TcpStream};
+use tokio::prelude::*;
 
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -11,11 +12,9 @@ use structopt::StructOpt;
 
 use std::collections::HashSet;
 
-use futures_util::FutureExt;
-use futures_util::StreamExt;
-
 type Error = Box<dyn std::error::Error>;
 
+
 /// This application recieves logfiles in ubiquitis weird logfile format from the network,
 /// splits each line into its fields and writes them to a sql database for further analysis.
 ///
@@ -24,20 +23,21 @@ type Error = Box<dyn std::error::Error>;
 #[derive(StructOpt)]
 struct Options {
     /// ip and (TCP) port to bind to
-    #[structopt(short, long, default_value = "[::]:514")]
+    #[structopt(short, long, default_value="[::]:514")]
-    addr: String,
+    addr : String,
     /// space separated list of daemon/service names to not write to the db
-    #[structopt(default_value = "dnsmasq-dhcp")]
+    #[structopt(default_value="dnsmasq-dhcp")]
-    blacklist: Vec<String>,
+    blacklist: Vec<String>
 }
 
-#[tokio::main(flavor = "current_thread")]
+#[tokio::main]
 async fn main() -> Result<(), Error> {
     let options = Options::from_args();
 
     let postgresurl = std::env::var("LOG2DB_PSQL")
         .map_err(|_| "please supply the LOG2DB_PSQL environment variable, containting a postgress connection string")?;
 
+
     let mut cfg = Config::from_str(&postgresurl)?;
     cfg.application_name("log2db");
     let (client, connection) = cfg.connect(NoTls).await?;
@@ -49,71 +49,51 @@ async fn main() -> Result<(), Error> {
     });
     tokio::spawn(connection);
 
-    client
+    client.execute("create table if not exists log(
-        .execute(
+        prio smallint,
-            "create table if not exists log(
+        rcv_ip inet,
-                prio smallint,
+        rcv_date timestamptz,
-                rcv_ip inet,
+        date timestamptz,
-                rcv_date timestamptz,
+        daemon varchar,
-                date timestamptz,
+        message varchar
-                daemon varchar,
-                message varchar
-            )
-            ",
-            &[],
         )
-        .await?;
+        ", &[]).await?;
 
     let client = Arc::new(client);
 
     let insert_statement = client
         .prepare("insert into log(prio, rcv_ip, rcv_date, date, daemon, message) values ($1, $2, $3, $4, $5, $6)").await?;
 
-    let listener = TcpListener::bind(&options.addr)
+    let mut listener = TcpListener::bind(&options.addr).await
-        .await
         .map_err(|e| format!("could not bind to {} with error {:?}", &options.addr, e))?;
 
     let blacklist: HashSet<_> = options.blacklist.into_iter().collect();
     let blacklist = Arc::new(blacklist);
 
+
     loop {
-        match listener.accept().await {
+        match listener.accept().await  {
             Ok((socket, peer)) => {
-                tokio::spawn(handle_peer_and_error(
+                tokio::spawn(
-                    socket,
+                    handle_peer_and_error(socket, peer, client.clone(), insert_statement.clone(), blacklist.clone())
-                    peer,
+                    );
-                    client.clone(),
+            },
-                    insert_statement.clone(),
-                    blacklist.clone(),
-                ));
-            }
             Err(e) => eprintln!("{:?}", e),
         }
     }
 }
 
-use chrono::{DateTime, FixedOffset, Local};
+use chrono::{Local, DateTime, FixedOffset};
 
-async fn handle_peer_and_error(
+async fn handle_peer_and_error(stream: TcpStream, peer: SocketAddr, db: Arc<Client>, insert_statement: Statement, blacklist: Arc<HashSet<String>>) {
-    stream: TcpStream,
-    peer: SocketAddr,
-    db: Arc<Client>,
-    insert_statement: Statement,
-    blacklist: Arc<HashSet<String>>,
-) {
     if let Err(e) = handle_peer(stream, peer, db, insert_statement, blacklist).await {
         eprintln!("{}", e);
     }
 }
 
-async fn handle_peer(
+async fn handle_peer(stream: TcpStream, peer: SocketAddr, db: Arc<Client>, insert_statement: Statement, blacklist: Arc<HashSet<String>>) 
-    stream: TcpStream,
+-> Result<(), Error> {
-    peer: SocketAddr,
+    use tokio::codec::{FramedRead, LinesCodec};
-    db: Arc<Client>,
-    insert_statement: Statement,
-    blacklist: Arc<HashSet<String>>,
-) -> Result<(), Error> {
-    use tokio_util::codec::{FramedRead, LinesCodec};
 
     let ip = peer.ip();
 
@@ -123,13 +103,9 @@ async fn handle_peer(
             Some(line) => {
                 let (prio, now, date, service, log) = parse_line(&line)?;
                 if !blacklist.contains(service) {
-                    db.execute(
+                    db.execute(&insert_statement, &[&prio, &ip, &now, &date, &service, &log]).await?;
-                        &insert_statement,
-                        &[&prio, &ip, &now, &date, &service, &log],
-                    )
-                    .await?;
                 }
-            }
+            },
             None => break,
         }
     }
@@ -144,30 +120,15 @@ async fn handle_peer(
  * , log entry
  * )
  */
-fn parse_line(
+fn parse_line(line: &'_ str) -> Result<(i16, DateTime<Local>, DateTime<FixedOffset>, &'_ str, &'_ str), Error> {
-    line: &'_ str,
-) -> Result<
-    (
-        i16,
-        DateTime<Local>,
-        DateTime<FixedOffset>,
-        &'_ str,
-        &'_ str,
-    ),
-    Error,
-> {
     let mut prio_and_remainder = line.splitn(2, '>');
-    let prio = prio_and_remainder
+    let prio = prio_and_remainder.next().ok_or("log did not contain priority")?;
-        .next()
-        .ok_or("log did not contain priority")?;
     let prio = &prio[1..];
-    let prio = prio
+    let prio = prio.parse()
-        .parse()
+        .map_err(|e| format!("could not parse priority {}: {}",prio, e))?;
-        .map_err(|e| format!("could not parse priority {}: {}", prio, e))?;
 
-    let line = prio_and_remainder
+
-        .next()
+    let line = prio_and_remainder.next().expect("splitn should always return a second part");
-        .expect("splitn should always return a second part");
     let (date, line) = line.split_at(16);
 
     // we need to prepend the current year and timezone, as that is not stated in the logfile
@@ -175,17 +136,14 @@ fn parse_line(
     let mut base = format!("{}", now.format("%Y %z "));
     base.push_str(date);
 
-    let date = DateTime::parse_from_str(&base, "%Y %z %b %e %H:%M:%S ")
+    let date = DateTime::parse_from_str(&base, "%Y %z %b %e %H:%M:%S ").
-        .map_err(|e| format!("could not parse {}{} {}", date, line, e))?;
+        map_err(|e| format!("could not parse {}{} {}", date, line, e))?;
 
     let mut parts = line.splitn(2, ':');
 
     let service_and_pid = parts.next().ok_or("could not parse service")?;
     let mut service_parts = service_and_pid.splitn(2, '[');
-    let service = service_parts
+    let service = service_parts.next().ok_or("could not split pid from service")?.trim();
-        .next()
-        .ok_or("could not split pid from service")?
-        .trim();
 
     let log = parts.next().ok_or("could not parse logfile")?.trim();
     Ok((prio, now, date, service, log))