tk/Ocelot
1
0
Fork 0
mirror of https://github.com/nsnail/Ocelot.git synced 2025-11-04 13:50:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

#529 removed raft (#543)

Browse Source
This commit is contained in:
Tom Pallister
2018-08-12 22:12:59 +05:30
committed by GitHub
parent 4e22b3cfc4
commit cbd0af6d75
23 changed files with 6 additions and 1437 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/Ocelot/Configuration/FileConfigurationController.cs
View File

@@ -4,11 +4,9 @@ using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using Ocelot.Configuration.File;
using Ocelot.Configuration.Setter;
using Ocelot.Raft;
namespace Ocelot.Configuration
{
using Rafty.Concensus.Node;
using Repository;
[Authorize]
@@ -44,20 +42,6 @@ namespace Ocelot.Configuration
{
try
{
//todo - this code is a bit shit sort it out..
var test = _provider.GetService(typeof(INode));
if (test != null)
{
var node = (INode)test;
var result = await node.Accept(new UpdateFileConfiguration(fileConfiguration));
if (result.GetType() == typeof(Rafty.Infrastructure.ErrorResponse<UpdateFileConfiguration>))
{
return new BadRequestObjectResult("There was a problem. This error message sucks raise an issue in GitHub.");
}
return new OkObjectResult(result.Command.Configuration);
}
var response = await _setter.Set(fileConfiguration);
if (response.IsError)

1
src/Ocelot/DependencyInjection/IOcelotAdministrationBuilder.cs
View File

@@ -2,6 +2,5 @@ namespace Ocelot.DependencyInjection
{
public interface IOcelotAdministrationBuilder
{
IOcelotAdministrationBuilder AddRafty();
}
}

27
src/Ocelot/DependencyInjection/OcelotAdministrationBuilder.cs
View File

@@ -1,36 +1,17 @@
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Ocelot.Raft;
using Rafty.Concensus;
using Rafty.FiniteStateMachine;
using Rafty.Infrastructure;
using Rafty.Log;
namespace Ocelot.DependencyInjection
{
using Rafty.Concensus.Node;
public class OcelotAdministrationBuilder : IOcelotAdministrationBuilder
{
private readonly IServiceCollection _services;
private readonly IConfiguration _configurationRoot;
private IServiceCollection Services { get; }
private IConfiguration ConfigurationRoot { get; }
public OcelotAdministrationBuilder(IServiceCollection services, IConfiguration configurationRoot)
{
_configurationRoot = configurationRoot;
_services = services;
}
public IOcelotAdministrationBuilder AddRafty()
{
var settings = new InMemorySettings(4000, 6000, 100, 10000);
_services.AddSingleton<ILog, SqlLiteLog>();
_services.AddSingleton<IFiniteStateMachine, OcelotFiniteStateMachine>();
_services.AddSingleton<ISettings>(settings);
_services.AddSingleton<IPeersProvider, FilePeersProvider>();
_services.AddSingleton<INode, Node>();
_services.Configure<FilePeers>(_configurationRoot);
return this;
ConfigurationRoot = configurationRoot;
Services = services;
}
}
}

34
src/Ocelot/Middleware/OcelotMiddlewareExtensions.cs
View File

@@ -15,10 +15,7 @@
using Ocelot.Configuration.Setter;
using Ocelot.Responses;
using Ocelot.Logging;
using Rafty.Concensus;
using Rafty.Infrastructure;
using Ocelot.Middleware.Pipeline;
using Rafty.Concensus.Node;
using Microsoft.Extensions.DependencyInjection;
public static class OcelotMiddlewareExtensions
@@ -42,11 +39,6 @@
CreateAdministrationArea(builder, configuration);
if (UsingRafty(builder))
{
SetUpRafty(builder);
}
ConfigureDiagnosticListener(builder);
return CreateOcelotPipeline(builder, pipelineConfiguration);
@@ -77,26 +69,6 @@
return builder;
}
private static bool UsingRafty(IApplicationBuilder builder)
{
var node = builder.ApplicationServices.GetService<INode>();
if (node != null)
{
return true;
}
return false;
}
private static void SetUpRafty(IApplicationBuilder builder)
{
var applicationLifetime = builder.ApplicationServices.GetService<IApplicationLifetime>();
applicationLifetime.ApplicationStopping.Register(() => OnShutdown(builder));
var node = builder.ApplicationServices.GetService<INode>();
var nodeId = builder.ApplicationServices.GetService<NodeId>();
node.Start(nodeId);
}
private static async Task<IInternalConfiguration> CreateConfiguration(IApplicationBuilder builder)
{
// make configuration from file system?
@@ -207,11 +179,5 @@
var diagnosticListener = builder.ApplicationServices.GetService<DiagnosticListener>();
diagnosticListener.SubscribeWithAdapter(listener);
}
private static void OnShutdown(IApplicationBuilder app)
{
var node = app.ApplicationServices.GetService<INode>();
node.Stop();
}
}
}

1
src/Ocelot/Ocelot.csproj
View File

@@ -48,6 +48,5 @@
<PackageReference Include="System.Text.RegularExpressions" Version="4.3.0" />
<PackageReference Include="Polly" Version="6.0.1" />
<PackageReference Include="IdentityServer4" Version="2.2.0" />
<PackageReference Include="Rafty" Version="0.4.4" />
</ItemGroup>
</Project>

17
src/Ocelot/Raft/BearerToken.cs
View File

@@ -1,17 +0,0 @@
using Newtonsoft.Json;
namespace Ocelot.Raft
{
[ExcludeFromCoverage]
internal class BearerToken
{
[JsonProperty("access_token")]
public string AccessToken { get; set; }
[JsonProperty("expires_in")]
public int ExpiresIn { get; set; }
[JsonProperty("token_type")]
public string TokenType { get; set; }
}
}

7
src/Ocelot/Raft/ExcludeFromCoverageAttribute.cs
View File

@@ -1,7 +0,0 @@
using System;
namespace Ocelot.Raft
{
[AttributeUsage(AttributeTargets.Class|AttributeTargets.Method|AttributeTargets.Property)]
public class ExcludeFromCoverageAttribute : Attribute{}
}

15
src/Ocelot/Raft/FakeCommand.cs
View File

@@ -1,15 +0,0 @@
using Rafty.FiniteStateMachine;
namespace Ocelot.Raft
{
[ExcludeFromCoverage]
public class FakeCommand : ICommand
{
public FakeCommand(string value)
{
this.Value = value;
}
public string Value { get; private set; }
}
}

8
src/Ocelot/Raft/FilePeer.cs
View File

@@ -1,8 +0,0 @@
namespace Ocelot.Raft
{
[ExcludeFromCoverage]
public class FilePeer
{
public string HostAndPort { get; set; }
}
}

15
src/Ocelot/Raft/FilePeers.cs
View File

@@ -1,15 +0,0 @@
using System.Collections.Generic;
namespace Ocelot.Raft
{
[ExcludeFromCoverage]
public class FilePeers
{
public FilePeers()
{
Peers = new List<FilePeer>();
}
public List<FilePeer> Peers {get; set;}
}
}

47
src/Ocelot/Raft/FilePeersProvider.cs
View File

@@ -1,47 +0,0 @@
using System.Collections.Generic;
using System.Net.Http;
using Microsoft.Extensions.Options;
using Ocelot.Configuration;
using Ocelot.Configuration.Repository;
using Ocelot.Middleware;
using Rafty.Concensus;
using Rafty.Infrastructure;
namespace Ocelot.Raft
{
using Rafty.Concensus.Peers;
[ExcludeFromCoverage]
public class FilePeersProvider : IPeersProvider
{
private readonly IOptions<FilePeers> _options;
private readonly List<IPeer> _peers;
private IBaseUrlFinder _finder;
private IInternalConfigurationRepository _repo;
private IIdentityServerConfiguration _identityServerConfig;
public FilePeersProvider(IOptions<FilePeers> options, IBaseUrlFinder finder, IInternalConfigurationRepository repo, IIdentityServerConfiguration identityServerConfig)
{
_identityServerConfig = identityServerConfig;
_repo = repo;
_finder = finder;
_options = options;
_peers = new List<IPeer>();
var config = _repo.Get();
foreach (var item in _options.Value.Peers)
{
var httpClient = new HttpClient();
//todo what if this errors?
var httpPeer = new HttpPeer(item.HostAndPort, httpClient, _finder, config.Data, _identityServerConfig);
_peers.Add(httpPeer);
}
}
public List<IPeer> Get()
{
return _peers;
}
}
}

131
src/Ocelot/Raft/HttpPeer.cs
View File

@@ -1,131 +0,0 @@
using System;
using System.Collections.Generic;
using System.Net.Http;
using System.Threading.Tasks;
using Newtonsoft.Json;
using Ocelot.Configuration;
using Ocelot.Middleware;
using Rafty.Concensus;
using Rafty.FiniteStateMachine;
namespace Ocelot.Raft
{
using Rafty.Concensus.Messages;
using Rafty.Concensus.Peers;
using Rafty.Infrastructure;
[ExcludeFromCoverage]
public class HttpPeer : IPeer
{
private readonly string _hostAndPort;
private readonly HttpClient _httpClient;
private readonly JsonSerializerSettings _jsonSerializerSettings;
private readonly string _baseSchemeUrlAndPort;
private BearerToken _token;
private readonly IInternalConfiguration _config;
private readonly IIdentityServerConfiguration _identityServerConfiguration;
public HttpPeer(string hostAndPort, HttpClient httpClient, IBaseUrlFinder finder, IInternalConfiguration config, IIdentityServerConfiguration identityServerConfiguration)
{
_identityServerConfiguration = identityServerConfiguration;
_config = config;
Id = hostAndPort;
_hostAndPort = hostAndPort;
_httpClient = httpClient;
_jsonSerializerSettings = new JsonSerializerSettings() {
TypeNameHandling = TypeNameHandling.All
};
_baseSchemeUrlAndPort = finder.Find();
}
public string Id { get; }
public async Task<RequestVoteResponse> Request(RequestVote requestVote)
{
if(_token == null)
{
await SetToken();
}
var json = JsonConvert.SerializeObject(requestVote, _jsonSerializerSettings);
var content = new StringContent(json);
content.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/json");
var response = await _httpClient.PostAsync($"{_hostAndPort}/administration/raft/requestvote", content);
if(response.IsSuccessStatusCode)
{
return JsonConvert.DeserializeObject<RequestVoteResponse>(await response.Content.ReadAsStringAsync(), _jsonSerializerSettings);
}
return new RequestVoteResponse(false, requestVote.Term);
}
public async Task<AppendEntriesResponse> Request(AppendEntries appendEntries)
{
try
{
if(_token == null)
{
await SetToken();
}
var json = JsonConvert.SerializeObject(appendEntries, _jsonSerializerSettings);
var content = new StringContent(json);
content.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/json");
var response = await _httpClient.PostAsync($"{_hostAndPort}/administration/raft/appendEntries", content);
if(response.IsSuccessStatusCode)
{
return JsonConvert.DeserializeObject<AppendEntriesResponse>(await response.Content.ReadAsStringAsync(), _jsonSerializerSettings);
}
return new AppendEntriesResponse(appendEntries.Term, false);
}
catch(Exception ex)
{
Console.WriteLine(ex);
return new AppendEntriesResponse(appendEntries.Term, false);
}
}
public async Task<Response<T>> Request<T>(T command)
where T : ICommand
{
Console.WriteLine("SENDING REQUEST....");
if(_token == null)
{
await SetToken();
}
var json = JsonConvert.SerializeObject(command, _jsonSerializerSettings);
var content = new StringContent(json);
content.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("application/json");
var response = await _httpClient.PostAsync($"{_hostAndPort}/administration/raft/command", content);
if(response.IsSuccessStatusCode)
{
Console.WriteLine("REQUEST OK....");
var okResponse = JsonConvert.DeserializeObject<OkResponse<ICommand>>(await response.Content.ReadAsStringAsync(), _jsonSerializerSettings);
return new OkResponse<T>((T)okResponse.Command);
}
Console.WriteLine("REQUEST NOT OK....");
return new ErrorResponse<T>(await response.Content.ReadAsStringAsync(), command);
}
private async Task SetToken()
{
var tokenUrl = $"{_baseSchemeUrlAndPort}{_config.AdministrationPath}/connect/token";
var formData = new List<KeyValuePair<string, string>>
{
new KeyValuePair<string, string>("client_id", _identityServerConfiguration.ApiName),
new KeyValuePair<string, string>("client_secret", _identityServerConfiguration.ApiSecret),
new KeyValuePair<string, string>("scope", _identityServerConfiguration.ApiName),
new KeyValuePair<string, string>("grant_type", "client_credentials")
};
var content = new FormUrlEncodedContent(formData);
var response = await _httpClient.PostAsync(tokenUrl, content);
var responseContent = await response.Content.ReadAsStringAsync();
response.EnsureSuccessStatusCode();
_token = JsonConvert.DeserializeObject<BearerToken>(responseContent);
_httpClient.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue(_token.TokenType, _token.AccessToken);
}
}
}

26
src/Ocelot/Raft/OcelotFiniteStateMachine.cs
View File

@@ -1,26 +0,0 @@
using System.Threading.Tasks;
using Ocelot.Configuration.Setter;
using Rafty.FiniteStateMachine;
using Rafty.Log;
namespace Ocelot.Raft
{
[ExcludeFromCoverage]
public class OcelotFiniteStateMachine : IFiniteStateMachine
{
private readonly IFileConfigurationSetter _setter;
public OcelotFiniteStateMachine(IFileConfigurationSetter setter)
{
_setter = setter;
}
public async Task Handle(LogEntry log)
{
//todo - handle an error
//hack it to just cast as at the moment we know this is the only command :P
var hack = (UpdateFileConfiguration)log.CommandData;
await _setter.Set(hack.Configuration);
}
}
}

101
src/Ocelot/Raft/RaftController.cs
View File

@@ -1,101 +0,0 @@
using System;
using System.IO;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Ocelot.Logging;
using Ocelot.Middleware;
using Ocelot.Raft;
using Rafty.Concensus;
using Rafty.FiniteStateMachine;
namespace Ocelot.Raft
{
using Rafty.Concensus.Messages;
using Rafty.Concensus.Node;
[ExcludeFromCoverage]
[Authorize]
[Route("raft")]
public class RaftController : Controller
{
private readonly INode _node;
private readonly IOcelotLogger _logger;
private readonly string _baseSchemeUrlAndPort;
private readonly JsonSerializerSettings _jsonSerialiserSettings;
public RaftController(INode node, IOcelotLoggerFactory loggerFactory, IBaseUrlFinder finder)
{
_jsonSerialiserSettings = new JsonSerializerSettings {
TypeNameHandling = TypeNameHandling.All
};
_baseSchemeUrlAndPort = finder.Find();
_logger = loggerFactory.CreateLogger<RaftController>();
_node = node;
}
[Route("appendentries")]
public async Task<IActionResult> AppendEntries()
{
using(var reader = new StreamReader(HttpContext.Request.Body))
{
var json = await reader.ReadToEndAsync();
var appendEntries = JsonConvert.DeserializeObject<AppendEntries>(json, _jsonSerialiserSettings);
_logger.LogDebug($"{_baseSchemeUrlAndPort}/appendentries called, my state is {_node.State.GetType().FullName}");
var appendEntriesResponse = await _node.Handle(appendEntries);
return new OkObjectResult(appendEntriesResponse);
}
}
[Route("requestvote")]
public async Task<IActionResult> RequestVote()
{
using(var reader = new StreamReader(HttpContext.Request.Body))
{
var json = await reader.ReadToEndAsync();
var requestVote = JsonConvert.DeserializeObject<RequestVote>(json, _jsonSerialiserSettings);
_logger.LogDebug($"{_baseSchemeUrlAndPort}/requestvote called, my state is {_node.State.GetType().FullName}");
var requestVoteResponse = await _node.Handle(requestVote);
return new OkObjectResult(requestVoteResponse);
}
}
[Route("command")]
public async Task<IActionResult> Command()
{
try
{
using(var reader = new StreamReader(HttpContext.Request.Body))
{
var json = await reader.ReadToEndAsync();
var command = JsonConvert.DeserializeObject<ICommand>(json, _jsonSerialiserSettings);
_logger.LogDebug($"{_baseSchemeUrlAndPort}/command called, my state is {_node.State.GetType().FullName}");
var commandResponse = await _node.Accept(command);
json = JsonConvert.SerializeObject(commandResponse, _jsonSerialiserSettings);
return StatusCode(200, json);
}
}
catch(Exception e)
{
_logger.LogError($"THERE WAS A PROBLEM ON NODE {_node.State.CurrentState.Id}", e);
throw;
}
}
}
}

335
src/Ocelot/Raft/SqlLiteLog.cs
View File

@@ -1,335 +0,0 @@
namespace Ocelot.Raft
{
using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Rafty.Infrastructure;
using Rafty.Log;
[ExcludeFromCoverage]
public class SqlLiteLog : ILog
{
private readonly string _path;
private readonly SemaphoreSlim _sempaphore = new SemaphoreSlim(1, 1);
private readonly ILogger _logger;
private readonly NodeId _nodeId;
public SqlLiteLog(NodeId nodeId, ILoggerFactory loggerFactory)
{
_logger = loggerFactory.CreateLogger<SqlLiteLog>();
_nodeId = nodeId;
_path = $"{nodeId.Id.Replace("/", "").Replace(":", "")}.db";
_sempaphore.Wait();
if (!File.Exists(_path))
{
var fs = File.Create(_path);
fs.Dispose();
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
const string sql = @"create table logs (
id integer primary key,
data text not null
)";
using (var command = new SqliteCommand(sql, connection))
{
var result = command.ExecuteNonQuery();
_logger.LogInformation(result == 0
? $"id: {_nodeId.Id} create database, result: {result}"
: $"id: {_nodeId.Id} did not create database., result: {result}");
}
}
}
_sempaphore.Release();
}
public async Task<int> LastLogIndex()
{
_sempaphore.Wait();
var result = 1;
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
var sql = @"select id from logs order by id desc limit 1";
using (var command = new SqliteCommand(sql, connection))
{
var index = Convert.ToInt32(await command.ExecuteScalarAsync());
if (index > result)
{
result = index;
}
}
}
_sempaphore.Release();
return result;
}
public async Task<long> LastLogTerm()
{
_sempaphore.Wait();
long result = 0;
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
var sql = @"select data from logs order by id desc limit 1";
using (var command = new SqliteCommand(sql, connection))
{
var data = Convert.ToString(await command.ExecuteScalarAsync());
var jsonSerializerSettings = new JsonSerializerSettings()
{
TypeNameHandling = TypeNameHandling.All
};
var log = JsonConvert.DeserializeObject<LogEntry>(data, jsonSerializerSettings);
if (log != null && log.Term > result)
{
result = log.Term;
}
}
}
_sempaphore.Release();
return result;
}
public async Task<int> Count()
{
_sempaphore.Wait();
var result = 0;
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(await command.ExecuteScalarAsync());
if (index > result)
{
result = index;
}
}
}
_sempaphore.Release();
return result;
}
public async Task<int> Apply(LogEntry log)
{
_sempaphore.Wait();
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
var jsonSerializerSettings = new JsonSerializerSettings()
{
TypeNameHandling = TypeNameHandling.All
};
var data = JsonConvert.SerializeObject(log, jsonSerializerSettings);
//todo - sql injection dont copy this..
var sql = $"insert into logs (data) values ('{data}')";
_logger.LogInformation($"id: {_nodeId.Id}, sql: {sql}");
using (var command = new SqliteCommand(sql, connection))
{
var result = await command.ExecuteNonQueryAsync();
_logger.LogInformation($"id: {_nodeId.Id}, insert log result: {result}");
}
sql = "select last_insert_rowid()";
using (var command = new SqliteCommand(sql, connection))
{
var result = await command.ExecuteScalarAsync();
_logger.LogInformation($"id: {_nodeId.Id}, about to release semaphore");
_sempaphore.Release();
_logger.LogInformation($"id: {_nodeId.Id}, saved log to sqlite");
return Convert.ToInt32(result);
}
}
}
public async Task DeleteConflictsFromThisLog(int index, LogEntry logEntry)
{
_sempaphore.Wait();
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
//todo - sql injection dont copy this..
var sql = $"select data from logs where id = {index};";
_logger.LogInformation($"id: {_nodeId.Id} sql: {sql}");
using (var command = new SqliteCommand(sql, connection))
{
var data = Convert.ToString(await command.ExecuteScalarAsync());
var jsonSerializerSettings = new JsonSerializerSettings()
{
TypeNameHandling = TypeNameHandling.All
};
_logger.LogInformation($"id {_nodeId.Id} got log for index: {index}, data is {data} and new log term is {logEntry.Term}");
var log = JsonConvert.DeserializeObject<LogEntry>(data, jsonSerializerSettings);
if (logEntry != null && log != null && logEntry.Term != log.Term)
{
//todo - sql injection dont copy this..
var deleteSql = $"delete from logs where id >= {index};";
_logger.LogInformation($"id: {_nodeId.Id} sql: {deleteSql}");
using (var deleteCommand = new SqliteCommand(deleteSql, connection))
{
var result = await deleteCommand.ExecuteNonQueryAsync();
}
}
}
}
_sempaphore.Release();
}
public async Task<bool> IsDuplicate(int index, LogEntry logEntry)
{
_sempaphore.Wait();
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
//todo - sql injection dont copy this..
var sql = $"select data from logs where id = {index};";
using (var command = new SqliteCommand(sql, connection))
{
var data = Convert.ToString(await command.ExecuteScalarAsync());
var jsonSerializerSettings = new JsonSerializerSettings()
{
TypeNameHandling = TypeNameHandling.All
};
var log = JsonConvert.DeserializeObject<LogEntry>(data, jsonSerializerSettings);
if (logEntry != null && log != null && logEntry.Term == log.Term)
{
_sempaphore.Release();
return true;
}
}
}
_sempaphore.Release();
return false;
}
public async Task<LogEntry> Get(int index)
{
_sempaphore.Wait();
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
//todo - sql injection dont copy this..
var sql = $"select data from logs where id = {index}";
using (var command = new SqliteCommand(sql, connection))
{
var data = Convert.ToString(await command.ExecuteScalarAsync());
var jsonSerializerSettings = new JsonSerializerSettings()
{
TypeNameHandling = TypeNameHandling.All
};
var log = JsonConvert.DeserializeObject<LogEntry>(data, jsonSerializerSettings);
_sempaphore.Release();
return log;
}
}
}
public async Task<List<(int index, LogEntry logEntry)>> GetFrom(int index)
{
_sempaphore.Wait();
var logsToReturn = new List<(int, LogEntry)>();
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
//todo - sql injection dont copy this..
var sql = $"select id, data from logs where id >= {index}";
using (var command = new SqliteCommand(sql, connection))
{
using (var reader = await command.ExecuteReaderAsync())
{
while (reader.Read())
{
var id = Convert.ToInt32(reader[0]);
var data = (string)reader[1];
var jsonSerializerSettings = new JsonSerializerSettings()
{
TypeNameHandling = TypeNameHandling.All
};
var log = JsonConvert.DeserializeObject<LogEntry>(data, jsonSerializerSettings);
logsToReturn.Add((id, log));
}
}
}
_sempaphore.Release();
return logsToReturn;
}
}
public async Task<long> GetTermAtIndex(int index)
{
_sempaphore.Wait();
long result = 0;
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
//todo - sql injection dont copy this..
var sql = $"select data from logs where id = {index}";
using (var command = new SqliteCommand(sql, connection))
{
var data = Convert.ToString(await command.ExecuteScalarAsync());
var jsonSerializerSettings = new JsonSerializerSettings()
{
TypeNameHandling = TypeNameHandling.All
};
var log = JsonConvert.DeserializeObject<LogEntry>(data, jsonSerializerSettings);
if (log != null && log.Term > result)
{
result = log.Term;
}
}
}
_sempaphore.Release();
return result;
}
public async Task Remove(int indexOfCommand)
{
_sempaphore.Wait();
using (var connection = new SqliteConnection($"Data Source={_path};"))
{
connection.Open();
//todo - sql injection dont copy this..
var deleteSql = $"delete from logs where id >= {indexOfCommand};";
_logger.LogInformation($"id: {_nodeId.Id} Remove {deleteSql}");
using (var deleteCommand = new SqliteCommand(deleteSql, connection))
{
var result = await deleteCommand.ExecuteNonQueryAsync();
}
}
_sempaphore.Release();
}
}
}

15
src/Ocelot/Raft/UpdateFileConfiguration.cs
View File

@@ -1,15 +0,0 @@
using Ocelot.Configuration.File;
using Rafty.FiniteStateMachine;
namespace Ocelot.Raft
{
public class UpdateFileConfiguration : ICommand
{
public UpdateFileConfiguration(FileConfiguration configuration)
{
Configuration = configuration;
}
public FileConfiguration Configuration {get;private set;}
}
}

1
test/Ocelot.AcceptanceTests/Ocelot.AcceptanceTests.csproj
View File

@@ -52,6 +52,5 @@
<PackageReference Include="TestStack.BDDfy" Version="4.3.2" />
<PackageReference Include="xunit" Version="2.3.1" />
<PackageReference Include="Butterfly.Client.AspNetCore" Version="0.0.8" />
<PackageReference Include="Rafty" Version="0.4.4" />
</ItemGroup>
</Project>

1
test/Ocelot.IntegrationTests/Ocelot.IntegrationTests.csproj
View File

@@ -42,7 +42,6 @@
<PackageReference Include="IdentityServer4" Version="2.2.0" />
<PackageReference Include="Shouldly" Version="3.0.0" />
<PackageReference Include="TestStack.BDDfy" Version="4.3.2" />
<PackageReference Include="Rafty" Version="0.4.4" />
<PackageReference Include="Microsoft.Data.SQLite" Version="2.1.0" />
</ItemGroup>
</Project>

515
test/Ocelot.IntegrationTests/RaftTests.cs
View File

@@ -1,515 +0,0 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Threading;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using Newtonsoft.Json;
using Ocelot.Configuration.File;
using Ocelot.Raft;
using Rafty.Concensus;
using Rafty.Infrastructure;
using Shouldly;
using Xunit;
using static Rafty.Infrastructure.Wait;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Configuration;
using Ocelot.DependencyInjection;
using Ocelot.Middleware;
namespace Ocelot.IntegrationTests
{
using System.Threading.Tasks;
using Xunit.Abstractions;
public class RaftTests : IDisposable
{
private readonly List<IWebHost> _builders;
private readonly List<IWebHostBuilder> _webHostBuilders;
private readonly List<Thread> _threads;
private FilePeers _peers;
private HttpClient _httpClient;
private readonly HttpClient _httpClientForAssertions;
private BearerToken _token;
private HttpResponseMessage _response;
private static readonly object _lock = new object();
private ITestOutputHelper _output;
public RaftTests(ITestOutputHelper output)
{
_output = output;
_httpClientForAssertions = new HttpClient();
_webHostBuilders = new List<IWebHostBuilder>();
_builders = new List<IWebHost>();
_threads = new List<Thread>();
}
[Fact(Skip = "Still not stable, more work required in rafty..")]
public async Task should_persist_command_to_five_servers()
{
var peers = new List<FilePeer>
{
new FilePeer {HostAndPort = "http://localhost:5000"},
new FilePeer {HostAndPort = "http://localhost:5001"},
new FilePeer {HostAndPort = "http://localhost:5002"},
new FilePeer {HostAndPort = "http://localhost:5003"},
new FilePeer {HostAndPort = "http://localhost:5004"}
};
var configuration = new FileConfiguration
{
GlobalConfiguration = new FileGlobalConfiguration
{
}
};
var updatedConfiguration = new FileConfiguration
{
GlobalConfiguration = new FileGlobalConfiguration
{
},
ReRoutes = new List<FileReRoute>()
{
new FileReRoute()
{
DownstreamHostAndPorts = new List<FileHostAndPort>
{
new FileHostAndPort
{
Host = "127.0.0.1",
Port = 80,
}
},
DownstreamScheme = "http",
DownstreamPathTemplate = "/geoffrey",
UpstreamHttpMethod = new List<string> { "get" },
UpstreamPathTemplate = "/"
},
new FileReRoute()
{
DownstreamHostAndPorts = new List<FileHostAndPort>
{
new FileHostAndPort
{
Host = "123.123.123",
Port = 443,
}
},
DownstreamScheme = "https",
DownstreamPathTemplate = "/blooper/{productId}",
UpstreamHttpMethod = new List<string> { "post" },
UpstreamPathTemplate = "/test"
}
}
};
var command = new UpdateFileConfiguration(updatedConfiguration);
GivenThePeersAre(peers);
GivenThereIsAConfiguration(configuration);
GivenFiveServersAreRunning();
await GivenIHaveAnOcelotToken("/administration");
await WhenISendACommandIntoTheCluster(command);
Thread.Sleep(5000);
await ThenTheCommandIsReplicatedToAllStateMachines(command);
}
[Fact(Skip = "Still not stable, more work required in rafty..")]
public async Task should_persist_command_to_five_servers_when_using_administration_api()
{
var peers = new List<FilePeer>
{
new FilePeer {HostAndPort = "http://localhost:5005"},
new FilePeer {HostAndPort = "http://localhost:5006"},
new FilePeer {HostAndPort = "http://localhost:5007"},
new FilePeer {HostAndPort = "http://localhost:5008"},
new FilePeer {HostAndPort = "http://localhost:5009"}
};
var configuration = new FileConfiguration
{
};
var updatedConfiguration = new FileConfiguration
{
ReRoutes = new List<FileReRoute>()
{
new FileReRoute()
{
DownstreamHostAndPorts = new List<FileHostAndPort>
{
new FileHostAndPort
{
Host = "127.0.0.1",
Port = 80,
}
},
DownstreamScheme = "http",
DownstreamPathTemplate = "/geoffrey",
UpstreamHttpMethod = new List<string> { "get" },
UpstreamPathTemplate = "/"
},
new FileReRoute()
{
DownstreamHostAndPorts = new List<FileHostAndPort>
{
new FileHostAndPort
{
Host = "123.123.123",
Port = 443,
}
},
DownstreamScheme = "https",
DownstreamPathTemplate = "/blooper/{productId}",
UpstreamHttpMethod = new List<string> { "post" },
UpstreamPathTemplate = "/test"
}
}
};
var command = new UpdateFileConfiguration(updatedConfiguration);
GivenThePeersAre(peers);
GivenThereIsAConfiguration(configuration);
GivenFiveServersAreRunning();
await GivenIHaveAnOcelotToken("/administration");
GivenIHaveAddedATokenToMyRequest();
await WhenIPostOnTheApiGateway("/administration/configuration", updatedConfiguration);
await ThenTheCommandIsReplicatedToAllStateMachines(command);
}
private void GivenThePeersAre(List<FilePeer> peers)
{
FilePeers filePeers = new FilePeers();
filePeers.Peers.AddRange(peers);
var json = JsonConvert.SerializeObject(filePeers);
File.WriteAllText("peers.json", json);
_httpClient = new HttpClient();
var ocelotBaseUrl = peers[0].HostAndPort;
_httpClient.BaseAddress = new Uri(ocelotBaseUrl);
}
private async Task WhenISendACommandIntoTheCluster(UpdateFileConfiguration command)
{
async Task<bool> SendCommand()
{
try
{
var p = _peers.Peers.First();
var json = JsonConvert.SerializeObject(command, new JsonSerializerSettings()
{
TypeNameHandling = TypeNameHandling.All
});
var httpContent = new StringContent(json);
httpContent.Headers.ContentType = new MediaTypeHeaderValue("application/json");
using (var httpClient = new HttpClient())
{
httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
var response = await httpClient.PostAsync($"{p.HostAndPort}/administration/raft/command", httpContent);
response.EnsureSuccessStatusCode();
var content = await response.Content.ReadAsStringAsync();
var errorResult = JsonConvert.DeserializeObject<ErrorResponse<UpdateFileConfiguration>>(content);
if (!string.IsNullOrEmpty(errorResult.Error))
{
return false;
}
var okResult = JsonConvert.DeserializeObject<OkResponse<UpdateFileConfiguration>>(content);
if (okResult.Command.Configuration.ReRoutes.Count == 2)
{
return true;
}
}
return false;
}
catch (Exception e)
{
Console.WriteLine(e);
return false;
}
}
var commandSent = await WaitFor(40000).Until(async () =>
{
var result = await SendCommand();
Thread.Sleep(1000);
return result;
});
commandSent.ShouldBeTrue();
}
private async Task ThenTheCommandIsReplicatedToAllStateMachines(UpdateFileConfiguration expecteds)
{
async Task<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 = await _httpClientForAssertions.GetAsync($"{peer.HostAndPort}/administration/configuration");
var json = await result.Content.ReadAsStringAsync();
var response = JsonConvert.DeserializeObject<FileConfiguration>(json, new JsonSerializerSettings{TypeNameHandling = TypeNameHandling.All});
response.GlobalConfiguration.RequestIdKey.ShouldBe(expecteds.Configuration.GlobalConfiguration.RequestIdKey);
response.GlobalConfiguration.ServiceDiscoveryProvider.Host.ShouldBe(expecteds.Configuration.GlobalConfiguration.ServiceDiscoveryProvider.Host);
response.GlobalConfiguration.ServiceDiscoveryProvider.Port.ShouldBe(expecteds.Configuration.GlobalConfiguration.ServiceDiscoveryProvider.Port);
for (var i = 0; i < response.ReRoutes.Count; i++)
{
for (var j = 0; j < response.ReRoutes[i].DownstreamHostAndPorts.Count; j++)
{
var res = response.ReRoutes[i].DownstreamHostAndPorts[j];
var expected = expecteds.Configuration.ReRoutes[i].DownstreamHostAndPorts[j];
res.Host.ShouldBe(expected.Host);
res.Port.ShouldBe(expected.Port);
}
response.ReRoutes[i].DownstreamPathTemplate.ShouldBe(expecteds.Configuration.ReRoutes[i].DownstreamPathTemplate);
response.ReRoutes[i].DownstreamScheme.ShouldBe(expecteds.Configuration.ReRoutes[i].DownstreamScheme);
response.ReRoutes[i].UpstreamPathTemplate.ShouldBe(expecteds.Configuration.ReRoutes[i].UpstreamPathTemplate);
response.ReRoutes[i].UpstreamHttpMethod.ShouldBe(expecteds.Configuration.ReRoutes[i].UpstreamHttpMethod);
}
passed++;
}
return passed == 5;
}
catch(Exception e)
{
//_output.WriteLine($"{e.Message}, {e.StackTrace}");
Console.WriteLine(e);
return false;
}
}
var commandOnAllStateMachines = await WaitFor(40000).Until(async () =>
{
var result = await CommandCalledOnAllStateMachines();
Thread.Sleep(1000);
return result;
});
commandOnAllStateMachines.ShouldBeTrue();
}
private async Task WhenIPostOnTheApiGateway(string url, FileConfiguration updatedConfiguration)
{
async Task<bool> SendCommand()
{
var json = JsonConvert.SerializeObject(updatedConfiguration);
var content = new StringContent(json);
content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
_response = await _httpClient.PostAsync(url, content);
var responseContent = await _response.Content.ReadAsStringAsync();
if(responseContent == "There was a problem. This error message sucks raise an issue in GitHub.")
{
return false;
}
if(string.IsNullOrEmpty(responseContent))
{
return false;
}
return _response.IsSuccessStatusCode;
}
var commandSent = await WaitFor(40000).Until(async () =>
{
var result = await SendCommand();
Thread.Sleep(1000);
return result;
});
commandSent.ShouldBeTrue();
}
private void GivenIHaveAddedATokenToMyRequest()
{
_httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", _token.AccessToken);
}
private async Task GivenIHaveAnOcelotToken(string adminPath)
{
async Task<bool> AddToken()
{
try
{
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 = await _httpClient.PostAsync(tokenUrl, content);
var responseContent = await response.Content.ReadAsStringAsync();
if(!response.IsSuccessStatusCode)
{
return false;
}
_token = JsonConvert.DeserializeObject<BearerToken>(responseContent);
var configPath = $"{adminPath}/.well-known/openid-configuration";
response = await _httpClient.GetAsync(configPath);
return response.IsSuccessStatusCode;
}
catch(Exception)
{
return false;
}
}
var addToken = await WaitFor(40000).Until(async () =>
{
var result = await AddToken();
Thread.Sleep(1000);
return result;
});
addToken.ShouldBeTrue();
}
private void GivenThereIsAConfiguration(FileConfiguration fileConfiguration)
{
var configurationPath = $"{Directory.GetCurrentDirectory()}/ocelot.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}/ocelot.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())
.ConfigureAppConfiguration((hostingContext, config) =>
{
config.SetBasePath(hostingContext.HostingEnvironment.ContentRootPath);
var env = hostingContext.HostingEnvironment;
config.AddJsonFile("appsettings.json", optional: true, reloadOnChange: false)
.AddJsonFile($"appsettings.{env.EnvironmentName}.json", optional: true, reloadOnChange: false);
config.AddJsonFile("ocelot.json", false, false);
config.AddJsonFile("peers.json", optional: true, reloadOnChange: false);
#pragma warning disable CS0618
config.AddOcelotBaseUrl(url);
#pragma warning restore CS0618
config.AddEnvironmentVariables();
})
.ConfigureServices(x =>
{
x.AddSingleton(new NodeId(url));
x
.AddOcelot()
.AddAdministration("/administration", "secret")
.AddRafty();
})
.Configure(app =>
{
app.UseOcelot().Wait();
});
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)
{
File.Delete(peer.HostAndPort.Replace("/", "").Replace(":", ""));
File.Delete($"{peer.HostAndPort.Replace("/", "").Replace(":", "")}.db");
var thread = new Thread(() => GivenAServerIsRunning(peer.HostAndPort));
thread.Start();
_threads.Add(thread);
}
}
public void Dispose()
{
foreach (var builder in _builders)
{
builder?.Dispose();
}
foreach (var peer in _peers.Peers)
{
try
{
File.Delete(peer.HostAndPort.Replace("/", "").Replace(":", ""));
File.Delete($"{peer.HostAndPort.Replace("/", "").Replace(":", "")}.db");
}
catch (Exception e)
{
Console.WriteLine(e);
}
}
}
}
}

58
test/Ocelot.UnitTests/Controllers/FileConfigurationControllerTests.cs
View File

@@ -8,14 +8,11 @@ using Ocelot.Responses;
using TestStack.BDDfy;
using Xunit;
using Shouldly;
using Ocelot.Raft;
using Rafty.Concensus;
using Ocelot.Configuration;
namespace Ocelot.UnitTests.Controllers
{
using Ocelot.Configuration.Repository;
using Rafty.Concensus.Node;
public class FileConfigurationControllerTests
{
@@ -25,7 +22,6 @@ namespace Ocelot.UnitTests.Controllers
private IActionResult _result;
private FileConfiguration _fileConfiguration;
private readonly Mock<IServiceProvider> _provider;
private Mock<INode> _node;
public FileConfigurationControllerTests()
{
@@ -70,33 +66,6 @@ namespace Ocelot.UnitTests.Controllers
.BDDfy();
}
[Fact]
public void should_post_file_configuration_using_raft_node()
{
var expected = new FileConfiguration();
this.Given(x => GivenTheFileConfiguration(expected))
.And(x => GivenARaftNodeIsRegistered())
.And(x => GivenTheNodeReturnsOK())
.And(x => GivenTheConfigSetterReturns(new OkResponse()))
.When(x => WhenIPostTheFileConfiguration())
.Then(x => x.ThenTheNodeIsCalledCorrectly())
.BDDfy();
}
[Fact]
public void should_return_error_when_cannot_set_config_using_raft_node()
{
var expected = new FileConfiguration();
this.Given(x => GivenTheFileConfiguration(expected))
.And(x => GivenARaftNodeIsRegistered())
.And(x => GivenTheNodeReturnsError())
.When(x => WhenIPostTheFileConfiguration())
.Then(x => ThenTheResponseIs<BadRequestObjectResult>())
.BDDfy();
}
[Fact]
public void should_return_error_when_cannot_set_config()
{
@@ -110,33 +79,6 @@ namespace Ocelot.UnitTests.Controllers
.BDDfy();
}
private void ThenTheNodeIsCalledCorrectly()
{
_node.Verify(x => x.Accept(It.IsAny<UpdateFileConfiguration>()), Times.Once);
}
private void GivenARaftNodeIsRegistered()
{
_node = new Mock<INode>();
_provider
.Setup(x => x.GetService(typeof(INode)))
.Returns(_node.Object);
}
private void GivenTheNodeReturnsOK()
{
_node
.Setup(x => x.Accept(It.IsAny<UpdateFileConfiguration>()))
.ReturnsAsync(new Rafty.Infrastructure.OkResponse<UpdateFileConfiguration>(new UpdateFileConfiguration(new FileConfiguration())));
}
private void GivenTheNodeReturnsError()
{
_node
.Setup(x => x.Accept(It.IsAny<UpdateFileConfiguration>()))
.ReturnsAsync(new Rafty.Infrastructure.ErrorResponse<UpdateFileConfiguration>("error", new UpdateFileConfiguration(new FileConfiguration())));
}
private void GivenTheConfigSetterReturns(Response response)
{
_setter

22
test/Ocelot.UnitTests/DependencyInjection/OcelotBuilderTests.cs
View File

@@ -75,16 +75,6 @@ namespace Ocelot.UnitTests.DependencyInjection
.BDDfy();
}
[Fact]
public void should_set_up_rafty()
{
this.Given(x => WhenISetUpOcelotServices())
.When(x => WhenISetUpRafty())
.Then(x => ThenAnExceptionIsntThrown())
.Then(x => ThenTheCorrectAdminPathIsRegitered())
.BDDfy();
}
[Fact]
public void should_set_up_administration_with_identity_server_options()
{
@@ -248,18 +238,6 @@ namespace Ocelot.UnitTests.DependencyInjection
first.ShouldBe(second);
}
private void WhenISetUpRafty()
{
try
{
_ocelotBuilder.AddAdministration("/administration", "secret").AddRafty();
}
catch (Exception e)
{
_ex = e;
}
}
private void ThenAnOcelotBuilderIsReturned()
{
_ocelotBuilder.ShouldBeOfType<OcelotBuilder>();

1
test/Ocelot.UnitTests/Ocelot.UnitTests.csproj
View File

@@ -57,7 +57,6 @@
<PackageReference Include="TestStack.BDDfy" Version="4.3.2" />
<PackageReference Include="xunit" Version="2.3.1" />
<PackageReference Include="Butterfly.Client.AspNetCore" Version="0.0.8" />
<PackageReference Include="Rafty" Version="0.4.4" />
</ItemGroup>
<ItemGroup>

45
test/Ocelot.UnitTests/Raft/OcelotFiniteStateMachineTests.cs
View File

@@ -1,45 +0,0 @@
using Moq;
using Ocelot.Configuration.Setter;
using Ocelot.Raft;
using TestStack.BDDfy;
using Xunit;
namespace Ocelot.UnitTests.Raft
{
public class OcelotFiniteStateMachineTests
{
private UpdateFileConfiguration _command;
private OcelotFiniteStateMachine _fsm;
private Mock<IFileConfigurationSetter> _setter;
public OcelotFiniteStateMachineTests()
{
_setter = new Mock<IFileConfigurationSetter>();
_fsm = new OcelotFiniteStateMachine(_setter.Object);
}
[Fact]
public void should_handle_update_file_configuration_command()
{
this.Given(x => GivenACommand(new UpdateFileConfiguration(new Ocelot.Configuration.File.FileConfiguration())))
.When(x => WhenTheCommandIsHandled())
.Then(x => ThenTheStateIsUpdated())
.BDDfy();
}
private void GivenACommand(UpdateFileConfiguration command)
{
_command = command;
}
private void WhenTheCommandIsHandled()
{
_fsm.Handle(new Rafty.Log.LogEntry(_command, _command.GetType(), 0));
}
private void ThenTheStateIsUpdated()
{
_setter.Verify(x => x.Set(_command.Configuration), Times.Once);
}
}
}