Merge pull request #376 from AdExNetwork/issue-319-event-aggregator-changes

Issue 319 event aggregator changes

Author: elpiel

sentry/src/analytics_recorder.rs

@@ -1,26 +1,26 @@
 use crate::epoch;
-use crate::payout::get_payout;
 use crate::Session;
 use primitives::sentry::Event;
 use primitives::sentry::{ChannelReport, PublisherReport};
-use primitives::{BigNum, Channel};
+use primitives::{BigNum, Channel, ValidatorId};
 use redis::aio::MultiplexedConnection;
 use redis::pipe;
 use slog::{error, Logger};
 
+// Records only payout events
 pub async fn record(
     mut conn: MultiplexedConnection,
     channel: Channel,
     session: Session,
-    events: Vec<Event>,
+    events: Vec<(Event, Option<(ValidatorId, BigNum)>)>,
     logger: Logger,
 ) {
     let mut db = pipe();
 
     events
         .iter()
-        .filter(|&ev| ev.is_click_event() || ev.is_impression_event())
-        .for_each(|event: &Event| match event {
+        .filter(|(ev, _)| ev.is_click_event() || ev.is_impression_event())
+        .for_each(|(event, payout)| match event {
             Event::Impression {
                 publisher,
                 ad_unit,
@@ -34,17 +34,13 @@ pub async fn record(
                 referrer,
             } => {
                 let divisor = BigNum::from(10u64.pow(18));
-
-                let pay_amount = match get_payout(&logger, &channel, event, &session) {
-                    Ok(Some((_, payout))) => payout.div_floor(&divisor)
+                let pay_amount = match payout {
+                    Some((_, payout)) => payout
+                        .div_floor(&divisor)
                         .to_f64()
                         .expect("Should always have a payout in f64 after division"),
                     // This should never happen, as the conditions we are checking for in the .filter are the same as getPayout's
-                    Ok(None) => return,
-                    Err(err) => {
-                        error!(&logger, "Getting the payout failed: {}", &err; "module" => "analytics-recorder", "err" => ?err);
-                        return
-                    },
+                    None => return,
                 };
 
                 if let Some(ad_unit) = ad_unit {

sentry/src/event_aggregator.rs

@@ -4,6 +4,7 @@ use crate::db::event_aggregate::insert_event_aggregate;
 use crate::db::DbPool;
 use crate::db::{get_channel_by_id, update_targeting_rules};
 use crate::event_reducer;
+use crate::payout::get_payout;
 use crate::Application;
 use crate::ResponseError;
 use crate::Session;
@@ -13,7 +14,7 @@ use chrono::Utc;
 use lazy_static::lazy_static;
 use primitives::adapter::Adapter;
 use primitives::sentry::{Event, EventAggregate};
-use primitives::{Channel, ChannelId};
+use primitives::{BigNum, Channel, ChannelId, ValidatorId};
 use slog::{error, Logger};
 use std::collections::HashMap;
 use std::env;
@@ -64,13 +65,13 @@ async fn store(db: &DbPool, channel_id: &ChannelId, logger: &Logger, recorder: R
 }
 
 impl EventAggregator {
-    pub async fn record<'a, A: Adapter>(
+    pub async fn record<A: Adapter>(
         &self,
-        app: &'a Application<A>,
+        app: &Application<A>,
         channel_id: &ChannelId,
         session: &Session,
         auth: Option<&Auth>,
-        events: &'a [Event],
+        events: Vec<Event>,
     ) -> Result<(), ResponseError> {
         let recorder = self.recorder.clone();
         let aggr_throttle = app.config.aggr_throttle;
@@ -97,7 +98,6 @@ impl EventAggregator {
                 // insert into
                 channel_recorder.insert(channel_id.to_owned(), record);
 
-                //
                 // spawn async task that persists
                 // the channel events to database
                 if aggr_throttle > 0 {
@@ -131,7 +131,7 @@ impl EventAggregator {
             auth,
             &app.config.ip_rate_limit,
             &record.channel,
-            events,
+            &events,
         )
         .await
         .map_err(|e| match e {
@@ -155,18 +155,29 @@ impl EventAggregator {
             update_targeting_rules(&app.pool, &channel_id, &new_rules).await?;
         }
 
-        events.iter().for_each(|ev| {
-            match event_reducer::reduce(
-                &app.logger,
-                &record.channel,
-                &mut record.aggregate,
-                ev,
-                &session,
-            ) {
-                Ok(_) => {}
-                Err(err) => error!(&app.logger, "Event Reducer failed"; "error" => ?err ),
-            }
-        });
+        // Pre-computing all payouts once
+        let events_with_payout: Vec<(Event, Option<(ValidatorId, BigNum)>)> = events
+            .iter()
+            .filter(|ev| ev.is_click_event() || ev.is_impression_event())
+            .map(|ev| {
+                let payout = match get_payout(&app.logger, &record.channel, &ev, &session) {
+                    Ok(payout) => payout,
+                    Err(err) => return Err(err),
+                };
+
+                match event_reducer::reduce(&record.channel, &mut record.aggregate, &ev, &payout) {
+                    Ok(_) => {}
+                    Err(err) => error!(&app.logger, "Event Reducred failed"; "error" => ?err),
+                }
+
+                Ok((ev.clone(), payout))
+            })
+            .collect::<Result<_, _>>()?;
+
+        // We don't want to save empty aggregates
+        if record.aggregate.events.is_empty() {
+            return Ok(());
+        }
 
         // only time we don't have session is during
         // an unauthenticated close event
@@ -175,7 +186,7 @@ impl EventAggregator {
                 redis.clone(),
                 record.channel.clone(),
                 session.clone(),
-                events.to_owned().to_vec(),
+                events_with_payout,
                 app.logger.clone(),
             ));
         }

sentry/src/event_reducer.rs

@@ -1,35 +1,35 @@
-use crate::{payout::get_payout, Session};
 use primitives::{
     sentry::{AggregateEvents, Event, EventAggregate},
     BigNum, Channel, ValidatorId,
 };
-use slog::Logger;
 
 pub(crate) fn reduce(
-    logger: &Logger,
     channel: &Channel,
     initial_aggr: &mut EventAggregate,
     ev: &Event,
-    session: &Session,
+    payout: &Option<(ValidatorId, BigNum)>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     let event_type = ev.to_string();
+
     match ev {
         Event::Impression { publisher, .. } => {
             let impression = initial_aggr.events.get(&event_type);
-            let payout = get_payout(logger, &channel, &ev, session)?;
             let merge = merge_payable_event(
                 impression,
-                payout.unwrap_or_else(|| (*publisher, Default::default())),
+                payout
+                    .to_owned()
+                    .unwrap_or_else(|| (*publisher, Default::default())),
             );
 
             initial_aggr.events.insert(event_type, merge);
         }
         Event::Click { publisher, .. } => {
             let clicks = initial_aggr.events.get(&event_type);
-            let payout = get_payout(logger, &channel, &ev, session)?;
             let merge = merge_payable_event(
                 clicks,
-                payout.unwrap_or_else(|| (*publisher, Default::default())),
+                payout
+                    .to_owned()
+                    .unwrap_or_else(|| (*publisher, Default::default())),
             );
 
             initial_aggr.events.insert(event_type, merge);
@@ -77,15 +77,13 @@ fn merge_payable_event(
 mod test {
     use super::*;
     use chrono::Utc;
-    use primitives::util::tests::{
-        discard_logger,
-        prep_db::{DUMMY_CHANNEL, IDS},
+    use primitives::{
+        util::tests::prep_db::{DUMMY_CHANNEL, IDS},
+        BigNum,
     };
-    use primitives::BigNum;
 
     #[test]
     fn test_reduce() {
-        let logger = discard_logger();
         let mut channel: Channel = DUMMY_CHANNEL.clone();
         channel.deposit_amount = 100.into();
         // make immutable again
@@ -103,16 +101,9 @@ mod test {
             ad_slot: None,
             referrer: None,
         };
-
-        let session = Session {
-            ip: Default::default(),
-            country: None,
-            referrer_header: None,
-            os: None,
-        };
-
+        let payout = Some((IDS["publisher"], BigNum::from(1)));
         for i in 0..101 {
-            reduce(&logger, &channel, &mut event_aggr, &event, &session)
+            reduce(&channel, &mut event_aggr, &event, &payout)
                 .expect(&format!("Should be able to reduce event #{}", i));
         }
 
@@ -132,9 +123,7 @@ mod test {
         assert_eq!(event_counts, &BigNum::from(101));
 
         let event_payouts = impression_event
-            .event_counts
-            .as_ref()
-            .expect("there should be event_counts set")
+            .event_payouts
             .get(&IDS["publisher"])
             .expect("There should be myAwesomePublisher event_payouts key");
         assert_eq!(event_payouts, &BigNum::from(101));

sentry/src/payout.rs

@@ -9,7 +9,7 @@ use primitives::{
 use slog::{error, Logger};
 use std::cmp::{max, min};
 
-type Result = std::result::Result<Option<(ValidatorId, BigNum)>, Error>;
+pub type Result = std::result::Result<Option<(ValidatorId, BigNum)>, Error>;
 
 pub fn get_payout(logger: &Logger, channel: &Channel, event: &Event, session: &Session) -> Result {
     let event_type = event.to_string();

sentry/src/routes/channel.rs

@@ -201,14 +201,14 @@ pub async fn insert_events<A: Adapter + 'static>(
     let channel_id = ChannelId::from_hex(route_params.index(0))?;
 
     let body_bytes = hyper::body::to_bytes(req_body).await?;
-    let request_body = serde_json::from_slice::<HashMap<String, Vec<Event>>>(&body_bytes)?;
+    let mut request_body = serde_json::from_slice::<HashMap<String, Vec<Event>>>(&body_bytes)?;
 
     let events = request_body
-        .get("events")
+        .remove("events")
         .ok_or_else(|| ResponseError::BadRequest("invalid request".to_string()))?;
 
     app.event_aggregator
-        .record(app, &channel_id, session, auth, &events)
+        .record(app, &channel_id, session, auth, events)
         .await?;
 
     Ok(Response::builder()