class GenericProcessor(BaseProcessor):
"""
Common signature of a processor in DPR-service.
NOTE: a new instance of this class is called for every endpoint call.
"""
def __init__(
self,
db_process_manager: PostgreSQLManager,
cluster_info: ClusterInfo,
local_mode_address: str = "",
tasktable_module: str = "",
tasktable_class: str = "",
): # pylint: disable=super-init-not-called
self.tasktable_module = tasktable_module
self.tasktable_class = tasktable_class
self.job_logger = JobLogger(db_process_manager)
self.cluster_handler = DaskClusterHandler(cluster_info, local_mode_address)
def get_processor_name(self):
"""Return the normalized processor name in lowercase"""
return self.__class__.__name__.lower().replace("processor", "")
async def get_tasktable(self):
"""Return the EOPF tasktable for a given module and class names"""
dask_client = self.cluster_handler.setup_dask_connection()
try:
# Extract span infos to send to Dask
span_context = trace.get_current_span().get_span_context()
dpr_processor = call_dask.ProcessorCaller(
caller_env=dict(os.environ),
span_context=span_context,
dask_gateway_address=self.cluster_handler.dask_gateway_address,
cluster_info=self.cluster_handler.cluster_info,
processor_name=self.get_processor_name(),
job_id=self.job_logger.job_id,
data={}, # not used for the tasktables
)
# Run processor in the dask client
task_table_task = dask_client.submit(
dpr_processor.get_tasktable,
module_name=self.tasktable_module,
class_name=self.tasktable_class,
pure=False, # disable cache
)
res = task_table_task.result()
return res
except Exception as e: # pylint: disable=broad-exception-caught
logger.exception(f"Submitting task to dask cluster failed. Reason: {traceback.format_exc()}")
raise e
finally:
# cleanup by disconnecting the dask client
dask_client.close()
# Override from BaseProcessor, execute is async in RSPYProcessor
async def execute( # pylint: disable=invalid-overridden-method
self,
data: dict,
_outputs=None,
) -> tuple[str, dict]:
"""
Asynchronously execute the dpr process in the dask cluster
"""
# Set experimental configuration from env var
if env_bool("DPR_LOCAL_CLUSTER", default=False):
data.setdefault("experimental_config", {})
data["experimental_config"].setdefault("local_cluster", {})
data["experimental_config"]["local_cluster"]["enabled"] = True
logger.debug(f"Executing processor for {data}")
self.job_logger.log_job_execution(JobStatus.running, 0, "Processor execution started")
# Start execution
loop = asyncio.get_event_loop()
if loop.is_running():
# If the loop is running, schedule the async function
_ = asyncio.create_task(self.start_processor(data))
else:
# If the loop is not running, run it until complete
loop.run_until_complete(self.start_processor(data))
return self.job_logger.get_execute_result()
async def start_processor(
self,
data: dict,
) -> tuple[str, dict]:
"""
Method used to trigger dask distributed streaming process.
It creates dask client object, gets the external dpr_payload sources access token
Prepares the tasks for execution
Manage eventual runtime exceptions
Args:
catalog_collection (str): Name of the catalog collection.
Returns:
tuple: tuple of MIME type and process response (dictionary containing the job ID and a
status message).
Example: ("application/json", {"running": <job_id>})
"""
logger.debug(f"Starting main loop for {data}")
try:
experimental_config = ExperimentalConfig(**data.get("experimental_config", {}))
# For testing: run the eopf-cpm scheduler on local.
# It will then init a dask LocalCluster instance itself.
if LOCAL_MODE and experimental_config.local_cluster.enabled:
dask_client = None
# Nominal case: run the eopf-cpm scheduler on a dedicated cluster pod, not locally.
else:
dask_client = self.cluster_handler.setup_dask_connection()
except KeyError as ke:
return self.job_logger.log_job_execution(
JobStatus.failed,
0,
f"Failed to start the dpr-service process: No env var {ke} found",
log_exception=True,
)
except RuntimeError:
return self.job_logger.log_job_execution(
JobStatus.failed,
0,
f"Failed to start the dpr-service process: {traceback.format_exc()}",
log_exception=True,
)
self.job_logger.log_job_execution(JobStatus.running, 0, "Sending task to the dask cluster")
# Manage dask tasks in a separate thread
# starting a thread for managing the dask callbacks
logger.debug("Starting tasks monitoring thread")
try:
await asyncio.to_thread(
self.manage_dask_tasks,
dask_client,
data,
)
except Exception: # pylint: disable=broad-exception-caught
self.job_logger.log_job_execution(
JobStatus.failed,
0,
f"Error from tasks monitoring thread: {traceback.format_exc()}",
log_exception=True,
)
# cleanup by disconnecting the dask client
if dask_client:
dask_client.close()
return self.job_logger.get_execute_result()
def manage_dask_tasks(self, dask_client: Client | None, data: dict):
"""
Manages Dask tasks where the dpr processor is started.
"""
if dask_client:
logger.info(f"Processing tasks monitoring started with dask client {dask_client}")
res = None
self.job_logger.log_job_execution(
JobStatus.running,
50,
"In progress",
)
processor_name = self.get_processor_name()
with start_span(__name__, f"dpr_manage_dask_tasks_{processor_name}") as span:
try:
# Build the processor caller with span context
dpr_processor = call_dask.ProcessorCaller(
caller_env=dict(os.environ),
span_context=span.get_span_context(),
dask_gateway_address=self.cluster_handler.dask_gateway_address,
cluster_info=self.cluster_handler.cluster_info,
processor_name=processor_name,
job_id=self.job_logger.job_id,
data=data,
)
# Nominal usecase: run processor in the dask client
if dask_client:
dpr_task = dask_client.submit(
dpr_processor.run_processor,
pure=False, # disable cache
)
# Specific case for local debugging
else:
dpr_task = None
res = dpr_processor.run_processor()
except Exception as e: # pylint: disable=broad-exception-caught
if not dask_client:
logger.exception(traceback.format_exc())
raise
self._handle_error(span, e, f"Submitting task to dask cluster failed: {traceback.format_exc()}")
return
try:
if dpr_task:
res = dpr_task.result() # This will raise the exception from the task if it failed
logger.info("%s Task streaming completed", dpr_task.key)
except Exception as e: # pylint: disable=broad-exception-caught
self._handle_error(span, e, f"Processing task failed: {traceback.format_exc()}")
return
# Update status and insert the result of the dask task in the jobs table
self.job_logger.log_job_execution(JobStatus.successful, 100, str(res))
# Update the subscribers for token refreshment
logger.info("Tasks monitoring finished")
def _handle_error(self, span: Span, e: Exception, error_msg: str):
record_error(span, e)
self.job_logger.log_job_execution(JobStatus.failed, None, error_msg, log_exception=True)