-
Notifications
You must be signed in to change notification settings - Fork 0
Add Wolfgang.Etl.ErrorPolicies package (lockstep 0.21.0) #347
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit
Hold shift + click to select a range
d770fa7
Add Wolfgang.Etl.ErrorPolicies package (lockstep 0.21.0)
Chris-Wolfgang a13cc1f
Merge vNext (#345 convenience bases) into feat/error-policies-package
Chris-Wolfgang 924059b
Add ErrorPolicy delegate property to the base stages
Chris-Wolfgang 3d506c0
Make full-channel dead-letter drops observable (#347)
Chris-Wolfgang File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,195 @@ | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Threading.Channels; | ||
| using Microsoft.Extensions.Logging; | ||
| using Wolfgang.Etl.Abstractions; | ||
|
|
||
| namespace Wolfgang.Etl.ErrorPolicies; | ||
|
|
||
| /// <summary> | ||
| /// Ready-made policies for an ETL stage's <c>OnError</c> hook. Each is a | ||
| /// <see cref="Func{T, TResult}"/> from an <see cref="ItemErrorContext"/> to an | ||
| /// <see cref="ItemErrorAction"/>, so it can be assigned directly to a stage's <c>OnError</c>: | ||
| /// <example><code> | ||
| /// var deadLetters = new List<ItemErrorContext>(); | ||
| /// var extractor = new SomeExtractor<Record>(source) | ||
| /// { | ||
| /// OnError = ItemErrorPolicy.SkipDeadLetterAndLog(deadLetters, logger) | ||
| /// }; | ||
| /// </code></example> | ||
| /// The dead-letter overloads write to a caller-owned sink (a collection or a channel), so its size — | ||
| /// and therefore the memory a bad feed can consume — stays under the caller's control. | ||
| /// </summary> | ||
|
Chris-Wolfgang marked this conversation as resolved.
|
||
| public static class ItemErrorPolicy | ||
| { | ||
| /// <summary> | ||
| /// A policy that discards the failed item and continues with the next one. The stage increments | ||
| /// its error-item count (<c>CurrentErrorItemCount</c>) so the skip is never silent. | ||
| /// </summary> | ||
| public static Func<ItemErrorContext, ItemErrorAction> Skip { get; } = _ => ItemErrorAction.Skip; | ||
|
|
||
|
|
||
|
|
||
| /// <summary> | ||
| /// A policy that re-throws the failure and stops the run. Equivalent to leaving <c>OnError</c> | ||
| /// unset, provided for symmetry and explicitness. | ||
| /// </summary> | ||
| public static Func<ItemErrorContext, ItemErrorAction> Abort { get; } = _ => ItemErrorAction.Abort; | ||
|
|
||
|
|
||
|
|
||
| /// <summary> | ||
| /// Logs the failure as a warning through <paramref name="logger"/>, then discards the item and | ||
| /// continues. | ||
| /// </summary> | ||
| /// <param name="logger">The logger the returned policy writes each failure to.</param> | ||
| /// <returns>A policy that logs and returns <see cref="ItemErrorAction.Skip"/>.</returns> | ||
| /// <exception cref="ArgumentNullException"><paramref name="logger"/> is <see langword="null"/>.</exception> | ||
| public static Func<ItemErrorContext, ItemErrorAction> SkipAndLog(ILogger logger) | ||
| { | ||
| if (logger is null) | ||
| { | ||
| throw new ArgumentNullException(nameof(logger)); | ||
| } | ||
|
|
||
| return context => | ||
| { | ||
| ItemErrorPolicyLog.ItemFailedAndSkipped(logger, context.ItemNumber, context.Exception); | ||
| return ItemErrorAction.Skip; | ||
| }; | ||
| } | ||
|
|
||
|
|
||
|
|
||
| /// <summary> | ||
| /// Records the failure in <paramref name="deadLetters"/> (a "dead-letter" queue the caller owns), | ||
| /// then discards the item and continues. | ||
| /// </summary> | ||
| /// <param name="deadLetters">The caller-owned collection each failed item is added to.</param> | ||
| /// <returns>A policy that dead-letters and returns <see cref="ItemErrorAction.Skip"/>.</returns> | ||
| /// <exception cref="ArgumentNullException"><paramref name="deadLetters"/> is <see langword="null"/>.</exception> | ||
| /// <remarks> | ||
| /// A single stage invokes this policy serially, so a plain <see cref="List{T}"/> is safe. If you | ||
| /// share one collection across stages running concurrently, either supply a thread-safe collection | ||
| /// or use the <see cref="SkipAndDeadLetter(ChannelWriter{ItemErrorContext})"/> overload — the policy | ||
| /// adds to the collection without locking. | ||
| /// </remarks> | ||
| public static Func<ItemErrorContext, ItemErrorAction> SkipAndDeadLetter(ICollection<ItemErrorContext> deadLetters) | ||
| { | ||
| if (deadLetters is null) | ||
| { | ||
| throw new ArgumentNullException(nameof(deadLetters)); | ||
| } | ||
|
|
||
| return context => | ||
| { | ||
| deadLetters.Add(context); | ||
| return ItemErrorAction.Skip; | ||
| }; | ||
| } | ||
|
|
||
|
|
||
|
|
||
| /// <summary> | ||
| /// Writes the failure to <paramref name="deadLetters"/> (a caller-owned channel) with | ||
| /// <see cref="ChannelWriter{T}.TryWrite(T)"/>, then discards the item and continues. Because the | ||
| /// hook is synchronous the non-blocking <c>TryWrite</c> is used, so a bounded channel that is full | ||
| /// drops the failure — size the channel, or use <see cref="BoundedChannelFullMode"/>, accordingly. | ||
| /// </summary> | ||
| /// <param name="deadLetters">The caller-owned channel each failed item is written to.</param> | ||
| /// <returns>A policy that dead-letters and returns <see cref="ItemErrorAction.Skip"/>.</returns> | ||
| /// <exception cref="ArgumentNullException"><paramref name="deadLetters"/> is <see langword="null"/>.</exception> | ||
| public static Func<ItemErrorContext, ItemErrorAction> SkipAndDeadLetter(ChannelWriter<ItemErrorContext> deadLetters) | ||
| { | ||
| if (deadLetters is null) | ||
| { | ||
| throw new ArgumentNullException(nameof(deadLetters)); | ||
| } | ||
|
|
||
| return context => | ||
| { | ||
| deadLetters.TryWrite(context); | ||
| return ItemErrorAction.Skip; | ||
| }; | ||
| } | ||
|
|
||
|
|
||
|
|
||
| /// <summary> | ||
| /// Records the failure in <paramref name="deadLetters"/> and logs it as a warning through | ||
| /// <paramref name="logger"/>, then discards the item and continues. | ||
| /// </summary> | ||
| /// <param name="deadLetters">The caller-owned collection each failed item is added to.</param> | ||
| /// <param name="logger">The logger the returned policy writes each failure to.</param> | ||
| /// <returns>A policy that dead-letters, logs, and returns <see cref="ItemErrorAction.Skip"/>.</returns> | ||
| /// <exception cref="ArgumentNullException"> | ||
| /// <paramref name="deadLetters"/> or <paramref name="logger"/> is <see langword="null"/>. | ||
| /// </exception> | ||
| /// <remarks> | ||
| /// A single stage invokes this policy serially, so a plain <see cref="List{T}"/> is safe. If you | ||
| /// share one collection across stages running concurrently, either supply a thread-safe collection | ||
| /// or use the <see cref="SkipDeadLetterAndLog(ChannelWriter{ItemErrorContext}, ILogger)"/> overload — | ||
| /// the policy adds to the collection without locking. | ||
| /// </remarks> | ||
| public static Func<ItemErrorContext, ItemErrorAction> SkipDeadLetterAndLog | ||
| ( | ||
| ICollection<ItemErrorContext> deadLetters, | ||
| ILogger logger | ||
| ) | ||
| { | ||
| if (deadLetters is null) | ||
| { | ||
| throw new ArgumentNullException(nameof(deadLetters)); | ||
| } | ||
|
|
||
| if (logger is null) | ||
| { | ||
| throw new ArgumentNullException(nameof(logger)); | ||
| } | ||
|
|
||
| return context => | ||
| { | ||
| deadLetters.Add(context); | ||
| ItemErrorPolicyLog.ItemFailedAndSkipped(logger, context.ItemNumber, context.Exception); | ||
| return ItemErrorAction.Skip; | ||
| }; | ||
| } | ||
|
|
||
|
|
||
|
|
||
| /// <summary> | ||
| /// Writes the failure to <paramref name="deadLetters"/> with | ||
| /// <see cref="ChannelWriter{T}.TryWrite(T)"/> and logs it as a warning through | ||
| /// <paramref name="logger"/>, then discards the item and continues. See | ||
| /// <see cref="SkipAndDeadLetter(ChannelWriter{ItemErrorContext})"/> for the <c>TryWrite</c> caveat. | ||
| /// </summary> | ||
| /// <param name="deadLetters">The caller-owned channel each failed item is written to.</param> | ||
| /// <param name="logger">The logger the returned policy writes each failure to.</param> | ||
| /// <returns>A policy that dead-letters, logs, and returns <see cref="ItemErrorAction.Skip"/>.</returns> | ||
| /// <exception cref="ArgumentNullException"> | ||
| /// <paramref name="deadLetters"/> or <paramref name="logger"/> is <see langword="null"/>. | ||
| /// </exception> | ||
| public static Func<ItemErrorContext, ItemErrorAction> SkipDeadLetterAndLog | ||
| ( | ||
| ChannelWriter<ItemErrorContext> deadLetters, | ||
| ILogger logger | ||
| ) | ||
| { | ||
| if (deadLetters is null) | ||
| { | ||
| throw new ArgumentNullException(nameof(deadLetters)); | ||
| } | ||
|
|
||
| if (logger is null) | ||
| { | ||
| throw new ArgumentNullException(nameof(logger)); | ||
| } | ||
|
|
||
| return context => | ||
| { | ||
| deadLetters.TryWrite(context); | ||
| ItemErrorPolicyLog.ItemFailedAndSkipped(logger, context.ItemNumber, context.Exception); | ||
| return ItemErrorAction.Skip; | ||
| }; | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| using System; | ||
| using Microsoft.Extensions.Logging; | ||
|
|
||
| namespace Wolfgang.Etl.ErrorPolicies; | ||
|
|
||
| /// <summary> | ||
| /// Cached <see cref="LoggerMessage"/> delegates for the logging error policies, so a per-item log | ||
| /// call allocates nothing on the hot path. | ||
| /// </summary> | ||
| internal static class ItemErrorPolicyLog | ||
| { | ||
| internal static readonly Action<ILogger, long, Exception?> ItemFailedAndSkipped = | ||
| LoggerMessage.Define<long> | ||
| ( | ||
| LogLevel.Warning, | ||
| new EventId(1, nameof(ItemFailedAndSkipped)), | ||
| "Item {ItemNumber} failed to process and was skipped." | ||
| ); | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| #nullable enable |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| #nullable enable | ||
| Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy | ||
| static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.Abort.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>! | ||
| static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.Skip.get -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>! | ||
| static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndDeadLetter(System.Collections.Generic.ICollection<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>! | ||
| static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndDeadLetter(System.Threading.Channels.ChannelWriter<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>! | ||
| static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipAndLog(Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>! | ||
| static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipDeadLetterAndLog(System.Collections.Generic.ICollection<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters, Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>! | ||
| static Wolfgang.Etl.ErrorPolicies.ItemErrorPolicy.SkipDeadLetterAndLog(System.Threading.Channels.ChannelWriter<Wolfgang.Etl.Abstractions.ItemErrorContext!>! deadLetters, Microsoft.Extensions.Logging.ILogger! logger) -> System.Func<Wolfgang.Etl.Abstractions.ItemErrorContext!, Wolfgang.Etl.Abstractions.ItemErrorAction>! |
61 changes: 61 additions & 0 deletions
61
src/Wolfgang.Etl.ErrorPolicies/Wolfgang.Etl.ErrorPolicies.csproj
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
| <PropertyGroup> | ||
| <TargetFrameworks>net462;net472;net48;net481;netstandard2.0;net5.0;net6.0;net7.0;net8.0;net9.0;net10.0</TargetFrameworks> | ||
| <LangVersion>latest</LangVersion> | ||
| <!-- Lockstep with Wolfgang.Etl.Abstractions: this package is built and released from the same | ||
| repo under the same version and tag (like Wolfgang.Etl.TestKit + Wolfgang.Etl.TestKit.Xunit). --> | ||
| <Version>0.21.0</Version> | ||
| <AssemblyVersion>1.0.0.0</AssemblyVersion> | ||
| <FileVersion>$([System.Text.RegularExpressions.Regex]::Replace("$(Version)", "[-+].*$", "")).0</FileVersion> | ||
| <!-- No EnablePackageValidation on the FIRST release: there is no previously-published baseline to | ||
| compare against. Enable it next cycle with PackageValidationBaselineVersion set to 0.21.0. --> | ||
| <GeneratePackageOnBuild>False</GeneratePackageOnBuild> | ||
| <Title>$(AssemblyName)</Title> | ||
| <Description>Ready-made item-error policies — skip, log, and dead-letter (to a collection or a channel) — for the OnError hook of Wolfgang.Etl extractors, loaders, and transformers. Built on Wolfgang.Etl.Abstractions.</Description> | ||
|
Chris-Wolfgang marked this conversation as resolved.
Outdated
|
||
| <PackageProjectUrl>https://github.com/Chris-Wolfgang/ETL-Abstractions</PackageProjectUrl> | ||
| <PackageReadmeFile>README.md</PackageReadmeFile> | ||
| <RepositoryUrl>https://github.com/Chris-Wolfgang/ETL-Abstractions</RepositoryUrl> | ||
| <PackageLicenseExpression>MIT</PackageLicenseExpression> | ||
| <PackageRequireLicenseAcceptance>True</PackageRequireLicenseAcceptance> | ||
| <PackageIcon>ETL-Abstractions.png</PackageIcon> | ||
| <GenerateDocumentationFile>True</GenerateDocumentationFile> | ||
| <SignAssembly>False</SignAssembly> | ||
| <!-- The 10.0.10 logging packages drop net5.0/net7.0 from their tested TFM list; they resolve the | ||
| netstandard2.0 asset there and work. Silence the NETSDK TFM-support warnings on those TFMs. --> | ||
| <SuppressTfmSupportBuildWarnings>true</SuppressTfmSupportBuildWarnings> | ||
| <PackageTags>ETL;Extract-Transform-Load;error-handling;dead-letter</PackageTags> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <InternalsVisibleTo Include="Wolfgang.Etl.ErrorPolicies.Tests.Unit" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <None Include="..\..\ETL-Abstractions.png"> | ||
| <Pack>True</Pack> | ||
| <PackagePath>\</PackagePath> | ||
| </None> | ||
| <None Include="..\..\README.md"> | ||
| <Pack>True</Pack> | ||
| <PackagePath>\</PackagePath> | ||
| </None> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\Wolfgang.Etl.Abstractions\Wolfgang.Etl.Abstractions.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| <!-- Logging.Abstractions is always a package reference (never in-box). --> | ||
| <ItemGroup> | ||
| <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.10" /> | ||
| </ItemGroup> | ||
|
|
||
| <!-- System.Threading.Channels is in-box on net5.0+; only the netFx / netstandard2.0 TFMs need the package. --> | ||
| <ItemGroup Condition="'$(TargetFramework)' == 'net462' OR '$(TargetFramework)' == 'net472' OR '$(TargetFramework)' == 'net48' OR '$(TargetFramework)' == 'net481' OR '$(TargetFramework)' == 'netstandard2.0'"> | ||
| <PackageReference Include="System.Threading.Channels" Version="10.0.10" /> | ||
| </ItemGroup> | ||
|
|
||
| <!-- Analyzer PackageReferences are centralized in Directory.Build.props --> | ||
|
|
||
|
|
||
| </Project> | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.