Activity

1. milindsrivastava1997 commented on Mar 4, 2026

   

   milindsrivastava1997

   on Mar 4, 2026

   ContributorAuthor

   More actions

   Originally by @linear[bot] on 2025-09-25:

   PRO-62 Explore how to decouple remote_monitor from PrometheusClient
2. milindsrivastava1997 commented on Mar 4, 2026

   

   milindsrivastava1997

   on Mar 4, 2026

   ContributorAuthor

   More actions

   Originally by @milindsrivastava1997 on 2025-09-26:

   From Claude code

   Decoupling remote_monitor.py from PrometheusClientService

   Overview

   This document explores how to decouple experiments/remote_monitor.py from PrometheusClientService. Currently, experiments/experiment_run_e2e.py orchestrates remote_monitor, which in turn orchestrates PrometheusClientService. They're tightly coupled around profiling PID management and lifecycle coordination.

   Current Architecture Analysis

   Key Files Examined

   • /home/milind/Desktop/cmu/research/sketch_db_for_prometheus/code/Utilities/experiments/remote_monitor.py
   • /home/milind/Desktop/cmu/research/sketch_db_for_prometheus/code/Utilities/experiments/experiment_run_e2e.py
   • /home/milind/Desktop/cmu/research/sketch_db_for_prometheus/code/Utilities/experiments/experiment_utils/services/prometheus_client_service.py
   • /home/milind/Desktop/cmu/research/sketch_db_for_prometheus/code/Utilities/experiments/experiment_utils/services/remote_monitor_service.py

   Current Coupling Points

   1. Direct Instantiation (remote_monitor.py:293-308):

      • remote_monitor.py directly imports and instantiates PrometheusClientService
      • Passes profiling PIDs (profile_query_engine_pid) directly to the service
      • Controls service lifecycle (start/stop) based on execution mode

   2. Profiling Coordination (remote_monitor.py:274-337):

      • remote_monitor.py manages profiling for Flink/Arroyo PIDs
      • Coordinates timing between profiling start/stop and PrometheusClient execution
      • Hard-coded profiling logic tied to PrometheusClient execution lifecycle

   3. Configuration Coupling:

      • Both services share same config files and experiment parameters
      • PrometheusClient health monitoring blocks remote_monitor execution
      • Shared responsibility for output directory management

   Architecture Flow

   experiment_run_e2e.py → RemoteMonitorService → remote_monitor.py → PrometheusClientService
   

   Current Code Examples

   Direct Coupling in remote_monitor.py

   # Lines 293-308 in remote_monitor.py
   if args.execution_mode == "prometheus_client":
       logger.debug("Starting prometheus client")
       prometheus_client_service = PrometheusClientService(
           username="user",
           hostname_suffix="localhost",
           use_container=args.use_container_prometheus_client,
       )
       prometheus_client_service.start(
           args.experiment_mode,
           args.config_file,
           args.query_engine_config_file,
           prometheus_client_output_dir,
           args.prometheus_client_output_file,
           export_cost_and_latency,
           profile_query_engine_pid,
           args.profile_prometheus_time,
           args.prometheus_client_parallel,
       )

   Profiling PID Management

   # Lines 233-250 in remote_monitor.py
   profile_query_engine_pid = None
   if args.profile_query_engine:
       if (
           constants.QUERY_ENGINE_PY_PROCESS_KEYWORD in args.keywords
           or constants.QUERY_ENGINE_PY_CONTAINER_NAME in args.keywords
       ):
           query_engine_pids = get_pids(constants.QUERY_ENGINE_PY_PROCESS_KEYWORD)
           profile_query_engine_pid = query_engine_pids[
               0
           ]  # Take first PID for profiling
       elif (
           constants.QUERY_ENGINE_RS_PROCESS_KEYWORD in args.keywords
           or constants.QUERY_ENGINE_RS_CONTAINER_NAME in args.keywords
       ):
           raise NotImplementedError(
               "Profiling for Rust query engine is not implemented yet"
           )

   Decoupling Strategies

   1. Event-Driven Architecture

   Replace direct coupling with event/message-based communication:

   # Event bus pattern
   class ExperimentEventBus:
       def publish(self, event_type: str, data: dict)
       def subscribe(self, event_type: str, callback: Callable)
   
   # Events: "profiling_started", "profiling_stopped", "experiment_phase_change"

   2. Service Registry Pattern

   Decouple service discovery and lifecycle management:

   class ServiceRegistry:
       def register_service(self, name: str, service: BaseService)
       def get_service(self, name: str) -> Optional[BaseService]
       def notify_service_event(self, event: ServiceEvent)

   3. Abstract Profiling Manager

   Extract profiling concerns into dedicated component:

   class ProfilingManager:
       def start_profiling(self, target: str, pids: List[int])
       def stop_profiling(self, target: str, store_results: bool)
       def register_profiling_listener(self, listener: ProfilingListener)

   4. Configuration Mediator

   Centralize configuration management:

   class ExperimentConfigMediator:
       def get_service_config(self, service_name: str) -> dict
       def notify_config_change(self, service_name: str, config: dict)

   Recommended Decoupling Approach

   Phase 1: Extract Profiling Logic

   • Create ProfilerService to handle Flink/Arroyo/QueryEngine profiling
   • Move profiling methods from remote_monitor.py to dedicated service
   • Use dependency injection to provide profiler to remote_monitor

   Phase 2: Abstract Client Execution

   • Create ClientExecutor interface with PrometheusClientExecutor implementation
   • Remove direct PrometheusClientService dependency from remote_monitor
   • Allow remote_monitor to work with any client executor

   Phase 3: Event-Driven Coordination

   • Implement experiment lifecycle events
   • PrometheusClient subscribes to profiling events
   • Remote monitor publishes experiment state changes

   This approach would:

   • Remove direct service dependencies
   • Enable independent testing of components
   • Support different client implementations
   • Simplify remote_monitor responsibility to process monitoring + coordination

   The key insight is that remote_monitor.py currently has two distinct responsibilities: process monitoring and experiment orchestration. Separating these concerns would enable the decoupling you're looking for.

   Event-Driven Implementation

   Simple In-Process Event Bus (No Dependencies)

   # events.py
   from dataclasses import dataclass
   from typing import Any, Dict, List, Callable
   from enum import Enum
   
   class EventType(Enum):
       PROFILING_STARTED = "profiling_started"
       PROFILING_STOPPED = "profiling_stopped"
       EXPERIMENT_PHASE_CHANGED = "experiment_phase_changed"
       PROMETHEUS_CLIENT_STARTED = "prometheus_client_started"
       PROMETHEUS_CLIENT_FINISHED = "prometheus_client_finished"
   
   @dataclass
   class Event:
       type: EventType
       source: str
       data: Dict[str, Any]
       timestamp: float = None
   
   class EventBus:
       def __init__(self):
           self._subscribers: Dict[EventType, List[Callable]] = {}
   
       def subscribe(self, event_type: EventType, callback: Callable[[Event], None]):
           if event_type not in self._subscribers:
               self._subscribers[event_type] = []
           self._subscribers[event_type].append(callback)
   
       def publish(self, event: Event):
           if event.type in self._subscribers:
               for callback in self._subscribers[event.type]:
                   try:
                       callback(event)
                   except Exception as e:
                       print(f"Error in event handler: {e}")
   
   # Global event bus instance
   event_bus = EventBus()

   Refactored Remote Monitor

   # remote_monitor_decoupled.py
   from events import EventBus, Event, EventType, event_bus
   
   class RemoteMonitor:
       def __init__(self, event_bus: EventBus):
           self.event_bus = event_bus
           self.profiler_service = None
           self.client_executor = None
   
           # Subscribe to events we care about
           self.event_bus.subscribe(
               EventType.PROMETHEUS_CLIENT_STARTED,
               self._on_prometheus_client_started
           )
           self.event_bus.subscribe(
               EventType.PROMETHEUS_CLIENT_FINISHED,
               self._on_prometheus_client_finished
           )
   
       def start_monitoring(self, args):
           # Start process monitoring
           self._start_process_monitors(args)
   
           # Publish event that monitoring started
           self.event_bus.publish(Event(
               type=EventType.EXPERIMENT_PHASE_CHANGED,
               source="remote_monitor",
               data={
                   "phase": "monitoring_started",
                   "experiment_mode": args.experiment_mode,
                   "pids": self.monitored_pids
               }
           ))
   
           # Start client execution (no longer directly coupled)
           if args.execution_mode == "prometheus_client":
               self.client_executor.start(args)
   
       def _on_prometheus_client_started(self, event: Event):
           """React to prometheus client starting"""
           logger.debug("Prometheus client started, adjusting monitoring")
   
           # Start profiling if needed
           if self.should_profile_flink(event.data):
               self.profiler_service.start_flink_profiling(
                   event.data.get("flink_pids", [])
               )
   
       def _on_prometheus_client_finished(self, event: Event):
           """React to prometheus client finishing"""
           logger.debug("Prometheus client finished, stopping profiling")
   
           # Stop profiling
           self.profiler_service.stop_all_profiling(store_results=True)

   Decoupled PrometheusClient Service

   # prometheus_client_decoupled.py
   from events import Event, EventType, event_bus
   
   class PrometheusClientService:
       def __init__(self, event_bus: EventBus):
           self.event_bus = event_bus
   
           # Subscribe to relevant events
           self.event_bus.subscribe(
               EventType.PROFILING_STARTED,
               self._on_profiling_started
           )
   
       def start(self, args):
           # Publish that we're starting
           self.event_bus.publish(Event(
               type=EventType.PROMETHEUS_CLIENT_STARTED,
               source="prometheus_client",
               data={
                   "profile_query_engine_pid": args.profile_query_engine_pid,
                   "flink_pids": getattr(args, 'flink_pids', []),
                   "experiment_mode": args.experiment_mode
               }
           ))
   
           # Do the actual work
           self._execute_queries(args)
   
           # Publish that we're done
           self.event_bus.publish(Event(
               type=EventType.PROMETHEUS_CLIENT_FINISHED,
               source="prometheus_client",
               data={"success": True}
           ))
   
       def _on_profiling_started(self, event: Event):
           """Adjust behavior based on profiling events"""
           if event.data.get("target") == "query_engine":
               self.profiling_active = True

   Profiling Service

   # profiling_service.py
   from events import Event, EventType, event_bus
   
   class ProfilingService:
       def __init__(self, event_bus: EventBus):
           self.event_bus = event_bus
           self.active_profiles = {}
   
       def start_flink_profiling(self, pids: List[int]):
           """Start profiling Flink processes"""
           for pid in pids:
               self._start_asprof(pid)
   
           self.active_profiles["flink"] = pids
   
           # Notify others that profiling started
           self.event_bus.publish(Event(
               type=EventType.PROFILING_STARTED,
               source="profiling_service",
               data={
                   "target": "flink",
                   "pids": pids
               }
           ))
   
       def stop_all_profiling(self, store_results: bool = True):
           """Stop all active profiling"""
           for target, pids in self.active_profiles.items():
               if target == "flink":
                   self._stop_asprof(pids, store_results)
               elif target == "arroyo":
                   self._stop_flamegraph(pids, store_results)
   
           self.event_bus.publish(Event(
               type=EventType.PROFILING_STOPPED,
               source="profiling_service",
               data={
                   "targets": list(self.active_profiles.keys()),
                   "stored": store_results
               }
           ))
   
           self.active_profiles.clear()

   Library Options for Event Bus

   1. No Dependencies (Recommended for Your Use Case)

   The simple event bus shown above requires no external libraries. This is often the best choice for:

   • Single-process applications
   • Simple pub/sub patterns
   • Avoiding dependency bloat

   2. Lightweight Libraries

   PyDispatcher (minimal dependency):

   # pip install PyDispatcher
   from pydispatch import dispatcher
   
   # Subscribe
   dispatcher.connect(handle_profiling_started, signal="profiling_started")
   
   # Publish
   dispatcher.send(signal="profiling_started", sender="profiler", pids=[123, 456])

   blinker (Flask's event system):

   # pip install blinker
   from blinker import Namespace
   
   signals = Namespace()
   profiling_started = signals.signal('profiling-started')
   
   # Subscribe
   profiling_started.connect(handle_profiling_started)
   
   # Publish
   profiling_started.send(sender="profiler", pids=[123, 456])

   3. Full-Featured Message Brokers

   Redis Pub/Sub (if you need cross-process events):

   # pip install redis
   import redis
   
   r = redis.Redis()
   
   # Publisher
   r.publish('profiling_started', json.dumps({"pids": [123, 456]}))
   
   # Subscriber
   pubsub = r.pubsub()
   pubsub.subscribe('profiling_started')
   for message in pubsub.listen():
       handle_event(json.loads(message['data']))

   RabbitMQ/Celery (overkill for your use case):

   # For distributed systems - probably too heavy

   Recommendation: Start Simple

   For your experiment orchestration use case, I'd recommend:

   Option 1: Custom Event Bus (No Dependencies)

   • ✅ Zero dependencies
   • ✅ Full control over behavior
   • ✅ Easy to debug and test
   • ✅ Fits your single-process model
   • ❌ Need to implement yourself

   Option 2: blinker (Minimal Dependency)

   • ✅ Battle-tested (used by Flask)
   • ✅ Clean API
   • ✅ Good documentation
   • ✅ Only ~10KB package
   • ❌ One more dependency

   Integration Example

   # experiment_coordinator.py
   from events import event_bus, EventType, Event
   
   class ExperimentCoordinator:
       def __init__(self):
           self.services = {
               'remote_monitor': RemoteMonitor(event_bus),
               'profiler': ProfilingService(event_bus),
               'prometheus_client': PrometheusClientService(event_bus)
           }
   
       def run_experiment(self, args):
           # Start monitoring first
           self.services['remote_monitor'].start_monitoring(args)
   
           # Services coordinate through events automatically
           # No more direct coupling!

   Benefits of This Approach

   The key benefits of this event-driven approach:

   1. Loose Coupling: Services only know about events, not each other
   2. Testability: Easy to mock events for unit testing
   3. Extensibility: New services can listen to existing events
   4. Debugging: Event flow is explicit and observable

   For your CloudLab experiment orchestration, I'd start with the custom event bus since it's simple, has no dependencies, and perfectly fits your single-process coordination needs. You can always upgrade to blinker later if you need more features.

   Next Steps

   1. Phase 1: Implement the simple event bus
   2. Phase 2: Extract profiling logic into ProfilingService
   3. Phase 3: Refactor remote_monitor.py to use events instead of direct coupling
   4. Phase 4: Update PrometheusClientService to publish/subscribe to events
   5. Phase 5: Test and validate the decoupled system

   This approach maintains all existing functionality while dramatically reducing coupling between components.
3. milindsrivastava1997 commented on Mar 4, 2026

   

   milindsrivastava1997

   on Mar 4, 2026

   ContributorAuthor

   More actions

   Originally by @linear[bot] on 2025-11-07:

   PRO-62
4. milindsrivastava1997

   added

   prometheus-client

   asap-tools

   on Mar 4, 2026
5. milindsrivastava1997

   closed this as completedon May 26, 2026