|
|
|
@ -33,7 +33,7 @@ mapping = {
|
|
|
|
|
"ts": {"type": "date", "format": "epoch_second"},
|
|
|
|
|
"match_ts": {"type": "date", "format": "iso8601"},
|
|
|
|
|
"file_tim": {"type": "date", "format": "epoch_millis"},
|
|
|
|
|
"rule_uuid": {"type": "keyword"},
|
|
|
|
|
"rule_id": {"type": "keyword"},
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
@ -43,7 +43,7 @@ for field in keyword_fields:
|
|
|
|
|
|
|
|
|
|
class ElasticsearchBackend(StorageBackend):
|
|
|
|
|
def __init__(self):
|
|
|
|
|
super().__init__("Elasticsearch")
|
|
|
|
|
super().__init__("elasticsearch")
|
|
|
|
|
self.client = None
|
|
|
|
|
self.async_client = None
|
|
|
|
|
|
|
|
|
@ -272,10 +272,12 @@ class ElasticsearchBackend(StorageBackend):
|
|
|
|
|
"""
|
|
|
|
|
if self.async_client is None:
|
|
|
|
|
await self.async_initialise()
|
|
|
|
|
print("MATCHES", matches)
|
|
|
|
|
for match in matches:
|
|
|
|
|
result = await self.async_client.index(
|
|
|
|
|
index=settings.INDEX_RULE_STORAGE, body=match
|
|
|
|
|
)
|
|
|
|
|
print("RESULT", result)
|
|
|
|
|
if not result["result"] == "created":
|
|
|
|
|
self.log.error(f"Indexing failed: {result}")
|
|
|
|
|
self.log.debug(f"Indexed {len(matches)} messages in ES")
|
|
|
|
@ -439,6 +441,7 @@ class ElasticsearchBackend(StorageBackend):
|
|
|
|
|
raise QueryError(error)
|
|
|
|
|
if len(response["hits"]["hits"]) == 0:
|
|
|
|
|
# No results, skip
|
|
|
|
|
result_map[index] = ({}, [])
|
|
|
|
|
continue
|
|
|
|
|
meta, response = self.parse(response, meta=True)
|
|
|
|
|
# print("Parsed response", response)
|
|
|
|
@ -500,7 +503,7 @@ class ElasticsearchBackend(StorageBackend):
|
|
|
|
|
|
|
|
|
|
if rule_object is not None:
|
|
|
|
|
index = settings.INDEX_RULE_STORAGE
|
|
|
|
|
add_bool.append({"rule_uuid": str(rule_object.id)})
|
|
|
|
|
add_bool.append({"rule_id": str(rule_object.id)})
|
|
|
|
|
else:
|
|
|
|
|
# I - Index
|
|
|
|
|
index = parse_index(request.user, query_params)
|
|
|
|
@ -520,14 +523,19 @@ class ElasticsearchBackend(StorageBackend):
|
|
|
|
|
if isinstance(sources, dict):
|
|
|
|
|
return sources
|
|
|
|
|
total_count = len(sources)
|
|
|
|
|
total_sources = len(settings.MAIN_SOURCES) + len(settings.SOURCES_RESTRICTED)
|
|
|
|
|
# Total is -1 due to the "all" source
|
|
|
|
|
total_sources = (
|
|
|
|
|
len(settings.MAIN_SOURCES) - 1 + len(settings.SOURCES_RESTRICTED)
|
|
|
|
|
)
|
|
|
|
|
print("total_count", total_count, "total_sources", total_sources)
|
|
|
|
|
if total_count != total_sources:
|
|
|
|
|
add_top_tmp = {"bool": {"should": []}}
|
|
|
|
|
for source_iter in sources:
|
|
|
|
|
add_top_tmp["bool"]["should"].append(
|
|
|
|
|
{"match_phrase": {"src": source_iter}}
|
|
|
|
|
)
|
|
|
|
|
add_top.append(add_top_tmp)
|
|
|
|
|
if rule_object is not None and query_params["source"] != "all":
|
|
|
|
|
add_top.append(add_top_tmp)
|
|
|
|
|
|
|
|
|
|
# R - Ranges
|
|
|
|
|
# date_query = False
|
|
|
|
@ -547,12 +555,17 @@ class ElasticsearchBackend(StorageBackend):
|
|
|
|
|
sort = parse_sort(query_params)
|
|
|
|
|
if isinstance(sort, dict):
|
|
|
|
|
return sort
|
|
|
|
|
|
|
|
|
|
if rule_object is not None:
|
|
|
|
|
field = "match_ts"
|
|
|
|
|
else:
|
|
|
|
|
field = "ts"
|
|
|
|
|
if sort:
|
|
|
|
|
# For Druid compatibility
|
|
|
|
|
sort_map = {"ascending": "asc", "descending": "desc"}
|
|
|
|
|
sorting = [
|
|
|
|
|
{
|
|
|
|
|
"ts": {
|
|
|
|
|
field: {
|
|
|
|
|
"order": sort_map[sort],
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
@ -594,6 +607,7 @@ class ElasticsearchBackend(StorageBackend):
|
|
|
|
|
search_query,
|
|
|
|
|
index=index,
|
|
|
|
|
)
|
|
|
|
|
print("query", search_query)
|
|
|
|
|
if "message" in response:
|
|
|
|
|
return response
|
|
|
|
|
|
|
|
|
|