test

src/centralstore/incremental/DataPublisher.cpp

@@ -37,9 +37,11 @@ DataPublisher::DataPublisher(int worker_port, std::string worker_address) {
     if ((sock = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
         data_publisher_logger.error("Socket creation error!");
     }
+    data_publisher_logger.info("Socket created");
     if (Utils::connect_wrapper(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {
         data_publisher_logger.error("Connection Failed!");
     }
+    data_publisher_logger.info("Connected to worker");
 }
 
 DataPublisher::~DataPublisher() { close(sock); }

src/frontend/JasmineGraphFrontEnd.cpp

@@ -120,12 +120,15 @@ void *frontendservicesesion(std::string masterIP, int connFd, SQLiteDBInterface
     KafkaConnector *kstream;
     Partitioner graphPartitioner(numberOfPartitions, 1, spt::Algorithms::HASH);
 
+    frontend_logger.log("Creating data publishers", "info");
+
     for (int i = 0; i < workerList.size(); i++) {
 
         Utils::worker currentWorker = workerList.at(i);
         string workerHost = currentWorker.hostname;
         string workerID = currentWorker.workerID;
         int workerPort = atoi(string(currentWorker.port).c_str());
+        frontend_logger.log("Worker Host : " + workerHost + "; Worker ID : " + workerID + "; Worker port : " + string(currentWorker.port), "info");
         DataPublisher *workerClient = new DataPublisher(workerPort, workerHost);
         workerClients.push_back(workerClient);
     }
@@ -142,10 +145,13 @@ void *frontendservicesesion(std::string masterIP, int connFd, SQLiteDBInterface
         }
 
         bzero(data, FRONTEND_DATA_LENGTH + 1);
+        frontend_logger.info("Reading ..");
         read(connFd, data, FRONTEND_DATA_LENGTH);
+        frontend_logger.info("Read done");
 
         string line(data);
         if (line.compare("\r\n") == 0) {
+            frontend_logger.info("Empty line");
             continue;
         }
         frontend_logger.log("Command received: " + line, "info");
@@ -1704,19 +1710,20 @@ int JasmineGraphFrontEnd::run() {
     int noThread = 0;
 
     while (true) {
-        frontend_logger.log("Frontend Listening", "info");
+        frontend_logger.log("Frontend Listening hello ... ", "info");
 
         //this is where client connects. svr will hang in this mode until client conn
         connFd = accept(listenFd, (struct sockaddr *) &clntAdd, &len);
 
         if (connFd < 0) {
-            frontend_logger.log("Cannot accept connection", "error");
+            frontend_logger.log("Cannot accept connection hello ...", "error");
             return 0;
         } else {
-            frontend_logger.log("Connection successful", "info");
+            frontend_logger.log("Connection successful hello ...", "info");
         }
 
         frontend_logger.log("Master IP" + masterIP, "info");
+        frontend_logger.log("Client IP" + std::string(inet_ntoa(clntAdd.sin_addr)), "info");
 
         struct frontendservicesessionargs *frontendservicesessionargs1 = (struct frontendservicesessionargs *) malloc(
                 sizeof(struct frontendservicesessionargs) * 1);;

src/util/Utils.cpp

@@ -596,12 +596,16 @@ std::string Utils::checkFlag(std::string flagPath){
 }
 
 int Utils::connect_wrapper(int sock, const sockaddr *addr, socklen_t slen) {
+    util_logger.info(string(inet_ntoa(((struct sockaddr_in *)addr)->sin_addr)));
     int retry = 0;
     do {
         if (retry) sleep(retry * 2);
+        util_logger.info("retry " + std::to_string(retry));
         if (connect(sock, addr, slen) == 0) {
+            util_logger.info("Connected to worker 12345");
             return 0;
         }
+        util_logger.info("Connection Failed! 12345");
     } while (retry++ < 4);
     return -1;
 }