add: validation for analytics rules and events (#1927)

* add: validation for analytics rules

* fix: tests for the same

* add: check for log_to_store and mandatory collection in rule for log type

* fix: missing of events

* fix: status codes in event manager

* fix: validation condition and tests for the same

include/analytics_manager.h

@@ -10,7 +10,8 @@
 
 struct event_type_collection {
     std::string event_type;
-    std::string collection;
+    std::string destination_collection;
+    std::vector<std::string> src_collections;
     bool log_to_store = false;
     std::string analytic_rule;
     QueryAnalytics* queries_ptr = nullptr;
@@ -90,8 +91,8 @@ private:
 
     struct suggestion_config_t {
         std::string name;
-        std::string suggestion_collection;
-        std::vector<std::string> query_collections;
+        std::string destination_collection;
+        std::vector<std::string> src_collections;
         size_t limit;
         std::string rule_type;
         bool expand_query = false;
@@ -103,8 +104,8 @@ private:
             obj["type"] = rule_type;
             obj["params"] = nlohmann::json::object();
             obj["params"]["limit"] = limit;
-            obj["params"]["source"]["collections"] = query_collections;
-            obj["params"]["destination"]["collection"] = suggestion_collection;
+            obj["params"]["source"]["collections"] = src_collections;
+            obj["params"]["destination"]["collection"] = destination_collection;
 
             if(rule_type == POPULAR_QUERIES_TYPE) {
                 obj["params"]["expand_query"] = expand_query;

src/analytics_manager.cpp

@@ -59,7 +59,8 @@ Option<bool> AnalyticsManager::create_index(nlohmann::json &payload, bool upsert
     }
 
     std::string counter_field;
-    std::string suggestion_collection = "generic";
+    std::string destination_collection;
+    std::vector<std::string> src_collections;
 
     suggestion_config_t suggestion_config;
     suggestion_config.name = suggestion_config_name;
@@ -67,25 +68,23 @@ Option<bool> AnalyticsManager::create_index(nlohmann::json &payload, bool upsert
     suggestion_config.expand_query = expand_query;
     suggestion_config.rule_type = payload["type"];
 
-    //for counter events source collections are not needed
-    if(params["source"].contains("collections")) {
-        if(!params["source"]["collections"].is_array()) {
-            return Option<bool>(400, "Must contain a valid list of source collections.");
-        }
-
+    //for all types source collection is needed.
+    if(!params["source"].contains("collections") || !params["source"]["collections"].is_array()) {
+        return Option<bool>(400, "Must contain a valid list of source collections.");
+    } else {
         for(const auto& coll: params["source"]["collections"]) {
             if (!coll.is_string()) {
-                return Option<bool>(400, "Must contain a valid list of source collection names.");
+                return Option<bool>(400, "Source collections value should be a string.");
+            }
+            auto collection = CollectionManager::get_instance().get_collection(coll.get<std::string>());
+            if (collection == nullptr) {
+                return Option<bool>(404, "Collection `" + coll.get<std::string>() + "` is not found");
             }
 
             const std::string &src_collection = coll.get<std::string>();
-            suggestion_config.query_collections.push_back(src_collection);
-
-            suggestion_collection = src_collection;
+            src_collections.push_back(src_collection);
+            destination_collection = src_collection;
         }
-    } else if(payload["type"] == POPULAR_QUERIES_TYPE || payload["type"] == NOHITS_QUERIES_TYPE) {
-        //for popular and nohits queries, source collection is mandatory
-        return Option<bool>(400, "Must contain a valid list of source collections.");
     }
 
     if((payload["type"] == POPULAR_QUERIES_TYPE || payload["type"] == NOHITS_QUERIES_TYPE)
@@ -126,30 +125,30 @@ Option<bool> AnalyticsManager::create_index(nlohmann::json &payload, bool upsert
             suggestion_config.counter_field = counter_field;
         }
 
-        suggestion_collection = params["destination"]["collection"].get<std::string>();
+        destination_collection = params["destination"]["collection"].get<std::string>();
     }
 
     if(payload["type"] == POPULAR_QUERIES_TYPE) {
-        if(!upsert && popular_queries.count(suggestion_collection) != 0) {
+        if(!upsert && popular_queries.count(destination_collection) != 0) {
             return Option<bool>(400, "There's already another configuration for this destination collection.");
         }
     } else if(payload["type"] == NOHITS_QUERIES_TYPE) {
-        if(!upsert && nohits_queries.count(suggestion_collection) != 0) {
+        if(!upsert && nohits_queries.count(destination_collection) != 0) {
             return Option<bool>(400, "There's already another configuration for this destination collection.");
         }
     } else if(payload["type"] == COUNTER_TYPE) {
-        if(!upsert && counter_events.count(suggestion_collection) != 0) {
+        if(!upsert && counter_events.count(destination_collection) != 0) {
             return Option<bool>(400, "There's already another configuration for this destination collection.");
         }
 
-        auto coll = CollectionManager::get_instance().get_collection(suggestion_collection).get();
+        auto coll = CollectionManager::get_instance().get_collection(destination_collection).get();
         if(coll != nullptr) {
             if(!coll->contains_field(counter_field)) {
                 return Option<bool>(404,
                                     "counter_field `" + counter_field + "` not found in destination collection.");
             }
         } else {
-            return Option<bool>(404, "Collection `" + suggestion_collection + "` not found.");
+            return Option<bool>(404, "Collection `" + destination_collection + "` not found.");
         }
     }
 
@@ -163,21 +162,28 @@ Option<bool> AnalyticsManager::create_index(nlohmann::json &payload, bool upsert
         }
     }
 
-    if(query_collection_events.count(suggestion_collection) == 0) {
+    if(query_collection_events.count(destination_collection) == 0) {
         std::vector<event_t> vec;
-        query_collection_events.emplace(suggestion_collection, vec);
+        query_collection_events.emplace(destination_collection, vec);
     }
 
     std::map<std::string, uint16_t> event_weight_map;
-    bool log_to_store = payload["type"] == LOG_TYPE ? true : false;
+    bool log_to_store = payload["type"] == LOG_TYPE;
+
+    for (const std::string coll: src_collections) {
+        if(query_collection_events.count(coll) == 0) {
+            std::vector<event_t> vec;
+            query_collection_events.emplace(coll, vec);
+        }
+    }
 
     if(payload["type"] == POPULAR_QUERIES_TYPE) {
         QueryAnalytics* popularQueries = new QueryAnalytics(limit, enable_auto_aggregation);
         popularQueries->set_expand_query(suggestion_config.expand_query);
-        popular_queries.emplace(suggestion_collection, popularQueries);
+        popular_queries.emplace(destination_collection, popularQueries);
     } else if(payload["type"] == NOHITS_QUERIES_TYPE) {
         QueryAnalytics* noresultsQueries = new QueryAnalytics(limit, enable_auto_aggregation);
-        nohits_queries.emplace(suggestion_collection, noresultsQueries);
+        nohits_queries.emplace(destination_collection, noresultsQueries);
     }
 
     if(valid_events_found) {
@@ -186,30 +192,28 @@ Option<bool> AnalyticsManager::create_index(nlohmann::json &payload, bool upsert
                 return Option<bool>(400, "Events must contain a unique name.");
             }
 
+            bool event_log_to_store = false;
             if(payload["type"] == COUNTER_TYPE) {
                 if(!event.contains("weight") || !event["weight"].is_number()) {
                     return Option<bool>(400, "Counter events must contain a weight value.");
                 }
-
-                //store event name to their weights
-                //which can be used to keep counter events separate from non counter events
-                if(event.contains("log_to_store")) {
-                    log_to_store = event["log_to_store"].get<bool>();
-
-                    if(log_to_store && !analytics_store) {
-                        return Option<bool>(400, "Event can't be logged when analytics-db is not defined.");
-                    }
-                }
                 event_weight_map[event["name"]] = event["weight"];
             }
 
-            event_type_collection ec{event["type"], suggestion_collection, log_to_store, suggestion_config_name};
+            if(event.contains("log_to_store")) {
+                event_log_to_store = event["log_to_store"].get<bool>();
+                if(event_log_to_store && !analytics_store) {
+                    return Option<bool>(400, "Event can't be logged when analytics-db is not defined.");
+                }
+            }
+
+            event_type_collection ec{event["type"], destination_collection, src_collections, event_log_to_store || log_to_store, suggestion_config_name};
 
             //keep pointer for /events API
             if(payload["type"] == POPULAR_QUERIES_TYPE) {
-                ec.queries_ptr = popular_queries.at(suggestion_collection);
+                ec.queries_ptr = popular_queries.at(destination_collection);
             } else if(payload["type"] == NOHITS_QUERIES_TYPE) {
-                ec.queries_ptr = nohits_queries.at(suggestion_collection);
+                ec.queries_ptr = nohits_queries.at(destination_collection);
             }
 
             event_collection_map.emplace(event["name"], ec);
@@ -217,16 +221,17 @@ Option<bool> AnalyticsManager::create_index(nlohmann::json &payload, bool upsert
 
         //store counter events data
         if(payload["type"] == COUNTER_TYPE) {
-            counter_events.emplace(suggestion_collection, counter_event_t{counter_field, {}, event_weight_map});
+            counter_events.emplace(destination_collection, counter_event_t{counter_field, {}, event_weight_map});
         }
     }
 
-    suggestion_config.suggestion_collection = suggestion_collection;
+    suggestion_config.destination_collection = destination_collection;
+    suggestion_config.src_collections = src_collections;
 
     suggestion_configs.emplace(suggestion_config_name, suggestion_config);
 
-    for(const auto& query_coll: suggestion_config.query_collections) {
-        query_collection_mapping[query_coll].push_back(suggestion_collection);
+    for(const auto& query_coll: suggestion_config.src_collections) {
+        query_collection_mapping[query_coll].push_back(destination_collection);
     }
 
     if(write_to_disk) {
@@ -315,9 +320,9 @@ Option<bool> AnalyticsManager::remove_index(const std::string &name) {
         return Option<bool>(404, "Rule not found.");
     }
 
-    const auto& suggestion_collection = suggestion_configs_it->second.suggestion_collection;
+    const auto& suggestion_collection = suggestion_configs_it->second.destination_collection;
 
-    for(const auto& query_collection: suggestion_configs_it->second.query_collections) {
+    for(const auto& query_collection: suggestion_configs_it->second.src_collections) {
         query_collection_mapping.erase(query_collection);
     }
 
@@ -388,9 +393,22 @@ Option<bool> AnalyticsManager::add_event(const std::string& client_ip, const std
         return Option<bool>(400, "event_type mismatch in analytic rules.");
     }
 
-    const auto& query_collection = event_collection_map_it->second.collection;
+    std::string destination_collection = event_collection_map_it->second.destination_collection;
+    std::vector<std::string> src_collections = event_collection_map_it->second.src_collections;
 
-    const auto& query_collection_events_it = query_collection_events.find(query_collection);
+    std::string src_collection;
+    if (!event_json.contains("collection") && src_collections.size() == 1) {
+        src_collection = src_collections[0];
+    } else if(!event_json.contains("collection") && src_collections.size() > 1) {
+        return Option<bool>(400, "Multiple source collections. 'collection' should be specified");
+    } else if (event_json.contains("collection")) {
+        if(std::find(src_collections.begin(), src_collections.end(), event_json["collection"]) == src_collections.end()) {
+            return Option<bool>(400, event_json["collection"].get<std::string>() + " not found in the rule " + event_name);
+        }
+        src_collection = event_json["collection"];
+    }
+
+    const auto& query_collection_events_it = query_collection_events.find(src_collection);
     if(query_collection_events_it != query_collection_events.end()) {
         auto &events_vec = query_collection_events_it->second;
 #ifdef TEST_BUILD
@@ -467,7 +485,7 @@ Option<bool> AnalyticsManager::add_event(const std::string& client_ip, const std
         }
 
         if (!counter_events.empty()) {
-            auto counter_events_it = counter_events.find(query_collection);
+            auto counter_events_it = counter_events.find(destination_collection);
             if (counter_events_it != counter_events.end()) {
                 auto event_weight_map_it = counter_events_it->second.event_weight_map.find(event_name);
                 if (event_weight_map_it != counter_events_it->second.event_weight_map.end()) {
@@ -478,9 +496,11 @@ Option<bool> AnalyticsManager::add_event(const std::string& client_ip, const std
                                << " not defined in analytic rule for counter events.";
                 }
             } else {
-                LOG(ERROR) << "collection " << query_collection << " not found in analytics rule.";
+                LOG(ERROR) << "collection " << destination_collection << " not found in analytics rule.";
             }
         }
+    } else {
+        return Option<bool>(500, "Failure in adding an event.");
     }
     return Option<bool>(true);
 }
@@ -581,7 +601,7 @@ void AnalyticsManager::persist_query_events(ReplicationState *raft_server, uint6
 
     for(const auto& suggestion_config: suggestion_configs) {
         const std::string& sink_name = suggestion_config.first;
-        const std::string& suggestion_coll = suggestion_config.second.suggestion_collection;
+        const std::string& suggestion_coll = suggestion_config.second.destination_collection;
 
         auto popular_queries_it = popular_queries.find(suggestion_coll);
         auto nohits_queries_it = nohits_queries.find(suggestion_coll);

src/event_manager.cpp

@@ -37,30 +37,34 @@ Option<bool> EventManager::add_event(const nlohmann::json& event, const std::str
             }
             const auto& event_name = event[EVENT_NAME];
             if(!event_data_val.is_object()) {
-                return Option<bool>(500, "event_data_val is not object.");
+                return Option<bool>(400, "data is not object.");
             }
 
             if(event_type == AnalyticsManager::SEARCH_EVENT) {
                 if(!event_data_val.contains("user_id") || !event_data_val["user_id"].is_string()) {
-                    return Option<bool>(500,
+                    return Option<bool>(400,
                                         "search event json data fields should contain `user_id` as string value.");
                 }
 
                 if(!event_data_val.contains("q") || !event_data_val["q"].is_string()) {
-                    return Option<bool>(500,
+                    return Option<bool>(400,
                                         "search event json data fields should contain `q` as string value.");
                 }
             } else {
                 if(!event_data_val.contains("doc_id") || !event_data_val["doc_id"].is_string()) {
-                    return Option<bool>(500, "event should have 'doc_id' as string value.");
+                    return Option<bool>(400, "event should have 'doc_id' as string value.");
                 }
 
-                if(event_data_val.contains("user_id") && !event_data_val["user_id"].is_string()) {
-                    return Option<bool>(500, "'user_id' should be a string value.");
+                if(event_data_val.contains("collection") && !event_data_val["collection"].is_string()) {
+                    return Option<bool>(400, "'collection' should be a  string value.");
+                }
+
+                if(!event_data_val.contains("user_id") || !event_data_val["user_id"].is_string()) {
+                    return Option<bool>(400, "event should have 'user_id' as string value.");
                 }
 
                 if(event_data_val.contains("q") && !event_data_val["q"].is_string()) {
-                    return Option<bool>(500, "'q' should be a string value.");
+                    return Option<bool>(400, "'q' should be a string value.");
                 }
             }
 
@@ -72,7 +76,7 @@ Option<bool> EventManager::add_event(const nlohmann::json& event, const std::str
             return Option<bool>(404, "event_type " + event_type + " not found.");
         }
     } else {
-        return Option<bool>(500, "`event_type` value should be string.");
+        return Option<bool>(400, "`event_type` value should be string.");
     }
 
     return Option(true);

src/typesense_server_utils.cpp

@@ -370,6 +370,12 @@ int run_server(const Config & config, const std::string & version, void (*master
         return 1;
     }
 
+    if (config.get_enable_search_analytics() && !config.get_analytics_dir().empty() && !directory_exists(config.get_analytics_dir())) {
+        LOG(ERROR) << "Typesense failed to start. " << "Analytics directory " << config.get_analytics_dir()
+                 << " does not exist.";
+        return 1;
+    }
+
     if(!config.get_master().empty()) {
         LOG(ERROR) << "The --master option has been deprecated. Please use clustering for high availability. "
                    << "Look for the --nodes configuration in the documentation.";

test/analytics_manager_test.cpp

@@ -172,6 +172,19 @@ TEST_F(AnalyticsManagerTest, AddSuggestionWithExpandedQuery) {
 }
 
 TEST_F(AnalyticsManagerTest, GetAndDeleteSuggestions) {
+    nlohmann::json titles_schema = R"({
+            "name": "titles",
+            "fields": [
+                {"name": "title", "type": "string"}
+            ]
+        })"_json;
+
+    Collection* titles_coll = collectionManager.create_collection(titles_schema).get();
+
+    nlohmann::json doc;
+    doc["title"] = "Cool trousers";
+    ASSERT_TRUE(titles_coll->add(doc.dump()).ok());
+
     nlohmann::json analytics_rule = R"({
         "name": "top_search_queries",
         "type": "popular_queries",
@@ -223,7 +236,7 @@ TEST_F(AnalyticsManagerTest, GetAndDeleteSuggestions) {
 
     create_op = analyticsManager.create_rule(analytics_rule, false, true);
     ASSERT_FALSE(create_op.ok());
-    ASSERT_EQ("Must contain a valid list of source collection names.", create_op.error());
+    ASSERT_EQ("Source collections value should be a string.", create_op.error());
 
     analytics_rule = R"({
         "name": "top_search_queries2",
@@ -296,6 +309,15 @@ TEST_F(AnalyticsManagerTest, EventsValidation) {
 
     Collection* titles_coll = collectionManager.create_collection(titles_schema).get();
 
+    nlohmann::json titles1_schema = R"({
+            "name": "titles_1",
+            "fields": [
+                {"name": "title", "type": "string"}
+            ]
+        })"_json;
+
+    Collection* titles1_coll = collectionManager.create_collection(titles1_schema).get();
+
     std::shared_ptr<http_req> req = std::make_shared<http_req>();
     std::shared_ptr<http_res> res = std::make_shared<http_res>(nullptr);
 
@@ -370,7 +392,7 @@ TEST_F(AnalyticsManagerTest, EventsValidation) {
 
     req->body = event3.dump();
     ASSERT_FALSE(post_create_event(req, res));
-    ASSERT_EQ("{\"message\": \"'user_id' should be a string value.\"}", res->body);
+    ASSERT_EQ("{\"message\": \"event should have 'user_id' as string value.\"}", res->body);
 
     event3 = R"({
         "type": "conversion",
@@ -504,12 +526,13 @@ TEST_F(AnalyticsManagerTest, EventsValidation) {
     create_op = analyticsManager.create_rule(analytics_rule, true, true);
     ASSERT_TRUE(create_op.ok());
 
-    // log based event should be created with only doc_id
+    // log based event should be created with only doc_id and user_id
     event5 = R"({
         "type": "click",
         "name": "AP",
         "data": {
-            "doc_id": "21"
+            "doc_id": "21",
+            "user_id": "123"
         }
     })"_json;
 
@@ -597,7 +620,7 @@ TEST_F(AnalyticsManagerTest, EventsValidation) {
     req->body = event9.dump();
     ASSERT_TRUE(post_create_event(req, res));
 
-    //for log events source collections is optional
+    //for log events source collections is not optional
     req->params["name"] = "product_events2";
     ASSERT_TRUE(del_analytics_rules(req, res));
     analytics_rule = R"({
@@ -611,7 +634,8 @@ TEST_F(AnalyticsManagerTest, EventsValidation) {
     })"_json;
 
     create_op = analyticsManager.create_rule(analytics_rule, true, true);
-    ASSERT_TRUE(create_op.ok());
+    ASSERT_FALSE(create_op.ok());
+    ASSERT_EQ("Must contain a valid list of source collections.", create_op.error());
 
     //try adding removed events
     ASSERT_TRUE(analyticsManager.remove_rule("product_events").ok());
@@ -629,6 +653,44 @@ TEST_F(AnalyticsManagerTest, EventsValidation) {
 
     create_op = analyticsManager.create_rule(analytics_rule, false, true);
     ASSERT_TRUE(create_op.ok());
+
+    analytics_rule = R"({
+        "name": "product_events2",
+        "type": "log",
+        "params": {
+            "source": {
+                 "collections": ["titles", "titles_1"],
+                 "events":  [{"type": "click", "name": "CP"}]
+            }
+        }
+    })"_json;
+
+    create_op = analyticsManager.create_rule(analytics_rule, true, true);
+    ASSERT_TRUE(create_op.ok());
+
+    event9 = R"({
+        "type": "click",
+        "name": "CP",
+        "data": {
+            "doc_id": "12",
+            "user_id": "11"
+        }
+    })"_json;
+    req->body = event9.dump();
+    ASSERT_FALSE(post_create_event(req, res));
+    ASSERT_EQ("{\"message\": \"Multiple source collections. 'collection' should be specified\"}", res->body);
+
+    event9 = R"({
+        "type": "click",
+        "name": "CP",
+        "data": {
+            "doc_id": "12",
+            "user_id": "11",
+            "collection": "titles"
+        }
+    })"_json;
+    req->body = event9.dump();
+    ASSERT_TRUE(post_create_event(req, res));
 }
 
 TEST_F(AnalyticsManagerTest, EventsPersist) {
@@ -748,60 +810,6 @@ TEST_F(AnalyticsManagerTest, EventsPersist) {
     ASSERT_EQ("13", parsed_json["user_id"]);
     ASSERT_EQ("21", parsed_json["doc_id"]);
     ASSERT_EQ("technology", parsed_json["query"]);
-
-    //create rule without source collections
-    analytics_rule = R"({
-        "name": "product_click_events2",
-        "type": "log",
-        "params": {
-            "source": {
-                 "events":  [{"type": "click", "name": "APCT"}]
-            }
-        }
-    })"_json;
-
-    create_op = analyticsManager.create_rule(analytics_rule, true, true);
-    ASSERT_TRUE(create_op.ok());
-
-    event = R"({
-        "type": "click",
-        "name": "APCT",
-        "data": {
-            "q": "technology",
-            "doc_id": "10",
-            "user_id": "1"
-        }
-    })"_json;
-
-    req->body = event.dump();
-    ASSERT_TRUE(post_create_event(req, res));
-
-    //get events
-    payload.clear();
-    collection_events_map = analyticsManager.get_log_events();
-    for (auto &events_collection_it: collection_events_map) {
-        const auto& collection = events_collection_it.first;
-        for(const auto& event: events_collection_it.second) {
-            event.to_json(event_data, collection);
-            payload.push_back(event_data);
-        }
-    }
-
-    //manually trigger write to db
-    ASSERT_TRUE(analyticsManager.write_to_db(payload));
-
-    values.clear();
-    analyticsManager.get_last_N_events("1", "*", 5, values);
-    ASSERT_EQ(1, values.size());
-
-    parsed_json = nlohmann::json::parse(values[0]);
-
-    //events will be fetched in LIFO order
-    ASSERT_EQ("APCT", parsed_json["name"]);
-    ASSERT_EQ("generic", parsed_json["collection"]); //without source collections events are classified into generic collection
-    ASSERT_EQ("1", parsed_json["user_id"]);
-    ASSERT_EQ("10", parsed_json["doc_id"]);
-    ASSERT_EQ("technology", parsed_json["query"]);
 }
 
 TEST_F(AnalyticsManagerTest, EventsRateLimitTest) {
@@ -1158,8 +1166,11 @@ TEST_F(AnalyticsManagerTest, PopularityScore) {
         "type": "counter",
         "params": {
             "source": {
-                "events":  [{"type": "click", "weight": 1, "name": "CLK1"}, {"type": "conversion", "weight": 5, "name": "CNV1"} ],
-                "log_to_store": true
+                "collections": ["products"],
+                "events":  [
+                    {"type": "click", "weight": 1, "name": "CLK1", "log_to_store": true},
+                    {"type": "conversion", "weight": 5, "name": "CNV1", "log_to_store": true}
+                ]
             },
             "destination": {
                 "collection": "products",
@@ -1287,22 +1298,6 @@ TEST_F(AnalyticsManagerTest, PopularityScore) {
     ASSERT_EQ(1, popular_clicks.size());
     ASSERT_EQ("popularity", popular_clicks["products"].counter_field);
     ASSERT_EQ(1, popular_clicks["products"].docid_counts.size());
-
-    //add with only doc_id
-    event5 = R"({
-        "type": "conversion",
-        "name": "CNV1",
-        "data": {
-            "doc_id": "5"
-        }
-    })"_json;
-    req->body = event5.dump();
-    ASSERT_TRUE(post_create_event(req, res));
-
-    popular_clicks = analyticsManager.get_popular_clicks();
-    ASSERT_EQ(1, popular_clicks.size());
-    ASSERT_EQ("popularity", popular_clicks["products"].counter_field);
-    ASSERT_EQ(2, popular_clicks["products"].docid_counts.size());
 }
 
 TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
@@ -1337,6 +1332,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                "collections": ["books"],
                 "events":  [{"type": "click", "weight": 1, "name": "CLK2"}, {"type": "conversion", "weight": 5, "name": "CNV2"} ]
             },
             "destination": {
@@ -1355,6 +1351,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                "collections": ["books"],
                 "events":  [{"type": "click", "weight": 1, "name": "CLK3"}, {"type": "conversion", "weight": 5, "name": "CNV3"} ]
             },
             "destination": {
@@ -1392,6 +1389,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                "collections": ["books"]
             },
             "destination": {
                 "collection": "books",
@@ -1409,6 +1407,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                "collections": ["books"],
                 "events":  []
             },
             "destination": {
@@ -1427,6 +1426,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                "collections": ["books"],
                 "events":  "query_click"
             },
             "destination": {
@@ -1445,6 +1445,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                "collections": ["books"],
                 "events":  [{"type": "click", "weight": 1}, {"type": "conversion", "weight": 5} ]
             },
             "destination": {
@@ -1467,6 +1468,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                 "collections": ["books"],
                  "events":  [{"type": "click", "name" : "CLK4"}, {"type": "conversion", "name": "CNV4", "log_to_store" : true} ]
             },
             "destination": {
@@ -1485,6 +1487,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                 "collections": ["books"],
                  "events":  [{"type": "click", "weight": 1, "name" : "CLK4"}, {"type": "conversion", "weight": 5, "name": "CNV4", "log_to_store" : true} ]
             },
             "destination": {
@@ -1520,7 +1523,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
     ASSERT_EQ(1, popular_clicks["books"].docid_counts.size());
     ASSERT_EQ(5, popular_clicks["books"].docid_counts["1"]);
 
-    //trigger persistance event manually
+    //trigger persistence event manually
     for (auto &popular_clicks_it: popular_clicks) {
         std::string docs;
         req->params["collection"] = popular_clicks_it.first;
@@ -1678,6 +1681,7 @@ TEST_F(AnalyticsManagerTest, PopularityScoreValidation) {
         "type": "counter",
         "params": {
             "source": {
+                 "collections": ["books"],
                  "events":  [{"type": "conversion", "weight": 5, "name": "CNV4"} ]
             },
             "destination": {
@@ -1718,6 +1722,15 @@ TEST_F(AnalyticsManagerTest, AnalyticsStoreTTL) {
     LOG(INFO) << "Truncating and creating: " << analytics_dir_path;
     system(("rm -rf "+ analytics_dir_path +" && mkdir -p "+analytics_dir_path).c_str());
 
+    nlohmann::json titles_schema = R"({
+            "name": "titles",
+            "fields": [
+                {"name": "title", "type": "string"}
+            ]
+        })"_json;
+
+    Collection* titles_coll = collectionManager.create_collection(titles_schema).get();
+
     analytic_store = new Store(analytics_dir_path, 24*60*60, 1024, true, FOURWEEKS_SECS);
     analyticsManager.init(store, analytic_store, analytics_minute_rate_limit);
 
@@ -1800,6 +1813,15 @@ TEST_F(AnalyticsManagerTest, AnalyticsStoreGetLastN) {
     analytic_store = new Store(analytics_dir_path, 24*60*60, 1024, true, FOURWEEKS_SECS);
     analyticsManager.init(store, analytic_store, analytics_minute_rate_limit);
 
+    nlohmann::json titles_schema = R"({
+            "name": "titles",
+            "fields": [
+                {"name": "title", "type": "string"}
+            ]
+        })"_json;
+
+    Collection* titles_coll = collectionManager.create_collection(titles_schema).get();
+
     auto analytics_rule = R"({
         "name": "product_events2",
         "type": "log",
@@ -2015,7 +2037,7 @@ TEST_F(AnalyticsManagerTest, AnalyticsStoreGetLastN) {
     }
 }
 
-TEST_F(AnalyticsManagerTest, AnalyticsWithAliases) {
+TEST_F(AnalyticsManagerTest, DISABLED_AnalyticsWithAliases) {
     nlohmann::json titles_schema = R"({
             "name": "titles",
             "fields": [