Timer workers (part 1)

src/ApiService/ApiService/Log.cs

@@ -8,7 +8,7 @@ namespace Microsoft.OneFuzz.Service;
 public interface ILog {
     void Log(Guid correlationId, String message, SeverityLevel level, IReadOnlyDictionary<string, string> tags, string? caller);
     void LogEvent(Guid correlationId, String evt, IReadOnlyDictionary<string, string> tags, IReadOnlyDictionary<string, double>? metrics, string? caller);
-    void LogException(Guid correlationId, Exception ex, IReadOnlyDictionary<string, string> tags, IReadOnlyDictionary<string, double>? metrics, string? caller);
+    void LogException(Guid correlationId, Exception ex, string message, IReadOnlyDictionary<string, string> tags, IReadOnlyDictionary<string, double>? metrics, string? caller);
     void Flush();
 }
 
@@ -37,7 +37,7 @@ public void LogEvent(Guid correlationId, String evt, IReadOnlyDictionary<string,
 
         _telemetryClient.TrackEvent(evt, properties: copyTags, metrics: copyMetrics);
     }
-    public void LogException(Guid correlationId, Exception ex, IReadOnlyDictionary<string, string> tags, IReadOnlyDictionary<string, double>? metrics, string? caller) {
+    public void LogException(Guid correlationId, Exception ex, string message, IReadOnlyDictionary<string, string> tags, IReadOnlyDictionary<string, double>? metrics, string? caller) {
         Dictionary<string, string> copyTags = new(tags);
         copyTags["Correlation ID"] = correlationId.ToString();
         if (caller is not null) copyTags["CalledBy"] = caller;
@@ -47,6 +47,8 @@ public void LogException(Guid correlationId, Exception ex, IReadOnlyDictionary<s
             copyMetrics = new(metrics);
         }
         _telemetryClient.TrackException(ex, copyTags, copyMetrics);
+
+        Log(correlationId, $"{message} : {ex.Message}", SeverityLevel.Error, tags, caller);
     }
 
     public void Flush() {
@@ -89,8 +91,8 @@ public void LogEvent(Guid correlationId, String evt, IReadOnlyDictionary<string,
         LogTags(correlationId, tags);
         LogMetrics(correlationId, metrics);
     }
-    public void LogException(Guid correlationId, Exception ex, IReadOnlyDictionary<string, string> tags, IReadOnlyDictionary<string, double>? metrics, string? caller) {
-        System.Console.Out.WriteLine($"[{correlationId}][Exception] {ex}");
+    public void LogException(Guid correlationId, Exception ex, string message, IReadOnlyDictionary<string, string> tags, IReadOnlyDictionary<string, double>? metrics, string? caller) {
+        System.Console.Out.WriteLine($"[{correlationId}][Exception] {message}:{ex}");
         LogTags(correlationId, tags);
         LogMetrics(correlationId, metrics);
     }
@@ -105,7 +107,7 @@ public interface ILogTracer {
     void Critical(string message);
     void Error(string message);
     void Event(string evt, IReadOnlyDictionary<string, double>? metrics);
-    void Exception(Exception ex, IReadOnlyDictionary<string, double>? metrics = null);
+    void Exception(Exception ex, string message = "", IReadOnlyDictionary<string, double>? metrics = null);
     void ForceFlush();
     void Info(string message);
     void Warning(string message);
@@ -238,10 +240,10 @@ public void Event(string evt, IReadOnlyDictionary<string, double>? metrics) {
         }
     }
 
-    public void Exception(Exception ex, IReadOnlyDictionary<string, double>? metrics) {
+    public void Exception(Exception ex, string message, IReadOnlyDictionary<string, double>? metrics) {
         var caller = GetCaller();
         foreach (var logger in _loggers) {
-            logger.LogException(CorrelationId, ex, Tags, metrics, caller);
+            logger.LogException(CorrelationId, ex, message, Tags, metrics, caller);
         }
     }
 

src/ApiService/ApiService/OneFuzzTypes/Enums.cs

@@ -121,9 +121,9 @@ public enum JobState {
 }
 
 public static class JobStateHelper {
-    private static readonly HashSet<JobState> _shuttingDown = new HashSet<JobState>(new[] { JobState.Stopping, JobState.Stopped });
-    private static readonly HashSet<JobState> _avaiable = new HashSet<JobState>(new[] { JobState.Init, JobState.Enabled });
-    private static readonly HashSet<JobState> _needsWork = new HashSet<JobState>(new[] { JobState.Init, JobState.Stopping });
+    private static readonly IReadOnlySet<JobState> _shuttingDown = new HashSet<JobState>(new[] { JobState.Stopping, JobState.Stopped });
+    private static readonly IReadOnlySet<JobState> _avaiable = new HashSet<JobState>(new[] { JobState.Init, JobState.Enabled });
+    private static readonly IReadOnlySet<JobState> _needsWork = new HashSet<JobState>(new[] { JobState.Init, JobState.Stopping });
 
     public static IReadOnlySet<JobState> Available => _avaiable;
     public static IReadOnlySet<JobState> NeedsWork => _needsWork;
@@ -354,3 +354,33 @@ public enum AgentMode {
     Repro,
     Proxy
 }
+
+
+public enum NodeState {
+    Init,
+    Free,
+    SettingUp,
+    Rebooting,
+    Ready,
+    Busy,
+    Done,
+    Shutdown,
+    Halt,
+}
+
+public static class NodeStateHelper {
+
+    private static readonly IReadOnlySet<NodeState> _needsWork = new HashSet<NodeState>(new[] { NodeState.Done, NodeState.Shutdown, NodeState.Halt });
+    private static readonly IReadOnlySet<NodeState> _readyForReset = new HashSet<NodeState>(new[] { NodeState.Done, NodeState.Shutdown, NodeState.Halt });
+    private static readonly IReadOnlySet<NodeState> _canProcessNewWork = new HashSet<NodeState>(new[] { NodeState.Free });
+
+
+    public static IReadOnlySet<NodeState> NeedsWork => _needsWork;
+
+    ///If Node is in one of these states, ignore updates from the agent.
+    public static IReadOnlySet<NodeState> ReadyForReset => _readyForReset;
+
+    public static IReadOnlySet<NodeState> CanProcessNewWork => _canProcessNewWork;
+}
+
+

src/ApiService/ApiService/OneFuzzTypes/Events.cs

@@ -60,6 +60,7 @@ public EventType GetEventType() {
                 EventScalesetFailed _ => EventType.ScalesetFailed,
                 EventScalesetResizeScheduled _ => EventType.ScalesetResizeScheduled,
                 EventScalesetStateUpdated _ => EventType.ScalesetStateUpdated,
+                EventNodeDeleted _ => EventType.NodeDeleted,
                 _ => throw new NotImplementedException(),
             };
 
@@ -81,7 +82,10 @@ public static Type GetTypeInfo(EventType eventType) {
             EventType.TaskFailed => typeof(EventTaskFailed),
             EventType.TaskStopped => typeof(EventTaskStopped),
             EventType.TaskStateUpdated => typeof(EventTaskStateUpdated),
-
+            EventType.ScalesetFailed => typeof(EventScalesetFailed),
+            EventType.ScalesetResizeScheduled => typeof(EventScalesetResizeScheduled),
+            EventType.ScalesetStateUpdated => typeof(EventScalesetStateUpdated),
+            EventType.NodeDeleted => typeof(EventNodeDeleted),
             _ => throw new ArgumentException($"invalid input {eventType}"),
 
         };
@@ -102,7 +106,7 @@ TaskConfig Config
 ) : BaseEvent();
 
 
-record EventTaskFailed(
+public record EventTaskFailed(
     Guid JobId,
     Guid TaskId,
     Error Error,
@@ -118,14 +122,14 @@ TaskConfig Config
 //    ) : BaseEvent();
 
 
-record JobTaskStopped(
+public record JobTaskStopped(
     Guid TaskId,
     TaskType TaskType,
     Error? Error
     ) : BaseEvent();
 
 
-record EventJobStopped(
+public record EventJobStopped(
     Guid JobId,
     JobConfig Config,
     UserInfo? UserInfo,
@@ -141,7 +145,7 @@ List<JobTaskStopped> TaskInfo
 //    ) : BaseEvent();
 
 
-record EventTaskStateUpdated(
+public record EventTaskStateUpdated(
     Guid JobId,
     Guid TaskId,
     TaskState State,
@@ -245,11 +249,11 @@ PoolName PoolName
     ) : BaseEvent();
 
 
-//    record EventNodeDeleted(
-//        Guid MachineId,
-//        Guid ScalesetId,
-//        PoolName PoolName
-//        ) : BaseEvent();
+public record EventNodeDeleted(
+    Guid MachineId,
+    Guid? ScalesetId,
+    PoolName PoolName
+) : BaseEvent();
 
 
 public record EventScalesetStateUpdated(

src/ApiService/ApiService/OneFuzzTypes/Model.cs

@@ -70,19 +70,8 @@ public record NodeTasks
     Guid MachineId,
     Guid TaskId,
     NodeTaskState State = NodeTaskState.Init
-);
+) : StatefulEntityBase<NodeTaskState>(State);
 
-public enum NodeState {
-    Init,
-    Free,
-    SettingUp,
-    Rebooting,
-    Ready,
-    Busy,
-    Done,
-    Shutdown,
-    Halt,
-}
 
 public record ProxyHeartbeat
 (

src/ApiService/ApiService/Program.cs

@@ -88,6 +88,9 @@ public static void Main() {
             .AddScoped<ILogAnalytics, LogAnalytics>()
             .AddScoped<IExtensions, Extensions>()
             .AddScoped<IVmssOperations, VmssOperations>()
+            .AddScoped<INodeTasksOperations, NodeTasksOperations>()
+            .AddScoped<INodeMessageOperations, NodeMessageOperations>()
+
             .AddSingleton<ICreds, Creds>()
             .AddSingleton<IServiceConfig, ServiceConfiguration>()
             .AddHttpClient()

src/ApiService/ApiService/TimerWorkers.cs

@@ -0,0 +1,40 @@
+namespace Microsoft.OneFuzz.Service;
+
+public class TimerWorkers {
+    ILogTracer _log;
+    IScalesetOperations _scaleSetOps;
+
+    public TimerWorkers(ILogTracer log, IScalesetOperations scaleSetOps) {
+        _log = log;
+        _scaleSetOps = scaleSetOps;
+    }
+
+    void ProcessScaleSets(Scaleset scaleset) {
+        _log.Verbose($"checking scaleset for updates: {scaleset.ScalesetId}");
+
+        _scaleSetOps.UpdateConfigs(scaleset);
+
+
+
+    }
+
+
+    //public async Async.Task Run([TimerTrigger("00:01:30")] TimerInfo t) {
+    // NOTE: Update pools first, such that scalesets impacted by pool updates
+    // (such as shutdown or resize) happen during this iteration `timer_worker`
+    // rather than the following iteration.
+
+
+
+
+    // NOTE: Nodes, and Scalesets should be processed in a consistent order such
+    // during 'pool scale down' operations. This means that pools that are
+    // scaling down will more likely remove from the same scalesets over time.
+    // By more likely removing from the same scalesets, we are more likely to
+    // get to empty scalesets, which can safely be deleted.
+
+
+    //}
+
+
+}

src/ApiService/ApiService/onefuzzlib/Events.cs

@@ -1,5 +1,4 @@
-using System.Text;
-using System.Text.Json;
+using System.Text.Json;
 using System.Text.Json.Serialization;
 using Microsoft.OneFuzz.Service.OneFuzzLib.Orm;
 
@@ -32,8 +31,7 @@ public Events(IQueue queue, IWebhookOperations webhook, ILogTracer log) {
 
         public async Async.Task QueueSignalrEvent(EventMessage eventMessage) {
             var message = new SignalREvent("events", new List<EventMessage>() { eventMessage });
-            var encodedMessage = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(message));
-            await _queue.SendMessage("signalr-events", encodedMessage, StorageType.Config);
+            await _queue.SendMessage("signalr-events", JsonSerializer.Serialize(message), StorageType.Config);
         }
 
         public async Async.Task SendEvent(BaseEvent anEvent) {