Comment déterminer la raison du message de file d'attente de poison

Question

J'ai une fonction azur avec un QueueTrigger qui traite une image et sort un blob ainsi qu'un enregistrement de table.

Parfois, lors du traitement de plusieurs grandes images, j'ai rencontré des exceptions OutOfMemory, ce qui peut entraîner la mise en file d'attente de l'élément de la file d'attente.

Parfois, il existe une condition de concurrence et l'enregistrement de la table en cours d'insertion entraîne une erreur car il existe déjà un enregistrement avec cette clé de partition et cette clé de ligne. Je pourrais contourner ces problèmes moi-même au sein des fonctions, mais ma façon préférée de gérer ceci serait que le message poison inclue une raison ou l'exception qui a causé que l'élément soit placé dans la file d'attente de poison. De cette façon, je peux avoir un autre déclencheur écouter sur la file d'attente de poison et agir en conséquence après avoir évalué ce qui s'est mal passé.

Comment cela devrait-il être géré? Je suis conscient de l'ErrorTrigger, ce qui est agréable d'obtenir l'exception, mais je ne vois pas comment je peux le relier à l'élément de file d'attente spécifique qui l'a causé.

Answer 1

Il est impossible de modifier une file d'attente de messages de file d'attente de stockage après l'échec de votre déclencheur de file d'attente.

Les fonctions Azure utilisent Web Jobs SDK. En utilisant directement Web Jobs SDK, il est possible d'écraser QueueProcessor.CopyMessageToPoisonQueueAsync. Quelques liens utiles:

https://github.com/Azure/azure-webjobs-sdk/issues/1204

https://github.com/Azure/azure-webjobs-sdk-samples/blob/master/BasicSamples/MiscOperations/CustomQueueProcessorFactory.cs

https://github.com/Azure/azure-webjobs-sdk/blob/ffae7c86ea87fd73748186bac2c38c5401b80f68/test/Microsoft.Azure.WebJobs.Host.EndToEndTests/AzureStorageEndToEndTests.cs

Answer 2

Il existe un moyen hacky pour suivre l'exception et le message qui a provoqué le message en utilisant un filtre FunctionInvocation.

public static class QueueFunction 
{ 
    [FunctionName("QueueFunction")] 
    [TrackFailedMessages] 
    public static void Run([QueueTrigger("myqueue")]string message) 
    { 
     throw new Exception("OMG"); 
    } 
} 

public class TrackFailedMessages : FunctionInvocationFilterAttribute 
{ 
    private static readonly ConcurrentDictionary<Guid, object> QueueParamters = new ConcurrentDictionary<Guid, object>(); 
    public override Task OnExecutingAsync(FunctionExecutingContext executingContext, CancellationToken cancellationToken) 
    { 
     if (executingContext.Arguments.TryGetValue("message", out var data)) 
     { 
      QueueParamters.AddOrUpdate(executingContext.FunctionInstanceId, data, (id, obj) => data); 
     } 
     return Task.CompletedTask; 
    } 

    public override Task OnExecutedAsync(FunctionExecutedContext executedContext, CancellationToken cancellationToken) 
    { 
     if (executedContext.FunctionResult.Exception != null && 
      QueueParamters.TryGetValue(executedContext.FunctionInstanceId, out var message)) 
     { 
      executedContext.Logger.LogCritical(
       "The message {message} caused the exception {exception}", 
       message, 
       executedContext.FunctionResult.Exception.ToString()); 
     } 
     QueueParamters.TryRemove(executedContext.FunctionInstanceId, out var _); 
     return Task.CompletedTask; 
    } 
} 

OnExecutinAsync sera appelée, lorsqu'une fonction est appelée. En utilisant l'executingContext, vous avez accès aux paramètres de la fonction. Vous pouvez rechercher votre paramètre QueueTrigger et stocker la valeur dans un dictionnaire, où FunctionInstanceId est la clé. FunctionInstanceId est un identifiant unique pour chaque invocation.

En OnExecuted vous pouvez vérifier si la fonction a soulevé et exception et le message du dictionnaire, puis stocker un autre endroit (base de données, journal, etc.)

Je ne sais pas si cela attrape le tableau de stockage Exception ...