mirror of
https://github.com/nsnail/Ocelot.git
synced 2025-06-19 06:48:16 +08:00
Raft round 2 (#182)
* brought in rafty * moved raft classes into Ocelot and deleted from int project * started to set up rafty in Ocelot * RAFTY INSIDE OCELOT...WOOT * more work adding rafty...just need to get auth working now * rudimentary authenticated raft requests working * asyn await stuff * hacked rafty into the fileconfigurationcontroller...everything seems to be working roughly but I have a lot of refactoring to do * updated to latest rafty that doesnt need an id * hacky but all tests passing * changed admin area set up to use builder not configuration.json, changed admin area auth to use client credentials * missing code coverage * ignore raft sectionf for code coverage * ignore raft sectionf for code coverage * back to normal filters * try exclude attr * missed these * moved client secret to builder for authentication and updated docs * lock to try and fix error accessing identity server created temprsa file on build server * updated postman scripts and changed Ocelot to not always use type handling as this looked crap when manually accessing the configuration endpoint * added rafty docs * changes I missed * added serialisation code we need for rafty to process commands when they proxy to leader * moved controllers into their feature slices
This commit is contained in:
Tom Pallister
431
test/Ocelot.IntegrationTests/RaftTests.cs
Normal file
431
test/Ocelot.IntegrationTests/RaftTests.cs
Normal file
@ -0,0 +1,431 @@
|
||||
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.Text;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Newtonsoft.Json;
|
||||
using Ocelot.Configuration.File;
|
||||
using Ocelot.Raft;
|
||||
using Rafty.Concensus;
|
||||
using Rafty.FiniteStateMachine;
|
||||
using Rafty.Infrastructure;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using static Rafty.Infrastructure.Wait;
|
||||
using Microsoft.Data.Sqlite;
|
||||
|
||||
namespace Ocelot.IntegrationTests
|
||||
{
|
||||
public class RaftTests : IDisposable
|
||||
{
|
||||
private List<IWebHost> _builders;
|
||||
private List<IWebHostBuilder> _webHostBuilders;
|
||||
private List<Thread> _threads;
|
||||
private FilePeers _peers;
|
||||
private HttpClient _httpClient;
|
||||
private HttpClient _httpClientForAssertions;
|
||||
private string _ocelotBaseUrl;
|
||||
private BearerToken _token;
|
||||
private HttpResponseMessage _response;
|
||||
private static object _lock = new object();
|
||||
|
||||
public RaftTests()
|
||||
{
|
||||
_httpClientForAssertions = new HttpClient();
|
||||
_httpClient = new HttpClient();
|
||||
_ocelotBaseUrl = "http://localhost:5000";
|
||||
_httpClient.BaseAddress = new Uri(_ocelotBaseUrl);
|
||||
_webHostBuilders = new List<IWebHostBuilder>();
|
||||
_builders = new List<IWebHost>();
|
||||
_threads = new List<Thread>();
|
||||
}
|
||||
public void Dispose()
|
||||
{
|
||||
foreach (var builder in _builders)
|
||||
{
|
||||
builder?.Dispose();
|
||||
}
|
||||
|
||||
foreach (var peer in _peers.Peers)
|
||||
{
|
||||
File.Delete(peer.HostAndPort.Replace("/","").Replace(":",""));
|
||||
File.Delete($"{peer.HostAndPort.Replace("/","").Replace(":","")}.db");
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void should_persist_command_to_five_servers()
|
||||
{
|
||||
var configuration = new FileConfiguration
|
||||
{
|
||||
GlobalConfiguration = new FileGlobalConfiguration
|
||||
{
|
||||
}
|
||||
};
|
||||
|
||||
var updatedConfiguration = new FileConfiguration
|
||||
{
|
||||
GlobalConfiguration = new FileGlobalConfiguration
|
||||
{
|
||||
},
|
||||
ReRoutes = new List<FileReRoute>()
|
||||
{
|
||||
new FileReRoute()
|
||||
{
|
||||
DownstreamHost = "127.0.0.1",
|
||||
DownstreamPort = 80,
|
||||
DownstreamScheme = "http",
|
||||
DownstreamPathTemplate = "/geoffrey",
|
||||
UpstreamHttpMethod = new List<string> { "get" },
|
||||
UpstreamPathTemplate = "/"
|
||||
},
|
||||
new FileReRoute()
|
||||
{
|
||||
DownstreamHost = "123.123.123",
|
||||
DownstreamPort = 443,
|
||||
DownstreamScheme = "https",
|
||||
DownstreamPathTemplate = "/blooper/{productId}",
|
||||
UpstreamHttpMethod = new List<string> { "post" },
|
||||
UpstreamPathTemplate = "/test"
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
var command = new UpdateFileConfiguration(updatedConfiguration);
|
||||
GivenThereIsAConfiguration(configuration);
|
||||
GivenFiveServersAreRunning();
|
||||
GivenALeaderIsElected();
|
||||
GivenIHaveAnOcelotToken("/administration");
|
||||
WhenISendACommandIntoTheCluster(command);
|
||||
ThenTheCommandIsReplicatedToAllStateMachines(command);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void should_persist_command_to_five_servers_when_using_administration_api()
|
||||
{
|
||||
var configuration = new FileConfiguration
|
||||
{
|
||||
};
|
||||
|
||||
var updatedConfiguration = new FileConfiguration
|
||||
{
|
||||
ReRoutes = new List<FileReRoute>()
|
||||
{
|
||||
new FileReRoute()
|
||||
{
|
||||
DownstreamHost = "127.0.0.1",
|
||||
DownstreamPort = 80,
|
||||
DownstreamScheme = "http",
|
||||
DownstreamPathTemplate = "/geoffrey",
|
||||
UpstreamHttpMethod = new List<string> { "get" },
|
||||
UpstreamPathTemplate = "/"
|
||||
},
|
||||
new FileReRoute()
|
||||
{
|
||||
DownstreamHost = "123.123.123",
|
||||
DownstreamPort = 443,
|
||||
DownstreamScheme = "https",
|
||||
DownstreamPathTemplate = "/blooper/{productId}",
|
||||
UpstreamHttpMethod = new List<string> { "post" },
|
||||
UpstreamPathTemplate = "/test"
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
var command = new UpdateFileConfiguration(updatedConfiguration);
|
||||
GivenThereIsAConfiguration(configuration);
|
||||
GivenFiveServersAreRunning();
|
||||
GivenALeaderIsElected();
|
||||
GivenIHaveAnOcelotToken("/administration");
|
||||
GivenIHaveAddedATokenToMyRequest();
|
||||
WhenIPostOnTheApiGateway("/administration/configuration", updatedConfiguration);
|
||||
ThenTheCommandIsReplicatedToAllStateMachines(command);
|
||||
}
|
||||
|
||||
private void WhenISendACommandIntoTheCluster(UpdateFileConfiguration command)
|
||||
{
|
||||
var p = _peers.Peers.First();
|
||||
var json = JsonConvert.SerializeObject(command,new JsonSerializerSettings() {
|
||||
TypeNameHandling = TypeNameHandling.All
|
||||
});
|
||||
var httpContent = new StringContent(json);
|
||||
httpContent.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/json");
|
||||
using(var httpClient = new HttpClient())
|
||||
{
|
||||
httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
|
||||
var response = httpClient.PostAsync($"{p.HostAndPort}/administration/raft/command", httpContent).GetAwaiter().GetResult();
|
||||
response.EnsureSuccessStatusCode();
|
||||
var content = response.Content.ReadAsStringAsync().GetAwaiter().GetResult();
|
||||
var result = JsonConvert.DeserializeObject<OkResponse<UpdateFileConfiguration>>(content);
|
||||
result.Command.Configuration.ReRoutes.Count.ShouldBe(2);
|
||||
}
|
||||
|
||||
//dirty sleep to make sure command replicated...
|
||||
var stopwatch = Stopwatch.StartNew();
|
||||
while(stopwatch.ElapsedMilliseconds < 10000)
|
||||
{
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private void ThenTheCommandIsReplicatedToAllStateMachines(UpdateFileConfiguration expected)
|
||||
{
|
||||
//dirty sleep to give a chance to replicate...
|
||||
var stopwatch = Stopwatch.StartNew();
|
||||
while(stopwatch.ElapsedMilliseconds < 2000)
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
bool CommandCalledOnAllStateMachines()
|
||||
{
|
||||
try
|
||||
{
|
||||
var passed = 0;
|
||||
foreach (var peer in _peers.Peers)
|
||||
{
|
||||
var path = $"{peer.HostAndPort.Replace("/","").Replace(":","")}.db";
|
||||
using(var connection = new SqliteConnection($"Data Source={path};"))
|
||||
{
|
||||
connection.Open();
|
||||
var sql = @"select count(id) from logs";
|
||||
using(var command = new SqliteCommand(sql, connection))
|
||||
{
|
||||
var index = Convert.ToInt32(command.ExecuteScalar());
|
||||
index.ShouldBe(1);
|
||||
}
|
||||
}
|
||||
_httpClientForAssertions.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
|
||||
var result = _httpClientForAssertions.GetAsync($"{peer.HostAndPort}/administration/configuration").Result;
|
||||
var json = result.Content.ReadAsStringAsync().Result;
|
||||
var response = JsonConvert.DeserializeObject<FileConfiguration>(json, new JsonSerializerSettings{TypeNameHandling = TypeNameHandling.All});
|
||||
response.GlobalConfiguration.RequestIdKey.ShouldBe(expected.Configuration.GlobalConfiguration.RequestIdKey);
|
||||
response.GlobalConfiguration.ServiceDiscoveryProvider.Host.ShouldBe(expected.Configuration.GlobalConfiguration.ServiceDiscoveryProvider.Host);
|
||||
response.GlobalConfiguration.ServiceDiscoveryProvider.Port.ShouldBe(expected.Configuration.GlobalConfiguration.ServiceDiscoveryProvider.Port);
|
||||
|
||||
for (var i = 0; i < response.ReRoutes.Count; i++)
|
||||
{
|
||||
response.ReRoutes[i].DownstreamHost.ShouldBe(expected.Configuration.ReRoutes[i].DownstreamHost);
|
||||
response.ReRoutes[i].DownstreamPathTemplate.ShouldBe(expected.Configuration.ReRoutes[i].DownstreamPathTemplate);
|
||||
response.ReRoutes[i].DownstreamPort.ShouldBe(expected.Configuration.ReRoutes[i].DownstreamPort);
|
||||
response.ReRoutes[i].DownstreamScheme.ShouldBe(expected.Configuration.ReRoutes[i].DownstreamScheme);
|
||||
response.ReRoutes[i].UpstreamPathTemplate.ShouldBe(expected.Configuration.ReRoutes[i].UpstreamPathTemplate);
|
||||
response.ReRoutes[i].UpstreamHttpMethod.ShouldBe(expected.Configuration.ReRoutes[i].UpstreamHttpMethod);
|
||||
}
|
||||
passed++;
|
||||
}
|
||||
|
||||
return passed == 5;
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
Console.WriteLine(e);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
var commandOnAllStateMachines = WaitFor(20000).Until(() => CommandCalledOnAllStateMachines());
|
||||
commandOnAllStateMachines.ShouldBeTrue();
|
||||
}
|
||||
|
||||
private void ThenTheResponseShouldBe(FileConfiguration expected)
|
||||
{
|
||||
var response = JsonConvert.DeserializeObject<FileConfiguration>(_response.Content.ReadAsStringAsync().Result);
|
||||
|
||||
response.GlobalConfiguration.RequestIdKey.ShouldBe(expected.GlobalConfiguration.RequestIdKey);
|
||||
response.GlobalConfiguration.ServiceDiscoveryProvider.Host.ShouldBe(expected.GlobalConfiguration.ServiceDiscoveryProvider.Host);
|
||||
response.GlobalConfiguration.ServiceDiscoveryProvider.Port.ShouldBe(expected.GlobalConfiguration.ServiceDiscoveryProvider.Port);
|
||||
|
||||
for (var i = 0; i < response.ReRoutes.Count; i++)
|
||||
{
|
||||
response.ReRoutes[i].DownstreamHost.ShouldBe(expected.ReRoutes[i].DownstreamHost);
|
||||
response.ReRoutes[i].DownstreamPathTemplate.ShouldBe(expected.ReRoutes[i].DownstreamPathTemplate);
|
||||
response.ReRoutes[i].DownstreamPort.ShouldBe(expected.ReRoutes[i].DownstreamPort);
|
||||
response.ReRoutes[i].DownstreamScheme.ShouldBe(expected.ReRoutes[i].DownstreamScheme);
|
||||
response.ReRoutes[i].UpstreamPathTemplate.ShouldBe(expected.ReRoutes[i].UpstreamPathTemplate);
|
||||
response.ReRoutes[i].UpstreamHttpMethod.ShouldBe(expected.ReRoutes[i].UpstreamHttpMethod);
|
||||
}
|
||||
}
|
||||
|
||||
private void WhenIGetUrlOnTheApiGateway(string url)
|
||||
{
|
||||
_response = _httpClient.GetAsync(url).Result;
|
||||
}
|
||||
|
||||
private void WhenIPostOnTheApiGateway(string url, FileConfiguration updatedConfiguration)
|
||||
{
|
||||
var json = JsonConvert.SerializeObject(updatedConfiguration);
|
||||
var content = new StringContent(json);
|
||||
content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
|
||||
_response = _httpClient.PostAsync(url, content).Result;
|
||||
}
|
||||
|
||||
private void GivenIHaveAddedATokenToMyRequest()
|
||||
{
|
||||
_httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
|
||||
}
|
||||
|
||||
private void GivenIHaveAnOcelotToken(string adminPath)
|
||||
{
|
||||
var tokenUrl = $"{adminPath}/connect/token";
|
||||
var formData = new List<KeyValuePair<string, string>>
|
||||
{
|
||||
new KeyValuePair<string, string>("client_id", "admin"),
|
||||
new KeyValuePair<string, string>("client_secret", "secret"),
|
||||
new KeyValuePair<string, string>("scope", "admin"),
|
||||
new KeyValuePair<string, string>("grant_type", "client_credentials")
|
||||
};
|
||||
var content = new FormUrlEncodedContent(formData);
|
||||
|
||||
var response = _httpClient.PostAsync(tokenUrl, content).Result;
|
||||
var responseContent = response.Content.ReadAsStringAsync().Result;
|
||||
response.EnsureSuccessStatusCode();
|
||||
_token = JsonConvert.DeserializeObject<BearerToken>(responseContent);
|
||||
var configPath = $"{adminPath}/.well-known/openid-configuration";
|
||||
response = _httpClient.GetAsync(configPath).Result;
|
||||
response.EnsureSuccessStatusCode();
|
||||
}
|
||||
|
||||
private void GivenThereIsAConfiguration(FileConfiguration fileConfiguration)
|
||||
{
|
||||
var configurationPath = $"{Directory.GetCurrentDirectory()}/configuration.json";
|
||||
|
||||
var jsonConfiguration = JsonConvert.SerializeObject(fileConfiguration);
|
||||
|
||||
if (File.Exists(configurationPath))
|
||||
{
|
||||
File.Delete(configurationPath);
|
||||
}
|
||||
|
||||
File.WriteAllText(configurationPath, jsonConfiguration);
|
||||
|
||||
var text = File.ReadAllText(configurationPath);
|
||||
|
||||
configurationPath = $"{AppContext.BaseDirectory}/configuration.json";
|
||||
|
||||
if (File.Exists(configurationPath))
|
||||
{
|
||||
File.Delete(configurationPath);
|
||||
}
|
||||
|
||||
File.WriteAllText(configurationPath, jsonConfiguration);
|
||||
|
||||
text = File.ReadAllText(configurationPath);
|
||||
}
|
||||
|
||||
private void GivenAServerIsRunning(string url)
|
||||
{
|
||||
lock(_lock)
|
||||
{
|
||||
IWebHostBuilder webHostBuilder = new WebHostBuilder();
|
||||
webHostBuilder.UseUrls(url)
|
||||
.UseKestrel()
|
||||
.UseContentRoot(Directory.GetCurrentDirectory())
|
||||
.ConfigureServices(x =>
|
||||
{
|
||||
x.AddSingleton(webHostBuilder);
|
||||
x.AddSingleton(new NodeId(url));
|
||||
})
|
||||
.UseStartup<RaftStartup>();
|
||||
|
||||
var builder = webHostBuilder.Build();
|
||||
builder.Start();
|
||||
|
||||
_webHostBuilders.Add(webHostBuilder);
|
||||
_builders.Add(builder);
|
||||
}
|
||||
}
|
||||
|
||||
private void GivenFiveServersAreRunning()
|
||||
{
|
||||
var bytes = File.ReadAllText("peers.json");
|
||||
_peers = JsonConvert.DeserializeObject<FilePeers>(bytes);
|
||||
|
||||
foreach (var peer in _peers.Peers)
|
||||
{
|
||||
var thread = new Thread(() => GivenAServerIsRunning(peer.HostAndPort));
|
||||
thread.Start();
|
||||
_threads.Add(thread);
|
||||
}
|
||||
}
|
||||
|
||||
private void GivenALeaderIsElected()
|
||||
{
|
||||
//dirty sleep to make sure we have a leader
|
||||
var stopwatch = Stopwatch.StartNew();
|
||||
while(stopwatch.ElapsedMilliseconds < 20000)
|
||||
{
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private void WhenISendACommandIntoTheCluster(FakeCommand command)
|
||||
{
|
||||
var p = _peers.Peers.First();
|
||||
var json = JsonConvert.SerializeObject(command,new JsonSerializerSettings() {
|
||||
TypeNameHandling = TypeNameHandling.All
|
||||
});
|
||||
var httpContent = new StringContent(json);
|
||||
httpContent.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/json");
|
||||
using(var httpClient = new HttpClient())
|
||||
{
|
||||
httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
|
||||
var response = httpClient.PostAsync($"{p.HostAndPort}/administration/raft/command", httpContent).GetAwaiter().GetResult();
|
||||
response.EnsureSuccessStatusCode();
|
||||
var content = response.Content.ReadAsStringAsync().GetAwaiter().GetResult();
|
||||
var result = JsonConvert.DeserializeObject<OkResponse<FakeCommand>>(content);
|
||||
result.Command.Value.ShouldBe(command.Value);
|
||||
}
|
||||
|
||||
//dirty sleep to make sure command replicated...
|
||||
var stopwatch = Stopwatch.StartNew();
|
||||
while(stopwatch.ElapsedMilliseconds < 10000)
|
||||
{
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private void ThenTheCommandIsReplicatedToAllStateMachines(FakeCommand command)
|
||||
{
|
||||
//dirty sleep to give a chance to replicate...
|
||||
var stopwatch = Stopwatch.StartNew();
|
||||
while(stopwatch.ElapsedMilliseconds < 2000)
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
bool CommandCalledOnAllStateMachines()
|
||||
{
|
||||
try
|
||||
{
|
||||
var passed = 0;
|
||||
foreach (var peer in _peers.Peers)
|
||||
{
|
||||
string fsmData;
|
||||
fsmData = File.ReadAllText(peer.HostAndPort.Replace("/","").Replace(":",""));
|
||||
fsmData.ShouldNotBeNullOrEmpty();
|
||||
var fakeCommand = JsonConvert.DeserializeObject<FakeCommand>(fsmData);
|
||||
fakeCommand.Value.ShouldBe(command.Value);
|
||||
passed++;
|
||||
}
|
||||
|
||||
return passed == 5;
|
||||
}
|
||||
catch(Exception e)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
var commandOnAllStateMachines = WaitFor(20000).Until(() => CommandCalledOnAllStateMachines());
|
||||
commandOnAllStateMachines.ShouldBeTrue();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user