如何模拟触发Azure CI、ADF等服务的告警以验证配置有效性？

Azure CI（DevOps Pipeline）告警模拟 #

• 用Azure CLI触发预设的失败管道：
  1. 提前在Pipeline中添加必失败步骤：

     steps:
     - script: |
         exit 1
       displayName: "Force Pipeline Failure"
     

  2. 执行CLI命令触发管道：

     az pipelines run --name <你的管道名称> --org <Azure DevOps组织URL> --project <项目名称>
     

• 用REST API直接触发：

  curl -X POST -u <用户名>:<个人访问令牌> "https://dev.azure.com/<组织>/<项目>/_apis/pipelines/<管道ID>/runs?api-version=7.1-preview.1" -H "Content-Type: application/json" -d '{"resources": {"repositories": {"self": {"refName": "refs/heads/main"}}}}'
  

Azure Data Factory（ADF）告警模拟 #

• 用Azure CLI触发含错误活动的管道：
  1. 创建管道并添加指向无效存储账户/数据集的Copy Activity
  2. 执行CLI命令触发运行：

     az datafactory pipeline create-run --resource-group <资源组名称> --factory-name <ADF名称> --name <管道名称>
     

• PowerShell触发失败管道：

  Invoke-AzDataFactoryV2Pipeline -ResourceGroupName <资源组名称> -DataFactoryName <ADF名称> -PipelineName <管道名称>
  

Databricks告警模拟 #

• 用Databricks CLI提交失败作业：
  1. 准备报错脚本fail_job.py：

     raise Exception("Trigger Databricks Alert Test")
     

  2. 提交作业命令：

     databricks jobs create --json '{
       "name": "Test Fail Job",
       "tasks": [
         {
           "task_key": "FailTask",
           "spark_python_task": {
             "python_file": "/dbfs/FileStore/fail_job.py"
           },
           "new_cluster": {
             "spark_version": "13.3.x-scala2.12",
             "node_type_id": "Standard_DS3_v2",
             "num_workers": 1
           }
         }
       ]
     }'
     

• REST API触发已存在的失败任务：

  curl -X POST -n https://<Databricks工作区URL>/api/2.1/jobs/runs/submit -d '{
    "job_id": <已存在的作业ID>,
    "notebook_task": {
      "notebook_path": "/path/to/fail_notebook"
    }
  }'
  

Azure机器学习告警模拟 #

• CLI触发失败训练作业：
  1. 准备报错训练脚本fail_train.py：

     assert False, "Trigger ML Alert Test"
     

  2. 提交作业（需提前编写job.yml指定运行该脚本）：

     az ml job create --file job.yml --resource-group <资源组名称> --workspace-name <ML工作区名称>
     

• 模拟模型部署失败：

  az ml model deploy --name <部署名称> --model <无效模型路径> --endpoint <端点名称> --resource-group <资源组名称> --workspace-name <ML工作区名称>
  

Event Hub/Event Grid告警模拟 #

Event Hub #

• 批量发消息模拟容量超限：

  from azure.eventhub import EventHubProducerClient, EventData
  import time
  
  producer = EventHubProducerClient.from_connection_string("<Event Hub连接字符串>", eventhub_name="<Event Hub名称>")
  with producer:
      for _ in range(10000):
          event_data_batch = producer.create_batch()
          event_data_batch.add(EventData(b"Test Message"))
          producer.send_batch(event_data_batch)
          time.sleep(0.01)
  

• 发无效格式消息触发解析错误：

  az eventhubs event send --resource-group <资源组名称> --namespace-name <命名空间名称> --eventhub-name <Event Hub名称> --message "Invalid Format Message" --properties "invalid_key=value"
  

Event Grid #

• 发事件到无效端点模拟投递失败：
  提前将事件订阅端点设为无效URL（如https://invalid-endpoint.example.com），再执行：

  az eventgrid event send --topic-name <事件主题名称> --resource-group <资源组名称> --event-data '{
    "id": "12345",
    "eventType": "Test.Failure",
    "subject": "TestAlert",
    "eventTime": "'$(date -u +%Y-%m-%dT%H:%M:%SZ)'",
    "data": {"key": "value"},
    "dataVersion": "1.0"
  }'
  

内容的提问来源于stack exchange，提问作者zay9899