Update requirements.txt to use latest opni-proto library

aiops/opni-opensearch-update-service/opensearch-update-service/app/main.py

@@ -93,16 +93,16 @@ async def receive_template_data(queue):
     es = await setup_es_connection()
     while True:
         df = await queue.get()
-        await update_template_data(es, df, queue)
+        await update_template_data(es, df)
 
 
 async def receive_logs(queue):
     es = await setup_es_connection()
     while True:
         df = await queue.get()
-        await update_logs(es, df, queue)
+        await update_logs(es, df)
 
-async def update_template_data(es, df, queue):
+async def update_template_data(es, df):
     try:
         async for ok, result in async_streaming_bulk(
                 es,
@@ -119,9 +119,8 @@ async def update_template_data(es, df, queue):
             "Failed to index data. Re-adding to queue"
         )
         logging.error(exception)
-        await queue.put(df)
 
-async def update_logs(es, df, queue):
+async def update_logs(es, df):
     # This function will be updating Opensearch logs which were inferred on by the DRAIN model.
     model_keywords_dict = {"drain":  ["_id", "masked_log", "template_matched","template_cluster_id","inference_model", "anomaly_level"],
                           "opnilog":  ["_id", "masked_log", "anomaly_level", "template_matched","template_cluster_id","opnilog_confidence", "inference_model"]}
@@ -151,7 +150,6 @@ async def update_logs(es, df, queue):
                     "Failed to index data. Re-adding to queue"
                 )
                 logging.error(exception)
-                await queue.put(anomaly_level_df[model_keywords_dict[model_name]])
 
 
 async def init_nats():

aiops/requirements.txt

@@ -1,5 +1,5 @@
 elasticsearch[async]==7.13.0
 numpy==1.22.0
 opni-nats==0.0.0.4
-dev-proto==0.6.0.0
+opni-proto==0.6.1.0
 pandas==1.2.3