Merge pull request #39 from zetroot/revert-35-feature/parallel-observations

Revert "Feature/parallel observations"

Author: zetroot

src/NTorSpectator.Observer/Program.cs

@@ -50,7 +50,6 @@
     var jobDetail = JobBuilder.Create<SpectatorJob>()
         .WithDescription("Tor spectator job")
         .WithIdentity("tor-spectator")
-        .DisallowConcurrentExecution()
         .Build();
     cfg.AddJob<SpectatorJob>(jobKey: jobDetail.Key, configure: j => {});
     cfg.AddTrigger(t => t.WithCronSchedule("0 0 * * * ?").ForJob(jobDetail));

src/NTorSpectator.Observer/Services/SpectatorJob.cs

@@ -1,4 +1,3 @@
-using System.Collections.Concurrent;
 using System.Diagnostics;
 using NTorSpectator.Observer.TorIntegration;
 using NTorSpectator.Services;
@@ -23,13 +22,15 @@ public class SpectatorJob : IJob
 
     private readonly ILogger<SpectatorJob> _logger;
     private readonly ISitesCatalogue _sitesCatalogue;
-    private readonly IServiceProvider _services;
-    
-    public SpectatorJob(ILogger<SpectatorJob> logger, ISitesCatalogue sitesCatalogue, IServiceProvider services)
+    private readonly TorControlManager _torControl;
+    private readonly ISiteObserver _siteObserver;
+
+    public SpectatorJob(ILogger<SpectatorJob> logger, ISitesCatalogue sitesCatalogue, TorControlManager torControl, ISiteObserver siteObserver)
     {
         _logger = logger;
         _sitesCatalogue = sitesCatalogue;
-        _services = services;
+        _torControl = torControl;
+        _siteObserver = siteObserver;
     }
 
     public async Task Execute(IJobExecutionContext context)
@@ -39,74 +40,50 @@ public async Task Execute(IJobExecutionContext context)
         _logger.LogDebug("Starting sites observations");
         var sites = await _sitesCatalogue.GetAllSites();
         _logger.LogDebug("Got {Count} sites to observe", sites.Count);
-        var queuedSites = sites.Select(x => new QueuedSite(x, 0));
-        var siteQueue = new ConcurrentQueue<QueuedSite>(queuedSites);
-        while(true)
+        
+        var siteQueue = new Queue<QueuedSite>(sites.Select(x => new QueuedSite(x, 0)));
+        while(siteQueue.TryDequeue(out var queuedSite))
         {
             QueueLength.Set(siteQueue.Count);
-            if (!siteQueue.Any())
-                break;
-            var chunk = TryDequeueMax(siteQueue, 4);
-            var tasks = chunk.Select(x => TryCheckSite(x, s => siteQueue.Enqueue(s))).ToArray();
-            await Task.WhenAll(tasks);
-        }
-        _logger.LogDebug("The queue is finally empty, observations finished");
-        sw.Stop();
-        TotalSessionDuration.Set(sw.ElapsedMilliseconds);
-    }
-
-
-    private QueuedSite[] TryDequeueMax(ConcurrentQueue<QueuedSite> queue, int max)
-    {
-        var result = new List<QueuedSite>(max);
-        for (int i = 0; i < max; ++i)
-        {
-            if (queue.TryDequeue(out var item))
-                result.Add(item);
-        }
-        return result.ToArray();
-    }
-    
-    private async Task TryCheckSite(QueuedSite queuedSite, Action<QueuedSite> putBackIntoQueue)
-    {
-        using var _ = _logger.BeginScope(new Dictionary<string, object> { { "HiddenService", queuedSite.Site.SiteUri } });
-        using var scope = _services.CreateScope();
-        var torControl = scope.ServiceProvider.GetRequiredService<TorControlManager>();
-        var siteObserver = scope.ServiceProvider.GetRequiredService<ISiteObserver>();
-        _logger.LogDebug("Starting observations on the next site");
-        try
-        {
-            var observations = await ObserveSite(queuedSite.Site.SiteUri, torControl);
-            ObservationsCount.Inc();
-            if (!observations.IsOk)
+            using var _ = _logger.BeginScope(new Dictionary<string, object> { { "HiddenService", queuedSite.Site.SiteUri } });
+            _logger.LogDebug("Starting observations on the next site");
+            try
             {
-                _logger.LogDebug("Site observed as not available");
-                var siteObservationsCount = queuedSite.ObservationsCount;
-                if (siteObservationsCount < 3)
+                var observations = await ObserveSite(queuedSite.Site.SiteUri);
+                ObservationsCount.Inc();
+                if (!observations.IsOk)
                 {
-                    _logger.LogDebug("Site has been observed {Count} times, returning it to queue", siteObservationsCount);
-                    putBackIntoQueue(queuedSite with { ObservationsCount = siteObservationsCount + 1 });
-                    RetriesCount.Inc();
-                    return;
+                    _logger.LogDebug("Site observed as not available");
+                    var siteObservationsCount = queuedSite.ObservationsCount;
+                    if (siteObservationsCount < 3)
+                    {
+                        _logger.LogDebug("Site has been observed {Count} times, returning it to queue", siteObservationsCount);
+                        siteQueue.Enqueue(queuedSite with{ObservationsCount = siteObservationsCount + 1});
+                        RetriesCount.Inc();
+                        continue;
+                    }
                 }
+                _logger.LogDebug("Site seems to be up");
+                await _siteObserver.AddNewObservation(queuedSite.Site.SiteUri, observations.IsOk);
+                SiteStatus.WithLabels(queuedSite.Site.SiteUri).Set(observations.IsOk ? 1 : 0);
+                _logger.LogInformation("Site observed");
+            }
+            catch (Exception e)
+            {
+                _logger.LogError(e, "Observation for site failed");
             }
-            _logger.LogDebug("Site seems to be up");
-            await siteObserver.AddNewObservation(queuedSite.Site.SiteUri, observations.IsOk);
-            SiteStatus.WithLabels(queuedSite.Site.SiteUri).Set(observations.IsOk ? 1 : 0);
-            _logger.LogInformation("Site observed");
-        }
-        catch (Exception e)
-        {
-            _logger.LogError(e, "Observation for site failed");
         }
+        _logger.LogDebug("The queue is finally empty, observations finished");
+        sw.Stop();
+        TotalSessionDuration.Set(sw.ElapsedMilliseconds);
     }
-
+    
     private record QueuedSite(Site Site, int ObservationsCount);
 
-    private async Task<TorWatchResults> ObserveSite(string site, TorControlManager torControl)
+    private async Task<TorWatchResults> ObserveSite(string site)
     {
         using var _ = RequestDuration.NewTimer();
-        var torReply = await torControl.HsFetch(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);