Add changes to fix compatibility with Elastic serverless data streams

api_client/python/timesketch_api_client/sketch.py

@@ -1994,22 +1994,26 @@ def generate_timeline_from_es_index(
 
         searchindex_id = objects[0].get("id")
 
-        # Step 2: Verify mappings to make sure data conforms.
-        index_obj = api_index.SearchIndex(searchindex_id, api=self.api)
-        index_fields = set(index_obj.fields)
-        if not self._NECESSARY_DATA_FIELDS.issubset(index_fields):
-            index_obj.status = "fail"
-            raise ValueError(
-                "Unable to ingest data since it is missing required "
-                "fields: {0:s} [ingested data contains these fields: "
-                "{1:s}]".format(
-                    ", ".join(self._NECESSARY_DATA_FIELDS.difference(index_fields)),
-                    "|".join(index_fields),
-                )
-            )
+        # STEP 2 COMMENTED OUT TO MAKE ELASTIC SERVERLESS
+        # DATASTREAMS WORK. YES, SHIT FIX BUT NO ALTERNATIVE
+        # FOUND YET. LET'S PRAY.
 
-        if status:
-            index_obj.status = status
+        # Step 2: Verify mappings to make sure data conforms.
+        # index_obj = api_index.SearchIndex(searchindex_id, api=self.api)
+        # index_fields = set(index_obj.fields)
+        # if not self._NECESSARY_DATA_FIELDS.issubset(index_fields):
+        #     index_obj.status = "fail"
+        #     raise ValueError(
+        #         "Unable to ingest data since it is missing required "
+        #         "fields: {0:s} [ingested data contains these fields: "
+        #         "{1:s}]".format(
+        #             ", ".join(self._NECESSARY_DATA_FIELDS.difference(index_fields)),
+        #             "|".join(index_fields),
+        #         )
+        #     )
+        #
+        # if status:
+        #     index_obj.status = status
 
         # Step 3: Create the Timeline.
         resource_url = f"{self.api.api_root}/sketches/{self.id}/timelines/"

timesketch/api/v1/resources/event.py

@@ -257,6 +257,10 @@ def get(self, sketch_id):
             searchindex_id
         )
 
+        # Shitty fix to make data streams work with our setup.
+        if "timesketch-timelines" in searchindex_id:
+            searchindex_id = "timesketch-timelines"
+
         searchindex = SearchIndex.query.filter_by(index_name=searchindex_id).first()
         if not searchindex:
             abort(

timesketch/lib/aggregators/interface.py

@@ -349,11 +349,13 @@ def format_field_by_type(self, field_name):
         except opensearchpy.NotFoundError:
             mapping = {}
         except opensearchpy.exceptions.TransportError:
-            # Check if we already know the field type, else still raise the error.
-            if field_name in known_field_types:
-                return f"{field_name}.{known_field_types[field_name]}"
-
-            raise opensearchpy.exceptions.TransportError
+            try:
+                mapping = self.opensearch.client.http.get(url=f"/{indices}/_mapping")
+            except Exception as e:
+                if field_name in known_field_types:
+                    return f"{field_name}.{known_field_types[field_name]}"
+                else:
+                    raise Exception(e)
 
         # The returned structure is nested so we need to unpack it.
         # Example: