We should be able to run multiple instances on the same background service on multiple servers. At the same time, only one instance should be run in "Active" mode, which means do work. Others should run in "Passive" mode, waiting until the "Active" one goes down, and then automatically switch from "Passive" to "Active".
Architecture
As you can see in the image above, we will implement a push-oriented service discovery functionality to fulfil requirements. Each instance will register itself at the start in the external data source (can be whatever Azure Database/SQL Server/No-SQL). All instances will contain a logic that will tell them which mode they should work. In case of an unexpected server failure, we will implement the logic of auto un-registration of instance.
Requirements:
Functional requirements:
Each Background Service has to be able to manipulate data.
Each Background Service has to be able to work in "Passive" mode.
Each Background Service has to be able to work in "Active" mode.
Each Background Service will register itself at the start.
Each Background Service will re-register itself in X seconds.
Each Background Service will un-register itself at the exit.
If Background Service is unable to unregister itself — for example, server crashes — then it should be un-registered by another service.
Non Functional Requirements:
NET 6 Worker Process.
Servers have limited access to the internet.
Servers can crash un-expected.
Background Services
NET 6 Host allow registering multiple Hosted Services as part of the same host. In this case, the application will contain two long-running background services that work in parallel.
Implementation
Let's start with background services implementation.
using ActivePassive.Services.Interfaces;
namespace ActivePassive
{
public class RegisterInstanceWorkerProcess : BackgroundService
{
private readonly ILogger<Worker> _logger;
private readonly IServiceProvider_serviceProvider;
public static readonly string InstanceName=$"
{nameof(RegisterInstanceWorkerProcess)}
{DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()}";
private IRegisterInstanceService? _registerInstanceService;
private readonly IHostApplicationLifetime_applicationLifetime;
public RegisterInstanceWorkerProcess(ILogger<Worker> logger,
IServiceProviderserviceProvider,
IHostApplicationLifetimeapplicationLifetime)
{
_logger=logger;
_serviceProvider=serviceProvider;
_applicationLifetime=applicationLifetime;
}
protected override async TaskExecuteAsync
(CancellationTokenstoppingToken)
{
using var scope = _serviceProvider.CreateScope();
_registerInstanceService = scope.ServiceProvider.
GetService<IRegisterInstanceService>();
ArgumentNullException.ThrowIfNull(_registerInstanceService);
while (!stoppingToken.IsCancellationRequested)
{
try
{
await _registerInstanceService.Register(InstanceName);
await _registerInstanceService.DeleteOrphanInstances();
_logger.LogInformation("Worker running at:
{time}", DateTimeOffset.Now);
await Task.Delay(1000, stoppingToken);
}
catch (Exceptionex)
{
_logger.LogError(ex, "Critical error forced worker to
shutdown");
_applicationLifetime.StopApplication();
}
}
}
public override async TaskStopAsync
(CancellationTokencancellationToken)
{
try
{
await _registerInstanceService.UnRegister(InstanceName);
_logger.LogInformation("UnRegister Instance
{instanceName} running at: {time}.",
InstanceName, DateTimeOffset.Now);
}
catch
{
_logger.LogError("Unable to UnRegister Instance
{
instanceName} running at: {time}.",
InstanceName, DateTimeOffset.Now);
}
awaitbase.StopAsync(cancellationToken);
}
}
public class Worker : BackgroundService
{
private readonly ILogger<Worker> _logger;
private readonly IServiceProvider_serviceProvider;
private readonly IHostApplicationLifetime_hostApplicationLifetime;
public Worker(ILogger<Worker> logger,
IServiceProviderserviceProvider,
IHostApplicationLifetimehostApplicationLifetime)
{
_logger = logger;
_serviceProvider = serviceProvider;
_hostApplicationLifetime = hostApplicationLifetime;
}
protected override async TaskExecuteAsync
(CancellationTokenstoppingToken)
{
try
{
using var scope = _serviceProvider.CreateScope();
var registerInstanceService = scope.ServiceProvider.
GetService<IRegisterInstanceService>();
ArgumentNullException.ThrowIfNull(registerInstanceService);
while (! stoppingToken.IsCancellationRequested)
{
if (await (registerInstanceService.IsActive
(RegisterInstanceWorkerProcess.InstanceName)))
{
await DoInActive(stoppingToken);
}
else
{
await DoInPassive(stoppingToken);
}
}
}
catch (Exceptionex)
{
_logger.LogError(ex, "Critical error forced worker to shutdown");
_hostApplicationLifetime.StopApplication();
}
}
private async TaskDoInActive(CancellationTokenstoppingToken)
{
_logger.LogInformation("Worker running at:
{time} in {mode} mode", DateTimeOffset.Now, "Active");
await Task.Delay(1000, stoppingToken);
}
private async TaskDoInPassive(CancellationTokenstoppingToken)
{
_logger.LogInformation("Worker running at:
{time} in {mode} mode", DateTimeOffset.Now, "Passive");
await Task.Delay(1000, stoppingToken);
}
}
}RegisterInstanceWorkerProcess
In Execute Async method (line 24) lives our whole logic. AS this is Worker Process, we have a while loop which iterates each 1second and do below:
Register worker process instance with unique instance id.
Delete orphaned instances
Wait 1 second and repeat the whole operation.
In case of any error, we log the error, stop the instance and try to unregister it (StopAsync method). StopAsync method will trigger only if we gracefully stop the worker process. In case of unexpected critical failure (for example, the server goes down as electricity is off), the method will not execute. In this scenario, another instance of the worker process will unregister this one (DeleteOprhantInstanes method in line 36).
Worker
The worker process is much simpler — in a while loop, we check if the instance should be active or passive. Depending on the outcome, we execute corresponding logic (DoInActive and DoInPassive methods).
RegisterInstanceService
The core of this article is our RegisterInstanceService class. This class contains four methods.
Let's take a look:
using ActivePassive.Data.Interfaces;
using ActivePassive.Model;
using ActivePassive.Services.Interfaces;
namespace ActivePassive.Services
{
publicclassRegisterInstanceService : IRegisterInstanceService
{
private readonly
IRegisterInstanceRepository_registerInstanceRepository;
private readonly int _timeUntilUnregisterInSeconds;
public RegisterInstanceService
(IRegisterInstanceRepositoryregisterInstanceRepository
, IConfigurationconfiguration)
{
_registerInstanceRepository=registerInstanceRepository;
if (!int.TryParse(configuration
["TimeUntilUnregisterInSeconds"],
out_timeUntilUnregisterInSeconds))
{
throw new ArgumentException(nameof
(_timeUntilUnregisterInSeconds));
}
}
public async TaskRegister(string instanceId)
{
var instances = await _registerInstanceRepository.GetAll();
var registeredInstance = instances.FirstOrDefault
(x=>x.InstanceRegistrationId == instanceId);
var instance = registeredInstance??
InstanceRegistration.New(instanceId);
instance.UpdateRegistrationDate();
if (!instances.Any() ||!instances.Any(x=>x.Active))
{
instance.Active=true;
}
await _registerInstanceRepository.Add(instance);
}
public async TaskUnRegister(string instanceId)
{
await _registerInstanceRepository.Delete(instanceId);
}
public async Task<bool> IsActive(string instanceId)
{
var instances = await _registerInstanceRepository.GetAll();
var instance=instances.FirstOrDefault
(x=>x.InstanceRegistrationId==instanceId);
return instance is notnull && instance.Active;
}
public async Task<List<string>> DeleteOrphanInstances()
{
var instances = await _registerInstanceRepository.GetAll();
var oldRegistrations = instances.Where
(x=>x.LastUpdated.Value.AddSeconds
(_timeUntilUnregisterInSeconds)
<DateTime.UtcNow).ToList();
foreach (var instanceRegistration in oldRegistrations)
{
await _registerInstanceRepository.Delete
(instanceRegistration.InstanceRegistrationId);
}
return oldRegistrations.Select
(x=>x.InstanceRegistrationId).ToList();
}
}
}In Line 20, we have a Register method.
The method fetches all registered instances, and If there isn't any currently active in registered instances, we register a new instance and set it as active. If our instance already exists (as we register each 10s), we just update the registration date.
Unregister method (line 36) simply deletes an instance for our datastore.
IsActive (line 41) method is quite simple too. From all registered instances, we try to find the one with the current id and return the active flag.
DeleteOrphantInstances is more interesting. This method allows us to remove all instances that are still in the data store but hasn't been updated for at least 1 minute. This method guarantee that even though some instance can crash without unregistering, we still automatically remove them from the datastore, allowing other instance to take over an "Active" mode.
Summary
Sometimes we cannot use dedicated tools/frameworks to achieve the functionality requested by our stakeholders. This time it was an active-pasive functionality.
Source: Medium - Norbert Dębosz
The Tech Platform