Auto replication (IntelLabs#40)

* Add Auto-Replication Support for VDMS db

* Add untar and restore option

* Add Auto-replicate test

* remove backups in tests folder

* remove the filter of Video tests

* remove archive_db.sh, remove the wrong backup message, and edit the config-vdms file

* Adjust the code to show the false backup option

config-vdms.json

@@ -1,6 +1,10 @@
 {
     "port": 55555,
+    "autoreplicate_interval":-1, // it should be > 0
+    "unit":"s", 
     "max_simultaneous_clients": 100,
+    // "backup_path":"backups_test", // set this if you want different path to store the back up file
     "db_root_path": "db",
+    "backup_flag" : "false",
     "more-info": "github.com/IntelLabs/vdms"
 }

src/QueryHandler.cc

@@ -113,7 +113,9 @@ void QueryHandler::init()
 
 QueryHandler::QueryHandler()
     : _pmgd_qh(),
-    _validator(valijson::Validator::kWeakTypes),_autodelete_init(false)
+    _validator(valijson::Validator::kWeakTypes),
+    _autodelete_init(false),
+    _autoreplicate_init(false)
 #ifdef CHRONO_TIMING
     ,ch_tx_total("ch_tx_total")
     ,ch_tx_query("ch_tx_query")
@@ -458,8 +460,55 @@ void QueryHandler::process_query(protobufs::queryMessage& proto_query,
         exception_handler();
     }
 }
+void QueryHandler::reset_autoreplicate_init_flag()
+{
+    _autoreplicate_init = true;
+}
+void QueryHandler::set_autoreplicate_init_flag( )
+{
+    _autoreplicate_init = false;
+}
+void QueryHandler::regualar_run_autoreplicate( std::string& backup_path, std::string& db_path, int& server_port)
+{
+    std::string command = "bsdtar cvfz ";
+    std::string name;
+    std::ostringstream oss;
+    Json::Value config_file; 
+    std::ofstream file_id;
+    name.clear();
+    auto t = std::time(nullptr);
+    auto tm = *std::localtime(&t);
+    oss<<asctime(&tm);
+    name=oss.str();
+    name.erase(remove(name.begin(), name.end(), ' '), name.end());
+    name.erase(std::remove(name.begin(), name.end(), '\n'),name.end());
+    // name.replace(name.find(":"),name.size(),"_");
+    std::string full_name=backup_path+"/" +name;
+
+    command = command + " "  + full_name +".tar.gz  " + db_path; // current_date_time
+    std::cout <<command <<std::endl;
+
+
+    system(command.c_str());
+
+    config_file["port"]=server_port;
+    config_file["db_root_path"]= full_name;
+    config_file["more-info"] ="github.com/IntelLabs/vdms";
+    std::string config_file_name =full_name+".json";
+    std::cout << "Name is" << config_file_name <<std::endl;
+    file_id.open(config_file_name);
+    Json::StyledWriter Writer;
+    file_id << Writer.write( config_file);
+    file_id.close();
+
+    command = "bsdtar cvfz ";
+    oss.str(std::string());
+    name.clear();
+    config_file.clear();
+
+
 
-
+}
 void QueryHandler::reset_autodelete_init_flag()
 {
     _autodelete_init = false;

src/QueryHandler.h

@@ -57,7 +57,8 @@ typedef ::google::protobuf::RepeatedPtrField<std::string> BlobArray;
 
         static std::unordered_map<std::string, RSCommand* > _rs_cmds;
         PMGDQueryHandler _pmgd_qh;
-        bool _autodelete_init; 
+        bool _autodelete_init;
+        bool _autoreplicate_init; 
 
         bool syntax_checker(const Json::Value &root, Json::Value& error);
         int parse_commands(const protobufs::queryMessage& proto_query,
@@ -89,5 +90,9 @@ typedef ::google::protobuf::RepeatedPtrField<std::string> BlobArray;
         void set_autodelete_init_flag();
         void regualar_run_autodelete();
         void build_autodelete_queue();
+        void set_autoreplicate_init_flag();
+        void reset_autoreplicate_init_flag();
+        void regualar_run_autoreplicate(std::string&, std::string&, int&);
+
     };
 }

src/Server.cc

@@ -30,6 +30,13 @@
  */
 
 #include <stdlib.h>     /* system, NULL, EXIT_FAILURE */
+#include <stdio.h>
+#include <thread>
+#include <chrono>
+
+#include <jsoncpp/json/json.h>
+#include <jsoncpp/json/value.h> //to create the config file
+
 
 #include "Server.h"
 #include "comm/Connection.h"
@@ -39,6 +46,7 @@
 #include "QueryHandler.h"
 #include "DescriptorsManager.h"
 
+
 #include "pmgdMessages.pb.h" // Protobuff implementation
 
 using namespace VDMS;
@@ -52,6 +60,19 @@ Server::Server(std::string config_file)
                         ->get_int_value("port", DEFAULT_PORT);
     _autodelete_interval = VDMSConfig::instance()
                         ->get_int_value("autodelete_interval_s", DEFAULT_AUTODELETE_INTERVAL);
+    _backup_flag =    VDMSConfig::instance()
+                        ->get_string_value("backup_flag", DEFAULT_AUTOREPLICATE_FLAG) ;                  
+
+    _autoreplecate_interval = VDMSConfig::instance()
+                        ->get_int_value("autoreplicate_interval", DEFAULT_AUTOREPLICATE_INTERVAL);
+    _replication_unit =  VDMSConfig::instance()
+                        ->get_string_value("unit", DEFAULT_AUTOREPLICATE_UNIT); 
+    _backup_path =  VDMSConfig::instance()
+                        ->get_string_value("backup_path", DEFAULT_BACKUP_PATH);  
+    _db_path =     VDMSConfig::instance()
+                        ->get_string_value("db_root_path", DEFAULT_DB_ROOT);                 
+
+
 
     PMGDQueryHandler::init();
     QueryHandler::init();
@@ -89,6 +110,8 @@ void Server::process_requests()
             comm::Connection *conn_server =
                 new comm::Connection(server->accept());
             _cm->add_connection(conn_server);
+
+
         }
         catch (comm::ExceptionComm e) {
             print_exception(e);
@@ -97,7 +120,45 @@ void Server::process_requests()
 
     delete server;
 }
-
+void Server::untar_data(std::string& name){
+
+
+        std::string command="tar -xvSf" + name;
+        system(command.c_str());
+
+}
+void Server::auto_replicate_data(){
+
+    long replication_period;
+    QueryHandler qh;
+ if(_backup_flag =="true"){
+       if (_autoreplecate_interval >0 ){
+            if (_replication_unit.compare("h") == 0){
+                replication_period =_autoreplecate_interval*60*60;
+            }
+            else if (_replication_unit.compare("m") == 0)
+                replication_period =_autoreplecate_interval*60;
+
+            else     
+                 replication_period= _autoreplecate_interval;
+       }       
+
+        if(_backup_path.empty()){
+            _backup_path=_db_path; //set the defualt path to be db
+        } 
+
+
+     if(replication_period > 0) //check to ensure valid autodelete_interval
+      { 
+
+        while(!shutdown)
+        {
+            sleep(replication_period);
+            qh.regualar_run_autoreplicate(_backup_path, _db_path, _server_port);
+        }
+    } 
+ } 
+}
 void Server::autodelete_expired_data()
 {
     if(_autodelete_interval > 0) //check to ensure valid autodelete_interval

src/Server.h

@@ -35,19 +35,36 @@
 
 #include "pmgd.h"
 #include "CommunicationManager.h"
+#include <chrono>
+
 
 namespace VDMS {
     class Server
     {
         static const int DEFAULT_PORT = 55555;
         static const int DEFAULT_AUTODELETE_INTERVAL = -1;
+        static const int DEFAULT_AUTOREPLICATE_INTERVAL = -1;
+        std::string DEFAULT_AUTOREPLICATE_UNIT ="s" ;
+        std::string DEFAULT_BACKUP_PATH =".";
+        std::string DEFAULT_DB_ROOT ="db";
+        std::string DEFAULT_AUTOREPLICATE_FLAG="false";
+
+
 
         CommunicationManager *_cm;
 
         // TODO: Partitioner here
 
         int _server_port;
         int _autodelete_interval;
+        int _autoreplecate_interval;
+        std::string _replication_unit;
+        std::string _backup_path;
+        std::string _db_path;
+        std::string _backup_flag;
+
+        bool _untar;
+
 
         // Handle ^c
         static bool shutdown;
@@ -61,6 +78,8 @@ namespace VDMS {
         Server(std::string config_file);
         void process_requests();
         void autodelete_expired_data();
+        void auto_replicate_data();
+        void untar_data(std::string&);
         ~Server();
     };
 };

src/vdms.cc

@@ -34,11 +34,14 @@
 */
 #include <iostream>
 
+
 #include "Server.h"
 
 void printUsage()
 {
     std::cout << "Usage: vdms -cfg config-file.json" << std::endl;
+
+    std::cout << "Usage: vdms -restore db.tar.gz" << std::endl;
     exit(0);
 }
 
@@ -49,6 +52,10 @@ static void* start_request_thread(void* server)
     ((VDMS::Server*)(server))->process_requests();
     return NULL;
 }
+static void* start_replication_thread(void* server){
+    ((VDMS::Server*)(server))->auto_replicate_data();
+}
+
 
 static void* start_autodelete_thread(void* server)
 {
@@ -59,8 +66,8 @@ static void* start_autodelete_thread(void* server)
 
 int main(int argc, char **argv)
 {
-    pthread_t request_thread, autodelete_thread;
-    int request_thread_flag, autodelete_thread_flag;
+    pthread_t request_thread, autodelete_thread, auto_replicate_thread;
+    int request_thread_flag, autodelete_thread_flag, auto_replcation_flag;
 
     printf("VDMS Server\n");
 
@@ -69,23 +76,64 @@ int main(int argc, char **argv)
     }
 
     std::string config_file = "config-vdms.json";
+    bool backup=false; //default of backing up the db folder to be false
+
 
     if (argc == 3){
         std::string option(argv[1]);
-        if (option != "-cfg")
+
+        if (option != "-cfg" && option!="-restore"  && option!="-backup")
             printUsage();
-
+        if(option =="-cfg")
         config_file = std::string (argv[2]);
+
+
+
+        else if (option=="-restore" ){
+            void* server;
+
+            std::string db_name(argv[2]);
+            size_t file_ext1 = db_name.find_last_of(".");
+
+           std::string temp_name_1= db_name.substr(0,file_ext1);
+
+           size_t file_ext2 = temp_name_1.find_last_of(".");
+
+            std::string temp_name_2= temp_name_1.substr(0,file_ext2);
+
+            ((VDMS::Server*)(server))->untar_data(db_name);
+
+            config_file = temp_name_2+".json";
+
+        }
+        //  else if (option=="-backup" ){
+        //     backup=true;
+        //     config_file = std::string (argv[2]);
+
+
+        //  }
+
+
     }
 
+
+
+    //
     printf("Server will start processing requests... \n");
     VDMS::Server server(config_file);
 
     //create a thread for processing request and a thread for the autodelete timer
     request_thread_flag = pthread_create(&request_thread, NULL, start_request_thread, (void*)( &server ) );
     autodelete_thread_flag = pthread_create(&autodelete_thread, NULL, start_autodelete_thread, (void*)( &server ) );
+    auto_replcation_flag = pthread_create(&auto_replicate_thread, NULL, start_replication_thread, (void*)( &server ) );
+
+
     pthread_join(request_thread, NULL);
     pthread_join(autodelete_thread, NULL);
+    pthread_join(auto_replicate_thread, NULL);
+
+
+
 
     printf("Server shutting down... \n");
 

tests/cleandbs.sh

@@ -1,4 +1,4 @@
-rm -r jsongraph qhgraph simpleAdd_db simpleAddx10_db simpleUpdate_db entitycheck_db datatypecheck_db
+rm -r jsongraph qhgraph simpleAdd_db simpleAddx10_db simpleUpdate_db entitycheck_db datatypecheck_db db_backup
 
 rm -r tdb
 rm -r dbs
@@ -8,3 +8,5 @@ rm -r vdms
 rm images/tdb_to_jpg.jpg
 rm images/tdb_to_png.png
 rm images/test_image.jpg
+rm -r backups
+

tests/run_tests.sh

@@ -3,6 +3,8 @@ sh cleandbs.sh
 mkdir dbs  # necessary for Descriptors
 mkdir temp # necessary for Videos
 mkdir videos_tests
+mkdir backups
+
 
 echo 'not the vdms application - this file is needed for shared key' > vdms
 

tests/server/config-auto-replicate-tests.json

@@ -0,0 +1,10 @@
+{
+    "port": 55555,
+    "autoreplicate_interval":5,
+    "unit":"s",
+    "max_simultaneous_clients": 100,
+    "backup_path":"",
+    "db_root_path": "db_backup",
+    "backup_flag" : "true",
+    "more-info": "github.com/IntelLabs/vdms"
+}