schedule_latency_test
Schedule network latency tests between hosts using pScheduler to measure packet transit times and identify performance issues.
Instructions
Schedule a latency test using pScheduler.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| source | No | Source host (optional) | |
| dest | Yes | Destination host | |
| packetCount | No | Number of packets | |
| packetInterval | No | Interval between packets |
Implementation Reference
- src/perfsonar_mcp/pscheduler.py:132-178 (handler)The actual implementation of scheduling a latency test in the PSchedulerClient class.
async def schedule_latency_test( self, source: Optional[str], dest: str, packet_count: int = 600, packet_interval: float = 0.1, slip: str = "PT10M", ) -> PSchedulerTaskResponse: """ Schedule a latency (one-way delay) test Args: source: Source host (None for local) dest: Destination host packet_count: Number of packets to send packet_interval: Interval between packets in seconds slip: Schedule slip time in ISO 8601 format (e.g., PT10M for 10 minutes) Returns: Task response """ logger.info( f"Scheduling latency test from {source or 'local'} to {dest} ({packet_count} packets)" ) # Determine which node to schedule on (prefer source if available) scheduler_node = source or dest scheduler_url = f"https://{scheduler_node}/pscheduler" logger.info(f"Using pScheduler at: {scheduler_url}") test_spec = LatencyTestSpec( source=source, dest=dest, packet_count=packet_count, packet_interval=packet_interval ) task_request = PSchedulerTaskRequest( test=PSchedulerTestSpec( type="latency", spec=test_spec.model_dump(by_alias=True, exclude_none=True) ), schedule={"slip": slip}, ) # Create a temporary client for this specific scheduler client = PSchedulerClient(scheduler_url) try: return await client.create_task(task_request) finally: await client.close()