Workflow Execution Filtering and Monitoring

1
{
2
  "type": "LOGICAL_CONDITION_GROUP",
3
  "conditions": [
4
    {
5
      "type": "LOGICAL_CONDITION",
6
      "lhs_variable": {"type": "STRING", "value": "field_name"},
7
      "operator": ">=",
8
      "rhs_variable": {"type": "STRING", "value": "filter_value"}
9
    }
10
  ],
11
  "combinator": "AND",
12
  "negated": false
13
}

1
// Example URL filter parameter
2
const encodedFilter = '%7B%22type%22%3A%22LOGICAL_CONDITION_GROUP%22%2C%22conditions%22%3A%5B%7B%22type%22%3A%22LOGICAL_CONDITION%22%2C%22lhs_variable%22%3A%7B%22type%22%3A%22STRING%22%2C%22value%22%3A%22timestamp%22%7D%2C%22operator%22%3A%22%3E%3D%22%2C%22rhs_variable%22%3A%7B%22type%22%3A%22STRING%22%2C%22value%22%3A%222025-09-22T20%3A42%3A09.361Z%22%7D%7D%5D%2C%22combinator%22%3A%22AND%22%2C%22negated%22%3Afalse%7D';
3
4
// Decode the filter
5
const decodedFilter = decodeURIComponent(encodedFilter);
6
const filterObject = JSON.parse(decodedFilter);
7
8
console.log('Filter structure:', filterObject);

1
import json
2
from datetime import datetime, timedelta
3
4
def monitor_failed_executions(deployment_id, hours_back=1):
5
    """Monitor for failed executions in the last N hours."""
6
    
7
    # Calculate timestamp for filtering
8
    cutoff_time = datetime.utcnow() - timedelta(hours=hours_back)
9
    cutoff_iso = cutoff_time.isoformat() + 'Z'
10
    
11
    # Build filter for recent failures
12
    filter_conditions = {
13
        "type": "LOGICAL_CONDITION_GROUP",
14
        "conditions": [
15
            {
16
                "type": "LOGICAL_CONDITION",
17
                "lhs_variable": {"type": "STRING", "value": "timestamp"},
18
                "operator": ">=",
19
                "rhs_variable": {"type": "STRING", "value": cutoff_iso}
20
            },
21
            {
22
                "type": "LOGICAL_CONDITION",
23
                "lhs_variable": {"type": "STRING", "value": "state"},
24
                "operator": "==",
25
                "rhs_variable": {"type": "STRING", "value": "REJECTED"}
26
            }
27
        ],
28
        "combinator": "AND",
29
        "negated": False
30
    }
31
    
32
    # Retrieve failed executions
33
    failed_executions = client.workflow_deployments.list_executions(
34
        id=deployment_id,
35
        filters=json.dumps(filter_conditions),
36
        ordering="-timestamp"
37
    )
38
    
39
    # Process failures
40
    for execution in failed_executions.results:
41
        print(f"ALERT: Execution {execution.execution_id} failed at {execution.timestamp}")
42
        
43
        # Get detailed error information
44
        detail = client.workflow_deployments.retrieve_execution_event(
45
            id=deployment_id,
46
            execution_id=execution.execution_id
47
        )
48
        
49
        if hasattr(detail, 'error') and detail.error:
50
            print(f"Error: {detail.error.message}")
51
52
# Usage
53
monitor_failed_executions("your-deployment-id", hours_back=2)

1
def analyze_execution_performance(deployment_id, days_back=7):
2
    """Analyze execution performance over time."""
3
    
4
    from datetime import datetime, timedelta
5
    
6
    # Get executions from the last week
7
    cutoff_time = datetime.utcnow() - timedelta(days=days_back)
8
    cutoff_iso = cutoff_time.isoformat() + 'Z'
9
    
10
    filter_conditions = {
11
        "type": "LOGICAL_CONDITION_GROUP",
12
        "conditions": [
13
            {
14
                "type": "LOGICAL_CONDITION",
15
                "lhs_variable": {"type": "STRING", "value": "timestamp"},
16
                "operator": ">=",
17
                "rhs_variable": {"type": "STRING", "value": cutoff_iso}
18
            }
19
        ],
20
        "combinator": "AND",
21
        "negated": False
22
    }
23
    
24
    executions = client.workflow_deployments.list_executions(
25
        id=deployment_id,
26
        filters=json.dumps(filter_conditions),
27
        limit=1000,  # Adjust based on your volume
28
        ordering="-timestamp"
29
    )
30
    
31
    # Analyze results
32
    total_count = len(executions.results)
33
    successful_count = len([e for e in executions.results if e.state == "FULFILLED"])
34
    failed_count = len([e for e in executions.results if e.state == "REJECTED"])
35
    
36
    print(f"Execution Summary (Last {days_back} days):")
37
    print(f"Total: {total_count}")
38
    print(f"Successful: {successful_count} ({successful_count/total_count*100:.1f}%)")
39
    print(f"Failed: {failed_count} ({failed_count/total_count*100:.1f}%)")
40
    
41
    return {
42
        "total": total_count,
43
        "successful": successful_count,
44
        "failed": failed_count,
45
        "success_rate": successful_count/total_count if total_count > 0 else 0
46
    }