Upgrade custom runners to be compatible with elasticsearch-py 8.2.0.

eventdata/runners/deleteindex_runner.py

@@ -67,7 +67,7 @@ def get_suffix(name, separator):
 
     if max_indices:
         response = await es.cat.indices(h="index")
-        indices = response.decode("utf-8").split("\n")
+        indices = response.split("\n")
         indices_by_suffix = {get_suffix(idx, suffix_separator): idx
             for idx in indices
             if fnmatch(idx, index_pattern) and
@@ -77,7 +77,7 @@ def get_suffix(name, separator):
         sorted_suffixes = sorted(list(indices_by_suffix.keys()))
         if len(sorted_suffixes) > max_indices:
             indices_to_delete = ",".join([indices_by_suffix[key] for key in sorted_suffixes[:(len(sorted_suffixes)-max_indices)]])
-            await es.indices.delete(indices_to_delete)
+            await es.indices.delete(index=indices_to_delete)
     else:
         await es.indices.delete(index=index_pattern)
 

eventdata/runners/indicesstats_runner.py

@@ -75,8 +75,8 @@ async def indicesstats(es, params):
                 response['total_segment_terms_memory_in_bytes'] = a['total']['segments']['terms_memory_in_bytes']
 
         if logger.isEnabledFor(logging.DEBUG):
-            logger.debug("Indices stats for {} => {}".format(index_pattern, json.dumps(result)))
-    except elasticsearch.TransportError as e:
+            logger.debug("Indices stats for {} => {}".format(index_pattern, result.body))
+    except (elasticsearch.ApiError, elasticsearch.TransportError) as e:
         logger.info("[indicesstats_runner] Error: {}".format(e))
 
     return response

eventdata/runners/mount_searchable_snapshot_runner.py

@@ -26,7 +26,7 @@ async def __call__(self, es, params):
         rename_pattern = params.get("rename_pattern", "(.*)")
         rename_replacement = params.get("rename_replacement", "\\1")
         query_params = params.get("query_params")
-        snapshots = await es.snapshot.get(repository_name, snapshot_name)
+        snapshots = await es.snapshot.get(repository=repository_name, snapshot=snapshot_name)
 
         # ES master
         if "responses" in snapshots:
@@ -43,7 +43,7 @@ async def __call__(self, es, params):
                         body={"index": index, "renamed_index": renamed_index}
 
                     await es.transport.perform_request(method="POST",
-                                                    url=f"/_snapshot/{repository_name}/{snapshot_name}/_mount",
+                                                    path=f"/_snapshot/{repository_name}/{snapshot_name}/_mount",
                                                     body=body,
                                                     params=query_params
                                                     )

eventdata/runners/nodestorage_runner.py

@@ -39,7 +39,7 @@ async def nodestorage(es, params):
     try:
         # get number of data nodes
         node_role_list = await es.cat.nodes(h="node.role")
-        node_role_list = node_role_list.decode("utf-8")
+        #node_role_list = node_role_list.decode("utf-8")
 
         data_node_count = 0
 
@@ -64,7 +64,7 @@ async def nodestorage(es, params):
         response['average_data_volume_per_node_bytes'] = volume_per_data_node
         response['average_data_volume_per_node_tb'] = volume_per_data_node_tb
 
-    except elasticsearch.TransportError as e:
+    except (elasticsearch.ApiError, elasticsearch.TransportError) as e:
         logger.info("[nodestorage_runner] Error: {}".format(e))
 
     return response