Repository: getsentry/sentry
Pull request: test(taskbroker): Add CLI command for sending taskbroker tasks
### Overview
This PR is responsible for adding a CLI tool for sending tasks to the taskbroker to assist with iterative local development.


Refs: https://github.com/getsentry/sentry/issues/81240

### Testing
```
sentry run taskbroker-send-tasks --task-function-path sentry.taskworker.tasks.example.say_hello --args '["foobar"]' --repeat 2
```

Problem statement:
Add command line tool for creating tasks
Would be great for local development if we could have a way to trigger a single task

```
sentry send_task namespace:task_id --args '["one", 2]' --kwargs '{"k": "v"}' --repeat 100
```

this would let folks do iterative development locally more easily as they can generate tasks, or script workloads for load-testing.

Hints from issue discussion:

Commit message: use args and kwargs

Patch to review:
diff --git a/src/sentry/runner/commands/run.py b/src/sentry/runner/commands/run.py
index 1c17cbf8d85712..c9a29ff3e5ce77 100644
--- a/src/sentry/runner/commands/run.py
+++ b/src/sentry/runner/commands/run.py
@@ -253,6 +253,61 @@ def taskworker(rpc_host: str, max_task_count: int, **options: Any) -> None:
         raise SystemExit(exitcode)
 
 
+@run.command()
+@log_options()
+@configuration
+@click.option(
+    "--repeat",
+    type=int,
+    help="Number of messages to send to the kafka topic",
+    default=100,
+    show_default=True,
+)
+@click.option(
+    "--kwargs",
+    type=str,
+    help="Task function keyword arguments",
+)
+@click.option(
+    "--args",
+    type=str,
+    help="Task function arguments",
+)
+@click.option(
+    "--task-function",
+    type=str,
+    help="The function name of the task to execute located in the module",
+    required=True,
+)
+@click.option(
+    "--path",
+    type=str,
+    help="The path to the task module",
+    required=True,
+)
+def taskbroker_send_tasks(
+    path: str,
+    task_function: str,
+    args: str,
+    kwargs: str,
+    repeat: int,
+) -> None:
+    import importlib
+
+    try:
+        module = importlib.import_module(path)
+        func = getattr(module, task_function)
+    except Exception as e:
+        click.echo(f"Error: {e}")
+        raise click.Abort()
+
+    for _ in range(repeat):
+        task_args = [] if not args else eval(args)
+        task_kwargs = {} if not kwargs else eval(kwargs)
+        func.delay(*task_args, **task_kwargs)
+    click.echo(message=f"Successfully sent {repeat} messages.")
+
+
 @run.command()
 @click.option(
     "--pidfile",
diff --git a/src/sentry/taskdemo/__init__.py b/src/sentry/taskdemo/__init__.py
new file mode 100644
index 00000000000000..39b932eb7e20e2
--- /dev/null
+++ b/src/sentry/taskdemo/__init__.py
@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+import logging
+
+from sentry.taskworker.registry import taskregistry
+
+logger = logging.getLogger(__name__)
+demotasks = taskregistry.create_namespace(name="demos")
+
+
+@demotasks.register(name="demos.say_hello")
+def say_hello(name):
+    print(f"{name}")  # noqa


Code review comments:
diff_hunk: @@ -0,0 +1,13 @@
+from __future__ import annotations
+
+import logging
+
+from sentry.taskworker.registry import taskregistry
+
+logger = logging.getLogger(__name__)
+demotasks = taskregistry.create_namespace(name="demos")
+
+
+@demotasks.register(name="demos.say_hello")
+def say_hello(name):
path: src/sentry/taskdemo/__init__.py
text: @user1:
We could put these demo functions in `sentry.taskworker.tasks.examples` 🤷 

@author:
Good idea

diff_hunk: @@ -253,6 +253,61 @@ def taskworker(rpc_host: str, max_task_count: int, **options: Any) -> None:
         raise SystemExit(exitcode)
 
 
+@run.command()
+@log_options()
+@configuration
+@click.option(
+    "--repeat",
+    type=int,
+    help="Number of messages to send to the kafka topic",
+    default=100,
+    show_default=True,
+)
+@click.option(
+    "--kwargs",
+    type=str,
+    help="Task function keyword arguments",
+)
+@click.option(
+    "--args",
+    type=str,
+    help="Task function arguments",
+)
+@click.option(
+    "--task-function",
+    type=str,
+    help="The function name of the task to execute located in the module",
+    required=True,
+)
+@click.option(
+    "--path",
+    type=str,
+    help="The path to the task module",
+    required=True,
+)
+def taskbroker_send_tasks(
+    path: str,
+    task_function: str,
+    args: str,
+    kwargs: str,
+    repeat: int,
+) -> None:
+    import importlib
+
+    try:
+        module = importlib.import_module(path)
+        func = getattr(module, task_function)
+    except Exception as e:
+        click.echo(f"Error: {e}")
+        raise click.Abort()
+
+    for _ in range(repeat):
+        task_args = [] if not args else eval(args)
+        task_kwargs = {} if not kwargs else eval(kwargs)
path: src/sentry/runner/commands/run.py
text: @user1:
If the args/kwargs aren't changing per call we could do them once.

@author:
Oops, yes

diff_hunk: @@ -253,6 +253,61 @@ def taskworker(rpc_host: str, max_task_count: int, **options: Any) -> None:
         raise SystemExit(exitcode)
 
 
+@run.command()
+@log_options()
+@configuration
+@click.option(
+    "--repeat",
+    type=int,
+    help="Number of messages to send to the kafka topic",
+    default=100,
path: src/sentry/runner/commands/run.py
text: @user1:
```suggestion
    default=1,
```

I think we should have the default start off small. I think it is more likely that someone will want to do local testing one task at a time instead of getting 100.

@author:
Makes sense, updated 👍