path_rename

extensions/rag/index/src/qdrant.rs

@@ -3,6 +3,7 @@ use dagrs::{Action, Content, EnvVar, InChannels, OutChannels, Output};
 use qdrant_client::Qdrant;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
+use std::fs;
 
 use crate::utils::CodeItem;
 
@@ -59,7 +60,7 @@ impl QdrantNode {
             }
         }
     }
-
+    
     // async fn ensure_collection(&self) {
     //     if self
     //         .client
@@ -118,6 +119,13 @@ impl Action for QdrantNode {
                 processed_count += 1;
                 if processed_count % 100 == 0 {
                     log::info!("Processed {processed_count} items");
+                    // 保存当前ID到文件
+                    let current_id = self.id_counter.load(Ordering::SeqCst);
+                    if let Err(e) = fs::write("/opt/data/last_id.json", current_id.to_string()) {
+                        log::error!("Failed to save ID to file: {e}");
+                    } else {
+                        log::info!("Saved current ID {} to file", current_id);
+                    }
                 }
             }
             out_channels.broadcast(Content::new(())).await;

extensions/update/.env

@@ -6,5 +6,5 @@ MEGA_URL=http://git.gitmega.nju:30080
 KAFKA_BROKER=172.17.0.1:30092
 KAFKA_TOPIC=ANALYSIS_TEST
 KAFKA_TOPIC_INDEX=INDEX_TEST
-ENABLE_CONCURRENCY=true
+ENABLE_CONCURRENCY=false
 MAX_WORKERS=32

extensions/update/deploy-ext-update.sh

@@ -0,0 +1,59 @@
+#!/bin/bash
+set -euxo pipefail
+
+# source
+CURRENT_DIR=$(pwd)
+MEGA_ROOT_DIR=$(dirname $(dirname "$CURRENT_DIR"))
+# deployment
+NAMESPACE=mega-rag
+CRONJOB_NAME=rag-update-cronjob
+CONTAINER_NAME=rag-update
+# build
+TIMESTAMP=$(date +%Y%m%d-%H%M)
+IMAGE_NAME=localhost:30500/mega-rag-update:local-$TIMESTAMP
+
+### Step 0: Verify we're in the correct directory
+if [[ ! "$CURRENT_DIR" == *"/mega/extensions/update" ]]; then
+    echo "Error: Must be run from .../mega/extensions/update directory"
+    exit 1
+fi
+
+### Step 1: Build Docker image
+docker build -t $IMAGE_NAME -f $CURRENT_DIR/Dockerfile $MEGA_ROOT_DIR
+
+### Step 2: Push Docker image
+docker push $IMAGE_NAME
+
+### Step 3: Update cronjob image
+kubectl set image cronjob/$CRONJOB_NAME -n $NAMESPACE $CONTAINER_NAME=$IMAGE_NAME
+
+### Step 4: Stop all running jobs from this cronjob and related pods
+# Get all jobs created by this cronjob
+JOBS=$(kubectl get jobs -n $NAMESPACE -o json | jq -r ".items[] | select(.metadata.ownerReferences[]?.name==\"$CRONJOB_NAME\") | .metadata.name")
+
+# Delete all running jobs
+for job in $JOBS; do
+    echo "Deleting job: $job"
+    kubectl delete job $job -n $NAMESPACE --ignore-not-found=true
+done
+
+# Delete all pods with label app=rag-update
+echo "Deleting pods with label app=rag-update..."
+kubectl delete pods -n $NAMESPACE -l app=rag-update --ignore-not-found=true
+
+# Wait until all jobs are terminated
+while kubectl get jobs -n $NAMESPACE -o json | jq -r ".items[] | select(.metadata.ownerReferences[]?.name==\"$CRONJOB_NAME\") | .metadata.name" | grep . > /dev/null; do
+    echo "Waiting for jobs to terminate..."
+    sleep 5
+done
+
+# Wait until all pods with label app=rag-update are terminated
+while kubectl get pods -n $NAMESPACE -l app=rag-update --no-headers 2>/dev/null | grep . > /dev/null; do
+    echo "Waiting for pods to terminate..."
+    sleep 5
+done
+
+### Step 5: Trigger a new job immediately
+kubectl create job --from=cronjob/$CRONJOB_NAME -n $NAMESPACE manual-$(date +%Y%m%d-%H%M%S)
+
+echo "Deployment completed successfully!"

extensions/update/do_utils.py

@@ -9,10 +9,11 @@ def update_repo_sync_result(crate_name, version, mega_url, status: SyncStatusEnu
     try:
         record = db.query(RepoSyncResult).filter_by(crate_name=crate_name).first()
         if record:
-            record.version = version
+            if record.version is None or vparse(version) > vparse(record.version):
+                record.version = version
+                record.mega_url = mega_url
             record.status = status
             record.err_message = err_message
-            record.mega_url = mega_url
             record.updated_at = datetime.utcnow()
         else:
             record = RepoSyncResult(

extensions/update/model.py

@@ -16,7 +16,7 @@ class SyncStatusEnum(str, enum.Enum):
     ANALYSED = "Analysed"
 
 class RepoSyncResult(Base):
-    __tablename__ = "repo_sync_result"
+    __tablename__ = "repo_sync_result_test"
 
     id = Column(Integer, primary_key=True, index=True)
     crate_name = Column(String, unique=True, nullable=False)   # 唯一约束