Does add_listener work?

[komikoni]

@events.init.add_listener
def on_locust_init(environment, **_kwargs):

invokust is a breakthrough. But, I couldn't get add_listner to work.

[FutureSharks]

I'm not sure what you mean @komikoni. Can you write your give your full locustfile and what you are trying to achieve?

[komikoni]

Thank you for your reply. (I'm sorry. It was a strange English sentence.)

I want to save the performance measurement results in a database And I want to see the results graphically at any time. And I want to run them serverless.

As a result of searching, with invokest locust-influxdb-listener was found. Later, I also found the Amazon Managed Service for Grafana (Preview).

However, influxdb has a Cloud service, but the version is different and it does not work, so I rewrote it for Amazon Timestream I made a locust-timestream-listener.

But it didn't work with invokest.(It worked with normal locust) I thought that the reason was that add_listener did not work in invokest, so I asked the question.

sample locustfile

from locust import between, events, tag, task, HttpUser

import sys,os
sys.path.append('##localpath##/locust-timestream-listener/') #TODO pip regist
from locust_timestream_listener import TimestreamListener, TimestreamSettings


@events.init.add_listener
def on_locust_init(environment, **_kwargs):
    print('on_locust_init')
    """
    Hook event that enables starting an timestream connection
    """
    # this settings matches the given docker-compose file
    timestreamSettings = TimestreamSettings(
        database = 'locust-result-1'
    )
    # start listerner with the given configuration
    TimestreamListener(env=environment, timestreamSettings=timestreamSettings)

class TestWebUser(HttpUser):

    wait_time = between(1,5)
         
    @tag('home_page')
    @task(1)
    def home_page(self):
        with self.client.get("/", catch_response=True) as response:
            if response.status_code != 200:
                response.failure("Got wrong response")
    
    @tag('connectors')
    @task(1)
    def workfront_connector(self):
        with self.client.get("/connectors/workfront", catch_response=True) as response:
            if 'Do More Work, Faster' not in response.text:
                response.failure("Expected test was not displayed")
            
    
    def on_start(self):
        print('New user was spawned')

[FutureSharks]

I made a locust-timestream-listener.

Nice 👍

2 things to try:

1. If you replace everything in on_locust_init with a print statement, do you see it in the logs?
2. Does it work when you run this locally in python, like shown in the readme?