diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 46ea77940..b1ac2396d 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -778,6 +778,34 @@ class LogIngestor(BaseModel): logging_level: LoggingLevelRust = "INFO" +class Spider(BaseModel): + host: DomainStr = "localhost" + port: Port = 6000 + + +class SpiderResourceGroup(BaseModel): + name: NonEmptyStr + + +class PollingBackoff(BaseModel): + init_backoff_millisecs: PositiveInt + max_backoff_millisecs: PositiveInt + + +class CompressionCoordinator(BaseModel): + resource_group: SpiderResourceGroup = SpiderResourceGroup(name="compression-coordinator") + job_polling_interval_millisecs: PositiveInt = 100 + result_polling: PollingBackoff = PollingBackoff( + init_backoff_millisecs=100, max_backoff_millisecs=1000 + ) + compression_task_max_retry: NonNegativeInt = 1 + commit_task_max_retry: NonNegativeInt = 1 + database_connection_pool_size: PositiveInt = 10 + termination_timeout_secs: PositiveInt = 30 + commit_task_soft_timeout_secs: PositiveInt = 45 + commit_task_hard_timeout_secs: PositiveInt = 60 + + class Presto(BaseModel): DEFAULT_PORT: ClassVar[int] = 8080 @@ -830,6 +858,8 @@ class ClpConfig(BaseModel): garbage_collector: GarbageCollector = GarbageCollector() api_server: ApiServer | None = ApiServer() log_ingestor: LogIngestor | None = LogIngestor() + spider: Spider | None = None + compression_coordinator: CompressionCoordinator | None = None credentials_file_path: SerializablePath = CLP_DEFAULT_CREDENTIALS_FILE_PATH mcp_server: McpServer | None = None @@ -1029,6 +1059,21 @@ def validate_log_ingestor_config(self): raise ValueError(msg) return self + @model_validator(mode="after") + def validate_compression_coordinator_config(self): + if self.compression_coordinator is None: + return self + if self.package.storage_engine != StorageEngine.CLP_S: + msg = ( + "compression-coordinator is only compatible with storage engine " + f"`{StorageEngine.CLP_S}`." + ) + raise ValueError(msg) + if self.spider is None: + msg = "compression-coordinator requires Spider to be configured." + raise ValueError(msg) + return self + @model_validator(mode="after") def validate_presto_config(self): query_engine = self.webui.query_engine diff --git a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py index 85dac3832..5af48908a 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py @@ -68,7 +68,8 @@ def main(argv): `status_msg` VARCHAR(512) NOT NULL DEFAULT '', `creation_time` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), `start_time` DATETIME(3) NULL DEFAULT NULL, - `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP(), + `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP() + ON UPDATE CURRENT_TIMESTAMP(), `duration` FLOAT NULL DEFAULT NULL, `original_size` BIGINT NOT NULL DEFAULT '0', `uncompressed_size` BIGINT NOT NULL DEFAULT '0', @@ -76,11 +77,14 @@ def main(argv): `num_tasks` INT NOT NULL DEFAULT '0', `num_tasks_completed` INT NOT NULL DEFAULT '0', `clp_binary_version` INT NULL DEFAULT NULL, + `spider_id` BIGINT UNSIGNED NULL DEFAULT NULL, + `dispatch_time` DATETIME NULL DEFAULT NULL, `clp_config` MEDIUMBLOB NOT NULL, PRIMARY KEY (`id`) USING BTREE, INDEX `JOB_STATUS` (`status`) USING BTREE, INDEX `JOB_UPDATE_TIME` (`update_time`) USING BTREE, - INDEX `JOB_START_TIME_STATUS` (`start_time`, `status`) USING BTREE + INDEX `JOB_START_TIME_STATUS` (`start_time`, `status`) USING BTREE, + INDEX `JOB_SPIDER_ID` (`spider_id`) USING BTREE ) ROW_FORMAT=DYNAMIC """ ) diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index 4f7ec4120..8093c6c1f 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -467,6 +467,16 @@ def _schedule_job( ) return elif input_type == InputType.S3_OBJECT_METADATA.value: + if clp_config.compression_coordinator is not None: + # NOTE: These jobs are left in PENDING and will eventually be picked up and + # scheduled by the compression-coordinator. + logger.info( + "compression-coordinator is configured to handle compression jobs submitted" + " by log-ingestor. Skipping job %d.", + job_id, + ) + return + try: _process_s3_object_metadata_input( input_config, paths_to_compress_buffer, db_context