lucretia/embeddingsearch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Renamed Worker.Scriptable to ScriptContainer, added cancellation token, Added LoggerWrapper to circumnavigate CLR issues, improved logging, added logging to ScriptToolSet

Browse Source
This commit is contained in:
LD-Reborn
2025-08-31 14:59:41 +02:00
parent 0647f10ca1
commit 14a6acf50f
9 changed files with 167 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
src/Indexer/Calls.cs
View File

@@ -40,7 +40,14 @@ public class RunOnceCall : ICall
public void Stop()
{
Worker.Scriptable.Stop();
if (IsEnabled)
{
Disable();
}
if (IsExecuting)
{
Worker.CancellationTokenSource.Cancel();
}
}
private async void IndexAsync()
@@ -51,7 +58,7 @@ public class RunOnceCall : ICall
IsExecuting = true;
try
{
await Task.Run(() => Worker.Scriptable.Update(new RunOnceCallbackInfos()));
await Task.Run(() => Worker.ScriptContainer.Update(new RunOnceCallbackInfos()));
}
finally
{
@@ -78,7 +85,7 @@ public class RunOnceCall : ICall
public class IntervalCall : ICall
{
public System.Timers.Timer Timer;
public IScriptContainer Scriptable;
public Worker Worker;
public ILogger _logger;
public bool IsEnabled { get; set; }
public bool IsExecuting { get; set; }
@@ -89,7 +96,7 @@ public class IntervalCall : ICall
public IntervalCall(Worker worker, ILogger logger, CallConfig callConfig)
{
Scriptable = worker.Scriptable;
Worker = worker;
_logger = logger;
CallConfig = callConfig;
IsEnabled = true;
@@ -115,7 +122,7 @@ public class IntervalCall : ICall
IsExecuting = true;
try
{
worker.Scriptable.Update(new IntervalCallbackInfos() { sender = sender, e = e });
worker.ScriptContainer.Update(new IntervalCallbackInfos() { sender = sender, e = e });
}
finally
{
@@ -141,7 +148,7 @@ public class IntervalCall : ICall
public void Disable()
{
Scriptable.Stop();
Worker.ScriptContainer.Stop();
Timer.Stop();
IsEnabled = false;
}
@@ -152,24 +159,31 @@ public class IntervalCall : ICall
public void Stop()
{
Scriptable.Stop();
if (IsEnabled)
{
Disable();
}
if (IsExecuting)
{
Worker.CancellationTokenSource.Cancel();
}
}
public HealthCheckResult HealthCheck()
{
if (!Scriptable.UpdateInfo.Successful)
if (!Worker.ScriptContainer.UpdateInfo.Successful)
{
_logger.LogWarning("HealthCheck revealed: The last execution of \"{name}\" was not successful", Scriptable.ToolSet.FilePath);
return HealthCheckResult.Unhealthy($"HealthCheck revealed: The last execution of \"{Scriptable.ToolSet.FilePath}\" was not successful");
_logger.LogWarning("HealthCheck revealed: The last execution of \"{name}\" was not successful", Worker.ScriptContainer.ToolSet.FilePath);
return HealthCheckResult.Unhealthy($"HealthCheck revealed: The last execution of \"{Worker.ScriptContainer.ToolSet.FilePath}\" was not successful");
}
double timerInterval = Timer.Interval; // In ms
DateTime lastRunDateTime = Scriptable.UpdateInfo.DateTime;
DateTime lastRunDateTime = Worker.ScriptContainer.UpdateInfo.DateTime;
DateTime now = DateTime.Now;
double millisecondsSinceLastExecution = now.Subtract(lastRunDateTime).TotalMilliseconds;
if (millisecondsSinceLastExecution >= 2 * timerInterval)
{
_logger.LogWarning("HealthCheck revealed: Since the last execution of \"{name}\" more than twice the interval has passed", Scriptable.ToolSet.FilePath);
return HealthCheckResult.Unhealthy($"HealthCheck revealed: Since the last execution of \"{Scriptable.ToolSet.FilePath}\" more than twice the interval has passed");
_logger.LogWarning("HealthCheck revealed: Since the last execution of \"{name}\" more than twice the interval has passed", Worker.ScriptContainer.ToolSet.FilePath);
return HealthCheckResult.Unhealthy($"HealthCheck revealed: Since the last execution of \"{Worker.ScriptContainer.ToolSet.FilePath}\" more than twice the interval has passed");
}
return HealthCheckResult.Healthy();
}
@@ -211,7 +225,7 @@ public class ScheduleCall : ICall
IsExecuting = true;
try
{
worker.Scriptable.Update(new ScheduleCallbackInfos());
worker.ScriptContainer.Update(new ScheduleCallbackInfos());
}
finally
{
@@ -253,7 +267,14 @@ public class ScheduleCall : ICall
public void Stop()
{
Worker.Scriptable.Stop();
if (IsEnabled)
{
Disable();
}
if (IsExecuting)
{
Worker.CancellationTokenSource.Cancel();
}
}
private async Task CreateJob()
@@ -358,7 +379,14 @@ public class FileUpdateCall : ICall
public void Stop()
{
Worker.Scriptable.Stop();
if (IsEnabled)
{
Disable();
}
if (IsExecuting)
{
Worker.CancellationTokenSource.Cancel();
}
}
private void OnFileChanged(object sender, FileSystemEventArgs e)
@@ -378,7 +406,7 @@ public class FileUpdateCall : ICall
IsExecuting = true;
try
{
Worker.Scriptable.Update(new FileUpdateCallbackInfos() {sender = sender, e = e});
Worker.ScriptContainer.Update(new FileUpdateCallbackInfos() {sender = sender, e = e});
}
finally
{

10
src/Indexer/Controllers/CallsController.cs
View File

@@ -87,7 +87,7 @@ public class CallsController : ControllerBase
}
[HttpGet("Disable")]
public ActionResult<CallDisableResult> Disable(string workerName, string? callName)
public ActionResult<CallDisableResult> Disable(string workerName, string? callName, bool? requestStop = false)
{
_workerCollection.Workers.TryGetValue(workerName, out Worker? worker);
if (worker is null)
@@ -101,6 +101,10 @@ public class CallsController : ControllerBase
foreach (ICall call in worker.Calls)
{
call.Disable();
if (requestStop == true)
{
call.Stop();
}
}
_logger.LogInformation("Stopped calls in worker {name}.", [workerName]);
} else
@@ -114,6 +118,10 @@ public class CallsController : ControllerBase
}
_logger.LogInformation("Starting call {callName} in worker {workerName}.", [callName, workerName]);
call.Disable();
if (requestStop == true)
{
call.Stop();
}
}
return new CallDisableResult { Success = true };
}

6
src/Indexer/Controllers/WorkerController.cs
View File

@@ -68,17 +68,17 @@ public class WorkerController : ControllerBase
}
_logger.LogInformation("triggering worker {name}.", [name]);
ManualTriggerCallbackInfos callbackInfos = new();
lock (worker.Scriptable)
lock (worker.ScriptContainer)
{
worker.IsExecuting = true;
worker.Scriptable.Update(callbackInfos);
worker.ScriptContainer.Update(callbackInfos);
worker.IsExecuting = false;
DateTime beforeExecution = DateTime.Now;
worker.IsExecuting = true;
try
{
worker.Scriptable.Update(callbackInfos);
worker.ScriptContainer.Update(callbackInfos);
}
finally
{

21
src/Indexer/Models/ScriptModels.cs
View File

@@ -13,21 +13,36 @@ public class ScriptToolSet
{
public string FilePath;
public Client.Client Client;
public ILogger Logger;
public LoggerWrapper Logger;
public ICallbackInfos? CallbackInfos;
public IConfiguration Configuration;
public CancellationToken CancellationToken;
public string Name;
public ScriptToolSet(string filePath, Client.Client client, ILogger logger, IConfiguration configuration, string name)
public ScriptToolSet(string filePath, Client.Client client, ILogger<WorkerManager> logger, IConfiguration configuration, CancellationToken cancellationToken, string name)
{
Configuration = configuration;
Name = name;
FilePath = filePath;
Client = client;
Logger = logger;
Logger = new LoggerWrapper(logger);
CancellationToken = cancellationToken;
}
}
public class LoggerWrapper
{
private readonly ILogger _logger;
public LoggerWrapper(ILogger logger) => _logger = logger;
public void LogTrace(string message, params object[]? args) => _logger.LogTrace(message, args);
public void LogDebug(string message, params object[]? args) => _logger.LogDebug(message, args);
public void LogInformation(string message, params object[]? args) => _logger.LogInformation(message, args);
public void LogWarning(string message, params object[]? args) => _logger.LogWarning(message, args);
public void LogError(string message, params object[]? args) => _logger.LogError(message, args);
public void LogCritical(string message, params object[]? args) => _logger.LogCritical(message, args);
}
public interface ICallbackInfos { }
public class RunOnceCallbackInfos : ICallbackInfos {}

4
src/Indexer/ScriptContainers/PythonScriptContainer.cs
View File

@@ -83,11 +83,11 @@ public class PythonScriptable : IScriptContainer
UpdateInfo = new() { DateTime = DateTime.Now, Successful = false, Exception = ex };
if (retryCounter < 3)
{
_logger.LogWarning("Execution of {name} function in script {Toolset.filePath} failed to an exception {ex.Message}", [name, ToolSet.FilePath, ex.Message]);
_logger.LogWarning("Execution of {name} function in script {Toolset.filePath} failed to an exception {ex}", [name, ToolSet.FilePath, ex]);
retryCounter++;
goto retry;
}
_logger.LogError("Execution of {name} function in script {Toolset.filePath} failed to an exception {ex.Message}", [name, ToolSet.FilePath, ex.Message]);
_logger.LogError("Execution of {name} function in script {Toolset.filePath} failed to an exception {ex}", [name, ToolSet.FilePath, ex]);
error = 1;
}
UpdateInfo = new() { DateTime = DateTime.Now, Successful = true };

34
src/Indexer/Scripts/example.py
View File

@@ -18,36 +18,42 @@ probmethod_entity = probmethod
def init(toolset: Toolset):
global example_counter
print("Py-DEBUG@init")
print("This is the init function from the python example script")
print(f"example_counter: {example_counter}")
toolset.Logger.LogInformation("{toolset.Name} - init", toolset.Name)
toolset.Logger.LogInformation("This is the init function from the python example script")
toolset.Logger.LogInformation(f"example_counter: {example_counter}")
searchdomainlist:SearchdomainListResults = toolset.Client.SearchdomainListAsync().Result
if example_searchdomain not in searchdomainlist.Searchdomains:
toolset.Client.SearchdomainCreateAsync(example_searchdomain).Result
searchdomainlist = toolset.Client.SearchdomainListAsync().Result
print("Currently these searchdomains exist:")
output = "Currently these searchdomains exist:\n"
for searchdomain in searchdomainlist.Searchdomains:
print(f" - {searchdomain}")
index_files(toolset)
output += f" - {searchdomain}\n"
toolset.Logger.LogInformation(output)
def update(toolset: Toolset):
global example_counter
print("Py-DEBUG@update")
print("This is the update function from the python example script")
toolset.Logger.LogInformation("{toolset.Name} - update", toolset.Name)
toolset.Logger.LogInformation("This is the update function from the python example script")
callbackInfos:ICallbackInfos = toolset.CallbackInfos
if (str(callbackInfos) == "Indexer.Models.IntervalCallbackInfos"):
print("It was called via an interval callback")
if (str(callbackInfos) == "Indexer.Models.RunOnceCallbackInfos"):
toolset.Logger.LogInformation("It was triggered by a runonce call")
elif (str(callbackInfos) == "Indexer.Models.IntervalCallbackInfos"):
toolset.Logger.LogInformation("It was triggered by an interval call")
elif (str(callbackInfos) == "Indexer.Models.ScheduleCallbackInfos"):
toolset.Logger.LogInformation("It was triggered by a schedule call")
elif (str(callbackInfos) == "Indexer.Models.FileUpdateCallbackInfos"):
toolset.Logger.LogInformation("It was triggered by a fileupdate call")
else:
print("It was called, but the origin of the call could not be determined")
toolset.Logger.LogInformation("It was triggered, but the origin of the call could not be determined")
example_counter += 1
print(f"example_counter: {example_counter}")
toolset.Logger.LogInformation(f"example_counter: {example_counter}")
index_files(toolset)
def index_files(toolset: Toolset):
jsonEntities:list = []
for filename in os.listdir(example_content):
qualified_filepath = example_content + "/" + filename
with open(qualified_filepath, "r", encoding='utf-8') as file:
with open(qualified_filepath, "r", encoding='utf-8', errors="replace") as file:
title = file.readline()
text = file.read()
datapoints:list = [
@@ -61,4 +67,4 @@ def index_files(toolset: Toolset):
timer_start = time.time()
result:EntityIndexResult = toolset.Client.EntityIndexAsync(jsonstring).Result
timer_end = time.time()
print(f"Update was successful: {result.Success} - and was done in {timer_end - timer_start} seconds.")
toolset.Logger.LogInformation(f"Update was successful: {result.Success} - and was done in {timer_end - timer_start} seconds.")

63
src/Indexer/Scripts/tools.py
View File

@@ -1,6 +1,8 @@
from __future__ import annotations
from dataclasses import dataclass
import array
from typing import Optional
from enum import Enum
@dataclass
class JSONDatapoint:
@@ -123,10 +125,63 @@ class IntervalCallbackInfos(ICallbackInfos):
sender: Optional[object]
e: object
@dataclass
class LoggerWrapper:
def LogTrace(message:str, args:list[object]) -> None:
pass
def LogDebug(message:str, args:list[object]) -> None:
pass
def LogInformation(message:str) -> None:
pass
def LogInformation(message:str, args:list[object]) -> None:
pass
def LogWarning(message:str, args:list[object]) -> None:
pass
def LogError(message:str, args:list[object]) -> None:
pass
def LogCritical(message:str, args:list[object]) -> None:
pass
@dataclass
class CancellationTokenRegistration:
Token: CancellationToken
def Dispose() -> None:
pass
def Unregister() -> None:
pass
@dataclass
class WaitHandle:
SafeWaitHandle: object
def Close() -> None:
pass
def Dispose() -> None:
pass
def WaitOne() -> bool:
pass
def WaitOne(timeout:int) -> bool:
pass
@dataclass
class CancellationToken:
CanBeCanceled: bool
IsCancellationRequested: bool
def ThrowIfCancellationRequested() -> None:
pass
def Register(callback: callable[[], any]) -> CancellationTokenRegistration:
pass
WaitHandle: WaitHandle
@dataclass
class Toolset:
filePath:str
client:Client
callbackInfos: Optional[ICallbackInfos] = None
Name:str
FilePath:str
Client:Client
Logger:LoggerWrapper
Configuration: object
CancellationToken: CancellationToken
Name:str
CallbackInfos: Optional[ICallbackInfos] = None

8
src/Indexer/Worker.cs
View File

@@ -5,17 +5,19 @@ public class Worker
{
public string Name { get; set; }
public WorkerConfig Config { get; set; }
public IScriptContainer Scriptable { get; set; }
public IScriptContainer ScriptContainer { get; set; }
public CancellationTokenSource CancellationTokenSource { get; }
public List<ICall> Calls { get; set; }
public bool IsExecuting { get; set; }
public DateTime? LastExecution { get; set; }
public DateTime? LastSuccessfulExecution { get; set; }
public Worker(string name, WorkerConfig workerConfig, IScriptContainer scriptable)
public Worker(string name, WorkerConfig workerConfig, IScriptContainer scriptable, CancellationTokenSource cancellationTokenSource)
{
Name = name;
Config = workerConfig;
Scriptable = scriptable;
ScriptContainer = scriptable;
CancellationTokenSource = cancellationTokenSource;
IsExecuting = false;
Calls = [];
}

11
src/Indexer/WorkerManager.cs
View File

@@ -6,7 +6,7 @@ public class WorkerManager
{
public Dictionary<string, Worker> Workers;
public List<Type> types;
private readonly ILogger _logger;
private readonly ILogger<WorkerManager> _logger;
private readonly IConfiguration _configuration;
private readonly Client.Client client;
@@ -35,8 +35,9 @@ public class WorkerManager
{
foreach (WorkerConfig workerConfig in sectionWorker.Worker)
{
ScriptToolSet toolSet = new(workerConfig.Script, client, _logger, _configuration, workerConfig.Name);
InitializeWorker(toolSet, workerConfig);
CancellationTokenSource cancellationTokenSource = new();
ScriptToolSet toolSet = new(workerConfig.Script, client, _logger, _configuration, cancellationTokenSource.Token, workerConfig.Name);
InitializeWorker(toolSet, workerConfig, cancellationTokenSource);
}
}
else
@@ -47,10 +48,10 @@ public class WorkerManager
_logger.LogInformation("Initialized workers");
}
public void InitializeWorker(ScriptToolSet toolSet, WorkerConfig workerConfig)
public void InitializeWorker(ScriptToolSet toolSet, WorkerConfig workerConfig, CancellationTokenSource cancellationTokenSource)
{
_logger.LogInformation("Initializing worker: {Name}", workerConfig.Name);
Worker worker = new(workerConfig.Name, workerConfig, GetScriptable(toolSet));
Worker worker = new(workerConfig.Name, workerConfig, GetScriptable(toolSet), cancellationTokenSource);
Workers[workerConfig.Name] = worker;
foreach (CallConfig callConfig in workerConfig.Calls)
{