Azure函数事件网格触发器

Question

我已经实现了一个EventGrid触发器来响应Blob存储事件，其逻辑简化如下：

public static async void Run(
    JObject eventGridEvent,
    TraceWriter log,
    ExecutionContext context)
{
    string eventContent = ParseEvent(eventGridEvent);

    HttpClient client = GetProxyClient();
    HttpResponseMessage response = await client.GetAsync("blabla/" + eventContent);
    string responseContent = await response.Content.ReadAsStringAsync();
    log.Info("Here is the response :" + responseContent);
}

外部API的响应时间不长(1秒或更短)，我对主机的配置设置为默认(因此允许无限数量的并发调用)。

当同时添加多个blob（从只有2个blob开始）时，我在日志中得到了很多重复的事件（脚本正在快速地一个接一个地上传blob，中间没有等待时间）。

我觉得这可能是由于我从不承认收到事件，我不知道我是否应该在我的代码中执行此操作，或者事件网格触发器是否自动执行此操作。

确认事件处理的逻辑应该在EventGrid触发器(Http 200响应)中实现还是自动处理？

如果没有，我还应该得到重复的事件吗？通常，当上传单个blob时，我会收到它的事件3-4次。

我问这个问题的原因是，当使用Http触发器并返回400响应时，我也会得到重复的事件，这是有意义的，因为我没有确认正确处理了事件。但是，当我返回200响应时，我不会收到重复的事件。

谢谢

Answer 1

您可以尝试使用 Data.api 字符串的 EventGrid Advanced Filter 以 FlushWithClose 结尾。我的 Azure 函数在 Blob 上传时多次执行的原因是，为 Blob 上传执行的每个 AppendFile 操作都创建了 EventGrid 消息。

我发现（通过反复试验）Azure 数据工厂使用一系列 API 调用将单个 blob 写入 Blob 存储。

最终看起来像这样:

< li> 创建文件路径 < li> 租赁文件 < li> AppendFile < li> AppendFile < li> AppendFile(每次追加都会放入blob的一个块，直到blob完成) < li> FlushFile(这是文件已完成的实际指示；因此上面显示了高级过滤器) < li> 租赁文件

下面是一个示例查询，您可以自己查看此上传流：

< li >注意:您将需要上载到blob容器的示例文件的< code>Uri

//==================================================//
// Author: Eric
// Created: 2021-05-26 0900 
// Query: ADF-to-Blob Storage reference flow
// Purpose: 
// To provide a reference flow of ADF-to-Blob Storage
// file uploads
//==================================================//
// Assign variables
//==================================================//
let varStart = ago(10d);
let varEnd = now();
let varStorageAccount = '<storageaccountname>';
let varStatus = 'Success';
let varSampleUri = 'https://<storageaccountname>.dfs.core.windows.net/<containername>/<parentfolder1>%2F<parentfolder2>%2F<samplefilename.extension>'
//==================================================//
// Filter table
//==================================================//
StorageBlobLogs
| where TimeGenerated between (varStart .. varEnd)
  and AccountName == varStorageAccount
  and StatusText == varStatus
  and split(Uri, '?')[0] == varSampleUri
//==================================================//
// Group and parse results
//==================================================//
| summarize 
  count() by OperationName,
  CorrelationId,
  TimeGenerated,
  UserAgent = tostring(split(UserAgentHeader, ' ')[0]),
  RequesterAppId,
  AccountName, 
  ContainerName = tostring(split(tostring(parse_url(url_decode(Uri))['Path']), '/')[1]),
  FileName = tostring(split(tostring(parse_url(url_decode(Uri))['Path']), '/')[-1]),
  ChunkSize = format_bytes(RequestBodySize, 2, 'MB'),
  StatusCode,
  StatusText
| order by TimeGenerated asc

从不同来源(Azure Data Factory、Azure Storage Explorer、Python/C# SDK、Azure Portal等)上传样本很有趣。)并了解他们使用的不同API方法。事实上，您可能需要这样做来获得您的日志记录和警报拨入。

太糟糕了，这些方法没有跨工具标准化，因为这个特定问题是你自己发现的巨大痛苦！

同样，在这种情况下，EventGridAdvanced Filters是您的朋友。

Answer 2

您不需要做任何特殊的事情来向事件网格表示成功。如果函数执行成功（未引发异常），触发器将自动响应成功状态代码。