From 387a2e7f8f48acb5cb4bc2b9949f2f9b2f9e0268 Mon Sep 17 00:00:00 2001 From: Jeroen van der Heijden Date: Mon, 7 Jan 2019 16:19:25 +0100 Subject: [PATCH] Added drop database option --- include/qpack/qpack.h | 4 + include/siri/db/db.h | 3 + include/siri/net/protocol.h | 4 +- include/siri/service/request.h | 1 + include/xpath/xpath.h | 1 + src/siri/db/auth.c | 6 +- src/siri/db/db.c | 83 ++++++++++++++++++- src/siri/net/bserver.c | 20 +++++ src/siri/net/protocol.c | 2 + src/siri/service/request.c | 142 ++++++++++++++++++++++++++++++--- src/xpath/xpath.c | 42 ++++++++++ 11 files changed, 290 insertions(+), 18 deletions(-) diff --git a/include/qpack/qpack.h b/include/qpack/qpack.h index 2528a1be..bb5168b6 100644 --- a/include/qpack/qpack.h +++ b/include/qpack/qpack.h @@ -160,6 +160,10 @@ static inline int qp_is_double(qp_types_t tp) { return tp == QP_DOUBLE; } +static inline int qp_is_bool(qp_types_t tp) +{ + return tp == QP_TRUE || tp == QP_FALSE; +} static inline int qp_is_raw_term(qp_obj_t * qp_obj) { return (qp_obj->tp == QP_RAW && diff --git a/include/siri/db/db.h b/include/siri/db/db.h index 952f96d8..d76c66ac 100644 --- a/include/siri/db/db.h +++ b/include/siri/db/db.h @@ -10,6 +10,7 @@ typedef struct siridb_s siridb_t; #define SIRIDB_MAX_DBNAME_LEN 256 /* 255 + NULL */ #define SIRIDB_SCHEMA 5 #define SIRIDB_FLAG_REINDEXING 1 +#define SIRIDB_FLAG_DROPPED 2 #define DEF_DROP_THRESHOLD 1.0 /* 100% */ #define DEF_SELECT_POINTS_LIMIT 1000000 /* one million */ @@ -43,11 +44,13 @@ int8_t siridb_get_idle_percentage(siridb_t * siridb); int siridb_is_db_path(const char * dbpath); siridb_t * siridb_new(const char * dbpath, int lock_flags); siridb_t * siridb_get(llist_t * siridb_list, const char * dbname); +siridb_t * siridb_get_by_qp(llist_t * siridb_list, qp_obj_t * qp_dbname); void siridb_decref_cb(siridb_t * siridb, void * args); ssize_t siridb_get_file(char ** buffer, siridb_t * siridb); int siridb_open_files(siridb_t * siridb); int siridb_save(siridb_t * siridb); void siridb__free(siridb_t * siridb); +void siridb_drop(siridb_t * siridb); #define siridb_incref(siridb) siridb->ref++ #define siridb_decref(_siridb) if (!--_siridb->ref) siridb__free(_siridb) diff --git a/include/siri/net/protocol.h b/include/siri/net/protocol.h index 455f6c98..f5a1f147 100644 --- a/include/siri/net/protocol.h +++ b/include/siri/net/protocol.h @@ -78,6 +78,7 @@ typedef enum BPROTO_ENABLE_BACKUP_MODE, /* empty */ BPROTO_DISABLE_BACKUP_MODE, /* empty */ BPROTO_TEE_PIPE_NAME_UPDATE, /* tee pipe name */ + BPROTO_DROP_DATABASE, /* empty */ } bproto_client_t; /* @@ -125,7 +126,8 @@ typedef enum BPROTO_ACK_ENABLE_BACKUP_MODE, /* empty */ BPROTO_ACK_DISABLE_BACKUP_MODE, /* empty */ BPROTO_RES_GROUPS, /* [[name, series], ...] */ - BPROTO_ACK_TEE_PIPE_NAME /* empty */ + BPROTO_ACK_TEE_PIPE_NAME, /* empty */ + BPROTO_ACK_DROP_DATABASE, /* empty */ } bproto_server_t; diff --git a/include/siri/service/request.h b/include/siri/service/request.h index afececf2..f991012b 100644 --- a/include/siri/service/request.h +++ b/include/siri/service/request.h @@ -12,6 +12,7 @@ typedef enum SERVICE_NEW_DATABASE_, SERVICE_NEW_POOL, SERVICE_NEW_REPLICA, + SERVICE_DROP_DATABASE, SERVICE_GET_VERSION=64, SERVICE_GET_ACCOUNTS, SERVICE_GET_DATABASES diff --git a/include/xpath/xpath.h b/include/xpath/xpath.h index 9524ef99..9c98df1d 100644 --- a/include/xpath/xpath.h +++ b/include/xpath/xpath.h @@ -17,5 +17,6 @@ int xpath_file_exist(const char * fn); int xpath_is_dir(const char * path); ssize_t xpath_get_content(char ** buffer, const char * fn); int xpath_get_exec_path(char * path); +int xpath_rmdir(const char * path); #endif /* XPATH_H_ */ diff --git a/src/siri/db/auth.c b/src/siri/db/auth.c index 5a72c5f1..6d3ff9dc 100644 --- a/src/siri/db/auth.c +++ b/src/siri/db/auth.c @@ -23,10 +23,6 @@ cproto_server_t siridb_auth_user_request( siridb_t * siridb; siridb_user_t * user; - char dbname[qp_dbname->len + 1]; - memcpy(dbname, qp_dbname->via.raw, qp_dbname->len); - dbname[qp_dbname->len] = 0; - char username[qp_username->len + 1]; memcpy(username, qp_username->via.raw, qp_username->len); username[qp_username->len] = 0; @@ -35,7 +31,7 @@ cproto_server_t siridb_auth_user_request( memcpy(password, qp_password->via.raw, qp_password->len); password[qp_password->len] = 0; - if ((siridb = siridb_get(siri.siridb_list, dbname)) == NULL) + if ((siridb = siridb_get_by_qp(siri.siridb_list, qp_dbname)) == NULL) { log_warning("User authentication request failed: unknown database"); return CPROTO_ERR_AUTH_UNKNOWN_DB; diff --git a/src/siri/db/db.c b/src/siri/db/db.c index bc105217..59da730b 100644 --- a/src/siri/db/db.c +++ b/src/siri/db/db.c @@ -48,6 +48,7 @@ static int siridb__from_unpacker( static siridb_t * siridb__from_dat(const char * dbpath); static int siridb__read_conf(siridb_t * siridb); static int siridb__lock(const char * dbpath, int lock_flags); +static inline int siridb__cmp_db(siridb_t * siridb, qp_obj_t * dbname); #define READ_DB_EXIT_WITH_ERROR(ERROR_MSG) \ strcpy(err_msg, ERROR_MSG); \ @@ -529,6 +530,34 @@ siridb_t * siridb_get(llist_t * siridb_list, const char * dbname) return NULL; } +/* + * Get a siridb object by qpack name. + */ +siridb_t * siridb_get_by_qp(llist_t * siridb_list, qp_obj_t * qp_dbname) +{ + assert (qp_dbname->tp == QP_RAW); + + llist_node_t * node = siridb_list->first; + siridb_t * siridb; + + while (node != NULL) + { + siridb = (siridb_t *) node->data; + if (qp_dbname->len == strlen(siridb->dbname) && + strncmp( + siridb->dbname, + (const char *) qp_dbname->via.raw, + qp_dbname->len) == 0) + { + return siridb; + } + node = node->next; + } + + return NULL; +} + + /* * Sometimes we need a callback function and cannot use a macro expansion. */ @@ -706,14 +735,53 @@ void siridb__free(siridb_t * siridb) } } + uv_mutex_destroy(&siridb->series_mutex); + uv_mutex_destroy(&siridb->shards_mutex); + + if (siridb->flags & SIRIDB_FLAG_DROPPED) + { + xpath_rmdir(siridb->dbpath); + } + free(siridb->dbpath); free(siridb->dbname); free(siridb->time); + free(siridb); +} - uv_mutex_destroy(&siridb->series_mutex); - uv_mutex_destroy(&siridb->shards_mutex); +void siridb_drop(siridb_t * siridb) +{ + if (siridb->flags & SIRIDB_FLAG_DROPPED) + { + return; + } - free(siridb); + log_warning("dropping database '%s'", siridb->dbname); + + siridb->flags |= SIRIDB_FLAG_DROPPED; + + uv_mutex_lock(&siri.siridb_mutex); + + (void *) llist_remove(siri.siridb_list, NULL, siridb); + + uv_mutex_unlock(&siri.siridb_mutex); + + if (siridb->replicate != NULL) + { + siridb_replicate_close(siridb->replicate); + } + + if (siridb->reindex != NULL && siridb->reindex->timer != NULL) + { + siridb_reindex_close(siridb->reindex); + } + + if (siridb->groups != NULL) + { + siridb_groups_destroy(siridb->groups); + } + + siridb_decref(siridb); } /* @@ -944,5 +1012,14 @@ static int siridb__lock(const char * dbpath, int lock_flags) return 0; } +static inline int siridb__cmp_db(siridb_t * siridb, qp_obj_t * dbname) +{ + size_t len = strlen(siridb->dbname); + return ( + dbname->len == len && + strncmp(siridb->dbname, (const char *) dbname->via.raw, len) == 0 + ); +} + diff --git a/src/siri/net/bserver.c b/src/siri/net/bserver.c index 3854db7e..b4792d44 100644 --- a/src/siri/net/bserver.c +++ b/src/siri/net/bserver.c @@ -46,6 +46,7 @@ static void on_log_level_update( static void on_tee_pipe_name_update( sirinet_stream_t * client, sirinet_pkg_t * pkg); +static void on_drop_database(sirinet_stream_t * client, sirinet_pkg_t * pkg); static void on_repl_finished(sirinet_stream_t * client, sirinet_pkg_t * pkg); static void on_query( sirinet_stream_t * client, @@ -279,6 +280,9 @@ static void on_data(sirinet_stream_t * client, sirinet_pkg_t * pkg) case BPROTO_TEE_PIPE_NAME_UPDATE: on_tee_pipe_name_update(client, pkg); break; + case BPROTO_DROP_DATABASE: + on_drop_database(client, pkg); + break; } } @@ -458,6 +462,22 @@ static void on_tee_pipe_name_update( } } +static void on_drop_database(sirinet_stream_t * client, sirinet_pkg_t * pkg) +{ + SERVER_CHECK_AUTHENTICATED(client, server) + + siridb_t * siridb = client->siridb; + sirinet_pkg_t * package = NULL; + + siridb_drop(siridb); + + package = sirinet_pkg_new(pkg->pid, 0, BPROTO_ACK_DROP_DATABASE, NULL); + if (package != NULL) + { + sirinet_pkg_send(client, package); + } +} + static void on_repl_finished(sirinet_stream_t * client, sirinet_pkg_t * pkg) { SERVER_CHECK_AUTHENTICATED(client, server) diff --git a/src/siri/net/protocol.c b/src/siri/net/protocol.c index b9b6122b..e0572c0d 100644 --- a/src/siri/net/protocol.c +++ b/src/siri/net/protocol.c @@ -85,6 +85,7 @@ const char * sirinet_bproto_client_str(bproto_client_t n) case BPROTO_ENABLE_BACKUP_MODE: return "BPROTO_ENABLE_BACKUP_MODE"; case BPROTO_DISABLE_BACKUP_MODE: return "BPROTO_DISABLE_BACKUP_MODE"; case BPROTO_TEE_PIPE_NAME_UPDATE: return "BPROTO_TEE_PIPE_NAME_UPDATE"; + case BPROTO_DROP_DATABASE: return "BPROTO_DROP_DATABASE"; default: sprintf(protocol_str, "BPROTO_CLIENT_TYPE_UNKNOWN (%d)", n); return protocol_str; @@ -121,6 +122,7 @@ const char * sirinet_bproto_server_str(bproto_server_t n) case BPROTO_ACK_DISABLE_BACKUP_MODE: return "BPROTO_ACK_DISABLE_BACKUP_MODE"; case BPROTO_RES_GROUPS: return "BPROTO_RES_GROUPS"; case BPROTO_ACK_TEE_PIPE_NAME: return "BPROTO_ACK_TEE_PIPE_NAME"; + case BPROTO_ACK_DROP_DATABASE: return "BPROTO_ACK_DROP_DATABASE"; default: sprintf(protocol_str, "BPROTO_SERVER_TYPE_UNKNOWN (%d)", n); return protocol_str; diff --git a/src/siri/service/request.c b/src/siri/service/request.c index c60ab7a6..1ca029d9 100644 --- a/src/siri/service/request.c +++ b/src/siri/service/request.c @@ -3,21 +3,22 @@ */ #define PCRE2_CODE_UNIT_WIDTH 8 +#include +#include +#include +#include +#include +#include +#include #include #include -#include #include #include -#include -#include -#include -#include +#include +#include #include #include -#include -#include -#include -#include +#include #define DEFAULT_TIME_PRECISION 1 #define DEFAULT_BUFFER_SIZE 1024 @@ -142,6 +143,9 @@ static cproto_server_t SERVICE_on_drop_account( qp_unpacker_t * qp_unpacker, qp_obj_t * qp_account, char * err_msg); +static cproto_server_t SERVICE_on_drop_database( + qp_unpacker_t * qp_unpacker, + char * err_msg); static cproto_server_t SERVICE_on_new_database( qp_unpacker_t * qp_unpacker, char * err_msg); @@ -170,6 +174,7 @@ static int SERVICE_find_database(siridb_t * siridb, qp_obj_t * dbname); static int SERVICE_list_accounts( siri_service_account_t * account, qp_packer_t * packer); +static void SERVICE_on_drop_database_cb(vec_t *, void *); static size_t max_filename_sz; @@ -263,6 +268,8 @@ cproto_server_t siri_service_request( client, SERVICE_NEW_REPLICA, err_msg); + case SERVICE_DROP_DATABASE: + return SERVICE_on_drop_database(qp_unpacker, err_msg); case SERVICE_GET_VERSION: return SERVICE_on_get_version(qp_unpacker, packaddr, err_msg); case SERVICE_GET_ACCOUNTS: @@ -438,6 +445,101 @@ static cproto_server_t SERVICE_on_drop_account( CPROTO_ERR_SERVICE : CPROTO_ACK_SERVICE; } +/* + * Returns CPROTO_ACK_SERVICE when successful. + * In case of an error CPROTO_ERR_SERVICE can be returned in which case err_msg + * is set, or CPROTO_ERR_SERVICE_INVALID_REQUEST is returned. + */ +static cproto_server_t SERVICE_on_drop_database( + qp_unpacker_t * qp_unpacker, + char * err_msg) +{ + _Bool ignore_offline; + siridb_t * siridb; + qp_obj_t qp_key, qp_target, qp_ignore_offline; + sirinet_pkg_t * pkg; + vec_t * servers; + + qp_target.tp = QP_HOOK; + qp_ignore_offline.tp = QP_HOOK; + + if (!qp_is_map(qp_next(qp_unpacker, NULL))) + { + return CPROTO_ERR_SERVICE_INVALID_REQUEST; + } + + while (qp_next(qp_unpacker, &qp_key) == QP_RAW) + { + if ( strncmp( + (const char *) qp_key.via.raw, + "database", + qp_key.len) == 0 && + qp_next(qp_unpacker, &qp_target) == QP_RAW) + { + continue; + } + + if ( strncmp( + (const char *) qp_key.via.raw, + "ignore_offline", + qp_key.len) == 0 && + qp_is_bool(qp_next(qp_unpacker, &qp_ignore_offline))) + { + continue; + } + return CPROTO_ERR_SERVICE_INVALID_REQUEST; + } + + if (qp_target.tp == QP_HOOK) + { + return CPROTO_ERR_SERVICE_INVALID_REQUEST; + } + + ignore_offline = ( + qp_ignore_offline.tp != QP_HOOK && + qp_ignore_offline.tp == QP_TRUE + ); + + siridb = siridb_get_by_qp(siri.siridb_list, &qp_target); + if (siridb == NULL) + { + sprintf(err_msg, "cannot find database '%.*s'", + (int) qp_target.len, (char *) qp_target.via.raw); + return CPROTO_ERR_SERVICE; + } + + if (!ignore_offline && !siridb_servers_online(siridb)) + { + sprintf(err_msg, + "at least one server is offline, " + "set `ignore_offline` to true if you want to " + "ignore offline servers"); + return CPROTO_ERR_SERVICE; + } + + pkg = sirinet_pkg_new(0, 0, BPROTO_DROP_DATABASE, NULL); + servers = siridb_servers_other2vec(siridb); + if (pkg == NULL || servers == NULL) + { + free(pkg); + vec_free(servers); + sprintf(err_msg, "memory allocation error"); + return CPROTO_ERR_SERVICE; + } + + siridb_servers_send_pkg( + servers, + pkg, + 0, + SERVICE_on_drop_database_cb, + NULL); + + siridb_drop(siridb); + vec_free(servers); + + return CPROTO_ACK_SERVICE; +} + /* * Returns CPROTO_ACK_SERVICE when successful. * In case of an error CPROTO_ERR_SERVICE can be returned in which case err_msg @@ -988,3 +1090,25 @@ static int SERVICE_list_accounts( { return qp_add_string(packer, account->account); } + +static void SERVICE_on_drop_database_cb( + vec_t * promises, + void * data __attribute__((unused))) +{ + log_debug("drop database has been send to all online servers"); + + if (promises != NULL) + { + size_t i; + sirinet_promise_t * promise; + for (i = 0; i < promises->len; i++) + { + promise = promises->data[i]; + if (promise != NULL) + { + free(promise->data); + sirinet_promise_decref(promise); + } + } + } +} diff --git a/src/xpath/xpath.c b/src/xpath/xpath.c index 20d26edb..8e1eb23e 100644 --- a/src/xpath/xpath.c +++ b/src/xpath/xpath.c @@ -8,6 +8,7 @@ #include #include #include +#include #include /* @@ -104,3 +105,44 @@ int xpath_get_exec_path(char * path) return 0; } + +int xpath_rmdir(const char * path) +{ + DIR * d = opendir(path); + if (!d) + return -1; + + size_t bufsz = 0, path_len = strlen(path); + const char * slash = (path[path_len - 1] == '/') ? "" : "/"; + struct dirent * p; + char * buf = NULL; + + while ((p = readdir(d))) + { + size_t len; + + /* Skip the names "." and ".." as we don't want to recurse on them. */ + if (!strcmp(p->d_name, ".") || !strcmp(p->d_name, "..")) + continue; + + len = path_len + strlen(p->d_name) + 2; + if (len > bufsz) + { + bufsz = len; + char * tmp = realloc(buf, bufsz); + if (!tmp) goto stop; + buf = tmp; + } + + snprintf(buf, len, "%s%s%s", path, slash, p->d_name); + + if (xpath_is_dir(buf) ? xpath_rmdir(buf) : unlink(buf)) + goto stop; + } + +stop: + free(buf); + closedir(d); + + return rmdir(path); +} -- 2.39.5