Compare commits

...

7 Commits

Author SHA1 Message Date
eric sciple
04ece46c6a temp tracing for client auth 2022-05-10 18:20:49 +00:00
eric sciple
e75d502ab1 use token for broker and actions service 2022-05-09 06:20:08 +00:00
eric sciple
5686904fbe temporarily dont renew the job lock 2022-04-28 03:04:35 +00:00
eric sciple
c278fb1736 working on running job 2022-04-28 02:32:40 +00:00
eric sciple
332b8f9240 send runner info on getNextMessage 2022-04-09 20:04:55 +00:00
eric sciple
25f6cc100f print retrieved message 2022-04-09 00:31:59 +00:00
eric sciple
6ec30ea522 adding compile-time const whether to use the broker service 2022-04-08 23:23:27 +00:00
26 changed files with 642 additions and 31 deletions

View File

@@ -47,6 +47,11 @@
<DefineConstants>$(DefineConstants);DEBUG</DefineConstants>
</PropertyGroup>
<!-- Set USE_BROKER vars -->
<PropertyGroup Condition="'$(USE_BROKER)' == 'true'">
<DefineConstants>$(DefineConstants);USE_BROKER</DefineConstants>
</PropertyGroup>
<!-- Set Treat tarnings as errors -->
<PropertyGroup>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>

View File

@@ -0,0 +1,59 @@
using System;
using System.Collections.Generic;
using GitHub.DistributedTask.WebApi;
using System.Threading;
using System.Threading.Tasks;
using GitHub.Runner.Common.Util;
using GitHub.Services.WebApi;
using GitHub.Services.Common;
using GitHub.Runner.Sdk;
using System.Net;
using System.Net.Http;
namespace GitHub.Runner.Common
{
[ServiceLocator(Default = typeof(BrokerServer))]
public interface IBrokerServer : IRunnerService
{
Task ConnectAsync(Uri serverUrl, CancellationToken cancellationToken);
Task<string> GetMessageAsync(TaskAgentSession session, RunnerSettings settings, long? lastMessageId, CancellationToken cancellationToken);
}
public sealed class BrokerServer : RunnerService, IBrokerServer
{
private HttpClient _httpClient;
public async Task ConnectAsync(Uri serverUrl, CancellationToken cancellationToken)
{
_httpClient = new HttpClient();
_httpClient.BaseAddress = serverUrl;
_httpClient.Timeout = TimeSpan.FromSeconds(100);
await _httpClient.GetAsync("health", cancellationToken);
}
public async Task<string> GetMessageAsync(TaskAgentSession session, RunnerSettings settings, long? lastMessageId, CancellationToken cancellationToken)
{
var response = await _httpClient.GetAsync($"message?tenant=org:github&root_tenant=org:github&group_id={settings.PoolId}&group_name={settings.PoolName}&runner_id={settings.AgentId}&runner_name={settings.AgentName}&labels=self-hosted,linux", cancellationToken);
if (!response.IsSuccessStatusCode)
{
var content = default(string);
try
{
content = await response.Content.ReadAsStringAsync();
}
catch
{
}
var error = $"HTTP {(int)response.StatusCode} {Enum.GetName(typeof(HttpStatusCode), response.StatusCode)}";
if (!string.IsNullOrEmpty(content))
{
error = $"{error}: {content}";
}
throw new Exception(error);
}
return await response.Content.ReadAsStringAsync();
}
}
}

View File

@@ -0,0 +1,120 @@
using GitHub.DistributedTask.WebApi;
using System;
using System.Collections.Generic;
using System.Runtime.Serialization;
using System.Threading;
using System.Threading.Tasks;
using GitHub.Runner.Common.Util;
using GitHub.Services.WebApi;
using GitHub.Services.Common;
using GitHub.Runner.Sdk;
using GitHub.DistributedTask.Pipelines;
namespace GitHub.Runner.Common
{
[ServiceLocator(Default = typeof(RunServer))]
public interface IRunServer : IRunnerService
{
Task ConnectAsync(Uri serverUrl, VssCredentials credentials);
Task<AgentJobRequestMessage> GetJobMessageAsync(Guid scopeId, Guid hostId, string planType, string planGroup, Guid planId, IList<InstanceRef> instanceRefsJson);
}
public sealed class RunServer : RunnerService, IRunServer
{
private bool _hasConnection;
private VssConnection _connection;
private TaskAgentHttpClient _taskAgentClient;
public async Task ConnectAsync(Uri serverUrl, VssCredentials credentials)
{
// System.Console.WriteLine("RunServer.ConnectAsync");
_connection = await EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(100));
_taskAgentClient = _connection.GetClient<TaskAgentHttpClient>();
_hasConnection = true;
}
private async Task<VssConnection> EstablishVssConnection(Uri serverUrl, VssCredentials credentials, TimeSpan timeout)
{
// System.Console.WriteLine("EstablishVssConnection");
Trace.Info($"EstablishVssConnection");
Trace.Info($"Establish connection with {timeout.TotalSeconds} seconds timeout.");
int attemptCount = 5;
while (attemptCount-- > 0)
{
var connection = VssUtil.CreateConnection(serverUrl, credentials, timeout: timeout);
try
{
await connection.ConnectAsync();
return connection;
}
catch (Exception ex) when (attemptCount > 0)
{
Trace.Info($"Catch exception during connect. {attemptCount} attempt left.");
Trace.Error(ex);
await HostContext.Delay(TimeSpan.FromMilliseconds(100), CancellationToken.None);
}
}
// should never reach here.
throw new InvalidOperationException(nameof(EstablishVssConnection));
}
private void CheckConnection()
{
if (!_hasConnection)
{
throw new InvalidOperationException($"SetConnection");
}
}
public Task<AgentJobRequestMessage> GetJobMessageAsync(
Guid scopeId,
Guid hostId,
string planType,
string planGroup,
Guid planId,
IList<InstanceRef> instanceRefsJson)
{
// System.Console.WriteLine("RunServer.GetMessageAsync");
CheckConnection();
return _taskAgentClient.GetJobMessageAsync(scopeId, hostId, planType, planGroup, planId, StringUtil.ConvertToJson(instanceRefsJson, Newtonsoft.Json.Formatting.None));
}
}
// todo: move to SDK?
[DataContract]
public sealed class MessageRef
{
[DataMember(Name = "url")]
public string Url { get; set; }
[DataMember(Name = "token")]
public string Token { get; set; }
[DataMember(Name = "scopeId")]
public Guid ScopeId { get; set; }
[DataMember(Name = "hostId")]
public Guid HostId { get; set; }
[DataMember(Name = "planType")]
public string PlanType { get; set; }
[DataMember(Name = "planGroup")]
public string PlanGroup { get; set; }
[DataMember(Name = "planId")]
public Guid PlanId { get; set; }
[DataMember(Name = "instanceRefs")]
public InstanceRef[] InstanceRefs { get; set; }
[DataMember(Name = "labels")]
public string[] Labels { get; set; }
}
[DataContract]
public sealed class InstanceRef
{
[DataMember(Name = "name")]
public string Name { get; set; }
[DataMember(Name = "instanceType")]
public string InstanceType { get; set; }
[DataMember(Name = "attempt")]
public int Attempt { get; set; }
}
}

View File

@@ -44,7 +44,7 @@ namespace GitHub.Runner.Common
// job request
Task<TaskAgentJobRequest> GetAgentRequestAsync(int poolId, long requestId, CancellationToken cancellationToken);
Task<TaskAgentJobRequest> RenewAgentRequestAsync(int poolId, long requestId, Guid lockToken, string orchestrationId, CancellationToken cancellationToken);
Task<TaskAgentJobRequest> FinishAgentRequestAsync(int poolId, long requestId, Guid lockToken, DateTime finishTime, TaskResult result, CancellationToken cancellationToken);
Task<TaskAgentJobRequest> FinishAgentRequestAsync(int poolId, long requestId, Guid lockToken, DateTime finishTime, TaskResult result, Guid targetHostId, CancellationToken cancellationToken);
// agent package
Task<List<PackageMetadata>> GetPackagesAsync(string packageType, string platform, int top, bool includeToken, CancellationToken cancellationToken);
@@ -68,11 +68,23 @@ namespace GitHub.Runner.Common
public async Task ConnectAsync(Uri serverUrl, VssCredentials credentials)
{
var createGenericConnection = EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(100));
// System.Console.WriteLine("RunnerServer.ConnectAsync: Create message connection");
var createMessageConnection = EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(60));
var createRequestConnection = EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(60));
await Task.WhenAll(createMessageConnection);
await Task.WhenAll(createGenericConnection, createMessageConnection, createRequestConnection);
// System.Console.WriteLine("RunnerServer.ConnectAsync: Create generic connection");
var createGenericConnection = EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(100));
await Task.WhenAll(createGenericConnection);
// System.Console.WriteLine("RunnerServer.ConnectAsync: Create request connection");
var createRequestConnection = EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(60));
await Task.WhenAll(createRequestConnection);
// var createGenericConnection = EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(100));
// var createMessageConnection = EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(60));
// var createRequestConnection = EstablishVssConnection(serverUrl, credentials, TimeSpan.FromSeconds(60));
// await Task.WhenAll(createGenericConnection, createMessageConnection, createRequestConnection);
_genericConnection = await createGenericConnection;
_messageConnection = await createMessageConnection;
@@ -182,6 +194,8 @@ namespace GitHub.Runner.Common
private async Task<VssConnection> EstablishVssConnection(Uri serverUrl, VssCredentials credentials, TimeSpan timeout)
{
// System.Console.WriteLine("EstablishVssConnection");
Trace.Info($"EstablishVssConnection");
Trace.Info($"Establish connection with {timeout.TotalSeconds} seconds timeout.");
int attemptCount = 5;
while (attemptCount-- > 0)
@@ -238,41 +252,48 @@ namespace GitHub.Runner.Common
public Task<List<TaskAgentPool>> GetAgentPoolsAsync(string agentPoolName = null, TaskAgentPoolType poolType = TaskAgentPoolType.Automation)
{
// System.Console.WriteLine("RunnerServer.GetAgentPoolsAsync");
CheckConnection(RunnerConnectionType.Generic);
return _genericTaskAgentClient.GetAgentPoolsAsync(agentPoolName, poolType: poolType);
}
public Task<TaskAgent> AddAgentAsync(Int32 agentPoolId, TaskAgent agent)
{
// System.Console.WriteLine("RunnerServer.AddAgentAsync");
CheckConnection(RunnerConnectionType.Generic);
return _genericTaskAgentClient.AddAgentAsync(agentPoolId, agent);
}
public Task<List<TaskAgent>> GetAgentsAsync(int agentPoolId, string agentName = null)
{
// System.Console.WriteLine("RunnerServer.GetAgentsAsync 1");
CheckConnection(RunnerConnectionType.Generic);
return _genericTaskAgentClient.GetAgentsAsync(agentPoolId, agentName, false);
}
public Task<List<TaskAgent>> GetAgentsAsync(string agentName)
{
// System.Console.WriteLine("RunnerServer.GetAgentsAsync 2");
return GetAgentsAsync(0, agentName); // search in all all agentPools
}
public Task<TaskAgent> ReplaceAgentAsync(int agentPoolId, TaskAgent agent)
{
// System.Console.WriteLine("RunnerServer.ReplaceAgentAsync");
CheckConnection(RunnerConnectionType.Generic);
return _genericTaskAgentClient.ReplaceAgentAsync(agentPoolId, agent);
}
public Task DeleteAgentAsync(int agentPoolId, int agentId)
{
// System.Console.WriteLine("RunnerServer.DeleteAgentAsync");
CheckConnection(RunnerConnectionType.Generic);
return _genericTaskAgentClient.DeleteAgentAsync(agentPoolId, agentId);
}
public Task DeleteAgentAsync(int agentId)
{
// System.Console.WriteLine("RunnerServer.DeleteAgentAsync");
return DeleteAgentAsync(0, agentId); // agentPool is ignored server side
}
@@ -282,24 +303,28 @@ namespace GitHub.Runner.Common
public Task<TaskAgentSession> CreateAgentSessionAsync(Int32 poolId, TaskAgentSession session, CancellationToken cancellationToken)
{
// System.Console.WriteLine("RunnerServer.CreateAgentSessionAsync");
CheckConnection(RunnerConnectionType.MessageQueue);
return _messageTaskAgentClient.CreateAgentSessionAsync(poolId, session, cancellationToken: cancellationToken);
}
public Task DeleteAgentMessageAsync(Int32 poolId, Int64 messageId, Guid sessionId, CancellationToken cancellationToken)
{
// System.Console.WriteLine("RunnerServer.DeleteAgentMessageAsync");
CheckConnection(RunnerConnectionType.MessageQueue);
return _messageTaskAgentClient.DeleteMessageAsync(poolId, messageId, sessionId, cancellationToken: cancellationToken);
}
public Task DeleteAgentSessionAsync(Int32 poolId, Guid sessionId, CancellationToken cancellationToken)
{
// System.Console.WriteLine("RunnerServer.DeleteAgentSessionAsync");
CheckConnection(RunnerConnectionType.MessageQueue);
return _messageTaskAgentClient.DeleteAgentSessionAsync(poolId, sessionId, cancellationToken: cancellationToken);
}
public Task<TaskAgentMessage> GetAgentMessageAsync(Int32 poolId, Guid sessionId, Int64? lastMessageId, CancellationToken cancellationToken)
{
// System.Console.WriteLine("RunnerServer.GetAgentMessageAsync");
CheckConnection(RunnerConnectionType.MessageQueue);
return _messageTaskAgentClient.GetMessageAsync(poolId, sessionId, lastMessageId, cancellationToken: cancellationToken);
}
@@ -310,18 +335,21 @@ namespace GitHub.Runner.Common
public Task<TaskAgentJobRequest> RenewAgentRequestAsync(int poolId, long requestId, Guid lockToken, string orchestrationId = null, CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("RunnerServer.RenewAgentRequestAsync");
CheckConnection(RunnerConnectionType.JobRequest);
return _requestTaskAgentClient.RenewAgentRequestAsync(poolId, requestId, lockToken, orchestrationId: orchestrationId, cancellationToken: cancellationToken);
}
public Task<TaskAgentJobRequest> FinishAgentRequestAsync(int poolId, long requestId, Guid lockToken, DateTime finishTime, TaskResult result, CancellationToken cancellationToken = default(CancellationToken))
public Task<TaskAgentJobRequest> FinishAgentRequestAsync(int poolId, long requestId, Guid lockToken, DateTime finishTime, TaskResult result, Guid targetHostId, CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("RunnerServer.FinishAgentRequestAsync");
CheckConnection(RunnerConnectionType.JobRequest);
return _requestTaskAgentClient.FinishAgentRequestAsync(poolId, requestId, lockToken, finishTime, result, cancellationToken: cancellationToken);
return _requestTaskAgentClient.FinishAgentRequestAsync(poolId, requestId, lockToken, finishTime, result, targetHostId, cancellationToken: cancellationToken);
}
public Task<TaskAgentJobRequest> GetAgentRequestAsync(int poolId, long requestId, CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("RunnerServer.GetAgentRequestAsync");
CheckConnection(RunnerConnectionType.JobRequest);
return _requestTaskAgentClient.GetAgentRequestAsync(poolId, requestId, cancellationToken: cancellationToken);
}
@@ -331,18 +359,21 @@ namespace GitHub.Runner.Common
//-----------------------------------------------------------------
public Task<List<PackageMetadata>> GetPackagesAsync(string packageType, string platform, int top, bool includeToken, CancellationToken cancellationToken)
{
// System.Console.WriteLine("RunnerServer.GetPackagesAsync");
CheckConnection(RunnerConnectionType.Generic);
return _genericTaskAgentClient.GetPackagesAsync(packageType, platform, top, includeToken, cancellationToken: cancellationToken);
}
public Task<PackageMetadata> GetPackageAsync(string packageType, string platform, string version, bool includeToken, CancellationToken cancellationToken)
{
// System.Console.WriteLine("RunnerServer.GetPackageAsync");
CheckConnection(RunnerConnectionType.Generic);
return _genericTaskAgentClient.GetPackageAsync(packageType, platform, version, includeToken, cancellationToken: cancellationToken);
}
public Task<TaskAgent> UpdateAgentUpdateStateAsync(int agentPoolId, int agentId, string currentState, string trace)
{
// System.Console.WriteLine("RunnerServer.UpdateAgentUpdateStateAsync");
CheckConnection(RunnerConnectionType.Generic);
return _genericTaskAgentClient.UpdateAgentUpdateStateAsync(agentPoolId, agentId, currentState, trace);
}

View File

@@ -40,6 +40,12 @@ namespace GitHub.Runner.Listener.Configuration
return creds;
}
#if USE_BROKER
public VssCredentials LoadCredentials()
{
return new VssCredentials();
}
#else
public VssCredentials LoadCredentials()
{
IConfigurationStore store = HostContext.GetService<IConfigurationStore>();
@@ -69,6 +75,7 @@ namespace GitHub.Runner.Listener.Configuration
return creds;
}
#endif
}
[DataContract]

View File

@@ -23,7 +23,7 @@ namespace GitHub.Runner.Listener
{
bool Busy { get; }
TaskCompletionSource<bool> RunOnceJobCompleted { get; }
void Run(Pipelines.AgentJobRequestMessage message, bool runOnce = false);
void Run(Guid targetHostId, Pipelines.AgentJobRequestMessage message, bool runOnce = false);
bool Cancel(JobCancelMessage message);
Task WaitAsync(CancellationToken token);
Task ShutdownAsync();
@@ -79,7 +79,7 @@ namespace GitHub.Runner.Listener
public bool Busy { get; private set; }
public void Run(Pipelines.AgentJobRequestMessage jobRequestMessage, bool runOnce = false)
public void Run(Guid targetHostId, Pipelines.AgentJobRequestMessage jobRequestMessage, bool runOnce = false)
{
Trace.Info($"Job request {jobRequestMessage.RequestId} for plan {jobRequestMessage.Plan.PlanId} job {jobRequestMessage.JobId} received.");
@@ -112,11 +112,11 @@ namespace GitHub.Runner.Listener
if (runOnce)
{
Trace.Info("Start dispatcher for one time used runner.");
newDispatch.WorkerDispatch = RunOnceAsync(jobRequestMessage, orchestrationId, currentDispatch, newDispatch.WorkerCancellationTokenSource.Token, newDispatch.WorkerCancelTimeoutKillTokenSource.Token);
newDispatch.WorkerDispatch = RunOnceAsync(targetHostId, jobRequestMessage, orchestrationId, currentDispatch, newDispatch.WorkerCancellationTokenSource.Token, newDispatch.WorkerCancelTimeoutKillTokenSource.Token);
}
else
{
newDispatch.WorkerDispatch = RunAsync(jobRequestMessage, orchestrationId, currentDispatch, newDispatch.WorkerCancellationTokenSource.Token, newDispatch.WorkerCancelTimeoutKillTokenSource.Token);
newDispatch.WorkerDispatch = RunAsync(targetHostId, jobRequestMessage, orchestrationId, currentDispatch, newDispatch.WorkerCancellationTokenSource.Token, newDispatch.WorkerCancelTimeoutKillTokenSource.Token);
}
_jobInfos.TryAdd(newDispatch.JobId, newDispatch);
@@ -317,11 +317,11 @@ namespace GitHub.Runner.Listener
}
}
private async Task RunOnceAsync(Pipelines.AgentJobRequestMessage message, string orchestrationId, WorkerDispatcher previousJobDispatch, CancellationToken jobRequestCancellationToken, CancellationToken workerCancelTimeoutKillToken)
private async Task RunOnceAsync(Guid targetHostId, Pipelines.AgentJobRequestMessage message, string orchestrationId, WorkerDispatcher previousJobDispatch, CancellationToken jobRequestCancellationToken, CancellationToken workerCancelTimeoutKillToken)
{
try
{
await RunAsync(message, orchestrationId, previousJobDispatch, jobRequestCancellationToken, workerCancelTimeoutKillToken);
await RunAsync(targetHostId, message, orchestrationId, previousJobDispatch, jobRequestCancellationToken, workerCancelTimeoutKillToken);
}
finally
{
@@ -330,7 +330,7 @@ namespace GitHub.Runner.Listener
}
}
private async Task RunAsync(Pipelines.AgentJobRequestMessage message, string orchestrationId, WorkerDispatcher previousJobDispatch, CancellationToken jobRequestCancellationToken, CancellationToken workerCancelTimeoutKillToken)
private async Task RunAsync(Guid targetHostId, Pipelines.AgentJobRequestMessage message, string orchestrationId, WorkerDispatcher previousJobDispatch, CancellationToken jobRequestCancellationToken, CancellationToken workerCancelTimeoutKillToken)
{
Busy = true;
try
@@ -383,7 +383,7 @@ namespace GitHub.Runner.Listener
await renewJobRequest;
// complete job request with result Cancelled
await CompleteJobRequestAsync(_poolId, message, lockToken, TaskResult.Canceled);
await CompleteJobRequestAsync(targetHostId, _poolId, message, lockToken, TaskResult.Canceled);
return;
}
@@ -540,7 +540,7 @@ namespace GitHub.Runner.Listener
await renewJobRequest;
// complete job request
await CompleteJobRequestAsync(_poolId, message, lockToken, result, detailInfo);
await CompleteJobRequestAsync(targetHostId, _poolId, message, lockToken, result, detailInfo);
// print out unhandled exception happened in worker after we complete job request.
// when we run out of disk space, report back to server has higher priority.
@@ -637,7 +637,7 @@ namespace GitHub.Runner.Listener
await renewJobRequest;
// complete job request
await CompleteJobRequestAsync(_poolId, message, lockToken, resultOnAbandonOrCancel);
await CompleteJobRequestAsync(targetHostId, _poolId, message, lockToken, resultOnAbandonOrCancel);
}
finally
{
@@ -666,17 +666,25 @@ namespace GitHub.Runner.Listener
{
try
{
request = await runnerServer.RenewAgentRequestAsync(poolId, requestId, lockToken, orchestrationId, token);
Trace.Info($"Successfully renew job request {requestId}, job is valid till {request.LockedUntil.Value}");
// #if USE_BROKER
if (!firstJobRequestRenewed.Task.IsCompleted)
{
// fire first renew succeed event.
firstJobRequestRenewed.TrySetResult(0);
// Update settings if the runner name has been changed server-side
UpdateAgentNameIfNeeded(request.ReservedAgent?.Name);
}
// #else
// request = await runnerServer.RenewAgentRequestAsync(poolId, requestId, lockToken, orchestrationId, token);
// Trace.Info($"Successfully renew job request {requestId}, job is valid till {request?.LockedUntil.Value}");
// if (!firstJobRequestRenewed.Task.IsCompleted)
// {
// // fire first renew succeed event.
// firstJobRequestRenewed.TrySetResult(0);
// // Update settings if the runner name has been changed server-side
// UpdateAgentNameIfNeeded(request.ReservedAgent?.Name);
// }
// #endif
if (encounteringError > 0)
{
@@ -911,7 +919,7 @@ namespace GitHub.Runner.Listener
}
}
private async Task CompleteJobRequestAsync(int poolId, Pipelines.AgentJobRequestMessage message, Guid lockToken, TaskResult result, string detailInfo = null)
private async Task CompleteJobRequestAsync(Guid targetHostId, int poolId, Pipelines.AgentJobRequestMessage message, Guid lockToken, TaskResult result, string detailInfo = null)
{
Trace.Entering();
@@ -928,7 +936,7 @@ namespace GitHub.Runner.Listener
{
try
{
await runnerServer.FinishAgentRequestAsync(poolId, message.RequestId, lockToken, DateTime.UtcNow, result, CancellationToken.None);
await runnerServer.FinishAgentRequestAsync(poolId, message.RequestId, lockToken, DateTime.UtcNow, result, targetHostId, CancellationToken.None);
return;
}
catch (TaskAgentJobNotFoundException)

View File

@@ -2,7 +2,11 @@ using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Net;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Runtime.InteropServices;
using System.Runtime.Serialization;
using System.Security.Cryptography;
using System.Text;
using System.Threading;
@@ -27,10 +31,13 @@ namespace GitHub.Runner.Listener
public sealed class MessageListener : RunnerService, IMessageListener
{
#if !USE_BROKER
private long? _lastMessageId;
#endif
private RunnerSettings _settings;
private ITerminal _term;
private IRunnerServer _runnerServer;
private IBrokerServer _brokerServer;
private TaskAgentSession _session;
private TimeSpan _getNextMessageRetryInterval;
private bool _accessTokenRevoked = false;
@@ -45,8 +52,44 @@ namespace GitHub.Runner.Listener
_term = HostContext.GetService<ITerminal>();
_runnerServer = HostContext.GetService<IRunnerServer>();
_brokerServer = HostContext.GetService<IBrokerServer>();
}
#if USE_BROKER
public async Task<Boolean> CreateSessionAsync(CancellationToken token)
{
Trace.Entering();
// Settings
var configManager = HostContext.GetService<IConfigurationManager>();
_settings = configManager.LoadSettings();
var serverUrl = _settings.ServerUrl;
Trace.Info(_settings);
// Connect
token.ThrowIfCancellationRequested();
Trace.Info($"Attempt to create session.");
Trace.Info("Connecting to the Runner Server...");
_term.WriteLine($"Connecting to {new Uri(serverUrl)}");
await _brokerServer.ConnectAsync(new Uri(serverUrl), token);
_term.WriteLine();
_term.WriteSuccessMessage("Connected to GitHub");
_term.WriteLine();
// Session info
var agent = new TaskAgentReference
{
Id = _settings.AgentId,
Name = _settings.AgentName,
Version = BuildConstants.RunnerPackage.Version,
OSDescription = RuntimeInformation.OSDescription,
};
string sessionName = $"{Environment.MachineName ?? "RUNNER"}";
_session = new TaskAgentSession(sessionName, agent);
return true;
}
#else
public async Task<Boolean> CreateSessionAsync(CancellationToken token)
{
Trace.Entering();
@@ -81,6 +124,7 @@ namespace GitHub.Runner.Listener
Trace.Info($"Attempt to create session.");
try
{
Trace.Info("Connecting to the Runner Server...");
Trace.Info("Connecting to the Runner Server...");
await _runnerServer.ConnectAsync(new Uri(serverUrl), creds);
Trace.Info("VssConnection created");
@@ -151,6 +195,7 @@ namespace GitHub.Runner.Listener
}
}
}
#endif
public async Task DeleteSessionAsync()
{
@@ -170,6 +215,167 @@ namespace GitHub.Runner.Listener
}
}
#if USE_BROKER
[DataContract]
public sealed class MessageRef
{
[DataMember(Name = "url")]
public string Url { get; set; }
[DataMember(Name = "token")]
public string Token { get; set; }
[DataMember(Name = "scopeId")]
public string ScopeId { get; set; }
[DataMember(Name = "planType")]
public string PlanType { get; set; }
[DataMember(Name = "planGroup")]
public string PlanGroup { get; set; }
[DataMember(Name = "instanceRefs")]
public InstanceRef[] InstanceRefs { get; set; }
[DataMember(Name = "labels")]
public string[] Labels { get; set; }
}
[DataContract]
public sealed class InstanceRef
{
[DataMember(Name = "name")]
public string Name { get; set; }
[DataMember(Name = "instanceType")]
public string InstanceType { get; set; }
[DataMember(Name = "attempt")]
public int Attempt { get; set; }
}
public async Task<TaskAgentMessage> GetNextMessageAsync(CancellationToken token)
{
Trace.Entering();
ArgUtil.NotNull(_session, nameof(_session));
ArgUtil.NotNull(_settings, nameof(_settings));
bool encounteringError = false;
int continuousError = 0;
string errorMessage = string.Empty;
Stopwatch heartbeat = new Stopwatch();
heartbeat.Restart();
while (true)
{
token.ThrowIfCancellationRequested();
string message = null;
try
{
message = await _brokerServer.GetMessageAsync(_session, _settings, null/*_lastMessageId*/, token);
_term.WriteLine($"{DateTime.UtcNow:u}: {message}");
if (!string.IsNullOrEmpty(message))
{
var messageRef = StringUtil.ConvertFromJson<MessageRef>(message);
var client = new HttpClient();
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", messageRef.Token);
var response = await client.GetAsync(messageRef.Url, token);
if (!response.IsSuccessStatusCode)
{
var content = default(string);
try
{
content = await response.Content.ReadAsStringAsync();
}
catch
{
}
var error = $"HTTP {(int)response.StatusCode} {Enum.GetName(typeof(HttpStatusCode), response.StatusCode)}";
if (!string.IsNullOrEmpty(content))
{
error = $"{error}: {content}";
}
throw new Exception(error);
}
var fullMessage = await response.Content.ReadAsStringAsync();
return StringUtil.ConvertFromJson<TaskAgentMessage>(fullMessage);
}
if (encounteringError) //print the message once only if there was an error
{
_term.WriteLine($"{DateTime.UtcNow:u}: Runner reconnected.");
encounteringError = false;
continuousError = 0;
}
}
catch (OperationCanceledException) when (token.IsCancellationRequested)
{
Trace.Info("Get next message has been cancelled.");
throw;
}
catch (TaskAgentAccessTokenExpiredException)
{
Trace.Info("Runner OAuth token has been revoked. Unable to pull message.");
_accessTokenRevoked = true;
throw;
}
catch (Exception ex)
{
Trace.Error("Catch exception during get next message.");
Trace.Error(ex);
// don't retry if SkipSessionRecover = true, DT service will delete agent session to stop agent from taking more jobs.
if (ex is TaskAgentSessionExpiredException && !_settings.SkipSessionRecover && await CreateSessionAsync(token))
{
Trace.Info($"{nameof(TaskAgentSessionExpiredException)} received, recovered by recreate session.");
}
else if (!IsGetNextMessageExceptionRetriable(ex))
{
throw;
}
else
{
continuousError++;
//retry after a random backoff to avoid service throttling
//in case of there is a service error happened and all agents get kicked off of the long poll and all agent try to reconnect back at the same time.
if (continuousError <= 5)
{
// random backoff [15, 30]
_getNextMessageRetryInterval = BackoffTimerHelper.GetRandomBackoff(TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30), _getNextMessageRetryInterval);
}
else
{
// more aggressive backoff [30, 60]
_getNextMessageRetryInterval = BackoffTimerHelper.GetRandomBackoff(TimeSpan.FromSeconds(30), TimeSpan.FromSeconds(60), _getNextMessageRetryInterval);
}
if (!encounteringError)
{
//print error only on the first consecutive error
_term.WriteError($"{DateTime.UtcNow:u}: Runner connect error: {ex.Message}. Retrying until reconnected.");
encounteringError = true;
}
// re-create VssConnection before next retry
await _runnerServer.RefreshConnectionAsync(RunnerConnectionType.MessageQueue, TimeSpan.FromSeconds(60));
Trace.Info("Sleeping for {0} seconds before retrying.", _getNextMessageRetryInterval.TotalSeconds);
await HostContext.Delay(_getNextMessageRetryInterval, token);
}
}
// if (message == null)
// {
// if (heartbeat.Elapsed > TimeSpan.FromMinutes(30))
// {
// Trace.Info($"No message retrieved from session '{_session.SessionId}' within last 30 minutes.");
// heartbeat.Restart();
// }
// else
// {
// Trace.Verbose($"No message retrieved from session '{_session.SessionId}'.");
// }
// continue;
// }
// Trace.Info($"Message '{message.MessageId}' received from session '{_session.SessionId}'.");
// return message;
}
}
#else
public async Task<TaskAgentMessage> GetNextMessageAsync(CancellationToken token)
{
Trace.Entering();
@@ -281,6 +487,7 @@ namespace GitHub.Runner.Listener
return message;
}
}
#endif
public async Task DeleteMessageAsync(TaskAgentMessage message)
{

View File

@@ -13,6 +13,10 @@ using GitHub.Runner.Sdk;
using System.Linq;
using GitHub.Runner.Listener.Check;
using System.Collections.Generic;
using System.Runtime.Serialization;
using System.Net;
using System.Net.Http;
using System.Net.Http.Headers;
namespace GitHub.Runner.Listener
{
@@ -449,7 +453,37 @@ namespace GitHub.Runner.Listener
{
Trace.Info($"Received job message of length {message.Body.Length} from service, with hash '{IOUtil.GetSha256Hash(message.Body)}'");
var jobMessage = StringUtil.ConvertFromJson<Pipelines.AgentJobRequestMessage>(message.Body);
jobDispatcher.Run(jobMessage, runOnce);
jobDispatcher.Run(Guid.Empty, jobMessage, runOnce);
if (runOnce)
{
Trace.Info("One time used runner received job message.");
runOnceJobReceived = true;
}
}
}
// Broker flow
else if (string.Equals(message.MessageType, JobRequestMessageTypes.RunnerJobRequest, StringComparison.OrdinalIgnoreCase))
{
if (autoUpdateInProgress || runOnceJobReceived)
{
skipMessageDeletion = true;
Trace.Info($"Skip message deletion for job request message '{message.MessageId}'.");
}
else
{
var messageRef = StringUtil.ConvertFromJson<MessageRef>(message.Body);
// Create connection
var credMgr = HostContext.GetService<ICredentialManager>();
var creds = credMgr.LoadCredentials();
// todo: add retries
var runServer = HostContext.CreateService<IRunServer>();
await runServer.ConnectAsync(new Uri(settings.ServerUrl), creds);
var jobMessage = await runServer.GetJobMessageAsync(messageRef.ScopeId, messageRef.HostId, messageRef.PlanType, messageRef.PlanGroup, messageRef.PlanId, messageRef.InstanceRefs);
// todo: Trace.Info($"Received job message of length {message.Body.Length} from service, with hash '{IOUtil.GetSha256Hash(message.Body)}'");
jobDispatcher.Run(messageRef.HostId, jobMessage, runOnce);
if (runOnce)
{
Trace.Info("One time used runner received job message.");

View File

@@ -36,6 +36,7 @@ namespace GitHub.Runner.Sdk
public static VssConnection CreateConnection(Uri serverUri, VssCredentials credentials, IEnumerable<DelegatingHandler> additionalDelegatingHandler = null, TimeSpan? timeout = null)
{
// System.Console.WriteLine("VssUtil.CreateConnection");
VssClientHttpRequestSettings settings = VssClientHttpRequestSettings.Default.Clone();
int maxRetryRequest;

View File

@@ -817,6 +817,7 @@ namespace GitHub.Runner.Worker
{
// Something else bad happened, let's go to our retry logic
response.EnsureSuccessStatusCode();
throw new Exception("Unexpected response code: " + response.StatusCode);
}
}
}

View File

@@ -18,6 +18,7 @@ namespace GitHub.Services.Common
public override bool IsAuthenticationChallenge(IHttpResponse webResponse)
{
// System.Console.WriteLine($"FederatedCredential.IsAuthenticationChallenge");
if (webResponse == null)
{
return false;

View File

@@ -88,6 +88,7 @@ namespace GitHub.Services.Common
IHttpResponse response,
IssuedToken failedToken)
{
// System.Console.WriteLine("IssuedTokenCredential.CreateTokenProvider");
if (response != null && !IsAuthenticationChallenge(response))
{
throw new InvalidOperationException();
@@ -99,12 +100,14 @@ namespace GitHub.Services.Common
{
throw new InvalidOperationException($"The {nameof(TokenStorageUrl)} property must have a value if the {nameof(Storage)} property is set on this instance of {GetType().Name}.");
}
// System.Console.WriteLine($"IssuedTokenCredential.CreateTokenProvider: TokenStorageUrl: {TokenStorageUrl}");
InitialToken = Storage.RetrieveToken(TokenStorageUrl, CredentialType);
}
IssuedTokenProvider provider = OnCreateTokenProvider(serverUrl, response);
if (provider != null)
{
// System.Console.WriteLine($"IssuedTokenCredential.CreateTokenProvider: provider: {provider}");
provider.TokenStorageUrl = TokenStorageUrl;
}
@@ -123,6 +126,7 @@ namespace GitHub.Services.Common
internal virtual string GetAuthenticationChallenge(IHttpResponse webResponse)
{
// System.Console.WriteLine($"IssuedTokenCredential.GetAuthenticationChallenge");
IEnumerable<String> values;
if (!webResponse.Headers.TryGetValues(Internal.HttpHeaders.WwwAuthenticate, out values))
{

View File

@@ -108,6 +108,7 @@ namespace GitHub.Services.Common
TaskScheduler scheduler,
IVssCredentialPrompt credentialPrompt)
{
// System.Console.WriteLine($"VssCredentials.ctor");
this.PromptType = promptType;
if (promptType == CredentialPromptType.PromptIfNeeded && scheduler == null)
@@ -150,6 +151,7 @@ namespace GitHub.Services.Common
{
get
{
// System.Console.WriteLine($"VssCredentials.get_PromptType");
return m_promptType;
}
set
@@ -170,6 +172,7 @@ namespace GitHub.Services.Common
{
get
{
// System.Console.WriteLine($"VssCredentials.get_Federated");
return m_federatedCredential;
}
}
@@ -184,6 +187,7 @@ namespace GitHub.Services.Common
{
get
{
// System.Console.WriteLine($"VssCredentials.get_Storage");
return m_credentialStorage;
}
set
@@ -203,6 +207,7 @@ namespace GitHub.Services.Common
/// </summary>
internal virtual bool TryGetValidAdalToken(IVssCredentialPrompt prompt)
{
// System.Console.WriteLine($"VssCredentials.TryGetValidAdalToken");
return false;
}
@@ -218,6 +223,7 @@ namespace GitHub.Services.Common
IHttpResponse webResponse,
IssuedToken failedToken)
{
// System.Console.WriteLine("VssCredential.CreateTokenProvider");
ArgumentUtility.CheckForNull(serverUrl, "serverUrl");
IssuedTokenProvider tokenProvider = null;
@@ -263,6 +269,7 @@ namespace GitHub.Services.Common
Uri serverUrl,
out IssuedTokenProvider provider)
{
// System.Console.WriteLine($"VssCredentials.TryGetTokenProvider");
ArgumentUtility.CheckForNull(serverUrl, "serverUrl");
lock (m_thisLock)
@@ -272,11 +279,13 @@ namespace GitHub.Services.Common
{
if (m_federatedCredential != null)
{
// System.Console.WriteLine($"VssCredentials.TryGetTokenProvider: Using federated credential");
m_currentProvider = m_federatedCredential.CreateTokenProvider(serverUrl, null, null);
}
if (m_currentProvider != null)
{
// System.Console.WriteLine($"VssCredentials.TryGetTokenProvider: Issued token provider created");
VssHttpEventSource.Log.IssuedTokenProviderCreated(VssTraceActivity.Current, m_currentProvider);
}
}
@@ -294,6 +303,7 @@ namespace GitHub.Services.Common
/// <returns>True if this is an token authentication redirect, false otherwise</returns>
internal bool IsAuthenticationChallenge(IHttpResponse webResponse)
{
// System.Console.WriteLine($"VssCredentials.IsAuthenticationChallenge");
if (webResponse == null)
{
return false;
@@ -313,6 +323,7 @@ namespace GitHub.Services.Common
Uri serviceLocation,
string identityProvider)
{
// System.Console.WriteLine($"VssCredentials.SignOut");
// Remove the token in the storage and the current token provider. Note that we don't
// call InvalidateToken here because we want to remove the whole token not just its value
if ((m_currentProvider != null) && (m_currentProvider.CurrentToken != null))
@@ -349,6 +360,7 @@ namespace GitHub.Services.Common
string token,
IDictionary<string, string> attributes)
{
// System.Console.WriteLine($"VssCredentials.WriteAuthorizationToken");
int i = 0;
for (int j = 0; j < token.Length; i++, j += 128)
{
@@ -360,6 +372,7 @@ namespace GitHub.Services.Common
protected static string ReadAuthorizationToken(IDictionary<string, string> attributes)
{
// System.Console.WriteLine($"VssCredentials.ReadAuthorizationToken");
string authTokenCountValue;
if (attributes.TryGetValue("AuthTokenSegmentCount", out authTokenCountValue))
{

View File

@@ -49,6 +49,7 @@ namespace GitHub.Services.Common
VssHttpRequestSettings settings,
HttpMessageHandler innerHandler)
{
// System.Console.WriteLine("VssHttpMessageHandler.ctor");
this.Credentials = credentials;
this.Settings = settings;
this.ExpectContinue = settings.ExpectContinue;
@@ -122,6 +123,7 @@ namespace GitHub.Services.Common
HttpRequestMessage request,
CancellationToken cancellationToken)
{
// System.Console.WriteLine("VssHttpMessageHandler.SendAsync");
VssTraceActivity traceActivity = VssTraceActivity.Current;
var traceInfo = VssHttpMessageHandlerTraceInfo.GetTraceInfo(request);
@@ -130,6 +132,7 @@ namespace GitHub.Services.Common
if (!m_appliedClientCertificatesToTransportHandler &&
request.RequestUri.Scheme == "https")
{
// System.Console.WriteLine("VssHttpMessageHandler.SendAsync: !appliedClientCertificatesToTransportHandler");
HttpClientHandler httpClientHandler = m_transportHandler as HttpClientHandler;
if (httpClientHandler != null &&
this.Settings.ClientCertificateManager != null &&
@@ -144,6 +147,7 @@ namespace GitHub.Services.Common
if (!m_appliedServerCertificateValidationCallbackToTransportHandler &&
request.RequestUri.Scheme == "https")
{
// System.Console.WriteLine("VssHttpMessageHandler.SendAsync: !appliedServerCertificateValidationCallbackToTransportHandler");
HttpClientHandler httpClientHandler = m_transportHandler as HttpClientHandler;
if (httpClientHandler != null &&
this.Settings.ServerCertificateValidationCallback != null)
@@ -165,6 +169,7 @@ namespace GitHub.Services.Common
IssuedTokenProvider provider;
if (this.Credentials.TryGetTokenProvider(request.RequestUri, out provider))
{
// System.Console.WriteLine("VssHttpMessageHandler.SendAsync: Got token provider from credentials");
token = provider.CurrentToken;
}
@@ -225,6 +230,7 @@ namespace GitHub.Services.Common
traceInfo?.TraceResponseContentTime();
// System.Console.WriteLine($"VssHttpMessageHandler.SendAsync: Creating response wrapper");
responseWrapper = new HttpResponseMessageWrapper(response);
if (!this.Credentials.IsAuthenticationChallenge(responseWrapper))
@@ -232,6 +238,7 @@ namespace GitHub.Services.Common
// Validate the token after it has been successfully authenticated with the server.
if (provider != null)
{
// System.Console.WriteLine("VssHttpMessageHandler.SendAsync: Validating token");
provider.ValidateToken(token, responseWrapper);
}
@@ -243,6 +250,7 @@ namespace GitHub.Services.Common
}
else
{
// System.Console.WriteLine($"VssHttpMessageHandler.SendAsync: Auth challenge. Response status code {response.StatusCode}; headers {response.Headers}");
// In the case of a Windows token, only apply it to the web proxy if it
// returned a 407 Proxy Authentication Required. If we didn't get this
// status code back, then the proxy (if there is one) is clearly working fine,
@@ -288,6 +296,7 @@ namespace GitHub.Services.Common
}
// Now invoke the provider and await the result
// System.Console.WriteLine($"VssHttpMessageHandler.SendAsync: Calling GetTokenAsync");
token = await provider.GetTokenAsync(token, tokenSource.Token).ConfigureAwait(false);
// I always see 0 here, but the method above could take more time so keep for now
@@ -432,6 +441,7 @@ namespace GitHub.Services.Common
activity != VssTraceActivity.Empty &&
!request.Headers.Contains(HttpHeaders.TfsSessionHeader))
{
// System.Console.WriteLine($"VssHttpMessageHandler.ApplyHeaders: Activity ID {activity.Id}");
request.Headers.Add(HttpHeaders.TfsSessionHeader, activity.Id.ToString("D"));
}
@@ -452,13 +462,16 @@ namespace GitHub.Services.Common
ICredentials credentialsToken = token as ICredentials;
if (credentialsToken != null)
{
// System.Console.WriteLine("VssHttpMessageHandler.ApplyToken: Credentials token != null");
if (applyICredentialsToWebProxy)
{
// System.Console.WriteLine("VssHttpMessageHandler.ApplyToken: Apply credentials to web proxy");
HttpClientHandler httpClientHandler = m_transportHandler as HttpClientHandler;
if (httpClientHandler != null &&
httpClientHandler.Proxy != null)
{
// System.Console.WriteLine("VssHttpMessageHandler.ApplyToken: Setting proxy crednetials");
httpClientHandler.Proxy.Credentials = credentialsToken;
}
}
@@ -467,6 +480,7 @@ namespace GitHub.Services.Common
}
else
{
// System.Console.WriteLine("VssHttpMessageHandler.ApplyToken: Applying credentials to request");
token.ApplyTo(new HttpRequestMessageWrapper(request));
}
}
@@ -479,7 +493,8 @@ namespace GitHub.Services.Common
HttpClientHandler httpClientHandler = handler as HttpClientHandler;
if (httpClientHandler != null)
{
httpClientHandler.AllowAutoRedirect = settings.AllowAutoRedirect;
// System.Console.WriteLine($"VssHttpMessageHandler.ApplySettings: Default credentials = {defaultCredentials} AllowAutoRedirect = {settings.AllowAutoRedirect}");
httpClientHandler.AllowAutoRedirect = true; //settings.AllowAutoRedirect;
httpClientHandler.ClientCertificateOptions = ClientCertificateOption.Manual;
//Setting httpClientHandler.UseDefaultCredentials to false in .Net Core, clears httpClientHandler.Credentials if
//credentials is already set to defaultcredentials. Therefore httpClientHandler.Credentials must be
@@ -550,6 +565,7 @@ namespace GitHub.Services.Common
Uri uri,
String authType)
{
// System.Console.WriteLine($"CredentialWrapper.GetCredential: InnerCredentials = {InnerCredentials}");
return InnerCredentials != null ? InnerCredentials.GetCredential(uri, authType) : null;
}
}

View File

@@ -27,6 +27,7 @@ using System.Net.Http.Headers;
using System.Net.Http.Formatting;
using System.Threading;
using System.Threading.Tasks;
using GitHub.DistributedTask.Pipelines;
using GitHub.Services.Common;
using GitHub.Services.WebApi;
@@ -378,6 +379,7 @@ namespace GitHub.DistributedTask.WebApi
/// <param name="requestId"></param>
/// <param name="lockToken"></param>
/// <param name="request"></param>
/// <param name="targetHostId"></param>
/// <param name="userState"></param>
/// <param name="cancellationToken">The cancellation token to cancel operation.</param>
[EditorBrowsable(EditorBrowsableState.Never)]
@@ -386,6 +388,7 @@ namespace GitHub.DistributedTask.WebApi
long requestId,
Guid lockToken,
TaskAgentJobRequest request,
Guid targetHostId,
object userState = null,
CancellationToken cancellationToken = default)
{
@@ -396,6 +399,7 @@ namespace GitHub.DistributedTask.WebApi
List<KeyValuePair<string, string>> queryParams = new List<KeyValuePair<string, string>>();
queryParams.Add("lockToken", lockToken.ToString());
queryParams.Add("targetHostId", targetHostId.ToString());
return SendAsync<TaskAgentJobRequest>(
httpMethod,
@@ -703,6 +707,47 @@ namespace GitHub.DistributedTask.WebApi
cancellationToken: cancellationToken);
}
/// <summary>
/// [Preview API]
/// </summary>
/// <param name="scopeId"></param>
/// <param name="planType"></param>
/// <param name="planGroup"></param>
/// <param name="planId"></param>
/// <param name="instanceRefsJson"></param>
/// <param name="userState"></param>
/// <param name="cancellationToken">The cancellation token to cancel operation.</param>
[EditorBrowsable(EditorBrowsableState.Never)]
public virtual Task<Pipelines.AgentJobRequestMessage> GetJobMessageAsync(
Guid scopeId,
Guid hostId,
string planType,
string planGroup,
Guid planId,
string instanceRefsJson,
object userState = null,
CancellationToken cancellationToken = default)
{
HttpMethod httpMethod = new HttpMethod("GET");
Guid locationId = new Guid("25adab70-1379-4186-be8e-b643061ebe3a");
List<KeyValuePair<string, string>> queryParams = new List<KeyValuePair<string, string>>();
queryParams.Add("scopeId", scopeId.ToString());
queryParams.Add("hostId", hostId.ToString());
queryParams.Add("planType", planType);
queryParams.Add("planGroup", planGroup);
queryParams.Add("planId", planId.ToString());
queryParams.Add("instanceRefsJson", instanceRefsJson);
return SendAsync<Pipelines.AgentJobRequestMessage>(
httpMethod,
locationId,
version: new ApiResourceVersion(6.0, 1),
queryParameters: queryParams,
userState: userState,
cancellationToken: cancellationToken);
}
/// <summary>
/// [Preview API]
/// </summary>
@@ -717,6 +762,7 @@ namespace GitHub.DistributedTask.WebApi
object userState = null,
CancellationToken cancellationToken = default)
{
// System.Console.WriteLine("TaskAgentHttpClientBase.CreateAgentSessionAsync");
HttpMethod httpMethod = new HttpMethod("POST");
Guid locationId = new Guid("134e239e-2df3-4794-a6f6-24f1f19ec8dc");
object routeValues = new { poolId = poolId };

View File

@@ -5,5 +5,6 @@ namespace GitHub.DistributedTask.WebApi
public static class JobRequestMessageTypes
{
public const String PipelineAgentJobRequest = "PipelineAgentJobRequest";
public const String RunnerJobRequest = "RunnerJobRequest";
}
}

View File

@@ -64,6 +64,7 @@ namespace GitHub.DistributedTask.WebApi
Guid lockToken,
DateTime finishTime,
TaskResult result,
Guid targetHostId,
Object userState = null,
CancellationToken cancellationToken = default(CancellationToken))
{
@@ -74,7 +75,7 @@ namespace GitHub.DistributedTask.WebApi
Result = result,
};
return UpdateAgentRequestAsync(poolId, requestId, lockToken, request, userState, cancellationToken);
return UpdateAgentRequestAsync(poolId, requestId, lockToken, request, targetHostId, userState, cancellationToken);
}
public Task<List<TaskAgent>> GetAgentsAsync(
@@ -152,6 +153,7 @@ namespace GitHub.DistributedTask.WebApi
CancellationToken cancellationToken = default(CancellationToken),
Func<HttpResponseMessage, CancellationToken, Task<T>> processResponse = null)
{
// System.Console.WriteLine("TaskAgentHttpClient.SendAsync 1");
return SendAsync<T>(method, null, locationId, routeValues, version, content, queryParameters, userState, cancellationToken, processResponse);
}
@@ -170,6 +172,7 @@ namespace GitHub.DistributedTask.WebApi
using (VssTraceActivity.GetOrCreate().EnterCorrelationScope())
using (HttpRequestMessage requestMessage = await CreateRequestMessageAsync(method, additionalHeaders, locationId, routeValues, version, content, queryParameters, userState, cancellationToken).ConfigureAwait(false))
{
// System.Console.WriteLine("TaskAgentHttpClient.SendAsync 2");
return await SendAsync<T>(requestMessage, userState, cancellationToken, processResponse).ConfigureAwait(false);
}
}
@@ -180,6 +183,7 @@ namespace GitHub.DistributedTask.WebApi
CancellationToken cancellationToken = default(CancellationToken),
Func<HttpResponseMessage, CancellationToken, Task<T>> processResponse = null)
{
// System.Console.WriteLine("TaskAgentHttpClient.SendAsync 3");
if (processResponse == null)
{
processResponse = ReadContentAsAsync<T>;

View File

@@ -801,6 +801,7 @@ namespace GitHub.Services.WebApi.Location
private async Task<ConnectionData> GetConnectionDataAsync(ConnectOptions connectOptions, int lastChangeId, CancellationToken cancellationToken)
{
// System.Console.WriteLine("ServerDataProvider.GetConnectionDataAsync");
int timeoutRetries = 1;
while (true)

View File

@@ -75,6 +75,7 @@ namespace GitHub.Services.OAuth
internal override void ApplyTo(IHttpRequest request)
{
// System.Console.WriteLine($"VssOAuthAccessToken.ApplyTo: Bearer {m_value}");
request.Headers.SetValue(Common.Internal.HttpHeaders.Authorization, $"Bearer {m_value}");
}

View File

@@ -60,6 +60,7 @@ namespace GitHub.Services.OAuth
Uri serverUrl,
IHttpResponse response)
{
// System.Console.WriteLine($"VssOAuthAccessTokenCredential.OnCreateTokenProvider");
return new VssOAuthAccessTokenProvider(this, serverUrl, null);
}
@@ -71,6 +72,7 @@ namespace GitHub.Services.OAuth
Uri signInUrl)
: base(credential, serverUrl, signInUrl)
{
// System.Console.WriteLine($"VssOAuthAccessTokenProvider.ctor");
}
public override Boolean GetTokenIsInteractive

View File

@@ -103,17 +103,23 @@ namespace GitHub.Services.OAuth
/// <returns>True if the web response indicates an authorization challenge; otherwise, false</returns>
public override Boolean IsAuthenticationChallenge(IHttpResponse webResponse)
{
// System.Console.WriteLine($"VssOAuthCredential.IsAuthenticationChallenge");
if (webResponse == null)
{
// System.Console.WriteLine($"VssOAuthCredential.IsAuthenticationChallenge: webResponse is null");
return false;
}
if (webResponse.StatusCode == HttpStatusCode.Found ||
webResponse.StatusCode == HttpStatusCode.Unauthorized)
{
return webResponse.Headers.GetValues(Common.Internal.HttpHeaders.WwwAuthenticate).Any(x => x.IndexOf("Bearer", StringComparison.OrdinalIgnoreCase) >= 0);
// System.Console.WriteLine($"VssOAuthCredential.IsAuthenticationChallenge: found or unauthorized");
var result = webResponse.Headers.GetValues(Common.Internal.HttpHeaders.WwwAuthenticate).Any(x => x.IndexOf("Bearer", StringComparison.OrdinalIgnoreCase) >= 0);
// System.Console.WriteLine($"VssOAuthCredential.IsAuthenticationChallenge: {result}");
return result;
}
// System.Console.WriteLine($"VssOAuthCredential.IsAuthenticationChallenge: false");
return false;
}
@@ -121,6 +127,7 @@ namespace GitHub.Services.OAuth
Uri serverUrl,
IHttpResponse response)
{
// System.Console.WriteLine($"VssOAuthCredential.OnCreateTokenProvider");
return new VssOAuthTokenProvider(this, serverUrl);
}

View File

@@ -54,6 +54,7 @@ namespace GitHub.Services.OAuth
VssOAuthTokenParameters tokenParameters = null,
CancellationToken cancellationToken = default(CancellationToken))
{
// todo: qqq
VssTraceActivity traceActivity = VssTraceActivity.Current;
using (HttpClient client = new HttpClient(CreateMessageHandler(this.AuthorizationUrl)))
{

View File

@@ -47,6 +47,7 @@ namespace GitHub.Services.OAuth
VssOAuthTokenParameters tokenParameters)
: base(credential, serverUrl, authorizationUrl)
{
// System.Console.WriteLine($"VssOAuthTokenProvider.ctor");
m_grant = grant;
m_tokenParameters = tokenParameters;
m_clientCredential = clientCrential;
@@ -59,6 +60,7 @@ namespace GitHub.Services.OAuth
{
get
{
// System.Console.WriteLine($"VssOAuthTokenProvider.get_Grant");
return m_grant;
}
}
@@ -70,6 +72,7 @@ namespace GitHub.Services.OAuth
{
get
{
// System.Console.WriteLine($"VssOAuthTokenProvider.get_ClientCredential");
return m_clientCredential;
}
}
@@ -81,6 +84,7 @@ namespace GitHub.Services.OAuth
{
get
{
// System.Console.WriteLine($"VssOAuthTokenProvider.get_TokenParameters");
return m_tokenParameters;
}
}
@@ -92,6 +96,7 @@ namespace GitHub.Services.OAuth
{
get
{
// System.Console.WriteLine($"VssOAuthTokenProvider.get_GetTokenIsInteractive");
return false;
}
}
@@ -100,6 +105,7 @@ namespace GitHub.Services.OAuth
{
get
{
// System.Console.WriteLine($"VssOAuthTokenProvider.get_AuthenticationParameter");
if (this.ClientCredential == null)
{
return null;
@@ -115,12 +121,14 @@ namespace GitHub.Services.OAuth
{
get
{
// System.Console.WriteLine($"VssOAuthTokenProvider.get_AuthenticationScheme");
return "Bearer";
}
}
public async Task<string> ValidateCredentialAsync(CancellationToken cancellationToken)
{
// System.Console.WriteLine($"VssOAuthTokenProvider.ValidateCredentialAsync: Calling VssOAuthTokenHttpClient.GetTokenAsync");
var tokenHttpClient = new VssOAuthTokenHttpClient(this.SignInUrl);
var tokenResponse = await tokenHttpClient.GetTokenAsync(this.Grant, this.ClientCredential, this.TokenParameters, cancellationToken);
@@ -139,6 +147,7 @@ namespace GitHub.Services.OAuth
IssuedToken failedToken,
CancellationToken cancellationToken)
{
// System.Console.WriteLine($"VssOAuthTokenProvider.OnGetTokenAsync");
if (this.SignInUrl == null ||
this.Grant == null ||
this.ClientCredential == null)
@@ -151,6 +160,7 @@ namespace GitHub.Services.OAuth
try
{
var tokenHttpClient = new VssOAuthTokenHttpClient(this.SignInUrl);
// System.Console.WriteLine($"VssOAuthTokenProvider.OnGetTokenAsync: Calling VssOAuthTokenHttpClient.GetTokenAsync; sign-in url {this.SignInUrl.AbsoluteUri}");
var tokenResponse = await tokenHttpClient.GetTokenAsync(this.Grant, this.ClientCredential, this.TokenParameters, cancellationToken).ConfigureAwait(false);
if (!String.IsNullOrEmpty(tokenResponse.AccessToken))
{
@@ -197,6 +207,7 @@ namespace GitHub.Services.OAuth
protected virtual IssuedToken CreateIssuedToken(VssOAuthTokenResponse tokenResponse)
{
// System.Console.WriteLine($"VssOAuthTokenProvider.CreateIssuedToken");
if (tokenResponse.ExpiresIn > 0)
{
return new VssOAuthAccessToken(tokenResponse.AccessToken, DateTime.UtcNow.AddSeconds(tokenResponse.ExpiresIn));

View File

@@ -100,6 +100,7 @@ namespace GitHub.Services.WebApi
IDictionary<String, String> parameters,
CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("VssConnection.ConnectAsync");
CheckForDisposed();
// Set the connectMode on the credential's FederatedPrompt
if (Credentials.Federated != null && Credentials.Federated.Prompt != null)

View File

@@ -390,6 +390,7 @@ namespace GitHub.Services.WebApi
Object userState = null,
CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("VssHttpClientBase.SendAsync 1");
return SendAsync<T>(method, null, locationId, routeValues, version, content, queryParameters, userState, cancellationToken);
}
@@ -404,6 +405,7 @@ namespace GitHub.Services.WebApi
Object userState = null,
CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("VssHttpClientBase.SendAsync 2");
using (VssTraceActivity.GetOrCreate().EnterCorrelationScope())
using (HttpRequestMessage requestMessage = await CreateRequestMessageAsync(method, additionalHeaders, locationId, routeValues, version, content, queryParameters, userState, cancellationToken).ConfigureAwait(false))
{
@@ -422,6 +424,7 @@ namespace GitHub.Services.WebApi
Object userState = null,
CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("VssHttpClientBase.SendAsync 3");
using (VssTraceActivity.GetOrCreate().EnterCorrelationScope())
using (HttpRequestMessage requestMessage = await CreateRequestMessageAsync(method, additionalHeaders, locationId, routeValues, version, content, queryParameters, userState, cancellationToken).ConfigureAwait(false))
{
@@ -455,6 +458,7 @@ namespace GitHub.Services.WebApi
Object userState = null,
CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("VssHttpClientBase.SendAsync 4");
using (VssTraceActivity.GetOrCreate().EnterCorrelationScope())
using (HttpRequestMessage requestMessage = await CreateRequestMessageAsync(method, locationId, routeValues, version, content, queryParameters, userState, cancellationToken).ConfigureAwait(false))
{
@@ -473,6 +477,7 @@ namespace GitHub.Services.WebApi
Object userState = null,
CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("VssHttpClientBase.SendAsync 5");
using (VssTraceActivity.GetOrCreate().EnterCorrelationScope())
using (HttpRequestMessage requestMessage = await CreateRequestMessageAsync(method, locationId, routeValues, version, content, queryParameters, userState, cancellationToken).ConfigureAwait(false))
{
@@ -501,6 +506,7 @@ namespace GitHub.Services.WebApi
CancellationToken cancellationToken = default(CancellationToken),
String mediaType = c_jsonMediaType)
{
// System.Console.WriteLine("VssHttpClientBase.CreateRequestMessageAsync 1");
return CreateRequestMessageAsync(method, null, locationId, routeValues, version, content, queryParameters, userState, cancellationToken, mediaType);
}
@@ -526,6 +532,7 @@ namespace GitHub.Services.WebApi
CancellationToken cancellationToken = default(CancellationToken),
String mediaType = c_jsonMediaType)
{
// System.Console.WriteLine("VssHttpClientBase.CreateRequestMessageAsync 2");
// Lookup the location
ApiResourceLocation location = await GetResourceLocationAsync(locationId, userState, cancellationToken).ConfigureAwait(false);
if (location == null)
@@ -555,6 +562,7 @@ namespace GitHub.Services.WebApi
IEnumerable<KeyValuePair<String, String>> queryParameters = null,
String mediaType = c_jsonMediaType)
{
// System.Console.WriteLine("VssHttpClientBase.CreateRequestMessageAsync 3");
return CreateRequestMessage(method, null, location, routeValues, version, content, queryParameters, mediaType);
}
@@ -578,6 +586,7 @@ namespace GitHub.Services.WebApi
IEnumerable<KeyValuePair<String, String>> queryParameters = null,
String mediaType = c_jsonMediaType)
{
// System.Console.WriteLine("VssHttpClientBase.CreateRequestMessageAsync 4");
CheckForDisposed();
// Negotiate the request version to send
ApiResourceVersion requestVersion = NegotiateRequestVersion(location, version);
@@ -749,12 +758,14 @@ namespace GitHub.Services.WebApi
//from deadlocking...
using (HttpResponseMessage response = await this.SendAsync(message, userState, cancellationToken).ConfigureAwait(false))
{
// System.Console.WriteLine("VssHttpClientBase.SendAsync 6");
return await ReadContentAsAsync<T>(response, cancellationToken).ConfigureAwait(false);
}
}
protected async Task<T> ReadContentAsAsync<T>(HttpResponseMessage response, CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine($"VssHttpClientBase.ReadContentAsAsync {response.Headers}");
CheckForDisposed();
Boolean isJson = IsJsonResponse(response);
bool mismatchContentType = false;
@@ -766,17 +777,20 @@ namespace GitHub.Services.WebApi
!typeof(Byte[]).GetTypeInfo().IsAssignableFrom(typeof(T).GetTypeInfo()) &&
!typeof(JObject).GetTypeInfo().IsAssignableFrom(typeof(T).GetTypeInfo()))
{
// System.Console.WriteLine("VssHttpClientBase.ReadContentAsAsync: isJson 1");
// expect it to come back wrapped, if it isn't it is a bug!
var wrapper = await ReadJsonContentAsync<VssJsonCollectionWrapper<T>>(response, cancellationToken).ConfigureAwait(false);
return wrapper.Value;
}
else if (isJson)
{
// System.Console.WriteLine("VssHttpClientBase.ReadContentAsAsync: isJson 2");
return await ReadJsonContentAsync<T>(response, cancellationToken).ConfigureAwait(false);
}
}
catch (JsonReaderException)
{
// System.Console.WriteLine("VssHttpClientBase.ReadContentAsAsync: mismatchContentType");
// We thought the content was JSON but failed to parse.
// In this case, do nothing and utilize the HandleUnknownContentType call below
mismatchContentType = true;
@@ -802,6 +816,7 @@ namespace GitHub.Services.WebApi
Object userState = null,
CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("VssHttpClientBase.SendAsync 7");
// the default in httpClient for HttpCompletionOption is ResponseContentRead so that is what we do here
return this.SendAsync(
message,
@@ -816,6 +831,7 @@ namespace GitHub.Services.WebApi
Object userState = null,
CancellationToken cancellationToken = default(CancellationToken))
{
// System.Console.WriteLine("VssHttpClientBase.SendAsync 8");
CheckForDisposed();
if (message.Headers.UserAgent != null)
{
@@ -851,6 +867,7 @@ namespace GitHub.Services.WebApi
//ConfigureAwait(false) enables the continuation to be run outside
//any captured SyncronizationContext (such as ASP.NET's) which keeps things
//from deadlocking...
// System.Console.WriteLine($"VssHttpClientBase.SendAsync 8: Calling Client.SendAsync {message}");
HttpResponseMessage response = await Client.SendAsync(message, completionOption, cancellationToken).ConfigureAwait(false);
// Inject delay or failure for testing
@@ -868,6 +885,7 @@ namespace GitHub.Services.WebApi
[Obsolete("Use VssHttpClientBase.HandleResponseAsync instead")]
protected virtual void HandleResponse(HttpResponseMessage response)
{
// System.Console.WriteLine("VssHttpClientBase.HandleResponse 1");
}
@@ -875,6 +893,7 @@ namespace GitHub.Services.WebApi
HttpResponseMessage response,
CancellationToken cancellationToken)
{
// System.Console.WriteLine($"VssHttpClientBase.HandleResponse 2 status code {response.StatusCode} headers {response.Headers}");
response.Trace();
VssHttpEventSource.Log.HttpRequestStop(VssTraceActivity.Current, response);
@@ -886,6 +905,7 @@ namespace GitHub.Services.WebApi
}
else if (ShouldThrowError(response))
{
// System.Console.WriteLine("VssHttpClientBase.HandleResponse: Should throw error");
Exception exToThrow = null;
if (IsJsonResponse(response))
{
@@ -909,6 +929,7 @@ namespace GitHub.Services.WebApi
{
message = response.ReasonPhrase;
}
// System.Console.WriteLine($"VssHttpClientBase.HandleResponse: Exception message {message}");
exToThrow = new VssServiceResponseException(response.StatusCode, message, exToThrow);
}

View File

@@ -2,7 +2,7 @@
###############################################################################
#
# ./dev.sh build/layout/test/package [Debug/Release]
# ./dev.sh build/layout/test/package [Debug/Release] [linux-x64|linux-x86|linux-arm64|linux-arm|osx-x64|win-x64|win-x86] [use-broker]
#
###############################################################################
@@ -11,6 +11,7 @@ set -e
DEV_CMD=$1
DEV_CONFIG=$2
DEV_TARGET_RUNTIME=$3
DEV_USE_BROKER=$4
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
LAYOUT_DIR="$SCRIPT_DIR/../_layout"
@@ -81,6 +82,13 @@ elif [[ "$CURRENT_PLATFORM" == 'darwin' ]]; then
fi
fi
if [ -n "$DEV_USE_BROKER" ]; then
USE_BROKER='-p:USE_BROKER="true"'
else
USE_BROKER=''
fi
function failed()
{
local error=${1:-Undefined error}
@@ -114,13 +122,13 @@ function heading()
function build ()
{
heading "Building ..."
dotnet msbuild -t:Build -p:PackageRuntime="${RUNTIME_ID}" -p:BUILDCONFIG="${BUILD_CONFIG}" -p:RunnerVersion="${RUNNER_VERSION}" ./dir.proj || failed build
dotnet msbuild -t:Build -p:PackageRuntime="${RUNTIME_ID}" -p:BUILDCONFIG="${BUILD_CONFIG}" -p:RunnerVersion="${RUNNER_VERSION}" $USE_BROKER ./dir.proj || failed build
}
function layout ()
{
heading "Create layout ..."
dotnet msbuild -t:layout -p:PackageRuntime="${RUNTIME_ID}" -p:BUILDCONFIG="${BUILD_CONFIG}" -p:RunnerVersion="${RUNNER_VERSION}" ./dir.proj || failed build
dotnet msbuild -t:layout -p:PackageRuntime="${RUNTIME_ID}" -p:BUILDCONFIG="${BUILD_CONFIG}" -p:RunnerVersion="${RUNNER_VERSION}" $USE_BROKER ./dir.proj || failed build
#change execution flag to allow running with sudo
if [[ ("$CURRENT_PLATFORM" == "linux") || ("$CURRENT_PLATFORM" == "darwin") ]]; then