From 4a54263aa713e564bb0026d2a4d794626d8aca33 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Wed, 15 Mar 2023 17:14:21 -0400 Subject: [PATCH 1/7] Also send Steps update to Results service --- src/Runner.Common/JobServer.cs | 18 +++++- src/Sdk/WebApi/WebApi/Contracts.cs | 41 ++++++++++++ src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 74 +++++++++++++++++++--- 3 files changed, 123 insertions(+), 10 deletions(-) diff --git a/src/Runner.Common/JobServer.cs b/src/Runner.Common/JobServer.cs index 9d845037868..aac568a41f2 100644 --- a/src/Runner.Common/JobServer.cs +++ b/src/Runner.Common/JobServer.cs @@ -359,8 +359,24 @@ public Task CreateTimelineAsync(Guid scopeIdentifier, string hubName, public Task> UpdateTimelineRecordsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, IEnumerable records, CancellationToken cancellationToken) { + // Send step updates to Results + var timelineRecords = records.ToList(); + if (_resultsClient != null) + { + try + { + _resultsClient.UpdateTimelineRecordsAsync(planId, new List(timelineRecords), cancellationToken: cancellationToken); + } + catch (Exception ex) + { + // Log error, but continue as this call is best-effort + Trace.Info($"Failed to update steps status due to {ex.GetType().Name}"); + Trace.Error(ex); + } + } + CheckConnection(); - return _taskClient.UpdateTimelineRecordsAsync(scopeIdentifier, hubName, planId, timelineId, records, cancellationToken: cancellationToken); + return _taskClient.UpdateTimelineRecordsAsync(scopeIdentifier, hubName, planId, timelineId, timelineRecords, cancellationToken: cancellationToken); } public Task RaisePlanEventAsync(Guid scopeIdentifier, string hubName, Guid planId, T eventData, CancellationToken cancellationToken) where T : JobEvent diff --git a/src/Sdk/WebApi/WebApi/Contracts.cs b/src/Sdk/WebApi/WebApi/Contracts.cs index 9165d832fa9..8da6bf236ff 100644 --- a/src/Sdk/WebApi/WebApi/Contracts.cs +++ b/src/Sdk/WebApi/WebApi/Contracts.cs @@ -1,3 +1,4 @@ +using System.Collections.Generic; using System.Runtime.Serialization; using Newtonsoft.Json; using Newtonsoft.Json.Serialization; @@ -126,6 +127,46 @@ public class CreateMetadataResponse public bool Ok; } + [DataContract] + [JsonObject(NamingStrategyType = typeof(SnakeCaseNamingStrategy))] + public class StepsUpdateRequest + { + [DataMember] + public IEnumerable Steps; + [DataMember] + public long ChangeOrder; + [DataMember] + public string WorkflowJobRunBackendId; + [DataMember] + public string WorkflowRunBackendId; + } + + [DataContract] + [JsonObject(NamingStrategyType = typeof(SnakeCaseNamingStrategy))] + public class Step + { + [DataMember] + public string ExternalId; + [DataMember] + public int Number; + [DataMember] + public string Name; + [DataMember] + public Status Status; + [DataMember] + public string StartedAt; + [DataMember] + public string CompletedAt; + } + + public enum Status + { + StatusUnknown = 0, + StatusInProgress = 3, + StatusPending = 5, + StatusCompleted = 6 + } + 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 b80ce4db3bc..c65235343f9 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -1,11 +1,16 @@ using System; +using System.Collections.Generic; +using System.Diagnostics; using System.IO; +using System.Linq; using System.Net.Http; using System.Net.Http.Headers; using System.Threading; using System.Threading.Tasks; -using GitHub.Services.Results.Contracts; using System.Net.Http.Formatting; +using GitHub.DistributedTask.WebApi; +using GitHub.Services.Common; +using GitHub.Services.Results.Contracts; using Sdk.WebApi.WebApi; namespace GitHub.Services.Results.Client @@ -22,6 +27,7 @@ public ResultsHttpClient( m_token = token; m_resultsServiceUrl = baseUrl; m_formatter = new JsonMediaTypeFormatter(); + m_changeIdCounter = 1; } // Get Sas URL calls @@ -86,7 +92,7 @@ private async Task GetJobLogUploadUrlAsync(string p // Create metadata calls - private async Task CreateMetadata(Uri uri, CancellationToken cancellationToken, R request, string timestamp) + private async Task SendRequest(Uri uri, CancellationToken cancellationToken, R request, string timestamp) { using (HttpRequestMessage requestMessage = new HttpRequestMessage(HttpMethod.Post, uri)) { @@ -121,7 +127,7 @@ private async Task StepSummaryUploadCompleteAsync(string planId, string jobId, G }; var createStepSummaryMetadataEndpoint = new Uri(m_resultsServiceUrl, Constants.CreateStepSummaryMetadata); - await CreateMetadata(createStepSummaryMetadataEndpoint, cancellationToken, request, timestamp); + await SendRequest(createStepSummaryMetadataEndpoint, cancellationToken, request, timestamp); } private async Task StepLogUploadCompleteAsync(string planId, string jobId, Guid stepId, long lineCount, CancellationToken cancellationToken) @@ -137,7 +143,7 @@ private async Task StepLogUploadCompleteAsync(string planId, string jobId, Guid }; var createStepLogsMetadataEndpoint = new Uri(m_resultsServiceUrl, Constants.CreateStepLogsMetadata); - await CreateMetadata(createStepLogsMetadataEndpoint, cancellationToken, request, timestamp); + await SendRequest(createStepLogsMetadataEndpoint, cancellationToken, request, timestamp); } private async Task JobLogUploadCompleteAsync(string planId, string jobId, long lineCount, CancellationToken cancellationToken) @@ -152,7 +158,7 @@ private async Task JobLogUploadCompleteAsync(string planId, string jobId, long l }; var createJobLogsMetadataEndpoint = new Uri(m_resultsServiceUrl, Constants.CreateJobLogsMetadata); - await CreateMetadata(createJobLogsMetadataEndpoint, cancellationToken, request, timestamp); + await SendRequest(createJobLogsMetadataEndpoint, cancellationToken, request, timestamp); } private async Task UploadBlockFileAsync(string url, string blobStorageType, FileStream file, CancellationToken cancellationToken) @@ -252,7 +258,7 @@ public async Task UploadStepSummaryAsync(string planId, string jobId, Guid stepI await StepSummaryUploadCompleteAsync(planId, jobId, stepId, fileSize, cancellationToken); } - // Handle file upload for step log + // Handle file upload for step log public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid stepId, string file, bool finalize, bool firstBlock, long lineCount, CancellationToken cancellationToken) { // Get the upload url @@ -262,7 +268,7 @@ public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid st throw new Exception("Failed to get step log upload url"); } - // Create the Append blob + // Create the Append blob if (firstBlock) { await CreateAppendFileAsync(uploadUrlResponse.LogsUrl, uploadUrlResponse.BlobStorageType, cancellationToken); @@ -283,7 +289,7 @@ public async Task UploadResultsStepLogAsync(string planId, string jobId, Guid st } } - // Handle file upload for job log + // Handle file upload for job log public async Task UploadResultsJobLogAsync(string planId, string jobId, string file, bool finalize, bool firstBlock, long lineCount, CancellationToken cancellationToken) { // Get the upload url @@ -293,7 +299,7 @@ public async Task UploadResultsJobLogAsync(string planId, string jobId, string f throw new Exception("Failed to get job log upload url"); } - // Create the Append blob + // Create the Append blob if (firstBlock) { await CreateAppendFileAsync(uploadUrlResponse.LogsUrl, uploadUrlResponse.BlobStorageType, cancellationToken); @@ -314,9 +320,57 @@ public async Task UploadResultsJobLogAsync(string planId, string jobId, string f } } + private Step ConvertTimelineRecordToStep(TimelineRecord r) + { + return new Step() + { + ExternalId = r.Id.ToString(), + Number = r.Order.GetValueOrDefault(), + Name = r.Name, + Status = ConvertStateToStatus(r.State.GetValueOrDefault()), + StartedAt = r.StartTime?.ToString(Constants.TimestampFormat), + CompletedAt = r.FinishTime?.ToString(Constants.TimestampFormat) + }; + } + + private Status ConvertStateToStatus(TimelineRecordState s) + { + switch (s) + { + case TimelineRecordState.Completed: + return Status.StatusCompleted; + case TimelineRecordState.Pending: + return Status.StatusPending; + case TimelineRecordState.InProgress: + return Status.StatusInProgress; + default: + return Status.StatusUnknown; + } + } + + public async Task UpdateTimelineRecordsAsync(Guid planId, IEnumerable records, CancellationToken cancellationToken) + { + var timestamp = DateTime.UtcNow.ToString(Constants.TimestampFormat); + var stepRecords = records.Where(r => String.Equals(r.RecordType, "Task", StringComparison.Ordinal)); + var stepUpdateRequests = stepRecords.GroupBy(r => r.ParentId).Select(sg => new StepsUpdateRequest() + { + WorkflowRunBackendId = planId.ToString(), + WorkflowJobRunBackendId = sg.Key.ToString(), + ChangeOrder = m_changeIdCounter++, + Steps = sg.Select(ConvertTimelineRecordToStep) + }); + + var stepUpdateEndpoint = new Uri(m_resultsServiceUrl, Constants.WorkflowStepsUpdate); + foreach (var request in stepUpdateRequests) + { + await SendRequest(stepUpdateEndpoint, cancellationToken, request, timestamp); + } + } + private MediaTypeFormatter m_formatter; private Uri m_resultsServiceUrl; private string m_token; + private int m_changeIdCounter; } // Constants specific to results @@ -331,6 +385,8 @@ public static class Constants public static readonly string CreateStepLogsMetadata = ResultsReceiverTwirpEndpoint + "CreateStepLogsMetadata"; public static readonly string GetJobLogsSignedBlobURL = ResultsReceiverTwirpEndpoint + "GetJobLogsSignedBlobURL"; public static readonly string CreateJobLogsMetadata = ResultsReceiverTwirpEndpoint + "CreateJobLogsMetadata"; + public static readonly string ResultsProtoApiV1Endpoint = "twirp/github.actions.results.api.v1.WorkflowStepUpdateService/"; + public static readonly string WorkflowStepsUpdate = ResultsProtoApiV1Endpoint + "WorkflowStepsUpdate"; public static readonly string AzureBlobSealedHeader = "x-ms-blob-sealed"; public static readonly string AzureBlobTypeHeader = "x-ms-blob-type"; From 506ee7ec17a3ff3e52e5ef83bdb18a68bf2e1184 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Tue, 28 Mar 2023 16:43:06 -0400 Subject: [PATCH 2/7] Refactor to separate results server from current job server --- src/Runner.Common/JobServer.cs | 54 +---------- src/Runner.Common/JobServerQueue.cs | 16 +++- src/Runner.Common/ResultsServer.cs | 103 +++++++++++++++++++++ src/Runner.Worker/JobRunner.cs | 2 + src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 2 +- 5 files changed, 118 insertions(+), 59 deletions(-) create mode 100644 src/Runner.Common/ResultsServer.cs diff --git a/src/Runner.Common/JobServer.cs b/src/Runner.Common/JobServer.cs index aac568a41f2..02612d427f8 100644 --- a/src/Runner.Common/JobServer.cs +++ b/src/Runner.Common/JobServer.cs @@ -24,15 +24,11 @@ public interface IJobServer : IRunnerService, IAsyncDisposable Task ConnectAsync(VssConnection jobConnection); void InitializeWebsocketClient(ServiceEndpoint serviceEndpoint); - void InitializeResultsClient(Uri uri, string token); // logging and console 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 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 CreateResultsJobLogAsync(string planId, string jobId, 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); @@ -46,7 +42,6 @@ public sealed class JobServer : RunnerService, IJobServer private bool _hasConnection; private VssConnection _connection; private TaskHttpClient _taskClient; - private ResultsHttpClient _resultsClient; private ClientWebSocket _websocketClient; private ServiceEndpoint _serviceEndpoint; @@ -150,12 +145,6 @@ public void InitializeWebsocketClient(ServiceEndpoint serviceEndpoint) InitializeWebsocketClient(TimeSpan.Zero); } - public void InitializeResultsClient(Uri uri, string token) - { - var httpMessageHandler = HostContext.CreateHttpClientHandler(); - this._resultsClient = new ResultsHttpClient(uri, httpMessageHandler, token, disposeHandler: true); - } - public ValueTask DisposeAsync() { CloseWebSocket(WebSocketCloseStatus.NormalClosure, CancellationToken.None); @@ -318,32 +307,6 @@ public Task CreateAttachmentAsync(Guid scopeIdentifier, string h return _taskClient.CreateAttachmentAsync(scopeIdentifier, hubName, planId, timelineId, timelineRecordId, type, name, uploadStream, cancellationToken: cancellationToken); } - public Task CreateStepSummaryAsync(string planId, string jobId, Guid stepId, string file, CancellationToken cancellationToken) - { - if (_resultsClient != null) - { - return _resultsClient.UploadStepSummaryAsync(planId, jobId, stepId, file, cancellationToken: cancellationToken); - } - throw new InvalidOperationException("Results client is not initialized."); - } - - 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, lineCount, cancellationToken: cancellationToken); - } - throw new InvalidOperationException("Results client is not initialized."); - } - - public Task CreateResultsJobLogAsync(string planId, string jobId, string file, bool finalize, bool firstBlock, long lineCount, CancellationToken cancellationToken) - { - if (_resultsClient != null) - { - return _resultsClient.UploadResultsJobLogAsync(planId, jobId, file, finalize, firstBlock, lineCount, cancellationToken: cancellationToken); - } - throw new InvalidOperationException("Results client is not initialized."); - } public Task CreateLogAsync(Guid scopeIdentifier, string hubName, Guid planId, TaskLog log, CancellationToken cancellationToken) { @@ -360,23 +323,8 @@ public Task CreateTimelineAsync(Guid scopeIdentifier, string hubName, public Task> UpdateTimelineRecordsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, IEnumerable records, CancellationToken cancellationToken) { // Send step updates to Results - var timelineRecords = records.ToList(); - if (_resultsClient != null) - { - try - { - _resultsClient.UpdateTimelineRecordsAsync(planId, new List(timelineRecords), cancellationToken: cancellationToken); - } - catch (Exception ex) - { - // Log error, but continue as this call is best-effort - Trace.Info($"Failed to update steps status due to {ex.GetType().Name}"); - Trace.Error(ex); - } - } - CheckConnection(); - return _taskClient.UpdateTimelineRecordsAsync(scopeIdentifier, hubName, planId, timelineId, timelineRecords, cancellationToken: cancellationToken); + return _taskClient.UpdateTimelineRecordsAsync(scopeIdentifier, hubName, planId, timelineId, records, cancellationToken: cancellationToken); } public Task RaisePlanEventAsync(Guid scopeIdentifier, string hubName, Guid planId, T eventData, CancellationToken cancellationToken) where T : JobEvent diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 1fadd99a8df..44943bfefb1 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -65,6 +65,7 @@ public sealed class JobServerQueue : RunnerService, IJobServerQueue // common private IJobServer _jobServer; + private IResultsServer _resultsServer; private Task[] _allDequeueTasks; private readonly TaskCompletionSource _jobCompletionSource = new(); private readonly TaskCompletionSource _jobRecordUpdated = new(); @@ -91,6 +92,7 @@ public override void Initialize(IHostContext hostContext) { base.Initialize(hostContext); _jobServer = hostContext.GetService(); + _resultsServer = hostContext.GetService(); } public void Start(Pipelines.AgentJobRequestMessage jobRequest) @@ -111,8 +113,11 @@ public void Start(Pipelines.AgentJobRequestMessage jobRequest) !string.IsNullOrEmpty(resultsReceiverEndpoint)) { Trace.Info("Initializing results client"); - _jobServer.InitializeResultsClient(new Uri(resultsReceiverEndpoint), accessToken); - _resultsClientInitiated = true; + if (_resultsServer != null) + { + _resultsServer.InitializeResultsClient(new Uri(resultsReceiverEndpoint), accessToken); + _resultsClientInitiated = true; + } } if (_queueInProcess) @@ -612,6 +617,7 @@ private async Task ProcessTimelinesUpdateQueueAsync(bool runOnce = false) try { await _jobServer.UpdateTimelineRecordsAsync(_scopeIdentifier, _hubName, _planId, update.TimelineId, update.PendingRecords, default(CancellationToken)); + await _resultsServer.UpdateResultsWorkflowStepsAsync(_scopeIdentifier, _hubName, _planId, update.TimelineId, update.PendingRecords, default(CancellationToken)); if (_bufferedRetryRecords.Remove(update.TimelineId)) { Trace.Verbose("Cleanup buffered timeline record for timeline: {0}.", update.TimelineId); @@ -819,7 +825,7 @@ private async Task UploadSummaryFile(ResultsUploadFileInfo file) Trace.Info($"Starting to upload summary file to results service {file.Name}, {file.Path}"); ResultsFileUploadHandler summaryHandler = async (file) => { - await _jobServer.CreateStepSummaryAsync(file.PlanId, file.JobId, file.RecordId, file.Path, CancellationToken.None); + await _resultsServer.CreateResultsStepSummaryAsync(file.PlanId, file.JobId, file.RecordId, file.Path, CancellationToken.None); }; await UploadResultsFile(file, summaryHandler); @@ -830,7 +836,7 @@ private async Task UploadResultsStepLogFile(ResultsUploadFileInfo file) Trace.Info($"Starting upload of step log file to results service {file.Name}, {file.Path}"); ResultsFileUploadHandler stepLogHandler = async (file) => { - await _jobServer.CreateResultsStepLogAsync(file.PlanId, file.JobId, file.RecordId, file.Path, file.Finalize, file.FirstBlock, file.TotalLines, CancellationToken.None); + await _resultsServer.CreateResultsStepLogAsync(file.PlanId, file.JobId, file.RecordId, file.Path, file.Finalize, file.FirstBlock, file.TotalLines, CancellationToken.None); }; await UploadResultsFile(file, stepLogHandler); @@ -841,7 +847,7 @@ private async Task UploadResultsJobLogFile(ResultsUploadFileInfo file) Trace.Info($"Starting upload of job log file to results service {file.Name}, {file.Path}"); ResultsFileUploadHandler jobLogHandler = async (file) => { - await _jobServer.CreateResultsJobLogAsync(file.PlanId, file.JobId, file.Path, file.Finalize, file.FirstBlock, file.TotalLines, CancellationToken.None); + await _resultsServer.CreateResultsJobLogAsync(file.PlanId, file.JobId, file.Path, file.Finalize, file.FirstBlock, file.TotalLines, CancellationToken.None); }; await UploadResultsFile(file, jobLogHandler); diff --git a/src/Runner.Common/ResultsServer.cs b/src/Runner.Common/ResultsServer.cs new file mode 100644 index 00000000000..a50a10903b5 --- /dev/null +++ b/src/Runner.Common/ResultsServer.cs @@ -0,0 +1,103 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using GitHub.DistributedTask.WebApi; +using GitHub.Services.Results.Client; + +namespace GitHub.Runner.Common +{ + [ServiceLocator(Default = typeof(ResultServer))] + public interface IResultsServer : IRunnerService, IAsyncDisposable + { + void InitializeResultsClient(Uri uri, string token); + + // logging and console + Task CreateResultsStepSummaryAsync(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 CreateResultsJobLogAsync(string planId, string jobId, string file, bool finalize, bool firstBlock, + long lineCount, CancellationToken cancellationToken); + + Task UpdateResultsWorkflowStepsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, + IEnumerable records, CancellationToken cancellationToken); + } + + public sealed class ResultServer : RunnerService, IResultsServer + { + private ResultsHttpClient _resultsClient; + + public void InitializeResultsClient(Uri uri, string token) + { + var httpMessageHandler = HostContext.CreateHttpClientHandler(); + this._resultsClient = new ResultsHttpClient(uri, httpMessageHandler, token, disposeHandler: true); + } + + public Task CreateResultsStepSummaryAsync(string planId, string jobId, Guid stepId, string file, + CancellationToken cancellationToken) + { + if (_resultsClient != null) + { + return _resultsClient.UploadStepSummaryAsync(planId, jobId, stepId, file, + cancellationToken: cancellationToken); + } + + throw new InvalidOperationException("Results client is not initialized."); + } + + 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, + lineCount, cancellationToken: cancellationToken); + } + + throw new InvalidOperationException("Results client is not initialized."); + } + + public Task CreateResultsJobLogAsync(string planId, string jobId, string file, bool finalize, bool firstBlock, + long lineCount, CancellationToken cancellationToken) + { + if (_resultsClient != null) + { + return _resultsClient.UploadResultsJobLogAsync(planId, jobId, file, finalize, firstBlock, lineCount, + cancellationToken: cancellationToken); + } + + throw new InvalidOperationException("Results client is not initialized."); + } + + public Task UpdateResultsWorkflowStepsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, + IEnumerable records, CancellationToken cancellationToken) + { + if (_resultsClient != null) + { + try + { + var timelineRecords = records.ToList(); + return _resultsClient.UpdateWorkflowStepsAsync(planId, new List(timelineRecords), + cancellationToken: cancellationToken); + } + catch (Exception ex) + { + // Log error, but continue as this call is best-effort + Trace.Info($"Failed to update steps status due to {ex.GetType().Name}"); + Trace.Error(ex); + } + } + + throw new InvalidOperationException("Results client is not initialized."); + } + + public ValueTask DisposeAsync() + { + return ValueTask.CompletedTask; + } + } +} diff --git a/src/Runner.Worker/JobRunner.cs b/src/Runner.Worker/JobRunner.cs index f63fbf71776..dd0e95f88d3 100644 --- a/src/Runner.Worker/JobRunner.cs +++ b/src/Runner.Worker/JobRunner.cs @@ -49,6 +49,8 @@ public async Task RunAsync(AgentJobRequestMessage message, Cancellat VssCredentials jobServerCredential = VssUtil.GetVssCredential(systemConnection); await runServer.ConnectAsync(systemConnection.Url, jobServerCredential); server = runServer; + + _jobServerQueue.Start(message); } else { diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index c65235343f9..335dd11b1d2 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -348,7 +348,7 @@ private Status ConvertStateToStatus(TimelineRecordState s) } } - public async Task UpdateTimelineRecordsAsync(Guid planId, IEnumerable records, CancellationToken cancellationToken) + public async Task UpdateWorkflowStepsAsync(Guid planId, IEnumerable records, CancellationToken cancellationToken) { var timestamp = DateTime.UtcNow.ToString(Constants.TimestampFormat); var stepRecords = records.Where(r => String.Equals(r.RecordType, "Task", StringComparison.Ordinal)); From 499bb7987f28c55262e1b28b23fb577d308cb183 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Tue, 28 Mar 2023 16:53:01 -0400 Subject: [PATCH 3/7] If hit any error while uploading to Results, skip Results upload --- src/Runner.Common/JobServer.cs | 1 - src/Runner.Common/JobServerQueue.cs | 24 ++++++++++++++++++- src/Runner.Worker/JobRunner.cs | 2 -- .../Common/Common/RawHttpMessageHandler.cs | 2 +- .../Common/Common/VssHttpMessageHandler.cs | 2 +- .../Common/Common/VssHttpRequestSettings.cs | 2 +- src/Sdk/WebApi/WebApi/ResultsHttpClient.cs | 8 +++---- 7 files changed, 30 insertions(+), 11 deletions(-) diff --git a/src/Runner.Common/JobServer.cs b/src/Runner.Common/JobServer.cs index 02612d427f8..04c244ad8a3 100644 --- a/src/Runner.Common/JobServer.cs +++ b/src/Runner.Common/JobServer.cs @@ -322,7 +322,6 @@ public Task CreateTimelineAsync(Guid scopeIdentifier, string hubName, public Task> UpdateTimelineRecordsAsync(Guid scopeIdentifier, string hubName, Guid planId, Guid timelineId, IEnumerable records, CancellationToken cancellationToken) { - // Send step updates to Results CheckConnection(); return _taskClient.UpdateTimelineRecordsAsync(scopeIdentifier, hubName, planId, timelineId, records, cancellationToken: cancellationToken); } diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 44943bfefb1..2db2d85d241 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -530,6 +530,9 @@ 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); errorCount++; + + // If we hit any exceptions uploading to Results, let's skip any additional uploads to Results + _resultsClientInitiated = false; } } @@ -617,7 +620,21 @@ private async Task ProcessTimelinesUpdateQueueAsync(bool runOnce = false) try { await _jobServer.UpdateTimelineRecordsAsync(_scopeIdentifier, _hubName, _planId, update.TimelineId, update.PendingRecords, default(CancellationToken)); - await _resultsServer.UpdateResultsWorkflowStepsAsync(_scopeIdentifier, _hubName, _planId, update.TimelineId, update.PendingRecords, default(CancellationToken)); + try + { + if (_resultsClientInitiated) + { + await _resultsServer.UpdateResultsWorkflowStepsAsync(_scopeIdentifier, _hubName, _planId, update.TimelineId, update.PendingRecords, default(CancellationToken)); + } + } + catch (Exception e) + { + Trace.Info("Catch exception during update steps, skip update Results."); + Trace.Error(e); + + _resultsClientInitiated = false; + } + if (_bufferedRetryRecords.Remove(update.TimelineId)) { Trace.Verbose("Cleanup buffered timeline record for timeline: {0}.", update.TimelineId); @@ -855,6 +872,11 @@ private async Task UploadResultsJobLogFile(ResultsUploadFileInfo file) private async Task UploadResultsFile(ResultsUploadFileInfo file, ResultsFileUploadHandler uploadHandler) { + if (!_resultsClientInitiated) + { + return; + } + bool uploadSucceed = false; try { diff --git a/src/Runner.Worker/JobRunner.cs b/src/Runner.Worker/JobRunner.cs index dd0e95f88d3..f63fbf71776 100644 --- a/src/Runner.Worker/JobRunner.cs +++ b/src/Runner.Worker/JobRunner.cs @@ -49,8 +49,6 @@ public async Task RunAsync(AgentJobRequestMessage message, Cancellat VssCredentials jobServerCredential = VssUtil.GetVssCredential(systemConnection); await runServer.ConnectAsync(systemConnection.Url, jobServerCredential); server = runServer; - - _jobServerQueue.Start(message); } else { diff --git a/src/Sdk/Common/Common/RawHttpMessageHandler.cs b/src/Sdk/Common/Common/RawHttpMessageHandler.cs index 8b3e75e1844..538a824efff 100644 --- a/src/Sdk/Common/Common/RawHttpMessageHandler.cs +++ b/src/Sdk/Common/Common/RawHttpMessageHandler.cs @@ -9,7 +9,7 @@ namespace GitHub.Services.Common { - public class RawHttpMessageHandler: HttpMessageHandler + public class RawHttpMessageHandler : HttpMessageHandler { public RawHttpMessageHandler( FederatedCredential credentials) diff --git a/src/Sdk/Common/Common/VssHttpMessageHandler.cs b/src/Sdk/Common/Common/VssHttpMessageHandler.cs index b40f7ea21a2..39be573b8aa 100644 --- a/src/Sdk/Common/Common/VssHttpMessageHandler.cs +++ b/src/Sdk/Common/Common/VssHttpMessageHandler.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; using System.Net; diff --git a/src/Sdk/Common/Common/VssHttpRequestSettings.cs b/src/Sdk/Common/Common/VssHttpRequestSettings.cs index 06c1554ce6d..8a86eea0822 100644 --- a/src/Sdk/Common/Common/VssHttpRequestSettings.cs +++ b/src/Sdk/Common/Common/VssHttpRequestSettings.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.ComponentModel; using System.Globalization; diff --git a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs index 335dd11b1d2..b80740f44a5 100644 --- a/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs +++ b/src/Sdk/WebApi/WebApi/ResultsHttpClient.cs @@ -354,10 +354,10 @@ public async Task UpdateWorkflowStepsAsync(Guid planId, IEnumerable String.Equals(r.RecordType, "Task", StringComparison.Ordinal)); var stepUpdateRequests = stepRecords.GroupBy(r => r.ParentId).Select(sg => new StepsUpdateRequest() { - WorkflowRunBackendId = planId.ToString(), - WorkflowJobRunBackendId = sg.Key.ToString(), - ChangeOrder = m_changeIdCounter++, - Steps = sg.Select(ConvertTimelineRecordToStep) + WorkflowRunBackendId = planId.ToString(), + WorkflowJobRunBackendId = sg.Key.ToString(), + ChangeOrder = m_changeIdCounter++, + Steps = sg.Select(ConvertTimelineRecordToStep) }); var stepUpdateEndpoint = new Uri(m_resultsServiceUrl, Constants.WorkflowStepsUpdate); From a55f68fe9a0cbd71cfdd7134166a25c5a1517b97 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Wed, 29 Mar 2023 11:39:41 -0400 Subject: [PATCH 4/7] Add proxy authentication and buffer request for WinHttpHandler --- src/Runner.Common/JobServerQueue.cs | 2 - .../Common/RawClientHttpRequestSettings.cs | 24 +++++++++++ .../Common/Common/RawHttpMessageHandler.cs | 41 +++++++++++++++++-- 3 files changed, 62 insertions(+), 5 deletions(-) diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 2db2d85d241..776ded431c7 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -931,8 +931,6 @@ internal class ResultsUploadFileInfo public long TotalLines { get; set; } } - - internal class ConsoleLineInfo { public ConsoleLineInfo(Guid recordId, string line, long? lineNumber) diff --git a/src/Sdk/Common/Common/RawClientHttpRequestSettings.cs b/src/Sdk/Common/Common/RawClientHttpRequestSettings.cs index 60fbbe093fa..c6956f29faa 100644 --- a/src/Sdk/Common/Common/RawClientHttpRequestSettings.cs +++ b/src/Sdk/Common/Common/RawClientHttpRequestSettings.cs @@ -184,9 +184,33 @@ private static RawClientHttpRequestSettings ConstructDefaultSettings() return settings; } + /// + /// Gets or sets the maximum size allowed for response content buffering. + /// + [DefaultValue(c_defaultContentBufferSize)] + public Int32 MaxContentBufferSize + { + get + { + return m_maxContentBufferSize; + } + set + { + ArgumentUtility.CheckForOutOfRange(value, nameof(value), 0, c_maxAllowedContentBufferSize); + m_maxContentBufferSize = value; + } + } + private static Lazy s_defaultSettings = new Lazy(ConstructDefaultSettings); + private Int32 m_maxContentBufferSize; + // We will buffer a maximum of 1024MB in the message handler + private const Int32 c_maxAllowedContentBufferSize = 1024 * 1024 * 1024; + + // We will buffer, by default, up to 512MB in the message handler + private const Int32 c_defaultContentBufferSize = 1024 * 1024 * 512; + private const Int32 c_defaultMaxRetry = 3; private static readonly TimeSpan s_defaultTimeout = TimeSpan.FromSeconds(100); //default WebAPI timeout private ICollection m_acceptLanguages = new List(); diff --git a/src/Sdk/Common/Common/RawHttpMessageHandler.cs b/src/Sdk/Common/Common/RawHttpMessageHandler.cs index 538a824efff..6774d371218 100644 --- a/src/Sdk/Common/Common/RawHttpMessageHandler.cs +++ b/src/Sdk/Common/Common/RawHttpMessageHandler.cs @@ -120,6 +120,7 @@ protected override async Task SendAsync( Boolean succeeded = false; HttpResponseMessageWrapper responseWrapper; + Boolean lastResponseDemandedProxyAuth = false; Int32 retries = m_maxAuthRetries; try { @@ -138,7 +139,13 @@ protected override async Task SendAsync( // Let's start with sending a token IssuedToken token = await m_tokenProvider.GetTokenAsync(null, tokenSource.Token).ConfigureAwait(false); - ApplyToken(request, token); + ApplyToken(request, token, applyICredentialsToWebProxy: lastResponseDemandedProxyAuth); + + // The WinHttpHandler will chunk any content that does not have a computed length which is + // not what we want. By loading into a buffer up-front we bypass this behavior and there is + // no difference in the normal HttpClientHandler behavior here since this is what they were + // already doing. + await BufferRequestContentAsync(request, tokenSource.Token).ConfigureAwait(false); // ConfigureAwait(false) enables the continuation to be run outside any captured // SyncronizationContext (such as ASP.NET's) which keeps things from deadlocking... @@ -147,7 +154,8 @@ protected override async Task SendAsync( responseWrapper = new HttpResponseMessageWrapper(response); var isUnAuthorized = responseWrapper.StatusCode == HttpStatusCode.Unauthorized; - if (!isUnAuthorized) + lastResponseDemandedProxyAuth = responseWrapper.StatusCode == HttpStatusCode.ProxyAuthenticationRequired; + if (!isUnAuthorized && !lastResponseDemandedProxyAuth) { // Validate the token after it has been successfully authenticated with the server. m_tokenProvider?.ValidateToken(token, responseWrapper); @@ -211,15 +219,42 @@ protected override async Task SendAsync( } } + private static async Task BufferRequestContentAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + if (request.Content != null && + request.Headers.TransferEncodingChunked != true) + { + Int64? contentLength = request.Content.Headers.ContentLength; + if (contentLength == null) + { + await request.Content.LoadIntoBufferAsync().EnforceCancellation(cancellationToken).ConfigureAwait(false); + } + + // Explicitly turn off chunked encoding since we have computed the request content size + request.Headers.TransferEncodingChunked = false; + } + } + private void ApplyToken( HttpRequestMessage request, - IssuedToken token) + IssuedToken token, + bool applyICredentialsToWebProxy = false) { switch (token) { case null: return; case ICredentials credentialsToken: + if (applyICredentialsToWebProxy) + { + HttpClientHandler httpClientHandler = m_transportHandler as HttpClientHandler; + if (httpClientHandler != null && httpClientHandler.Proxy != null) + { + httpClientHandler.Proxy.Credentials = credentialsToken; + } + } m_credentialWrapper.InnerCredentials = credentialsToken; break; default: From f77cba4ad44cb8bd25c6d514ae296e12ee690700 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Wed, 29 Mar 2023 11:45:22 -0400 Subject: [PATCH 5/7] Remove unnecessary null guard --- src/Runner.Common/JobServerQueue.cs | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index 776ded431c7..f4490860126 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -113,11 +113,8 @@ public void Start(Pipelines.AgentJobRequestMessage jobRequest) !string.IsNullOrEmpty(resultsReceiverEndpoint)) { Trace.Info("Initializing results client"); - if (_resultsServer != null) - { - _resultsServer.InitializeResultsClient(new Uri(resultsReceiverEndpoint), accessToken); - _resultsClientInitiated = true; - } + _resultsServer.InitializeResultsClient(new Uri(resultsReceiverEndpoint), accessToken); + _resultsClientInitiated = true; } if (_queueInProcess) From 43773b0b391be0c2d944d1885bf9ecd42c0fe0b5 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Wed, 29 Mar 2023 11:54:23 -0400 Subject: [PATCH 6/7] Also send Results telemetry when step update fails --- src/Runner.Common/JobServerQueue.cs | 28 +++++++++++++++++----------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/src/Runner.Common/JobServerQueue.cs b/src/Runner.Common/JobServerQueue.cs index f4490860126..20ad27a89fb 100644 --- a/src/Runner.Common/JobServerQueue.cs +++ b/src/Runner.Common/JobServerQueue.cs @@ -514,22 +514,14 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) } catch (Exception ex) { - 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() - { - Id = Constants.Runner.TelemetryRecordId, - }; - telemetryRecord.Issues.Add(issue); - QueueTimelineRecordUpdate(_jobTimelineId, telemetryRecord); - Trace.Info("Catch exception during file upload to results, keep going since the process is best effort."); Trace.Error(ex); errorCount++; // If we hit any exceptions uploading to Results, let's skip any additional uploads to Results _resultsClientInitiated = false; + + SendResultsTelemetry(ex); } } @@ -547,6 +539,19 @@ private async Task ProcessResultsUploadQueueAsync(bool runOnce = false) } } + private void SendResultsTelemetry(Exception ex) + { + var issue = new Issue() { Type = IssueType.Warning, Message = $"Caught exception with results. {ex.Message}" }; + issue.Data[Constants.Runner.InternalTelemetryIssueDataKey] = Constants.Runner.ResultsUploadFailure; + + var telemetryRecord = new TimelineRecord() + { + Id = Constants.Runner.TelemetryRecordId, + }; + telemetryRecord.Issues.Add(issue); + QueueTimelineRecordUpdate(_jobTimelineId, telemetryRecord); + } + private async Task ProcessTimelinesUpdateQueueAsync(bool runOnce = false) { while (!_jobCompletionSource.Task.IsCompleted || runOnce) @@ -628,8 +633,9 @@ private async Task ProcessTimelinesUpdateQueueAsync(bool runOnce = false) { Trace.Info("Catch exception during update steps, skip update Results."); Trace.Error(e); - _resultsClientInitiated = false; + + SendResultsTelemetry(e); } if (_bufferedRetryRecords.Remove(update.TimelineId)) From 5f1697f0709865a17c03fb3209c48f95a3e00641 Mon Sep 17 00:00:00 2001 From: Yang Cao Date: Mon, 3 Apr 2023 15:38:25 -0400 Subject: [PATCH 7/7] IResultsServer is not disposable --- src/Runner.Common/ResultsServer.cs | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/src/Runner.Common/ResultsServer.cs b/src/Runner.Common/ResultsServer.cs index a50a10903b5..4d24d55c86a 100644 --- a/src/Runner.Common/ResultsServer.cs +++ b/src/Runner.Common/ResultsServer.cs @@ -9,7 +9,7 @@ namespace GitHub.Runner.Common { [ServiceLocator(Default = typeof(ResultServer))] - public interface IResultsServer : IRunnerService, IAsyncDisposable + public interface IResultsServer : IRunnerService { void InitializeResultsClient(Uri uri, string token); @@ -94,10 +94,5 @@ public Task UpdateResultsWorkflowStepsAsync(Guid scopeIdentifier, string hubName throw new InvalidOperationException("Results client is not initialized."); } - - public ValueTask DisposeAsync() - { - return ValueTask.CompletedTask; - } } }