mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-05 00:09:35 +00:00
# Nested Functions and Joins - User Guide ## Overview This guide covers two powerful features that enable sophisticated data analysis in OpenObserve: - **Nested Functions**: Combine multiple functions together to create complex calculations - **Joins**: Merge data from multiple streams (data sources) based on common fields These features work in the **Auto SQL** mode of dashboard panels, allowing you to write advanced queries without deep SQL knowledge. --- ## What's New ### Why These Features? **Before:** - Limited function composition - you could only use basic, single-level functions - No way to combine data from multiple streams in one query - Complex analytics required exporting data and manual processing **Now:** - Compose unlimited nested functions with full DataFusion support - Join multiple streams directly in the dashboard panel editor - Create sophisticated analytics queries with a visual interface ### Key Benefits ✓ Calculate complex metrics (error rates, percentiles, conditional aggregations) ✓ Build advanced dashboards without custom code ✓ Support for all DataFusion aggregate and scalar functions --- ## Nested Functions ### What Are Nested Functions? Nested functions let you combine multiple functions together - using the output of one function as input to another. Think of it like a cooking recipe where each step builds on the previous one. **Simple Example:** - Basic: `AVG(response_time)` - calculates average response time - Nested: `ROUND(AVG(response_time), 2)` - calculates average AND rounds it to 2 decimal places ### Supported Functions OpenObserve supports **all DataFusion functions**, giving you powerful analytics capabilities: **Aggregate Functions** (work across multiple rows): - `count`, `sum`, `avg`, `min`, `max` - basic aggregations - `median`, `stddev`, `variance` - statistical functions - `percentile_cont` - calculate percentiles (e.g., p95, p99) - `string_agg` - combine text values **Mathematical Functions**: - `round`, `ceil`, `floor` - rounding operations - `abs`, `power`, `sqrt` - mathematical operations - `log`, `log10`, `log2` - logarithmic functions - `sin`, `cos`, `tan` - trigonometric functions **String Functions**: - `concat`, `upper`, `lower`, `trim` - `substring`, `length`, `replace` **Conditional Functions**: - `CASE WHEN ... THEN ... ELSE ... END` - conditional logic - `COALESCE` - return first non-null value 📖 **Full Reference**: [DataFusion Aggregate Functions](https://datafusion.apache.org/user-guide/sql/aggregate_functions.html) | [DataFusion Scalar Functions](https://datafusion.apache.org/user-guide/sql/scalar_functions.html) --- ### Using Nested Functions (Step-by-Step) #### Step 1: Create or Edit a Dashboard Panel Open your dashboard and create a new panel or edit an existing one. Make sure you're in **Auto SQL** mode (not Custom SQL). #### Step 2: Select a Function In the Y-Axis or column configuration, click **Select Function** to choose your base function. <img width="795" alt="Nested Function - Adding Layers" src="https://github.com/user-attachments/assets/0b6b134e-59b8-48b1-a4ae-52fb4800d772" /> In this example, we're using the `Concat` function to combine multiple fields (namespace and pod name). #### Step 3: Add Nested Layers Click the function icon to add another function layer. You can nest functions multiple times. <img width="793" alt="Nested Function - Select Base Function" src="https://github.com/user-attachments/assets/4bd9e03e-1104-4e82-b2ec-a34e6bb64870" /> Here we're building a nested function: - Inner layer: `Avg(severity)` - calculates average severity - Outer layer: `Ceil()` - rounds up the average to the nearest integer #### Step 4: Build Complex Expressions For advanced analytics, you can combine nested functions with conditional logic using `CASE WHEN` statements. <img width="1340" alt="Complex Nested Function with CASE WHEN" src="https://github.com/user-attachments/assets/4c3e3623-bfad-44c4-a760-e10cc94d4a95" /> This example shows a sophisticated calculation: ```sql ROUND(100.0 * SUM(CASE WHEN COALESCE(severity, 0) >= 4 THEN 1 ELSE 0 END) / COUNT(_timestamp), 2) ``` **What this does:** 1. `COALESCE(severity, 0)` - replaces null severity with 0 2. `CASE WHEN severity >= 4 THEN 1 ELSE 0` - returns 1 for high severity, 0 otherwise 3. `SUM(...)` - counts high severity events 4. Divide by `COUNT(_timestamp)` - calculates percentage 5. Multiply by `100.0` - converts to percentage 6. `ROUND(..., 2)` - rounds to 2 decimal places **Result**: Percentage of high-severity events, rounded to 2 decimals #### Step 5: View the Generated SQL Switch to the **Raw** tab to see the complete SQL query that was generated from your nested function configuration. <img width="1388" alt="View Generated SQL" src="https://github.com/user-attachments/assets/d7f6111a-26c6-4dd0-9e9f-e26092fe6368" /> You can review the auto-generated SQL in the Query section at the bottom of the panel editor. The chart visualization updates in real-time as you build your nested functions. --- ## Joins ### What Are Joins? Joins allow you to combine data from two or more different streams (data sources) based on common fields between them. **Example:** **employee_data_join** stream: ```json [ {"employee_id": 1, "name": "Alice", "department_id": 1}, {"employee_id": 2, "name": "Bob", "department_id": 2}, {"employee_id": 3, "name": "Charlie", "department_id": null}, {"employee_id": 4, "name": "David", "department_id": 1}, {"employee_id": 5, "name": "Eve", "department_id": 3} ] ``` **dept_data_join** stream: ```json [ {"department_id": 1, "department_name": "HR"}, {"department_id": 2, "department_name": "Engineering"}, {"department_id": 3, "department_name": "Sales"}, {"department_id": 4, "department_name": "Marketing"} ] ``` A join lets you combine these streams to see the department name alongside each employee, matched by `department_id`. ### Types of Joins OpenObserve supports three types of joins: | Join Type | Description | When to Use | |-----------|-------------|-------------| | **INNER JOIN** | Returns only records with matches in both streams | When you need data that exists in both sources | | **LEFT JOIN** | Returns all records from left stream + matching records from right | When you want all data from your primary source, with supplemental info where available | | **RIGHT JOIN** | Returns all records from right stream + matching records from left | When you want all data from your secondary source, with supplemental info where available | > **Note:** UNION and CROSS joins are not currently supported. --- ### Using Joins (Step-by-Step) #### Step 1: Access the Joins Section In your dashboard panel editor, locate the **Joins** section. Click the **+** button to add a join. <img width="1725" alt="Access Joins Interface" src="https://github.com/user-attachments/assets/35db4f28-c526-43a2-bdd1-470e188f53d7" /> In this example: - **Primary stream**: `employee_data_join` (contains employee names) - We'll join it with `dept_data_join` (contains department information) - The interface shows available fields from both streams in the left panel #### Step 2: Configure the Join Click on the **Joins** dropdown and select the stream you want to join. Then configure: 1. **Joining Stream**: Select the stream to join (e.g., `employee_data_join`) 2. **Join Type**: Choose from LEFT, INNER, or RIGHT using the visual toggle 3. **On Stream**: Select the target stream (e.g., `dept_data_join`) <img width="1725" alt="Configure Join Type" src="https://github.com/user-attachments/assets/35db4f28-c526-43a2-bdd1-470e188f53d7" /> The interface displays: **"Performing a Left Join between employee_data_join and dept_data_join"** Notice the visual toggle showing three join types: - **Left** (selected in this example) - **Inner** - **Right** #### Step 3: Define the Joining Clause Set up the join condition by specifying which fields to match between the two streams: **Clause 1:** - **Select Field** (Left): Choose a field from the first stream (e.g., `department_id`) - **Select Operator**: Choose comparison operator (usually `=`) - **Select Field** (Right): Choose the matching field from the second stream (e.g., `department_id`) You can add multiple clauses using the **+** button for complex join conditions (e.g., joining on multiple fields or adding additional filters). #### Step 4: View the Results After configuring the join, the data table will show combined results from both streams. <img width="1724" alt="Join Results" src="https://github.com/user-attachments/assets/94ca9f33-6cfa-4ad1-9b84-bee9db34f912" /> In this example: - **Name** column from `employee_data_join` (Alice, Bob, Charlie, David, Eve) - **Department Name** column from `dept_data_join` (HR, Engineering, HR, Sales) - Data successfully joined on `department_id` #### Step 5: Review the Generated SQL Check the **Query** section at the bottom to see the generated SQL: ```sql SELECT employee_data_join.name as "x_axis_1", stream_0.department_name as "x_axis_2" FROM "employee_data_join" LEFT JOIN "dept_data_join" AS stream_0 ON employee_data_join.department_id = stream_0.department_id ``` The SQL query shows exactly how OpenObserve is combining your data streams. --- --- ## Examples ### Example 1: String Concatenation with Nested Functions **Use Case**: Combine multiple fields into a single display value ```sql Concat(k8s_namespace_name, '-', k8s_pod_name) ``` This joins namespace and pod names with a hyphen separator. ### Example 2: Statistical Analysis with Nested Math Functions **Use Case**: Calculate rounded average of a metric ```sql Ceil(Avg(severity)) ``` This calculates the average severity level and rounds up to the nearest whole number. ### Example 3: Conditional Percentage Calculation **Use Case**: Calculate percentage of high-severity events ```sql ROUND(100.0 * SUM(CASE WHEN COALESCE(severity, 0) >= 4 THEN 1 ELSE 0 END) / COUNT(_timestamp), 2) ``` **Breakdown:** - Handles null values with `COALESCE` - Counts events where severity >= 4 - Calculates percentage of total events - Rounds to 2 decimal places ### Example 4: Employee-Department Join **Use Case**: Show employee names with their department information ```sql SELECT employee_data_join.name as "Name", dept_data_join.department_name as "Department Name" FROM employee_data_join LEFT JOIN dept_data_join ON employee_data_join.department_id = dept_data_join.department_id ``` This displays all employees with their department names, showing null for employees without a department assignment. ### Example 5: Complex Analytics - Joins + Nested Functions **Use Case**: Calculate error rate per service with deployment information ```sql SELECT logs.service_name, deployments.version, COUNT(*) as total_requests, ROUND((COUNT(CASE WHEN logs.status_code >= 500 THEN 1 END) * 100.0 / COUNT(*)), 2) as error_rate_percent FROM service_logs as logs LEFT JOIN deployments ON logs.service_name = deployments.service_name GROUP BY logs.service_name, deployments.version ORDER BY error_rate_percent DESC ``` This combines: - Join between service logs and deployment data - Nested functions for conditional counting and percentage calculation - Grouping and sorting for analysis --- ## Tips and Best Practices ### Nested Functions Best Practices ✅ **Start Simple**: Build your function one layer at a time - Begin with inner function: `AVG(response_time)` - Add outer layers: `ROUND(AVG(response_time), 2)` ✅ **Test Each Layer**: Verify results before adding complexity - Test the inner function first - Add one layer at a time and check output ✅ **Handle Null Values**: Use `COALESCE` to avoid unexpected nulls ```sql COALESCE(AVG(response_time), 0) ``` ✅ **Use Clear Aliases**: Name your calculated fields descriptively ```sql ROUND(AVG(response_time), 2) as avg_response_ms ``` ✅ **Check for Division by Zero**: Protect against zero denominators ```sql CASE WHEN COUNT(*) > 0 THEN (SUM(errors) * 100.0 / COUNT(*)) ELSE 0 END as error_rate ``` ### Joins Best Practices ✅ **Choose the Right Join Type**: - **INNER JOIN**: Both streams must have matching data - **LEFT JOIN**: Keep all records from primary stream - **RIGHT JOIN**: Keep all records from secondary stream ✅ **Verify Field Compatibility**: - Match data types (don't join string to number) - Ensure field names are correct - Check that values actually match between streams ✅ **Consider Performance**: - Join on indexed fields when possible - Test with limited data first --- ## FAQ ### Q: Can I nest more than 2 functions? **A**: Yes! You can nest as many functions as needed. Example: ```sql ROUND(CEIL(AVG(COALESCE(response_time, 0))), 2) ``` ### Q: What's the difference between LEFT and RIGHT joins? **A**: - **LEFT JOIN**: Keeps all records from the first (left) stream - **RIGHT JOIN**: Keeps all records from the second (right) stream - Choose based on which dataset you want to preserve completely ### Q: Can I join more than 2 streams? **A**: Yes! Click the **+** button in the Joins section to add additional joins. You can chain multiple joins together. ### Q: Do nested functions impact performance? **A**: Complex nested functions may take longer to compute, especially on large datasets. Test with small time ranges first and optimize your expressions. ### Q: Can I use Custom SQL instead of Auto SQL? **A**: Yes! If you prefer writing raw SQL, switch to **Custom SQL** mode. However, Auto SQL provides a visual interface that's easier for building queries. --- ## Additional Resources 📚 **DataFusion Documentation**: - [Aggregate Functions](https://datafusion.apache.org/user-guide/sql/aggregate_functions.html) - COUNT, SUM, AVG, etc. - [Scalar Functions](https://datafusion.apache.org/user-guide/sql/scalar_functions.html) - ROUND, CONCAT, CASE, etc. 🔗 **OpenObserve Resources**: - [GitHub Repository](https://github.com/openobserve/openobserve) - [PR #9343 - Feature Implementation](https://github.com/openobserve/openobserve/pull/9343) - [OpenObserve Documentation](https://openobserve.ai/docs/) 💬 **Get Help**: - [GitHub Issues](https://github.com/openobserve/openobserve/issues) - Report bugs or request features - [Community Discussions](https://github.com/openobserve/openobserve/discussions) - Ask questions and share ideas --- Design at: https://github.com/openobserve/designs/tree/main/joins --------- Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: Shrinath Rao <shnath@openobserve.ai>
631 lines
27 KiB
Python
631 lines
27 KiB
Python
import os # For environment variables
|
|
from requests.auth import HTTPBasicAuth
|
|
from datetime import datetime, timezone, timedelta # For timestamp conversion
|
|
import time # For sleep
|
|
from pathlib import Path # For file paths
|
|
import pytest # For fixtures
|
|
import random
|
|
import uuid
|
|
import string
|
|
import requests # or your HTTP client of choice
|
|
|
|
# Constants for WebSocket URL and user credentials
|
|
ZO_BASE_URL_SC = os.environ.get("ZO_BASE_URL_SC") # Use environment variable
|
|
ZO_ROOT_USER_EMAIL = os.environ.get("ZO_ROOT_USER_EMAIL") # Use environment variable
|
|
ZO_ROOT_USER_PASSWORD = os.environ.get("ZO_ROOT_USER_PASSWORD") # Use environment variable
|
|
|
|
# Directory for test data
|
|
root_dir = Path(__file__).parent.parent.parent
|
|
|
|
def generate_random_string(length=5):
|
|
"""Generate a random string of lowercase letters."""
|
|
characters = string.ascii_lowercase # Only lowercase letters
|
|
return ''.join(random.choice(characters) for _ in range(length))
|
|
|
|
# Generate a random stream name
|
|
random_string = generate_random_string()
|
|
|
|
stream_name = "tdef" + random_string
|
|
print("Random Stream:", stream_name)
|
|
|
|
@pytest.fixture(scope="module")
|
|
def org_id(create_session, base_url, random_string):
|
|
"""Create a new organization and return its identifier."""
|
|
session = create_session
|
|
session.auth = HTTPBasicAuth(ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD)
|
|
|
|
org_id_new = random_string(5)
|
|
print("Random Org ID:", org_id_new)
|
|
|
|
payload = {
|
|
"name": org_id_new
|
|
}
|
|
|
|
resp_create_organization = session.post(
|
|
f"{base_url}api/organizations", json=payload
|
|
)
|
|
|
|
print(resp_create_organization.content)
|
|
|
|
assert (
|
|
resp_create_organization.status_code == 200
|
|
), f"Expected 200, but got {resp_create_organization.status_code} {resp_create_organization.content}"
|
|
|
|
assert resp_create_organization.json().get("name") == org_id_new
|
|
|
|
org_id = resp_create_organization.json().get("identifier")
|
|
print("Org identifier:", org_id)
|
|
|
|
yield org_id
|
|
|
|
# Cleanup: Delete the organization after all tests are done
|
|
resp_delete_organization = session.delete(
|
|
f"{base_url}api/organizations/{org_id}"
|
|
)
|
|
print(f"Deleted organization {org_id}")
|
|
|
|
@pytest.mark.order(1)
|
|
def test_data(create_session, base_url, org_id):
|
|
"""Ingest data into the OpenObserve running instance."""
|
|
session = create_session
|
|
session.auth = HTTPBasicAuth(ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD)
|
|
|
|
# Open the JSON data file and read it
|
|
with open(root_dir / "test-data/logs_data.json") as f:
|
|
data = f.read()
|
|
|
|
url_ing = f"{base_url}api/{org_id}/{stream_name}/_json"
|
|
|
|
print("Ingested URL", url_ing) # Add this for debugging
|
|
|
|
resp_ing = session.post(url_ing, data=data, headers={"Content-Type": "application/json"})
|
|
|
|
print("Data ingested successfully for Summary, status code:", resp_ing.status_code)
|
|
print("Response content:", resp_ing.text)
|
|
assert resp_ing.status_code == 200, f"Data ingestion failed, status code: {resp_ing.status_code}"
|
|
|
|
# Wait for a few seconds to allow the data to be ingested
|
|
time.sleep(5) # Increase this time if necessary
|
|
|
|
# Payload for pipeline creation
|
|
pipeline_payload = {
|
|
"name": "p12345",
|
|
"description": "",
|
|
"source": {"source_type": "realtime"},
|
|
"nodes": [
|
|
{
|
|
"id": "1e1fa129-cfbd-4082-8d39-c4dc418a670f",
|
|
"type": "input",
|
|
"data": {
|
|
"node_type": "stream",
|
|
"stream_name": stream_name,
|
|
"stream_type": "logs",
|
|
"org_id": org_id,
|
|
},
|
|
"position": {"x": 169.83333, "y": 55},
|
|
"io_type": "input",
|
|
},
|
|
{
|
|
"id": "8e0ed123-7737-4801-9466-bd909c425c72",
|
|
"type": "output",
|
|
"data": {
|
|
"node_type": "stream",
|
|
"stream_name": stream_name,
|
|
"stream_type": "logs",
|
|
"org_id": org_id,
|
|
},
|
|
"position": {"x": 305.33334, "y": 276},
|
|
"io_type": "output",
|
|
},
|
|
],
|
|
"edges": [
|
|
{
|
|
"id": "e1e1fa129-cfbd-4082-8d39-c4dc418a670f-8e0ed123-7737-4801-9466-bd909c425c72",
|
|
"source": "1e1fa129-cfbd-4082-8d39-c4dc418a670f",
|
|
"target": "8e0ed123-7737-4801-9466-bd909c425c72",
|
|
}
|
|
],
|
|
"org": org_id,
|
|
}
|
|
|
|
# Create a pipeline
|
|
resp_create_pipeline = session.post(f"{base_url}api/{org_id}/pipelines", json=pipeline_payload)
|
|
assert resp_create_pipeline.status_code == 200, f"Expected status code 200 but got {resp_create_pipeline.status_code}"
|
|
print("Create pipeline response:", resp_create_pipeline.json())
|
|
|
|
# Wait for a few seconds to allow the pipeline to be created
|
|
time.sleep(5) # Increase this time if necessary
|
|
|
|
# Create new alert
|
|
alert_name = f"newalert_{random.randint(1000, 9999)}" # Make the name unique
|
|
folder_alert = f"newfloder_{random.randint(1000, 9999)}" # Make the name unique
|
|
|
|
headers = {"Content-Type": "application/json", "Custom-Header": "value"}
|
|
|
|
# create folder
|
|
url=base_url
|
|
payload_folder = {"description": "newfoldernvp", "name": folder_alert}
|
|
resp_create_folder = session.post(
|
|
f"{url}api/v2/{org_id}/folders/alerts", json=payload_folder)
|
|
|
|
print(resp_create_folder.content)
|
|
|
|
folder_id = resp_create_folder.json()["folderId"]
|
|
|
|
assert (
|
|
resp_create_folder.status_code == 200
|
|
), f"Expected 200, but got {resp_create_folder.status_code} {resp_create_folder.content}"
|
|
|
|
# createtemplate
|
|
template_alert = f"newtemp_{random.randint(10000, 99999)}" # Make the name unique
|
|
destination_alert = f"newdest_{random.randint(10000, 99999)}" # Make the name unique
|
|
|
|
payload_temp_alert = {
|
|
"name":template_alert,
|
|
"body":"{\n \"text\": \"For stream {stream_name} of organization {org_name} alert {alert_name} of type {alert_type} is active\"\n}",
|
|
"type":"http",
|
|
"title":""
|
|
}
|
|
|
|
# Create template for alerts
|
|
resp_create_templates_alert = session.post(
|
|
f"{url}api/{org_id}/alerts/templates", json=payload_temp_alert
|
|
)
|
|
|
|
print(resp_create_templates_alert.content)
|
|
assert (
|
|
resp_create_templates_alert.status_code == 200
|
|
), f"Create template 200, but got {resp_create_templates_alert.status_code} {resp_create_templates_alert.content}"
|
|
|
|
skip_tls_verify_value = False # Define the skip_tls_verify_value
|
|
|
|
payload_dest_alert = {
|
|
"url":"https://jsonplaceholder.typicode.com/todos",
|
|
"method":"get",
|
|
"skip_tls_verify":skip_tls_verify_value,
|
|
"template":template_alert,
|
|
"headers":{},
|
|
"name":destination_alert
|
|
}
|
|
|
|
# Create destination
|
|
resp_create_destinations_alert = session.post(
|
|
f"{url}api/{org_id}/alerts/destinations",
|
|
json=payload_dest_alert,
|
|
)
|
|
print(resp_create_destinations_alert.content)
|
|
|
|
assert (
|
|
resp_create_destinations_alert.status_code == 200
|
|
), f"Failed to create destination: {resp_create_destinations_alert.status_code} {resp_create_destinations_alert.content}"
|
|
|
|
# Get destination
|
|
resp_get_destinations_alert = session.get(
|
|
f"{url}api/{org_id}/alerts/destinations/{destination_alert}"
|
|
)
|
|
print(resp_get_destinations_alert.content)
|
|
|
|
assert (
|
|
resp_get_destinations_alert.status_code == 200
|
|
), f"Get all destinations list 200, but got {resp_get_destinations_alert.status_code} {resp_get_destinations_alert.content}"
|
|
|
|
# Check if the destination name matches
|
|
assert (
|
|
resp_get_destinations_alert.json()["name"] == destination_alert
|
|
), f"Destination {destination_alert} not found in the response"
|
|
|
|
time.sleep(5)
|
|
|
|
# Create logs
|
|
payload_logs = [
|
|
{
|
|
"Athlete": "newtemp",
|
|
"City": "Athens",
|
|
"Country": "HUN",
|
|
"Discipline": "Swimming",
|
|
"Sport": "Aquatics",
|
|
"Year": 1896,
|
|
},
|
|
{
|
|
"Athlete": "HERSCHMANN",
|
|
"City": "Athens",
|
|
"Country": "CHN",
|
|
"Discipline": "Swimming",
|
|
"Sport": "Aquatics",
|
|
"Year": 1896,
|
|
},
|
|
]
|
|
|
|
resp_create_logstream = session.post(
|
|
f"{url}api/{org_id}/{stream_name}/_json", json=payload_logs
|
|
)
|
|
|
|
print(resp_create_logstream.content)
|
|
assert (
|
|
resp_create_logstream.status_code == 200
|
|
), f"Get all logs list 200, but got {resp_create_logstream.status_code} {resp_create_logstream.content}"
|
|
|
|
time.sleep(5)
|
|
payload_alert = {
|
|
"name": alert_name,
|
|
"row_template": template_alert,
|
|
"stream_type": "logs",
|
|
"stream_name": stream_name,
|
|
"is_real_time": False,
|
|
"context_attributes": {},
|
|
"query_condition": {
|
|
"conditions": [ {
|
|
"column": "log",
|
|
"operator": "=",
|
|
"value": "200",
|
|
"type": None, # Set it to None or remove it
|
|
"id": str(uuid.uuid4())
|
|
} ],
|
|
"search_event_type": "ui",
|
|
"sql": "",
|
|
"promql": "",
|
|
"type": "custom",
|
|
"promql_condition": None,
|
|
"vrl_function": None,
|
|
"multi_time_range": []
|
|
},
|
|
"trigger_condition": {
|
|
"period": 10,
|
|
"operator": ">=",
|
|
"frequency": 1,
|
|
"cron": "",
|
|
"threshold": 3,
|
|
"silence": 10,
|
|
"frequency_type": "minutes",
|
|
"timezone": "UTC",
|
|
"tolerance_in_secs": 0
|
|
},
|
|
"org_id": "default",
|
|
"destinations": [destination_alert],
|
|
"enabled": True,
|
|
"description": "test",
|
|
"folderId": folder_id
|
|
}
|
|
resp_post_alertnew = session.post(
|
|
f"{url}api/v2/{org_id}/alerts",
|
|
json=payload_alert,
|
|
headers=headers,
|
|
)
|
|
print(resp_post_alertnew.content)
|
|
|
|
assert ( resp_post_alertnew.status_code == 200
|
|
), f"Post alert expected 200, but got {resp_post_alertnew.status_code} {resp_post_alertnew.content}"
|
|
print(f"Alert {alert_name} created successfully")
|
|
|
|
time.sleep(5) # Increase this time if necessary
|
|
|
|
# Make the request to create function
|
|
payload_function = {
|
|
"function": ".a=190025552",
|
|
"name": "pytestfunction",
|
|
"params": "row",
|
|
"transType": 0,
|
|
}
|
|
resp_create_function = session.post(
|
|
f"{base_url}api/{org_id}/functions", json=payload_function
|
|
)
|
|
print(resp_create_function.content)
|
|
assert (
|
|
resp_create_function.status_code == 200
|
|
), f"Expected 200, but got {resp_create_function.status_code} {resp_create_function.content}"
|
|
|
|
time.sleep(5) # Increase this time if necessary
|
|
|
|
# Make the request to create dashboard
|
|
payload_dashboard = {
|
|
"description": "pydashboards",
|
|
"title": "pytestdashboard",
|
|
"owner": ZO_ROOT_USER_EMAIL,
|
|
}
|
|
resp_create_dashboard = session.post(
|
|
f"{base_url}api/{org_id}/dashboards", json=payload_dashboard
|
|
)
|
|
print(resp_create_dashboard.content)
|
|
assert (
|
|
resp_create_dashboard.status_code == 200
|
|
), f"Expected 200, but got {resp_create_dashboard.status_code} {resp_create_dashboard.content}"
|
|
print("Dashboard created successfully")
|
|
|
|
@pytest.fixture
|
|
def base_url_sc():
|
|
"""Provide the base URL for the API of Super Cluster."""
|
|
return ZO_BASE_URL_SC
|
|
|
|
@pytest.mark.skip(reason="Skipped due to leftover resources from previous test runs causing failures. See MD Files/test_summary_failures.md for details.")
|
|
@pytest.mark.order(2)
|
|
def test_summary(create_session, base_url, base_url_sc, org_id):
|
|
"""Run an E2E test for summary mode."""
|
|
session = create_session
|
|
session.auth = HTTPBasicAuth(ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD) # Add this line
|
|
# Use base_url_sc if available, otherwise fall back to base_url
|
|
url_sc = base_url_sc if base_url_sc is not None else base_url
|
|
print("URL for super cluster:", url_sc)
|
|
time.sleep(5) # Increase this time if necessary
|
|
res_summary_sc = session.get(f"{url_sc}api/{org_id}/summary")
|
|
print("Summary SC response:", res_summary_sc.content) # Add this for debugging
|
|
assert res_summary_sc.status_code == 200, f"Summary failed, status code: {res_summary_sc.status_code}"
|
|
res_data_summary_sc = res_summary_sc.json()
|
|
print("Summary SC data:", res_data_summary_sc) # Add this for debugging
|
|
|
|
# Validate the number of streams
|
|
actual_num_streams_sc= res_data_summary_sc["streams"]["num_streams"]
|
|
expected_num_streams_sc = 1 # Adjust based on our expectations
|
|
assert actual_num_streams_sc == expected_num_streams_sc, f"Expected to be {expected_num_streams_sc}, but got {actual_num_streams_sc}"
|
|
|
|
# Validate the number of pipelines
|
|
pipelines_data_sc = res_data_summary_sc.get("pipelines", {})
|
|
actual_num_pipelines_sc = pipelines_data_sc.get("num_realtime", 0) # Default to 0 if not found
|
|
expected_num_pipelines_sc = 1 # Adjust based on your expectations
|
|
assert actual_num_pipelines_sc == expected_num_pipelines_sc, f"Expected to be {expected_num_pipelines_sc}, but got {actual_num_pipelines_sc}"
|
|
|
|
# Validate the number of alerts
|
|
alerts_data_sc = res_data_summary_sc.get("alerts", {})
|
|
# Validate the number of scheduled alerts
|
|
actual_num_scheduled_alerts_sc = alerts_data_sc.get("num_scheduled", 0) # Default to 0 if not found
|
|
expected_num_scheduled_alerts_sc = 1 # Adjust based on your expectations
|
|
assert actual_num_scheduled_alerts_sc == expected_num_scheduled_alerts_sc, (
|
|
f"Expected {expected_num_scheduled_alerts_sc} scheduled alerts, "
|
|
f"but got {actual_num_scheduled_alerts_sc}"
|
|
)
|
|
|
|
# Validate the total number of functions
|
|
actual_num_functions_sc = res_data_summary_sc.get("total_functions", 0) # Directly get the value
|
|
expected_num_functions_sc = 1 # Adjust based on your expectations
|
|
assert actual_num_functions_sc == expected_num_functions_sc, (
|
|
f"Expected total_functions to be {expected_num_functions_sc}, "
|
|
f"but got {actual_num_functions_sc}"
|
|
)
|
|
|
|
# Validate the number of dashboards
|
|
actual_num_dashboards_sc = res_data_summary_sc.get("total_dashboards", 0) # Directly get the value
|
|
expected_num_dashboards_sc = 1 # Adjust based on your expectations
|
|
assert actual_num_dashboards_sc == expected_num_dashboards_sc, (
|
|
f"Expected to be {expected_num_dashboards_sc}, but got {actual_num_dashboards_sc}"
|
|
)
|
|
|
|
# Retrieve the pipeline list (use base_url since it was created there)
|
|
resp_list_pipelines = session.get(f"{base_url}api/{org_id}/pipelines")
|
|
assert resp_list_pipelines.status_code == 200, f"Expected status code 200 but got {resp_list_pipelines.status_code}"
|
|
|
|
# Parse the pipeline list response
|
|
pipelines_list = resp_list_pipelines.json().get("list", [])
|
|
assert pipelines_list, "No pipelines found in the list response."
|
|
|
|
# Extract the pipeline_id for the created pipeline
|
|
pipeline_id = None
|
|
for pipeline in pipelines_list:
|
|
if pipeline["name"] == "p12345":
|
|
pipeline_id = pipeline["pipeline_id"]
|
|
break
|
|
|
|
assert pipeline_id, "Pipeline ID not found for the created pipeline."
|
|
print(f"Pipeline ID: {pipeline_id}")
|
|
resp_get_pipeline = session.get(f"{base_url}api/{org_id}/pipelines")
|
|
print(resp_get_pipeline.json())
|
|
assert resp_get_pipeline.status_code == 200
|
|
pipeline_list = resp_get_pipeline.json()["list"]
|
|
target_pipeline = next((p for p in pipeline_list if p["pipeline_id"] == pipeline_id), None)
|
|
assert target_pipeline is not None, f"Pipeline {pipeline_id} not found in the list"
|
|
|
|
# Delete the pipeline
|
|
resp_delete_pipeline = session.delete(f"{base_url}api/{org_id}/pipelines/{pipeline_id}")
|
|
assert resp_delete_pipeline.status_code == 200, f"Expected status code 200 but got {resp_delete_pipeline.status_code}"
|
|
print(f"Deleted pipeline {pipeline_id}")
|
|
|
|
# Make the request to get the list of alerts (use base_url since they were created there)
|
|
resp_get_allalertsnew = session.get(f"{base_url}api/v2/{org_id}/alerts")
|
|
# Ensure the response is successful
|
|
assert resp_get_allalertsnew.status_code == 200, f"Failed to fetch alerts: {resp_get_allalertsnew.status_code}"
|
|
# Parse the response JSON
|
|
response_json = resp_get_allalertsnew.json()
|
|
# Check if "list" is in the response and proceed
|
|
assert "list" in response_json, "Response does not contain 'list'"
|
|
# Get the list of alerts from the response
|
|
alerts = response_json["list"]
|
|
# Store destination and folder info before deleting alerts
|
|
destinations_to_delete = set()
|
|
folders_to_delete = set()
|
|
# Now you can iterate over the alerts
|
|
for alert in alerts:
|
|
alert_id = alert.get("alert_id")
|
|
assert alert_id, f"Alert ID is missing for alert: {alert}"
|
|
print(f"Extracted alert_id: {alert_id}")
|
|
# Store destinations and folder for cleanup
|
|
if alert.get("destinations"):
|
|
destinations_to_delete.update(alert["destinations"])
|
|
if alert.get("folderId"):
|
|
folders_to_delete.add(alert["folderId"])
|
|
# Validate the alert existence first
|
|
resp_check_alert = session.get(f"{base_url}api/v2/{org_id}/alerts/{alert_id}")
|
|
assert resp_check_alert.status_code == 200, f"Alert {alert_id} does not exist or cannot be retrieved."
|
|
print(f"Alert {alert_id} exists and is retrievable.")
|
|
# Proceed to delete the alert
|
|
resp_delete_alertnew = session.delete(f"{base_url}api/v2/{org_id}/alerts/{alert_id}")
|
|
print(f"Deleted Alert Response: {resp_delete_alertnew.text}")
|
|
assert resp_delete_alertnew.status_code == 200, f"Failed to delete alert {alert_id}"
|
|
print(f"Successfully deleted alert {alert_id}")
|
|
|
|
# Clean up alert destinations
|
|
for destination_name in destinations_to_delete:
|
|
resp_delete_dest = session.delete(f"{base_url}api/{org_id}/alerts/destinations/{destination_name}")
|
|
print(f"Deleted destination {destination_name}: {resp_delete_dest.status_code}")
|
|
|
|
# Clean up alert templates (get list and delete all)
|
|
resp_get_templates = session.get(f"{base_url}api/{org_id}/alerts/templates")
|
|
if resp_get_templates.status_code == 200:
|
|
templates_data = resp_get_templates.json()
|
|
# Handle both list and dict response formats
|
|
templates = templates_data if isinstance(templates_data, list) else templates_data.get("list", [])
|
|
for template in templates:
|
|
template_name = template.get("name")
|
|
if template_name:
|
|
resp_delete_template = session.delete(f"{base_url}api/{org_id}/alerts/templates/{template_name}")
|
|
print(f"Deleted template {template_name}: {resp_delete_template.status_code}")
|
|
|
|
# Clean up alert folders
|
|
for folder_id in folders_to_delete:
|
|
resp_delete_folder = session.delete(f"{base_url}api/v2/{org_id}/folders/{folder_id}")
|
|
print(f"Deleted folder {folder_id}: {resp_delete_folder.status_code}")
|
|
|
|
# Proceed to delete the function (use base_url since it was created there)
|
|
resp_delete_function = session.delete(
|
|
f"{base_url}api/{org_id}/functions/pytestfunction"
|
|
)
|
|
assert (
|
|
resp_delete_function.status_code == 200
|
|
), f"Deleting this function, but got {resp_delete_function.status_code} {resp_delete_function.content}"
|
|
print(f"Function deleted successfully")
|
|
|
|
# Proceed to delete the dashboard (use base_url since it was created there)
|
|
resp_get_alldashboards = session.get(f"{base_url}api/{org_id}/dashboards")
|
|
assert resp_get_alldashboards.status_code == 200, f"Expected status code 200 but got {resp_get_alldashboards.status_code}"
|
|
# Parse the response JSON
|
|
dashboards_response = resp_get_alldashboards.json()
|
|
print(f"Dashboard list response: {dashboards_response}")
|
|
# Extract the dashboard ID - handle both v5 and v6 schemas
|
|
dashboards = dashboards_response.get('dashboards', [])
|
|
if dashboards:
|
|
dashboard = dashboards[0]
|
|
# Try v7 first, then v6, then v5
|
|
dashboard_id = None
|
|
if dashboard.get('v8'):
|
|
dashboard_id = dashboard['v8']['dashboardId']
|
|
elif dashboard.get('v6'):
|
|
dashboard_id = dashboard['v6']['dashboardId']
|
|
elif dashboard.get('v5'):
|
|
dashboard_id = dashboard['v5']['dashboardId']
|
|
else:
|
|
raise AssertionError(f"Dashboard schema version not supported: {dashboard}")
|
|
|
|
print(f"Extracted Dashboard ID: {dashboard_id}")
|
|
# Now you can delete the dashboard using the extracted ID
|
|
resp_delete_dashboard = session.delete(
|
|
f"{base_url}api/{org_id}/dashboards/{dashboard_id}"
|
|
)
|
|
# Assert that the deletion was successful
|
|
assert resp_delete_dashboard.status_code == 200, f"Failed to delete dashboard, status code: {resp_delete_dashboard.status_code}"
|
|
print(f"Successfully deleted dashboard with ID: {dashboard_id}")
|
|
else:
|
|
print("No dashboards found to delete")
|
|
|
|
# Wait for deletion to propagate
|
|
time.sleep(3)
|
|
|
|
# Proceed to delete the stream (use base_url since it was created there)
|
|
resp_delete_stream= session.delete(f"{base_url}api/{org_id}/streams/{stream_name}?type=logs")
|
|
print(f"Deleted Stream Response: {resp_delete_stream.text}")
|
|
assert resp_delete_stream.status_code == 200, f"Failed to delete {stream_name}"
|
|
print(f"Successfully deleted stream {stream_name}")
|
|
|
|
@pytest.mark.skip(reason="Skipped due to dependency on test_summary. See MD Files/test_summary_failures.md for details.")
|
|
@pytest.mark.order(3)
|
|
def test_summary_validate(create_session, base_url, org_id):
|
|
"""Run an E2E test for summary mode."""
|
|
session = create_session
|
|
session.auth = HTTPBasicAuth(ZO_ROOT_USER_EMAIL, ZO_ROOT_USER_PASSWORD) # Add this line
|
|
print("Base URL:", base_url)
|
|
time.sleep(5) # Increase this time if necessary
|
|
|
|
# Retrieve the pipeline list
|
|
resp_list_pipelines = session.get(f"{base_url}api/{org_id}/pipelines")
|
|
assert resp_list_pipelines.status_code == 200, f"Expected status code 200 but got {resp_list_pipelines.status_code}"
|
|
# Parse the pipeline list response
|
|
pipelines_list = resp_list_pipelines.json().get("list", [])
|
|
# Assert that the pipelines_list is either empty or contains elements
|
|
if not pipelines_list:
|
|
print("No pipelines found in the list response.")
|
|
else:
|
|
print(f"Found {len(pipelines_list)} pipelines.")
|
|
|
|
# Make the request to get the list of alerts
|
|
resp_get_allalertsnew = session.get(f"{base_url}api/v2/{org_id}/alerts")
|
|
# Ensure the response is successful
|
|
assert resp_get_allalertsnew.status_code == 200, f"Failed to fetch alerts: {resp_get_allalertsnew.status_code}"
|
|
# Parse the response JSON
|
|
response_json = resp_get_allalertsnew.json()
|
|
# Check if "list" is in the response and proceed
|
|
assert "list" in response_json, "Response does not contain 'list'"
|
|
# Get the list of alerts from the response
|
|
alerts = response_json["list"]
|
|
# Check if the alerts list is empty
|
|
if not alerts:
|
|
print("No alerts found.")
|
|
else:
|
|
# Now you can iterate over the alerts
|
|
for alert in alerts:
|
|
alert_id = alert.get("alert_id")
|
|
assert alert_id, f"Alert ID is missing for alert: {alert}"
|
|
|
|
print(f"Extracted alert_id: {alert_id}")
|
|
|
|
# Fetch the functions
|
|
resp_ver_function = session.get(f"{base_url}api/{org_id}/functions")
|
|
print(f"Function verification response: {resp_ver_function.status_code}, content: {resp_ver_function.content}")
|
|
# Parse the response JSON
|
|
response_json = resp_ver_function.json()
|
|
# Assert that the status code indicates success
|
|
assert resp_ver_function.status_code == 200, f"Expected status code 200 but got {resp_ver_function.status_code}"
|
|
# Check if the "list" is present and assert that it's empty
|
|
assert "list" in response_json, "Response does not contain 'list'"
|
|
assert response_json["list"] == [], "Expected 'list' to be empty, but it contains elements."
|
|
|
|
# Fetch the dashboards
|
|
resp_ver_dashboard = session.get(f"{base_url}api/{org_id}/dashboards")
|
|
print(f"Dashboard verification response: {resp_ver_dashboard.status_code}, content: {resp_ver_dashboard.content}")
|
|
# Fetch the list of dashboards after deletion
|
|
assert resp_ver_dashboard.status_code == 200, f"Expected status code 200 but got {resp_ver_dashboard.status_code}"
|
|
# Parse the response JSON
|
|
dashboards_response = resp_ver_dashboard.json()
|
|
print(f"Dashboard list response: {dashboards_response}")
|
|
# Assert that the dashboards list is empty
|
|
assert 'dashboards' in dashboards_response, "Response does not contain 'dashboards'"
|
|
assert dashboards_response['dashboards'] == [], "Expected the dashboards list to be empty after deletion."
|
|
print("Successfully verified that the dashboard list is empty after deletion.")
|
|
|
|
res_summary = session.get(f"{base_url}api/{org_id}/summary")
|
|
print("Summary response:", res_summary.content) # Add this for debugging
|
|
assert res_summary.status_code == 200, f"Summary failed, status code: {res_summary.status_code}"
|
|
res_data_summary = res_summary.json()
|
|
print("Summary data:", res_data_summary) # Add this for debugging
|
|
|
|
# Validate the number of streams
|
|
actual_num_streams= res_data_summary["streams"]["num_streams"]
|
|
expected_num_streams = 0 # Adjust based on our expectations
|
|
assert actual_num_streams == expected_num_streams, f"Expected to be {expected_num_streams}, but got {actual_num_streams}"
|
|
print("Successfully verified that the number of streams is correct after deletion.")
|
|
|
|
# Validate the number of pipelines
|
|
pipelines_data = res_data_summary.get("pipelines", {})
|
|
actual_num_pipelines = pipelines_data.get("num_realtime", 0) # Default to 0 if not found
|
|
expected_num_pipelines = 0 # Adjust based on your expectations
|
|
assert actual_num_pipelines == expected_num_pipelines, f"Expected to be {expected_num_pipelines}, but got {actual_num_pipelines}"
|
|
|
|
# Validate the number of alerts
|
|
alerts_data = res_data_summary.get("alerts", {})
|
|
# Validate the number of scheduled alerts
|
|
actual_num_scheduled_alerts = alerts_data.get("num_scheduled", 0) # Default to 0 if not found
|
|
expected_num_scheduled_alerts = 0 # Adjust based on your expectations
|
|
assert actual_num_scheduled_alerts == expected_num_scheduled_alerts, (
|
|
f"Expected {expected_num_scheduled_alerts} scheduled alerts, "
|
|
f"but got {actual_num_scheduled_alerts}"
|
|
)
|
|
|
|
# Validate the total number of functions
|
|
actual_num_functions = res_data_summary.get("total_functions", 0) # Directly get the value
|
|
expected_num_functions = 0 # Adjust based on your expectations
|
|
assert actual_num_functions == expected_num_functions, (
|
|
f"Expected total_functions to be {expected_num_functions}, "
|
|
f"but got {actual_num_functions}"
|
|
)
|
|
|
|
# Validate the number of dashboards
|
|
actual_num_dashboards = res_data_summary.get("total_dashboards", 0) # Directly get the value
|
|
expected_num_dashboards = 0 # Adjust based on your expectations
|
|
assert actual_num_dashboards == expected_num_dashboards, (
|
|
f"Expected to be {expected_num_dashboards}, but got {actual_num_dashboards}"
|
|
) |