Log Analytics and Query Rules

def create_or_update(resource_group_name: str, rule_name: str, parameters: ScheduledQueryRuleResource, **kwargs: Any) -> ScheduledQueryRuleResource:
    """
    Creates or updates a scheduled query rule.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - rule_name: str - Name of the scheduled query rule
    - parameters: ScheduledQueryRuleResource - Query rule configuration
    
    Returns:
    ScheduledQueryRuleResource - The created or updated query rule
    """

def get(resource_group_name: str, rule_name: str, **kwargs: Any) -> ScheduledQueryRuleResource:
    """
    Retrieve an scheduled query rule definition.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - rule_name: str - Name of the scheduled query rule
    
    Returns:
    ScheduledQueryRuleResource - The query rule details
    """

def delete(resource_group_name: str, rule_name: str, **kwargs: Any) -> None:
    """
    Deletes a scheduled query rule.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - rule_name: str - Name of the scheduled query rule
    """

def update(resource_group_name: str, rule_name: str, parameters: ScheduledQueryRuleResourcePatch, **kwargs: Any) -> ScheduledQueryRuleResource:
    """
    Update a scheduled query rule.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - rule_name: str - Name of the scheduled query rule
    - parameters: ScheduledQueryRuleResourcePatch - Properties to update
    
    Returns:
    ScheduledQueryRuleResource - The updated query rule
    """

def list_by_subscription(**kwargs: Any) -> ItemPaged[ScheduledQueryRuleResource]:
    """
    Retrieve scheduled query rule definitions in a subscription.
    
    Returns:
    ItemPaged[ScheduledQueryRuleResource] - List of scheduled query rules
    """

def list_by_resource_group(resource_group_name: str, **kwargs: Any) -> ItemPaged[ScheduledQueryRuleResource]:
    """
    Retrieve scheduled query rule definitions in a resource group.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    
    Returns:
    ItemPaged[ScheduledQueryRuleResource] - List of scheduled query rules
    """

def create_or_update(log_profile_name: str, parameters: LogProfileResource, **kwargs: Any) -> LogProfileResource:
    """
    Create or update a log profile.
    
    Parameters:
    - log_profile_name: str - Name of the log profile
    - parameters: LogProfileResource - Log profile configuration
    
    Returns:
    LogProfileResource - The created or updated log profile
    """

def get(log_profile_name: str, **kwargs: Any) -> LogProfileResource:
    """
    Gets the log profile.
    
    Parameters:
    - log_profile_name: str - Name of the log profile
    
    Returns:
    LogProfileResource - The log profile details
    """

def delete(log_profile_name: str, **kwargs: Any) -> None:
    """
    Deletes the log profile.
    
    Parameters:
    - log_profile_name: str - Name of the log profile
    """

def list(**kwargs: Any) -> ItemPaged[LogProfileResource]:
    """
    List the log profiles.
    
    Returns:
    ItemPaged[LogProfileResource] - List of log profiles
    """

from azure.mgmt.monitor.models import (
    ScheduledQueryRuleResource, ScheduledQueryRuleCriteria,
    Condition, ConditionFailingPeriods, Actions
)

# Define KQL query and condition
condition = Condition(
    query="Heartbeat | summarize AggregatedValue = count() by Computer | where AggregatedValue < 1",
    time_aggregation="Count",
    metric_measure_column="AggregatedValue",
    operator="LessThan",
    threshold=1.0,
    failing_periods=ConditionFailingPeriods(
        number_of_evaluation_periods=2,
        min_failing_periods_to_alert=2
    )
)

# Define criteria
criteria = ScheduledQueryRuleCriteria(
    all_of=[condition]
)

# Define actions
actions = Actions(
    action_groups=[
        f"/subscriptions/{subscription_id}/resourceGroups/monitoring-rg/providers/Microsoft.Insights/actionGroups/admin-alerts"
    ]
)

# Create scheduled query rule
query_rule = ScheduledQueryRuleResource(
    location="East US",
    description="Alert when computers stop sending heartbeats",
    enabled=True,
    scopes=[
        f"/subscriptions/{subscription_id}/resourceGroups/monitoring-rg/providers/Microsoft.OperationalInsights/workspaces/central-workspace"
    ],
    evaluation_frequency="PT5M",  # Run every 5 minutes
    window_size="PT15M",  # Look at last 15 minutes
    criteria=criteria,
    actions=actions,
    severity=1,  # Error severity
    auto_mitigate=True
)

result = client.scheduled_query_rules.create_or_update(
    resource_group_name="monitoring-rg",
    rule_name="missing-heartbeat-alert",
    parameters=query_rule
)

# Alert on high application response times
app_performance_condition = Condition(
    query="""
    AppRequests
    | where TimeGenerated > ago(5m)
    | summarize AverageResponseTime = avg(DurationMs) by bin(TimeGenerated, 1m)
    | where AverageResponseTime > 2000
    """,
    time_aggregation="Average",
    metric_measure_column="AverageResponseTime",
    operator="GreaterThan",
    threshold=2000.0,
    failing_periods=ConditionFailingPeriods(
        number_of_evaluation_periods=3,
        min_failing_periods_to_alert=2
    )
)

app_perf_rule = ScheduledQueryRuleResource(
    location="East US",
    description="Alert when application response time exceeds 2 seconds",
    enabled=True,
    scopes=[workspace_id],
    evaluation_frequency="PT1M",
    window_size="PT5M",
    criteria=ScheduledQueryRuleCriteria(all_of=[app_performance_condition]),
    actions=actions,
    severity=2
)

# Alert based on custom application logs
custom_log_condition = Condition(
    query="""
    CustomApp_CL
    | where TimeGenerated > ago(10m)
    | where Status_s == "ERROR"
    | summarize ErrorCount = count() by bin(TimeGenerated, 1m)
    | where ErrorCount > 5
    """,
    time_aggregation="Total",
    metric_measure_column="ErrorCount",
    operator="GreaterThan",
    threshold=5.0,
    failing_periods=ConditionFailingPeriods(
        number_of_evaluation_periods=1,
        min_failing_periods_to_alert=1
    )
)

custom_log_rule = ScheduledQueryRuleResource(
    location="East US", 
    description="Alert on high error rate in custom application logs",
    enabled=True,
    scopes=[workspace_id],
    evaluation_frequency="PT5M",
    window_size="PT10M",
    criteria=ScheduledQueryRuleCriteria(all_of=[custom_log_condition]),
    actions=actions,
    severity=1
)

from azure.mgmt.monitor.models import LogProfileResource, RetentionPolicy

# Create log profile for activity log export
log_profile = LogProfileResource(
    location="global",
    locations=["East US", "West US"],  # Regions to collect from
    categories=["Write", "Delete", "Action"],  # Activity log categories
    retention_policy=RetentionPolicy(
        enabled=True,
        days=90  # Retain for 90 days
    ),
    storage_account_id=f"/subscriptions/{subscription_id}/resourceGroups/storage-rg/providers/Microsoft.Storage/storageAccounts/activitylogs",
    service_bus_rule_id=f"/subscriptions/{subscription_id}/resourceGroups/messaging-rg/providers/Microsoft.ServiceBus/namespaces/logs-namespace/authorizationrules/RootManageSharedAccessKey"
)

profile_result = client.log_profiles.create_or_update(
    log_profile_name="default-log-profile",
    parameters=log_profile
)

class ScheduledQueryRuleResource:
    """Scheduled query rule resource."""
    location: str  # Resource location
    description: Optional[str]  # Rule description
    enabled: bool  # Whether rule is enabled
    scopes: List[str]  # Log Analytics workspace scopes
    evaluation_frequency: str  # Query evaluation frequency (ISO 8601)
    window_size: str  # Query time window (ISO 8601)
    override_query_time_range: Optional[str]  # Override time range
    target_resource_types: Optional[List[str]]  # Target resource types
    criteria: ScheduledQueryRuleCriteria  # Query criteria
    mute_actions_duration: Optional[str]  # Mute actions duration
    actions: Optional[Actions]  # Actions to execute
    is_workspace_alerts_storage_configured: Optional[bool]  # Workspace alerts storage
    check_workspace_alerts_storage_configured: Optional[bool]  # Check storage
    skip_query_validation: Optional[bool]  # Skip query validation
    auto_mitigate: Optional[bool]  # Auto-resolve alerts
    display_name: Optional[str]  # Display name
    severity: Optional[int]  # Alert severity (0-4)
    created_with_api_version: Optional[str]  # Creation API version
    is_legacy_log_analytics_rule: Optional[bool]  # Legacy rule indicator

class ScheduledQueryRuleCriteria:
    """Scheduled query rule criteria."""
    all_of: List[Condition]  # All conditions must be met (AND logic)

class Condition:
    """Query rule condition."""
    query: str  # KQL query to execute
    time_aggregation: TimeAggregationType  # Time aggregation method
    metric_measure_column: Optional[str]  # Column to measure
    resource_id_column: Optional[str]  # Resource ID column
    operator: ConditionOperator  # Comparison operator
    threshold: float  # Threshold value
    failing_periods: ConditionFailingPeriods  # Failing period configuration
    metric_name: Optional[str]  # Metric name for multi-dimensional alerts

class ConditionFailingPeriods:
    """Failing periods configuration."""
    number_of_evaluation_periods: int  # Number of evaluation periods
    min_failing_periods_to_alert: int  # Minimum failing periods to alert

class Actions:
    """Actions to execute when alert fires."""
    action_groups: Optional[List[str]]  # Action group resource IDs
    custom_properties: Optional[Dict[str, str]]  # Custom properties

class LogProfileResource:
    """Log profile resource (deprecated)."""
    location: str  # Resource location
    locations: List[str]  # Locations to collect from
    categories: List[str]  # Activity log categories
    retention_policy: RetentionPolicy  # Retention policy
    storage_account_id: Optional[str]  # Storage account ID
    service_bus_rule_id: Optional[str]  # Service Bus authorization rule ID

class RetentionPolicy:
    """Retention policy configuration."""
    enabled: bool  # Whether retention is enabled
    days: int  # Number of days to retain

TimeAggregationType = Union["Count", "Average", "Minimum", "Maximum", "Total"]
ConditionOperator = Union["Equals", "GreaterThan", "GreaterThanOrEqual", "LessThan", "LessThanOrEqual"]