alibaba
diff --git a/‎rock/admin/main.py‎
Lines changed: 10 additions & 1 deletion b/‎rock/admin/main.py‎
Lines changed: 10 additions & 1 deletion
diff --git a/‎rock/admin/scheduler/scheduler.py‎
Lines changed: 141 additions & 30 deletions b/‎rock/admin/scheduler/scheduler.py‎
Lines changed: 141 additions & 30 deletions
diff --git a/‎rock/admin/scheduler/task_base.py‎
Lines changed: 41 additions & 0 deletions b/‎rock/admin/scheduler/task_base.py‎
Lines changed: 41 additions & 0 deletions
diff --git a/‎rock/admin/scheduler/task_factory.py‎
Lines changed: 3 additions & 32 deletions b/‎rock/admin/scheduler/task_factory.py‎
Lines changed: 3 additions & 32 deletions
@@ -21,7 +21,7 @@
 from rock.admin.entrypoints.warmup_api import set_warmup_service, warmup_router
 from rock.admin.gem.api import gem_router, set_env_service
 from rock.admin.scheduler.scheduler import SchedulerThread
-from rock.config import DatabaseConfig, RockConfig
+from rock.config import DatabaseConfig, RockConfig, SchedulerConfig
 from rock.logger import init_logger
 from rock.sandbox.gem_manager import GemManager
 from rock.sandbox.operator.factory import OperatorContext, OperatorFactory
@@ -51,6 +51,14 @@ async def lifespan(app: FastAPI):
         else env_vars.ROCK_CONFIG
     )
     rock_config = RockConfig.from_env(config_file_path)
+
+    # Override scheduler config from Nacos if available
+    if rock_config.nacos_provider:
+        nacos_config = await rock_config.nacos_provider.get_config()
+        if nacos_config and "scheduler" in nacos_config:
+            rock_config.scheduler = SchedulerConfig(**nacos_config["scheduler"])
+            logger.info(f"Overrode scheduler config from Nacos with {len(rock_config.scheduler.tasks)} tasks")
+
     env_vars.ROCK_ADMIN_ENV = args.env
     env_vars.ROCK_ADMIN_ROLE = args.role
 
@@ -128,6 +136,7 @@ async def lifespan(app: FastAPI):
         if rock_config.scheduler.enabled and is_primary_pod():
             scheduler_thread = SchedulerThread(
                 scheduler_config=rock_config.scheduler,
+                nacos_provider=rock_config.nacos_provider,
             )
             scheduler_thread.start()
             logger.info("Scheduler thread started on primary pod")
 
@@ -1,19 +1,23 @@
 # rock/admin/scheduler/scheduler.py
 import asyncio
+import hashlib
+import json
 import threading
 import time
+from dataclasses import asdict
 from datetime import datetime, timedelta
 
 import pytz
 import ray
+import yaml
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
 from rock import env_vars
 from rock.admin.scheduler.task_base import BaseTask
-from rock.admin.scheduler.task_registry import TaskRegistry
 from rock.common.constants import SCHEDULER_LOG_NAME
-from rock.config import SchedulerConfig
+from rock.config import SchedulerConfig, TaskConfig
 from rock.logger import init_logger
+from rock.utils.providers import NacosConfigProvider
 
 logger = init_logger(name="scheduler", file_name=SCHEDULER_LOG_NAME)
 
@@ -65,27 +69,140 @@ def get_alive_workers(self, force_refresh: bool = False) -> list[str]:
 
 
 class TaskScheduler:
-    """Manages task scheduling using APScheduler."""
+    """Manages task scheduling using APScheduler with optional Nacos dynamic config."""
 
-    def __init__(self, scheduler_config: SchedulerConfig):
+    def __init__(
+        self,
+        scheduler_config: SchedulerConfig,
+        nacos_provider: NacosConfigProvider | None = None,
+    ):
         self.scheduler_config = scheduler_config
         self.local_tz = pytz.timezone(env_vars.ROCK_TIME_ZONE)
         self._scheduler: AsyncIOScheduler | None = None
         self._stop_event: asyncio.Event | None = None
         self._worker_cache: WorkerIPCache | None = None
-        self._loop: asyncio.AbstractEventLoop | None = None
+        self._nacos_provider = nacos_provider
+        self._event_loop: asyncio.AbstractEventLoop | None = None
+        self._last_scheduler_config_hash: str | None = None
+        self._task_hashes: dict[str, str] = {}
+        self._tasks_by_class: dict[str, BaseTask] = {}
 
     def _init_worker_cache(self) -> None:
         """Initialize the worker IP cache."""
         self._worker_cache = WorkerIPCache(
             cache_ttl=self.scheduler_config.worker_cache_ttl,
         )
 
-    def _register_tasks(self) -> None:
-        """Register all tasks from configuration."""
+    def _on_nacos_config_changed(self, new_config: dict) -> None:
+        """Callback invoked by Nacos watcher when config changes (runs in Nacos polling thread)."""
+        logger.info("Nacos config changed, checking scheduler section...")
+        try:
+            config_dict = yaml.safe_load(new_config["content"])
+            if not config_dict or "scheduler" not in config_dict:
+                logger.warning("No 'scheduler' section in updated Nacos config, skipping")
+                return
+
+            # Compare scheduler section hash to avoid unnecessary reloads
+            scheduler_raw = json.dumps(config_dict["scheduler"], sort_keys=True)
+            config_hash = hashlib.md5(scheduler_raw.encode()).hexdigest()
+            if config_hash == self._last_scheduler_config_hash:
+                logger.info("Scheduler config unchanged, skipping reload")
+                return
+            self._last_scheduler_config_hash = config_hash
+
+            new_scheduler_config = SchedulerConfig(**config_dict["scheduler"])
+            # Schedule the async reload on the event loop (thread-safe)
+            if self._event_loop and self._event_loop.is_running():
+                asyncio.run_coroutine_threadsafe(self._reload_scheduler_config(new_scheduler_config), self._event_loop)
+            else:
+                logger.warning("Event loop not available, cannot reload tasks dynamically")
+        except yaml.YAMLError as e:
+            logger.error(f"Failed to parse updated Nacos YAML config: {e}")
+        except Exception as e:
+            logger.error(f"Failed to process Nacos config change: {e}")
+
+    async def _reload_scheduler_config(self, new_scheduler_config: SchedulerConfig) -> None:
+        """Reload scheduler config by clearing and rebuilding all tasks."""
+        self.scheduler_config = new_scheduler_config
+        self._worker_cache.cache_ttl = new_scheduler_config.worker_cache_ttl
+        await self._rebuild_tasks()
+
+    @staticmethod
+    def _compute_task_hash(task_config: TaskConfig) -> str:
+        raw = json.dumps(asdict(task_config), sort_keys=True)
+        return hashlib.md5(raw.encode()).hexdigest()
+
+    def _install_task(self, task_config: TaskConfig) -> None:
+        """Create, register, and schedule a single task."""
         from rock.admin.scheduler.task_factory import TaskFactory
 
-        TaskFactory.register_all_tasks(self.scheduler_config)
+        try:
+            task = TaskFactory.create_task(task_config)
+        except Exception as e:
+            logger.error(f"Failed to create task '{task_config.task_class}': {e}")
+            return
+
+        self._tasks_by_class[task_config.task_class] = task
+        self._task_hashes[task_config.task_class] = self._compute_task_hash(task_config)
+        self._scheduler.add_job(
+            self._run_task,
+            trigger="interval",
+            seconds=task.interval_seconds,
+            args=[task],
+            id=task.type,
+            name=task.type,
+            replace_existing=True,
+            next_run_time=datetime.now(self.local_tz) + timedelta(seconds=2),
+        )
+        logger.info(f"Installed task '{task.type}' with interval {task.interval_seconds}s")
+
+    async def _uninstall_task(self, task_class: str) -> None:
+        """Remove a single task from the scheduler and clean up its worker-side processes."""
+        task = self._tasks_by_class.pop(task_class, None)
+        self._task_hashes.pop(task_class, None)
+        if task is None:
+            return
+        try:
+            self._scheduler.remove_job(task.type)
+        except Exception as e:
+            logger.warning(f"Failed to remove scheduler job '{task.type}': {e}")
+        if self._worker_cache is not None:
+            worker_ips = self._worker_cache.get_alive_workers()
+            if worker_ips:
+                await task.cleanup(worker_ips)
+        logger.info(f"Uninstalled task '{task.type}'")
+
+    async def _rebuild_tasks(self) -> None:
+        """Apply config changes by diffing old vs new tasks; only touch the ones that changed."""
+        if not self.scheduler_config.enabled:
+            for task_class in list(self._tasks_by_class):
+                await self._uninstall_task(task_class)
+            logger.info("Scheduler disabled, all tasks removed")
+            return
+
+        new_by_class: dict[str, TaskConfig] = {}
+        new_hashes: dict[str, str] = {}
+        for task_config in self.scheduler_config.tasks:
+            if not task_config.enabled or not task_config.task_class:
+                continue
+            new_by_class[task_config.task_class] = task_config
+            new_hashes[task_config.task_class] = self._compute_task_hash(task_config)
+
+        old_keys = set(self._task_hashes)
+        new_keys = set(new_hashes)
+        removed = old_keys - new_keys
+        added = new_keys - old_keys
+        changed = {k for k in (old_keys & new_keys) if self._task_hashes[k] != new_hashes[k]}
+
+        for task_class in removed | changed:
+            await self._uninstall_task(task_class)
+        for task_class in added | changed:
+            self._install_task(new_by_class[task_class])
+
+        if removed or added or changed:
+            logger.info(f"Scheduler tasks updated: removed={len(removed)}, added={len(added)}, changed={len(changed)}")
+        else:
+            logger.info("No task changes detected")
 
     async def _run_task(self, task: BaseTask) -> None:
         """Run a single task on alive workers."""
@@ -99,28 +216,18 @@ async def _run_task(self, task: BaseTask) -> None:
         except Exception as e:
             logger.error(f"Task '{task.type}' failed: {e}")
 
-    def _add_jobs(self) -> None:
-        """Add all registered tasks as scheduler jobs."""
-        for task in TaskRegistry.get_all_tasks().values():
-            self._scheduler.add_job(
-                self._run_task,
-                trigger="interval",
-                seconds=task.interval_seconds,
-                args=[task],
-                id=task.type,
-                name=task.type,
-                replace_existing=True,
-                next_run_time=datetime.now(self.local_tz) + timedelta(seconds=2),
-            )
-            logger.info(f"Added job '{task.type}' with interval {task.interval_seconds}s")
-
     async def run(self) -> None:
         """Run the scheduler until stopped."""
+        self._event_loop = asyncio.get_running_loop()
+
         self._init_worker_cache()
-        self._register_tasks()
+
+        if self._nacos_provider:
+            self._nacos_provider.add_listener(self._on_nacos_config_changed)
+            logger.info("Nacos dynamic config listener registered for scheduler")
 
         self._scheduler = AsyncIOScheduler(timezone=self.local_tz)
-        self._add_jobs()
+        await self._rebuild_tasks()
 
         # Pre-cache worker IPs before starting
         self._worker_cache.refresh()
@@ -129,7 +236,6 @@ async def run(self) -> None:
         logger.info("Scheduler started")
 
         self._stop_event = asyncio.Event()
-        self._loop = asyncio.get_event_loop()
 
         try:
             await self._stop_event.wait()
@@ -141,22 +247,27 @@ async def run(self) -> None:
 
     def stop(self) -> None:
         """Thread-safe stop: signal the scheduler to shut down."""
-        if self._stop_event and self._loop:
-            self._loop.call_soon_threadsafe(self._stop_event.set)
+        if self._stop_event and self._event_loop:
+            self._event_loop.call_soon_threadsafe(self._stop_event.set)
 
 
 class SchedulerThread:
     """Scheduler thread manager - runs APScheduler in a daemon thread with its own event loop."""
 
-    def __init__(self, scheduler_config: SchedulerConfig):
+    def __init__(
+        self,
+        scheduler_config: SchedulerConfig,
+        nacos_provider: NacosConfigProvider | None = None,
+    ):
         self.scheduler_config = scheduler_config
+        self.nacos_provider = nacos_provider
         self._thread: threading.Thread | None = None
         self._task_scheduler: TaskScheduler | None = None
 
     def _run_scheduler_in_thread(self) -> None:
         """Entry point for running scheduler in a thread with a dedicated event loop."""
         try:
-            self._task_scheduler = TaskScheduler(self.scheduler_config)
+            self._task_scheduler = TaskScheduler(self.scheduler_config, self.nacos_provider)
             asyncio.run(self._task_scheduler.run())
         except Exception:
             logger.exception("Scheduler thread encountered an error")
 
@@ -171,6 +171,47 @@ async def save_task_status(self, runtime: RemoteSandboxRuntime, status: TaskStat
         """Save task status to worker file."""
         await runtime.write_file(WriteFileRequest(path=self.status_file_path, content=status.to_json()))
 
+    async def _clear_task_status(self, runtime: RemoteSandboxRuntime) -> None:
+        """Remove the status file from worker."""
+        await runtime.execute(Command(command=f"rm -f {self.status_file_path}", shell=True))
+
+    async def cleanup_on_worker(self, ip: str) -> None:
+        """Stop any long-running process spawned by this task on a single worker.
+
+        For idempotent tasks this is a no-op (no daemon process to kill).
+        """
+        if self.idempotency == IdempotencyType.IDEMPOTENT:
+            return
+        runtime = self._get_runtime(ip)
+        status = await self.get_task_status(runtime)
+        if status is None or not status.pid:
+            return
+        if await runtime.check_pid_exists(status.pid):
+            kill_cmd = f"pkill -9 -P {status.pid}; kill -9 {status.pid}"
+            await runtime.execute(Command(command=kill_cmd, shell=True))
+            logger.info(f"[{self.type}] killed pid {status.pid} on worker[{ip}]")
+        await self._clear_task_status(runtime)
+
+    async def cleanup(self, worker_ips: list[str], max_concurrency: int = 50) -> None:
+        """Cleanup task across all workers, parallel and best-effort.
+
+        Idempotent tasks return immediately. For non-idempotent tasks, kills the
+        recorded daemon process and clears the status file on each worker. Failures
+        on individual workers are logged but do not propagate.
+        """
+        if self.idempotency == IdempotencyType.IDEMPOTENT:
+            return
+        semaphore = asyncio.Semaphore(max_concurrency)
+
+        async def cleanup_with_limit(ip: str) -> None:
+            async with semaphore:
+                try:
+                    await self.cleanup_on_worker(ip)
+                except Exception as e:
+                    logger.warning(f"[{self.type}] cleanup failed on worker[{ip}]: {e}")
+
+        await asyncio.gather(*[cleanup_with_limit(ip) for ip in worker_ips])
+
     async def should_run(self, runtime: RemoteSandboxRuntime) -> bool:
         """Determine if the task should be run."""
         if self.idempotency == IdempotencyType.IDEMPOTENT:
 
@@ -2,16 +2,11 @@
 import importlib
 
 from rock.admin.scheduler.task_base import BaseTask
-from rock.admin.scheduler.task_registry import TaskRegistry
-from rock.common.constants import SCHEDULER_LOG_NAME
-from rock.config import SchedulerConfig, TaskConfig
-from rock.logger import init_logger
-
-logger = init_logger("task_factory", file_name=SCHEDULER_LOG_NAME)
+from rock.config import TaskConfig
 
 
 class TaskFactory:
-    """Task factory - dynamically creates and registers tasks from config."""
+    """Task factory - dynamically creates tasks from config."""
 
     @staticmethod
     def _load_task_class(class_path: str) -> type[BaseTask]:
@@ -39,29 +34,5 @@ def create_task(cls, task_config: TaskConfig) -> BaseTask:
         Returns:
             Task instance
         """
-        # Dynamically load task class
         task_class = cls._load_task_class(task_config.task_class)
-
-        # Create task instance with config params
-        task = task_class.from_config(task_config)
-
-        return task
-
-    @classmethod
-    def register_all_tasks(cls, scheduler_config: SchedulerConfig):
-        """Register all enabled tasks from config."""
-        for task_config in scheduler_config.tasks:
-            if not task_config.enabled:
-                logger.info(f"Task '{task_config.task_class}' is disabled, skipping")
-                continue
-
-            if not task_config.task_class:
-                logger.warning(f"Task '{task_config.task_class}' has no task_class, skipping")
-                continue
-
-            try:
-                task = cls.create_task(task_config)
-                TaskRegistry.register(task)
-                logger.info(f"Registered task '{task.type}' with interval {task.interval_seconds}s")
-            except Exception as e:
-                logger.error(f"Failed to create task '{task_config.task_class}': {e}")
+        return task_class.from_config(task_config)