Add load tests for conversational analytics

[bwebs]

This PR adds comprehensive load testing capabilities for Looker's Conversational Analytics feature. It includes two new commands:

1. cookieless-embed-conversational-analytics: Tests the UI experience via cookieless embedding, including the initial handshake and iframe loading.
2. conversational-analytics-api: Tests the underlying API by simulating users sending messages to conversational agents. It supports using existing agents or creating new ones on the fly with custom prompts and explores.

Key changes:

• Created lkr/load_test/locustfile_conversational_analytics_api.py
• Created lkr/load_test/locustfile_cookieless_embed_conversational_analytics.py
• Added support files in lkr/load_test/embed_cookieless_conversational_analytics/
• Updated lkr/main.py to expose the new load test commands.

---

PR created automatically by Jules for task 12040390812617948698 started by @bwebs

[google-labs-jules]

👋 Jules, reporting for duty! I'm here to lend a hand with this pull request.

When you start a review, I'll add a 👀 emoji to each comment to let you know I've read it. I'll focus on feedback directed at me and will do my best to stay out of conversations between you and other bots or reviewers to keep the noise down.

I'll push a commit with your requested changes shortly after. Please note there might be a delay between these steps, but rest assured I'm on the job!

For more direct control, you can switch me to Reactive Mode. When this mode is on, I will only act on comments where you specifically mention me with @jules. You can find this option in the Pull Request section of your global Jules UI settings. You can always switch back!

New to Jules? Learn more at jules.google/docs.

---

For security, I will only act on instructions from the user who triggered this task.

[gemini-code-assist]

Code Review

This pull request introduces load testing for Looker's Conversational Analytics, adding a mock cookieless embed server, a Selenium-based Locust user, and an API-based Locust user, along with integrating these into the main CLI. The review feedback highlights several important areas for improvement, including avoiding Looker SDK re-initialization on every request to prevent performance bottlenecks, cleaning up created agents and conversations to avoid database leaks, handling missing User-Agent headers to prevent TypeErrors, preventing background process leaks upon driver initialization failures, and using monotonic timers for accurate benchmarking.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Initializing the Looker SDK (looker_sdk.init40()) inside the __init__ method of CookielessEmbedHandler means a new SDK instance is created for every single HTTP request (since HTTPServer instantiates the handler class per request). This is extremely expensive and will severely limit the throughput of the embed server during load tests.

Instead, initialize the SDK once (e.g., as a class-level attribute) and reuse it across requests.

Suggested change

	self.sdk = looker_sdk.init40()
	self.debug = debug
	if not hasattr(CookielessEmbedHandler, "_shared_sdk"):
	CookielessEmbedHandler._shared_sdk = looker_sdk.init40()
	self.sdk = CookielessEmbedHandler._shared_sdk

[gemini-code-assist]

If any exception occurs during the initialization of the Chrome driver or while waiting for the server to start, the constructor will raise an exception and exit. Because the object construction fails, Locust will not call on_stop(), resulting in the self.server_process background process being leaked as an orphan/zombie process.

Wrap the post-spawn initialization in a try...except block to ensure the server process is terminated and cleaned up if initialization fails.

        try:
            # Wait for the server to be ready with 10 second timeout
            is_server_ready = False
            for _ in range(20):
                try:
                    with socket.create_connection(("127.0.0.1", self.port), timeout=0.5):
                        is_server_ready = True
                        break
                except (socket.timeout, ConnectionRefusedError):
                    time.sleep(0.5)

            if not is_server_ready:
                raise Exception("Embed server failed to start")

            chrome_options = Options()
            chrome_options.add_argument("--headless=new")
            chrome_options.add_argument("--no-sandbox")
            chrome_options.add_argument("--disable-dev-shm-usage")
            chrome_options.add_argument("--disable-gpu")
            chrome_options.add_argument("--user-agent=Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36")
            chrome_options.add_argument("--enable-logging")
            chrome_options.add_argument("--v=1")

            # Speed up page loading by not waiting for non-critical subresources (CSS, images, fonts)
            chrome_options.page_load_strategy = "eager"

            # In VPCSC, block everything except required hosts to trigger immediate failure
            # instead of waiting for a 60-second network timeout.
            looker_url = os.environ.get("LOOKERSDK_BASE_URL", "")
            looker_host = urlparse(looker_url).hostname

            rules = "MAP * ~NOTFOUND, EXCLUDE localhost, EXCLUDE 127.0.0.1"
            if looker_host:
                rules += f", EXCLUDE {looker_host}"

            chrome_options.add_argument(f"--host-resolver-rules={rules}")

            chrome_options.add_experimental_option(
                "prefs",
                {
                    "profile.cookie_controls_mode": 1, # 1 = Block third-party cookies
                },
            )
            if self.debug:
                chrome_options.set_capability("goog:loggingPrefs", {"browser": "ALL"})
            self.driver = webdriver.Chrome(options=chrome_options)
        except Exception as e:
            self.server_process.terminate()
            self.server_process.wait()
            raise e

[gemini-code-assist]

If the User-Agent header is missing from the incoming request, self.headers.get('User-Agent') will return None. Passing None as a header value in transport_options can cause a TypeError in some HTTP clients or underlying libraries.

It is safer to defensively check if the User-Agent is present or default to an empty string.

Suggested change

	transport_options={'headers':{'User-Agent': self.headers.get('User-Agent')}}
	transport_options={'headers':{'User-Agent': self.headers.get('User-Agent') or ''}}

[gemini-code-assist]

If the User-Agent header is missing from the incoming request, self.headers.get('User-Agent') will return None. Passing None as a header value in transport_options can cause a TypeError in some HTTP clients or underlying libraries.

It is safer to defensively check if the User-Agent is present or default to an empty string.

Suggested change

	transport_options={'headers':{'User-Agent': self.headers.get('User-Agent')}}
	transport_options={'headers':{'User-Agent': self.headers.get('User-Agent') or ''}}

[gemini-code-assist]

The load test creates agents and conversations on the fly but never cleans them up when the test stops. Over a long load test run, this will leak orphaned agents and conversations in the Looker database, which can degrade database performance and skew the test results.

Implement an on_stop method to delete any agents or conversations created during the test.

    def on_start(self):
        self.sdk = self._init_sdk()
        self._created_agent_id = None
        initial_agent_id = self.agent_id
        self.agent_id = self._get_or_create_agent(self.sdk)
        if not initial_agent_id:
            self._created_agent_id = self.agent_id
        if self.continue_conversation:
            self.conversation_id = self._create_conversation(self.sdk, self.agent_id)

    def on_stop(self):
        if self.sdk:
            if self.conversation_id:
                try:
                    self.sdk.delete_conversation(conversation_id=self.conversation_id)
                except Exception as e:
                    logger.error("failed_to_delete_conversation", error=str(e), conversation_id=self.conversation_id)
            if getattr(self, "_created_agent_id", None):
                try:
                    self.sdk.delete_agent(agent_id=self._created_agent_id)
                except Exception as e:
                    logger.error("failed_to_delete_agent", error=str(e), agent_id=self._created_agent_id)

[gemini-code-assist]

Using datetime.datetime.now() to measure durations is prone to inaccuracies if the system clock is adjusted (e.g., via NTP synchronization) during the request.

For precise and robust duration measurements, use time.perf_counter(), which is monotonic and specifically designed for benchmarking and profiling.

Suggested change

	start_time = datetime.datetime.now()
	try:
	self.sdk.conversational_analytics_chat(
	body=models40.ConversationalAnalyticsChatRequest(
	conversation_id=cid,
	user_message=question
	)
	)
	end_time = datetime.datetime.now()
	duration = (end_time - start_time).total_seconds()
	import time
	start_time = time.perf_counter()
	try:
	self.sdk.conversational_analytics_chat(
	body=models40.ConversationalAnalyticsChatRequest(
	conversation_id=cid,
	user_message=question
	)
	)
	duration = time.perf_counter() - start_time

[gemini-code-assist]

The load test creates conversations on the fly but never cleans them up when continue_conversation is False. Over a long load test run, this will leak thousands of orphaned conversations in the Looker database, which can degrade database performance and skew the test results.

Implement immediate cleanup of the conversation if continue_conversation is False.

Suggested change

	if not self.continue_conversation:
	# If not continuing, we don't save the conversation_id for the next task
	# (In standard Looker, we might want to delete it too, but maybe not for load test)
	pass
	if not self.continue_conversation:
	try:
	self.sdk.delete_conversation(conversation_id=cid)
	except Exception as e:
	logger.error("failed_to_delete_conversation", error=str(e), conversation_id=cid)

[gemini-code-assist]

If self.driver.get_log('browser') raises an exception (e.g., if the driver is not fully initialized or doesn't support logging), the exception will propagate out of the finally block and mask any original exception raised in the try block.

Wrap the log retrieval in a nested try...except block to prevent masking other errors.

        finally:
            try:
                for entry in self.driver.get_log('browser'):
                    print(entry)
            except Exception as log_err:
                print(f"Notice: Failed to retrieve browser logs: {log_err}")