feature: 支持启动时安装算子包

Author: hhhhsc701

tools/install.sh

@@ -39,6 +39,7 @@ OPERATOR_PVC=""
 DATASET_PVC=""
 PORT="30000"
 ADDRESS_TYPE="management"
+PACKAGE_PATH=""
 
 
 cd "$(dirname "$0")" || exit
@@ -223,6 +224,12 @@ function add_route_to_haproxy() {
     log_info "Finish config haproxy"
 }
 
+function install_package() {
+  if [[ -n $PACKAGE_PATH ]]; then
+    bash "$UTILS_PATH/load_operators.sh" "$NAMESPACE" "$PACKAGE_PATH"
+  fi
+}
+
 function main() {
   while [[ "$#" -gt 0 ]]; do
     case $1 in
@@ -237,6 +244,7 @@ function main() {
       --skip-push) SKIP_PUSH=true; shift ;;
       --skip-load) SKIP_LOAD=true; shift ;;
       --skip-milvus) INSTALL_MILVUS=false; shift ;;
+      --package) PACKAGE_PATH="$2"; shift 2 ;;
       -h|--help) print_help "${SCRIPT_PATH}"; exit 0 ;;
       *) log_info "错误: 未知参数: $1"; shift ;;
     esac
@@ -248,6 +256,9 @@ function main() {
   [ "$INSTALL_MILVUS" == "true" ] && load_images "milvus"
   install
   add_route_to_haproxy
+
+  kubectl wait --for=condition=Ready pod -l app.kubernetes.io/instance=datamate -n "$NAMESPACE" --timeout=300s >/dev/null
+  bash "$UTILS_PATH/load_images.sh"
 }
 
 main "$@"

tools/uninstall.sh

@@ -82,6 +82,8 @@ function main() {
 
   uninstall
   remove_route_from_haproxy
+
+  kubectl wait --for=delete pod -l app.kubernetes.io/instance=datamate -n "$NAMESPACE" --timeout=300s >/dev/null
 }
 
 main "$@"

tools/utils/load_operators.sh

@@ -29,18 +29,18 @@ fi
 WORK_DIR="$current_dir/operators"
 
 # 退出时自动清理
-cleanup() {
-    log_info "清理临时文件..."
-    rm -rf "$WORK_DIR"
-}
-trap cleanup EXIT
+#cleanup() {
+#    log_info "清理临时文件..."
+#    rm -rf "$WORK_DIR"
+#}
+#trap cleanup EXIT
 
 # 2. 处理输入源（解压或复制到工作区）
 SOURCE_DIR="$WORK_DIR/source"
 mkdir -p "$SOURCE_DIR"
 
 if [ -d "$INPUT_PATH" ]; then
-    log_info "输入为目录，复制文件..."
+    log_info "输入为目录，跳过解压..."
     SOURCE_DIR=$INPUT_PATH
 elif [[ "$INPUT_PATH" == *.zip ]]; then
     log_info "输入为zip包，解压中..."
@@ -78,9 +78,11 @@ while read -r pkg; do
 
     # 解压子包
     if [[ "$pkg" == *.zip ]]; then
-        unzip -q "$pkg" -d "$PKG_EXTRACT_DIR"
-    else
+        unzip -oq "$pkg" -d "$PKG_EXTRACT_DIR"
+    elif [[ "$pkg" == *.tar ]]; then
         tar -xf "$pkg" -C "$PKG_EXTRACT_DIR"
+    else
+      continue
     fi
 
     # 检查关键文件是否存在
@@ -93,24 +95,24 @@ while read -r pkg; do
     # 假设每个包在目标目录下应该有一个独立的文件夹
     REMOTE_PATH="$EXTRACT_DIR/$PKG_BASE"
 
-    log_info "  -> 部署文件到容器 $BACKEND_POD_NAME:$REMOTE_PATH"
+    log_info "拷贝目录${PKG_BASE}到容器..."
 
-#    kubectl cp "$SOURCE_DIR/$PKG_NAME" "$BACKEND_POD_NAME:$UPLOAD_DIR/" -n "$NAMESPACE"
-    kubectl cp "$PKG_EXTRACT_DIR" "$BACKEND_POD_NAME:$REMOTE_PATH/" -n "$NAMESPACE"
+#    kubectl cp "$SOURCE_DIR/$PKG_NAME" "$BACKEND_POD_NAME:$UPLOAD_DIR/$PKG_BASE/" -n "$NAMESPACE"
+    kubectl exec "$BACKEND_POD_NAME" -n "$NAMESPACE" -- sh -c "rm -rf $REMOTE_PATH/ && mkdir -p $REMOTE_PATH"
+    kubectl cp "$PKG_EXTRACT_DIR/." "$BACKEND_POD_NAME:$REMOTE_PATH/" -n "$NAMESPACE"
 
-    if [ -f "$PKG_EXTRACT_DIR/wheels" ]; then
+    if [ -d "$PKG_EXTRACT_DIR/wheels" ]; then
+      log_info "安装算子依赖..."
       kubectl exec "$HEAD_POD_NAME" -n "$NAMESPACE" -- bash -c "uv pip install --target $PACKAGE_DIR /opt/runtime/datamate/ops/user/$PKG_BASE/wheels/*.whl"
     fi
 done < <(find "$SOURCE_DIR" -maxdepth 1 -type f \( -name "*.zip" -o -name "*.tar" \))
 
 
-FULL_SQL=$(kubectl exec -i "$HEAD_POD_NAME" -n "$NAMESPACE" -- python3 - << EOF
+FULL_SQL=$(kubectl exec -i "$HEAD_POD_NAME" -n "$NAMESPACE" -c ray-head -- python3 - << EOF
 from pathlib import Path
-import sys, yaml
+import json, sys, yaml
 
-operator_sql='INSERT IGNORE INTO t_operator
-(id, name, description, version, inputs, outputs, runtime, settings, file_name, is_star)
-VALUES '
+operator_sql = 'INSERT IGNORE INTO t_operator (id, name, description, version, inputs, outputs, runtime, settings, file_name, is_star) VALUES '
 category_sql='INSERT IGNORE INTO t_operator_category_relation(category_id, operator_id) VALUES '
 modal_map = {
     'text': 'd8a5df7a-52a9-42c2-83c4-01062e60f597',
@@ -129,27 +131,33 @@ for metadata_file in base_path.rglob('metadata.yml'):
     try:
         with open(metadata_file, 'r') as f:
             data = yaml.safe_load(f)
-            id = data.get('raw_id')
-            name = data.get('name')
-            desc = data.get('description')
-            version = data.get('version')
-            modal = data.get('modal').lower()
-            language = data.get('language').lower()
-            inputs = data.get('inputs')
-            outputs = data.get('outputs')
+            id = data.get('raw_id').strip("'\"")
+            name = data.get('name').strip("'\"")
+            desc = data.get('description').strip("'\"")
+            version = data.get('version').strip("'\"")
+            modal = data.get('modal').lower().strip("'\"")
+            language = data.get('language').lower().strip("'\"")
+            inputs = data.get('inputs').strip("'\"")
+            outputs = data.get('outputs').strip("'\"")
+            runtime = f"'{json.dumps(data.get('runtime'), ensure_ascii=False)}'" if 'runtime' in data else 'null'
+            settings = f"'{json.dumps(data.get('settings'), ensure_ascii=False)}'" if 'settings' in data else 'null'
             file_name = Path(metadata_file).parent.name
 
-            operator_sql += "('{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', '{}', 'false'),".format(id, name, desc, version, inputs, outputs, runtime, settings, file_name)
-            category_sql += "('{}', '{}'),".format(id, modal_map.get(modal, 'd8a5df7a-52a9-42c2-83c4-01062e60f597'))
-            category_sql += "('{}', '{}'),".format(id, language_map.get(language, '9eda9d5d-072b-499b-916c-797a0a8750e1'))
+            operator_sql += "('{}', '{}', '{}', '{}', '{}', '{}', {}, {}, '{}', 'false'),".format(id, name, desc, version, inputs, outputs, runtime, settings, file_name)
+            category_sql += "('{}', '{}'),".format(modal_map.get(modal, 'd8a5df7a-52a9-42c2-83c4-01062e60f597'), id)
+            category_sql += "('{}', '{}'),".format(language_map.get(language, '9eda9d5d-072b-499b-916c-797a0a8750e1'), id)
+            category_sql += "('ec2cdd17-8b93-4a81-88c4-ac9e98d10757', '{}'),".format(id)
     except Exception as e:
         print(f'ERROR: {e}', file=sys.stderr)
         sys.exit(1)
     print(operator_sql[:-1] + ';\n' + category_sql[:-1] + ';')
 EOF
 )
 
+log_info "插入数据库..."
 DATABASE_POD_NAME=$(kubectl get pod -n "$NAMESPACE" -l app.kubernetes.io/name=datamate-database -o jsonpath='{.items[*].metadata.name}')
-kubectl exec -i "$DATABASE_POD_NAME" -n "$NAMESPACE" -- mysql -uroot -p"\$MYSQL_ROOT_PASSWORD" "datamate" -e "$FULL_SQL" 2>/dev/null
+kubectl exec -i "$DATABASE_POD_NAME" -n "$NAMESPACE" -c database -- sh -c 'MYSQL_PWD="$MYSQL_ROOT_PASSWORD" mysql -uroot datamate' << EOF
+$FULL_SQL
+EOF
 
 log_info "所有任务执行完毕。"