DPDV-4549: Do not crash when invalid query is used (#96)

* DPDV-4549: Do not crash when invalid query is used

* Added error handling also for long running queries

* Increase timeout to 180s

* Try to wait little bit longer

* Add unit test to test the function - search_error_exit

Author: martin-majlis-s1

TA_dataset/bin/dataset_api.py

@@ -5,7 +5,7 @@
 
 # adjust paths to make the Splunk app working
 import import_declare_test  # noqa: F401
-from dataset_common import normalize_time
+from dataset_common import logger, normalize_time
 
 # Dataset V2 API client (generated)
 from dataset_query_api_client import AuthenticatedClient
@@ -24,6 +24,13 @@
 )
 
 
+# APIException stores response payload received by API.
+# Payload is passed as is into search_error_exit.
+class APIException(Exception):
+    def __init__(self, payload):
+        self.payload = payload
+
+
 # TODO: Convert to the expected format
 # https://www.python-httpx.org/advanced/#http-proxying
 def convert_proxy(proxy):
@@ -39,7 +46,13 @@ def convert_proxy(proxy):
 
 # Executes Dataset LongRunningQuery for log events
 def ds_lrq_log_query(
-    base_url, api_key, start_time, end_time, filter_expr, limit, proxy
+    base_url,
+    api_key,
+    start_time,
+    end_time,
+    filter_expr,
+    limit,
+    proxy,
 ):
     client = AuthenticatedClient(
         base_url=base_url, token=api_key, proxy=convert_proxy(proxy)
@@ -69,7 +82,14 @@ def ds_lrq_power_query(base_url, api_key, start_time, end_time, query, proxy):
 
 # Executes Dataset LongRunningQuery to fetch facet values
 def ds_lrq_facet_values(
-    base_url, api_key, start_time, end_time, filter, name, max_values, proxy
+    base_url,
+    api_key,
+    start_time,
+    end_time,
+    filter,
+    name,
+    max_values,
+    proxy,
 ):
     client = AuthenticatedClient(
         base_url=base_url, token=api_key, proxy=convert_proxy(proxy)
@@ -87,29 +107,53 @@ def ds_lrq_facet_values(
 
 # Executes LRQ run loop of launch-ping-remove API requests until the query completes
 # with a result
+# Returns tuple - value, error message
 def ds_lrq_run_loop(
     client: AuthenticatedClient, body: PostQueriesLaunchQueryRequestBody
 ):
     body.query_priority = PostQueriesLaunchQueryRequestBodyQueryPriority.HIGH
     response = post_queries.sync_detailed(client=client, json_body=body)
+    logger().debug(response)
     result = response.parsed
-    forward_tag = response.headers["x-dataset-query-forward-tag"]
-    steps_done = result.steps_completed
-    steps_total = result.steps_total
-    query_id = result.id
-    while steps_done < steps_total:
-        response = get_queries.sync_detailed(
-            id=query_id,
-            query_type=body.query_type,
-            client=client,
-            last_step_seen=steps_done,
-            forward_tag=forward_tag,
-        )
-        result = response.parsed
+    if result:
+        forward_tag = response.headers["x-dataset-query-forward-tag"]
         steps_done = result.steps_completed
-    delete_queries.sync_detailed(id=query_id, client=client, forward_tag=forward_tag)
+        steps_total = result.steps_total
+        query_id = result.id
+        retry = 0
+        while steps_done < steps_total:
+            response = get_queries.sync_detailed(
+                id=query_id,
+                query_type=body.query_type,
+                client=client,
+                last_step_seen=steps_done,
+                forward_tag=forward_tag,
+            )
+            logger().debug(response)
+            result = response.parsed
+            if result:
+                steps_done = result.steps_completed
+                retry = 0
+            else:
+                # 2023-11-01: QA server is sometimes returns 500 Operation not
+                # permitted, after several batches have been received.
+                # Idea is to try few retries. However, based on my handful of examples
+                # when this happened, retries has never helped.
+                retry += 1
+                logger().warning("Retrying: {}; {}".format(retry, response))
+                if retry > 5:
+                    logger().error(response)
+                    raise APIException(json.loads(response.content))
+                else:
+                    time.sleep(retry)
+
+        delete_queries.sync_detailed(
+            id=query_id, client=client, forward_tag=forward_tag
+        )
+        return result
 
-    return result
+    logger().error(response)
+    raise APIException(json.loads(response.content))
 
 
 # Returns a valid PowerQuery incorporating provided filter, columns and limit

TA_dataset/bin/dataset_common.py

@@ -4,6 +4,8 @@
 import os.path as op
 import sys
 import time
+from logging import Logger
+from typing import Optional  # noqa: F401
 
 # adjust paths to make the Splunk app working
 import import_declare_test  # noqa: F401
@@ -12,6 +14,8 @@
 APP_NAME = __file__.split(op.sep)[-3]
 CONF_NAME = "ta_dataset"
 
+LOGGER = None  # type: Optional[Logger]
+
 
 # define DataSet API URL for all environments
 def get_url(base_url, ds_method):
@@ -29,12 +33,22 @@ def get_url(base_url, ds_method):
     return base_url + "/api/" + ds_api_endpoint
 
 
+def logger() -> Logger:
+    if LOGGER:
+        return LOGGER
+    return logging.getLogger()
+
+
 # returns logger that logs data into file
 # /opt/splunk/var/log/splunk/${APP_NAME}/${suffix}
-def get_logger(session_key, suffix: str):
+# you should call this function as soon as possible, to set up proper logging
+def get_logger(session_key, suffix: str) -> Logger:
     logger = log.Logs().get_logger("{}_{}".format(APP_NAME, suffix))
     log_level = get_log_level(session_key, logging)
     logger.setLevel(log_level)
+
+    global LOGGER
+    LOGGER = logger
     return logger
 
 

TA_dataset/bin/dataset_search_command.py

@@ -2,16 +2,17 @@
 # -*- coding: utf-8 -*-
 
 import json
-import logging
 import re
 import sys
 import time
+from typing import Any, Dict, Union
 
 # ignore flake8 rule unused import, see
 # https://splunk.github.io/addonfactory-ucc-generator/troubleshooting/#modulenotfounderror-no-module-named-library-name
 import import_declare_test  # noqa: F401
 import requests
 from dataset_api import (
+    APIException,
     build_payload,
     ds_build_pq,
     ds_lrq_facet_values,
@@ -26,6 +27,7 @@
     get_logger,
     get_proxy,
     get_url,
+    logger,
     relative_to_epoch,
 )
 
@@ -116,17 +118,19 @@ def get_search_arguments(self):
     )
 
 
-def search_error_exit(self, r_json):
+def search_error_exit(self, r_json: Union[Dict[str, Any], str]) -> None:
+    logger().error(r_json)
     if "message" in r_json:
-        logging.error(r_json["message"])
         if r_json["message"].startswith("Couldn't decode API token"):
             error_message = (  # make API error more user-friendly
                 "API token rejected, check add-on configuration"
             )
         else:
             error_message = r_json["message"]
+
+        if "code" in r_json:
+            error_message += " (" + r_json["code"] + ")"
     else:
-        logging.error(r_json)
         try:
             error_message = str(r_json)
         except Exception as e:
@@ -313,13 +317,13 @@ def generate(self):
                         limit=ds_maxcount,
                         proxy=proxy,
                     )
+
                     logger.debug("QUERY RESULT, result={}".format(result))
 
                     matches_list = result.data.matches  # List<LogEvent>
 
                     if len(matches_list) == 0:
                         logger.warning("DataSet response success, no matches returned")
-                        # logger.warning(r_json)
 
                     for event in matches_list:
                         ds_event = json.loads(
@@ -429,6 +433,9 @@ def generate(self):
                         "DataSetFunction=getResponse, elapsed={}".format(r.elapsed)
                     )
                     r_json = r.json()
+                    status = r_json.get("status", "error")
+                    if status != "success":
+                        search_error_exit(self, r_json)
                     if "results" in r_json:
                         bucket_time = get_bucket_increments(
                             ds_start, ds_end, ts_buckets
@@ -470,6 +477,8 @@ def generate(self):
                         )
                     )
                     GeneratingCommand.flush
+            except APIException as e:
+                search_error_exit(self, e.payload)
             except Exception as e:
                 search_error_exit(self, str(e))
 

TA_dataset/default/data/ui/views/sdl_by_example.xml

@@ -101,7 +101,7 @@
       <input type="time" token="myTime" searchWhenChanged="true">
         <label>Time</label>
         <default>
-          <earliest>-4h@m</earliest>
+          <earliest>-15m@m</earliest>
           <latest>now</latest>
         </default>
       </input>

e2e/alert_action.spec.ts

@@ -4,15 +4,15 @@ import {setTimeout} from 'timers/promises';
 
 test('Alert action - create and delete alert with results propagation to DataSet', async ({page}) => {
     // GIVEN
-    test.setTimeout(120000); //default 60s may time out since job are scheduled for every 60s
+    test.setTimeout(200000); //default 60s may time out since job are scheduled for every 60s
     const serverHost = 'dataset_addon_for_splunk_playwright_CI_CD_e2e_test_host';
     const alertName = 'splunk_addon_test_alert_'+ Math.random().toString(36).substring(2,7);
     await removeAlertIfExists(page, alertName);
     // WHEN create alert from query
     await searchDataSet(page, "| dataset");
     await saveAsAlertWithDataSetTrigger(page, alertName, serverHost);
     // AND wait for alert job to be triggered (cron job every 1 minute)
-    const waitTotalS = 60;
+    const waitTotalS = 120;
     const waitStepS = 5;
     for (let i = 0; i <= waitTotalS; i = i + waitStepS) {
         console.log(`Waiting for splunk Job (${i}/${waitTotalS})`)

requirements-dev.txt

@@ -2,3 +2,4 @@ flake8==6.0.0; python_version <= '3.8'
 pip==23.2.1; python_version <= '3.8'
 pre-commit==3.3.3; python_version <= '3.8'
 pytest==7.4.0; python_version <= '3.8'
+pytest-mock==3.12.0; python_version <= '3.8'

tests/test_dataset_search_command.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+import pytest
+from dataset_search_command import DataSetSearch, search_error_exit
+
+
+@pytest.mark.parametrize(
+    "payload,expected",
+    [
+        ({"message": "foo - A"}, "foo - A"),
+        ({"message": "foo - B", "code": "bar"}, "foo - B (bar)"),
+        (
+            {"message": "Couldn't decode API token"},
+            "API token rejected, check add-on configuration",
+        ),
+        ("foo - C", "foo - C"),
+    ],
+)
+def test_search_error_exit(mocker, payload, expected):
+    s = DataSetSearch()
+    m = mocker.patch.object(s, "error_exit", return_value=True, autospec=True)
+    search_error_exit(s, payload)
+    m.assert_called_with(error="ERROR", message=expected)