Merge pull request #22 from zetroot/feature/better-schedulling

+semver: feature

Author: zetroot

src/NTorSpectator.Observer/NTorSpectator.Observer.csproj

@@ -13,6 +13,7 @@
         <PackageReference Include="Microsoft.AspNetCore.OpenApi" Version="7.0.5" />
         <PackageReference Include="NeoSmart.AsyncLock" Version="3.2.1" />
         <PackageReference Include="prometheus-net.AspNetCore" Version="8.0.0" />
+        <PackageReference Include="Quartz.AspNetCore" Version="3.6.2" />
         <PackageReference Include="Refit" Version="6.3.2" />
         <PackageReference Include="Refit.HttpClientFactory" Version="6.3.2" />
         <PackageReference Include="Serilog.AspNetCore" Version="6.1.0" />

src/NTorSpectator.Observer/Program.cs

@@ -6,6 +6,7 @@
 using NTorSpectator.Observer.TorIntegration;
 using NTorSpectator.Services;
 using Prometheus;
+using Quartz;
 using Refit;
 using Serilog;
 using Serilog.Formatting.Compact;
@@ -39,7 +40,21 @@
     .AddBizLogic()
     .AddDatabase(builder.Configuration)
     .AddTransient<TorControlManager>()
-    .AddHostedService<Spectator>();
+    .AddTransient<SpectatorJob>()
+    .AddHostedService<SitesUpdater>();
+
+builder.Services.AddQuartz(cfg =>
+{
+    cfg.UseMicrosoftDependencyInjectionJobFactory();
+    
+    var jobDetail = JobBuilder.Create<SpectatorJob>()
+        .WithDescription("Tor spectator job")
+        .WithIdentity("tor-spectator")
+        .Build();
+    cfg.AddJob<SpectatorJob>(jobKey: jobDetail.Key, configure: j => {});
+    cfg.AddTrigger(t => t.WithCronSchedule("0 0 * * * ?").ForJob(jobDetail));
+});
+builder.Services.AddQuartzServer(cfg => cfg.WaitForJobsToComplete = false);
 
 builder.Services
     .AddRefitClient<IMastodonClient>(sp =>

src/NTorSpectator.Observer/Services/SitesUpdater.cs

@@ -0,0 +1,36 @@
+using Microsoft.Extensions.Options;
+using NTorSpectator.Services;
+
+namespace NTorSpectator.Observer.Services;
+
+public class SitesUpdater : BackgroundService
+{
+    private readonly ILogger<SitesUpdater> _logger;
+    private readonly string _torSitesFile;
+    private readonly ISitesCatalogue _sitesCatalogue;
+
+    public SitesUpdater(ILogger<SitesUpdater> logger, IOptions<SpectatorSettings> opts, ISitesCatalogue sitesCatalogue)
+    {
+        _logger = logger;
+        _sitesCatalogue = sitesCatalogue;
+        _torSitesFile = opts.Value.SiteList;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        if (!File.Exists(_torSitesFile))
+        {
+            _logger.LogWarning("Tor sites file is not found at path {TorSitesFilePath}. Exiting", _torSitesFile);
+            return;
+        }
+        
+        var sites = await File.ReadAllLinesAsync(_torSitesFile);
+        foreach (var site in sites)
+        {
+            if (string.IsNullOrWhiteSpace(site))
+                continue;
+            await _sitesCatalogue.AddIfNotExists(site);
+        }
+        _logger.LogDebug("Finished adding sites");
+    }
+}

src/NTorSpectator.Observer/Services/Spectator.cs renamed to src/NTorSpectator.Observer/Services/SpectatorJob.cs

@@ -1,56 +1,36 @@
-using Microsoft.Extensions.Options;
 using NTorSpectator.Observer.TorIntegration;
 using NTorSpectator.Services;
 using NTorSpectator.Services.Models;
+using Quartz;
 
 namespace NTorSpectator.Observer.Services;
 
-public class Spectator : BackgroundService
+public class SpectatorJob : IJob
 {
-    private readonly ILogger<Spectator> _logger;
-    private readonly TorControlManager _torControl;
-    private readonly string _torSitesFile;
-    private readonly TimeSpan _cooldown;
+    private readonly ILogger<SpectatorJob> _logger;
     private readonly ISitesCatalogue _sitesCatalogue;
+    private readonly TorControlManager _torControl;
     private readonly ISiteObserver _siteObserver;
 
-    public Spectator(ILogger<Spectator> logger, TorControlManager torControl, IOptions<SpectatorSettings> opts, ISitesCatalogue sitesCatalogue, ISiteObserver siteObserver)
+    public SpectatorJob(ILogger<SpectatorJob> logger, ISitesCatalogue sitesCatalogue, TorControlManager torControl, ISiteObserver siteObserver)
     {
         _logger = logger;
-        _torControl = torControl;
         _sitesCatalogue = sitesCatalogue;
+        _torControl = torControl;
         _siteObserver = siteObserver;
-        _torSitesFile = opts.Value.SiteList;
-        _cooldown = opts.Value.CooldownInterval;
-    }
-
-    public bool IsRunning { get; private set; }
-    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
-    {
-        try
-        {
-            IsRunning = true;
-            await UpdateSites();
-            while (!stoppingToken.IsCancellationRequested)
-            {
-                await Watch();
-                await Task.Delay(_cooldown, stoppingToken);
-            }
-        }
-        finally
-        {
-            IsRunning = false;
-        }
     }
 
-    private async Task Watch()
+    public async Task Execute(IJobExecutionContext context)
     {
+        _logger.LogDebug("Starting sites observations");
         var sites = await _sitesCatalogue.GetAllSites();
-        var siteQueue = new Queue<QueuedSite>(sites.Select(x => new QueuedSite(x, 0)));
+        _logger.LogDebug("Got {Count} sites to observe", sites.Count);
 
+        var siteQueue = new Queue<QueuedSite>(sites.Select(x => new QueuedSite(x, 0)));
         while(siteQueue.TryDequeue(out var queuedSite))
         {
             using var _ = _logger.BeginScope(new Dictionary<string, object> { { "HiddenService", queuedSite.Site.SiteUri } });
+            _logger.LogDebug("Starting observations on the next site");
             try
             {
                 var observations = await ObserveSite(queuedSite.Site.SiteUri);
@@ -65,6 +45,7 @@ private async Task Watch()
                         continue;
                     }
                 }
+                _logger.LogDebug("Site seems to be up");
                 await _siteObserver.AddNewObservation(queuedSite.Site.SiteUri, observations.IsOk);
                 _logger.LogInformation("Site observed");
             }
@@ -73,28 +54,16 @@ private async Task Watch()
                 _logger.LogError(e, "Observation for site failed");
             }
         }
+        _logger.LogDebug("The queue is finally empty, observations finished");
     }
-
-    private async Task UpdateSites()
-    {
-        var sites = await File.ReadAllLinesAsync(_torSitesFile);
-        foreach (var site in sites)
-        {
-            await _sitesCatalogue.AddIfNotExists(site);
-        }
-    }
+    
+    private record QueuedSite(Site Site, int ObservationsCount);
+    
     private async Task<TorWatchResults> ObserveSite(string site)
     {
         var torReply = await _torControl.HsFetch(site);
         var positive = torReply.Count(x => x.Action == HsDescAction.Received);
         var negative = torReply.Count(x => x.Action == HsDescAction.Failed);
         return new(site, positive, negative);
     }
-
-    private record QueuedSite(Site Site, int ObservationsCount);
-}
-
-public record TorWatchResults(string Site, int Positive, int Negative)
-{
-    public bool IsOk => Positive > 0;
 }

src/NTorSpectator.Observer/Services/TorWatchResults.cs

@@ -0,0 +1,6 @@
+namespace NTorSpectator.Observer.Services;
+
+public record TorWatchResults(string Site, int Positive, int Negative)
+{
+    public bool IsOk => Positive > 0;
+}

src/NTorSpectator.Observer/TorIntegration/TorControlManager.cs

@@ -86,7 +86,7 @@ public async Task<IReadOnlyCollection<HsDescEventReply>> HsFetch(string serviceD
                 _logger.LogTrace("Received {Count} messages", responses.Length);
 
                 var replies = responses.Select(x => TorReply.TryParse(x, out var reply) ? reply : null).Where(x => x != null).ToList();
-                _logger.LogInformation("Parsed {Count}", replies.Count);
+                _logger.LogDebug("Parsed {Count}", replies.Count);
                 var hsDescEvents = replies.OfType<HsDescEventReply>().Where(x => x.Action is HsDescAction.Received or HsDescAction.Failed).ToList();
                 results.AddRange(hsDescEvents);
                 if (results.Any(x => x.Action == HsDescAction.Received))