Дмитрий Шицков

Кажется, вы сильно непонимание как работают сети. Стоит прочитать цикл статей https://linkmeup.ru/blog/1188/

Пайплайн исполняется в контейнере, создаваемом из имейджа ubuntu:20.04 - он "чистый" при каждом запуске пайплайна и не сохраняет состояний между запусками, в нем нет никакого test_user.
По окончании работы пайплайна, контейнер и его содержимое будут удалены.

Варианты следующие - процесс работающий по расписанию выделить в отдельный сервис-воркер и запускать лишь в единственном экземпляре, возможно используя планировщик задач.
Ну или сделать так чтобы реплики знали друг о друге и единовременно регулярную задачу выполняла только одна, например используя redis в качестве распределенного лока, или raft, или consul, или zookeeper и т.п..

Из своего приложения развернутого в кубе подключаться на test-streaming:4222 или test-streaming.<namespace>:4222 если nats и ваш сервис в разных неймспейсах

Советую поступить как нравится.

Оркестрация кубом в пределах одного физического сервера скорее прихоть, чем необходимость. Можно поиграться разве что для знакомства с системой. И k8s скорее всего точно мимо - впустую потраченные ресурсы одного сервера. minikube, k3s и подобные будут лучшим выбором.
Для поставки в куб все равно придется организовать сборку и публикацию контейнеров. Для доставки и helm хватит.

Что бы посоветовал - baremetal с оркестрация через Ansible.
Как вариант - гипервизор: виртуалки или lxс; возможно Proxmox. Просто на случай необходимости в изоляции или разных конфигураций ОС.

Вот все решения которые пришли в голову от простого и правильного к нетипичному.

1. Не использовать новые потоки, потому что код и так асинхронный
   
2. Сделать функцию aaa синхронной - проблема с запуском нового потока уйдет
   
3. Если aaa невозможно изменить - нужно сделать обертку в виде синхронной функции (или декоратора), в который создать новый eventloop (предположим используется asyncio)
   

def wrapper(*args, **kwargs):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(aaa(*args, **kwargs))
    loop.close()

Нужно удвоить слеши, так как их экранирование требуется и в json.

>>> data = {"path": r"C:\Test\123"}
>>> data
{'path': 'C:\\Test\\123'}
>>> print(json.dumps(data))
{"path": "C:\\Test\\123"}

Ввиду отсутствия какой-либо синхронизации потоков, работающих с одним и тем же объектом, результат получается неопределен.
В представленном коде запускаются 2 потока подряд, без ожидания завершения каждого из них. А результат, вероятно, проверялся в главном потоке, пока оба потока ещё работали.

Придется сделать 2 базы - оригинальную и после изменений.

После этого есть варианты:
Можно воспользоваться mysqldbcompare в режиме --changes-for

Или попросту сделать два дампа и сравнить их

mysqldump --skip-comments --skip-extended-insert -u user database1 > file1.txt
mysqldump --skip-comments --skip-extended-insert -u user database2 > file2.txt
diff file1.txt file2.txt

Ну или включить Audit Log в mysql и посмотреть по нему какие изменения будут вноситься.

+ https://liquibase.org/

+ Здесь в самом вопросе перечислено пяток инструментов https://stackoverflow.com/q/225772

И кажется Mysql Workbench может сравнивать.

data = programs_dict
Или
data = subkey_dict
Ну или... Откуда нам знать чему ваша data равняться должна...

1. При использовании транзакции, в сентри попадет информация о времени исполнения части кода и операции от открытия этой транзакции до закрытия. Но операция может состоять из множества "тяжелых" секций кода, каждую из которых можно замерить отдельно в рамках транзакции, создав span. Т.е. транзакция может включать несколько span как дочерние операции для детализации происходящего процесса. К примеру, в транзакции "запрос на получение данных" длительностью 15 мс. может иметь span "запрос в БД" с длительностью 5 мс.
   
2. Воспользуйтесь первыми двумя примерами - создавайте транзакцию перед вызовом обработки каждого из запросов (1ый пример) и добавляйте к транзакции span'ы по необходимости(2ой пример). Если у вашего сервиса одна точка входа - grpc, то у вас всегда ситуация определена. Входящие запросы гарантировано создадут транзакцию, которую будет использовать весь остальной код.
   
3. Да, если если dsn указан корректно.
   

Не забудьте понизить traces_sample_rate для продакшена, чтобы это не сказалось на производительности сервиса.