Skip to content
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.

Sign up for GitHub

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

Checkpointer testing cont. #47125

Merged
merged 6 commits into from
Nov 13, 2024
Merged

Checkpointer testing cont. #47125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
24bbdc9
change checkpointer API
jaschrep-msft Nov 11, 2024
6f1df3f
move serialze closer to checkpoint write
jaschrep-msft Nov 11, 2024
dcae8dc
error report bug fix
jaschrep-msft Nov 11, 2024
c67465e
adapt tests to new interface
jaschrep-msft Nov 12, 2024
e1de08c
Merge branch 'main' into tm-checkpointer-testing-cont
jaschrep-msft Nov 12, 2024
4b86451
exportapi
jaschrep-msft Nov 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ...rage/Azure.Storage.DataMovement.Blobs/tests/Azure.Storage.DataMovement.Blobs.Tests.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
<Compile Include="$(AzureStorageDataMovementTestSharedSources)TestEventsRaised.cs" LinkBase="Shared\DataMovement" />
<Compile Include="$(AzureStorageDataMovementTestSharedSources)TestTransferWithTimeout.cs" LinkBase="Shared\DataMovement" />
<Compile Include="$(AzureStorageDataMovementTestSharedSources)DisposingLocalDirectory.cs" LinkBase="Shared\DataMovement" />
<Compile Include="$(AzureStorageDataMovementTestSharedSources)MemoryTransferCheckpointer.cs" LinkBase="Shared\DataMovement" />
<Compile Include="$(AzureStorageDataMovementTestSharedSources)MockQueueInternalTasks.cs" LinkBase="Shared\DataMovement" />
<Compile Include="$(AzureStorageDataMovementTestSharedSources)MockResourceCheckpointData.cs" LinkBase="Shared\DataMovement" />
<Compile Include="$(AzureStorageDataMovementTestSharedSources)StartTransferUploadTestBase.cs" LinkBase="Shared\DataMovement" />
Expand Down
6 changes: 3 additions & 3 deletions sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.net6.0.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,10 +237,10 @@ public partial class TransferItemCompletedEventArgs : Azure.Storage.DataMovement
}
public partial class TransferItemFailedEventArgs : Azure.Storage.DataMovement.DataTransferEventArgs
{
public TransferItemFailedEventArgs(string transferId, Azure.Storage.DataMovement.StorageResourceItem sourceResource, Azure.Storage.DataMovement.StorageResourceItem destinationResource, System.Exception exception, bool isRunningSynchronously, System.Threading.CancellationToken cancellationToken) : base (default(string), default(bool), default(System.Threading.CancellationToken)) { }
public Azure.Storage.DataMovement.StorageResourceItem DestinationResource { get { throw null; } }
public TransferItemFailedEventArgs(string transferId, Azure.Storage.DataMovement.StorageResource sourceResource, Azure.Storage.DataMovement.StorageResource destinationResource, System.Exception exception, bool isRunningSynchronously, System.Threading.CancellationToken cancellationToken) : base (default(string), default(bool), default(System.Threading.CancellationToken)) { }
public Azure.Storage.DataMovement.StorageResource DestinationResource { get { throw null; } }
public System.Exception Exception { get { throw null; } }
public Azure.Storage.DataMovement.StorageResourceItem SourceResource { get { throw null; } }
public Azure.Storage.DataMovement.StorageResource SourceResource { get { throw null; } }
}
public partial class TransferItemSkippedEventArgs : Azure.Storage.DataMovement.DataTransferEventArgs
{
Expand Down
6 changes: 3 additions & 3 deletions sdk/storage/Azure.Storage.DataMovement/api/Azure.Storage.DataMovement.netstandard2.0.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,10 +237,10 @@ public partial class TransferItemCompletedEventArgs : Azure.Storage.DataMovement
}
public partial class TransferItemFailedEventArgs : Azure.Storage.DataMovement.DataTransferEventArgs
{
public TransferItemFailedEventArgs(string transferId, Azure.Storage.DataMovement.StorageResourceItem sourceResource, Azure.Storage.DataMovement.StorageResourceItem destinationResource, System.Exception exception, bool isRunningSynchronously, System.Threading.CancellationToken cancellationToken) : base (default(string), default(bool), default(System.Threading.CancellationToken)) { }
public Azure.Storage.DataMovement.StorageResourceItem DestinationResource { get { throw null; } }
public TransferItemFailedEventArgs(string transferId, Azure.Storage.DataMovement.StorageResource sourceResource, Azure.Storage.DataMovement.StorageResource destinationResource, System.Exception exception, bool isRunningSynchronously, System.Threading.CancellationToken cancellationToken) : base (default(string), default(bool), default(System.Threading.CancellationToken)) { }
public Azure.Storage.DataMovement.StorageResource DestinationResource { get { throw null; } }
public System.Exception Exception { get { throw null; } }
public Azure.Storage.DataMovement.StorageResourceItem SourceResource { get { throw null; } }
public Azure.Storage.DataMovement.StorageResource SourceResource { get { throw null; } }
}
public partial class TransferItemSkippedEventArgs : Azure.Storage.DataMovement.DataTransferEventArgs
{
Expand Down
2 changes: 1 addition & 1 deletion sdk/storage/Azure.Storage.DataMovement/src/DisabledTransferCheckpointer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ public Task AddNewJobAsync(string transferId, StorageResource source, StorageRes
return Task.CompletedTask;
}

public Task AddNewJobPartAsync(string transferId, int partNumber, Stream headerStream, CancellationToken cancellationToken = default)
public Task AddNewJobPartAsync(string transferId, int partNumber, JobPartPlanHeader header, CancellationToken cancellationToken = default)
{
return Task.CompletedTask;
}
Expand Down
2 changes: 1 addition & 1 deletion sdk/storage/Azure.Storage.DataMovement/src/ITransferCheckpointer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ Task AddNewJobAsync(
Task AddNewJobPartAsync(
string transferId,
int partNumber,
Stream headerStream,
JobPartPlanHeader header,
CancellationToken cancellationToken = default);

Task<bool> IsEnumerationCompleteAsync(
Expand Down
15 changes: 5 additions & 10 deletions sdk/storage/Azure.Storage.DataMovement/src/JobPartInternal.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -473,16 +473,11 @@ public async virtual Task CleanupAbortedJobPartAsync()
/// </summary>
public async virtual Task AddJobPartToCheckpointerAsync()
{
JobPartPlanHeader header = this.ToJobPartPlanHeader();
using (Stream stream = new MemoryStream())
{
header.Serialize(stream);
await _checkpointer.AddNewJobPartAsync(
transferId: _dataTransfer.Id,
partNumber: PartNumber,
headerStream: stream,
cancellationToken: _cancellationToken).ConfigureAwait(false);
}
await _checkpointer.AddNewJobPartAsync(
transferId: _dataTransfer.Id,
partNumber: PartNumber,
header: this.ToJobPartPlanHeader(),
cancellationToken: _cancellationToken).ConfigureAwait(false);
}

internal async virtual Task SetCheckpointerStatusAsync()
Expand Down
13 changes: 8 additions & 5 deletions sdk/storage/Azure.Storage.DataMovement/src/JobPlan/JobPartPlanFile.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,21 +38,21 @@ public static async Task<JobPartPlanFile> CreateJobPartPlanFileAsync(
string checkpointerPath,
string id,
int jobPart,
Stream headerStream,
JobPartPlanHeader header,
CancellationToken cancellationToken = default)
{
Argument.AssertNotNullOrEmpty(checkpointerPath, nameof(checkpointerPath));
Argument.AssertNotNullOrEmpty(id, nameof(id));
Argument.AssertNotNull(jobPart, nameof(jobPart));
Argument.AssertNotNull(headerStream, nameof(headerStream));
Argument.AssertNotNull(header, nameof(header));

JobPartPlanFileName fileName = new JobPartPlanFileName(checkpointerPath: checkpointerPath, id: id, jobPartNumber: jobPart);
return await CreateJobPartPlanFileAsync(fileName, headerStream, cancellationToken).ConfigureAwait(false);
return await CreateJobPartPlanFileAsync(fileName, header, cancellationToken).ConfigureAwait(false);
}

public static async Task<JobPartPlanFile> CreateJobPartPlanFileAsync(
JobPartPlanFileName fileName,
Stream headerStream,
JobPartPlanHeader header,
CancellationToken cancellationToken = default)
{
JobPartPlanFile result = new JobPartPlanFile()
Expand All @@ -63,8 +63,11 @@ public static async Task<JobPartPlanFile> CreateJobPartPlanFileAsync(
try
{
using (FileStream fileStream = File.Create(result.FileName.ToString()))
using (MemoryStream ms = new())
{
await headerStream.CopyToAsync(
header.Serialize(ms);
ms.Position = 0;
await ms.CopyToAsync(
fileStream,
DataMovementConstants.DefaultStreamCopyBufferSize,
cancellationToken).ConfigureAwait(false);
Expand Down
7 changes: 3 additions & 4 deletions sdk/storage/Azure.Storage.DataMovement/src/LocalTransferCheckpointer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,14 +104,13 @@ public override async Task AddNewJobAsync(
public override async Task AddNewJobPartAsync(
string transferId,
int partNumber,
Stream headerStream,
JobPartPlanHeader header,
CancellationToken cancellationToken = default)
{
Argument.AssertNotNullOrEmpty(transferId, nameof(transferId));
Argument.AssertNotNull(partNumber, nameof(partNumber));
Argument.AssertNotNull(headerStream, nameof(headerStream));
Argument.AssertNotNull(header, nameof(header));
CancellationHelper.ThrowIfCancellationRequested(cancellationToken);
headerStream.Position = 0;

if (!_transferStates.ContainsKey(transferId))
{
Expand All @@ -124,7 +123,7 @@ public override async Task AddNewJobPartAsync(
_pathToCheckpointer,
transferId,
partNumber,
headerStream,
header,
cancellationToken).ConfigureAwait(false);

// Add the job part into the current state
Expand Down
4 changes: 2 additions & 2 deletions sdk/storage/Azure.Storage.DataMovement/src/SerializerTransferCheckpointer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,15 +45,15 @@ public abstract Task AddNewJobAsync(
/// </summary>
/// <param name="transferId">The transfer ID.</param>
/// <param name="partNumber">The job part number.</param>
/// <param name="headerStream">A <see cref="Stream"/> to the job part plan header.</param>
/// <param name="header">A <see cref="Stream"/> to the job part plan header.</param>
/// <param name="cancellationToken">
/// Optional <see cref="CancellationToken"/> to propagate
/// notifications that the operation should be canceled.
/// </param>
public abstract Task AddNewJobPartAsync(
string transferId,
int partNumber,
Stream headerStream,
JobPartPlanHeader header,
CancellationToken cancellationToken = default);

/// <summary>
Expand Down
12 changes: 6 additions & 6 deletions sdk/storage/Azure.Storage.DataMovement/src/TransferItemFailedEventArgs.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,14 @@ namespace Azure.Storage.DataMovement
public class TransferItemFailedEventArgs : DataTransferEventArgs
{
/// <summary>
/// Gets the <see cref="StorageResourceItem"/> that was the source resource for the transfer.
/// Gets the <see cref="StorageResource"/> that was the source resource for the transfer.
/// </summary>
public StorageResourceItem SourceResource { get; }
public StorageResource SourceResource { get; }

/// <summary>
/// Gets the <see cref="StorageResourceItem"/> that was the destination resource for the transfer.
/// Gets the <see cref="StorageResource"/> that was the destination resource for the transfer.
/// </summary>
public StorageResourceItem DestinationResource { get; }
public StorageResource DestinationResource { get; }

/// <summary>
/// Gets the <see cref="Exception"/> that was thrown during the job.
Expand Down Expand Up @@ -53,8 +53,8 @@ public class TransferItemFailedEventArgs : DataTransferEventArgs
/// </exception>
public TransferItemFailedEventArgs(
string transferId,
StorageResourceItem sourceResource,
StorageResourceItem destinationResource,
StorageResource sourceResource,
StorageResource destinationResource,
Exception exception,
bool isRunningSynchronously,
CancellationToken cancellationToken)
Expand Down
4 changes: 2 additions & 2 deletions sdk/storage/Azure.Storage.DataMovement/src/TransferJobInternal.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -493,8 +493,8 @@ ex is not TaskCanceledException &&
await TransferFailedEventHandler.RaiseAsync(
new TransferItemFailedEventArgs(
_dataTransfer.Id,
_sourceResource,
_destinationResource,
(StorageResource)_sourceResource ?? _sourceResourceContainer,
(StorageResource)_destinationResource ?? _destinationResourceContainer,
ex,
false,
_cancellationToken),
Expand Down
30 changes: 28 additions & 2 deletions sdk/storage/Azure.Storage.DataMovement/tests/JobPartPlanFileTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,20 @@ public async Task CreateJobPartPlanFileAsync_Base()
checkpointerPath: test.DirectoryPath,
id: transferId,
jobPart: jobPart,
headerStream: stream);
header: new(
DataMovementConstants.JobPartPlanFile.SchemaVersion_b3,
transferId,
jobPart,
System.DateTimeOffset.Now,
"mock",
"mock",
"mock",
"mock",
default,
default,
default,
default,
new()));
}

JobPartPlanFileName fileName = new JobPartPlanFileName(
Expand Down Expand Up @@ -63,7 +76,20 @@ public async Task CreateJobPartPlanFileAsync_FileName()
{
file = await JobPartPlanFile.CreateJobPartPlanFileAsync(
fileName: fileName,
headerStream: stream);
header: new(
DataMovementConstants.JobPartPlanFile.SchemaVersion_b3,
transferId,
jobPart,
System.DateTimeOffset.Now,
"mock",
"mock",
"mock",
"mock",
default,
default,
default,
default,
new()));
}

Assert.NotNull(file);
Expand Down
Loading