mode updates

Author: milindsrivastava1997

asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md

@@ -136,9 +136,9 @@ These parameters must be provided for all experiment scripts:
 
 #### `streaming.engine` (string, optional)
 - **Description**: Which streaming engine to use
-- **Default**: `"flink"`
-- **Choices**: `"flink"`, `"arroyo"`
-- **Example**: `"arroyo"`
+- **Default**: `"precompute"`
+- **Choices**: `"flink"`, `"arroyo"`, `"precompute"`
+- **Example**: `"precompute"`
 - **Usage**: Selects streaming processing framework
 
 #### `streaming.flink_input_format` (string, optional)

asap-tools/experiments/HYDRA_CONFIG_USAGE.md

@@ -52,7 +52,7 @@ flow.no_teardown: true/false
 flow.steady_state_wait: 300  # seconds
 
 # Streaming engine
-streaming.engine: flink|arroyo
+streaming.engine: flink|arroyo|precompute
 streaming.flink_input_format: json|avro-json|avro-binary
 streaming.flink_output_format: json|byte
 streaming.enable_object_reuse: true/false

asap-tools/experiments/config/config.yaml

@@ -78,10 +78,7 @@ fake_exporter_language: "rust"  # choices: ["python", "rust"]
 # Cluster data exporter configuration
 cluster_data_directory: "/data/cluster_traces"  # Path to directory containing Google/Alibaba cluster trace data
 
-# Query engine language
-query_engine_language: "rust"  # choices: ["python", "rust"]
-
-# Query language (SQL vs PROMQL) - only used by Rust query engine
+# Query language (SQL vs PROMQL)
 query_language: "PROMQL"  # choices: ["SQL", "PROMQL"]
 
 # Query engine options
@@ -120,7 +117,7 @@ prometheus_client:
 
 # Container deployment settings
 use_container:
-  query_engine: true       # QueryEngineService - containerized query engine
+  query_engine: true       # QueryEngineRustService - containerized query engine
   arroyo: true            # ArroyoService - containerized Arroyo streaming engine
   controller: true        # ControllerService - containerized controller
   fake_exporter: true     # ExporterServiceFactory - containerized fake exporters

asap-tools/experiments/constants.py

@@ -14,9 +14,7 @@
 FLINK_OUTPUT_TOPIC = "flink_output"
 KAFKA_BROKER = "localhost:9092"
 
-QUERY_ENGINE_PY_PROCESS_KEYWORD = "main_query_engine.py"
 QUERY_ENGINE_RS_PROCESS_KEYWORD = "query_engine_rust"
-QUERY_ENGINE_PY_CONTAINER_NAME = "sketchdb-queryengine"
 QUERY_ENGINE_RS_CONTAINER_NAME = "sketchdb-queryengine-rust"
 
 ARROYO_IMAGE = "ghcr.io/projectasap/asap-arroyo:v0.1.0"

asap-tools/experiments/experiment_only_ingest_path.py

@@ -242,12 +242,16 @@ def main(cfg: DictConfig):
             "controller_client_configs",
             f"{experiment_mode}.yaml",
         )
+        prometheus_url = (
+            f"http://localhost:{prometheus_service.get_query_endpoint_port()}"
+        )
         controller_service.start(
             controller_input_file=controller_client_config,
             prometheus_scrape_interval=prometheus_scrape_interval,
             streaming_engine=args.streaming_engine,
             controller_remote_output_dir=CONTROLLER_REMOTE_OUTPUT_DIR,
             punting=args.controller_punting,
+            prometheus_url=prometheus_url,
         )
         sync.rsync_controller_config_remote_to_local(
             provider,
@@ -257,19 +261,21 @@ def main(cfg: DictConfig):
         )
 
         # Start Kafka
-        kafka_service.start()
-        kafka_service.wait_until_ready()
-        kafka_service.delete_topics()
-        kafka_service.create_topics()
+        if args.streaming_engine != "precompute":
+            kafka_service.start()
+            kafka_service.wait_until_ready()
+            kafka_service.delete_topics()
+            kafka_service.create_topics()
 
         # Start Arroyo
-        arroyo_service.stop()
-        time.sleep(10)
-        arroyo_service.start(
-            experiment_output_dir=experiment_output_dir,
-            remote_write_base_port=args.remote_write_base_port,
-            parallelism=args.parallelism,
-        )
+        if args.streaming_engine != "precompute":
+            arroyo_service.stop()
+            time.sleep(10)
+            arroyo_service.start(
+                experiment_output_dir=experiment_output_dir,
+                remote_write_base_port=args.remote_write_base_port,
+                parallelism=args.parallelism,
+            )
 
     # Start fake exporter if configured
     if config.check_exporter_and_queries_exist("fake_exporter", cfg.experiment_params):
@@ -317,7 +323,7 @@ def main(cfg: DictConfig):
         prometheus_service.start(experiment_output_dir)
 
     # Start V2-specific: Run ArroyoSketch pipeline
-    if is_v2:
+    if is_v2 and args.streaming_engine != "precompute":
         print("Starting ArroyoSketch pipeline...")
         arroyosketch_pipeline_id = arroyo_service.run_arroyosketch(
             experiment_name=args.experiment_name,
@@ -405,8 +411,9 @@ def main(cfg: DictConfig):
         if arroyosketch_pipeline_id:
             arroyo_service.stop_arroyosketch(arroyosketch_pipeline_id)
         arroyo_service.stop()
-        kafka_service.delete_topics()
-        kafka_service.stop()
+        if args.streaming_engine != "precompute":
+            kafka_service.delete_topics()
+            kafka_service.stop()
         controller_service.stop()
 
     # Stop core services

asap-tools/experiments/experiment_run_e2e.py

@@ -12,7 +12,7 @@
 from experiment_utils.services import (
     KafkaService,
     FlinkService,
-    QueryEngineServiceFactory,
+    QueryEngineRustService,
     ExporterServiceFactory,
     PrometheusKafkaAdapterService,
     ArroyoService,
@@ -135,9 +135,7 @@ def main(cfg: DictConfig):
     # Initialize services
     kafka_service = KafkaService(provider, args.node_offset, num_tries=KAFKA_NUM_TRIES)
     flink_service = FlinkService(provider, args.node_offset)
-    # Initialize query engine service based on language
-    query_engine_service = QueryEngineServiceFactory.create_query_engine_service(
-        args.query_engine_language,
+    query_engine_service = QueryEngineRustService(
         provider,
         use_container=args.use_container_query_engine,
         node_offset=args.node_offset,

asap-tools/experiments/experiment_run_grafana_demo.py

@@ -11,7 +11,7 @@
 from experiment_utils.providers.factory import create_provider
 from experiment_utils.services import (
     KafkaService,
-    QueryEngineServiceFactory,
+    QueryEngineRustService,
     ExporterServiceFactory,
     ArroyoService,
     ArroyoThroughputMonitor,
@@ -120,9 +120,7 @@ def main(cfg: DictConfig):
 
     # Initialize services
     kafka_service = KafkaService(provider, args.node_offset, num_tries=KAFKA_NUM_TRIES)
-    # Initialize query engine service based on language
-    query_engine_service = QueryEngineServiceFactory.create_query_engine_service(
-        args.query_engine_language,
+    query_engine_service = QueryEngineRustService(
         provider,
         use_container=args.use_container_query_engine,
         node_offset=args.node_offset,
@@ -289,23 +287,28 @@ def main(cfg: DictConfig):
 
     # copy_controller_client_config(args.controller_client_config, local_experiment_dir)
     if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME:
+        prometheus_url = (
+            f"http://localhost:{prometheus_service.get_query_endpoint_port()}"
+        )
         controller_service.start(
             controller_input_file=controller_client_config,
             prometheus_scrape_interval=prometheus_scrape_interval,
             streaming_engine=args.streaming_engine,
             controller_remote_output_dir=CONTROLLER_REMOTE_OUTPUT_DIR,
             punting=args.controller_punting,
+            prometheus_url=prometheus_url,
         )
         sync.rsync_controller_config_remote_to_local(
             provider,
             CONTROLLER_REMOTE_OUTPUT_DIR,
             CONTROLLER_LOCAL_OUTPUT_DIR,
             node_offset=args.node_offset,
         )
-        kafka_service.start()
-        kafka_service.wait_until_ready()
-        kafka_service.delete_topics()
-        kafka_service.create_topics()
+        if args.streaming_engine != "precompute":
+            kafka_service.start()
+            kafka_service.wait_until_ready()
+            kafka_service.delete_topics()
+            kafka_service.create_topics()
 
     if config.check_exporter_and_queries_exist("fake_exporter", cfg.experiment_params):
         # this DOES NOT block

asap-tools/experiments/experiment_teardown_everything.py

@@ -18,7 +18,7 @@
 from experiment_utils.services import (
     KafkaService,
     FlinkService,
-    QueryEngineServiceFactory,
+    QueryEngineRustService,
     ExporterServiceFactory,
     PrometheusKafkaAdapterService,
     ArroyoService,
@@ -72,22 +72,11 @@ def main(cfg: DictConfig):
     kafka_service = KafkaService(provider, args.node_offset, num_tries=KAFKA_NUM_TRIES)
     flink_service = FlinkService(provider, args.node_offset)
 
-    # Initialize both query engine languages
-    query_engine_service_rust = QueryEngineServiceFactory.create_query_engine_service(
-        "rust", provider, use_container=True, node_offset=args.node_offset
-    )
-    query_engine_service_python = QueryEngineServiceFactory.create_query_engine_service(
-        "python", provider, use_container=True, node_offset=args.node_offset
-    )
-    query_engine_service_rust_native = (
-        QueryEngineServiceFactory.create_query_engine_service(
-            "rust", provider, use_container=False, node_offset=args.node_offset
-        )
+    query_engine_service_container = QueryEngineRustService(
+        provider, use_container=True, node_offset=args.node_offset
     )
-    query_engine_service_python_native = (
-        QueryEngineServiceFactory.create_query_engine_service(
-            "python", provider, use_container=False, node_offset=args.node_offset
-        )
+    query_engine_service_native = QueryEngineRustService(
+        provider, use_container=False, node_offset=args.node_offset
     )
 
     system_exporters_service = SystemExportersService(
@@ -176,10 +165,8 @@ def main(cfg: DictConfig):
         ("Prometheus Client (container)", prometheus_client_service_container),
         ("Prometheus Client (native)", prometheus_client_service_native),
         ("Remote Monitor", remote_monitor_service),
-        ("Query Engine Rust (container)", query_engine_service_rust),
-        ("Query Engine Python (container)", query_engine_service_python),
-        ("Query Engine Rust (native)", query_engine_service_rust_native),
-        ("Query Engine Python (native)", query_engine_service_python_native),
+        ("Query Engine (container)", query_engine_service_container),
+        ("Query Engine (native)", query_engine_service_native),
         ("Kafka", kafka_service),
         ("Prometheus-Kafka Adapter", prometheus_kafka_adapter_service),
         ("System Exporters", system_exporters_service),

asap-tools/experiments/experiment_utils/config.py

@@ -413,9 +413,6 @@ def __init__(self, cfg: DictConfig):
         # Fake exporter language
         self.fake_exporter_language = cfg.fake_exporter_language
 
-        # Query engine language
-        self.query_engine_language = cfg.query_engine_language
-
         # Query language (SQL vs PROMQL) - only used by Rust query engine
         self.query_language = cfg.query_language
 
@@ -511,17 +508,6 @@ def validate_config(cfg: DictConfig, script_name: str = "experiment_run_e2e"):
                 f"Valid options: {valid_policies}"
             )
 
-        # Validate Python query engine only supports no_cleanup
-        if (
-            hasattr(cfg, "query_engine_language")
-            and cfg.query_engine_language == "python"
-            and policy != "no_cleanup"
-        ):
-            raise ValueError(
-                f"aggregate_cleanup.policy='{policy}' is not supported by the Python query engine. "
-                "Either use query_engine_language='rust' or set aggregate_cleanup.policy='no_cleanup'"
-            )
-
 
 def generate_and_copy_prometheus_config(
     num_nodes_in_experiment,

asap-tools/experiments/experiment_utils/services/__init__.py

@@ -9,9 +9,7 @@
 from .kafka import KafkaService
 from .flink import FlinkService
 from .query_engine import (
-    QueryEngineService,
     QueryEngineRustService,
-    QueryEngineServiceFactory,
 )
 from .monitoring import MonitoringService
 from .fake_exporters import (
@@ -131,9 +129,7 @@ def create_prometheus_service(cfg, provider, num_nodes: int, node_offset: int):
     "DockerServiceBase",
     "KafkaService",
     "FlinkService",
-    "QueryEngineService",
     "QueryEngineRustService",
-    "QueryEngineServiceFactory",
     "MonitoringService",
     "ExporterServiceFactory",
     "PythonExporterService",