Raw kg contruction

.github/workflows/build.yml

@@ -65,6 +65,21 @@ jobs:
     - name: Docker Build
       run: docker build -t jasminegraph .
 
+    - uses: actions/setup-python@v4
+      with:
+        python-version: 3.11
+    - name: Install python dependencies Build
+      run: |
+        pip install fastapi uvicorn httpx
+
+      # 4. Start mock Ollama server in the background
+    - name: Start Mock LLM Server
+      run: |
+        nohup uvicorn mock_ollama_server:app --host 0.0.0.0 --port 11450 &
+
+      # 5. Wait a few seconds for server to be ready
+    - run: sleep 2
+
     - name: Run Integration Tests
       run: |
         chmod +x test-docker.sh
@@ -129,62 +144,3 @@ jobs:
           mkdir /var/tmp/worker0 /var/tmp/worker1
           chmod +x test-k8s.sh
           TEST_NAME="main" ./test-k8s.sh
-
-  k8s-integration-tests-graph-rag:
-    runs-on: ubuntu-latest
-    timeout-minutes: 20
-    if: ${{!contains(github.event.pull_request.labels.*.name, 'Skip k8s integration')}}
-
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          ref: ${{github.head_ref}}
-          repository: ${{ github.event.pull_request.head.repo.full_name || github.repository }}
-
-      - uses: jupyterhub/action-k3s-helm@v4
-        with:
-          docker-enabled: true
-          network-policy: false
-
-      - name: Show system resources and StorageClass
-        run: |
-          echo "=== CPU Info ==="
-          lscpu | grep -E '^CPU\(s\)|^Model name'
-          echo
-          echo "=== Memory Info ==="
-          free -h
-          echo
-          echo "=== Disk Info ==="
-          df -h
-          echo
-          echo "=== Kubernetes StorageClasses ==="
-          kubectl get storageclass
-          kubectl describe storageclass
-
-      - name: Grant permissions for default user
-        run: kubectl apply -f ./k8s/rbac.yaml
-
-      - uses: actions/setup-python@v4
-        with:
-          python-version: 3.11
-      - name: Install python dependencies Build
-        run: |
-          pip install fastapi uvicorn httpx
-
-      # 4. Start mock Ollama server in the background
-      - name: Start Mock LLM Server
-        run: |
-          nohup uvicorn mock_ollama_server:app --host 0.0.0.0 --port 11450 &
-
-      # 5. Wait a few seconds for server to be ready
-      - run: sleep 2
-
-      - name: Docker Build
-        run: docker build -t jasminegraph .
-
-
-      - name: K8S integration tests
-        run: |
-          mkdir /var/tmp/worker0 /var/tmp/worker1
-          chmod +x test-k8s.sh
-          TEST_NAME="graphRAG" ./test-k8s.sh

CMakeLists.txt

@@ -252,8 +252,6 @@ if(NOT GENERATED_SRC)
     set(GENERATED_SRC ""
             src/frontend/JasmineGraphFrontEnd.cpp
             src/localstore/incremental/JasmineGraphIncrementalLocalStore.cpp
-            src/localstore/incremental/JasmineGraphIncrementalLocalStore.cpp
-            src/localstore/incremental/JasmineGraphIncrementalLocalStore.cpp
             src/frontend/JasmineGraphFrontEnd.cpp)
 endif()
 

conf/jasminegraph-server.properties

@@ -108,7 +108,7 @@ org.jasminegraph.autopartition.enabled=false
 #--------------------------------------------------------------------------------
 
 #This parameter holds the maximum label size of Node Block
-org.jasminegraph.nativestore.max.label.size=256
+org.jasminegraph.nativestore.max.label.size=43
 
 
 # knowledge graph construction
@@ -133,5 +133,11 @@ org.jasminegraph.query.threadpool.maxworkers=32
 org.jasminegraph.vectorstore.enabled=true
 org.jasminegraph.vectorstore.dimension=512
 org.jasminegraph.vectorstore.embedding.model=jina/jina-embeddings-v2-small-en
-org.jasminegraph.vectorstore.embedding.ollama.endpoint=http://gemma3_container:11441
+org.jasminegraph.vectorstore.embedding.ollama.endpoint=http://gemma3:11434
+#
+#org.jasminegraph.vectorstore.dimension=768
+#org.jasminegraph.vectorstore.embedding.model=nomic-embed-text
+#org.jasminegraph.vectorstore.embedding.ollama.endpoint=https://sajeenthiranp-21--nomic-embedder-embeddings-serve.modal.run
+##org.jasminegraph.vectorstore.embedding.ollama.endpoint=http://192.168.1.19:11441
+##org.jasminegraph.vectorstore.embedding.ollama.endpoint=http://10.8.100.248:11441
 #

conf/prometheus.yaml

@@ -6,10 +6,10 @@ scrape_configs:
   - job_name: "prometheus"
     scrape_interval: 5s
     static_configs:
-      - targets: ["prometheus:9090"]
+      - targets: ["10.8.100.248:9090"]
 
   - job_name: "pushgateway"
     scrape_interval: 2s
     static_configs:
-      - targets: ["pushgateway:9091"]
+      - targets: ["10.8.100.248:9091"]
 

mock_ollama_server.py

@@ -1,4 +1,4 @@
-"""Copyright 2025 JasmineGraph Team
+"""Copyright 2026 JasmineGraph Team
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
@@ -40,8 +40,8 @@ async def streamer( model: str):
 
     # Example array-of-arrays tuples
     tuples = [
-        ["Radio City", "is", "India's first private FM radio station"],
-        ["Radio City", "was started on", "3 July 2001", "Organization", "Date"],
+        ["Radio City", "is", "India's first private FM radio station" ,"Organization", "Description"],
+        ["Radio City", "was started on", "3 July 2001", "Organization", "Date", ],
         ["Radio City", "broadcasts on", "91.1", "Organization", "Frequency"]
     ]
 

src/frontend/JasmineGraphFrontEnd.cpp

@@ -2037,6 +2037,45 @@ bool JasmineGraphFrontEnd::constructKGStreamHDFSCommand(std::string masterIP, in
     std::string hdfsFilePathS(hdfsFilePath);
     hdfsFilePathS = Utils::trim_copy(hdfsFilePathS);
 
+    CURL* curl = curl_easy_init();
+    if (!curl) {
+        frontend_logger.error("Failed to initialize CURL");
+        *loop_exit_p = true;
+        return false;
+    }
+
+    std::string url =
+        "http://" + hdfsServerIp + ":9870/webhdfs/v1/?op=GETHOMEDIRECTORY";
+
+    long http_code = 0;
+
+    curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
+    curl_easy_setopt(curl, CURLOPT_NOBODY, 1L);          // no body
+    curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 10L);   // wait up to 10s to connect
+    curl_easy_setopt(curl, CURLOPT_TIMEOUT, 30L);        // max 30s for whole request
+    curl_easy_setopt(curl, CURLOPT_FAILONERROR, 1L);    // HTTP 4xx/5xx => error
+    curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L);       // avoid DNS delays
+
+    CURLcode res = curl_easy_perform(curl);
+    curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &http_code);
+    curl_easy_cleanup(curl);
+
+    // Immediate failure
+    if (res != CURLE_OK || http_code == 0) {
+        frontend_logger.error(
+            "HDFS file System Not reachable at: " +
+            hdfsServerIp + " port: " + hdfsPort);
+        frontend_logger.error("CURL response code: " +
+                                               std::to_string(res));
+        std::string error_message = "HDFS file System Not reachable.";
+        write(connFd, error_message.c_str(), error_message.length());
+        write(connFd,
+              Conts::CARRIAGE_RETURN_NEW_LINE.c_str(),
+              Conts::CARRIAGE_RETURN_NEW_LINE.size());
+
+        *loop_exit_p = true;
+        return false;
+    }
     HDFSConnector* hdfsConnector = new HDFSConnector(hdfsServerIp, hdfsPort);
 
     if (!hdfsConnector->isPathValid(hdfsFilePathS)) {
@@ -2051,11 +2090,15 @@ bool JasmineGraphFrontEnd::constructKGStreamHDFSCommand(std::string masterIP, in
 
     std::string path = "hdfs:" + hdfsFilePathS;
     double_t total_file_size = hdfsGetPathInfo(hdfsConnector->getFileSystem(), hdfsFilePathS.c_str())->mSize;
-    std::time_t time = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
-    std::string uploadStartTime = ctime(&time);
+    std::time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
+    struct tm localTime;
+    localtime_r(&now, &localTime);  // thread-safe version of localtime()
+    char buffer[100];
+    strftime(buffer, sizeof(buffer), "%Y-%m-%d %H:%M:%S", &localTime);
+    std::string uploadStartTime(buffer);
+
 
     // 2. Prepare new graph insertion
-    std::time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
     uploadStartTime.erase(uploadStartTime.find_last_not_of(Conts::CARRIAGE_RETURN_NEW_LINE) + 1);  // remove newline
 
     std::string llmRunnerMSG = "LLM runner hostname:port: ";
@@ -2078,6 +2121,12 @@ bool JasmineGraphFrontEnd::constructKGStreamHDFSCommand(std::string masterIP, in
     std::string hostnamePortS(hostnamePort);
     hostnamePortS = Utils::trim_copy(hostnamePortS);
 
+    if (hostnamePortS.find("exit") != std::string::npos) {
+        *loop_exit_p = true;
+        return false;
+    }
+
+
     frontend_logger.info("Received LLM runners: " + hostnamePortS);
 
     std::string llmInferenceMSG = "LLM inference engine? ollama/vllm? ";
@@ -2102,6 +2151,49 @@ bool JasmineGraphFrontEnd::constructKGStreamHDFSCommand(std::string masterIP, in
 
     frontend_logger.info("received Inference Engine: " + llmInferenceEngineS);
 
+    vector<std::string> llmServers = Utils::getUniqueLLMRunners(hostnamePortS);
+
+    for (auto llmServer : llmServers) {
+        std::string url;
+        bool modelFound = false;
+        std::string endpointPath;
+        if (llmInferenceEngineS == "ollama") {
+            endpointPath = "api/tags";
+        } else if (llmInferenceEngineS == "vllm") {
+            endpointPath = "/v1/models";
+        } else {
+            frontend_logger.error("Unknown inference engine: " + llmInferenceEngineS);
+            std::string msg = "Unknown inference engine '" + llmInferenceEngineS + "'";
+            write(connFd, msg.c_str(), msg.length());
+            write(connFd, Conts::CARRIAGE_RETURN_NEW_LINE.c_str(), Conts::CARRIAGE_RETURN_NEW_LINE.size());
+            *loop_exit_p = true;
+            return false;
+        }
+
+        url = Utils::normalizeURL(llmServer, endpointPath);
+        frontend_logger.info("Final LLM endpoint: " + url);
+
+        CURL* curl = curl_easy_init();
+        if (curl) {
+            std::string response;
+            curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
+            curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, WriteCallback);
+            curl_easy_setopt(curl, CURLOPT_WRITEDATA, &response);
+            curl_easy_setopt(curl, CURLOPT_TIMEOUT, 5L);
+
+            CURLcode res = curl_easy_perform(curl);
+            curl_easy_cleanup(curl);
+
+            if (res != CURLE_OK) {
+                frontend_logger.error("Failed to reach " + llmInferenceEngineS + " server at " + llmServer);
+                std::string msg = "Could not connect to " + llmInferenceEngineS + " server.";
+                write(connFd, msg.c_str(), msg.length());
+                write(connFd, Conts::CARRIAGE_RETURN_NEW_LINE.c_str(), Conts::CARRIAGE_RETURN_NEW_LINE.size());
+                *loop_exit_p = true;
+                return false;
+            }
+        }
+    }
     std::string LLM_MSG = "What is the LLM you want to use?:";
     resultWr = write(connFd, LLM_MSG.c_str(), LLM_MSG.length());
     if (resultWr < 0) {
@@ -2122,9 +2214,10 @@ bool JasmineGraphFrontEnd::constructKGStreamHDFSCommand(std::string masterIP, in
     std::string llmS(llm);
     llmS = Utils::trim_copy(llmS);
     frontend_logger.info("Received LLM " + llmS);
-
-    vector<std::string> llmServers = Utils::getUniqueLLMRunners(hostnamePortS);
-
+    if (llmS.find("exit") != std::string::npos) {
+        *loop_exit_p = true;
+        return false;
+    }
     for (auto llmServer : llmServers) {
         std::string url;
         bool modelFound = false;
@@ -2338,7 +2431,11 @@ bool JasmineGraphFrontEnd::constructKGStreamHDFSCommand(std::string masterIP, in
         }
 
         std::time_t time = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());
-        std::string uploadEndTime = ctime(&time);
+        struct tm localTime;
+        localtime_r(&time, &localTime);
+        char buffer[100];
+        strftime(buffer, sizeof(buffer), "%Y-%m-%d %H:%M:%S", &localTime);
+        std::string uploadEndTime(buffer);
 
         std::string sqlStatementUpdateEndTime = "UPDATE graph SET upload_end_time = \"" + uploadEndTime +
                                                 "\" WHERE idgraph = " + std::to_string(newGraphID);

src/frontend/ui/JasmineGraphFrontEndUI.cpp

@@ -1182,6 +1182,9 @@ static void semantic_beam_search_command(int connFd, std::string command, int nu
         *loop_exit_p = true;
         return;
     }
+    string done =  R"({"done":"true"})";
+    write(connFd,  done.c_str(), done.size());
+    write(connFd, Conts::CARRIAGE_RETURN_NEW_LINE.c_str(), Conts::CARRIAGE_RETURN_NEW_LINE.size());
 
     ui_frontend_logger.info("Semantic beam search completed successfully for graph " + graph_id);
 }

src/knowledgegraph/construction/Pipeline.cpp

@@ -457,9 +457,9 @@ json Pipeline::processTupleAndSaveInPartition(const std::vector<std::unique_ptr<
     using namespace std::chrono;
 
     auto nextTick = steady_clock::now();
-
-    while (metaThreadRunning.load(std::memory_order_relaxed)) {
         std::this_thread::sleep_for(std::chrono::milliseconds(10000));
+    while (metaThreadRunning.load(std::memory_order_relaxed)) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(60000));
 
         kg_pipeline_stream_handler_logger.debug("Meta thread running");
 
@@ -804,7 +804,7 @@ void Pipeline::extractTuples(std::string host, int port, std::string masterIP, i
 
             char ack3[ACK_MESSAGE_SIZE] = {0};
             int converted_number = htonl(chunk.length());
-            kg_pipeline_stream_handler_logger.info("Sending chunk length: " +
+            kg_pipeline_stream_handler_logger.debug("Sending chunk length: " +
                 std::to_string(chunk.length()));
             if (!Utils::sendIntExpectResponse(sockfd, ack3,
                                               JasmineGraphInstanceProtocol::GRAPH_STREAM_C_length_ACK.length(),
@@ -816,7 +816,7 @@ void Pipeline::extractTuples(std::string host, int port, std::string masterIP, i
                 break;
                                               }
 
-            kg_pipeline_stream_handler_logger.info("Sending chunk data");
+            kg_pipeline_stream_handler_logger.debug("Sending chunk data");
             if (!Utils::send_str_wrapper(sockfd, chunk)) {
                 kg_pipeline_stream_handler_logger.error("Failed to send chunk data");
                 retry = true;
@@ -840,7 +840,7 @@ void Pipeline::extractTuples(std::string host, int port, std::string masterIP, i
                 break;
                                               }
 
-            kg_pipeline_stream_handler_logger.info("Sending currentTraceContext data:" +currentTraceContext);
+            kg_pipeline_stream_handler_logger.debug("Sending currentTraceContext data:" +currentTraceContext);
             if (!Utils::send_str_wrapper(sockfd, currentTraceContext)) {
                 kg_pipeline_stream_handler_logger.error("Failed to send chunk data");
                 retry = true;

src/knowledgegraph/construction/VLLMTupleStreamer.cpp

@@ -102,7 +102,7 @@ size_t VLLMTupleStreamer::StreamCallback(char* ptr, size_t size, size_t nmemb,
                 auto triple = json::parse(ctx->current_tuple);
 
                   if (!triple.is_array() || triple.size() < 5) {
-                      vllm_tuple_streamer_logger.error(
+                      vllm_tuple_streamer_logger.warn(
                           "Invalid tuple size detected. Retrying entire chunk.");
 
                       ctx->isSuccess = false;
@@ -158,10 +158,10 @@ size_t VLLMTupleStreamer::StreamCallback(char* ptr, size_t size, size_t nmemb,
                       "✅ Added formatted triple: " + formattedTriple.dump());
                 }
               } catch (const std::exception& ex) {
-                vllm_tuple_streamer_logger.error(
+                vllm_tuple_streamer_logger.warn(
                     "❌ JSON array parse failed: " + std::string(ex.what()) + ". Invalid Tuple: " + std::string
                     (ctx->current_tuple));
-                  vllm_tuple_streamer_logger.error(
+                  vllm_tuple_streamer_logger.warn(
                          "Invalid tuple  detected. Retrying entire chunk.");
 
                   ctx->isSuccess = false;
@@ -176,7 +176,7 @@ size_t VLLMTupleStreamer::StreamCallback(char* ptr, size_t size, size_t nmemb,
         }
       }
     } catch (const std::exception& ex) {
-      vllm_tuple_streamer_logger.debug("JSON parse error: " +
+      vllm_tuple_streamer_logger.warn("JSON parse error: " +
                                        std::string(ex.what()));
     }
   }
@@ -261,7 +261,7 @@ void VLLMTupleStreamer::streamChunk(const std::string& chunkKey,
     jsonRequest["max_tokens"] = 10000;
 
     std::string postFields = jsonRequest.dump();
-    vllm_tuple_streamer_logger.info("Post fields: " + postFields);
+    // vllm_tuple_streamer_logger.info("Post fields: " + postFields);
     ctx.current_tuple = "";
     curl_easy_setopt(curl, CURLOPT_POSTFIELDS, postFields.c_str());
     curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, postFields.size());