Cleanup

src/Runner.Common/BrokerServer.cs

@@ -0,0 +1,56 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using GitHub.Actions.RunService.WebApi;
+using GitHub.DistributedTask.Pipelines;
+using GitHub.DistributedTask.WebApi;
+using GitHub.Runner.Sdk;
+using GitHub.Services.Common;
+using Sdk.RSWebApi.Contracts;
+using Sdk.WebApi.WebApi.RawClient;
+
+namespace GitHub.Runner.Common
+{
+    [ServiceLocator(Default = typeof(BrokerServer))]
+    public interface IBrokerServer : IRunnerService
+    {
+        Task ConnectAsync(Uri serverUrl, VssCredentials credentials);
+
+        Task<TaskAgentMessage> GetRunnerMessageAsync(CancellationToken token, TaskAgentStatus status, string version);
+    }
+
+    public sealed class BrokerServer : RunnerService, IBrokerServer
+    {
+        private bool _hasConnection;
+        private Uri _brokerUri;
+        private RawConnection _connection;
+        private BrokerHttpClient _brokerHttpClient;
+
+        public async Task ConnectAsync(Uri serverUri, VssCredentials credentials)
+        {
+            _brokerUri = serverUri;
+
+            _connection = VssUtil.CreateRawConnection(serverUri, credentials);
+            _brokerHttpClient = await _connection.GetClientAsync<BrokerHttpClient>();
+            _hasConnection = true;
+        }
+
+        private void CheckConnection()
+        {
+            if (!_hasConnection)
+            {
+                throw new InvalidOperationException($"SetConnection");
+            }
+        }
+
+        public Task<TaskAgentMessage> GetRunnerMessageAsync(CancellationToken cancellationToken, TaskAgentStatus status, string version)
+        {
+            CheckConnection();
+            var jobMessage = RetryRequest<TaskAgentMessage>(
+                async () => await _brokerHttpClient.GetRunnerMessageAsync(version, status, cancellationToken), cancellationToken);
+
+            return jobMessage;
+        }
+    }
+}

src/Runner.Common/ConfigurationStore.cs

@@ -53,6 +53,9 @@ namespace GitHub.Runner.Common
         [DataMember(EmitDefaultValue = false)]
         public bool UseV2Flow { get; set; }
 
+        [DataMember(EmitDefaultValue = false)]
+        public string ServerUrlV2 { get; set; }
+
         [IgnoreDataMember]
         public bool IsHostedServer
         {

src/Runner.Common/RunnerDotcomServer.cs

@@ -17,7 +17,7 @@ namespace GitHub.Runner.Common
     {
         Task<List<TaskAgent>> GetRunnersAsync(int runnerGroupId, string githubUrl, string githubToken, string agentName);
 
-        Task<TaskAgent> AddRunnerAsync(int runnerGroupId, TaskAgent agent, string githubUrl, string githubToken);
+        Task<DistributedTask.WebApi.Runner> AddRunnerAsync(int runnerGroupId, TaskAgent agent, string githubUrl, string githubToken, string publicKey);
         Task<List<TaskAgentPool>> GetRunnerGroupsAsync(string githubUrl, string githubToken);
 
         string GetGitHubRequestId(HttpResponseHeaders headers);
@@ -136,7 +136,7 @@ namespace GitHub.Runner.Common
             return agentPools?.ToAgentPoolList();
         }
 
-        public async Task<TaskAgent> AddRunnerAsync(int runnerGroupId, TaskAgent agent, string githubUrl, string githubToken)
+        public async Task<DistributedTask.WebApi.Runner> AddRunnerAsync(int runnerGroupId, TaskAgent agent, string githubUrl, string githubToken, string publicKey)
         {
             var gitHubUrlBuilder = new UriBuilder(githubUrl);
             var path = gitHubUrlBuilder.Path.Split('/', '\\', StringSplitOptions.RemoveEmptyEntries);
@@ -159,12 +159,12 @@ namespace GitHub.Runner.Common
                         {"updates_disabled", agent.DisableUpdate},
                         {"ephemeral", agent.Ephemeral},
                         {"labels", agent.Labels},
+                        {"public_key", publicKey}
                     };
 
             var body = new StringContent(StringUtil.ConvertToJson(bodyObject), null, "application/json");
-            var responseAgent = await RetryRequest<TaskAgent>(githubApiUrl, githubToken, RequestType.Post, 3, "Failed to add agent", body);
-            agent.Id = responseAgent.Id;
-            return agent;
+
+            return await RetryRequest<DistributedTask.WebApi.Runner>(githubApiUrl, githubToken, RequestType.Post, 3, "Failed to add agent", body);
         }
 
         private async Task<T> RetryRequest<T>(string githubApiUrl, string githubToken, RequestType requestType, int maxRetryAttemptsCount = 5, string errorMessage = null, StringContent body = null)

src/Runner.Listener/BrokerMessageListener.cs

@@ -0,0 +1,209 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Runtime.InteropServices;
+using System.Security.Cryptography;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using GitHub.DistributedTask.WebApi;
+using GitHub.Runner.Common;
+using GitHub.Runner.Listener.Configuration;
+using GitHub.Runner.Sdk;
+using GitHub.Services.Common;
+using GitHub.Runner.Common.Util;
+using GitHub.Services.OAuth;
+
+namespace GitHub.Runner.Listener
+{
+    public sealed class BrokerMessageListener : RunnerService, IMessageListener
+    {
+        private RunnerSettings _settings;
+        private ITerminal _term;
+        private TimeSpan _getNextMessageRetryInterval;
+        private TaskAgentStatus runnerStatus = TaskAgentStatus.Online;
+        private CancellationTokenSource _getMessagesTokenSource;
+        private IBrokerServer _brokerServer;
+
+        public override void Initialize(IHostContext hostContext)
+        {
+            base.Initialize(hostContext);
+
+            _term = HostContext.GetService<ITerminal>();
+            _brokerServer = HostContext.GetService<IBrokerServer>();
+        }
+
+        public async Task<Boolean> CreateSessionAsync(CancellationToken token)
+        {
+            await RefreshBrokerConnection();
+            return await Task.FromResult(true);
+        }
+
+        public async Task DeleteSessionAsync()
+        {
+            await Task.CompletedTask;
+        }
+
+        public void OnJobStatus(object sender, JobStatusEventArgs e)
+        {
+            Trace.Info("Received job status event. JobState: {0}", e.Status);
+            runnerStatus = e.Status;
+            try
+            {
+                _getMessagesTokenSource?.Cancel();
+            }
+            catch (ObjectDisposedException)
+            {
+                Trace.Info("_getMessagesTokenSource is already disposed.");
+            }
+        }
+
+        public async Task<TaskAgentMessage> GetNextMessageAsync(CancellationToken token)
+        {
+            bool encounteringError = false;
+            int continuousError = 0;
+            Stopwatch heartbeat = new();
+            heartbeat.Restart();
+            var maxRetryCount = 10;
+
+            while (true)
+            {
+                TaskAgentMessage message = null;
+                _getMessagesTokenSource = CancellationTokenSource.CreateLinkedTokenSource(token);
+                try
+                {
+                    message = await _brokerServer.GetRunnerMessageAsync(_getMessagesTokenSource.Token, runnerStatus, BuildConstants.RunnerPackage.Version);
+
+                    if (message == null)
+                    {
+                        continue;
+                    }
+
+                    return message;
+                }
+                catch (OperationCanceledException) when (_getMessagesTokenSource.Token.IsCancellationRequested && !token.IsCancellationRequested)
+                {
+                    Trace.Info("Get messages has been cancelled using local token source. Continue to get messages with new status.");
+                    continue;
+                }
+                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.");
+                    throw;
+                }
+                catch (AccessDeniedException e) when (e.InnerException is InvalidTaskAgentVersionException)
+                {
+                    throw;
+                }
+                catch (Exception ex)
+                {
+                    Trace.Error("Catch exception during get next message.");
+                    Trace.Error(ex);
+
+                    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 if (continuousError >= maxRetryCount)
+                        {
+                            throw;
+                        }
+                        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 RefreshBrokerConnection();
+
+                        Trace.Info("Sleeping for {0} seconds before retrying.", _getNextMessageRetryInterval.TotalSeconds);
+                        await HostContext.Delay(_getNextMessageRetryInterval, token);
+                    }
+                }
+                finally
+                {
+                    _getMessagesTokenSource.Dispose();
+                }
+
+                if (message == null)
+                {
+                    if (heartbeat.Elapsed > TimeSpan.FromMinutes(30))
+                    {
+                        Trace.Info($"No message retrieved within last 30 minutes.");
+                        heartbeat.Restart();
+                    }
+                    else
+                    {
+                        Trace.Verbose($"No message retrieved.");
+                    }
+
+                    continue;
+                }
+
+                Trace.Info($"Message '{message.MessageId}' received.");
+            }
+        }
+
+        public async Task DeleteMessageAsync(TaskAgentMessage message)
+        {
+            await Task.CompletedTask;
+        }
+
+        private bool IsGetNextMessageExceptionRetriable(Exception ex)
+        {
+            if (ex is TaskAgentNotFoundException ||
+                ex is TaskAgentPoolNotFoundException ||
+                ex is TaskAgentSessionExpiredException ||
+                ex is AccessDeniedException ||
+                ex is VssUnauthorizedException)
+            {
+                Trace.Info($"Non-retriable exception: {ex.Message}");
+                return false;
+            }
+            else
+            {
+                Trace.Info($"Retriable exception: {ex.Message}");
+                return true;
+            }
+        }
+
+        private async Task RefreshBrokerConnection()
+        {
+            var configManager = HostContext.GetService<IConfigurationManager>();
+            _settings = configManager.LoadSettings();
+
+            if (_settings.ServerUrlV2 == null)
+            {
+                throw new InvalidOperationException("ServerUrlV2 is not set");
+            }
+
+            var credMgr = HostContext.GetService<ICredentialManager>();
+            VssCredentials creds = credMgr.LoadCredentials();
+            await _brokerServer.ConnectAsync(new Uri(_settings.ServerUrlV2), creds);
+        }
+    }
+}

src/Runner.Listener/Configuration/ConfigurationManager.cs

@@ -14,6 +14,7 @@ using GitHub.Runner.Sdk;
 using GitHub.Services.Common;
 using GitHub.Services.Common.Internal;
 using GitHub.Services.OAuth;
+using GitHub.Services.WebApi.Jwt;
 
 namespace GitHub.Runner.Listener.Configuration
 {
@@ -181,9 +182,11 @@ namespace GitHub.Runner.Listener.Configuration
             // We want to use the native CSP of the platform for storage, so we use the RSACSP directly
             RSAParameters publicKey;
             var keyManager = HostContext.GetService<IRSAKeyManager>();
+            string publicKeyXML;
             using (var rsa = keyManager.CreateKey())
             {
                 publicKey = rsa.ExportParameters(false);
+                publicKeyXML = rsa.ToXmlString(includePrivateParameters: false);
             }
 
             _term.WriteSection("Runner Registration");
@@ -297,7 +300,9 @@ namespace GitHub.Runner.Listener.Configuration
                     {
                         if (runnerSettings.UseV2Flow)
                         {
-                            agent = await _dotcomServer.AddRunnerAsync(runnerSettings.PoolId, agent, runnerSettings.GitHubUrl, registerToken);
+                            var runner = await _dotcomServer.AddRunnerAsync(runnerSettings.PoolId, agent, runnerSettings.GitHubUrl, registerToken, publicKeyXML);
+                            runner.ApplyToTaskAgent(agent);
+                            runnerSettings.ServerUrlV2 = runner.RunnerAuthorization.ServerUrl;
                         }
                         else
                         {
@@ -354,6 +359,9 @@ namespace GitHub.Runner.Listener.Configuration
             }
 
             // Testing agent connection, detect any potential connection issue, like local clock skew that cause OAuth token expired.
+
+            if (!runnerSettings.UseV2Flow)
+            {
                 var credMgr = HostContext.GetService<ICredentialManager>();
                 VssCredentials credential = credMgr.LoadCredentials();
                 try
@@ -373,6 +381,7 @@ namespace GitHub.Runner.Listener.Configuration
                     Trace.Error(ex);
                     throw new Exception("The local machine's clock may be out of sync with the server time by more than five minutes. Please sync your clock with your domain or internet time and try again.");
                 }
+            }
 
             _term.WriteSection("Runner settings");
 

src/Runner.Listener/Runner.cs

@@ -339,13 +339,25 @@ namespace GitHub.Runner.Listener
             }
         }
 
+        private IMessageListener GetMesageListener(RunnerSettings settings)
+        {
+            if (settings.UseV2Flow)
+            {
+                var brokerListener = new BrokerMessageListener();
+                brokerListener.Initialize(HostContext);
+                return brokerListener;
+            }
+
+            return HostContext.GetService<IMessageListener>();
+        }
+
         //create worker manager, create message listener and start listening to the queue
         private async Task<int> RunAsync(RunnerSettings settings, bool runOnce = false)
         {
             try
             {
                 Trace.Info(nameof(RunAsync));
-                _listener = HostContext.GetService<IMessageListener>();
+                _listener = GetMesageListener(settings);
                 if (!await _listener.CreateSessionAsync(HostContext.RunnerShutdownToken))
                 {
                     return Constants.Runner.ReturnCode.TerminatedError;

src/Sdk/DTWebApi/WebApi/ListRunnersResponse.cs

@@ -41,8 +41,6 @@ namespace GitHub.DistributedTask.WebApi
 
         public List<TaskAgent> ToTaskAgents()
         {
-            List<TaskAgent> taskAgents = new List<TaskAgent>();
-
             return Runners.Select(runner => new TaskAgent() { Name = runner.Name }).ToList();
         }
     }

src/Sdk/DTWebApi/WebApi/Runner.cs

@@ -1,12 +1,44 @@
+using System;
 using Newtonsoft.Json;
 
 namespace GitHub.DistributedTask.WebApi
 {
     public class Runner
+    {
+
+        public class Authorization
         {
             /// <summary>
-        /// Name of the agent
+            /// The url to refresh tokens
             /// </summary> 
+            [JsonProperty("authorization_url")]
+            public Uri AuthorizationUrl
+            {
+                get;
+                internal set;
+            }
+
+            /// <summary>
+            /// The url to connect to to poll for messages
+            /// </summary> 
+            [JsonProperty("server_url")]
+            public string ServerUrl
+            {
+                get;
+                internal set;
+            }
+
+            /// <summary>
+            /// The client id to use when connecting to the authorization_url
+            /// </summary>
+            [JsonProperty("client_id")]
+            public string ClientId
+            {
+                get;
+                internal set;
+            }
+        }
+
         [JsonProperty("name")]
         public string Name
         {
@@ -14,5 +46,29 @@ namespace GitHub.DistributedTask.WebApi
             internal set;
         }
 
+        [JsonProperty("id")]
+        public Int32 Id
+        {
+            get;
+            internal set;
+        }
+
+        [JsonProperty("authorization")]
+        public Authorization RunnerAuthorization
+        {
+            get;
+            internal set;
+        }
+
+        public TaskAgent ApplyToTaskAgent(TaskAgent agent)
+        {
+            agent.Id = this.Id;
+            agent.Authorization = new TaskAgentAuthorization()
+            {
+                AuthorizationUrl = this.RunnerAuthorization.AuthorizationUrl,
+                ClientId = new Guid(this.RunnerAuthorization.ClientId)
+            };
+            return agent;
+        }
     }
 }

src/Sdk/WebApi/WebApi/BrokerHttpClient.cs

@@ -0,0 +1,84 @@
+using System;
+using System.Collections.Generic;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+using GitHub.DistributedTask.Pipelines;
+using GitHub.DistributedTask.WebApi;
+using GitHub.Services.Common;
+using GitHub.Services.OAuth;
+using GitHub.Services.WebApi;
+using Sdk.RSWebApi.Contracts;
+using Sdk.WebApi.WebApi;
+
+namespace GitHub.Actions.RunService.WebApi
+{
+    public class BrokerHttpClient : RawHttpClientBase
+    {
+        public BrokerHttpClient(
+            Uri baseUrl,
+            VssOAuthCredential credentials)
+            : base(baseUrl, credentials)
+        {
+        }
+
+        public BrokerHttpClient(
+            Uri baseUrl,
+            VssOAuthCredential credentials,
+            RawClientHttpRequestSettings settings)
+            : base(baseUrl, credentials, settings)
+        {
+        }
+
+        public BrokerHttpClient(
+            Uri baseUrl,
+            VssOAuthCredential credentials,
+            params DelegatingHandler[] handlers)
+            : base(baseUrl, credentials, handlers)
+        {
+        }
+
+        public BrokerHttpClient(
+            Uri baseUrl,
+            VssOAuthCredential credentials,
+            RawClientHttpRequestSettings settings,
+            params DelegatingHandler[] handlers)
+            : base(baseUrl, credentials, settings, handlers)
+        {
+        }
+
+        public BrokerHttpClient(
+            Uri baseUrl,
+            HttpMessageHandler pipeline,
+            Boolean disposeHandler)
+            : base(baseUrl, pipeline, disposeHandler)
+        {
+        }
+
+        public Task<TaskAgentMessage> GetRunnerMessageAsync(
+            string runnerVersion,
+            TaskAgentStatus? status,
+            CancellationToken cancellationToken = default
+        )
+        {
+            var requestUri = new Uri(Client.BaseAddress, "message");
+
+            List<KeyValuePair<string, string>> queryParams = new List<KeyValuePair<string, string>>();
+
+            if (status != null)
+            {
+                queryParams.Add("status", status.Value.ToString());
+            }
+            if (runnerVersion != null)
+            {
+                queryParams.Add("runnerVersion", runnerVersion);
+            }
+
+            return SendAsync<TaskAgentMessage>(
+                new HttpMethod("GET"),
+                requestUri: requestUri,
+                queryParameters: queryParams,
+                cancellationToken: cancellationToken);
+        }
+    }
+}