postgres/typesense
1
0
Fork 0
mirror of https://github.com/typesense/typesense.git synced 2025-05-21 06:02:26 +08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Add upsert + get APIs for analytics rules.

Browse Source
This commit is contained in:
Kishore Nallan 2023-07-07 12:26:54 +05:30
parent beb95a659a
commit 8e1f6caaf1
7 changed files with 196 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
include/analytics_manager.h
View File

@ -24,10 +24,11 @@ private:
void to_json(nlohmann::json& obj) const {
obj["name"] = name;
obj["type"] = POPULAR_QUERIES_TYPE;
obj["params"] = nlohmann::json::object();
obj["params"]["suggestion_collection"] = suggestion_collection;
obj["params"]["query_collections"] = query_collections;
obj["params"]["limit"] = limit;
obj["params"]["source"]["collections"] = query_collections;
obj["params"]["destination"]["collection"] = suggestion_collection;
}
};
@ -48,7 +49,9 @@ private:
Option<bool> remove_popular_queries_index(const std::string& name);
Option<bool> create_popular_queries_index(nlohmann::json &payload, bool write_to_disk);
Option<bool> create_popular_queries_index(nlohmann::json &payload,
bool upsert,
bool write_to_disk);
public:
@ -69,12 +72,14 @@ public:
Option<nlohmann::json> list_rules();
Option<bool> create_rule(nlohmann::json& payload, bool write_to_disk = true);
Option<nlohmann::json> get_rule(const std::string& name);
Option<bool> create_rule(nlohmann::json& payload, bool upsert, bool write_to_disk);
Option<bool> remove_rule(const std::string& name);
void add_suggestion(const std::string& query_collection,
std::string& query, const bool live_query, const std::string& user_id);
std::string& query, bool live_query, const std::string& user_id);
void stop();

4
include/core_api.h
View File

@ -147,8 +147,12 @@ bool post_create_event(const std::shared_ptr<http_req>& req, const std::shared_p
bool get_analytics_rules(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res);
bool get_analytics_rule(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res);
bool post_create_analytics_rules(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res);
bool put_upsert_analytics_rules(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res);
bool del_analytics_rules(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res);
// Misc helpers

43
src/analytics_manager.cpp
View File

@ -5,7 +5,7 @@
#include "http_client.h"
#include "collection_manager.h"
Option<bool> AnalyticsManager::create_rule(nlohmann::json& payload, bool write_to_disk) {
Option<bool> AnalyticsManager::create_rule(nlohmann::json& payload, bool upsert, bool write_to_disk) {
/*
Sample payload:
@ -37,16 +37,23 @@ Option<bool> AnalyticsManager::create_rule(nlohmann::json& payload, bool write_t
}
if(payload["type"] == POPULAR_QUERIES_TYPE) {
return create_popular_queries_index(payload, write_to_disk);
return create_popular_queries_index(payload, upsert, write_to_disk);
}
return Option<bool>(400, "Invalid type.");
}
Option<bool> AnalyticsManager::create_popular_queries_index(nlohmann::json &payload, bool write_to_disk) {
Option<bool> AnalyticsManager::create_popular_queries_index(nlohmann::json &payload, bool upsert, bool write_to_disk) {
// params and name are validated upstream
const auto& params = payload["params"];
const std::string& suggestion_config_name = payload["name"].get<std::string>();
bool already_exists = suggestion_configs.find(suggestion_config_name) != suggestion_configs.end();
if(!upsert && already_exists) {
return Option<bool>(400, "There's already another configuration with the name `" +
suggestion_config_name + "`.");
}
const auto& params = payload["params"];
if(!params.contains("source") || !params["source"].is_object()) {
return Option<bool>(400, "Bad or missing source.");
@ -56,18 +63,12 @@ Option<bool> AnalyticsManager::create_popular_queries_index(nlohmann::json &payl
return Option<bool>(400, "Bad or missing destination.");
}
size_t limit = 1000;
if(params.contains("limit") && params["limit"].is_number_integer()) {
limit = params["limit"].get<size_t>();
}
if(suggestion_configs.find(suggestion_config_name) != suggestion_configs.end()) {
return Option<bool>(400, "There's already another configuration with the name `" +
suggestion_config_name + "`.");
}
if(!params["source"].contains("collections") || !params["source"]["collections"].is_array()) {
return Option<bool>(400, "Must contain a valid list of source collections.");
}
@ -93,6 +94,14 @@ Option<bool> AnalyticsManager::create_popular_queries_index(nlohmann::json &payl
std::unique_lock lock(mutex);
if(already_exists) {
// remove the previous configuration with same name (upsert)
Option<bool> remove_op = remove_popular_queries_index(suggestion_config_name);
if(!remove_op.ok()) {
return Option<bool>(500, "Error erasing the existing configuration.");;
}
}
suggestion_configs.emplace(suggestion_config_name, suggestion_config);
for(const auto& query_coll: suggestion_config.query_collections) {
@ -130,13 +139,25 @@ Option<nlohmann::json> AnalyticsManager::list_rules() {
for(const auto& suggestion_config: suggestion_configs) {
nlohmann::json rule;
suggestion_config.second.to_json(rule);
rule["type"] = POPULAR_QUERIES_TYPE;
rules["rules"].push_back(rule);
}
return Option<nlohmann::json>(rules);
}
Option<nlohmann::json> AnalyticsManager::get_rule(const string& name) {
nlohmann::json rule;
std::unique_lock lock(mutex);
auto suggestion_config_it = suggestion_configs.find(name);
if(suggestion_config_it == suggestion_configs.end()) {
return Option<nlohmann::json>(404, "Rule not found.");
}
suggestion_config_it->second.to_json(rule);
return Option<nlohmann::json>(rule);
}
Option<bool> AnalyticsManager::remove_rule(const string &name) {
std::unique_lock lock(mutex);

22
src/collection_manager.cpp
View File

@ -285,6 +285,17 @@ Option<bool> CollectionManager::load(const size_t collection_batch_size, const s
iter->Next();
}
// restore query suggestions configs
std::vector<std::string> analytics_config_jsons;
store->scan_fill(AnalyticsManager::ANALYTICS_RULE_PREFIX,
std::string(AnalyticsManager::ANALYTICS_RULE_PREFIX) + "`",
analytics_config_jsons);
for(const auto& analytics_config_json: analytics_config_jsons) {
nlohmann::json analytics_config = nlohmann::json::parse(analytics_config_json);
AnalyticsManager::get_instance().create_rule(analytics_config, false, false);
}
delete iter;
LOG(INFO) << "Loaded " << num_collections << " collection(s).";
@ -1312,17 +1323,6 @@ Option<bool> CollectionManager::load_collection(const nlohmann::json &collection
collection->add_synonym(collection_synonym, false);
}
// restore query suggestions configs
std::vector<std::string> analytics_config_jsons;
cm.store->scan_fill(AnalyticsManager::ANALYTICS_RULE_PREFIX,
std::string(AnalyticsManager::ANALYTICS_RULE_PREFIX) + "`",
analytics_config_jsons);
for(const auto& analytics_config_json: analytics_config_jsons) {
nlohmann::json analytics_config = nlohmann::json::parse(analytics_config_json);
AnalyticsManager::get_instance().create_rule(analytics_config, false);
}
// Fetch records from the store and re-create memory index
const std::string seq_id_prefix = collection->get_seq_id_collection_prefix();
std::string upper_bound_key = collection->get_seq_id_collection_prefix() + "`"; // cannot inline this

52
src/core_api.cpp
View File

@ -2078,13 +2078,25 @@ bool post_create_event(const std::shared_ptr<http_req>& req, const std::shared_p
bool get_analytics_rules(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res) {
auto rules_op = AnalyticsManager::get_instance().list_rules();
if(rules_op.ok()) {
res->set_200(rules_op.get().dump());
return true;
if(!rules_op.ok()) {
res->set(rules_op.code(), rules_op.error());
return false;
}
res->set(rules_op.code(), rules_op.error());
return false;
res->set_200(rules_op.get().dump());
return true;
}
bool get_analytics_rule(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res) {
auto rules_op = AnalyticsManager::get_instance().get_rule(req->params["name"]);
if(!rules_op.ok()) {
res->set(rules_op.code(), rules_op.error());
return false;
}
res->set_200(rules_op.get().dump());
return true;
}
bool post_create_analytics_rules(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res) {
@ -2098,7 +2110,7 @@ bool post_create_analytics_rules(const std::shared_ptr<http_req>& req, const std
return false;
}
auto op = AnalyticsManager::get_instance().create_rule(req_json);
auto op = AnalyticsManager::get_instance().create_rule(req_json, false, true);
if(!op.ok()) {
res->set(op.code(), op.error());
@ -2109,6 +2121,29 @@ bool post_create_analytics_rules(const std::shared_ptr<http_req>& req, const std
return true;
}
bool put_upsert_analytics_rules(const std::shared_ptr<http_req> &req, const std::shared_ptr<http_res> &res) {
nlohmann::json req_json;
try {
req_json = nlohmann::json::parse(req->body);
} catch(const std::exception& e) {
LOG(ERROR) << "JSON error: " << e.what();
res->set_400("Bad JSON.");
return false;
}
req_json["name"] = req->params["name"];
auto op = AnalyticsManager::get_instance().create_rule(req_json, true, true);
if(!op.ok()) {
res->set(op.code(), op.error());
return false;
}
res->set_200(req_json.dump());
return true;
}
bool del_analytics_rules(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res) {
auto op = AnalyticsManager::get_instance().remove_rule(req->params["name"]);
if(!op.ok()) {
@ -2116,11 +2151,10 @@ bool del_analytics_rules(const std::shared_ptr<http_req>& req, const std::shared
return false;
}
res->set_200(R"({"ok": true)");
res->set_200(R"({"ok": true})");
return true;
}
bool post_proxy(const std::shared_ptr<http_req>& req, const std::shared_ptr<http_res>& res) {
HttpProxy& proxy = HttpProxy::get_instance();
@ -2180,4 +2214,4 @@ bool post_proxy(const std::shared_ptr<http_req>& req, const std::shared_ptr<http
res->set_200(response.body);
return true;
}
}

2
src/main/typesense_server.cpp
View File

@ -70,7 +70,9 @@ void master_server_routes() {
// analytics
server->get("/analytics/rules", get_analytics_rules);
server->get("/analytics/rules/:name", get_analytics_rule);
server->post("/analytics/rules", post_create_analytics_rules);
server->put("/analytics/rules/:name", put_upsert_analytics_rules);
server->del("/analytics/rules/:name", del_analytics_rules);
server->post("/analytics/events", post_create_event);

95
test/analytics_manager_test.cpp
View File

@ -78,7 +78,7 @@ TEST_F(AnalyticsManagerTest, AddSuggestion) {
}
})"_json;
auto create_op = analyticsManager.create_rule(analytics_rule);
auto create_op = analyticsManager.create_rule(analytics_rule, false, true);
ASSERT_TRUE(create_op.ok());
std::string q = "foobar";
@ -88,4 +88,97 @@ TEST_F(AnalyticsManagerTest, AddSuggestion) {
auto userQueries = popularQueries["top_queries"]->get_user_prefix_queries()["1"];
ASSERT_EQ(1, userQueries.size());
ASSERT_EQ("foobar", userQueries[0].query);
// add another query which is more popular
q = "buzzfoo";
analyticsManager.add_suggestion("titles", q, true, "1");
analyticsManager.add_suggestion("titles", q, true, "2");
analyticsManager.add_suggestion("titles", q, true, "3");
popularQueries = analyticsManager.get_popular_queries();
userQueries = popularQueries["top_queries"]->get_user_prefix_queries()["1"];
ASSERT_EQ(2, userQueries.size());
ASSERT_EQ("foobar", userQueries[0].query);
ASSERT_EQ("buzzfoo", userQueries[1].query);
}
TEST_F(AnalyticsManagerTest, GetAndDeleteSuggestions) {
nlohmann::json analytics_rule = R"({
"name": "top_search_queries",
"type": "popular_queries",
"params": {
"limit": 100,
"source": {
"collections": ["titles"]
},
"destination": {
"collection": "top_queries"
}
}
})"_json;
auto create_op = analyticsManager.create_rule(analytics_rule, false, true);
ASSERT_TRUE(create_op.ok());
analytics_rule = R"({
"name": "top_search_queries2",
"type": "popular_queries",
"params": {
"limit": 100,
"source": {
"collections": ["titles"]
},
"destination": {
"collection": "top_queries"
}
}
})"_json;
create_op = analyticsManager.create_rule(analytics_rule, false, true);
ASSERT_TRUE(create_op.ok());
auto rules = analyticsManager.list_rules().get()["rules"];
ASSERT_EQ(2, rules.size());
ASSERT_TRUE(analyticsManager.get_rule("top_search_queries").ok());
ASSERT_TRUE(analyticsManager.get_rule("top_search_queries2").ok());
auto missing_rule_op = analyticsManager.get_rule("top_search_queriesX");
ASSERT_FALSE(missing_rule_op.ok());
ASSERT_EQ(404, missing_rule_op.code());
ASSERT_EQ("Rule not found.", missing_rule_op.error());
// upsert rule that already exists
analytics_rule = R"({
"name": "top_search_queries2",
"type": "popular_queries",
"params": {
"limit": 100,
"source": {
"collections": ["titles"]
},
"destination": {
"collection": "top_queriesUpdated"
}
}
})"_json;
create_op = analyticsManager.create_rule(analytics_rule, true, true);
ASSERT_TRUE(create_op.ok());
auto existing_rule = analyticsManager.get_rule("top_search_queries2").get();
ASSERT_EQ("top_queriesUpdated", existing_rule["params"]["destination"]["collection"].get<std::string>());
// reject when upsert is not enabled
create_op = analyticsManager.create_rule(analytics_rule, false, true);
ASSERT_FALSE(create_op.ok());
ASSERT_EQ("There's already another configuration with the name `top_search_queries2`.", create_op.error());
// try deleting both rules
analyticsManager.remove_rule("top_search_queries");
analyticsManager.remove_rule("top_search_queries2");
missing_rule_op = analyticsManager.get_rule("top_search_queries");
ASSERT_FALSE(missing_rule_op.ok());
missing_rule_op = analyticsManager.get_rule("top_search_queries2");
ASSERT_FALSE(missing_rule_op.ok());
}