Skip to content

Commit

Permalink
Merge pull request #118 from lsst-ts/tickets/DM-41536
Browse files Browse the repository at this point in the history 
DM-41536: Refactor LOVE stress and uptime tests for configuring URLs instead of domains
  • Loading branch information
@sebastian-aranda
sebastian-aranda authored Nov 9, 2023
2 parents 2105a83 + 0cd8306 commit 81e0ddc
Show file tree
Hide file tree
Showing 7 changed files with 124 additions and 192 deletions.
3 changes: 3 additions & 0 deletions doc/news/DM-41536.misc.rst
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
* In ``love_manager_client``, ``make_love_stress_tests`` and ``make_love_uptime_tests`` change location attribute to be an URL instead of a domain
* In ``love_manager_client`` remove ``command_url``
* In ``make_love_stress_tests`` and ``make_love_uptime_tests`` make both ``USER_USERNAME`` and ``USER_USER_PASS`` environment variables required
18 changes: 12 additions & 6 deletions python/lsst/ts/externalscripts/love_manager_client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,8 @@ class LoveManagerClient:
Parameters
----------
location : `str`
Host of the running LOVE-manager instance
Complete URL of the running LOVE instance
e.g. https://base-lsp.lsst.codes/love or http://love01.ls.lsst.org
username: `str`
LOVE username to use as authenticator
password: `str`
Expand Down Expand Up @@ -73,7 +74,6 @@ def __init__(

self.token = None
self.websocket_url = None
self.command_url = None
self.__api_headers = None

self.__msg_tracing = msg_tracing
Expand All @@ -84,6 +84,9 @@ def __init__(
self.__password = password
self.__websocket = None

self.__secure = location.split(":")[0] == "https"
self.__domain = location.split("//")[1]

self.start_task = utils.make_done_future()

async def __request_token(self):
Expand All @@ -96,7 +99,7 @@ async def __request_token(self):
RuntimeError
If the token cannot be retrieved.
"""
url = f"http://{self.__location}/manager/api/get-token/"
url = f"{self.__location}/manager/api/get-token/"
data = {
"username": self.username,
"password": self.__password,
Expand All @@ -108,14 +111,15 @@ async def __request_token(self):
json_data = await resp.json()
token = json_data["token"]
self.websocket_url = (
f"ws://{self.__location}/manager/ws/subscription?token={token}"
f"ws://{self.__domain}/manager/ws/subscription?token={token}"
if not self.__secure
else f"wss://{self.__domain}/manager/ws/subscription?token={token}"
)
self.__api_headers = {
"Authorization": f"Token {token}",
"Accept": "application/json",
"Content-Type": "application/json",
}
self.command_url = f"http://{self.__location}/manager/api/cmd/"
except Exception as e:
raise RuntimeError("Authentication failed.") from e

Expand Down Expand Up @@ -195,6 +199,8 @@ async def send_sal_command(self, csc, salindex, cmd_name, params):
params: `dict`
Parameters of the command to be sent
"""

url = f"{self.__location}/manager/api/cmd/"
data = {
"csc": csc,
"salindex": salindex,
Expand All @@ -204,7 +210,7 @@ async def send_sal_command(self, csc, salindex, cmd_name, params):
async with aiohttp.ClientSession() as session:
try:
async with session.post(
self.command_url, data=json.dumps(data), headers=self.__api_headers
url, data=json.dumps(data), headers=self.__api_headers
) as resp:
json_data = await resp.json()
self.log.info("Command sent: ", json_data)
Expand Down
174 changes: 26 additions & 148 deletions python/lsst/ts/externalscripts/make_love_stress_tests.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,154 +21,12 @@
__all__ = ["StressLOVE"]

import asyncio
import json
import os

import aiohttp
import yaml
from lsst.ts import salobj, utils


class ManagerClient:
"""Connect to a LOVE-manager instance.
Parameters
----------
location : `str`
Host of the running LOVE-manager instance
username: `str`
LOVE username to use as authenticator
password: `str`
Password of the choosen LOVE user
event_streams: `dict`
Dictionary whith each item as <CSC:salindex>: <events_names_tuple>
e.g. {"ATDome:0": ('allAxesInPosition', 'authList',
'azimuthCommandedState', 'azimuthInPosition', ...)
telemetry_streams: `dict`
Dictionary whith each item as <CSC:salindex>: <telemetries_names_tuple>
e.g. {"ATDome:0": ('position', ...)
from lsst.ts import salobj

Notes
-----
**Details**
* Generate websocket connections using provided credentials
by token authentication and subscribe to every
event and telemetry specified.
"""

def __init__(self, location, username, password, event_streams, telemetry_streams):
self.username = username
self.event_streams = event_streams
self.telemetry_streams = telemetry_streams

self.websocket_url = None
self.num_received_messages = 0
self.msg_traces = []

self.__location = location
self.__password = password
self.__websocket = None

self.start_task = utils.make_done_future()

async def __request_token(self):
"""Authenticate on the LOVE-manager instance
to get an authorization token and set the
corresponding websocket_url.
Raises
------
RuntimeError
If the token cannot be retrieved.
"""

url = f"http://{self.__location}/manager/api/get-token/"
data = {
"username": self.username,
"password": self.__password,
}

async with aiohttp.ClientSession() as session:
try:
async with session.post(url, data=data) as resp:
json_data = await resp.json()
token = json_data["token"]
self.websocket_url = (
f"ws://{self.__location}/manager/ws/subscription?token={token}"
)
except Exception as e:
raise RuntimeError("Authentication failed.") from e

async def __handle_message_reception(self):
"""Handles the reception of messages."""

if self.__websocket:
async for message in self.__websocket:
if message.type == aiohttp.WSMsgType.TEXT:
msg = json.loads(message.data)
if "category" not in msg or (
"option" in msg and msg["option"] == "subscribe"
):
continue
self.num_received_messages = self.num_received_messages + 1
tracing = msg["tracing"]
tracing["client_rcv"] = utils.current_tai()
self.msg_traces.append(tracing)

async def __subscribe_to(self, csc, salindex, topic, topic_type):
"""Subscribes to the specified CSC stream in order to
receive LOVE-producer(s) data
Parameters
----------
csc : `str`
Name of the CSC stream
salindex: `int`
Salindex of the CSC stream
topic: `str`
Topic of the CSC stream
topic_type: `str`
Type of topic: `event` or `telemetry`
"""

subscribe_msg = {
"option": "subscribe",
"category": topic_type,
"csc": csc,
"salindex": salindex,
"stream": topic,
}
await self.__websocket.send_str(json.dumps(subscribe_msg))

async def start_ws_client(self):
"""Start client websocket connection"""

try:
await self.__request_token()
if self.websocket_url is not None:
async with aiohttp.ClientSession() as session:
self.__websocket = await session.ws_connect(self.websocket_url)
for name in self.event_streams:
csc, salindex = salobj.name_to_name_index(name)
for stream in self.event_streams[name]:
await self.__subscribe_to(csc, salindex, stream, "event")
for name in self.telemetry_streams:
csc, salindex = salobj.name_to_name_index(name)
for stream in self.telemetry_streams[name]:
await self.__subscribe_to(
csc, salindex, stream, "telemetry"
)
await self.__handle_message_reception()
except Exception as e:
raise RuntimeError("Manager Client connection failed.") from e

def create_start_task(self):
self.start_task = asyncio.create_task(self.start_ws_client())

async def close(self):
if self.__websocket:
await self.__websocket.close()
from .love_manager_client import LoveManagerClient


class StressLOVE(salobj.BaseScript):
Expand Down Expand Up @@ -220,7 +78,8 @@ def get_schema(cls):
type: object
properties:
location:
description: Host of the running LOVE instance (web server) to stress
description: Complete URL of the running LOVE instance (web server) to stress
e.g. https://base-lsp.lsst.codes/love or http://love01.ls.lsst.org
type: string
number_of_clients:
description: The number of clients to create
Expand Down Expand Up @@ -256,10 +115,15 @@ def set_metadata(self, metadata):
async def configure(self, config):
"""Configure the script.
Look for credentials configured with environment variables:
- USER_USERNAME
- USER_USER_PASS
These should match the credentials used to log into the LOVE instance.
Specify the Stress test configurations:
- LOVE host location
- Number of clients
- Number of messages to store
- Number of messages
- CSCs
Parameters
Expand All @@ -271,12 +135,20 @@ async def configure(self, config):
-----
Saves the results on several attributes:
* username : `str`, LOVE username to use as authenticator
* password : `str`, Password of the choosen LOVE user
* config : `types.SimpleNamespace`, same as config param
* remotes : a dict, with each item as
CSC_name[:index]: `lsst.ts.salobj.Remote`
Constructing a `salobj.Remote` is slow (DM-17904), so configuration
may take a 10s or 100s of seconds per CSC.
Raises
------
RuntimeError
If environment variables USER_USERNAME or
USER_USER_PASS are not defined.
"""
self.log.info("Configure started")

Expand All @@ -286,6 +158,10 @@ async def configure(self, config):
# get credentials
self.username = os.environ.get("USER_USERNAME")
self.password = os.environ.get("USER_USER_PASS")
if self.username is None:
raise RuntimeError(
"Configuration failed: environment variable USER_USERNAME not defined"
)
if self.password is None:
raise RuntimeError(
"Configuration failed: environment variable USER_USER_PASS not defined"
Expand Down Expand Up @@ -330,12 +206,13 @@ async def run(self):
f"Waiting for {self.config.number_of_clients} Manager Clients to be ready"
)
for i in range(self.config.number_of_clients):
client = ManagerClient(
client = LoveManagerClient(
self.config.location,
self.username,
self.password,
event_streams,
telemetry_streams,
msg_tracing=True,
)
self.clients.append(client)
client.create_start_task()
Expand All @@ -359,7 +236,8 @@ async def cleanup(self):

# Close all ManagerClients
for client in self.clients:
await client.close()
if client is not None:
await client.close()

def get_mean_latency(self):
"""Calculate the mean latency of all received messages."""
Expand Down
Loading

0 comments on commit 81e0ddc

Please sign in to comment.