"""CrawlOperator""" import traceback import os import math from pprint import pprint from typing import Optional, Any, Sequence from datetime import datetime from uuid import UUID import json import humanize from kubernetes.utils import parse_quantity from redis import asyncio as exceptions from btrixcloud.models import ( TYPE_NON_RUNNING_STATES, TYPE_RUNNING_STATES, TYPE_ALL_CRAWL_STATES, NON_RUNNING_STATES, RUNNING_STATES, WAITING_STATES, RUNNING_AND_STARTING_ONLY, RUNNING_AND_WAITING_STATES, SUCCESSFUL_STATES, FAILED_STATES, CrawlStats, CrawlFile, CrawlCompleteIn, StorageRef, ) from btrixcloud.utils import str_to_date, date_to_str, dt_now from .baseoperator import BaseOperator, Redis from .models import ( CrawlSpec, CrawlStatus, StopReason, MCBaseRequest, MCSyncData, PodInfo, POD, CMAP, PVC, CJS, BTRIX_API, ) METRICS_API = "metrics.k8s.io/v1beta1" METRICS = f"PodMetrics.{METRICS_API}" DEFAULT_TTL = 30 REDIS_TTL = 60 # time in seconds before a crawl is deemed 'waiting' instead of 'starting' STARTING_TIME_SECS = 150 # how often to update execution time seconds EXEC_TIME_UPDATE_SECS = 60 # scale up if exceeded this threshold of mem usage (eg. 90%) MEM_SCALE_UP_THRESHOLD = 0.90 # scale up by this much MEM_SCALE_UP = 1.2 # soft OOM if exceeded this threshold of mem usage (eg. 100%) MEM_SOFT_OOM_THRESHOLD = 1.0 # set memory limit to this much of request for extra padding MEM_LIMIT_PADDING = 1.2 # pylint: disable=too-many-public-methods, too-many-locals, too-many-branches, too-many-statements # pylint: disable=invalid-name, too-many-lines, too-many-return-statements # ============================================================================ class CrawlOperator(BaseOperator): """CrawlOperator Handler""" done_key: str pages_key: str errors_key: str fast_retry_secs: int log_failed_crawl_lines: int min_avail_storage_ratio: float def __init__(self, *args): super().__init__(*args) self.done_key = "crawls-done" self.pages_key = "pages" self.errors_key = "e" self.fast_retry_secs = int(os.environ.get("FAST_RETRY_SECS") or 0) self.log_failed_crawl_lines = int(os.environ.get("LOG_FAILED_CRAWL_LINES") or 0) # ensure available storage is at least this much times used storage self.min_avail_storage_ratio = float( os.environ.get("CRAWLER_MIN_AVAIL_STORAGE_RATIO") or 0 ) def init_routes(self, app): """init routes for this operator""" @app.post("/op/crawls/sync") async def mc_sync_crawls(data: MCSyncData): return await self.sync_crawls(data) # reuse sync path, but distinct endpoint for better logging @app.post("/op/crawls/finalize") async def mc_sync_finalize(data: MCSyncData): return await self.sync_crawls(data) @app.post("/op/crawls/customize") async def mc_related(data: MCBaseRequest): return self.get_related(data) async def sync_crawls(self, data: MCSyncData): """sync crawls""" status = CrawlStatus(**data.parent.get("status", {})) status.last_state = status.state spec = data.parent.get("spec", {}) # spec is the data from crawl_job.yaml crawl_id = spec["id"] cid = spec["cid"] oid = spec["oid"] redis_url = self.k8s.get_redis_url(crawl_id) params = {} params.update(self.k8s.shared_params) params["id"] = crawl_id params["cid"] = cid params["oid"] = oid params["userid"] = spec.get("userid", "") pods = data.children[POD] org = await self.org_ops.get_org_by_id(UUID(oid)) crawl = CrawlSpec( id=crawl_id, cid=cid, oid=oid, org=org, storage=StorageRef(spec["storageName"]), crawler_channel=spec.get("crawlerChannel", "default"), proxy_id=spec.get("proxyId"), scale=spec.get("scale", 1), started=data.parent["metadata"]["creationTimestamp"], stopping=spec.get("stopping", False), timeout=spec.get("timeout") or 0, max_crawl_size=int(spec.get("maxCrawlSize") or 0), scheduled=spec.get("manual") != "1", qa_source_crawl_id=spec.get("qaSourceCrawlId"), ) if crawl.qa_source_crawl_id: crawl.scale = int(params.get("qa_scale", 1)) # if finalizing, crawl is being deleted if data.finalizing: if not status.finished: # if can't cancel, already finished await self.cancel_crawl(crawl, status, data.children[POD]) # instead of fetching the state (that was already set) # return exception to ignore this request, keep previous # finished state # raise HTTPException(status_code=400, detail="out_of_sync_status") return await self.finalize_response( crawl, status, spec, data.children, params, ) # just in case, finished but not deleted, can only get here if # do_crawl_finished_tasks() doesn't reach the end or taking too long if status.finished: print( f"warn crawl {crawl_id} finished but not deleted, post-finish taking too long?" ) self.run_task(self.k8s.delete_crawl_job(crawl.id)) return await self.finalize_response( crawl, status, spec, data.children, params, ) # shouldn't get here, crawl should already be finalizing when canceled # just in case, handle canceled-but-not-finalizing here if status.state == "canceled": await self.k8s.delete_crawl_job(crawl.id) return {"status": status.dict(exclude_none=True), "children": []} # first, check storage quota, and fail immediately if quota reached if status.state in ( "starting", "skipped_storage_quota_reached", "skipped_time_quota_reached", ): # only check on very first run, before any pods/pvcs created # for now, allow if crawl has already started (pods/pvcs created) if not pods and not data.children[PVC]: if self.org_ops.storage_quota_reached(org): await self.mark_finished( crawl, status, "skipped_storage_quota_reached" ) return self._empty_response(status) if self.org_ops.exec_mins_quota_reached(org): await self.mark_finished( crawl, status, "skipped_time_quota_reached" ) return self._empty_response(status) if status.state in ("starting", "waiting_org_limit"): if not await self.can_start_new(crawl, data, status): return self._empty_response(status) await self.set_state( "starting", status, crawl, allowed_from=["waiting_org_limit"] ) if len(pods): for pod_name, pod in pods.items(): self.sync_resources(status, pod_name, pod, data.children) status = await self.sync_crawl_state(redis_url, crawl, status, pods, data) if self.k8s.enable_auto_resize: # auto sizing handled here await self.handle_auto_size(status.podStatus) if status.finished: return await self.finalize_response( crawl, status, spec, data.children, params, ) await self.increment_pod_exec_time( pods, crawl, status, EXEC_TIME_UPDATE_SECS ) else: status.scale = 1 children = self._load_redis(params, status, data.children) storage_path = crawl.storage.get_storage_extra_path(oid) storage_secret = crawl.storage.get_storage_secret_name(oid) if not crawl.is_qa: params["profile_filename"] = spec.get("profile_filename", "") else: storage_path += "qa/" params["storage_path"] = storage_path params["storage_secret"] = storage_secret status.crawlerImage = self.crawl_config_ops.get_channel_crawler_image( crawl.crawler_channel ) params["crawler_image"] = status.crawlerImage if crawl.proxy_id and not crawl.is_qa: proxy = self.crawl_config_ops.get_crawler_proxy(crawl.proxy_id) if proxy: params["proxy_id"] = crawl.proxy_id params["proxy_url"] = proxy.url params["proxy_ssh_private_key"] = proxy.has_private_key params["proxy_ssh_host_public_key"] = proxy.has_host_public_key params["storage_filename"] = spec["storage_filename"] params["restart_time"] = spec.get("restartTime") params["warc_prefix"] = spec.get("warcPrefix") params["redis_url"] = redis_url if spec.get("restartTime") != status.restartTime: # pylint: disable=invalid-name status.restartTime = spec.get("restartTime") status.resync_after = self.fast_retry_secs params["force_restart"] = True else: params["force_restart"] = False config_update_needed = ( spec.get("lastConfigUpdate", "") != status.lastConfigUpdate ) status.lastConfigUpdate = spec.get("lastConfigUpdate", "") children.extend( await self._load_crawl_configmap( crawl, data.children, params, config_update_needed ) ) if crawl.qa_source_crawl_id: params["qa_source_crawl_id"] = crawl.qa_source_crawl_id children.extend(await self._load_qa_configmap(params, data.children)) for i in range(0, status.scale): children.extend(self._load_crawler(params, i, status, data.children)) return { "status": status.dict(exclude_none=True), "children": children, "resyncAfterSeconds": status.resync_after, } def _load_redis(self, params, status: CrawlStatus, children): name = f"redis-{params['id']}" has_pod = name in children[POD] pod_info = status.podStatus[name] params["name"] = name params["cpu"] = pod_info.newCpu or params.get("redis_cpu") params["memory"] = pod_info.newMemory or params.get("redis_memory") restart_reason = None if has_pod: restart_reason = pod_info.should_restart_pod() if restart_reason: print(f"Restarting {name}, reason: {restart_reason}") params["init_redis"] = status.initRedis and not restart_reason return self.load_from_yaml("redis.yaml", params) def _filter_autoclick_behavior( self, behaviors: Optional[str], crawler_image: str ) -> Optional[str]: """Remove autoclick behavior if crawler version doesn't support it""" min_autoclick_crawler_image = os.environ.get("MIN_AUTOCLICK_CRAWLER_IMAGE") if ( min_autoclick_crawler_image and behaviors and "autoclick" in behaviors and crawler_image and crawler_image < min_autoclick_crawler_image ): print( "Crawler version < min_autoclick_crawler_image, removing autoclick behavior", flush=True, ) behaviors_list = behaviors.split(",") filtered_behaviors = [ behavior for behavior in behaviors_list if behavior != "autoclick" ] return ",".join(filtered_behaviors) return behaviors async def _load_crawl_configmap( self, crawl: CrawlSpec, children, params, config_update_needed: bool ): name = f"crawl-config-{crawl.id}" configmap = children[CMAP].get(name) if configmap and not config_update_needed: metadata = configmap["metadata"] configmap["metadata"] = { "name": metadata["name"], "namespace": metadata["namespace"], "labels": metadata["labels"], } return [configmap] params["name"] = name crawlconfig = await self.crawl_config_ops.get_crawl_config(crawl.cid, crawl.oid) self.crawl_config_ops.ensure_quota_page_limit(crawlconfig, crawl.org) raw_config = crawlconfig.get_raw_config() raw_config["behaviors"] = self._filter_autoclick_behavior( raw_config["behaviors"], params["crawler_image"] ) params["config"] = json.dumps(raw_config) if config_update_needed: print(f"Updating config for {crawl.id}") return self.load_from_yaml("crawl_configmap.yaml", params) async def _load_qa_configmap(self, params, children): qa_source_crawl_id = params["qa_source_crawl_id"] name = f"qa-replay-{qa_source_crawl_id}" configmap = children[CMAP].get(name) if configmap and not self._qa_configmap_update_needed(name, configmap): metadata = configmap["metadata"] configmap["metadata"] = { "name": metadata["name"], "namespace": metadata["namespace"], "labels": metadata["labels"], } return [configmap] crawl_replay = await self.crawl_ops.get_internal_crawl_out(qa_source_crawl_id) params["name"] = name params["qa_source_replay_json"] = crawl_replay.json(include={"resources"}) return self.load_from_yaml("qa_configmap.yaml", params) def _load_crawler(self, params, i, status: CrawlStatus, children): name = f"crawl-{params['id']}-{i}" has_pod = name in children[POD] if params.get("qa_source_crawl_id"): cpu_field = "qa_cpu" mem_field = "qa_memory" worker_field = "qa_workers" pri_class = f"qa-crawl-pri-{i}" else: cpu_field = "crawler_cpu" mem_field = "crawler_memory" worker_field = "crawler_workers" pri_class = f"crawl-pri-{i}" pod_info = status.podStatus[name] params["name"] = name params["priorityClassName"] = pri_class params["cpu"] = pod_info.newCpu or params.get(cpu_field) params["memory"] = pod_info.newMemory or params.get(mem_field) if self.k8s.enable_auto_resize: params["memory_limit"] = float(params["memory"]) * MEM_LIMIT_PADDING else: params["memory_limit"] = self.k8s.max_crawler_memory_size params["storage"] = pod_info.newStorage or params.get("crawler_storage") params["workers"] = params.get(worker_field) or 1 params["do_restart"] = False if has_pod: restart_reason = pod_info.should_restart_pod(params.get("force_restart")) if restart_reason: print(f"Restarting {name}, reason: {restart_reason}") params["do_restart"] = True return self.load_from_yaml("crawler.yaml", params) def _qa_configmap_update_needed(self, name, configmap): try: now = dt_now() resources = json.loads(configmap["data"]["qa-config.json"])["resources"] for resource in resources: expire_at = str_to_date(resource["expireAt"]) if expire_at and expire_at <= now: print(f"Refreshing QA configmap for QA run: {name}") return True # pylint: disable=broad-exception-caught except Exception as e: print(e) return False # pylint: disable=too-many-arguments async def _resolve_scale( self, crawl_id: str, desired_scale: int, redis: Redis, status: CrawlStatus, pods: dict[str, dict], ): """Resolve scale If desired_scale >= actual scale, just set (also limit by number of pages found). If desired scale < actual scale, attempt to shut down each crawl instance via redis setting. If contiguous instances shutdown (successful exit), lower scale and clean up previous scale state. """ # actual scale (minus redis pod) actual_scale = len(pods) if pods.get(f"redis-{crawl_id}"): actual_scale -= 1 # ensure at least enough pages for the scale if status.pagesFound < desired_scale: desired_scale = max(1, status.pagesFound) # if desired_scale same or scaled up, return desired_scale if desired_scale >= actual_scale: return desired_scale new_scale = actual_scale for i in range(actual_scale - 1, desired_scale - 1, -1): name = f"crawl-{crawl_id}-{i}" pod = pods.get(name) if pod: print(f"Attempting scaling down of pod {i}") await redis.hset(f"{crawl_id}:stopone", name, "1") # check if this pod can be scaled down if new_scale == i + 1: # if status key doesn't exist, this pod never actually ran, so just scale down if not await redis.hexists(f"{crawl_id}:status", name): new_scale = i print(f"Scaled down pod index {i + 1} -> {i}, no previous pod") elif pod and pod["status"].get("phase") == "Succeeded": new_scale = i print(f"Scaled down pod index {i + 1} -> {i}, pod completed") if new_scale < actual_scale: for i in range(new_scale, actual_scale): name = f"crawl-{crawl_id}-{i}" await redis.hdel(f"{crawl_id}:stopone", name) await redis.hdel(f"{crawl_id}:status", name) return new_scale def sync_resources(self, status, name, pod, children): """set crawljob status from current resources""" resources = status.podStatus[name].allocated src = pod["spec"]["containers"][0]["resources"]["requests"] resources.memory = int(parse_quantity(src.get("memory"))) resources.cpu = float(parse_quantity(src.get("cpu"))) pvc = children[PVC].get(name) if pvc: try: src = pvc["status"]["capacity"] resources.storage = int(parse_quantity(src.get("storage"))) # pylint: disable=bare-except except: pass async def set_state( self, state: TYPE_ALL_CRAWL_STATES, status: CrawlStatus, crawl: CrawlSpec, allowed_from: Sequence[TYPE_ALL_CRAWL_STATES], finished: Optional[datetime] = None, stats: Optional[CrawlStats] = None, ): """set status state and update db, if changed if allowed_from passed in, can only transition from allowed_from state, otherwise get current state from db and return the following state transitions are supported: from starting to org concurrent crawl limit and back: - starting -> waiting_org_capacity -> starting from starting to running: - starting -> running from running to complete or complete[:stopReason]: - running -> complete[:stopReason] - running -> complete from starting or running to waiting for capacity (pods pending) and back: - starting -> waiting_capacity - running -> waiting_capacity - waiting_capacity -> running from any state to canceled or failed: - not complete[:stopReason] -> canceled - not complete[:stopReason] -> failed """ if not allowed_from or status.state in allowed_from: res = await self.crawl_ops.update_crawl_state_if_allowed( crawl.db_crawl_id, crawl.is_qa, state=state, allowed_from=allowed_from, finished=finished, stats=stats, ) if res and status.state != state: print(f"Setting state: {status.state} -> {state}, {crawl.id}") status.state = state return True # get actual crawl state actual_state, finished = await self.crawl_ops.get_crawl_state( crawl.db_crawl_id, crawl.is_qa ) if actual_state: status.state = actual_state if finished: status.finished = date_to_str(finished) if actual_state != state: print( f"State mismatch, actual state {actual_state}, requested {state}, {crawl.id}" ) if not actual_state and state == "canceled": return True if status.state != state: print( f"Not setting state: {status.state} -> {state}, not allowed, {crawl.id}" ) return False def get_related(self, data: MCBaseRequest): """return objects related to crawl pods""" spec = data.parent.get("spec", {}) crawl_id = spec["id"] oid = spec.get("oid") # filter by role as well (job vs qa-job) role = data.parent.get("metadata", {}).get("labels", {}).get("role") related_resources = [ { "apiVersion": BTRIX_API, "resource": "crawljobs", "labelSelector": {"matchLabels": {"btrix.org": oid, "role": role}}, }, ] if self.k8s.enable_auto_resize: related_resources.append( { "apiVersion": METRICS_API, "resource": "pods", "labelSelector": {"matchLabels": {"crawl": crawl_id}}, } ) return {"relatedResources": related_resources} async def can_start_new( self, crawl: CrawlSpec, data: MCSyncData, status: CrawlStatus, ): """return true if crawl can start, otherwise set crawl to 'queued' state until more crawls for org finish""" max_crawls = crawl.org.quotas.maxConcurrentCrawls or 0 if not max_crawls: return True if len(data.related[CJS]) <= max_crawls: return True name = data.parent.get("metadata", {}).get("name") i = 0 for crawl_sorted in data.related[CJS].values(): if crawl_sorted.get("status", {}).get("state") in NON_RUNNING_STATES: continue if crawl_sorted.get("metadata").get("name") == name: if i < max_crawls: return True break i += 1 await self.set_state( "waiting_org_limit", status, crawl, allowed_from=["starting"] ) return False async def cancel_crawl( self, crawl: CrawlSpec, status: CrawlStatus, pods: dict, ) -> bool: """Mark crawl as canceled""" if not await self.mark_finished(crawl, status, "canceled"): return False await self.mark_for_cancelation(crawl.id) if not status.canceled: for name, pod in pods.items(): pstatus = pod["status"] role = pod["metadata"]["labels"]["role"] if role != "crawler": continue if "containerStatuses" not in pstatus: continue cstatus = pstatus["containerStatuses"][0] self.handle_terminated_pod( name, role, status, cstatus["state"].get("terminated") ) status.canceled = True return status.canceled async def fail_crawl( self, crawl: CrawlSpec, status: CrawlStatus, pods: dict, stats: Optional[CrawlStats] = None, ) -> bool: """Mark crawl as failed, log crawl state and print crawl logs, if possible""" prev_state = status.state if not await self.mark_finished(crawl, status, "failed", stats=stats): return False if not self.log_failed_crawl_lines or prev_state == "failed": return True pod_names = list(pods.keys()) for name in pod_names: print(f"============== POD STATUS: {name} ==============") pprint(pods[name]["status"]) self.run_task(self.k8s.print_pod_logs(pod_names, self.log_failed_crawl_lines)) return True def _empty_response(self, status): """done response for removing crawl""" return { "status": status.dict(exclude_none=True), "children": [], } async def finalize_response( self, crawl: CrawlSpec, status: CrawlStatus, spec: dict, children: dict, params: dict, ): """ensure crawl id ready for deletion""" redis_pod = f"redis-{crawl.id}" new_children = [] finalized = False pods = children[POD] if redis_pod in pods: # if has other pods, keep redis pod until they are removed if len(pods) > 1: new_children = self._load_redis(params, status, children) await self.increment_pod_exec_time(pods, crawl, status) # keep pvs until pods are removed if new_children: new_children.extend(list(children[PVC].values())) if not children[POD] and not children[PVC]: # keep parent until ttl expired, if any if status.finished: ttl = spec.get("ttlSecondsAfterFinished", DEFAULT_TTL) finished = str_to_date(status.finished) if finished and (dt_now() - finished).total_seconds() > ttl >= 0: print("CrawlJob expired, deleting: " + crawl.id) finalized = True else: finalized = True if finalized and crawl.is_qa: self.run_task(self.crawl_ops.qa_run_finished(crawl.db_crawl_id)) return { "status": status.dict(exclude_none=True), "children": new_children, "finalized": finalized, } async def _get_redis(self, redis_url: str) -> Optional[Redis]: """init redis, ensure connectivity""" redis = None try: redis = await self.k8s.get_redis_client(redis_url) # test connection await redis.ping() return redis # pylint: disable=bare-except except: if redis: await redis.close() return None async def sync_crawl_state( self, redis_url: str, crawl: CrawlSpec, status: CrawlStatus, pods: dict[str, dict], data: MCSyncData, ): """sync crawl state for running crawl""" # check if at least one crawler pod started running crawler_running, redis_running, pod_done_count = self.sync_pod_status( pods, status ) redis = None metrics = data.related.get(METRICS, {}) try: if redis_running: redis = await self._get_redis(redis_url) await self.add_used_stats(crawl.id, status.podStatus, redis, metrics) # skip if no newly exited pods if status.anyCrawlPodNewExit: await self.log_crashes(crawl.id, status.podStatus, redis) if not crawler_running or not redis: # if either crawler is not running or redis is inaccessible if not pod_done_count and self.should_mark_waiting( status.state, crawl.started ): # mark as waiting (if already running) await self.set_state( "waiting_capacity", status, crawl, allowed_from=RUNNING_AND_STARTING_ONLY, ) if not crawler_running and redis: # if crawler is not running for REDIS_TTL seconds, also stop redis # but not right away in case crawler pod is just restarting. # avoids keeping redis pods around while no crawler pods are up # (eg. due to resource constraints) last_active_time = str_to_date(status.lastActiveTime) if last_active_time and ( (dt_now() - last_active_time).total_seconds() > REDIS_TTL ): print( f"Pausing redis, no running crawler pods for >{REDIS_TTL} secs" ) status.initRedis = False elif crawler_running and not redis: # if crawler is running, but no redis, init redis status.initRedis = True # if no crawler / no redis, resync after N seconds status.resync_after = self.fast_retry_secs return status # update lastActiveTime if crawler is running if crawler_running: status.lastActiveTime = date_to_str(dt_now()) file_done = await redis.lpop(self.done_key) while file_done: msg = json.loads(file_done) # add completed file if msg.get("filename"): await self.add_file_to_crawl(msg, crawl, redis) await redis.incr("filesAdded") # get next file done file_done = await redis.lpop(self.done_key) page_crawled = await redis.lpop(f"{crawl.id}:{self.pages_key}") qa_run_id = crawl.id if crawl.is_qa else None while page_crawled: page_dict = json.loads(page_crawled) await self.page_ops.add_page_to_db( page_dict, crawl.db_crawl_id, qa_run_id, crawl.oid ) page_crawled = await redis.lpop(f"{crawl.id}:{self.pages_key}") crawl_error = await redis.lpop(f"{crawl.id}:{self.errors_key}") while crawl_error: await self.crawl_ops.add_crawl_error( crawl.db_crawl_id, crawl.is_qa, crawl_error ) crawl_error = await redis.lpop(f"{crawl.id}:{self.errors_key}") # ensure filesAdded and filesAddedSize always set status.filesAdded = int(await redis.get("filesAdded") or 0) status.filesAddedSize = int(await redis.get("filesAddedSize") or 0) # update stats and get status return await self.update_crawl_state( redis, crawl, status, pods, pod_done_count, data ) # pylint: disable=broad-except except Exception as exc: traceback.print_exc() print(f"Crawl get failed: {exc}, will try again") return status finally: if redis: await redis.close() def sync_pod_status( self, pods: dict[str, dict], status: CrawlStatus ) -> tuple[bool, bool, int]: """check status of pods""" crawler_running = False redis_running = False pod_done_count = 0 try: for name, pod in pods.items(): running = False evicted = False pstatus = pod["status"] phase = pstatus["phase"] role = pod["metadata"]["labels"]["role"] if phase in ("Running", "Succeeded"): running = True elif phase == "Failed" and pstatus.get("reason") == "Evicted": evicted = True status.podStatus[name].evicted = evicted if "containerStatuses" in pstatus: cstatus = pstatus["containerStatuses"][0] # don't consider 'ContainerCreating' as running for now # may be stuck in this state for other reasons # # waiting = cstatus["state"].get("waiting") # if ( # phase == "Pending" # and waiting # and waiting.get("reason") == "ContainerCreating" # ): # running = True self.handle_terminated_pod( name, role, status, cstatus["state"].get("terminated") ) if role == "crawler": crawler_running = crawler_running or running if phase == "Succeeded": pod_done_count += 1 elif role == "redis": redis_running = redis_running or running # pylint: disable=broad-except except Exception as exc: print(exc) return crawler_running, redis_running, pod_done_count def handle_terminated_pod( self, name, role, status: CrawlStatus, terminated: Optional[dict[str, Any]] ) -> None: """handle terminated pod state""" if not terminated: return exit_time = terminated.get("finishedAt") if not exit_time: print("warn: terminated pod missing finishedAt", flush=True) return pod_status = status.podStatus[name] pod_status.isNewExit = pod_status.exitTime != exit_time if pod_status.isNewExit and role == "crawler": pod_status.exitTime = exit_time status.anyCrawlPodNewExit = True # detect reason exit_code = terminated.get("exitCode") if exit_code == 0: pod_status.reason = "done" elif terminated.get("reason") == "OOMKilled" or exit_code == 137: pod_status.reason = "oom" else: pod_status.reason = "interrupt: " + str(exit_code) pod_status.exitCode = exit_code async def increment_pod_exec_time( self, pods: dict[str, dict], crawl: CrawlSpec, status: CrawlStatus, min_duration=0, ) -> None: """inc exec time tracking""" now = dt_now() # don't count time crawl is not running if status.state in WAITING_STATES: # reset lastUpdatedTime if at least 2 consecutive updates of non-running state if status.last_state in WAITING_STATES: status.lastUpdatedTime = date_to_str(now) return update_start_time = await self.crawl_ops.get_crawl_exec_last_update_time( crawl.db_crawl_id, crawl.is_qa ) if not update_start_time: print("Crawl first started, webhooks called", now, crawl.id) # call initial running webhook if not crawl.qa_source_crawl_id: self.run_task( self.event_webhook_ops.create_crawl_started_notification( crawl.id, crawl.oid, scheduled=crawl.scheduled ) ) else: self.run_task( self.event_webhook_ops.create_qa_analysis_started_notification( crawl.id, crawl.oid, crawl.qa_source_crawl_id ) ) await self.crawl_ops.inc_crawl_exec_time( crawl.db_crawl_id, crawl.is_qa, 0, now ) status.lastUpdatedTime = date_to_str(now) return reason = None update_duration = (now - update_start_time).total_seconds() if status.anyCrawlPodNewExit: reason = "new pod exit" elif status.canceled: reason = "crawl canceled" elif now.month != update_start_time.month: reason = "month change" elif update_duration >= min_duration: reason = "duration reached" if min_duration else "finalizing" if not reason: return exec_time = 0 max_duration = 0 print( f"Exec Time Update: {reason}: {now} - {update_start_time} = {update_duration}" ) for name, pod in pods.items(): pstatus = pod["status"] role = pod["metadata"]["labels"]["role"] if role != "crawler": continue if "containerStatuses" not in pstatus: continue cstate = pstatus["containerStatuses"][0]["state"] end_time = None start_time = None pod_state = "" if "running" in cstate: pod_state = "running" state = cstate["running"] start_time = str_to_date(state.get("startedAt")) if update_start_time and start_time and update_start_time > start_time: start_time = update_start_time end_time = now elif "terminated" in cstate: pod_state = "terminated" state = cstate["terminated"] start_time = str_to_date(state.get("startedAt")) end_time = str_to_date(state.get("finishedAt")) if update_start_time and start_time and update_start_time > start_time: start_time = update_start_time # already counted if update_start_time and end_time and end_time < update_start_time: print( f" - {name}: {pod_state}: skipping already counted, " + f"{end_time} < {start_time}" ) continue if end_time and start_time: duration = int((end_time - start_time).total_seconds()) print( f" - {name}: {pod_state}: {end_time} - {start_time} = {duration}" ) exec_time += duration max_duration = max(duration, max_duration) if exec_time: await self.org_ops.inc_org_time_stats( crawl.oid, exec_time, True, crawl.is_qa ) status.crawlExecTime += exec_time status.elapsedCrawlTime += max_duration print( f" Exec Time Total: {status.crawlExecTime}, Incremented By: {exec_time}", flush=True, ) await self.crawl_ops.inc_crawl_exec_time( crawl.db_crawl_id, crawl.is_qa, exec_time, now ) status.lastUpdatedTime = date_to_str(now) def should_mark_waiting(self, state: TYPE_ALL_CRAWL_STATES, started: str) -> bool: """Should the crawl be marked as waiting for capacity?""" if state in RUNNING_STATES: return True if state == "starting": started_dt = str_to_date(started) if started_dt: return (dt_now() - started_dt).total_seconds() > STARTING_TIME_SECS return False async def add_used_stats( self, crawl_id, pod_status: dict[str, PodInfo], redis, metrics ): """load current usage stats""" if redis: stats = await redis.info("persistence") storage = int(stats.get("aof_current_size", 0)) + int( stats.get("current_cow_size", 0) ) pod_info = pod_status[f"redis-{crawl_id}"] pod_info.used.storage = storage # if no pod metrics, get memory estimate from redis itself if not self.k8s.enable_auto_resize: stats = await redis.info("memory") pod_info.used.memory = int(stats.get("used_memory_rss", 0)) # stats = await redis.info("cpu") # pod_info.used.cpu = float(stats.get("used_cpu_sys", 0)) for name, metric in metrics.items(): usage = metric["containers"][0]["usage"] pod_info = pod_status[name] pod_info.used.memory = int(parse_quantity(usage["memory"])) pod_info.used.cpu = float(parse_quantity(usage["cpu"])) async def handle_auto_size(self, pod_status: dict[str, PodInfo]) -> None: """auto scale pods here, experimental""" for name, pod in pod_status.items(): mem_usage = pod.get_percent_memory() new_memory = int(float(pod.allocated.memory) * MEM_SCALE_UP) send_sig = False # if pod is using >MEM_SCALE_UP_THRESHOLD of its memory, increase mem if mem_usage > MEM_SCALE_UP_THRESHOLD: if new_memory > self.k8s.max_crawler_memory_size: print( f"Mem {mem_usage}: Not resizing pod {name}: " + f"mem {new_memory} > max allowed {self.k8s.max_crawler_memory_size}" ) return pod.newMemory = new_memory print( f"Mem {mem_usage}: Resizing pod {name} -> mem {pod.newMemory} - Scale Up" ) # if crawler pod is using its OOM threshold, attempt a soft OOM # via a second SIGTERM if ( mem_usage >= MEM_SOFT_OOM_THRESHOLD and name.startswith("crawl") and pod.signalAtMem != pod.newMemory ): send_sig = True # if any pod crashed due to OOM, increase mem elif pod.isNewExit and pod.reason == "oom": pod.newMemory = new_memory print( f"Mem {mem_usage}: Resizing pod {name} -> mem {pod.newMemory} - OOM Detected" ) send_sig = True # avoid resending SIGTERM multiple times after it already succeeded if send_sig and await self.k8s.send_signal_to_pod(name, "SIGTERM"): pod.signalAtMem = pod.newMemory async def log_crashes(self, crawl_id, pod_status: dict[str, PodInfo], redis): """report/log any pod crashes here""" for name, pod in pod_status.items(): # log only unexpected exits as crashes # - 0 is success / intended shutdown # - 11 is default interrupt / intended restart # - 13 is force interrupt / intended restart if not pod.isNewExit or pod.exitCode in (0, 11, 13): continue log = self.get_log_line( "Crawler Instance Crashed", {"reason": pod.reason, "pod": name} ) if not redis: print(log) else: await redis.lpush(f"{crawl_id}:e", log) def get_log_line(self, message, details): """get crawler error line for logging""" err = { "timestamp": date_to_str(dt_now()), "logLevel": "error", "context": "k8s", "message": message, "details": details, } return json.dumps(err) async def add_file_to_crawl(self, cc_data, crawl: CrawlSpec, redis): """Handle finished CrawlFile to db""" filecomplete = CrawlCompleteIn(**cc_data) filename = self.storage_ops.get_org_relative_path( crawl.org, crawl.storage, filecomplete.filename ) crawl_file = CrawlFile( filename=filename, size=filecomplete.size, hash=filecomplete.hash, storage=crawl.storage, ) await redis.incr("filesAddedSize", filecomplete.size) await self.crawl_ops.add_crawl_file( crawl.db_crawl_id, crawl.is_qa, crawl_file, filecomplete.size ) # no replicas for QA for now if crawl.is_qa: return True try: await self.background_job_ops.create_replica_jobs( crawl.oid, crawl_file, crawl.id, "crawl" ) # pylint: disable=broad-except except Exception as exc: print("Replicate Exception", exc, flush=True) return True async def is_crawl_stopping( self, crawl: CrawlSpec, status: CrawlStatus, data: MCSyncData ) -> Optional[StopReason]: """check if crawl is stopping and set reason""" # if user requested stop, then enter stopping phase if crawl.stopping: return "stopped_by_user" # check timeout if timeout time exceeds elapsed time if crawl.timeout: elapsed = status.elapsedCrawlTime last_updated_time = str_to_date(status.lastUpdatedTime) if last_updated_time: elapsed += int((dt_now() - last_updated_time).total_seconds()) if elapsed > crawl.timeout: return "time-limit" # crawl size limit if crawl.max_crawl_size and status.size > crawl.max_crawl_size: return "size-limit" # gracefully stop crawl if current running crawl sizes reach storage quota org = crawl.org if org.readOnly: return "stopped_org_readonly" if org.quotas.storageQuota: running_crawls_total_size = status.size for crawl_job in data.related[CJS].values(): # if the job id matches current crawl job, then skip # this job to avoid double-counting # using the more up-to-date 'status.size' for this job if crawl_job.get("spec", {}).get("id") == crawl.id: continue crawl_status = crawl_job.get("status", {}) if crawl_status: running_crawls_total_size += crawl_status.get("size", 0) if self.org_ops.storage_quota_reached(org, running_crawls_total_size): return "stopped_storage_quota_reached" # gracefully stop crawl is execution time quota is reached if self.org_ops.exec_mins_quota_reached(org): return "stopped_time_quota_reached" return None async def get_redis_crawl_stats( self, redis: Redis, crawl_id: str ) -> tuple[CrawlStats, dict[str, Any]]: """get page stats""" try: # crawler >0.9.0, done key is a value pages_done = int(await redis.get(f"{crawl_id}:d") or 0) except exceptions.ResponseError: # crawler <=0.9.0, done key is a list pages_done = await redis.llen(f"{crawl_id}:d") pages_found = await redis.scard(f"{crawl_id}:s") # account for extra seeds and subtract from seen list extra_seeds = await redis.llen(f"{crawl_id}:extraSeeds") if extra_seeds: pages_found -= extra_seeds sizes = await redis.hgetall(f"{crawl_id}:size") archive_size = sum(int(x) for x in sizes.values()) stats = CrawlStats(found=pages_found, done=pages_done, size=archive_size) return stats, sizes async def update_crawl_state( self, redis: Redis, crawl: CrawlSpec, status: CrawlStatus, pods: dict[str, dict], pod_done_count: int, data: MCSyncData, ) -> CrawlStatus: """update crawl state and check if crawl is now done""" results = await redis.hgetall(f"{crawl.id}:status") stats, sizes = await self.get_redis_crawl_stats(redis, crawl.id) # need to add size of previously completed WACZ files as well! stats.size += status.filesAddedSize # update status status.pagesDone = stats.done status.pagesFound = stats.found status.size = stats.size status.sizeHuman = humanize.naturalsize(status.size) await self.crawl_ops.update_running_crawl_stats( crawl.db_crawl_id, crawl.is_qa, stats ) for key, value in sizes.items(): increase_storage = False value = int(value) if value > 0 and status.podStatus: pod_info = status.podStatus[key] pod_info.used.storage = value if ( status.state == "running" and self.min_avail_storage_ratio and pod_info.allocated.storage and pod_info.used.storage * self.min_avail_storage_ratio > pod_info.allocated.storage ): increase_storage = True # out of storage if pod_info.isNewExit and pod_info.exitCode == 3: pod_info.used.storage = pod_info.allocated.storage increase_storage = True if increase_storage: new_storage = math.ceil( pod_info.used.storage * self.min_avail_storage_ratio / 1_000_000_000 ) pod_info.newStorage = f"{new_storage}Gi" print( f"Attempting to adjust storage to {pod_info.newStorage} for {key}" ) if not status.stopReason: status.stopReason = await self.is_crawl_stopping(crawl, status, data) status.stopping = status.stopReason is not None if status.stopping: print("Crawl gracefully stopping: {status.stopReason}, id: {crawl.id}") # mark crawl as stopping if status.stopping: await redis.set(f"{crawl.id}:stopping", "1") # backwards compatibility with older crawler await redis.set("crawl-stop", "1") # resolve scale if crawl.scale != status.scale: status.scale = await self._resolve_scale( crawl.id, crawl.scale, redis, status, pods ) # check if done / failed status_count: dict[str, int] = {} for i in range(status.scale): res = results.get(f"crawl-{crawl.id}-{i}") if res: status_count[res] = status_count.get(res, 0) + 1 num_done = status_count.get("done", 0) num_failed = status_count.get("failed", 0) # all expected pods are either done or failed all_completed = (num_done + num_failed) >= status.scale # if at least one is done according to redis, consider crawl successful # ensure pod successfully exited as well # pylint: disable=chained-comparison if all_completed and num_done >= 1 and pod_done_count >= num_done: # check if one-page crawls actually succeeded # if only one page found, and no files, assume failed if status.pagesFound == 1 and not status.filesAdded: await self.fail_crawl(crawl, status, pods, stats) return status state: TYPE_NON_RUNNING_STATES if status.stopReason == "stopped_by_user": state = "stopped_by_user" elif status.stopReason == "stopped_storage_quota_reached": state = "stopped_storage_quota_reached" elif status.stopReason == "stopped_time_quota_reached": state = "stopped_time_quota_reached" elif status.stopReason == "stopped_org_readonly": state = "stopped_org_readonly" else: state = "complete" await self.mark_finished(crawl, status, state, stats) # check if all crawlers failed -- no crawl data was generated elif all_completed and num_done == 0 and num_failed > 0: # if stopping, and no pages finished, mark as canceled if status.stopping and not status.pagesDone: await self.mark_finished(crawl, status, "canceled", stats) else: await self.fail_crawl(crawl, status, pods, stats) # check for other statuses, default to "running" else: new_status: TYPE_RUNNING_STATES = "running" if status_count.get("generate-wacz"): new_status = "generate-wacz" elif status_count.get("uploading-wacz"): new_status = "uploading-wacz" elif status_count.get("pending-wait"): new_status = "pending-wait" await self.set_state( new_status, status, crawl, allowed_from=RUNNING_AND_WAITING_STATES ) return status # pylint: disable=too-many-arguments async def mark_finished( self, crawl: CrawlSpec, status: CrawlStatus, state: TYPE_NON_RUNNING_STATES, stats: Optional[CrawlStats] = None, ) -> bool: """mark crawl as finished, set finished timestamp and final state""" finished = dt_now() allowed_from = RUNNING_AND_WAITING_STATES # if set_state returns false, already set to same status, return if not await self.set_state( state, status, crawl, allowed_from=allowed_from, finished=finished, stats=stats, ): print("already finished, ignoring mark_finished") if not status.finished: status.finished = date_to_str(finished) return False status.finished = date_to_str(finished) if state in SUCCESSFUL_STATES: await self.inc_crawl_complete_stats(crawl, finished) # Regular Crawl Finished if not crawl.is_qa: self.run_task(self.do_crawl_finished_tasks(crawl, status, state)) # QA Run Finished else: self.run_task(self.do_qa_run_finished_tasks(crawl, state)) return True # pylint: disable=too-many-arguments async def do_crawl_finished_tasks( self, crawl: CrawlSpec, status: CrawlStatus, state: TYPE_NON_RUNNING_STATES, ) -> None: """Run tasks after crawl completes in asyncio.task coroutine.""" await self.crawl_config_ops.stats_recompute_last( crawl.cid, status.filesAddedSize, 1 ) if state in SUCCESSFUL_STATES and crawl.oid: await self.page_ops.set_archived_item_page_counts(crawl.id) await self.org_ops.inc_org_bytes_stored( crawl.oid, status.filesAddedSize, "crawl" ) await self.org_ops.set_last_crawl_finished(crawl.oid) await self.coll_ops.add_successful_crawl_to_collections(crawl.id, crawl.cid) if state in FAILED_STATES: await self.crawl_ops.delete_crawl_files(crawl.id, crawl.oid) await self.page_ops.delete_crawl_pages(crawl.id, crawl.oid) await self.event_webhook_ops.create_crawl_finished_notification( crawl.id, crawl.oid, state ) # finally, delete job await self.k8s.delete_crawl_job(crawl.id) # pylint: disable=too-many-arguments async def do_qa_run_finished_tasks( self, crawl: CrawlSpec, state: TYPE_NON_RUNNING_STATES, ) -> None: """Run tasks after qa run completes in asyncio.task coroutine.""" if state in FAILED_STATES: await self.page_ops.delete_qa_run_from_pages(crawl.db_crawl_id, crawl.id) # finally, delete job await self.k8s.delete_crawl_job(crawl.id) async def inc_crawl_complete_stats(self, crawl: CrawlSpec, finished: datetime): """Increment Crawl Stats""" started = str_to_date(crawl.started) if not started: print("Missing crawl start time, unable to increment crawl stats") return duration = int((finished - started).total_seconds()) print(f"Duration: {duration}", flush=True) await self.org_ops.inc_org_time_stats(crawl.oid, duration, False, crawl.is_qa) async def mark_for_cancelation(self, crawl_id): """mark crawl as canceled in redis""" try: redis_url = self.k8s.get_redis_url(crawl_id) redis = await self._get_redis(redis_url) if not redis: return False await redis.set(f"{crawl_id}:canceled", "1") return True finally: if redis: await redis.close()