From 335e4422afebaf6210efcbac490ab11553504d35 Mon Sep 17 00:00:00 2001 From: Brittany Ellich Date: Wed, 25 Jan 2023 14:46:30 -0800 Subject: [PATCH 01/14] Rename queue to results queue --- src/Runner.Common/JobServer.cs | 10 ++ src/Runner.Common/JobServerQueue.cs | 107 ++++++++++++++++----- src/Runner.Common/Logging.cs | 2 +- src/Runner.Worker/ExecutionContext.cs | 2 +- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 23 +++++ 5 files changed, 118 insertions(+), 26 deletions(-) diff --git a/src/Runner.Common/JobServer.cs b/src/Runner.Common/JobServer.cs index 8a6c4a6a76e..42feca70e77 100644 --- a/src/Runner.Common/JobServer.cs +++ b/src/Runner.Common/JobServer.cs @@ -31,6 +31,7 @@ public interface IJobServer : IRunnerService, IAsyncDisposable Task AppendTimelineRecordFeedAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, Guid stepId, IList lines, long? startLine, CancellationToken cancellationToken); Task CreateAttachmentAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, String type, String name, Stream uploadStream, CancellationToken cancellationToken); Task CreateStepSymmaryAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken); + Task CreateResultsLogAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken); Task CreateLogAsync(Guid scopeIdentifier, string hubName, Guid planId, TaskLog log, CancellationToken cancellationToken); Task CreateTimelineAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, CancellationToken cancellationToken); Task> UpdateTimelineRecordsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, IEnumerable records, CancellationToken cancellationToken); @@ -325,6 +326,15 @@ public Task CreateStepSymmaryAsync(string planId, string jobId, string stepId, s throw new InvalidOperationException("Results client is not initialized."); } + public Task CreateResultsLogAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken) + { + if (_resultsClient != null) + { + return _resultsClient.UploadResultsLogAsync(planId, jobId, stepId, file, cancellationToken: cancellationToken); + } + throw new InvalidOperationException("Results client is not initialized."); + } + public Task CreateLogAsync(Guid scopeIdentifier, string hubName, Guid planId, TaskLog log, CancellationToken cancellationToken) { diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 6440da73607..cd513552af0 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -20,7 +20,7 @@ public interface IJobServerQueue : IRunnerService, IThrottlingReporter void Start(Pipelines.AgentJobRequestMessage jobRequest); void QueueWebConsoleLine(Guid stepRecordId, string line, long? lineNumber = null); void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, string name, string path, bool deleteSource); - void QueueSummaryUpload(Guid stepRecordId, string name, string path, bool deleteSource); + void QueueResultsUpload(Guid stepRecordId, string name, string path, string type, bool deleteSource); void QueueTimelineRecordUpdate(Guid timelineId, TimelineRecord timelineRecord); } @@ -31,7 +31,7 @@ public sealed class JobServerQueue : RunnerService, IJobServerQueue private static readonly TimeSpan _delayForWebConsoleLineDequeue = TimeSpan.FromMilliseconds(500); private static readonly TimeSpan _delayForTimelineUpdateDequeue = TimeSpan.FromMilliseconds(500); private static readonly TimeSpan _delayForFileUploadDequeue = TimeSpan.FromMilliseconds(1000); - private static readonly TimeSpan _delayForSummaryUploadDequeue = TimeSpan.FromMilliseconds(1000); + private static readonly TimeSpan _delayForResultsUploadDequeue = TimeSpan.FromMilliseconds(1000); // Job message information private Guid _scopeIdentifier; @@ -46,7 +46,7 @@ public sealed class JobServerQueue : RunnerService, IJobServerQueue // queue for file upload (log file or attachment) private readonly ConcurrentQueue _fileUploadQueue = new(); - private readonly ConcurrentQueue _summaryFileUploadQueue = new(); + private readonly ConcurrentQueue _resultsFileUploadQueue = new(); // queue for timeline or timeline record update (one queue per timeline) private readonly ConcurrentDictionary> _timelineUpdateQueue = new(); @@ -60,7 +60,7 @@ public sealed class JobServerQueue : RunnerService, IJobServerQueue // Task for each queue's dequeue process private Task _webConsoleLineDequeueTask; private Task _fileUploadDequeueTask; - private Task _summaryUploadDequeueTask; + private Task _resultsUploadDequeueTask; private Task _timelineUpdateDequeueTask; // common @@ -140,12 +140,12 @@ public void Start(Pipelines.AgentJobRequestMessage jobRequest) _fileUploadDequeueTask = ProcessFilesUploadQueueAsync(); Trace.Info("Start results file upload queue."); - _summaryUploadDequeueTask = ProcessSummaryUploadQueueAsync(); + _resultsUploadDequeueTask = ProcessResultsUploadQueueAsync(); Trace.Info("Start process timeline update queue."); _timelineUpdateDequeueTask = ProcessTimelinesUpdateQueueAsync(); - _allDequeueTasks = new Task[] { _webConsoleLineDequeueTask, _fileUploadDequeueTask, _timelineUpdateDequeueTask, _summaryUploadDequeueTask }; + _allDequeueTasks = new Task[] { _webConsoleLineDequeueTask, _fileUploadDequeueTask, _timelineUpdateDequeueTask, _resultsUploadDequeueTask }; _queueInProcess = true; } @@ -176,9 +176,9 @@ public async Task ShutdownAsync() await ProcessFilesUploadQueueAsync(runOnce: true); Trace.Info("File upload queue drained."); - Trace.Verbose("Draining results summary upload queue."); - await ProcessSummaryUploadQueueAsync(runOnce: true); - Trace.Info("Results summary upload queue drained."); + Trace.Verbose("Draining results upload queue."); + await ProcessResultsUploadQueueAsync(runOnce: true); + Trace.Info("Results upload queue drained."); // ProcessTimelinesUpdateQueueAsync() will throw exception during shutdown // if there is any timeline records that failed to update contains output variabls. @@ -226,17 +226,36 @@ public void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, DeleteSource = deleteSource }; + // Dual upload the same files to Results Service + if (String.Equals(type, CoreAttachmentType.Log, StringComparison.OrdinalIgnoreCase)) + { + newFile.Type = CoreAttachmentType.Log; + Trace.Verbose("Log was sent from logging file", newFile.Path, timelineRecordId); + // jobRequest.Variables.TryGetValue("system.github.results_endpoint", out VariableValue resultsEndpointVariable); + // context.Global.Variables.TryGetValue("system.github.results_endpoint", out string resultsReceiverEndpoint); + // if (resultsReceiverEndpoint != null) + // { + // Trace.Info($"Queueing results file ({path})"); + // var stepId = context.Id; + // // Attachments must be added to the parent context (job), not the current context (step) + // QueueResultsUpload(attachmentName, scrubbedFilePath, stepId); + // } + } + Trace.Verbose("Enqueue file upload queue: file '{0}' attach to record {1}", newFile.Path, timelineRecordId); _fileUploadQueue.Enqueue(newFile); + + } - public void QueueSummaryUpload(Guid stepRecordId, string name, string path, bool deleteSource) + public void QueueResultsUpload(Guid stepRecordId, string name, string path, string type, bool deleteSource) { // all parameter not null, file path exist. - var newFile = new SummaryUploadFileInfo() + var newFile = new ResultsUploadFileInfo() { Name = name, Path = path, + Type = type, PlanId = _planId.ToString(), JobId = _jobTimelineRecordId.ToString(), StepId = stepRecordId.ToString(), @@ -244,7 +263,7 @@ public void QueueSummaryUpload(Guid stepRecordId, string name, string path, bool }; Trace.Verbose("Enqueue results file upload queue: file '{0}' attach to job {1} step {2}", newFile.Path, _jobTimelineRecordId, stepRecordId); - _summaryFileUploadQueue.Enqueue(newFile); + _resultsFileUploadQueue.Enqueue(newFile); } public void QueueTimelineRecordUpdate(Guid timelineId, TimelineRecord timelineRecord) @@ -437,15 +456,15 @@ private async Task ProcessFilesUploadQueueAsync(bool runOnce = false) } } - private async Task ProcessSummaryUploadQueueAsync(bool runOnce = false) + private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) { Trace.Info("Starting results-based upload queue..."); while (!_jobCompletionSource.Task.IsCompleted || runOnce) { - List filesToUpload = new(); - SummaryUploadFileInfo dequeueFile; - while (_summaryFileUploadQueue.TryDequeue(out dequeueFile)) + List filesToUpload = new(); + ResultsUploadFileInfo dequeueFile; + while (_resultsFileUploadQueue.TryDequeue(out dequeueFile)) { filesToUpload.Add(dequeueFile); // process at most 10 file upload. @@ -459,7 +478,7 @@ private async Task ProcessSummaryUploadQueueAsync(bool runOnce = false) { if (runOnce) { - Trace.Info($"Uploading {filesToUpload.Count} summary files in one shot through results service."); + Trace.Info($"Uploading {filesToUpload.Count} files in one shot through results service."); } int errorCount = 0; @@ -467,11 +486,19 @@ private async Task ProcessSummaryUploadQueueAsync(bool runOnce = false) { try { - await UploadSummaryFile(file); + if (String.Equals(file.Type, ChecksAttachmentType.StepSummary, StringComparison.OrdinalIgnoreCase)) + { + await UploadSummaryFile(file); + } else if (String.Equals(file.Type, CoreAttachmentType.Log, StringComparison.OrdinalIgnoreCase)) + { + Trace.Info($"Got a log file to send to results service."); + await UploadResultsStepLogFile(file); + } + } catch (Exception ex) { - var issue = new Issue() { Type = IssueType.Warning, Message = $"Caught exception during summary file upload to results. {ex.Message}" }; + var issue = new Issue() { Type = IssueType.Warning, Message = $"Caught exception during file upload to results. {ex.Message}" }; issue.Data[Constants.Runner.InternalTelemetryIssueDataKey] = Constants.Runner.ResultsUploadFailure; var telemetryRecord = new TimelineRecord() @@ -481,7 +508,7 @@ private async Task ProcessSummaryUploadQueueAsync(bool runOnce = false) telemetryRecord.Issues.Add(issue); QueueTimelineRecordUpdate(_jobTimelineId, telemetryRecord); - Trace.Info("Catch exception during summary file upload to results, keep going since the process is best effort."); + Trace.Info("Catch exception during file upload to results, keep going since the process is best effort."); Trace.Error(ex); } finally @@ -490,7 +517,7 @@ private async Task ProcessSummaryUploadQueueAsync(bool runOnce = false) } } - Trace.Info("Tried to upload {0} summary files to results, success rate: {1}/{0}.", filesToUpload.Count, filesToUpload.Count - errorCount); + Trace.Info("Tried to upload {0} files to results, success rate: {1}/{0}.", filesToUpload.Count, filesToUpload.Count - errorCount); } if (runOnce) @@ -499,7 +526,7 @@ private async Task ProcessSummaryUploadQueueAsync(bool runOnce = false) } else { - await Task.Delay(_delayForSummaryUploadDequeue); + await Task.Delay(_delayForResultsUploadDequeue); } } } @@ -736,15 +763,18 @@ private async Task UploadFile(UploadFileInfo file) if (String.Equals(file.Type, CoreAttachmentType.Log, StringComparison.OrdinalIgnoreCase)) { // Create the log + Trace.Info("Create log async!"); var taskLog = await _jobServer.CreateLogAsync(_scopeIdentifier, _hubName, _planId, new TaskLog(String.Format(@"logs\{0:D}", file.TimelineRecordId)), default(CancellationToken)); // Upload the contents using (FileStream fs = File.Open(file.Path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)) { + Trace.Info("Uploading a log!!"); var logUploaded = await _jobServer.AppendLogContentAsync(_scopeIdentifier, _hubName, _planId, taskLog.Id, fs, default(CancellationToken)); } // Create a new record and only set the Log field + Trace.Info("Updating timeline record!"); var attachmentUpdataRecord = new TimelineRecord() { Id = file.TimelineRecordId, Log = taskLog }; QueueTimelineRecordUpdate(file.TimelineId, attachmentUpdataRecord); } @@ -776,7 +806,7 @@ private async Task UploadFile(UploadFileInfo file) } } - private async Task UploadSummaryFile(SummaryUploadFileInfo file) + private async Task UploadSummaryFile(ResultsUploadFileInfo file) { bool uploadSucceed = false; try @@ -804,6 +834,34 @@ private async Task UploadSummaryFile(SummaryUploadFileInfo file) } } } + + private async Task UploadResultsStepLogFile(ResultsUploadFileInfo file) + { + bool uploadSucceed = false; + try + { + Trace.Info($"Starting to step log file to results service {file.Name}, {file.Path}"); + var cancellationTokenSource = new CancellationTokenSource(); + await _jobServer.CreateResultsLogAsync(file.PlanId, file.JobId, file.StepId, file.Path, cancellationTokenSource.Token); + + uploadSucceed = true; + } + finally + { + if (uploadSucceed && file.DeleteSource) + { + try + { + File.Delete(file.Path); + } + catch (Exception ex) + { + Trace.Info("Catch exception during delete success results uploaded step log file."); + Trace.Error(ex); + } + } + } + } } internal class PendingTimelineRecord @@ -822,9 +880,10 @@ internal class UploadFileInfo public bool DeleteSource { get; set; } } - internal class SummaryUploadFileInfo + internal class ResultsUploadFileInfo { public string Name { get; set; } + public string Type { get; set; } public string Path { get; set; } public string PlanId { get; set; } public string JobId { get; set; } diff --git a/src/Runner.Common/Logging.cs b/src/Runner.Common/Logging.cs index 40be5cdcff0..a3f98ec68ec 100644 --- a/src/Runner.Common/Logging.cs +++ b/src/Runner.Common/Logging.cs @@ -114,7 +114,7 @@ private void EndPage() _pageWriter.Dispose(); _pageWriter = null; _pageData = null; - _jobServerQueue.QueueFileUpload(_timelineId, _timelineRecordId, "DistributedTask.Core.Log", "CustomToolLog", _dataFileName, true); + _jobServerQueue.QueueFileUpload(_timelineId, _timelineRecordId, "DistributedTask.Core.NotLog", "CustomToolLog", _dataFileName, true); } } } diff --git a/src/Runner.Worker/ExecutionContext.cs b/src/Runner.Worker/ExecutionContext.cs index c4a7db6fee4..f0f92f79a7a 100644 --- a/src/Runner.Worker/ExecutionContext.cs +++ b/src/Runner.Worker/ExecutionContext.cs @@ -857,7 +857,7 @@ public void QueueSummaryFile(string name, string filePath, Guid stepRecordId) throw new FileNotFoundException($"Can't upload (name:{name}) file: {filePath}. File does not exist."); } - _jobServerQueue.QueueSummaryUpload(stepRecordId, name, filePath, deleteSource: false); + _jobServerQueue.QueueResultsUpload(stepRecordId, name, filePath, ChecksAttachmentType.StepSummary, deleteSource: false); } // Add OnMatcherChanged diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index 77a733eaeba..ecf7f957585 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -135,6 +135,29 @@ public async Task UploadStepSummaryAsync(string planId, string jobId, string ste await StepSummaryUploadCompleteAsync(planId, jobId, stepId, fileSize, cancellationToken); } + // Handle file upload for step summary + public async Task UploadResultsLogAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken) + { + // Get the upload url + var uploadUrlResponse = await GetStepSummaryUploadUrlAsync(planId, jobId, stepId, cancellationToken); + if (uploadUrlResponse == null) + { + throw new Exception("Failed to get step summary upload url"); + } + + // Do we want to throw an exception here or should we just be uploading/truncating the data + var fileSize = new FileInfo(file).Length; + + // Upload the file + using (var fileStream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, true)) + { + var response = await UploadFileAsync(uploadUrlResponse.SummaryUrl, uploadUrlResponse.BlobStorageType, fileStream, cancellationToken); + } + + // Send step summary upload complete message + await StepSummaryUploadCompleteAsync(planId, jobId, stepId, fileSize, cancellationToken); + } + private MediaTypeFormatter m_formatter; private Uri m_resultsServiceUrl; private string m_token; From 3bc52cfe2dd40292b4885cdf1a4d139b9de513b4 Mon Sep 17 00:00:00 2001 From: Brittany Ellich Date: Thu, 26 Jan 2023 16:22:32 -0800 Subject: [PATCH 02/14] Add results contracts --- src/Runner.Common/Logging.cs | 5 +- src/Sdk/WebApi/WebApi/Contracts.cs | 48 ++++++++++++++++ src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 67 +++++++++++++++++++++- 3 files changed, 116 insertions(+), 4 deletions(-) diff --git a/src/Runner.Common/Logging.cs b/src/Runner.Common/Logging.cs index a3f98ec68ec..64b8b727d31 100644 --- a/src/Runner.Common/Logging.cs +++ b/src/Runner.Common/Logging.cs @@ -114,7 +114,10 @@ private void EndPage() _pageWriter.Dispose(); _pageWriter = null; _pageData = null; - _jobServerQueue.QueueFileUpload(_timelineId, _timelineRecordId, "DistributedTask.Core.NotLog", "CustomToolLog", _dataFileName, true); + // Figure out if this is a results log and also queue the results log + _jobServerQueue.QueueFileUpload(_timelineId, _timelineRecordId, "DistributedTask.Core.Log", "CustomToolLog", _dataFileName, true); + // Need to get FF here + _jobServerQueue.QueueResultsUpload(_timelineRecordId, "CustomToolLog", _dataFileName, "DistributedTask.Core.Log", true); } } } diff --git a/src/Sdk/WebApi/WebApi/Contracts.cs b/src/Sdk/WebApi/WebApi/Contracts.cs index d240cc1e230..a23de2e83c6 100644 --- a/src/Sdk/WebApi/WebApi/Contracts.cs +++ b/src/Sdk/WebApi/WebApi/Contracts.cs @@ -28,6 +28,30 @@ public class GetSignedStepSummaryURLResponse public string BlobStorageType; } + [DataContract] + [JsonObject(NamingStrategyType = typeof(SnakeCaseNamingStrategy))] + public class GetSignedStepLogsURLRequest + { + [DataMember] + public string WorkflowJobRunBackendId; + [DataMember] + public string WorkflowRunBackendId; + [DataMember] + public string StepBackendId; + } + + [DataContract] + [JsonObject(NamingStrategyType = typeof(SnakeCaseNamingStrategy))] + public class GetSignedStepLogsURLResponse + { + [DataMember] + public string SummaryUrl; + [DataMember] + public long SoftSizeLimit; + [DataMember] + public string BlobStorageType; + } + [DataContract] [JsonObject(NamingStrategyType = typeof(SnakeCaseNamingStrategy))] public class StepSummaryMetadataCreate @@ -52,6 +76,30 @@ public class CreateStepSummaryMetadataResponse public bool Ok; } + [DataContract] + [JsonObject(NamingStrategyType = typeof(SnakeCaseNamingStrategy))] + public class StepLogsMetadataCreate + { + [DataMember] + public string StepBackendId; + [DataMember] + public string WorkflowRunBackendId; + [DataMember] + public string WorkflowJobRunBackendId; + [DataMember] + public long Size; + [DataMember] + public string UploadedAt; + } + + [DataContract] + [JsonObject(NamingStrategyType = typeof(SnakeCaseNamingStrategy))] + public class CreateStepLogsMetadataResponse + { + [DataMember] + public bool Ok; + } + public static class BlobStorageTypes { public static readonly string AzureBlobStorage = "BLOB_STORAGE_TYPE_AZURE"; diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index ecf7f957585..74d78bb3a59 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -51,6 +51,33 @@ public async Task GetStepSummaryUploadUrlAsync( } } + public async Task GetStepLogUploadUrlAsync(string planId, string jobId, string stepId, CancellationToken cancellationToken) + { + var request = new GetSignedStepLogsURLRequest() + { + WorkflowJobRunBackendId= jobId, + WorkflowRunBackendId= planId, + StepBackendId= stepId + }; + + var stepLogsUploadRequest = new Uri(m_resultsServiceUrl, "twirp/results.services.receiver.Receiver/GetStepStepLogsSignedBlobURL"); + + using (HttpRequestMessage requestMessage = new HttpRequestMessage(HttpMethod.Post, stepLogsUploadRequest)) + { + requestMessage.Headers.Authorization = new AuthenticationHeaderValue("Bearer", m_token); + requestMessage.Headers.Accept.Add(MediaTypeWithQualityHeaderValue.Parse("application/json")); + + using (HttpContent content = new ObjectContent(request, m_formatter)) + { + requestMessage.Content = content; + using (var response = await SendAsync(requestMessage, HttpCompletionOption.ResponseContentRead, cancellationToken: cancellationToken)) + { + return await ReadJsonContentAsync(response, cancellationToken); + } + } + } + } + private async Task StepSummaryUploadCompleteAsync(string planId, string jobId, string stepId, long size, CancellationToken cancellationToken) { var timestamp = DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss.fffK"); @@ -85,6 +112,40 @@ private async Task StepSummaryUploadCompleteAsync(string planId, string jobId, s } } + private async Task StepLogUploadCompleteAsync(string planId, string jobId, string stepId, long size, CancellationToken cancellationToken) + { + var timestamp = DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss.fffK"); + var request = new StepLogsMetadataCreate() + { + WorkflowJobRunBackendId= jobId, + WorkflowRunBackendId= planId, + StepBackendId = stepId, + Size = size, + UploadedAt = timestamp + }; + + var stepLogsUploadCompleteRequest = new Uri(m_resultsServiceUrl, "twirp/results.services.receiver.Receiver/CreateStepLogsMetadata"); + + using (HttpRequestMessage requestMessage = new HttpRequestMessage(HttpMethod.Post, stepLogsUploadCompleteRequest)) + { + requestMessage.Headers.Authorization = new AuthenticationHeaderValue("Bearer", m_token); + requestMessage.Headers.Accept.Add(MediaTypeWithQualityHeaderValue.Parse("application/json")); + + using (HttpContent content = new ObjectContent(request, m_formatter)) + { + requestMessage.Content = content; + using (var response = await SendAsync(requestMessage, HttpCompletionOption.ResponseContentRead, cancellationToken: cancellationToken)) + { + var jsonResponse = await ReadJsonContentAsync(response, cancellationToken); + if (!jsonResponse.Ok) + { + throw new Exception($"Failed to mark step log upload as complete, status code: {response.StatusCode}, ok: {jsonResponse.Ok}, size: {size}, timestamp: {timestamp}"); + } + } + } + } + } + private async Task UploadFileAsync(string url, string blobStorageType, FileStream file, CancellationToken cancellationToken) { // Upload the file to the url @@ -139,10 +200,10 @@ public async Task UploadStepSummaryAsync(string planId, string jobId, string ste public async Task UploadResultsLogAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken) { // Get the upload url - var uploadUrlResponse = await GetStepSummaryUploadUrlAsync(planId, jobId, stepId, cancellationToken); + var uploadUrlResponse = await GetStepLogUploadUrlAsync(planId, jobId, stepId, cancellationToken); if (uploadUrlResponse == null) { - throw new Exception("Failed to get step summary upload url"); + throw new Exception("Failed to get step log upload url"); } // Do we want to throw an exception here or should we just be uploading/truncating the data @@ -155,7 +216,7 @@ public async Task UploadResultsLogAsync(string planId, string jobId, string step } // Send step summary upload complete message - await StepSummaryUploadCompleteAsync(planId, jobId, stepId, fileSize, cancellationToken); + await StepLogUploadCompleteAsync(planId, jobId, stepId, fileSize, cancellationToken); } private MediaTypeFormatter m_formatter; From 94d39245aad65294985258bb852e17b68de4c387 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Mon, 30 Jan 2023 11:25:14 -0500 Subject: [PATCH 03/14] Add Results logging handling --- src/Runner.Common/JobServerQueue.cs | 4 +- src/Runner.Common/Logging.cs | 66 ++++++++++++++++++---- src/Runner.Worker/ExecutionContext.cs | 2 +- src/Sdk/WebApi/WebApi/Contracts.cs | 2 +- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 2 +- 5 files changed, 60 insertions(+), 16 deletions(-) diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index cd513552af0..3c57a019569 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -20,7 +20,7 @@ public interface IJobServerQueue : IRunnerService, IThrottlingReporter void Start(Pipelines.AgentJobRequestMessage jobRequest); void QueueWebConsoleLine(Guid stepRecordId, string line, long? lineNumber = null); void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, string name, string path, bool deleteSource); - void QueueResultsUpload(Guid stepRecordId, string name, string path, string type, bool deleteSource); + void QueueResultsUpload(Guid stepRecordId, string name, string path, string type, bool deleteSource, bool finalize); void QueueTimelineRecordUpdate(Guid timelineId, TimelineRecord timelineRecord); } @@ -248,7 +248,7 @@ public void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, } - public void QueueResultsUpload(Guid stepRecordId, string name, string path, string type, bool deleteSource) + public void QueueResultsUpload(Guid stepRecordId, string name, string path, string type, bool deleteSource, bool finalize) { // all parameter not null, file path exist. var newFile = new ResultsUploadFileInfo() diff --git a/src/Runner.Common/Logging.cs b/src/Runner.Common/Logging.cs index 64b8b727d31..5d6ffad4a80 100644 --- a/src/Runner.Common/Logging.cs +++ b/src/Runner.Common/Logging.cs @@ -32,6 +32,19 @@ public class PagingLogger : RunnerService, IPagingLogger private string _pagesFolder; private IJobServerQueue _jobServerQueue; + // For Results + public static string BlocksFolder = "blocks"; + + // 2 MB + public const int BlockSize = 2 * 1024 * 1024; + + private string _resultsDataFileName; + private FileStream _resultsBlockData; + private StreamWriter _resultsBlockWriter; + private string _resultsBlockFolder; + private int _blockByteCount; + private int _blockCount; + public long TotalLines => _totalLines; public override void Initialize(IHostContext hostContext) @@ -39,8 +52,10 @@ public override void Initialize(IHostContext hostContext) base.Initialize(hostContext); _totalLines = 0; _pagesFolder = Path.Combine(hostContext.GetDirectory(WellKnownDirectory.Diag), PagingFolder); - _jobServerQueue = HostContext.GetService(); Directory.CreateDirectory(_pagesFolder); + _resultsBlockFolder = Path.Combine(hostContext.GetDirectory(WellKnownDirectory.Diag), BlocksFolder); + Directory.CreateDirectory(_resultsBlockFolder); + _jobServerQueue = HostContext.GetService(); } public void Setup(Guid timelineId, Guid timelineRecordId) @@ -60,11 +75,17 @@ public void Write(string message) // lazy creation on write if (_pageWriter == null) { - Create(); + NewPage(); + } + + if (_resultsBlockWriter == null) + { + NewBlock(); } string line = $"{DateTime.UtcNow.ToString("O")} {message}"; _pageWriter.WriteLine(line); + _resultsBlockWriter.WriteLine(line); _totalLines++; if (line.IndexOf('\n') != -1) @@ -78,21 +99,25 @@ public void Write(string message) } } - _byteCount += System.Text.Encoding.UTF8.GetByteCount(line); + var bytes = System.Text.Encoding.UTF8.GetByteCount(line); + _byteCount += bytes; + _blockByteCount += bytes; if (_byteCount >= PageSize) { NewPage(); } + + if (_blockByteCount >= BlockSize) + { + NewBlock(); + } + } public void End() { EndPage(); - } - - private void Create() - { - NewPage(); + EndBlock(true); } private void NewPage() @@ -114,10 +139,29 @@ private void EndPage() _pageWriter.Dispose(); _pageWriter = null; _pageData = null; - // Figure out if this is a results log and also queue the results log _jobServerQueue.QueueFileUpload(_timelineId, _timelineRecordId, "DistributedTask.Core.Log", "CustomToolLog", _dataFileName, true); - // Need to get FF here - _jobServerQueue.QueueResultsUpload(_timelineRecordId, "CustomToolLog", _dataFileName, "DistributedTask.Core.Log", true); + } + } + + private void NewBlock() + { + EndBlock(false); + _blockByteCount = 0; + _resultsDataFileName = Path.Combine(_resultsBlockFolder, $"{_timelineId}_{_timelineRecordId}_{++_blockCount}.block"); + _resultsBlockData = new FileStream(_resultsDataFileName, FileMode.CreateNew, FileAccess.ReadWrite, FileShare.ReadWrite); + _resultsBlockWriter = new StreamWriter(_resultsBlockData, System.Text.Encoding.UTF8); + } + + private void EndBlock(bool finalize) + { + if (_resultsBlockWriter != null) + { + _resultsBlockWriter.Flush(); + _resultsBlockData.Flush(); + _resultsBlockWriter.Dispose(); + _resultsBlockWriter = null; + _resultsBlockData = null; + _jobServerQueue.QueueResultsUpload(_timelineRecordId, "ResultsLog", _resultsDataFileName, "Results.Core.Log", true, finalize); } } } diff --git a/src/Runner.Worker/ExecutionContext.cs b/src/Runner.Worker/ExecutionContext.cs index f0f92f79a7a..e4650ccf885 100644 --- a/src/Runner.Worker/ExecutionContext.cs +++ b/src/Runner.Worker/ExecutionContext.cs @@ -857,7 +857,7 @@ public void QueueSummaryFile(string name, string filePath, Guid stepRecordId) throw new FileNotFoundException($"Can't upload (name:{name}) file: {filePath}. File does not exist."); } - _jobServerQueue.QueueResultsUpload(stepRecordId, name, filePath, ChecksAttachmentType.StepSummary, deleteSource: false); + _jobServerQueue.QueueResultsUpload(stepRecordId, name, filePath, ChecksAttachmentType.StepSummary, deleteSource: false, finalize: false); } // Add OnMatcherChanged diff --git a/src/Sdk/WebApi/WebApi/Contracts.cs b/src/Sdk/WebApi/WebApi/Contracts.cs index a23de2e83c6..42b1041e72b 100644 --- a/src/Sdk/WebApi/WebApi/Contracts.cs +++ b/src/Sdk/WebApi/WebApi/Contracts.cs @@ -45,7 +45,7 @@ public class GetSignedStepLogsURLRequest public class GetSignedStepLogsURLResponse { [DataMember] - public string SummaryUrl; + public string BlobUrl; [DataMember] public long SoftSizeLimit; [DataMember] diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index 74d78bb3a59..e5bee89251e 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -212,7 +212,7 @@ public async Task UploadResultsLogAsync(string planId, string jobId, string step // Upload the file using (var fileStream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, true)) { - var response = await UploadFileAsync(uploadUrlResponse.SummaryUrl, uploadUrlResponse.BlobStorageType, fileStream, cancellationToken); + var response = await UploadFileAsync(uploadUrlResponse.BlobUrl, uploadUrlResponse.BlobStorageType, fileStream, cancellationToken); } // Send step summary upload complete message From f5413dfbc816eadfa121bc333d6a56da83fcb801 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Mon, 30 Jan 2023 17:09:47 -0500 Subject: [PATCH 04/14] Adding calls to create and finalize append blob --- src/Runner.Common/JobServer.cs | 6 +-- src/Runner.Common/JobServerQueue.cs | 23 ++++++--- src/Runner.Common/Logging.cs | 4 +- src/Runner.Worker/ExecutionContext.cs | 2 +- src/Sdk/DTWebApi/WebApi/TaskAttachment.cs | 1 + src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 57 ++++++++++++++++++++-- 6 files changed, 77 insertions(+), 16 deletions(-) diff --git a/src/Runner.Common/JobServer.cs b/src/Runner.Common/JobServer.cs index 42feca70e77..a4add9b0633 100644 --- a/src/Runner.Common/JobServer.cs +++ b/src/Runner.Common/JobServer.cs @@ -31,7 +31,7 @@ public interface IJobServer : IRunnerService, IAsyncDisposable Task AppendTimelineRecordFeedAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, Guid stepId, IList lines, long? startLine, CancellationToken cancellationToken); Task CreateAttachmentAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, String type, String name, Stream uploadStream, CancellationToken cancellationToken); Task CreateStepSymmaryAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken); - Task CreateResultsLogAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken); + Task CreateResultsLogAsync(string planId, string jobId, string stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken); Task CreateLogAsync(Guid scopeIdentifier, string hubName, Guid planId, TaskLog log, CancellationToken cancellationToken); Task CreateTimelineAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, CancellationToken cancellationToken); Task> UpdateTimelineRecordsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, IEnumerable records, CancellationToken cancellationToken); @@ -326,11 +326,11 @@ public Task CreateStepSymmaryAsync(string planId, string jobId, string stepId, s throw new InvalidOperationException("Results client is not initialized."); } - public Task CreateResultsLogAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken) + public Task CreateResultsLogAsync(string planId, string jobId, string stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken) { if (_resultsClient != null) { - return _resultsClient.UploadResultsLogAsync(planId, jobId, stepId, file, cancellationToken: cancellationToken); + return _resultsClient.UploadResultsLogAsync(planId, jobId, stepId, file, finalize, firstBlock, cancellationToken: cancellationToken); } throw new InvalidOperationException("Results client is not initialized."); } diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 3c57a019569..1326b750639 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -20,7 +20,7 @@ public interface IJobServerQueue : IRunnerService, IThrottlingReporter void Start(Pipelines.AgentJobRequestMessage jobRequest); void QueueWebConsoleLine(Guid stepRecordId, string line, long? lineNumber = null); void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, string name, string path, bool deleteSource); - void QueueResultsUpload(Guid stepRecordId, string name, string path, string type, bool deleteSource, bool finalize); + void QueueResultsUpload(Guid recordId, string name, string path, string type, bool deleteSource, bool finalize, bool firstBlock); void QueueTimelineRecordUpdate(Guid timelineId, TimelineRecord timelineRecord); } @@ -248,8 +248,14 @@ public void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, } - public void QueueResultsUpload(Guid stepRecordId, string name, string path, string type, bool deleteSource, bool finalize) + public void QueueResultsUpload(Guid recordId, string name, string path, string type, bool deleteSource, bool finalize, bool firstBlock) { + if (recordId == _jobTimelineRecordId) + { + Trace.Verbose("Skipping job log {0} for record {1}", path, recordId); + return; + } + // all parameter not null, file path exist. var newFile = new ResultsUploadFileInfo() { @@ -258,11 +264,12 @@ public void QueueResultsUpload(Guid stepRecordId, string name, string path, stri Type = type, PlanId = _planId.ToString(), JobId = _jobTimelineRecordId.ToString(), - StepId = stepRecordId.ToString(), - DeleteSource = deleteSource + StepId = recordId.ToString(), + DeleteSource = deleteSource, + Finalize = finalize, }; - Trace.Verbose("Enqueue results file upload queue: file '{0}' attach to job {1} step {2}", newFile.Path, _jobTimelineRecordId, stepRecordId); + Trace.Verbose("Enqueue results file upload queue: file '{0}' attach to job {1} step {2}", newFile.Path, _jobTimelineRecordId, recordId); _resultsFileUploadQueue.Enqueue(newFile); } @@ -489,7 +496,7 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) if (String.Equals(file.Type, ChecksAttachmentType.StepSummary, StringComparison.OrdinalIgnoreCase)) { await UploadSummaryFile(file); - } else if (String.Equals(file.Type, CoreAttachmentType.Log, StringComparison.OrdinalIgnoreCase)) + } else if (String.Equals(file.Type, CoreAttachmentType.ResultsLog, StringComparison.OrdinalIgnoreCase)) { Trace.Info($"Got a log file to send to results service."); await UploadResultsStepLogFile(file); @@ -842,7 +849,7 @@ private async Task UploadResultsStepLogFile(ResultsUploadFileInfo file) { Trace.Info($"Starting to step log file to results service {file.Name}, {file.Path}"); var cancellationTokenSource = new CancellationTokenSource(); - await _jobServer.CreateResultsLogAsync(file.PlanId, file.JobId, file.StepId, file.Path, cancellationTokenSource.Token); + await _jobServer.CreateResultsLogAsync(file.PlanId, file.JobId, file.StepId, file.Path, file.Finalize, file.FirstBlock, cancellationTokenSource.Token); uploadSucceed = true; } @@ -889,6 +896,8 @@ internal class ResultsUploadFileInfo public string JobId { get; set; } public string StepId { get; set; } public bool DeleteSource { get; set; } + public bool Finalize { get; set; } + public bool FirstBlock { get; set; } } diff --git a/src/Runner.Common/Logging.cs b/src/Runner.Common/Logging.cs index 5d6ffad4a80..9131261641e 100644 --- a/src/Runner.Common/Logging.cs +++ b/src/Runner.Common/Logging.cs @@ -147,7 +147,7 @@ private void NewBlock() { EndBlock(false); _blockByteCount = 0; - _resultsDataFileName = Path.Combine(_resultsBlockFolder, $"{_timelineId}_{_timelineRecordId}_{++_blockCount}.block"); + _resultsDataFileName = Path.Combine(_resultsBlockFolder, $"{_timelineId}_{_timelineRecordId}.{++_blockCount}"); _resultsBlockData = new FileStream(_resultsDataFileName, FileMode.CreateNew, FileAccess.ReadWrite, FileShare.ReadWrite); _resultsBlockWriter = new StreamWriter(_resultsBlockData, System.Text.Encoding.UTF8); } @@ -161,7 +161,7 @@ private void EndBlock(bool finalize) _resultsBlockWriter.Dispose(); _resultsBlockWriter = null; _resultsBlockData = null; - _jobServerQueue.QueueResultsUpload(_timelineRecordId, "ResultsLog", _resultsDataFileName, "Results.Core.Log", true, finalize); + _jobServerQueue.QueueResultsUpload(_timelineRecordId, "ResultsLog", _resultsDataFileName, "Results.Core.Log", true, finalize, _resultsDataFileName.EndsWith(".0")); } } } diff --git a/src/Runner.Worker/ExecutionContext.cs b/src/Runner.Worker/ExecutionContext.cs index e4650ccf885..63a97b5a6cc 100644 --- a/src/Runner.Worker/ExecutionContext.cs +++ b/src/Runner.Worker/ExecutionContext.cs @@ -857,7 +857,7 @@ public void QueueSummaryFile(string name, string filePath, Guid stepRecordId) throw new FileNotFoundException($"Can't upload (name:{name}) file: {filePath}. File does not exist."); } - _jobServerQueue.QueueResultsUpload(stepRecordId, name, filePath, ChecksAttachmentType.StepSummary, deleteSource: false, finalize: false); + _jobServerQueue.QueueResultsUpload(stepRecordId, name, filePath, ChecksAttachmentType.StepSummary, deleteSource: false, finalize: false, firstBlock: false); } // Add OnMatcherChanged diff --git a/src/Sdk/DTWebApi/WebApi/TaskAttachment.cs b/src/Sdk/DTWebApi/WebApi/TaskAttachment.cs index 17027d1261a..572b1f6426b 100644 --- a/src/Sdk/DTWebApi/WebApi/TaskAttachment.cs +++ b/src/Sdk/DTWebApi/WebApi/TaskAttachment.cs @@ -100,6 +100,7 @@ public class CoreAttachmentType public static readonly String Summary = "DistributedTask.Core.Summary"; public static readonly String FileAttachment = "DistributedTask.Core.FileAttachment"; public static readonly String DiagnosticLog = "DistributedTask.Core.DiagnosticLog"; + public static readonly String ResultsLog = "Results.Core.Log"; } [GenerateAllConstants] diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index e5bee89251e..a8aee166ad4 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -169,6 +169,49 @@ private async Task UploadFileAsync(string url, string blobS } } + private async Task CreateAppendFileAsync(string url, string blobStorageType, CancellationToken cancellationToken) + { + var request = new HttpRequestMessage(HttpMethod.Put, url); + if (blobStorageType == BlobStorageTypes.AzureBlobStorage) + { + request.Content.Headers.Add("x-ms-blob-type", "AppendBlob"); + request.Content.Headers.Add("Content-Length", "0"); + } + + using (var response = await SendAsync(request, HttpCompletionOption.ResponseHeadersRead, userState: null, cancellationToken)) + { + if (!response.IsSuccessStatusCode) + { + throw new Exception($"Failed to create append file, status code: {response.StatusCode}, reason: {response.ReasonPhrase}"); + } + return response; + } + } + + private async Task UploadAppendFileAsync(string url, string blobStorageType, FileStream file, bool finalize, CancellationToken cancellationToken) + { + // Upload the file to the url + var request = new HttpRequestMessage(HttpMethod.Put, url) + { + Content = new StreamContent(file) + }; + + if (blobStorageType == BlobStorageTypes.AzureBlobStorage) + { + request.Content.Headers.Add("x-ms-blob-type", "AppendBlob"); + request.Content.Headers.Add("x-ms-blob-sealed", finalize.ToString()); + } + + using (var response = await SendAsync(request, HttpCompletionOption.ResponseHeadersRead, userState: null, cancellationToken)) + { + if (!response.IsSuccessStatusCode) + { + throw new Exception($"Failed to upload append file, status code: {response.StatusCode}, reason: {response.ReasonPhrase}"); + } + return response; + } + } + // Handle file upload for step summary public async Task UploadStepSummaryAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken) { @@ -196,8 +239,8 @@ public async Task UploadStepSummaryAsync(string planId, string jobId, string ste await StepSummaryUploadCompleteAsync(planId, jobId, stepId, fileSize, cancellationToken); } - // Handle file upload for step summary - public async Task UploadResultsLogAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken) + // Handle file upload for step log + public async Task UploadResultsLogAsync(string planId, string jobId, string stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken) { // Get the upload url var uploadUrlResponse = await GetStepLogUploadUrlAsync(planId, jobId, stepId, cancellationToken); @@ -210,9 +253,17 @@ public async Task UploadResultsLogAsync(string planId, string jobId, string step var fileSize = new FileInfo(file).Length; // Upload the file + // if this is the first block, we need to create + // otherwise we just need to append + // if this is the last block we also need to seal + if (firstBlock) + { + var response = await CreateAppendFileAsync(uploadUrlResponse.BlobUrl, uploadUrlResponse.BlobStorageType, cancellationToken); + } + using (var fileStream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, true)) { - var response = await UploadFileAsync(uploadUrlResponse.BlobUrl, uploadUrlResponse.BlobStorageType, fileStream, cancellationToken); + var response = await UploadAppendFileAsync(uploadUrlResponse.BlobUrl, uploadUrlResponse.BlobStorageType, fileStream, finalize, cancellationToken); } // Send step summary upload complete message From 6bc23ca6f070282c102ad9daaf2d870c80296be3 Mon Sep 17 00:00:00 2001 From: Brittany Ellich Date: Tue, 31 Jan 2023 10:32:49 -0800 Subject: [PATCH 05/14] Modifications for azurite upload --- src/Runner.Common/JobServerQueue.cs | 4 +--- src/Runner.Common/Logging.cs | 2 +- src/Sdk/WebApi/WebApi/Contracts.cs | 2 +- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 22 ++++++++++++---------- 4 files changed, 15 insertions(+), 15 deletions(-) diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 1326b750639..601d1d7d66e 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -267,6 +267,7 @@ public void QueueResultsUpload(Guid recordId, string name, string path, string t StepId = recordId.ToString(), DeleteSource = deleteSource, Finalize = finalize, + FirstBlock = firstBlock, }; Trace.Verbose("Enqueue results file upload queue: file '{0}' attach to job {1} step {2}", newFile.Path, _jobTimelineRecordId, recordId); @@ -517,9 +518,6 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) Trace.Info("Catch exception during file upload to results, keep going since the process is best effort."); Trace.Error(ex); - } - finally - { errorCount++; } } diff --git a/src/Runner.Common/Logging.cs b/src/Runner.Common/Logging.cs index 9131261641e..d446d0a9e66 100644 --- a/src/Runner.Common/Logging.cs +++ b/src/Runner.Common/Logging.cs @@ -161,7 +161,7 @@ private void EndBlock(bool finalize) _resultsBlockWriter.Dispose(); _resultsBlockWriter = null; _resultsBlockData = null; - _jobServerQueue.QueueResultsUpload(_timelineRecordId, "ResultsLog", _resultsDataFileName, "Results.Core.Log", true, finalize, _resultsDataFileName.EndsWith(".0")); + _jobServerQueue.QueueResultsUpload(_timelineRecordId, "ResultsLog", _resultsDataFileName, "Results.Core.Log", true, finalize, _resultsDataFileName.EndsWith(".1")); } } } diff --git a/src/Sdk/WebApi/WebApi/Contracts.cs b/src/Sdk/WebApi/WebApi/Contracts.cs index 42b1041e72b..a23de2e83c6 100644 --- a/src/Sdk/WebApi/WebApi/Contracts.cs +++ b/src/Sdk/WebApi/WebApi/Contracts.cs @@ -45,7 +45,7 @@ public class GetSignedStepLogsURLRequest public class GetSignedStepLogsURLResponse { [DataMember] - public string BlobUrl; + public string SummaryUrl; [DataMember] public long SoftSizeLimit; [DataMember] diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index a8aee166ad4..b2ae3a3a798 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -171,7 +171,10 @@ private async Task UploadFileAsync(string url, string blobS private async Task CreateAppendFileAsync(string url, string blobStorageType, CancellationToken cancellationToken) { - var request = new HttpRequestMessage(HttpMethod.Put, url); + var request = new HttpRequestMessage(HttpMethod.Put, url) + { + Content = new StringContent("") + }; if (blobStorageType == BlobStorageTypes.AzureBlobStorage) { request.Content.Headers.Add("x-ms-blob-type", "AppendBlob"); @@ -188,17 +191,19 @@ private async Task CreateAppendFileAsync(string url, string } } - private async Task UploadAppendFileAsync(string url, string blobStorageType, FileStream file, bool finalize, CancellationToken cancellationToken) + private async Task UploadAppendFileAsync(string url, string blobStorageType, FileStream file, bool finalize, long fileSize, CancellationToken cancellationToken) { + var comp = finalize ? "&comp=appendblock&seal=true" : "&comp=appendblock"; // Upload the file to the url - var request = new HttpRequestMessage(HttpMethod.Put, url) + var request = new HttpRequestMessage(HttpMethod.Put, url + comp) { Content = new StreamContent(file) }; if (blobStorageType == BlobStorageTypes.AzureBlobStorage) { - request.Content.Headers.Add("x-ms-blob-type", "AppendBlob"); + // request.Content.Headers.Add("x-ms-blob-type", "AppendBlock"); + request.Content.Headers.Add("Content-Length", fileSize.ToString()); request.Content.Headers.Add("x-ms-blob-sealed", finalize.ToString()); } @@ -206,7 +211,7 @@ private async Task UploadAppendFileAsync(string url, string { if (!response.IsSuccessStatusCode) { - throw new Exception($"Failed to upload append file, status code: {response.StatusCode}, reason: {response.ReasonPhrase}"); + throw new Exception($"Failed to upload append file, status code: {response.StatusCode}, reason: {response.ReasonPhrase}, object: {response}, fileSize: {fileSize}"); } return response; } @@ -253,17 +258,14 @@ public async Task UploadResultsLogAsync(string planId, string jobId, string step var fileSize = new FileInfo(file).Length; // Upload the file - // if this is the first block, we need to create - // otherwise we just need to append - // if this is the last block we also need to seal if (firstBlock) { - var response = await CreateAppendFileAsync(uploadUrlResponse.BlobUrl, uploadUrlResponse.BlobStorageType, cancellationToken); + await CreateAppendFileAsync(uploadUrlResponse.SummaryUrl, uploadUrlResponse.BlobStorageType, cancellationToken); } using (var fileStream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, true)) { - var response = await UploadAppendFileAsync(uploadUrlResponse.BlobUrl, uploadUrlResponse.BlobStorageType, fileStream, finalize, cancellationToken); + var response = await UploadAppendFileAsync(uploadUrlResponse.SummaryUrl, uploadUrlResponse.BlobStorageType, fileStream, finalize, fileSize, cancellationToken); } // Send step summary upload complete message From 86a22457372555ccaccdddf9b0d5c9394590f65b Mon Sep 17 00:00:00 2001 From: Brittany Ellich Date: Tue, 31 Jan 2023 10:38:14 -0800 Subject: [PATCH 06/14] Only call upload complete on final section and remove size --- src/Sdk/WebApi/WebApi/Contracts.cs | 2 -- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 12 +++++++----- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/Sdk/WebApi/WebApi/Contracts.cs b/src/Sdk/WebApi/WebApi/Contracts.cs index a23de2e83c6..16ec391498c 100644 --- a/src/Sdk/WebApi/WebApi/Contracts.cs +++ b/src/Sdk/WebApi/WebApi/Contracts.cs @@ -87,8 +87,6 @@ public class StepLogsMetadataCreate [DataMember] public string WorkflowJobRunBackendId; [DataMember] - public long Size; - [DataMember] public string UploadedAt; } diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index b2ae3a3a798..1a67dfd0554 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -112,7 +112,7 @@ private async Task StepSummaryUploadCompleteAsync(string planId, string jobId, s } } - private async Task StepLogUploadCompleteAsync(string planId, string jobId, string stepId, long size, CancellationToken cancellationToken) + private async Task StepLogUploadCompleteAsync(string planId, string jobId, string stepId, CancellationToken cancellationToken) { var timestamp = DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss.fffK"); var request = new StepLogsMetadataCreate() @@ -120,7 +120,6 @@ private async Task StepLogUploadCompleteAsync(string planId, string jobId, strin WorkflowJobRunBackendId= jobId, WorkflowRunBackendId= planId, StepBackendId = stepId, - Size = size, UploadedAt = timestamp }; @@ -139,7 +138,7 @@ private async Task StepLogUploadCompleteAsync(string planId, string jobId, strin var jsonResponse = await ReadJsonContentAsync(response, cancellationToken); if (!jsonResponse.Ok) { - throw new Exception($"Failed to mark step log upload as complete, status code: {response.StatusCode}, ok: {jsonResponse.Ok}, size: {size}, timestamp: {timestamp}"); + throw new Exception($"Failed to mark step log upload as complete, status code: {response.StatusCode}, ok: {jsonResponse.Ok}, timestamp: {timestamp}"); } } } @@ -268,8 +267,11 @@ public async Task UploadResultsLogAsync(string planId, string jobId, string step var response = await UploadAppendFileAsync(uploadUrlResponse.SummaryUrl, uploadUrlResponse.BlobStorageType, fileStream, finalize, fileSize, cancellationToken); } - // Send step summary upload complete message - await StepLogUploadCompleteAsync(planId, jobId, stepId, fileSize, cancellationToken); + if (finalize) + { + // Send step summary upload complete message + await StepLogUploadCompleteAsync(planId, jobId, stepId, cancellationToken); + } } private MediaTypeFormatter m_formatter; From 42d7fe9c2cab13c83a9bddae75ddfe1e848e8be6 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Tue, 31 Jan 2023 13:51:23 -0500 Subject: [PATCH 07/14] Make method specific to step log so we can support job log later --- src/Runner.Common/JobServer.cs | 10 +++++----- src/Runner.Common/JobServerQueue.cs | 16 +++++++++------- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 20 ++++++++++---------- 3 files changed, 24 insertions(+), 22 deletions(-) diff --git a/src/Runner.Common/JobServer.cs b/src/Runner.Common/JobServer.cs index a4add9b0633..18f4b2bcf24 100644 --- a/src/Runner.Common/JobServer.cs +++ b/src/Runner.Common/JobServer.cs @@ -30,8 +30,8 @@ public interface IJobServer : IRunnerService, IAsyncDisposable Task AppendLogContentAsync(Guid scopeIdentifier, string hubName, Guid planId, int logId, Stream uploadStream, CancellationToken cancellationToken); Task AppendTimelineRecordFeedAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, Guid stepId, IList lines, long? startLine, CancellationToken cancellationToken); Task CreateAttachmentAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, String type, String name, Stream uploadStream, CancellationToken cancellationToken); - Task CreateStepSymmaryAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken); - Task CreateResultsLogAsync(string planId, string jobId, string stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken); + Task CreateStepSymmaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken); + Task CreateResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken); Task CreateLogAsync(Guid scopeIdentifier, string hubName, Guid planId, TaskLog log, CancellationToken cancellationToken); Task CreateTimelineAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, CancellationToken cancellationToken); Task> UpdateTimelineRecordsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, IEnumerable records, CancellationToken cancellationToken); @@ -317,7 +317,7 @@ public Task CreateAttachmentAsync(Guid scopeIdentifier, string h return _taskClient.CreateAttachmentAsync(scopeIdentifier, hubName, planId, timelineId, timelineRecordId, type, name, uploadStream, cancellationToken: cancellationToken); } - public Task CreateStepSymmaryAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken) + public Task CreateStepSymmaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken) { if (_resultsClient != null) { @@ -326,11 +326,11 @@ public Task CreateStepSymmaryAsync(string planId, string jobId, string stepId, s throw new InvalidOperationException("Results client is not initialized."); } - public Task CreateResultsLogAsync(string planId, string jobId, string stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken) + public Task CreateResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken) { if (_resultsClient != null) { - return _resultsClient.UploadResultsLogAsync(planId, jobId, stepId, file, finalize, firstBlock, cancellationToken: cancellationToken); + return _resultsClient.UploadResultsStepLogAsync(planId, jobId, stepId, file, finalize, firstBlock, cancellationToken: cancellationToken); } throw new InvalidOperationException("Results client is not initialized."); } diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 601d1d7d66e..5bae7c957e6 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -264,7 +264,7 @@ public void QueueResultsUpload(Guid recordId, string name, string path, string t Type = type, PlanId = _planId.ToString(), JobId = _jobTimelineRecordId.ToString(), - StepId = recordId.ToString(), + RecordId = recordId, DeleteSource = deleteSource, Finalize = finalize, FirstBlock = firstBlock, @@ -499,10 +499,12 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) await UploadSummaryFile(file); } else if (String.Equals(file.Type, CoreAttachmentType.ResultsLog, StringComparison.OrdinalIgnoreCase)) { - Trace.Info($"Got a log file to send to results service."); - await UploadResultsStepLogFile(file); + if (file.RecordId != _jobTimelineRecordId) + { + Trace.Info($"Got a step log file to send to results service."); + await UploadResultsStepLogFile(file); + } } - } catch (Exception ex) { @@ -819,7 +821,7 @@ private async Task UploadSummaryFile(ResultsUploadFileInfo file) // Upload the step summary Trace.Info($"Starting to upload summary file to results service {file.Name}, {file.Path}"); var cancellationTokenSource = new CancellationTokenSource(); - await _jobServer.CreateStepSymmaryAsync(file.PlanId, file.JobId, file.StepId, file.Path, cancellationTokenSource.Token); + await _jobServer.CreateStepSymmaryAsync(file.PlanId, file.JobId, file.RecordId, file.Path, cancellationTokenSource.Token); uploadSucceed = true; } @@ -847,7 +849,7 @@ private async Task UploadResultsStepLogFile(ResultsUploadFileInfo file) { Trace.Info($"Starting to step log file to results service {file.Name}, {file.Path}"); var cancellationTokenSource = new CancellationTokenSource(); - await _jobServer.CreateResultsLogAsync(file.PlanId, file.JobId, file.StepId, file.Path, file.Finalize, file.FirstBlock, cancellationTokenSource.Token); + await _jobServer.CreateResultsStepLogAsync(file.PlanId, file.JobId, file.RecordId, file.Path, file.Finalize, file.FirstBlock, cancellationTokenSource.Token); uploadSucceed = true; } @@ -892,7 +894,7 @@ internal class ResultsUploadFileInfo public string Path { get; set; } public string PlanId { get; set; } public string JobId { get; set; } - public string StepId { get; set; } + public Guid RecordId { get; set; } public bool DeleteSource { get; set; } public bool Finalize { get; set; } public bool FirstBlock { get; set; } diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index 1a67dfd0554..51e4d6d5753 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -24,13 +24,13 @@ public ResultsHttpClient( m_formatter = new JsonMediaTypeFormatter(); } - public async Task GetStepSummaryUploadUrlAsync(string planId, string jobId, string stepId, CancellationToken cancellationToken) + public async Task GetStepSummaryUploadUrlAsync(string planId, string jobId, Guid stepId, CancellationToken cancellationToken) { var request = new GetSignedStepSummaryURLRequest() { WorkflowJobRunBackendId= jobId, WorkflowRunBackendId= planId, - StepBackendId= stepId + StepBackendId= stepId.ToString() }; var stepSummaryUploadRequest = new Uri(m_resultsServiceUrl, "twirp/results.services.receiver.Receiver/GetStepSummarySignedBlobURL"); @@ -51,13 +51,13 @@ public async Task GetStepSummaryUploadUrlAsync( } } - public async Task GetStepLogUploadUrlAsync(string planId, string jobId, string stepId, CancellationToken cancellationToken) + public async Task GetStepLogUploadUrlAsync(string planId, string jobId, Guid stepId, CancellationToken cancellationToken) { var request = new GetSignedStepLogsURLRequest() { WorkflowJobRunBackendId= jobId, WorkflowRunBackendId= planId, - StepBackendId= stepId + StepBackendId= stepId.ToString(), }; var stepLogsUploadRequest = new Uri(m_resultsServiceUrl, "twirp/results.services.receiver.Receiver/GetStepStepLogsSignedBlobURL"); @@ -78,14 +78,14 @@ public async Task GetStepLogUploadUrlAsync(string } } - private async Task StepSummaryUploadCompleteAsync(string planId, string jobId, string stepId, long size, CancellationToken cancellationToken) + private async Task StepSummaryUploadCompleteAsync(string planId, string jobId, Guid stepId, long size, CancellationToken cancellationToken) { var timestamp = DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss.fffK"); var request = new StepSummaryMetadataCreate() { WorkflowJobRunBackendId= jobId, WorkflowRunBackendId= planId, - StepBackendId = stepId, + StepBackendId = stepId.ToString(), Size = size, UploadedAt = timestamp }; @@ -112,14 +112,14 @@ private async Task StepSummaryUploadCompleteAsync(string planId, string jobId, s } } - private async Task StepLogUploadCompleteAsync(string planId, string jobId, string stepId, CancellationToken cancellationToken) + private async Task StepLogUploadCompleteAsync(string planId, string jobId, Guid stepId, CancellationToken cancellationToken) { var timestamp = DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss.fffK"); var request = new StepLogsMetadataCreate() { WorkflowJobRunBackendId= jobId, WorkflowRunBackendId= planId, - StepBackendId = stepId, + StepBackendId = stepId.ToString(), UploadedAt = timestamp }; @@ -217,7 +217,7 @@ private async Task UploadAppendFileAsync(string url, string } // Handle file upload for step summary - public async Task UploadStepSummaryAsync(string planId, string jobId, string stepId, string file, CancellationToken cancellationToken) + public async Task UploadStepSummaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken) { // Get the upload url var uploadUrlResponse = await GetStepSummaryUploadUrlAsync(planId, jobId, stepId, cancellationToken); @@ -244,7 +244,7 @@ public async Task UploadStepSummaryAsync(string planId, string jobId, string ste } // Handle file upload for step log - public async Task UploadResultsLogAsync(string planId, string jobId, string stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken) + public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken) { // Get the upload url var uploadUrlResponse = await GetStepLogUploadUrlAsync(planId, jobId, stepId, cancellationToken); From e2759778f63b1f522888ddc3b3fda6ce51c8b958 Mon Sep 17 00:00:00 2001 From: Brittany Ellich Date: Tue, 31 Jan 2023 14:17:00 -0800 Subject: [PATCH 08/14] Change contract for results --- src/Sdk/WebApi/WebApi/Contracts.cs | 2 +- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/Sdk/WebApi/WebApi/Contracts.cs b/src/Sdk/WebApi/WebApi/Contracts.cs index 16ec391498c..8f435f01728 100644 --- a/src/Sdk/WebApi/WebApi/Contracts.cs +++ b/src/Sdk/WebApi/WebApi/Contracts.cs @@ -45,7 +45,7 @@ public class GetSignedStepLogsURLRequest public class GetSignedStepLogsURLResponse { [DataMember] - public string SummaryUrl; + public string LogsUrl; [DataMember] public long SoftSizeLimit; [DataMember] diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index 51e4d6d5753..9e8e356772f 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -259,12 +259,12 @@ public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid st // Upload the file if (firstBlock) { - await CreateAppendFileAsync(uploadUrlResponse.SummaryUrl, uploadUrlResponse.BlobStorageType, cancellationToken); + await CreateAppendFileAsync(uploadUrlResponse.LogsUrl, uploadUrlResponse.BlobStorageType, cancellationToken); } using (var fileStream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, true)) { - var response = await UploadAppendFileAsync(uploadUrlResponse.SummaryUrl, uploadUrlResponse.BlobStorageType, fileStream, finalize, fileSize, cancellationToken); + var response = await UploadAppendFileAsync(uploadUrlResponse.LogsUrl, uploadUrlResponse.BlobStorageType, fileStream, finalize, fileSize, cancellationToken); } if (finalize) From 792c8adeaaa70e8f959fbcb357295fbc0f70adcf Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Wed, 1 Feb 2023 10:03:21 -0500 Subject: [PATCH 09/14] Add totalline count to the result log upload file --- src/Runner.Common/JobServerQueue.cs | 6 ++++-- src/Runner.Common/Logging.cs | 2 +- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 5bae7c957e6..238b669ac94 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -20,7 +20,7 @@ public interface IJobServerQueue : IRunnerService, IThrottlingReporter void Start(Pipelines.AgentJobRequestMessage jobRequest); void QueueWebConsoleLine(Guid stepRecordId, string line, long? lineNumber = null); void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, string name, string path, bool deleteSource); - void QueueResultsUpload(Guid recordId, string name, string path, string type, bool deleteSource, bool finalize, bool firstBlock); + void QueueResultsUpload(Guid timelineRecordId, string name, string path, string type, bool deleteSource, bool finalize, bool firstBlock, long totalLines = 0); void QueueTimelineRecordUpdate(Guid timelineId, TimelineRecord timelineRecord); } @@ -248,7 +248,7 @@ public void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, } - public void QueueResultsUpload(Guid recordId, string name, string path, string type, bool deleteSource, bool finalize, bool firstBlock) + public void QueueResultsUpload(Guid recordId, string name, string path, string type, bool deleteSource, bool finalize, bool firstBlock, long totalLines) { if (recordId == _jobTimelineRecordId) { @@ -268,6 +268,7 @@ public void QueueResultsUpload(Guid recordId, string name, string path, string t DeleteSource = deleteSource, Finalize = finalize, FirstBlock = firstBlock, + TotalLines = totalLines, }; Trace.Verbose("Enqueue results file upload queue: file '{0}' attach to job {1} step {2}", newFile.Path, _jobTimelineRecordId, recordId); @@ -898,6 +899,7 @@ internal class ResultsUploadFileInfo public bool DeleteSource { get; set; } public bool Finalize { get; set; } public bool FirstBlock { get; set; } + public long TotalLines { get; set; } } diff --git a/src/Runner.Common/Logging.cs b/src/Runner.Common/Logging.cs index d446d0a9e66..8cf23749d57 100644 --- a/src/Runner.Common/Logging.cs +++ b/src/Runner.Common/Logging.cs @@ -161,7 +161,7 @@ private void EndBlock(bool finalize) _resultsBlockWriter.Dispose(); _resultsBlockWriter = null; _resultsBlockData = null; - _jobServerQueue.QueueResultsUpload(_timelineRecordId, "ResultsLog", _resultsDataFileName, "Results.Core.Log", true, finalize, _resultsDataFileName.EndsWith(".1")); + _jobServerQueue.QueueResultsUpload(_timelineRecordId, "ResultsLog", _resultsDataFileName, "Results.Core.Log", deleteSource: true, finalize, firstBlock: _resultsDataFileName.EndsWith(".1"), totalLines: _totalLines); } } } From f09318f126361a83bd8f3e20d93a9021c75a8add Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Mon, 6 Feb 2023 08:00:19 -0800 Subject: [PATCH 10/14] Actually pass lineCount to Results Service --- src/Runner.Common/JobServer.cs | 6 +++--- src/Runner.Common/JobServerQueue.cs | 2 +- src/Sdk/WebApi/WebApi/Contracts.cs | 2 ++ src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 11 ++++++----- 4 files changed, 12 insertions(+), 9 deletions(-) diff --git a/src/Runner.Common/JobServer.cs b/src/Runner.Common/JobServer.cs index 18f4b2bcf24..7a2f96ae077 100644 --- a/src/Runner.Common/JobServer.cs +++ b/src/Runner.Common/JobServer.cs @@ -31,7 +31,7 @@ public interface IJobServer : IRunnerService, IAsyncDisposable Task AppendTimelineRecordFeedAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, Guid stepId, IList lines, long? startLine, CancellationToken cancellationToken); Task CreateAttachmentAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, String type, String name, Stream uploadStream, CancellationToken cancellationToken); Task CreateStepSymmaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken); - Task CreateResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken); + Task CreateResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, long lineCount, CancellationToken cancellationToken); Task CreateLogAsync(Guid scopeIdentifier, string hubName, Guid planId, TaskLog log, CancellationToken cancellationToken); Task CreateTimelineAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, CancellationToken cancellationToken); Task> UpdateTimelineRecordsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, IEnumerable records, CancellationToken cancellationToken); @@ -326,11 +326,11 @@ public Task CreateStepSymmaryAsync(string planId, string jobId, Guid stepId, str throw new InvalidOperationException("Results client is not initialized."); } - public Task CreateResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken) + public Task CreateResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, long lineCount, CancellationToken cancellationToken) { if (_resultsClient != null) { - return _resultsClient.UploadResultsStepLogAsync(planId, jobId, stepId, file, finalize, firstBlock, cancellationToken: cancellationToken); + return _resultsClient.UploadResultsStepLogAsync(planId, jobId, stepId, file, finalize, firstBlock, lineCount, cancellationToken: cancellationToken); } throw new InvalidOperationException("Results client is not initialized."); } diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 238b669ac94..56305890bf6 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -850,7 +850,7 @@ private async Task UploadResultsStepLogFile(ResultsUploadFileInfo file) { Trace.Info($"Starting to step log file to results service {file.Name}, {file.Path}"); var cancellationTokenSource = new CancellationTokenSource(); - await _jobServer.CreateResultsStepLogAsync(file.PlanId, file.JobId, file.RecordId, file.Path, file.Finalize, file.FirstBlock, cancellationTokenSource.Token); + await _jobServer.CreateResultsStepLogAsync(file.PlanId, file.JobId, file.RecordId, file.Path, file.Finalize, file.FirstBlock, file.TotalLines, cancellationTokenSource.Token); uploadSucceed = true; } diff --git a/src/Sdk/WebApi/WebApi/Contracts.cs b/src/Sdk/WebApi/WebApi/Contracts.cs index 8f435f01728..bc6361d62bc 100644 --- a/src/Sdk/WebApi/WebApi/Contracts.cs +++ b/src/Sdk/WebApi/WebApi/Contracts.cs @@ -88,6 +88,8 @@ public class StepLogsMetadataCreate public string WorkflowJobRunBackendId; [DataMember] public string UploadedAt; + [DataMember] + public long LineCount; } [DataContract] diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index 9e8e356772f..f6efecbc4c8 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -112,7 +112,7 @@ private async Task StepSummaryUploadCompleteAsync(string planId, string jobId, G } } - private async Task StepLogUploadCompleteAsync(string planId, string jobId, Guid stepId, CancellationToken cancellationToken) + private async Task StepLogUploadCompleteAsync(string planId, string jobId, Guid stepId, long lineCount, CancellationToken cancellationToken) { var timestamp = DateTime.UtcNow.ToString("yyyy-MM-dd'T'HH:mm:ss.fffK"); var request = new StepLogsMetadataCreate() @@ -120,7 +120,8 @@ private async Task StepLogUploadCompleteAsync(string planId, string jobId, Guid WorkflowJobRunBackendId= jobId, WorkflowRunBackendId= planId, StepBackendId = stepId.ToString(), - UploadedAt = timestamp + UploadedAt = timestamp, + LineCount = lineCount, }; var stepLogsUploadCompleteRequest = new Uri(m_resultsServiceUrl, "twirp/results.services.receiver.Receiver/CreateStepLogsMetadata"); @@ -244,7 +245,7 @@ public async Task UploadStepSummaryAsync(string planId, string jobId, Guid stepI } // Handle file upload for step log - public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, CancellationToken cancellationToken) + public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, long lineCount, CancellationToken cancellationToken) { // Get the upload url var uploadUrlResponse = await GetStepLogUploadUrlAsync(planId, jobId, stepId, cancellationToken); @@ -269,8 +270,8 @@ public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid st if (finalize) { - // Send step summary upload complete message - await StepLogUploadCompleteAsync(planId, jobId, stepId, cancellationToken); + // Send step log upload complete message + await StepLogUploadCompleteAsync(planId, jobId, stepId, lineCount, cancellationToken); } } From 69f7fcbc117f661a7983875d9ba752d9d4bfca07 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Mon, 6 Feb 2023 14:34:28 -0800 Subject: [PATCH 11/14] Fix typos --- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index f6efecbc4c8..31cc4b68f69 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -60,7 +60,7 @@ public async Task GetStepLogUploadUrlAsync(string StepBackendId= stepId.ToString(), }; - var stepLogsUploadRequest = new Uri(m_resultsServiceUrl, "twirp/results.services.receiver.Receiver/GetStepStepLogsSignedBlobURL"); + var stepLogsUploadRequest = new Uri(m_resultsServiceUrl, "twirp/results.services.receiver.Receiver/GetStepLogsSignedBlobURL"); using (HttpRequestMessage requestMessage = new HttpRequestMessage(HttpMethod.Post, stepLogsUploadRequest)) { @@ -249,7 +249,7 @@ public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid st { // Get the upload url var uploadUrlResponse = await GetStepLogUploadUrlAsync(planId, jobId, stepId, cancellationToken); - if (uploadUrlResponse == null) + if (uploadUrlResponse == null || uploadUrlResponse.LogsUrl == null) { throw new Exception("Failed to get step log upload url"); } From 421715af65765fdae79ba0f78a526bb503e8ff60 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Tue, 7 Feb 2023 09:33:13 -0500 Subject: [PATCH 12/14] Code cleanup --- src/Runner.Common/JobServerQueue.cs | 26 +++------------------- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 5 +++-- 2 files changed, 6 insertions(+), 25 deletions(-) diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 56305890bf6..cb3f49f5d80 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -226,26 +226,8 @@ public void QueueFileUpload(Guid timelineId, Guid timelineRecordId, string type, DeleteSource = deleteSource }; - // Dual upload the same files to Results Service - if (String.Equals(type, CoreAttachmentType.Log, StringComparison.OrdinalIgnoreCase)) - { - newFile.Type = CoreAttachmentType.Log; - Trace.Verbose("Log was sent from logging file", newFile.Path, timelineRecordId); - // jobRequest.Variables.TryGetValue("system.github.results_endpoint", out VariableValue resultsEndpointVariable); - // context.Global.Variables.TryGetValue("system.github.results_endpoint", out string resultsReceiverEndpoint); - // if (resultsReceiverEndpoint != null) - // { - // Trace.Info($"Queueing results file ({path})"); - // var stepId = context.Id; - // // Attachments must be added to the parent context (job), not the current context (step) - // QueueResultsUpload(attachmentName, scrubbedFilePath, stepId); - // } - } - Trace.Verbose("Enqueue file upload queue: file '{0}' attach to record {1}", newFile.Path, timelineRecordId); _fileUploadQueue.Enqueue(newFile); - - } public void QueueResultsUpload(Guid recordId, string name, string path, string type, bool deleteSource, bool finalize, bool firstBlock, long totalLines) @@ -495,10 +477,11 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) { try { - if (String.Equals(file.Type, ChecksAttachmentType.StepSummary, StringComparison.OrdinalIgnoreCase)) + if (String.Equals(file.Type, ChecksAttachmentType.StepSummary, StringComparison.OrdinalIgnoreCase)) { await UploadSummaryFile(file); - } else if (String.Equals(file.Type, CoreAttachmentType.ResultsLog, StringComparison.OrdinalIgnoreCase)) + } + else if (String.Equals(file.Type, CoreAttachmentType.ResultsLog, StringComparison.OrdinalIgnoreCase)) { if (file.RecordId != _jobTimelineRecordId) { @@ -771,18 +754,15 @@ private async Task UploadFile(UploadFileInfo file) if (String.Equals(file.Type, CoreAttachmentType.Log, StringComparison.OrdinalIgnoreCase)) { // Create the log - Trace.Info("Create log async!"); var taskLog = await _jobServer.CreateLogAsync(_scopeIdentifier, _hubName, _planId, new TaskLog(String.Format(@"logs\{0:D}", file.TimelineRecordId)), default(CancellationToken)); // Upload the contents using (FileStream fs = File.Open(file.Path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite)) { - Trace.Info("Uploading a log!!"); var logUploaded = await _jobServer.AppendLogContentAsync(_scopeIdentifier, _hubName, _planId, taskLog.Id, fs, default(CancellationToken)); } // Create a new record and only set the Log field - Trace.Info("Updating timeline record!"); var attachmentUpdataRecord = new TimelineRecord() { Id = file.TimelineRecordId, Log = taskLog }; QueueTimelineRecordUpdate(file.TimelineId, attachmentUpdataRecord); } diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index 31cc4b68f69..989bc239563 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -202,7 +202,6 @@ private async Task UploadAppendFileAsync(string url, string if (blobStorageType == BlobStorageTypes.AzureBlobStorage) { - // request.Content.Headers.Add("x-ms-blob-type", "AppendBlock"); request.Content.Headers.Add("Content-Length", fileSize.ToString()); request.Content.Headers.Add("x-ms-blob-sealed", finalize.ToString()); } @@ -257,17 +256,19 @@ public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid st // Do we want to throw an exception here or should we just be uploading/truncating the data var fileSize = new FileInfo(file).Length; - // Upload the file + // Create the Append blob if (firstBlock) { await CreateAppendFileAsync(uploadUrlResponse.LogsUrl, uploadUrlResponse.BlobStorageType, cancellationToken); } + // Upload content using (var fileStream = new FileStream(file, FileMode.Open, FileAccess.Read, FileShare.Read, 4096, true)) { var response = await UploadAppendFileAsync(uploadUrlResponse.LogsUrl, uploadUrlResponse.BlobStorageType, fileStream, finalize, fileSize, cancellationToken); } + // Update metadata if (finalize) { // Send step log upload complete message From 57c33d2c2e78cd732a36a426cf92b9e7b71f9918 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Tue, 7 Feb 2023 12:51:06 -0500 Subject: [PATCH 13/14] Fixing typos --- src/Runner.Common/JobServer.cs | 4 ++-- src/Runner.Common/JobServerQueue.cs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/Runner.Common/JobServer.cs b/src/Runner.Common/JobServer.cs index 7a2f96ae077..085cc84979d 100644 --- a/src/Runner.Common/JobServer.cs +++ b/src/Runner.Common/JobServer.cs @@ -30,7 +30,7 @@ public interface IJobServer : IRunnerService, IAsyncDisposable Task AppendLogContentAsync(Guid scopeIdentifier, string hubName, Guid planId, int logId, Stream uploadStream, CancellationToken cancellationToken); Task AppendTimelineRecordFeedAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, Guid stepId, IList lines, long? startLine, CancellationToken cancellationToken); Task CreateAttachmentAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, Guid timelineRecordId, String type, String name, Stream uploadStream, CancellationToken cancellationToken); - Task CreateStepSymmaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken); + Task CreateStepSummaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken); Task CreateResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, long lineCount, CancellationToken cancellationToken); Task CreateLogAsync(Guid scopeIdentifier, string hubName, Guid planId, TaskLog log, CancellationToken cancellationToken); Task CreateTimelineAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, CancellationToken cancellationToken); @@ -317,7 +317,7 @@ public Task CreateAttachmentAsync(Guid scopeIdentifier, string h return _taskClient.CreateAttachmentAsync(scopeIdentifier, hubName, planId, timelineId, timelineRecordId, type, name, uploadStream, cancellationToken: cancellationToken); } - public Task CreateStepSymmaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken) + public Task CreateStepSummaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken) { if (_resultsClient != null) { diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index cb3f49f5d80..e736822fba0 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -802,7 +802,7 @@ private async Task UploadSummaryFile(ResultsUploadFileInfo file) // Upload the step summary Trace.Info($"Starting to upload summary file to results service {file.Name}, {file.Path}"); var cancellationTokenSource = new CancellationTokenSource(); - await _jobServer.CreateStepSymmaryAsync(file.PlanId, file.JobId, file.RecordId, file.Path, cancellationTokenSource.Token); + await _jobServer.CreateStepSummaryAsync(file.PlanId, file.JobId, file.RecordId, file.Path, cancellationTokenSource.Token); uploadSucceed = true; } From f3110f82cd626f5620bbb38ed42414960fb098cb Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Tue, 7 Feb 2023 13:35:47 -0500 Subject: [PATCH 14/14] Apply suggestions from code review Co-authored-by: Konrad Pabjan --- src/Runner.Common/JobServerQueue.cs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index e736822fba0..bc20fed4404 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -458,7 +458,7 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) while (_resultsFileUploadQueue.TryDequeue(out dequeueFile)) { filesToUpload.Add(dequeueFile); - // process at most 10 file upload. + // process at most 10 file uploads. if (!runOnce && filesToUpload.Count > 10) { break; @@ -469,7 +469,7 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) { if (runOnce) { - Trace.Info($"Uploading {filesToUpload.Count} files in one shot through results service."); + Trace.Info($"Uploading {filesToUpload.Count} file(s) in one shot through results service."); } int errorCount = 0; @@ -508,7 +508,7 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) } } - Trace.Info("Tried to upload {0} files to results, success rate: {1}/{0}.", filesToUpload.Count, filesToUpload.Count - errorCount); + Trace.Info("Tried to upload {0} file(s) to results, success rate: {1}/{0}.", filesToUpload.Count, filesToUpload.Count - errorCount); } if (runOnce) @@ -828,7 +828,7 @@ private async Task UploadResultsStepLogFile(ResultsUploadFileInfo file) bool uploadSucceed = false; try { - Trace.Info($"Starting to step log file to results service {file.Name}, {file.Path}"); + Trace.Info($"Starting upload of step log file to results service {file.Name}, {file.Path}"); var cancellationTokenSource = new CancellationTokenSource(); await _jobServer.CreateResultsStepLogAsync(file.PlanId, file.JobId, file.RecordId, file.Path, file.Finalize, file.FirstBlock, file.TotalLines, cancellationTokenSource.Token); @@ -844,7 +844,7 @@ private async Task UploadResultsStepLogFile(ResultsUploadFileInfo file) } catch (Exception ex) { - Trace.Info("Catch exception during delete success results uploaded step log file."); + Trace.Info("Exception encountered during deletion of a temporary file that was already successfully uploaded to results."); Trace.Error(ex); } }