Merge pull request #1208 from splunk/ssa_etet_ig

fixed index search

bin/ssa-end-to-end-testing/tests/modules/spl/detection2.spl

@@ -17,5 +17,4 @@
 | eval start_time = timestamp,
 end_time = timestamp,
 entities = mvappend(dest_device_id, dest_user_id),
-body = "TBD"
-| into index("mc", "main");
+body = "TBD";

bin/ssa-end-to-end-testing/tests/modules/spl/firehose.spl

@@ -1 +1 @@
-| from read_splunk_firehose();
+| from read_splunk_firehose();

bin/ssa-end-to-end-testing/tests/modules/spl/firehose2.spl

@@ -1,2 +1 @@
-| from read_splunk_firehose()
-| into index("mc", "main");
+| from read_splunk_firehose();

bin/ssa-end-to-end-testing/tests/modules/spl/troubleshoot.spl

@@ -0,0 +1 @@
+| from read_ssa_enriched_events();

bin/ssa-end-to-end-testing/tests/modules/streams_service_api_helper.py

@@ -14,9 +14,11 @@ from http import HTTPStatus
 from modules.utils import request_headers
 
 
-DSP_URL = f"https://api.playground.scp.splunk.com/research2/"
+TENANT = "research2"
+DSP_URL = f"https://api.playground.scp.splunk.com/{TENANT}/"
 STREAMS_ENDPOINT = f"{DSP_URL}streams/v3beta1/"
 SEARCH_ENDPOINT = f"{DSP_URL}search/v2beta1/"
+CATALOG_ENDPOINT = f"{DSP_URL}catalog/v2beta1/"
 
 # Streaming Pipelines REST endpoints
 CONNECTIONS_ENDPOINT = f"{STREAMS_ENDPOINT}connections"
@@ -28,6 +30,7 @@ PREVIEW_SESSION_ENDPOINT = f"{STREAMS_ENDPOINT}preview-session"
 PREVIEW_DATA_ENDPOINT = f"{STREAMS_ENDPOINT}preview-data"
 INGEST_ENDPOINT = f"{DSP_URL}ingest/v1beta2/events"
 SUBMIT_SEARCH_ENDPOINT = f"{SEARCH_ENDPOINT}jobs"
+DATASETS_ENDPOINT = f"{CATALOG_ENDPOINT}datasets"
 
 # Logger
 logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
@@ -498,7 +501,7 @@ def ingest_data(header_token, data):
     return response.json()
 
 
-def submit_search_job(header_token, query):
+def submit_search_job(header_token, module, query):
     """
     Submit Search job
 
@@ -513,7 +516,8 @@ def submit_search_job(header_token, query):
         sid to search job
     """
     data = {
-        "query": query
+        "query": query,
+        "module": module
     }
     LOGGER.debug(f"Submit Search Job")
     response = requests.post(SUBMIT_SEARCH_ENDPOINT, json=data, headers=request_headers(header_token))
@@ -540,3 +544,43 @@ def get_search_job_results(header_token, sid):
     response = requests.get(results_search_job_endpoint, headers=request_headers(header_token))
     response_body = response.json()
     return response_body
+
+
+def create_temp_index(header_token, module):
+    """
+    Creates an index under module
+
+    Parameters
+    @param header_token: str
+         IAC token for DSP playground environment
+    @param module: str
+        module under this index will be created
+    @return:
+        index object dictionary
+    """
+    index_name = f"temp_st_{uuid.uuid1()}".replace("-", "_")
+    data = {
+        "module": module,
+        "name": index_name,
+        "kind": "index",
+        "disabled": False
+    }
+    response = requests.post(DATASETS_ENDPOINT, headers=request_headers(header_token), json=data)
+    return response.json()
+
+
+def delete_temp_index(header_token, index_id):
+    """
+    Deletes an index
+
+    @param header_token: str
+         IAC token for DSP playground environment
+    @param index_id:
+        Index ID
+    @return:
+        response status code from API
+    """
+    delete_url = f"{DATASETS_ENDPOINT}/{index_id}"
+    response = requests.delete(delete_url, headers=request_headers(header_token))
+    return response.status_code
+

bin/ssa-end-to-end-testing/tests/modules/utils.py

@@ -66,10 +66,16 @@ def request_headers(header_token):
     return headers
 
 
-def read_spl(file_name):
+def read_spl(file_name, results_index=None):
     file_path = os.path.join(os.path.dirname(__file__), 'spl', file_name)
     spl = open(file_path, "r").read()
     spl = replace_ssa_macros(spl)
+    if results_index is not None:
+        # When an index is defined for a test, it writes the output of this pipeline to this index.
+        # original_pipeline; => original_pipeline | into index("module", "index");
+        module = results_index["module"]
+        index = results_index["name"]
+        spl = spl[:spl.rindex(";")] + f"| into index(\"{module}\", \"{index}\");"
     return spl
 
 def replace_ssa_macros(spl):

bin/ssa-end-to-end-testing/tests/test_ssa_detections_playground.py

@@ -5,7 +5,7 @@ import os
 import time
 
 from http import HTTPStatus
-from modules.streams_service_api_helper import compile_spl, create_pipeline_from_spl, pipeline_status, activate_pipeline, ingest_data, get_preview_id_from_spl, get_preview_data, submit_search_job, get_search_job_results, stop_preview_session, deactivate_pipeline, delete_pipeline
+from modules.streams_service_api_helper import compile_spl, create_pipeline_from_spl, pipeline_status, activate_pipeline, ingest_data, get_preview_id_from_spl, get_preview_data, submit_search_job, get_search_job_results, stop_preview_session, deactivate_pipeline, delete_pipeline, create_temp_index, delete_temp_index
 from modules.utils import read_spl, read_data
 import pytest_check as check
 
@@ -14,6 +14,26 @@ logging.basicConfig(level=os.environ.get("LOGLEVEL", "INFO"))
 LOGGER = logging.getLogger(__name__)
 
 
+@pytest.fixture
+def results_index(token):
+    """
+    Fixture that creates an temporary index, and tears it down.
+
+    @todo
+        Implement a safe tear down version of this. When an exception happens in the test, we can't guarantee
+        that we won't have orphan indexes.
+    @param token:
+        This parameter is passed by `pytest`
+    @return:
+        Returns a descriptor of the index as a dictionary
+    """
+    header_token = f"Bearer {token}"
+    temp_index = create_temp_index(header_token, module="mc")
+    yield temp_index
+    # tear down the index
+    delete_temp_index(header_token, temp_index["id"])
+
+
 def test_data_ingestion_preview(token):
     assert (token is not None), "scloud token is missing"
     header_token = f"Bearer {token}"
@@ -21,11 +41,11 @@ def test_data_ingestion_preview(token):
     ssa_detection_in_dsp_with_preview_session(header_token, 'firehose.spl')
 
 
-def test_data_ingestion(token):
+def test_data_ingestion_index(token, results_index):
     assert (token is not None), "scloud token is missing"
     header_token = f"Bearer {token}"
 
-    ssa_detection_in_dsp(header_token, 'firehose2.spl')
+    ssa_detection_in_dsp(header_token, 'firehose2.spl', results_index)
 
 
 def test_ssa_example_detection_preview(token):
@@ -35,11 +55,11 @@ def test_ssa_example_detection_preview(token):
     ssa_detection_in_dsp_with_preview_session(header_token, 'detection.spl')
 
 
-def test_ssa_example_detection(token):
+def test_ssa_example_detection(token, results_index):
     assert (token is not None), "scloud token is missing"
     header_token = f"Bearer {token}"
 
-    ssa_detection_in_dsp(header_token, 'detection2.spl')
+    ssa_detection_in_dsp(header_token, 'detection2.spl', results_index)
 
 
 ## Helper Functions ##
@@ -52,7 +72,7 @@ def ssa_detection_in_dsp_with_preview_session(header_token, spl):
     preview_id = get_preview_id_from_spl(header_token, spl)
     assert preview_id is not None
 
-    time.sleep(120)
+    time.sleep(30)
 
     data = read_data(f"example.txt")
     response_body = ingest_data(header_token, data)
@@ -63,8 +83,8 @@ def ssa_detection_in_dsp_with_preview_session(header_token, spl):
     response = stop_preview_session(header_token, preview_id)
 
 
-def ssa_detection_in_dsp(header_token, spl):
-    spl = read_spl(spl)
+def ssa_detection_in_dsp(header_token, spl, results_index):
+    spl = read_spl(spl, results_index)
     assert (spl is not None), "fail to read dummy spl file"
 
     pipeline_id = create_pipeline_from_spl(header_token, spl)
@@ -76,14 +96,18 @@ def ssa_detection_in_dsp(header_token, spl):
     response_body = activate_pipeline(header_token, pipeline_id)
     assert response_body.get("activated") == pipeline_id, f"pipeline {pipeline_id} should be successfully activate."
 
-    time.sleep(120)
+    time.sleep(30)
 
     data = read_data(f"example.txt")
     response_body = ingest_data(header_token, data)
 
-    sid = submit_search_job(header_token, "from index:main")
+    time.sleep(30)
+
+    sid = submit_search_job(header_token, results_index['module'], f"from index:{results_index['name']} | search source!=\"Search Catalog\"")
     assert sid is not None
 
+    time.sleep(30)
+
     response_body = get_search_job_results(header_token, sid)
     check.greater(len(response_body.get("results")), 0, "Search job didn't return any results")