Initial
This commit is contained in:
Zurazan
@@ -0,0 +1,53 @@
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSyncStaticFilesServer.Services;
|
||||
using LightlessSyncStaticFilesServer.Utils;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Controllers;
|
||||
|
||||
[Route(LightlessFiles.Cache)]
|
||||
public class CacheController : ControllerBase
|
||||
{
|
||||
private readonly RequestFileStreamResultFactory _requestFileStreamResultFactory;
|
||||
private readonly CachedFileProvider _cachedFileProvider;
|
||||
private readonly RequestQueueService _requestQueue;
|
||||
private readonly FileStatisticsService _fileStatisticsService;
|
||||
|
||||
public CacheController(ILogger<CacheController> logger, RequestFileStreamResultFactory requestFileStreamResultFactory,
|
||||
CachedFileProvider cachedFileProvider, RequestQueueService requestQueue, FileStatisticsService fileStatisticsService) : base(logger)
|
||||
{
|
||||
_requestFileStreamResultFactory = requestFileStreamResultFactory;
|
||||
_cachedFileProvider = cachedFileProvider;
|
||||
_requestQueue = requestQueue;
|
||||
_fileStatisticsService = fileStatisticsService;
|
||||
}
|
||||
|
||||
[HttpGet(LightlessFiles.Cache_Get)]
|
||||
public async Task<IActionResult> GetFiles(Guid requestId)
|
||||
{
|
||||
_logger.LogDebug($"GetFile:{LightlessUser}:{requestId}");
|
||||
|
||||
if (!_requestQueue.IsActiveProcessing(requestId, LightlessUser, out var request)) return BadRequest();
|
||||
|
||||
_requestQueue.ActivateRequest(requestId);
|
||||
|
||||
Response.ContentType = "application/octet-stream";
|
||||
|
||||
long requestSize = 0;
|
||||
List<BlockFileDataSubstream> substreams = new();
|
||||
|
||||
foreach (var fileHash in request.FileIds)
|
||||
{
|
||||
var fs = await _cachedFileProvider.DownloadAndGetLocalFileInfo(fileHash).ConfigureAwait(false);
|
||||
if (fs == null) continue;
|
||||
|
||||
substreams.Add(new(fs));
|
||||
|
||||
requestSize += fs.Length;
|
||||
}
|
||||
|
||||
_fileStatisticsService.LogRequest(requestSize);
|
||||
|
||||
return _requestFileStreamResultFactory.Create(requestId, new BlockFileDataStream(substreams));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
using LightlessSyncShared.Utils;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Controllers;
|
||||
|
||||
public class ControllerBase : Controller
|
||||
{
|
||||
protected ILogger _logger;
|
||||
|
||||
public ControllerBase(ILogger logger)
|
||||
{
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected string LightlessUser => HttpContext.User.Claims.First(f => string.Equals(f.Type, LightlessClaimTypes.Uid, StringComparison.Ordinal)).Value;
|
||||
protected string Continent => HttpContext.User.Claims.FirstOrDefault(f => string.Equals(f.Type, LightlessClaimTypes.Continent, StringComparison.Ordinal))?.Value ?? "*";
|
||||
protected bool IsPriority => !string.IsNullOrEmpty(HttpContext.User.Claims.FirstOrDefault(f => string.Equals(f.Type, LightlessClaimTypes.Alias, StringComparison.Ordinal))?.Value ?? string.Empty);
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSyncStaticFilesServer.Services;
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Controllers;
|
||||
|
||||
[Route(LightlessFiles.Distribution)]
|
||||
public class DistributionController : ControllerBase
|
||||
{
|
||||
private readonly CachedFileProvider _cachedFileProvider;
|
||||
|
||||
public DistributionController(ILogger<DistributionController> logger, CachedFileProvider cachedFileProvider) : base(logger)
|
||||
{
|
||||
_cachedFileProvider = cachedFileProvider;
|
||||
}
|
||||
|
||||
[HttpGet(LightlessFiles.Distribution_Get)]
|
||||
[Authorize(Policy = "Internal")]
|
||||
public async Task<IActionResult> GetFile(string file)
|
||||
{
|
||||
_logger.LogInformation($"GetFile:{LightlessUser}:{file}");
|
||||
|
||||
var fs = await _cachedFileProvider.DownloadAndGetLocalFileInfo(file);
|
||||
if (fs == null) return NotFound();
|
||||
|
||||
return PhysicalFile(fs.FullName, "application/octet-stream");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using LightlessSyncStaticFilesServer.Services;
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Controllers;
|
||||
|
||||
[Route(LightlessFiles.Main)]
|
||||
[Authorize(Policy = "Internal")]
|
||||
public class MainController : ControllerBase
|
||||
{
|
||||
private readonly IClientReadyMessageService _messageService;
|
||||
private readonly MainServerShardRegistrationService _shardRegistrationService;
|
||||
|
||||
public MainController(ILogger<MainController> logger, IClientReadyMessageService lightlessHub,
|
||||
MainServerShardRegistrationService shardRegistrationService) : base(logger)
|
||||
{
|
||||
_messageService = lightlessHub;
|
||||
_shardRegistrationService = shardRegistrationService;
|
||||
}
|
||||
|
||||
[HttpGet(LightlessFiles.Main_SendReady)]
|
||||
public async Task<IActionResult> SendReadyToClients(string uid, Guid requestId)
|
||||
{
|
||||
await _messageService.SendDownloadReady(uid, requestId).ConfigureAwait(false);
|
||||
return Ok();
|
||||
}
|
||||
|
||||
[HttpPost("shardRegister")]
|
||||
public IActionResult RegisterShard([FromBody] ShardConfiguration shardConfiguration)
|
||||
{
|
||||
try
|
||||
{
|
||||
_shardRegistrationService.RegisterShard(LightlessUser, shardConfiguration);
|
||||
return Ok();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Shard could not be registered {shard}", LightlessUser);
|
||||
return BadRequest();
|
||||
}
|
||||
}
|
||||
|
||||
[HttpPost("shardUnregister")]
|
||||
public IActionResult UnregisterShard()
|
||||
{
|
||||
_shardRegistrationService.UnregisterShard(LightlessUser);
|
||||
return Ok();
|
||||
}
|
||||
|
||||
[HttpPost("shardHeartbeat")]
|
||||
public IActionResult ShardHeartbeat()
|
||||
{
|
||||
try
|
||||
{
|
||||
_shardRegistrationService.ShardHeartbeat(LightlessUser);
|
||||
return Ok();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Shard not registered: {shard}", LightlessUser);
|
||||
return BadRequest();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSyncStaticFilesServer.Services;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Controllers;
|
||||
|
||||
[Route(LightlessFiles.Request)]
|
||||
public class RequestController : ControllerBase
|
||||
{
|
||||
private readonly CachedFileProvider _cachedFileProvider;
|
||||
private readonly RequestQueueService _requestQueue;
|
||||
|
||||
public RequestController(ILogger<RequestController> logger, CachedFileProvider cachedFileProvider, RequestQueueService requestQueue) : base(logger)
|
||||
{
|
||||
_cachedFileProvider = cachedFileProvider;
|
||||
_requestQueue = requestQueue;
|
||||
}
|
||||
|
||||
[HttpGet]
|
||||
[Route(LightlessFiles.Request_Cancel)]
|
||||
public async Task<IActionResult> CancelQueueRequest(Guid requestId)
|
||||
{
|
||||
try
|
||||
{
|
||||
_requestQueue.RemoveFromQueue(requestId, LightlessUser, IsPriority);
|
||||
return Ok();
|
||||
}
|
||||
catch (OperationCanceledException) { return BadRequest(); }
|
||||
}
|
||||
|
||||
[HttpPost]
|
||||
[Route(LightlessFiles.Request_Enqueue)]
|
||||
public async Task<IActionResult> PreRequestFilesAsync([FromBody] IEnumerable<string> files)
|
||||
{
|
||||
try
|
||||
{
|
||||
foreach (var file in files)
|
||||
{
|
||||
_logger.LogDebug("Prerequested file: " + file);
|
||||
await _cachedFileProvider.DownloadFileWhenRequired(file).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
Guid g = Guid.NewGuid();
|
||||
await _requestQueue.EnqueueUser(new(g, LightlessUser, files.ToList()), IsPriority, HttpContext.RequestAborted);
|
||||
|
||||
return Ok(g);
|
||||
}
|
||||
catch (OperationCanceledException) { return BadRequest(); }
|
||||
}
|
||||
|
||||
[HttpGet]
|
||||
[Route(LightlessFiles.Request_Check)]
|
||||
public async Task<IActionResult> CheckQueueAsync(Guid requestId, [FromBody] IEnumerable<string> files)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!_requestQueue.StillEnqueued(requestId, LightlessUser, IsPriority))
|
||||
await _requestQueue.EnqueueUser(new(requestId, LightlessUser, files.ToList()), IsPriority, HttpContext.RequestAborted);
|
||||
return Ok();
|
||||
}
|
||||
catch (OperationCanceledException) { return BadRequest(); }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,363 @@
|
||||
using K4os.Compression.LZ4.Legacy;
|
||||
using LightlessSync.API.Dto.Files;
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSync.API.SignalR;
|
||||
using LightlessSyncServer.Hubs;
|
||||
using LightlessSyncShared.Data;
|
||||
using LightlessSyncShared.Metrics;
|
||||
using LightlessSyncShared.Models;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using LightlessSyncStaticFilesServer.Services;
|
||||
using LightlessSyncStaticFilesServer.Utils;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text.Json;
|
||||
using System.Text.RegularExpressions;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Controllers;
|
||||
|
||||
[Route(LightlessFiles.ServerFiles)]
|
||||
public class ServerFilesController : ControllerBase
|
||||
{
|
||||
private static readonly SemaphoreSlim _fileLockDictLock = new(1);
|
||||
private static readonly ConcurrentDictionary<string, SemaphoreSlim> _fileUploadLocks = new(StringComparer.Ordinal);
|
||||
private readonly string _basePath;
|
||||
private readonly CachedFileProvider _cachedFileProvider;
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configuration;
|
||||
private readonly IHubContext<LightlessHub> _hubContext;
|
||||
private readonly IDbContextFactory<LightlessDbContext> _lightlessDbContext;
|
||||
private readonly LightlessMetrics _metricsClient;
|
||||
private readonly MainServerShardRegistrationService _shardRegistrationService;
|
||||
|
||||
public ServerFilesController(ILogger<ServerFilesController> logger, CachedFileProvider cachedFileProvider,
|
||||
IConfigurationService<StaticFilesServerConfiguration> configuration,
|
||||
IHubContext<LightlessHub> hubContext,
|
||||
IDbContextFactory<LightlessDbContext> lightlessDbContext, LightlessMetrics metricsClient,
|
||||
MainServerShardRegistrationService shardRegistrationService) : base(logger)
|
||||
{
|
||||
_basePath = configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UseColdStorage), false)
|
||||
? configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.ColdStorageDirectory))
|
||||
: configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
||||
_cachedFileProvider = cachedFileProvider;
|
||||
_configuration = configuration;
|
||||
_hubContext = hubContext;
|
||||
_lightlessDbContext = lightlessDbContext;
|
||||
_metricsClient = metricsClient;
|
||||
_shardRegistrationService = shardRegistrationService;
|
||||
}
|
||||
|
||||
[HttpPost(LightlessFiles.ServerFiles_DeleteAll)]
|
||||
public async Task<IActionResult> FilesDeleteAll()
|
||||
{
|
||||
using var dbContext = await _lightlessDbContext.CreateDbContextAsync();
|
||||
var ownFiles = await dbContext.Files.Where(f => f.Uploaded && f.Uploader.UID == LightlessUser).ToListAsync().ConfigureAwait(false);
|
||||
bool isColdStorage = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UseColdStorage), false);
|
||||
|
||||
foreach (var dbFile in ownFiles)
|
||||
{
|
||||
var fi = FilePathUtil.GetFileInfoForHash(_basePath, dbFile.Hash);
|
||||
if (fi != null)
|
||||
{
|
||||
_metricsClient.DecGauge(isColdStorage ? MetricsAPI.GaugeFilesTotalColdStorage : MetricsAPI.GaugeFilesTotal, fi == null ? 0 : 1);
|
||||
_metricsClient.DecGauge(isColdStorage ? MetricsAPI.GaugeFilesTotalSizeColdStorage : MetricsAPI.GaugeFilesTotalSize, fi?.Length ?? 0);
|
||||
|
||||
fi?.Delete();
|
||||
}
|
||||
}
|
||||
|
||||
dbContext.Files.RemoveRange(ownFiles);
|
||||
await dbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
|
||||
return Ok();
|
||||
}
|
||||
|
||||
[HttpGet(LightlessFiles.ServerFiles_GetSizes)]
|
||||
public async Task<IActionResult> FilesGetSizes([FromBody] List<string> hashes)
|
||||
{
|
||||
using var dbContext = await _lightlessDbContext.CreateDbContextAsync();
|
||||
var forbiddenFiles = await dbContext.ForbiddenUploadEntries.
|
||||
Where(f => hashes.Contains(f.Hash)).ToListAsync().ConfigureAwait(false);
|
||||
List<DownloadFileDto> response = new();
|
||||
|
||||
var cacheFile = await dbContext.Files.AsNoTracking()
|
||||
.Where(f => hashes.Contains(f.Hash))
|
||||
.Select(k => new { k.Hash, k.Size, k.RawSize })
|
||||
.ToListAsync().ConfigureAwait(false);
|
||||
|
||||
var allFileShards = _shardRegistrationService.GetConfigurationsByContinent(Continent);
|
||||
|
||||
foreach (var file in cacheFile)
|
||||
{
|
||||
var forbiddenFile = forbiddenFiles.SingleOrDefault(f => string.Equals(f.Hash, file.Hash, StringComparison.OrdinalIgnoreCase));
|
||||
Uri? baseUrl = null;
|
||||
|
||||
if (forbiddenFile == null)
|
||||
{
|
||||
var matchingShards = allFileShards.Where(f => new Regex(f.FileMatch).IsMatch(file.Hash)).ToList();
|
||||
|
||||
var shard = matchingShards.SelectMany(g => g.RegionUris)
|
||||
.OrderBy(g => Guid.NewGuid()).FirstOrDefault();
|
||||
|
||||
baseUrl = shard.Value ?? _configuration.GetValue<Uri>(nameof(StaticFilesServerConfiguration.CdnFullUrl));
|
||||
}
|
||||
|
||||
response.Add(new DownloadFileDto
|
||||
{
|
||||
FileExists = file.Size > 0,
|
||||
ForbiddenBy = forbiddenFile?.ForbiddenBy ?? string.Empty,
|
||||
IsForbidden = forbiddenFile != null,
|
||||
Hash = file.Hash,
|
||||
Size = file.Size,
|
||||
Url = baseUrl?.ToString() ?? string.Empty,
|
||||
RawSize = file.RawSize
|
||||
});
|
||||
}
|
||||
|
||||
return Ok(JsonSerializer.Serialize(response));
|
||||
}
|
||||
|
||||
[HttpGet(LightlessFiles.ServerFiles_DownloadServers)]
|
||||
public async Task<IActionResult> GetDownloadServers()
|
||||
{
|
||||
var allFileShards = _shardRegistrationService.GetConfigurationsByContinent(Continent);
|
||||
return Ok(JsonSerializer.Serialize(allFileShards.SelectMany(t => t.RegionUris.Select(v => v.Value.ToString()))));
|
||||
}
|
||||
|
||||
[HttpPost(LightlessFiles.ServerFiles_FilesSend)]
|
||||
public async Task<IActionResult> FilesSend([FromBody] FilesSendDto filesSendDto)
|
||||
{
|
||||
using var dbContext = await _lightlessDbContext.CreateDbContextAsync();
|
||||
|
||||
var userSentHashes = new HashSet<string>(filesSendDto.FileHashes.Distinct(StringComparer.Ordinal).Select(s => string.Concat(s.Where(c => char.IsLetterOrDigit(c)))), StringComparer.Ordinal);
|
||||
var notCoveredFiles = new Dictionary<string, UploadFileDto>(StringComparer.Ordinal);
|
||||
var forbiddenFiles = await dbContext.ForbiddenUploadEntries.AsNoTracking().Where(f => userSentHashes.Contains(f.Hash)).AsNoTracking().ToDictionaryAsync(f => f.Hash, f => f).ConfigureAwait(false);
|
||||
var existingFiles = await dbContext.Files.AsNoTracking().Where(f => userSentHashes.Contains(f.Hash)).AsNoTracking().ToDictionaryAsync(f => f.Hash, f => f).ConfigureAwait(false);
|
||||
|
||||
List<FileCache> fileCachesToUpload = new();
|
||||
foreach (var hash in userSentHashes)
|
||||
{
|
||||
// Skip empty file hashes, duplicate file hashes, forbidden file hashes and existing file hashes
|
||||
if (string.IsNullOrEmpty(hash)) { continue; }
|
||||
if (notCoveredFiles.ContainsKey(hash)) { continue; }
|
||||
if (forbiddenFiles.ContainsKey(hash))
|
||||
{
|
||||
notCoveredFiles[hash] = new UploadFileDto()
|
||||
{
|
||||
ForbiddenBy = forbiddenFiles[hash].ForbiddenBy,
|
||||
Hash = hash,
|
||||
IsForbidden = true,
|
||||
};
|
||||
|
||||
continue;
|
||||
}
|
||||
if (existingFiles.TryGetValue(hash, out var file) && file.Uploaded) { continue; }
|
||||
|
||||
notCoveredFiles[hash] = new UploadFileDto()
|
||||
{
|
||||
Hash = hash,
|
||||
};
|
||||
}
|
||||
|
||||
if (notCoveredFiles.Any(p => !p.Value.IsForbidden))
|
||||
{
|
||||
await _hubContext.Clients.Users(filesSendDto.UIDs).SendAsync(nameof(ILightlessHub.Client_UserReceiveUploadStatus), new LightlessSync.API.Dto.User.UserDto(new(LightlessUser)))
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return Ok(JsonSerializer.Serialize(notCoveredFiles.Values.ToList()));
|
||||
}
|
||||
|
||||
[HttpPost(LightlessFiles.ServerFiles_Upload + "/{hash}")]
|
||||
[RequestSizeLimit(200 * 1024 * 1024)]
|
||||
public async Task<IActionResult> UploadFile(string hash, CancellationToken requestAborted)
|
||||
{
|
||||
using var dbContext = await _lightlessDbContext.CreateDbContextAsync();
|
||||
|
||||
_logger.LogInformation("{user}|{file}: Uploading", LightlessUser, hash);
|
||||
|
||||
hash = hash.ToUpperInvariant();
|
||||
var existingFile = await dbContext.Files.SingleOrDefaultAsync(f => f.Hash == hash);
|
||||
if (existingFile != null) return Ok();
|
||||
|
||||
SemaphoreSlim fileLock = await CreateFileLock(hash, requestAborted).ConfigureAwait(false);
|
||||
|
||||
try
|
||||
{
|
||||
var existingFileCheck2 = await dbContext.Files.SingleOrDefaultAsync(f => f.Hash == hash);
|
||||
if (existingFileCheck2 != null)
|
||||
{
|
||||
return Ok();
|
||||
}
|
||||
|
||||
// copy the request body to memory
|
||||
using var memoryStream = new MemoryStream();
|
||||
await Request.Body.CopyToAsync(memoryStream, requestAborted).ConfigureAwait(false);
|
||||
|
||||
_logger.LogDebug("{user}|{file}: Finished uploading", LightlessUser, hash);
|
||||
|
||||
await StoreData(hash, dbContext, memoryStream).ConfigureAwait(false);
|
||||
|
||||
return Ok();
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError(e, "{user}|{file}: Error during file upload", LightlessUser, hash);
|
||||
return BadRequest();
|
||||
}
|
||||
finally
|
||||
{
|
||||
try
|
||||
{
|
||||
fileLock.Release();
|
||||
fileLock.Dispose();
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// it's disposed whatever
|
||||
}
|
||||
finally
|
||||
{
|
||||
_fileUploadLocks.TryRemove(hash, out _);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
[HttpPost(LightlessFiles.ServerFiles_UploadMunged + "/{hash}")]
|
||||
[RequestSizeLimit(200 * 1024 * 1024)]
|
||||
public async Task<IActionResult> UploadFileMunged(string hash, CancellationToken requestAborted)
|
||||
{
|
||||
using var dbContext = await _lightlessDbContext.CreateDbContextAsync();
|
||||
|
||||
_logger.LogInformation("{user}|{file}: Uploading munged", LightlessUser, hash);
|
||||
hash = hash.ToUpperInvariant();
|
||||
var existingFile = await dbContext.Files.SingleOrDefaultAsync(f => f.Hash == hash);
|
||||
if (existingFile != null) return Ok();
|
||||
|
||||
SemaphoreSlim fileLock = await CreateFileLock(hash, requestAborted).ConfigureAwait(false);
|
||||
|
||||
try
|
||||
{
|
||||
var existingFileCheck2 = await dbContext.Files.SingleOrDefaultAsync(f => f.Hash == hash);
|
||||
if (existingFileCheck2 != null)
|
||||
{
|
||||
return Ok();
|
||||
}
|
||||
|
||||
// copy the request body to memory
|
||||
using var compressedMungedStream = new MemoryStream();
|
||||
await Request.Body.CopyToAsync(compressedMungedStream, requestAborted).ConfigureAwait(false);
|
||||
var unmungedFile = compressedMungedStream.ToArray();
|
||||
MungeBuffer(unmungedFile.AsSpan());
|
||||
await using MemoryStream unmungedMs = new(unmungedFile);
|
||||
|
||||
_logger.LogDebug("{user}|{file}: Finished uploading, unmunged stream", LightlessUser, hash);
|
||||
|
||||
await StoreData(hash, dbContext, unmungedMs);
|
||||
|
||||
return Ok();
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError(e, "{user}|{file}: Error during file upload", LightlessUser, hash);
|
||||
return BadRequest();
|
||||
}
|
||||
finally
|
||||
{
|
||||
try
|
||||
{
|
||||
fileLock.Release();
|
||||
fileLock.Dispose();
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// it's disposed whatever
|
||||
}
|
||||
finally
|
||||
{
|
||||
_fileUploadLocks.TryRemove(hash, out _);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task StoreData(string hash, LightlessDbContext dbContext, MemoryStream compressedFileStream)
|
||||
{
|
||||
var decompressedData = LZ4Wrapper.Unwrap(compressedFileStream.ToArray());
|
||||
// reset streams
|
||||
compressedFileStream.Seek(0, SeekOrigin.Begin);
|
||||
|
||||
// compute hash to verify
|
||||
var hashString = BitConverter.ToString(SHA1.HashData(decompressedData))
|
||||
.Replace("-", "", StringComparison.Ordinal).ToUpperInvariant();
|
||||
if (!string.Equals(hashString, hash, StringComparison.Ordinal))
|
||||
throw new InvalidOperationException($"{LightlessUser}|{hash}: Hash does not match file, computed: {hashString}, expected: {hash}");
|
||||
|
||||
// save file
|
||||
var path = FilePathUtil.GetFilePath(_basePath, hash);
|
||||
using var fileStream = new FileStream(path, FileMode.Create);
|
||||
await compressedFileStream.CopyToAsync(fileStream).ConfigureAwait(false);
|
||||
_logger.LogDebug("{user}|{file}: Uploaded file saved to {path}", LightlessUser, hash, path);
|
||||
|
||||
// update on db
|
||||
await dbContext.Files.AddAsync(new FileCache()
|
||||
{
|
||||
Hash = hash,
|
||||
UploadDate = DateTime.UtcNow,
|
||||
UploaderUID = LightlessUser,
|
||||
Size = compressedFileStream.Length,
|
||||
Uploaded = true,
|
||||
RawSize = decompressedData.LongLength
|
||||
}).ConfigureAwait(false);
|
||||
|
||||
await dbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
|
||||
_logger.LogDebug("{user}|{file}: Uploaded file saved to DB", LightlessUser, hash);
|
||||
|
||||
bool isColdStorage = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UseColdStorage), false);
|
||||
|
||||
_metricsClient.IncGauge(isColdStorage ? MetricsAPI.GaugeFilesTotalColdStorage : MetricsAPI.GaugeFilesTotal, 1);
|
||||
_metricsClient.IncGauge(isColdStorage ? MetricsAPI.GaugeFilesTotalSizeColdStorage : MetricsAPI.GaugeFilesTotalSize, compressedFileStream.Length);
|
||||
}
|
||||
|
||||
|
||||
private async Task<SemaphoreSlim> CreateFileLock(string hash, CancellationToken requestAborted)
|
||||
{
|
||||
SemaphoreSlim? fileLock = null;
|
||||
bool successfullyWaited = false;
|
||||
while (!successfullyWaited && !requestAborted.IsCancellationRequested)
|
||||
{
|
||||
lock (_fileUploadLocks)
|
||||
{
|
||||
if (!_fileUploadLocks.TryGetValue(hash, out fileLock))
|
||||
{
|
||||
_logger.LogDebug("{user}|{file}: Creating filelock", LightlessUser, hash);
|
||||
_fileUploadLocks[hash] = fileLock = new SemaphoreSlim(1);
|
||||
}
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_logger.LogDebug("{user}|{file}: Waiting for filelock", LightlessUser, hash);
|
||||
await fileLock.WaitAsync(requestAborted).ConfigureAwait(false);
|
||||
successfullyWaited = true;
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
_logger.LogWarning("{user}|{file}: Semaphore disposed, recreating", LightlessUser, hash);
|
||||
}
|
||||
}
|
||||
|
||||
return fileLock;
|
||||
}
|
||||
|
||||
private static void MungeBuffer(Span<byte> buffer)
|
||||
{
|
||||
for (int i = 0; i < buffer.Length; ++i)
|
||||
{
|
||||
buffer[i] ^= 42;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
using Microsoft.Extensions.Caching.Memory;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Controllers;
|
||||
|
||||
[Route(LightlessFiles.Speedtest)]
|
||||
public class SpeedTestController : ControllerBase
|
||||
{
|
||||
private readonly IMemoryCache _memoryCache;
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configurationService;
|
||||
private const string RandomByteDataName = "SpeedTestRandomByteData";
|
||||
private static readonly SemaphoreSlim _speedtestSemaphore = new(10, 10);
|
||||
|
||||
public SpeedTestController(ILogger<SpeedTestController> logger, IMemoryCache memoryCache,
|
||||
IConfigurationService<StaticFilesServerConfiguration> configurationService) : base(logger)
|
||||
{
|
||||
_memoryCache = memoryCache;
|
||||
_configurationService = configurationService;
|
||||
}
|
||||
|
||||
[HttpGet(LightlessFiles.Speedtest_Run)]
|
||||
public async Task<IActionResult> DownloadTest(CancellationToken cancellationToken)
|
||||
{
|
||||
var user = HttpContext.User.Claims.First(f => string.Equals(f.Type, LightlessClaimTypes.Uid, StringComparison.Ordinal)).Value;
|
||||
var speedtestLimit = _configurationService.GetValueOrDefault(nameof(StaticFilesServerConfiguration.SpeedTestHoursRateLimit), 0.5);
|
||||
if (_memoryCache.TryGetValue<DateTime>(user, out var value))
|
||||
{
|
||||
var hoursRemaining = value.Subtract(DateTime.UtcNow).TotalHours;
|
||||
return StatusCode(429, $"Can perform speedtest every {speedtestLimit} hours. {hoursRemaining:F2} hours remain.");
|
||||
}
|
||||
|
||||
await _speedtestSemaphore.WaitAsync(cancellationToken);
|
||||
|
||||
try
|
||||
{
|
||||
var expiry = DateTime.UtcNow.Add(TimeSpan.FromHours(speedtestLimit));
|
||||
_memoryCache.Set(user, expiry, TimeSpan.FromHours(speedtestLimit));
|
||||
|
||||
var randomByteData = _memoryCache.GetOrCreate(RandomByteDataName, (entry) =>
|
||||
{
|
||||
byte[] data = new byte[100 * 1024 * 1024];
|
||||
new Random().NextBytes(data);
|
||||
return data;
|
||||
});
|
||||
|
||||
return File(randomByteData, "application/octet-stream", "speedtest.dat");
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
return StatusCode(499, "Cancelled");
|
||||
}
|
||||
finally
|
||||
{
|
||||
_speedtestSemaphore.Release();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
|
||||
// this is a very hacky way to attach this file server to the main lightless hub signalr instance via redis
|
||||
// signalr publishes the namespace and hubname into the redis backend so this needs to be equal to the original
|
||||
// but I don't need to reimplement the hub completely as I only exclusively use it for internal connection calling
|
||||
// from the queue service so I keep the namespace and name of the class the same so it can connect to the same channel
|
||||
// if anyone finds a better way to do this let me know
|
||||
|
||||
#pragma warning disable IDE0130 // Namespace does not match folder structure
|
||||
#pragma warning disable MA0048 // File name must match type name
|
||||
namespace LightlessSyncServer.Hubs;
|
||||
public class LightlessHub : Hub
|
||||
{
|
||||
public override Task OnConnectedAsync()
|
||||
{
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
|
||||
public override Task OnDisconnectedAsync(Exception exception)
|
||||
{
|
||||
throw new NotSupportedException();
|
||||
}
|
||||
}
|
||||
#pragma warning restore IDE0130 // Namespace does not match folder structure
|
||||
#pragma warning restore MA0048 // File name must match type name
|
||||
@@ -0,0 +1,38 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk.Web">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net9.0</TargetFramework>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Content Remove="appsettings.Development.json" />
|
||||
<Content Remove="appsettings.json" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<None Include="appsettings.Development.json" />
|
||||
<None Include="appsettings.json">
|
||||
<CopyToOutputDirectory>Never</CopyToOutputDirectory>
|
||||
</None>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="IDisposableAnalyzers" Version="4.0.8">
|
||||
<PrivateAssets>all</PrivateAssets>
|
||||
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
|
||||
</PackageReference>
|
||||
<PackageReference Include="K4os.Compression.LZ4.Legacy" Version="1.3.8" />
|
||||
<PackageReference Include="Meziantou.Analyzer" Version="2.0.184">
|
||||
<PrivateAssets>all</PrivateAssets>
|
||||
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
|
||||
</PackageReference>
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting.Systemd" Version="9.0.0" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\LightlessAPI\LightlessSyncAPI\LightlessSync.API.csproj" />
|
||||
<ProjectReference Include="..\LightlessSyncShared\LightlessSyncShared.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
@@ -0,0 +1,63 @@
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer;
|
||||
|
||||
public class Program
|
||||
{
|
||||
public static void Main(string[] args)
|
||||
{
|
||||
var hostBuilder = CreateHostBuilder(args);
|
||||
var host = hostBuilder.Build();
|
||||
|
||||
using (var scope = host.Services.CreateScope())
|
||||
{
|
||||
var options = host.Services.GetService<IConfigurationService<StaticFilesServerConfiguration>>();
|
||||
var optionsServer = host.Services.GetService<IConfigurationService<LightlessConfigurationBase>>();
|
||||
var logger = host.Services.GetService<ILogger<Program>>();
|
||||
logger.LogInformation("Loaded LightlessSync Static Files Server Configuration (IsMain: {isMain})", options.IsMain);
|
||||
logger.LogInformation(options.ToString());
|
||||
logger.LogInformation("Loaded LightlessSync Server Auth Configuration (IsMain: {isMain})", optionsServer.IsMain);
|
||||
logger.LogInformation(optionsServer.ToString());
|
||||
}
|
||||
|
||||
host.Run();
|
||||
}
|
||||
|
||||
public static IHostBuilder CreateHostBuilder(string[] args)
|
||||
{
|
||||
var loggerFactory = LoggerFactory.Create(builder =>
|
||||
{
|
||||
builder.ClearProviders();
|
||||
builder.AddConsole();
|
||||
});
|
||||
var logger = loggerFactory.CreateLogger<Startup>();
|
||||
return Host.CreateDefaultBuilder(args)
|
||||
.UseSystemd()
|
||||
.UseConsoleLifetime()
|
||||
.ConfigureAppConfiguration((ctx, config) =>
|
||||
{
|
||||
var appSettingsPath = Environment.GetEnvironmentVariable("APPSETTINGS_PATH");
|
||||
if (!string.IsNullOrEmpty(appSettingsPath))
|
||||
{
|
||||
config.AddJsonFile(appSettingsPath, optional: true, reloadOnChange: true);
|
||||
}
|
||||
else
|
||||
{
|
||||
config.AddJsonFile("appsettings.json", optional: true, reloadOnChange: true);
|
||||
}
|
||||
|
||||
config.AddEnvironmentVariables();
|
||||
})
|
||||
.ConfigureLogging((ctx, builder) =>
|
||||
{
|
||||
builder.AddConfiguration(ctx.Configuration.GetSection("Logging"));
|
||||
builder.AddFile(o => o.RootPath = AppContext.BaseDirectory);
|
||||
})
|
||||
.ConfigureWebHostDefaults(webBuilder =>
|
||||
{
|
||||
webBuilder.UseContentRoot(AppContext.BaseDirectory);
|
||||
webBuilder.UseStartup(ctx => new Startup(ctx.Configuration, logger));
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
{
|
||||
"iisSettings": {
|
||||
"windowsAuthentication": false,
|
||||
"anonymousAuthentication": true,
|
||||
"iisExpress": {
|
||||
"applicationUrl": "http://localhost:21378",
|
||||
"sslPort": 44331
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"LightlessSyncStaticFilesServer": {
|
||||
"commandName": "Project",
|
||||
"dotnetRunMessages": true,
|
||||
"launchBrowser": true,
|
||||
"applicationUrl": "https://localhost:7094;http://localhost:5094",
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_ENVIRONMENT": "Development"
|
||||
}
|
||||
},
|
||||
"IIS Express": {
|
||||
"commandName": "IISExpress",
|
||||
"launchBrowser": true,
|
||||
"environmentVariables": {
|
||||
"ASPNETCORE_ENVIRONMENT": "Development"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,214 @@
|
||||
using LightlessSyncShared.Metrics;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncStaticFilesServer.Utils;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Net.Http.Headers;
|
||||
using LightlessSyncShared.Utils;
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public sealed class CachedFileProvider : IDisposable
|
||||
{
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configuration;
|
||||
private readonly ILogger<CachedFileProvider> _logger;
|
||||
private readonly FileStatisticsService _fileStatisticsService;
|
||||
private readonly LightlessMetrics _metrics;
|
||||
private readonly ServerTokenGenerator _generator;
|
||||
private readonly Uri _remoteCacheSourceUri;
|
||||
private readonly string _hotStoragePath;
|
||||
private readonly ConcurrentDictionary<string, Task> _currentTransfers = new(StringComparer.Ordinal);
|
||||
private readonly HttpClient _httpClient;
|
||||
private readonly SemaphoreSlim _downloadSemaphore = new(1, 1);
|
||||
private bool _disposed;
|
||||
|
||||
private bool IsMainServer => _remoteCacheSourceUri == null && _isDistributionServer;
|
||||
private bool _isDistributionServer;
|
||||
|
||||
public CachedFileProvider(IConfigurationService<StaticFilesServerConfiguration> configuration, ILogger<CachedFileProvider> logger,
|
||||
FileStatisticsService fileStatisticsService, LightlessMetrics metrics, ServerTokenGenerator generator)
|
||||
{
|
||||
_configuration = configuration;
|
||||
_logger = logger;
|
||||
_fileStatisticsService = fileStatisticsService;
|
||||
_metrics = metrics;
|
||||
_generator = generator;
|
||||
_remoteCacheSourceUri = configuration.GetValueOrDefault<Uri>(nameof(StaticFilesServerConfiguration.DistributionFileServerAddress), null);
|
||||
_isDistributionServer = configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.IsDistributionNode), false);
|
||||
_hotStoragePath = configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
||||
_httpClient = new();
|
||||
_httpClient.DefaultRequestHeaders.UserAgent.Add(new ProductInfoHeaderValue("LightlessSyncServer", "1.0.0.0"));
|
||||
_httpClient.Timeout = TimeSpan.FromSeconds(300);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_disposed = true;
|
||||
_httpClient?.Dispose();
|
||||
}
|
||||
|
||||
private async Task DownloadTask(string hash)
|
||||
{
|
||||
var destinationFilePath = FilePathUtil.GetFilePath(_hotStoragePath, hash);
|
||||
|
||||
// first check cold storage
|
||||
if (TryCopyFromColdStorage(hash, destinationFilePath)) return;
|
||||
|
||||
// if cold storage is not configured or file not found or error is present try to download file from remote
|
||||
var downloadUrl = LightlessFiles.DistributionGetFullPath(_remoteCacheSourceUri, hash);
|
||||
_logger.LogInformation("Did not find {hash}, downloading from {server}", hash, downloadUrl);
|
||||
|
||||
using var requestMessage = new HttpRequestMessage(HttpMethod.Get, downloadUrl);
|
||||
requestMessage.Headers.Authorization = new AuthenticationHeaderValue("Bearer", _generator.Token);
|
||||
HttpResponseMessage? response = null;
|
||||
|
||||
try
|
||||
{
|
||||
response = await _httpClient.SendAsync(requestMessage).ConfigureAwait(false);
|
||||
response.EnsureSuccessStatusCode();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to download {url}", downloadUrl);
|
||||
response?.Dispose();
|
||||
return;
|
||||
}
|
||||
|
||||
var tempFileName = destinationFilePath + ".dl";
|
||||
var fileStream = new FileStream(tempFileName, FileMode.Create, FileAccess.ReadWrite);
|
||||
var bufferSize = response.Content.Headers.ContentLength > 1024 * 1024 ? 4096 : 1024;
|
||||
var buffer = new byte[bufferSize];
|
||||
|
||||
var bytesRead = 0;
|
||||
using var content = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
|
||||
while ((bytesRead = await content.ReadAsync(buffer).ConfigureAwait(false)) > 0)
|
||||
{
|
||||
await fileStream.WriteAsync(buffer.AsMemory(0, bytesRead)).ConfigureAwait(false);
|
||||
}
|
||||
await fileStream.FlushAsync().ConfigureAwait(false);
|
||||
await fileStream.DisposeAsync().ConfigureAwait(false);
|
||||
File.Move(tempFileName, destinationFilePath, true);
|
||||
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesTotalSize, FilePathUtil.GetFileInfoForHash(_hotStoragePath, hash).Length);
|
||||
response.Dispose();
|
||||
}
|
||||
|
||||
private bool TryCopyFromColdStorage(string hash, string destinationFilePath)
|
||||
{
|
||||
if (!_configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UseColdStorage), false)) return false;
|
||||
|
||||
string coldStorageDir = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.ColdStorageDirectory), string.Empty);
|
||||
if (string.IsNullOrEmpty(coldStorageDir)) return false;
|
||||
|
||||
var coldStorageFilePath = FilePathUtil.GetFileInfoForHash(coldStorageDir, hash);
|
||||
if (coldStorageFilePath == null) return false;
|
||||
|
||||
try
|
||||
{
|
||||
_logger.LogDebug("Copying {hash} from cold storage: {path}", hash, coldStorageFilePath);
|
||||
var tempFileName = destinationFilePath + ".dl";
|
||||
File.Copy(coldStorageFilePath.FullName, tempFileName, true);
|
||||
File.Move(tempFileName, destinationFilePath, true);
|
||||
coldStorageFilePath.LastAccessTimeUtc = DateTime.UtcNow;
|
||||
var destinationFile = new FileInfo(destinationFilePath);
|
||||
destinationFile.LastAccessTimeUtc = DateTime.UtcNow;
|
||||
destinationFile.CreationTimeUtc = DateTime.UtcNow;
|
||||
destinationFile.LastWriteTimeUtc = DateTime.UtcNow;
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesTotalSize, new FileInfo(destinationFilePath).Length);
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Could not copy {coldStoragePath} from cold storage", coldStorageFilePath);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
public async Task DownloadFileWhenRequired(string hash)
|
||||
{
|
||||
var fi = FilePathUtil.GetFileInfoForHash(_hotStoragePath, hash);
|
||||
if (fi == null && IsMainServer)
|
||||
{
|
||||
TryCopyFromColdStorage(hash, FilePathUtil.GetFilePath(_hotStoragePath, hash));
|
||||
return;
|
||||
}
|
||||
|
||||
await _downloadSemaphore.WaitAsync().ConfigureAwait(false);
|
||||
if ((fi == null || (fi?.Length ?? 0) == 0)
|
||||
&& (!_currentTransfers.TryGetValue(hash, out var downloadTask)
|
||||
|| (downloadTask?.IsCompleted ?? true)))
|
||||
{
|
||||
_currentTransfers[hash] = Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesDownloadingFromCache);
|
||||
await DownloadTask(hash).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during Download Task for {hash}", hash);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesDownloadingFromCache);
|
||||
_currentTransfers.Remove(hash, out _);
|
||||
}
|
||||
});
|
||||
}
|
||||
_downloadSemaphore.Release();
|
||||
}
|
||||
|
||||
public FileInfo? GetLocalFilePath(string hash)
|
||||
{
|
||||
var fi = FilePathUtil.GetFileInfoForHash(_hotStoragePath, hash);
|
||||
if (fi == null) return null;
|
||||
|
||||
fi.LastAccessTimeUtc = DateTime.UtcNow;
|
||||
|
||||
_fileStatisticsService.LogFile(hash, fi.Length);
|
||||
|
||||
return new FileInfo(fi.FullName);
|
||||
}
|
||||
|
||||
public async Task<FileInfo?> DownloadAndGetLocalFileInfo(string hash)
|
||||
{
|
||||
await DownloadFileWhenRequired(hash).ConfigureAwait(false);
|
||||
|
||||
if (_currentTransfers.TryGetValue(hash, out var downloadTask))
|
||||
{
|
||||
try
|
||||
{
|
||||
using CancellationTokenSource cts = new();
|
||||
cts.CancelAfter(TimeSpan.FromSeconds(300));
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesTasksWaitingForDownloadFromCache);
|
||||
await downloadTask.WaitAsync(cts.Token).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogWarning(e, "Failed while waiting for download task for {hash}", hash);
|
||||
return null;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTasksWaitingForDownloadFromCache);
|
||||
}
|
||||
}
|
||||
|
||||
return GetLocalFilePath(hash);
|
||||
}
|
||||
|
||||
public bool AnyFilesDownloading(List<string> hashes)
|
||||
{
|
||||
return hashes.Exists(_currentTransfers.Keys.Contains);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
using LightlessSyncShared.Metrics;
|
||||
using System.Collections.Concurrent;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public class FileStatisticsService : IHostedService
|
||||
{
|
||||
private readonly LightlessMetrics _metrics;
|
||||
private readonly ILogger<FileStatisticsService> _logger;
|
||||
private CancellationTokenSource _resetCancellationTokenSource;
|
||||
private ConcurrentDictionary<string, long> _pastHourFiles = new(StringComparer.Ordinal);
|
||||
private ConcurrentDictionary<string, long> _pastDayFiles = new(StringComparer.Ordinal);
|
||||
|
||||
public FileStatisticsService(LightlessMetrics metrics, ILogger<FileStatisticsService> logger)
|
||||
{
|
||||
_metrics = metrics;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public void LogFile(string fileHash, long length)
|
||||
{
|
||||
if (!_pastHourFiles.ContainsKey(fileHash))
|
||||
{
|
||||
_pastHourFiles[fileHash] = length;
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesUniquePastHour);
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesUniquePastHourSize, length);
|
||||
}
|
||||
if (!_pastDayFiles.ContainsKey(fileHash))
|
||||
{
|
||||
_pastDayFiles[fileHash] = length;
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesUniquePastDay);
|
||||
_metrics.IncGauge(MetricsAPI.GaugeFilesUniquePastDaySize, length);
|
||||
}
|
||||
}
|
||||
|
||||
public void LogRequest(long requestSize)
|
||||
{
|
||||
_metrics.IncCounter(MetricsAPI.CounterFileRequests, 1);
|
||||
_metrics.IncCounter(MetricsAPI.CounterFileRequestSize, requestSize);
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Starting FileStatisticsService");
|
||||
_resetCancellationTokenSource = new();
|
||||
_ = ResetHourlyFileData();
|
||||
_ = ResetDailyFileData();
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task ResetHourlyFileData()
|
||||
{
|
||||
while (!_resetCancellationTokenSource.Token.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogInformation("Resetting 1h Data");
|
||||
|
||||
_pastHourFiles = new(StringComparer.Ordinal);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesUniquePastHour, 0);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesUniquePastHourSize, 0);
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
TimeOnly currentTime = new(now.Hour, now.Minute, now.Second);
|
||||
TimeOnly futureTime = new(now.Hour, 0, 0);
|
||||
var span = futureTime.AddHours(1) - currentTime;
|
||||
|
||||
await Task.Delay(span, _resetCancellationTokenSource.Token).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task ResetDailyFileData()
|
||||
{
|
||||
while (!_resetCancellationTokenSource.Token.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogInformation("Resetting 24h Data");
|
||||
|
||||
_pastDayFiles = new(StringComparer.Ordinal);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesUniquePastDay, 0);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesUniquePastDaySize, 0);
|
||||
|
||||
var now = DateTime.UtcNow;
|
||||
DateTime midnight = new(new DateOnly(now.Date.Year, now.Date.Month, now.Date.Day), new(0, 0, 0));
|
||||
var span = midnight.AddDays(1) - now;
|
||||
|
||||
await Task.Delay(span, _resetCancellationTokenSource.Token).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_resetCancellationTokenSource.Cancel();
|
||||
_logger.LogInformation("Stopping FileStatisticsService");
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public interface IClientReadyMessageService
|
||||
{
|
||||
Task SendDownloadReady(string uid, Guid requestId);
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using LightlessSync.API.SignalR;
|
||||
using LightlessSyncServer.Hubs;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public class MainClientReadyMessageService : IClientReadyMessageService
|
||||
{
|
||||
private readonly ILogger<MainClientReadyMessageService> _logger;
|
||||
private readonly IHubContext<LightlessHub> _lightlessHub;
|
||||
|
||||
public MainClientReadyMessageService(ILogger<MainClientReadyMessageService> logger, IHubContext<LightlessHub> lightlessHub)
|
||||
{
|
||||
_logger = logger;
|
||||
_lightlessHub = lightlessHub;
|
||||
}
|
||||
|
||||
public async Task SendDownloadReady(string uid, Guid requestId)
|
||||
{
|
||||
_logger.LogInformation("Sending Client Ready for {uid}:{requestId} to SignalR", uid, requestId);
|
||||
await _lightlessHub.Clients.User(uid).SendAsync(nameof(ILightlessHub.Client_DownloadReady), requestId).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,367 @@
|
||||
using ByteSizeLib;
|
||||
using K4os.Compression.LZ4.Legacy;
|
||||
using LightlessSyncShared.Data;
|
||||
using LightlessSyncShared.Metrics;
|
||||
using LightlessSyncShared.Models;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using LightlessSyncStaticFilesServer.Utils;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public class MainFileCleanupService : IHostedService
|
||||
{
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configuration;
|
||||
private readonly IDbContextFactory<LightlessDbContext> _dbContextFactory;
|
||||
private readonly ILogger<MainFileCleanupService> _logger;
|
||||
private readonly LightlessMetrics _metrics;
|
||||
private CancellationTokenSource _cleanupCts;
|
||||
|
||||
public MainFileCleanupService(LightlessMetrics metrics, ILogger<MainFileCleanupService> logger,
|
||||
IConfigurationService<StaticFilesServerConfiguration> configuration,
|
||||
IDbContextFactory<LightlessDbContext> dbContextFactory)
|
||||
{
|
||||
_metrics = metrics;
|
||||
_logger = logger;
|
||||
_configuration = configuration;
|
||||
_dbContextFactory = dbContextFactory;
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Cleanup Service started");
|
||||
|
||||
_cleanupCts = new();
|
||||
|
||||
_ = Task.Run(() => CleanUpTask(_cleanupCts.Token)).ConfigureAwait(false);
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_cleanupCts.Cancel();
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private List<FileInfo> CleanUpFilesBeyondSizeLimit(List<FileInfo> files, double sizeLimit, bool deleteFromDb, LightlessDbContext dbContext, CancellationToken ct)
|
||||
{
|
||||
if (sizeLimit <= 0)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_logger.LogInformation("Cleaning up files beyond the cache size limit of {cacheSizeLimit} GiB", sizeLimit);
|
||||
var allLocalFiles = files
|
||||
.OrderBy(f => f.LastAccessTimeUtc).ToList();
|
||||
var totalCacheSizeInBytes = allLocalFiles.Sum(s => s.Length);
|
||||
long cacheSizeLimitInBytes = (long)ByteSize.FromGibiBytes(sizeLimit).Bytes;
|
||||
while (totalCacheSizeInBytes > cacheSizeLimitInBytes && allLocalFiles.Count != 0 && !ct.IsCancellationRequested)
|
||||
{
|
||||
var oldestFile = allLocalFiles[0];
|
||||
allLocalFiles.RemoveAt(0);
|
||||
totalCacheSizeInBytes -= oldestFile.Length;
|
||||
_logger.LogInformation("Deleting {oldestFile} with size {size}MiB", oldestFile.FullName, ByteSize.FromBytes(oldestFile.Length).MebiBytes);
|
||||
oldestFile.Delete();
|
||||
FileCache f = new() { Hash = oldestFile.Name.ToUpperInvariant() };
|
||||
if (deleteFromDb)
|
||||
dbContext.Entry(f).State = EntityState.Deleted;
|
||||
}
|
||||
|
||||
return allLocalFiles;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during cache size limit cleanup");
|
||||
}
|
||||
|
||||
return [];
|
||||
}
|
||||
|
||||
private List<FileInfo> CleanUpOrphanedFiles(List<FileCache> allFiles, List<FileInfo> allPhysicalFiles, CancellationToken ct)
|
||||
{
|
||||
var allFilesHashes = new HashSet<string>(allFiles.Select(a => a.Hash.ToUpperInvariant()), StringComparer.Ordinal);
|
||||
foreach (var file in allPhysicalFiles.ToList())
|
||||
{
|
||||
if (!allFilesHashes.Contains(file.Name.ToUpperInvariant()))
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, file.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
file.Delete();
|
||||
_logger.LogInformation("File not in DB, deleting: {fileName}", file.Name);
|
||||
allPhysicalFiles.Remove(file);
|
||||
}
|
||||
|
||||
ct.ThrowIfCancellationRequested();
|
||||
}
|
||||
|
||||
return allPhysicalFiles;
|
||||
}
|
||||
|
||||
private async Task<List<FileInfo>> CleanUpOutdatedFiles(string dir, List<FileInfo> allFilesInDir, int unusedRetention, int forcedDeletionAfterHours,
|
||||
bool deleteFromDb, LightlessDbContext dbContext, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
_logger.LogInformation("Cleaning up files older than {filesOlderThanDays} days", unusedRetention);
|
||||
if (forcedDeletionAfterHours > 0)
|
||||
{
|
||||
_logger.LogInformation("Cleaning up files written to longer than {hours}h ago", forcedDeletionAfterHours);
|
||||
}
|
||||
|
||||
// clean up files in DB but not on disk or last access is expired
|
||||
var prevTime = DateTime.Now.Subtract(TimeSpan.FromDays(unusedRetention));
|
||||
var prevTimeForcedDeletion = DateTime.Now.Subtract(TimeSpan.FromHours(forcedDeletionAfterHours));
|
||||
List<FileCache> allDbFiles = await dbContext.Files.ToListAsync(ct).ConfigureAwait(false);
|
||||
List<string> removedFileHashes;
|
||||
|
||||
if (!deleteFromDb)
|
||||
{
|
||||
removedFileHashes = CleanupViaFiles(allFilesInDir, forcedDeletionAfterHours, prevTime, prevTimeForcedDeletion, ct);
|
||||
}
|
||||
else
|
||||
{
|
||||
removedFileHashes = await CleanupViaDb(dir, forcedDeletionAfterHours, dbContext, prevTime, prevTimeForcedDeletion, allDbFiles, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// clean up files that are on disk but not in DB anymore
|
||||
return CleanUpOrphanedFiles(allDbFiles, allFilesInDir.Where(c => !removedFileHashes.Contains(c.Name, StringComparer.OrdinalIgnoreCase)).ToList(), ct);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during file cleanup of old files");
|
||||
}
|
||||
|
||||
return [];
|
||||
}
|
||||
|
||||
private void CleanUpStuckUploads(LightlessDbContext dbContext)
|
||||
{
|
||||
var pastTime = DateTime.UtcNow.Subtract(TimeSpan.FromMinutes(20));
|
||||
var stuckUploads = dbContext.Files.Where(f => !f.Uploaded && f.UploadDate < pastTime);
|
||||
dbContext.Files.RemoveRange(stuckUploads);
|
||||
}
|
||||
|
||||
private async Task CleanUpTask(CancellationToken ct)
|
||||
{
|
||||
InitializeGauges();
|
||||
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
var cleanupCheckMinutes = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.CleanupCheckInMinutes), 15);
|
||||
bool useColdStorage = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UseColdStorage), false);
|
||||
var hotStorageDir = _configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
||||
var coldStorageDir = _configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.ColdStorageDirectory));
|
||||
using var dbContext = await _dbContextFactory.CreateDbContextAsync().ConfigureAwait(false);
|
||||
|
||||
_logger.LogInformation("Running File Cleanup Task");
|
||||
|
||||
try
|
||||
{
|
||||
using CancellationTokenSource timedCts = new();
|
||||
timedCts.CancelAfter(TimeSpan.FromMinutes(cleanupCheckMinutes - 1));
|
||||
using var linkedTokenCts = CancellationTokenSource.CreateLinkedTokenSource(timedCts.Token, ct);
|
||||
var linkedToken = linkedTokenCts.Token;
|
||||
|
||||
DirectoryInfo dirHotStorage = new(hotStorageDir);
|
||||
_logger.LogInformation("File Cleanup Task gathering hot storage files");
|
||||
var allFilesInHotStorage = dirHotStorage.GetFiles("*", SearchOption.AllDirectories).ToList();
|
||||
|
||||
var unusedRetention = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UnusedFileRetentionPeriodInDays), 14);
|
||||
var forcedDeletionAfterHours = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.ForcedDeletionOfFilesAfterHours), -1);
|
||||
var sizeLimit = _configuration.GetValueOrDefault<double>(nameof(StaticFilesServerConfiguration.CacheSizeHardLimitInGiB), -1);
|
||||
|
||||
_logger.LogInformation("File Cleanup Task cleaning up outdated hot storage files");
|
||||
var remainingHotFiles = await CleanUpOutdatedFiles(hotStorageDir, allFilesInHotStorage, unusedRetention, forcedDeletionAfterHours,
|
||||
deleteFromDb: !useColdStorage, dbContext: dbContext,
|
||||
ct: linkedToken).ConfigureAwait(false);
|
||||
|
||||
_logger.LogInformation("File Cleanup Task cleaning up hot storage file beyond size limit");
|
||||
var finalRemainingHotFiles = CleanUpFilesBeyondSizeLimit(remainingHotFiles, sizeLimit,
|
||||
deleteFromDb: !useColdStorage, dbContext: dbContext,
|
||||
ct: linkedToken);
|
||||
|
||||
if (useColdStorage)
|
||||
{
|
||||
DirectoryInfo dirColdStorage = new(coldStorageDir);
|
||||
_logger.LogInformation("File Cleanup Task gathering cold storage files");
|
||||
var allFilesInColdStorageDir = dirColdStorage.GetFiles("*", SearchOption.AllDirectories).ToList();
|
||||
|
||||
var coldStorageRetention = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.ColdStorageUnusedFileRetentionPeriodInDays), 60);
|
||||
var coldStorageSize = _configuration.GetValueOrDefault<double>(nameof(StaticFilesServerConfiguration.ColdStorageSizeHardLimitInGiB), -1);
|
||||
|
||||
// clean up cold storage
|
||||
_logger.LogInformation("File Cleanup Task cleaning up outdated cold storage files");
|
||||
var remainingColdFiles = await CleanUpOutdatedFiles(coldStorageDir, allFilesInColdStorageDir, coldStorageRetention, forcedDeletionAfterHours: -1,
|
||||
deleteFromDb: true, dbContext: dbContext,
|
||||
ct: linkedToken).ConfigureAwait(false);
|
||||
_logger.LogInformation("File Cleanup Task cleaning up cold storage file beyond size limit");
|
||||
var finalRemainingColdFiles = CleanUpFilesBeyondSizeLimit(remainingColdFiles, coldStorageSize,
|
||||
deleteFromDb: true, dbContext: dbContext,
|
||||
ct: linkedToken);
|
||||
}
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError(e, "Error during cleanup task");
|
||||
}
|
||||
finally
|
||||
{
|
||||
CleanUpStuckUploads(dbContext);
|
||||
|
||||
await dbContext.SaveChangesAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
if (useColdStorage)
|
||||
{
|
||||
DirectoryInfo dirColdStorageAfterCleanup = new(coldStorageDir);
|
||||
var allFilesInColdStorageAfterCleanup = dirColdStorageAfterCleanup.GetFiles("*", SearchOption.AllDirectories).ToList();
|
||||
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotalSizeColdStorage, allFilesInColdStorageAfterCleanup.Sum(f => { try { return f.Length; } catch { return 0; } }));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotalColdStorage, allFilesInColdStorageAfterCleanup.Count);
|
||||
}
|
||||
|
||||
DirectoryInfo dirHotStorageAfterCleanup = new(hotStorageDir);
|
||||
var allFilesInHotStorageAfterCleanup = dirHotStorageAfterCleanup.GetFiles("*", SearchOption.AllDirectories).ToList();
|
||||
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotalSize, allFilesInHotStorageAfterCleanup.Sum(f => { try { return f.Length; } catch { return 0; } }));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotal, allFilesInHotStorageAfterCleanup.Count);
|
||||
|
||||
var now = DateTime.Now;
|
||||
TimeOnly currentTime = new(now.Hour, now.Minute, now.Second);
|
||||
TimeOnly futureTime = new(now.Hour, now.Minute - now.Minute % cleanupCheckMinutes, 0);
|
||||
var span = futureTime.AddMinutes(cleanupCheckMinutes) - currentTime;
|
||||
|
||||
_logger.LogInformation("File Cleanup Complete, next run at {date}", now.Add(span));
|
||||
await Task.Delay(span, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
private async Task<List<string>> CleanupViaDb(string dir, int forcedDeletionAfterHours,
|
||||
LightlessDbContext dbContext, DateTime lastAccessCutoffTime, DateTime forcedDeletionCutoffTime, List<FileCache> allDbFiles, CancellationToken ct)
|
||||
{
|
||||
int fileCounter = 0;
|
||||
List<string> removedFileHashes = new();
|
||||
foreach (var fileCache in allDbFiles.Where(f => f.Uploaded))
|
||||
{
|
||||
bool deleteCurrentFile = false;
|
||||
var file = FilePathUtil.GetFileInfoForHash(dir, fileCache.Hash);
|
||||
if (file == null)
|
||||
{
|
||||
_logger.LogInformation("File does not exist anymore: {fileName}", fileCache.Hash);
|
||||
deleteCurrentFile = true;
|
||||
}
|
||||
else if (file != null && file.LastAccessTime < lastAccessCutoffTime)
|
||||
{
|
||||
_logger.LogInformation("File outdated: {fileName}, {fileSize}MiB", file.Name, ByteSize.FromBytes(file.Length).MebiBytes);
|
||||
deleteCurrentFile = true;
|
||||
}
|
||||
else if (file != null && forcedDeletionAfterHours > 0 && file.LastWriteTime < forcedDeletionCutoffTime)
|
||||
{
|
||||
_logger.LogInformation("File forcefully deleted: {fileName}, {fileSize}MiB", file.Name, ByteSize.FromBytes(file.Length).MebiBytes);
|
||||
deleteCurrentFile = true;
|
||||
}
|
||||
|
||||
// only used if file in db has no raw size for whatever reason
|
||||
if (!deleteCurrentFile && file != null && fileCache.RawSize == 0)
|
||||
{
|
||||
try
|
||||
{
|
||||
var length = LZ4Wrapper.Unwrap(File.ReadAllBytes(file.FullName)).LongLength;
|
||||
_logger.LogInformation("Setting Raw File Size of " + fileCache.Hash + " to " + length);
|
||||
fileCache.RawSize = length;
|
||||
if (fileCounter % 1000 == 0)
|
||||
await dbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Could not unpack {fileName}", file.FullName);
|
||||
}
|
||||
}
|
||||
|
||||
// do actual deletion of file and remove also from db if needed
|
||||
if (deleteCurrentFile)
|
||||
{
|
||||
if (file != null) file.Delete();
|
||||
|
||||
removedFileHashes.Add(fileCache.Hash);
|
||||
|
||||
dbContext.Files.Remove(fileCache);
|
||||
}
|
||||
|
||||
// only used if file in db has no size for whatever reason
|
||||
if (!deleteCurrentFile && file != null && fileCache.Size == 0)
|
||||
{
|
||||
_logger.LogInformation("Setting File Size of " + fileCache.Hash + " to " + file.Length);
|
||||
fileCache.Size = file.Length;
|
||||
// commit every 1000 files to db
|
||||
if (fileCounter % 1000 == 0)
|
||||
await dbContext.SaveChangesAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
fileCounter++;
|
||||
|
||||
ct.ThrowIfCancellationRequested();
|
||||
}
|
||||
|
||||
return removedFileHashes;
|
||||
}
|
||||
|
||||
private List<string> CleanupViaFiles(List<FileInfo> allFilesInDir, int forcedDeletionAfterHours,
|
||||
DateTime lastAccessCutoffTime, DateTime forcedDeletionCutoffTime, CancellationToken ct)
|
||||
{
|
||||
List<string> removedFileHashes = new List<string>();
|
||||
|
||||
foreach (var file in allFilesInDir)
|
||||
{
|
||||
bool deleteCurrentFile = false;
|
||||
if (file != null && file.LastAccessTime < lastAccessCutoffTime)
|
||||
{
|
||||
_logger.LogInformation("File outdated: {fileName}, {fileSize}MiB", file.Name, ByteSize.FromBytes(file.Length).MebiBytes);
|
||||
deleteCurrentFile = true;
|
||||
}
|
||||
else if (file != null && forcedDeletionAfterHours > 0 && file.LastWriteTime < forcedDeletionCutoffTime)
|
||||
{
|
||||
_logger.LogInformation("File forcefully deleted: {fileName}, {fileSize}MiB", file.Name, ByteSize.FromBytes(file.Length).MebiBytes);
|
||||
deleteCurrentFile = true;
|
||||
}
|
||||
|
||||
if (deleteCurrentFile)
|
||||
{
|
||||
if (file != null) file.Delete();
|
||||
|
||||
removedFileHashes.Add(file.Name);
|
||||
}
|
||||
|
||||
ct.ThrowIfCancellationRequested();
|
||||
}
|
||||
|
||||
return removedFileHashes;
|
||||
}
|
||||
|
||||
private void InitializeGauges()
|
||||
{
|
||||
bool useColdStorage = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UseColdStorage), false);
|
||||
|
||||
if (useColdStorage)
|
||||
{
|
||||
var coldStorageDir = _configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.ColdStorageDirectory));
|
||||
|
||||
DirectoryInfo dirColdStorage = new(coldStorageDir);
|
||||
var allFilesInColdStorageDir = dirColdStorage.GetFiles("*", SearchOption.AllDirectories).ToList();
|
||||
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotalSizeColdStorage, allFilesInColdStorageDir.Sum(f => f.Length));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotalColdStorage, allFilesInColdStorageDir.Count);
|
||||
}
|
||||
|
||||
var hotStorageDir = _configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
||||
DirectoryInfo dirHotStorage = new(hotStorageDir);
|
||||
var allFilesInHotStorage = dirHotStorage.GetFiles("*", SearchOption.AllDirectories).ToList();
|
||||
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotalSize, allFilesInHotStorage.Sum(f => { try { return f.Length; } catch { return 0; } }));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotal, allFilesInHotStorage.Count);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Frozen;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public class MainServerShardRegistrationService : IHostedService
|
||||
{
|
||||
private readonly ILogger<MainServerShardRegistrationService> _logger;
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configurationService;
|
||||
private readonly ConcurrentDictionary<string, ShardConfiguration> _shardConfigs = new(StringComparer.Ordinal);
|
||||
private readonly ConcurrentDictionary<string, DateTime> _shardHeartbeats = new(StringComparer.Ordinal);
|
||||
private readonly CancellationTokenSource _periodicCheckCts = new();
|
||||
|
||||
public MainServerShardRegistrationService(ILogger<MainServerShardRegistrationService> logger,
|
||||
IConfigurationService<StaticFilesServerConfiguration> configurationService)
|
||||
{
|
||||
_logger = logger;
|
||||
_configurationService = configurationService;
|
||||
}
|
||||
|
||||
public void RegisterShard(string shardName, ShardConfiguration shardConfiguration)
|
||||
{
|
||||
if (shardConfiguration == null || shardConfiguration == default)
|
||||
throw new InvalidOperationException("Empty configuration provided");
|
||||
|
||||
if (_shardConfigs.ContainsKey(shardName))
|
||||
_logger.LogInformation("Re-Registering Shard {name}", shardName);
|
||||
else
|
||||
_logger.LogInformation("Registering Shard {name}", shardName);
|
||||
|
||||
_shardHeartbeats[shardName] = DateTime.UtcNow;
|
||||
_shardConfigs[shardName] = shardConfiguration;
|
||||
}
|
||||
|
||||
public void UnregisterShard(string shardName)
|
||||
{
|
||||
_logger.LogInformation("Unregistering Shard {name}", shardName);
|
||||
|
||||
_shardHeartbeats.TryRemove(shardName, out _);
|
||||
_shardConfigs.TryRemove(shardName, out _);
|
||||
}
|
||||
|
||||
public List<ShardConfiguration> GetConfigurationsByContinent(string continent)
|
||||
{
|
||||
var shardConfigs = _shardConfigs.Values.Where(v => v.Continents.Contains(continent, StringComparer.OrdinalIgnoreCase)).ToList();
|
||||
if (shardConfigs.Any()) return shardConfigs;
|
||||
shardConfigs = _shardConfigs.Values.Where(v => v.Continents.Contains("*", StringComparer.OrdinalIgnoreCase)).ToList();
|
||||
if (shardConfigs.Any()) return shardConfigs;
|
||||
return [new ShardConfiguration() {
|
||||
Continents = ["*"],
|
||||
FileMatch = ".*",
|
||||
RegionUris = new(StringComparer.Ordinal) {
|
||||
{ "Central", _configurationService.GetValue<Uri>(nameof(StaticFilesServerConfiguration.CdnFullUrl)) }
|
||||
} }];
|
||||
}
|
||||
|
||||
public void ShardHeartbeat(string shardName)
|
||||
{
|
||||
if (!_shardConfigs.ContainsKey(shardName))
|
||||
throw new InvalidOperationException("Shard not registered");
|
||||
|
||||
_logger.LogInformation("Heartbeat from {name}", shardName);
|
||||
_shardHeartbeats[shardName] = DateTime.UtcNow;
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_ = Task.Run(() => PeriodicHeartbeatCleanup(_periodicCheckCts.Token), cancellationToken).ConfigureAwait(false);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await _periodicCheckCts.CancelAsync().ConfigureAwait(false);
|
||||
_periodicCheckCts.Dispose();
|
||||
}
|
||||
|
||||
private async Task PeriodicHeartbeatCleanup(CancellationToken ct)
|
||||
{
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
foreach (var kvp in _shardHeartbeats.ToFrozenDictionary())
|
||||
{
|
||||
if (DateTime.UtcNow.Subtract(kvp.Value) > TimeSpan.FromMinutes(1))
|
||||
{
|
||||
_shardHeartbeats.TryRemove(kvp.Key, out _);
|
||||
_shardConfigs.TryRemove(kvp.Key, out _);
|
||||
}
|
||||
}
|
||||
|
||||
await Task.Delay(5000, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,228 @@
|
||||
using LightlessSyncShared.Metrics;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using LightlessSyncStaticFilesServer.Utils;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Timers;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public class RequestQueueService : IHostedService
|
||||
{
|
||||
private readonly IClientReadyMessageService _clientReadyMessageService;
|
||||
private readonly CachedFileProvider _cachedFileProvider;
|
||||
private readonly ILogger<RequestQueueService> _logger;
|
||||
private readonly LightlessMetrics _metrics;
|
||||
private readonly ConcurrentQueue<UserRequest> _queue = new();
|
||||
private readonly ConcurrentQueue<UserRequest> _priorityQueue = new();
|
||||
private readonly int _queueExpirationSeconds;
|
||||
private readonly SemaphoreSlim _queueProcessingSemaphore = new(1);
|
||||
private readonly UserQueueEntry[] _userQueueRequests;
|
||||
private int _queueLimitForReset;
|
||||
private readonly int _queueReleaseSeconds;
|
||||
private System.Timers.Timer _queueTimer;
|
||||
|
||||
public RequestQueueService(LightlessMetrics metrics, IConfigurationService<StaticFilesServerConfiguration> configurationService,
|
||||
ILogger<RequestQueueService> logger, IClientReadyMessageService hubContext, CachedFileProvider cachedFileProvider)
|
||||
{
|
||||
_userQueueRequests = new UserQueueEntry[configurationService.GetValueOrDefault(nameof(StaticFilesServerConfiguration.DownloadQueueSize), 50)];
|
||||
_queueExpirationSeconds = configurationService.GetValueOrDefault(nameof(StaticFilesServerConfiguration.DownloadTimeoutSeconds), 5);
|
||||
_queueLimitForReset = configurationService.GetValueOrDefault(nameof(StaticFilesServerConfiguration.DownloadQueueClearLimit), 15000);
|
||||
_queueReleaseSeconds = configurationService.GetValueOrDefault(nameof(StaticFilesServerConfiguration.DownloadQueueReleaseSeconds), 15);
|
||||
_metrics = metrics;
|
||||
_logger = logger;
|
||||
_clientReadyMessageService = hubContext;
|
||||
_cachedFileProvider = cachedFileProvider;
|
||||
}
|
||||
|
||||
public void ActivateRequest(Guid request)
|
||||
{
|
||||
_logger.LogDebug("Activating request {guid}", request);
|
||||
var req = _userQueueRequests.First(f => f != null && f.UserRequest.RequestId == request);
|
||||
req.MarkActive();
|
||||
}
|
||||
|
||||
public async Task EnqueueUser(UserRequest request, bool isPriority, CancellationToken token)
|
||||
{
|
||||
while (_queueProcessingSemaphore.CurrentCount == 0)
|
||||
{
|
||||
await Task.Delay(50, token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_logger.LogDebug("Enqueueing req {guid} from {user} for {file}", request.RequestId, request.User, string.Join(", ", request.FileIds));
|
||||
|
||||
GetQueue(isPriority).Enqueue(request);
|
||||
}
|
||||
|
||||
public void FinishRequest(Guid request)
|
||||
{
|
||||
var req = _userQueueRequests.FirstOrDefault(f => f != null && f.UserRequest.RequestId == request);
|
||||
if (req != null)
|
||||
{
|
||||
var idx = Array.IndexOf(_userQueueRequests, req);
|
||||
_logger.LogDebug("Finishing Request {guid}, clearing slot {idx}", request, idx);
|
||||
_userQueueRequests[idx] = null;
|
||||
}
|
||||
else
|
||||
{
|
||||
_logger.LogDebug("Request {guid} already cleared", request);
|
||||
}
|
||||
}
|
||||
|
||||
public bool IsActiveProcessing(Guid request, string user, out UserRequest userRequest)
|
||||
{
|
||||
var userQueueRequest = _userQueueRequests.FirstOrDefault(u => u != null && u.UserRequest.RequestId == request && string.Equals(u.UserRequest.User, user, StringComparison.Ordinal));
|
||||
userRequest = userQueueRequest?.UserRequest ?? null;
|
||||
return userQueueRequest != null && userRequest != null && userQueueRequest.ExpirationDate > DateTime.UtcNow;
|
||||
}
|
||||
|
||||
public void RemoveFromQueue(Guid requestId, string user, bool isPriority)
|
||||
{
|
||||
var existingRequest = GetQueue(isPriority).FirstOrDefault(f => f.RequestId == requestId && string.Equals(f.User, user, StringComparison.Ordinal));
|
||||
if (existingRequest == null)
|
||||
{
|
||||
var activeSlot = _userQueueRequests.FirstOrDefault(r => r != null && string.Equals(r.UserRequest.User, user, StringComparison.Ordinal) && r.UserRequest.RequestId == requestId);
|
||||
if (activeSlot != null)
|
||||
{
|
||||
var idx = Array.IndexOf(_userQueueRequests, activeSlot);
|
||||
if (idx >= 0)
|
||||
{
|
||||
_userQueueRequests[idx] = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
existingRequest.IsCancelled = true;
|
||||
}
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_queueTimer = new System.Timers.Timer(500);
|
||||
_queueTimer.Elapsed += ProcessQueue;
|
||||
_queueTimer.AutoReset = true;
|
||||
_queueTimer.Start();
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private ConcurrentQueue<UserRequest> GetQueue(bool isPriority) => isPriority ? _priorityQueue : _queue;
|
||||
|
||||
public bool StillEnqueued(Guid request, string user, bool isPriority)
|
||||
{
|
||||
return GetQueue(isPriority).Any(c => c.RequestId == request && string.Equals(c.User, user, StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_queueTimer.Stop();
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private void DequeueIntoSlot(UserRequest userRequest, int slot)
|
||||
{
|
||||
_logger.LogDebug("Dequeueing {req} into {i}: {user} with {file}", userRequest.RequestId, slot, userRequest.User, string.Join(", ", userRequest.FileIds));
|
||||
_userQueueRequests[slot] = new(userRequest, DateTime.UtcNow.AddSeconds(_queueExpirationSeconds));
|
||||
_clientReadyMessageService.SendDownloadReady(userRequest.User, userRequest.RequestId);
|
||||
}
|
||||
|
||||
private void ProcessQueue(object src, ElapsedEventArgs e)
|
||||
{
|
||||
_logger.LogDebug("Periodic Processing Queue Start");
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeQueueFree, _userQueueRequests.Count(c => c == null));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeQueueActive, _userQueueRequests.Count(c => c != null && c.IsActive));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeQueueInactive, _userQueueRequests.Count(c => c != null && !c.IsActive));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeDownloadQueue, _queue.Count(q => !q.IsCancelled));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeDownloadQueueCancelled, _queue.Count(q => q.IsCancelled));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeDownloadPriorityQueue, _priorityQueue.Count(q => !q.IsCancelled));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeDownloadPriorityQueueCancelled, _priorityQueue.Count(q => q.IsCancelled));
|
||||
|
||||
if (_queueProcessingSemaphore.CurrentCount == 0)
|
||||
{
|
||||
_logger.LogDebug("Aborting Periodic Processing Queue, processing still running");
|
||||
return;
|
||||
}
|
||||
|
||||
_queueProcessingSemaphore.Wait();
|
||||
|
||||
try
|
||||
{
|
||||
if (_queue.Count(c => !c.IsCancelled) > _queueLimitForReset)
|
||||
{
|
||||
_queue.Clear();
|
||||
return;
|
||||
}
|
||||
|
||||
for (int i = 0; i < _userQueueRequests.Length; i++)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (_userQueueRequests[i] != null
|
||||
&& (((!_userQueueRequests[i].IsActive && _userQueueRequests[i].ExpirationDate < DateTime.UtcNow))
|
||||
|| (_userQueueRequests[i].IsActive && _userQueueRequests[i].ActivationDate < DateTime.UtcNow.Subtract(TimeSpan.FromSeconds(_queueReleaseSeconds))))
|
||||
)
|
||||
{
|
||||
_logger.LogDebug("Expiring request {guid} slot {slot}", _userQueueRequests[i].UserRequest.RequestId, i);
|
||||
_userQueueRequests[i] = null;
|
||||
}
|
||||
|
||||
if (_userQueueRequests[i] != null) continue;
|
||||
|
||||
while (true)
|
||||
{
|
||||
if (!_priorityQueue.All(u => _cachedFileProvider.AnyFilesDownloading(u.FileIds))
|
||||
&& _priorityQueue.TryDequeue(out var prioRequest))
|
||||
{
|
||||
if (prioRequest.IsCancelled)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
if (_cachedFileProvider.AnyFilesDownloading(prioRequest.FileIds))
|
||||
{
|
||||
_priorityQueue.Enqueue(prioRequest);
|
||||
continue;
|
||||
}
|
||||
|
||||
DequeueIntoSlot(prioRequest, i);
|
||||
break;
|
||||
}
|
||||
|
||||
if (!_queue.All(u => _cachedFileProvider.AnyFilesDownloading(u.FileIds))
|
||||
&& _queue.TryDequeue(out var request))
|
||||
{
|
||||
if (request.IsCancelled)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
if (_cachedFileProvider.AnyFilesDownloading(request.FileIds))
|
||||
{
|
||||
_queue.Enqueue(request);
|
||||
continue;
|
||||
}
|
||||
|
||||
DequeueIntoSlot(request, i);
|
||||
break;
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during inside queue processing");
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error during Queue processing");
|
||||
}
|
||||
finally
|
||||
{
|
||||
_queueProcessingSemaphore.Release();
|
||||
_logger.LogDebug("Periodic Processing Queue End");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using System.Net.Http.Headers;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public class ShardClientReadyMessageService : IClientReadyMessageService
|
||||
{
|
||||
private readonly ILogger<ShardClientReadyMessageService> _logger;
|
||||
private readonly ServerTokenGenerator _tokenGenerator;
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configurationService;
|
||||
private readonly HttpClient _httpClient;
|
||||
|
||||
public ShardClientReadyMessageService(ILogger<ShardClientReadyMessageService> logger, ServerTokenGenerator tokenGenerator, IConfigurationService<StaticFilesServerConfiguration> configurationService)
|
||||
{
|
||||
_logger = logger;
|
||||
_tokenGenerator = tokenGenerator;
|
||||
_configurationService = configurationService;
|
||||
_httpClient = new();
|
||||
_httpClient.DefaultRequestHeaders.UserAgent.Add(new ProductInfoHeaderValue("LightlessSyncServer", "1.0.0.0"));
|
||||
}
|
||||
|
||||
public async Task SendDownloadReady(string uid, Guid requestId)
|
||||
{
|
||||
var mainUrl = _configurationService.GetValue<Uri>(nameof(StaticFilesServerConfiguration.MainFileServerAddress));
|
||||
var path = LightlessFiles.MainSendReadyFullPath(mainUrl, uid, requestId);
|
||||
using HttpRequestMessage msg = new()
|
||||
{
|
||||
RequestUri = path
|
||||
};
|
||||
msg.Headers.Authorization = new AuthenticationHeaderValue("Bearer", _tokenGenerator.Token);
|
||||
|
||||
_logger.LogInformation("Sending Client Ready for {uid}:{requestId} to {path}", uid, requestId, path);
|
||||
try
|
||||
{
|
||||
using var result = await _httpClient.SendAsync(msg).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Failure to send for {uid}:{requestId}", uid, requestId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
using ByteSizeLib;
|
||||
using LightlessSyncShared.Metrics;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public class ShardFileCleanupService : IHostedService
|
||||
{
|
||||
private readonly string _cacheDir;
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configuration;
|
||||
private readonly ILogger<MainFileCleanupService> _logger;
|
||||
private readonly LightlessMetrics _metrics;
|
||||
private CancellationTokenSource _cleanupCts;
|
||||
|
||||
public ShardFileCleanupService(LightlessMetrics metrics, ILogger<MainFileCleanupService> logger, IConfigurationService<StaticFilesServerConfiguration> configuration)
|
||||
{
|
||||
_metrics = metrics;
|
||||
_logger = logger;
|
||||
_configuration = configuration;
|
||||
_cacheDir = _configuration.GetValue<string>(nameof(StaticFilesServerConfiguration.CacheDirectory));
|
||||
}
|
||||
|
||||
public async Task CleanUpTask(CancellationToken ct)
|
||||
{
|
||||
_logger.LogInformation("Starting periodic cleanup task");
|
||||
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
DirectoryInfo dir = new(_cacheDir);
|
||||
var allFiles = dir.GetFiles("*", SearchOption.AllDirectories);
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotalSize, allFiles.Sum(f => f.Length));
|
||||
_metrics.SetGaugeTo(MetricsAPI.GaugeFilesTotal, allFiles.Length);
|
||||
|
||||
CleanUpOutdatedFiles(ct);
|
||||
|
||||
CleanUpFilesBeyondSizeLimit(ct);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
_logger.LogError(e, "Error during cleanup task");
|
||||
}
|
||||
|
||||
var cleanupCheckMinutes = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.CleanupCheckInMinutes), 15);
|
||||
|
||||
var now = DateTime.Now;
|
||||
TimeOnly currentTime = new(now.Hour, now.Minute, now.Second);
|
||||
TimeOnly futureTime = new(now.Hour, now.Minute - now.Minute % cleanupCheckMinutes, 0);
|
||||
var span = futureTime.AddMinutes(cleanupCheckMinutes) - currentTime;
|
||||
|
||||
_logger.LogInformation("File Cleanup Complete, next run at {date}", now.Add(span));
|
||||
await Task.Delay(span, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Cleanup Service started");
|
||||
|
||||
_cleanupCts = new();
|
||||
|
||||
_ = CleanUpTask(_cleanupCts.Token);
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_cleanupCts.Cancel();
|
||||
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private void CleanUpFilesBeyondSizeLimit(CancellationToken ct)
|
||||
{
|
||||
var sizeLimit = _configuration.GetValueOrDefault<double>(nameof(StaticFilesServerConfiguration.CacheSizeHardLimitInGiB), -1);
|
||||
if (sizeLimit <= 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_logger.LogInformation("Cleaning up files beyond the cache size limit of {cacheSizeLimit} GiB", sizeLimit);
|
||||
var allLocalFiles = Directory.EnumerateFiles(_cacheDir, "*", SearchOption.AllDirectories)
|
||||
.Where(f => !f.EndsWith("dl", StringComparison.OrdinalIgnoreCase))
|
||||
.Select(f => new FileInfo(f)).ToList()
|
||||
.OrderBy(f => f.LastAccessTimeUtc).ToList();
|
||||
var totalCacheSizeInBytes = allLocalFiles.Sum(s => s.Length);
|
||||
long cacheSizeLimitInBytes = (long)ByteSize.FromGibiBytes(sizeLimit).Bytes;
|
||||
while (totalCacheSizeInBytes > cacheSizeLimitInBytes && allLocalFiles.Any() && !ct.IsCancellationRequested)
|
||||
{
|
||||
var oldestFile = allLocalFiles[0];
|
||||
allLocalFiles.Remove(oldestFile);
|
||||
totalCacheSizeInBytes -= oldestFile.Length;
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, oldestFile.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_logger.LogInformation("Deleting {oldestFile} with size {size}MiB", oldestFile.FullName, ByteSize.FromBytes(oldestFile.Length).MebiBytes);
|
||||
oldestFile.Delete();
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during cache size limit cleanup");
|
||||
}
|
||||
}
|
||||
|
||||
private void CleanUpOutdatedFiles(CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
var unusedRetention = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.UnusedFileRetentionPeriodInDays), 14);
|
||||
var forcedDeletionAfterHours = _configuration.GetValueOrDefault(nameof(StaticFilesServerConfiguration.ForcedDeletionOfFilesAfterHours), -1);
|
||||
|
||||
_logger.LogInformation("Cleaning up files older than {filesOlderThanDays} days", unusedRetention);
|
||||
if (forcedDeletionAfterHours > 0)
|
||||
{
|
||||
_logger.LogInformation("Cleaning up files written to longer than {hours}h ago", forcedDeletionAfterHours);
|
||||
}
|
||||
|
||||
var prevTime = DateTime.Now.Subtract(TimeSpan.FromDays(unusedRetention));
|
||||
var prevTimeForcedDeletion = DateTime.Now.Subtract(TimeSpan.FromHours(forcedDeletionAfterHours));
|
||||
DirectoryInfo dir = new(_cacheDir);
|
||||
var allFilesInDir = dir.GetFiles("*", SearchOption.AllDirectories)
|
||||
.Where(f => !f.Name.EndsWith("dl", StringComparison.OrdinalIgnoreCase))
|
||||
.ToList();
|
||||
|
||||
foreach (var file in allFilesInDir)
|
||||
{
|
||||
if (file.LastAccessTime < prevTime)
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, file.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_logger.LogInformation("File outdated: {fileName}, {fileSize}MiB", file.Name, ByteSize.FromBytes(file.Length).MebiBytes);
|
||||
file.Delete();
|
||||
}
|
||||
else if (forcedDeletionAfterHours > 0 && file.LastWriteTime < prevTimeForcedDeletion)
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, file.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_logger.LogInformation("File forcefully deleted: {fileName}, {fileSize}MiB", file.Name, ByteSize.FromBytes(file.Length).MebiBytes);
|
||||
file.Delete();
|
||||
}
|
||||
else if (file.Length == 0 && !string.Equals(file.Extension, ".dl", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotalSize, file.Length);
|
||||
_metrics.DecGauge(MetricsAPI.GaugeFilesTotal);
|
||||
_logger.LogInformation("File with size 0 deleted: {filename}", file.Name);
|
||||
file.Delete();
|
||||
}
|
||||
|
||||
ct.ThrowIfCancellationRequested();
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Error during file cleanup of old files");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
using LightlessSync.API.Routes;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
public class ShardRegistrationService : IHostedService
|
||||
{
|
||||
private readonly ILogger<ShardRegistrationService> _logger;
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configurationService;
|
||||
private readonly HttpClient _httpClient = new();
|
||||
private readonly CancellationTokenSource _heartBeatCts = new();
|
||||
private bool _isRegistered = false;
|
||||
|
||||
public ShardRegistrationService(ILogger<ShardRegistrationService> logger,
|
||||
IConfigurationService<StaticFilesServerConfiguration> configurationService,
|
||||
ServerTokenGenerator serverTokenGenerator)
|
||||
{
|
||||
_logger = logger;
|
||||
_configurationService = configurationService;
|
||||
_httpClient.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", serverTokenGenerator.Token);
|
||||
}
|
||||
|
||||
private void OnConfigChanged(object sender, EventArgs e)
|
||||
{
|
||||
_isRegistered = false;
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Starting");
|
||||
_configurationService.ConfigChangedEvent += OnConfigChanged;
|
||||
_ = Task.Run(() => HeartbeatLoop(_heartBeatCts.Token));
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Stopping");
|
||||
|
||||
_configurationService.ConfigChangedEvent -= OnConfigChanged;
|
||||
_heartBeatCts.Cancel();
|
||||
_heartBeatCts.Dispose();
|
||||
// call unregister
|
||||
await UnregisterShard().ConfigureAwait(false);
|
||||
_httpClient.Dispose();
|
||||
}
|
||||
|
||||
private async Task HeartbeatLoop(CancellationToken ct)
|
||||
{
|
||||
while (!_heartBeatCts.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await ProcessHeartbeat(ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Issue during Heartbeat");
|
||||
_isRegistered = false;
|
||||
}
|
||||
|
||||
await Task.Delay(TimeSpan.FromSeconds(30), ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessHeartbeat(CancellationToken ct)
|
||||
{
|
||||
if (!_isRegistered)
|
||||
{
|
||||
await TryRegisterShard(ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await ShardHeartbeat(ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task ShardHeartbeat(CancellationToken ct)
|
||||
{
|
||||
Uri mainServer = _configurationService.GetValue<Uri>(nameof(StaticFilesServerConfiguration.MainFileServerAddress));
|
||||
_logger.LogInformation("Running heartbeat against Main {server}", mainServer);
|
||||
|
||||
using var heartBeat = await _httpClient.PostAsync(new Uri(mainServer, LightlessFiles.Main + "/shardHeartbeat"), null, ct).ConfigureAwait(false);
|
||||
heartBeat.EnsureSuccessStatusCode();
|
||||
}
|
||||
|
||||
private async Task TryRegisterShard(CancellationToken ct)
|
||||
{
|
||||
Uri mainServer = _configurationService.GetValue<Uri>(nameof(StaticFilesServerConfiguration.MainFileServerAddress));
|
||||
_logger.LogInformation("Registering Shard with Main {server}", mainServer);
|
||||
var config = _configurationService.GetValue<ShardConfiguration>(nameof(StaticFilesServerConfiguration.ShardConfiguration));
|
||||
_logger.LogInformation("Config Value {varName}: {value}", nameof(ShardConfiguration.Continents), string.Join(", ", config.Continents));
|
||||
_logger.LogInformation("Config Value {varName}: {value}", nameof(ShardConfiguration.FileMatch), config.FileMatch);
|
||||
_logger.LogInformation("Config Value {varName}: {value}", nameof(ShardConfiguration.RegionUris), string.Join("; ", config.RegionUris.Select(k => k.Key + ":" + k.Value)));
|
||||
|
||||
using var register = await _httpClient.PostAsJsonAsync(new Uri(mainServer, LightlessFiles.Main + "/shardRegister"), config, ct).ConfigureAwait(false);
|
||||
register.EnsureSuccessStatusCode();
|
||||
_isRegistered = true;
|
||||
}
|
||||
|
||||
private async Task UnregisterShard()
|
||||
{
|
||||
Uri mainServer = _configurationService.GetValue<Uri>(nameof(StaticFilesServerConfiguration.MainFileServerAddress));
|
||||
_logger.LogInformation("Unregistering Shard with Main {server}", mainServer);
|
||||
using var heartBeat = await _httpClient.PostAsync(new Uri(mainServer, LightlessFiles.Main + "/shardUnregister"), null).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
272
LightlessSyncServer/LightlessSyncStaticFilesServer/Startup.cs
Normal file
272
LightlessSyncServer/LightlessSyncStaticFilesServer/Startup.cs
Normal file
@@ -0,0 +1,272 @@
|
||||
using LightlessSyncShared.Data;
|
||||
using LightlessSyncShared.Metrics;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils;
|
||||
using LightlessSyncStaticFilesServer.Controllers;
|
||||
using LightlessSyncStaticFilesServer.Services;
|
||||
using LightlessSyncStaticFilesServer.Utils;
|
||||
using MessagePack;
|
||||
using MessagePack.Resolvers;
|
||||
using Microsoft.AspNetCore.Authentication.JwtBearer;
|
||||
using Microsoft.AspNetCore.Authorization;
|
||||
using Microsoft.AspNetCore.Mvc.Controllers;
|
||||
using Microsoft.AspNetCore.Server.Kestrel.Core;
|
||||
using Microsoft.AspNetCore.SignalR;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.IdentityModel.Tokens;
|
||||
using Prometheus;
|
||||
using StackExchange.Redis.Extensions.Core.Configuration;
|
||||
using StackExchange.Redis.Extensions.System.Text.Json;
|
||||
using StackExchange.Redis;
|
||||
using System.Net;
|
||||
using System.Text;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer;
|
||||
|
||||
public class Startup
|
||||
{
|
||||
private bool _isMain;
|
||||
private bool _isDistributionNode;
|
||||
private readonly ILogger<Startup> _logger;
|
||||
|
||||
public Startup(IConfiguration configuration, ILogger<Startup> logger)
|
||||
{
|
||||
Configuration = configuration;
|
||||
_logger = logger;
|
||||
var lightlessSettings = Configuration.GetRequiredSection("LightlessSync");
|
||||
_isDistributionNode = lightlessSettings.GetValue(nameof(StaticFilesServerConfiguration.IsDistributionNode), false);
|
||||
_isMain = string.IsNullOrEmpty(lightlessSettings.GetValue(nameof(StaticFilesServerConfiguration.MainFileServerAddress), string.Empty)) && _isDistributionNode;
|
||||
}
|
||||
|
||||
public IConfiguration Configuration { get; }
|
||||
|
||||
public void ConfigureServices(IServiceCollection services)
|
||||
{
|
||||
services.AddHttpContextAccessor();
|
||||
|
||||
services.AddLogging();
|
||||
|
||||
services.Configure<StaticFilesServerConfiguration>(Configuration.GetRequiredSection("LightlessSync"));
|
||||
services.Configure<LightlessConfigurationBase>(Configuration.GetRequiredSection("LightlessSync"));
|
||||
services.Configure<KestrelServerOptions>(Configuration.GetSection("Kestrel"));
|
||||
services.AddSingleton(Configuration);
|
||||
|
||||
var lightlessConfig = Configuration.GetRequiredSection("LightlessSync");
|
||||
|
||||
// metrics configuration
|
||||
services.AddSingleton(m => new LightlessMetrics(m.GetService<ILogger<LightlessMetrics>>(), new List<string>
|
||||
{
|
||||
MetricsAPI.CounterFileRequests,
|
||||
MetricsAPI.CounterFileRequestSize
|
||||
}, new List<string>
|
||||
{
|
||||
MetricsAPI.GaugeFilesTotalColdStorage,
|
||||
MetricsAPI.GaugeFilesTotalSizeColdStorage,
|
||||
MetricsAPI.GaugeFilesTotalSize,
|
||||
MetricsAPI.GaugeFilesTotal,
|
||||
MetricsAPI.GaugeFilesUniquePastDay,
|
||||
MetricsAPI.GaugeFilesUniquePastDaySize,
|
||||
MetricsAPI.GaugeFilesUniquePastHour,
|
||||
MetricsAPI.GaugeFilesUniquePastHourSize,
|
||||
MetricsAPI.GaugeCurrentDownloads,
|
||||
MetricsAPI.GaugeDownloadQueue,
|
||||
MetricsAPI.GaugeDownloadQueueCancelled,
|
||||
MetricsAPI.GaugeDownloadPriorityQueue,
|
||||
MetricsAPI.GaugeDownloadPriorityQueueCancelled,
|
||||
MetricsAPI.GaugeQueueFree,
|
||||
MetricsAPI.GaugeQueueInactive,
|
||||
MetricsAPI.GaugeQueueActive,
|
||||
MetricsAPI.GaugeFilesDownloadingFromCache,
|
||||
MetricsAPI.GaugeFilesTasksWaitingForDownloadFromCache
|
||||
}));
|
||||
|
||||
// generic services
|
||||
services.AddSingleton<CachedFileProvider>();
|
||||
services.AddSingleton<FileStatisticsService>();
|
||||
services.AddSingleton<RequestFileStreamResultFactory>();
|
||||
services.AddSingleton<ServerTokenGenerator>();
|
||||
services.AddSingleton<RequestQueueService>();
|
||||
services.AddHostedService(p => p.GetService<RequestQueueService>());
|
||||
services.AddHostedService(m => m.GetService<FileStatisticsService>());
|
||||
services.AddSingleton<IConfigurationService<LightlessConfigurationBase>, LightlessConfigurationServiceClient<LightlessConfigurationBase>>();
|
||||
services.AddHostedService(p => (LightlessConfigurationServiceClient<LightlessConfigurationBase>)p.GetService<IConfigurationService<LightlessConfigurationBase>>());
|
||||
|
||||
// specific services
|
||||
if (_isMain)
|
||||
{
|
||||
services.AddSingleton<IClientReadyMessageService, MainClientReadyMessageService>();
|
||||
services.AddHostedService<MainFileCleanupService>();
|
||||
services.AddSingleton<IConfigurationService<StaticFilesServerConfiguration>, LightlessConfigurationServiceServer<StaticFilesServerConfiguration>>();
|
||||
services.AddSingleton<MainServerShardRegistrationService>();
|
||||
services.AddHostedService(s => s.GetRequiredService<MainServerShardRegistrationService>());
|
||||
services.AddDbContextPool<LightlessDbContext>(options =>
|
||||
{
|
||||
options.UseNpgsql(Configuration.GetConnectionString("DefaultConnection"), builder =>
|
||||
{
|
||||
builder.MigrationsHistoryTable("_efmigrationshistory", "public");
|
||||
}).UseSnakeCaseNamingConvention();
|
||||
options.EnableThreadSafetyChecks(false);
|
||||
}, lightlessConfig.GetValue(nameof(LightlessConfigurationBase.DbContextPoolSize), 1024));
|
||||
services.AddDbContextFactory<LightlessDbContext>(options =>
|
||||
{
|
||||
options.UseNpgsql(Configuration.GetConnectionString("DefaultConnection"), builder =>
|
||||
{
|
||||
builder.MigrationsHistoryTable("_efmigrationshistory", "public");
|
||||
builder.MigrationsAssembly("LightlessSyncShared");
|
||||
}).UseSnakeCaseNamingConvention();
|
||||
options.EnableThreadSafetyChecks(false);
|
||||
});
|
||||
|
||||
var signalRServiceBuilder = services.AddSignalR(hubOptions =>
|
||||
{
|
||||
hubOptions.MaximumReceiveMessageSize = long.MaxValue;
|
||||
hubOptions.EnableDetailedErrors = true;
|
||||
hubOptions.MaximumParallelInvocationsPerClient = 10;
|
||||
hubOptions.StreamBufferCapacity = 200;
|
||||
}).AddMessagePackProtocol(opt =>
|
||||
{
|
||||
var resolver = CompositeResolver.Create(StandardResolverAllowPrivate.Instance,
|
||||
BuiltinResolver.Instance,
|
||||
AttributeFormatterResolver.Instance,
|
||||
// replace enum resolver
|
||||
DynamicEnumAsStringResolver.Instance,
|
||||
DynamicGenericResolver.Instance,
|
||||
DynamicUnionResolver.Instance,
|
||||
DynamicObjectResolver.Instance,
|
||||
PrimitiveObjectResolver.Instance,
|
||||
// final fallback(last priority)
|
||||
StandardResolver.Instance);
|
||||
|
||||
opt.SerializerOptions = MessagePackSerializerOptions.Standard
|
||||
.WithCompression(MessagePackCompression.Lz4Block)
|
||||
.WithResolver(resolver);
|
||||
});
|
||||
|
||||
// configure redis for SignalR
|
||||
var redisConnection = lightlessConfig.GetValue(nameof(ServerConfiguration.RedisConnectionString), string.Empty);
|
||||
signalRServiceBuilder.AddStackExchangeRedis(redisConnection, options => { });
|
||||
|
||||
var options = ConfigurationOptions.Parse(redisConnection);
|
||||
|
||||
var endpoint = options.EndPoints[0];
|
||||
string address = "";
|
||||
int port = 0;
|
||||
if (endpoint is DnsEndPoint dnsEndPoint) { address = dnsEndPoint.Host; port = dnsEndPoint.Port; }
|
||||
if (endpoint is IPEndPoint ipEndPoint) { address = ipEndPoint.Address.ToString(); port = ipEndPoint.Port; }
|
||||
var redisConfiguration = new RedisConfiguration()
|
||||
{
|
||||
AbortOnConnectFail = true,
|
||||
KeyPrefix = "",
|
||||
Hosts = new RedisHost[]
|
||||
{
|
||||
new RedisHost(){ Host = address, Port = port },
|
||||
},
|
||||
AllowAdmin = true,
|
||||
ConnectTimeout = options.ConnectTimeout,
|
||||
Database = 0,
|
||||
Ssl = false,
|
||||
Password = options.Password,
|
||||
ServerEnumerationStrategy = new ServerEnumerationStrategy()
|
||||
{
|
||||
Mode = ServerEnumerationStrategy.ModeOptions.All,
|
||||
TargetRole = ServerEnumerationStrategy.TargetRoleOptions.Any,
|
||||
UnreachableServerAction = ServerEnumerationStrategy.UnreachableServerActionOptions.Throw,
|
||||
},
|
||||
MaxValueLength = 1024,
|
||||
PoolSize = lightlessConfig.GetValue(nameof(ServerConfiguration.RedisPool), 50),
|
||||
SyncTimeout = options.SyncTimeout,
|
||||
};
|
||||
|
||||
services.AddStackExchangeRedisExtensions<SystemTextJsonSerializer>(redisConfiguration);
|
||||
}
|
||||
else
|
||||
{
|
||||
services.AddSingleton<ShardRegistrationService>();
|
||||
services.AddHostedService(s => s.GetRequiredService<ShardRegistrationService>());
|
||||
services.AddSingleton<IClientReadyMessageService, ShardClientReadyMessageService>();
|
||||
services.AddHostedService<ShardFileCleanupService>();
|
||||
services.AddSingleton<IConfigurationService<StaticFilesServerConfiguration>, LightlessConfigurationServiceClient<StaticFilesServerConfiguration>>();
|
||||
services.AddHostedService(p => (LightlessConfigurationServiceClient<StaticFilesServerConfiguration>)p.GetService<IConfigurationService<StaticFilesServerConfiguration>>());
|
||||
}
|
||||
|
||||
services.AddMemoryCache();
|
||||
|
||||
// controller setup
|
||||
services.AddControllers().ConfigureApplicationPartManager(a =>
|
||||
{
|
||||
a.FeatureProviders.Remove(a.FeatureProviders.OfType<ControllerFeatureProvider>().First());
|
||||
if (_isMain)
|
||||
{
|
||||
a.FeatureProviders.Add(new AllowedControllersFeatureProvider(typeof(LightlessStaticFilesServerConfigurationController),
|
||||
typeof(CacheController), typeof(RequestController), typeof(ServerFilesController),
|
||||
typeof(DistributionController), typeof(MainController), typeof(SpeedTestController)));
|
||||
}
|
||||
else if (_isDistributionNode)
|
||||
{
|
||||
a.FeatureProviders.Add(new AllowedControllersFeatureProvider(typeof(CacheController), typeof(RequestController), typeof(DistributionController), typeof(SpeedTestController)));
|
||||
}
|
||||
else
|
||||
{
|
||||
a.FeatureProviders.Add(new AllowedControllersFeatureProvider(typeof(CacheController), typeof(RequestController), typeof(SpeedTestController)));
|
||||
}
|
||||
});
|
||||
|
||||
// authentication and authorization
|
||||
services.AddOptions<JwtBearerOptions>(JwtBearerDefaults.AuthenticationScheme)
|
||||
.Configure<IConfigurationService<LightlessConfigurationBase>>((o, s) =>
|
||||
{
|
||||
o.TokenValidationParameters = new()
|
||||
{
|
||||
ValidateIssuer = false,
|
||||
ValidateLifetime = true,
|
||||
ValidateAudience = false,
|
||||
ValidateIssuerSigningKey = true,
|
||||
IssuerSigningKey = new SymmetricSecurityKey(Encoding.ASCII.GetBytes(s.GetValue<string>(nameof(LightlessConfigurationBase.Jwt))))
|
||||
};
|
||||
});
|
||||
services.AddAuthentication(o =>
|
||||
{
|
||||
o.DefaultAuthenticateScheme = JwtBearerDefaults.AuthenticationScheme;
|
||||
o.DefaultChallengeScheme = JwtBearerDefaults.AuthenticationScheme;
|
||||
o.DefaultScheme = JwtBearerDefaults.AuthenticationScheme;
|
||||
}).AddJwtBearer();
|
||||
services.AddAuthorization(options =>
|
||||
{
|
||||
options.FallbackPolicy = new AuthorizationPolicyBuilder().RequireAuthenticatedUser().Build();
|
||||
options.AddPolicy("Internal", new AuthorizationPolicyBuilder().RequireClaim(LightlessClaimTypes.Internal, "true").Build());
|
||||
});
|
||||
services.AddSingleton<IUserIdProvider, IdBasedUserIdProvider>();
|
||||
|
||||
services.AddHealthChecks();
|
||||
services.AddHttpLogging(e => e = new Microsoft.AspNetCore.HttpLogging.HttpLoggingOptions());
|
||||
}
|
||||
|
||||
public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
|
||||
{
|
||||
app.UseHttpLogging();
|
||||
|
||||
app.UseRouting();
|
||||
|
||||
var config = app.ApplicationServices.GetRequiredService<IConfigurationService<LightlessConfigurationBase>>();
|
||||
|
||||
var metricServer = new KestrelMetricServer(config.GetValueOrDefault<int>(nameof(LightlessConfigurationBase.MetricsPort), 4981));
|
||||
metricServer.Start();
|
||||
|
||||
app.UseHttpMetrics();
|
||||
|
||||
app.UseAuthentication();
|
||||
app.UseAuthorization();
|
||||
|
||||
app.UseEndpoints(e =>
|
||||
{
|
||||
if (_isMain)
|
||||
{
|
||||
e.MapHub<LightlessSyncServer.Hubs.LightlessHub>("/dummyhub");
|
||||
}
|
||||
|
||||
e.MapControllers();
|
||||
e.MapHealthChecks("/health").WithMetadata(new AllowAnonymousAttribute());
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Utils;
|
||||
|
||||
public sealed class BlockFileDataStream : Stream
|
||||
{
|
||||
private readonly IEnumerable<BlockFileDataSubstream> _substreams;
|
||||
private int _currentStreamIndex = 0;
|
||||
|
||||
public BlockFileDataStream(IEnumerable<BlockFileDataSubstream> substreams)
|
||||
{
|
||||
_substreams = substreams;
|
||||
}
|
||||
|
||||
public override bool CanRead => true;
|
||||
public override bool CanSeek => false;
|
||||
public override bool CanWrite => false;
|
||||
public override long Length => throw new NotSupportedException();
|
||||
public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
|
||||
|
||||
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
|
||||
{
|
||||
int totalRead = 0;
|
||||
int currentOffset = 0;
|
||||
int remainingCount = count;
|
||||
while (totalRead < count && _currentStreamIndex < _substreams.Count())
|
||||
{
|
||||
var lastReadBytes = await _substreams.ElementAt(_currentStreamIndex).ReadAsync(buffer, currentOffset, remainingCount, cancellationToken).ConfigureAwait(false);
|
||||
if (lastReadBytes < remainingCount)
|
||||
{
|
||||
_substreams.ElementAt(_currentStreamIndex).Dispose();
|
||||
_currentStreamIndex++;
|
||||
}
|
||||
|
||||
totalRead += lastReadBytes;
|
||||
currentOffset += lastReadBytes;
|
||||
remainingCount -= lastReadBytes;
|
||||
}
|
||||
|
||||
return totalRead;
|
||||
}
|
||||
|
||||
public override int Read(byte[] buffer, int offset, int count)
|
||||
{
|
||||
int totalRead = 0;
|
||||
int currentOffset = 0;
|
||||
int remainingCount = count;
|
||||
while (totalRead < count && _currentStreamIndex < _substreams.Count())
|
||||
{
|
||||
var lastReadBytes = _substreams.ElementAt(_currentStreamIndex).Read(buffer, currentOffset, remainingCount);
|
||||
if (lastReadBytes < remainingCount)
|
||||
{
|
||||
_substreams.ElementAt(_currentStreamIndex).Dispose();
|
||||
_currentStreamIndex++;
|
||||
}
|
||||
|
||||
totalRead += lastReadBytes;
|
||||
currentOffset += lastReadBytes;
|
||||
remainingCount -= lastReadBytes;
|
||||
}
|
||||
|
||||
return totalRead;
|
||||
}
|
||||
|
||||
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
|
||||
{
|
||||
int totalRead = 0;
|
||||
|
||||
while (totalRead < buffer.Length && _currentStreamIndex < _substreams.Count())
|
||||
{
|
||||
var substream = _substreams.ElementAt(_currentStreamIndex);
|
||||
int bytesRead = await substream.ReadAsync(buffer.Slice(totalRead), cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (bytesRead == 0)
|
||||
{
|
||||
substream.Dispose();
|
||||
_currentStreamIndex++;
|
||||
}
|
||||
else
|
||||
{
|
||||
totalRead += bytesRead;
|
||||
}
|
||||
}
|
||||
|
||||
return totalRead;
|
||||
}
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (disposing)
|
||||
{
|
||||
foreach (var substream in _substreams)
|
||||
{
|
||||
// probably unnecessary but better safe than sorry
|
||||
substream.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
|
||||
public override void Flush() => throw new NotSupportedException();
|
||||
|
||||
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
|
||||
|
||||
public override void SetLength(long value) => throw new NotSupportedException();
|
||||
|
||||
public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
using System.Globalization;
|
||||
using System.Text;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Utils;
|
||||
|
||||
public sealed class BlockFileDataSubstream : IDisposable
|
||||
{
|
||||
private readonly MemoryStream _headerStream;
|
||||
private bool _disposed = false;
|
||||
private readonly Lazy<FileStream> _dataStreamLazy;
|
||||
private FileStream DataStream => _dataStreamLazy.Value;
|
||||
|
||||
public BlockFileDataSubstream(FileInfo file)
|
||||
{
|
||||
_dataStreamLazy = new(() => File.Open(file.FullName, GetFileStreamOptions(file.Length)));
|
||||
_headerStream = new MemoryStream(Encoding.ASCII.GetBytes("#" + file.Name + ":" + file.Length.ToString(CultureInfo.InvariantCulture) + "#"));
|
||||
}
|
||||
|
||||
private static FileStreamOptions GetFileStreamOptions(long fileSize)
|
||||
{
|
||||
int bufferSize = fileSize switch
|
||||
{
|
||||
<= 128 * 1024 => 0,
|
||||
<= 512 * 1024 => 4096,
|
||||
<= 1 * 1024 * 1024 => 65536,
|
||||
<= 10 * 1024 * 1024 => 131072,
|
||||
<= 100 * 1024 * 1024 => 524288,
|
||||
_ => 1048576
|
||||
};
|
||||
|
||||
FileStreamOptions opts = new()
|
||||
{
|
||||
Mode = FileMode.Open,
|
||||
Access = FileAccess.Read,
|
||||
Share = FileShare.Read | FileShare.Inheritable,
|
||||
BufferSize = bufferSize
|
||||
};
|
||||
|
||||
return opts;
|
||||
}
|
||||
|
||||
public int Read(byte[] inputBuffer, int offset, int count)
|
||||
{
|
||||
int bytesRead = 0;
|
||||
|
||||
// Read from header stream if it has remaining data
|
||||
if (_headerStream.Position < _headerStream.Length)
|
||||
{
|
||||
int headerBytesToRead = (int)Math.Min(count, _headerStream.Length - _headerStream.Position);
|
||||
bytesRead += _headerStream.Read(inputBuffer, offset, headerBytesToRead);
|
||||
offset += bytesRead;
|
||||
count -= bytesRead;
|
||||
}
|
||||
|
||||
// Read from data stream if there is still space in buffer
|
||||
if (count > 0 && DataStream.Position < DataStream.Length)
|
||||
{
|
||||
bytesRead += DataStream.Read(inputBuffer, offset, count);
|
||||
}
|
||||
|
||||
return bytesRead;
|
||||
}
|
||||
|
||||
public async Task<int> ReadAsync(byte[] inputBuffer, int offset, int count, CancellationToken cancellationToken = default)
|
||||
{
|
||||
int bytesRead = 0;
|
||||
|
||||
// Async read from header stream
|
||||
if (_headerStream.Position < _headerStream.Length)
|
||||
{
|
||||
int headerBytesToRead = (int)Math.Min(count, _headerStream.Length - _headerStream.Position);
|
||||
bytesRead += await _headerStream.ReadAsync(inputBuffer.AsMemory(offset, headerBytesToRead), cancellationToken).ConfigureAwait(false);
|
||||
offset += bytesRead;
|
||||
count -= bytesRead;
|
||||
}
|
||||
|
||||
// Async read from data stream
|
||||
if (count > 0 && DataStream.Position < DataStream.Length)
|
||||
{
|
||||
bytesRead += await DataStream.ReadAsync(inputBuffer.AsMemory(offset, count), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return bytesRead;
|
||||
}
|
||||
|
||||
public async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
|
||||
{
|
||||
int bytesRead = 0;
|
||||
|
||||
// Async read from header stream
|
||||
if (_headerStream.Position < _headerStream.Length)
|
||||
{
|
||||
int headerBytesToRead = (int)Math.Min(buffer.Length, _headerStream.Length - _headerStream.Position);
|
||||
bytesRead += await _headerStream.ReadAsync(buffer.Slice(0, headerBytesToRead), cancellationToken).ConfigureAwait(false);
|
||||
buffer = buffer.Slice(headerBytesToRead);
|
||||
}
|
||||
|
||||
// Async read from data stream
|
||||
if (buffer.Length > 0 && DataStream.Position < DataStream.Length)
|
||||
{
|
||||
bytesRead += await DataStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return bytesRead;
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_headerStream.Dispose();
|
||||
if (_dataStreamLazy.IsValueCreated)
|
||||
DataStream.Dispose();
|
||||
_disposed = true;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
namespace LightlessSyncStaticFilesServer.Utils;
|
||||
|
||||
public static partial class FilePathUtil
|
||||
{
|
||||
public static FileInfo GetFileInfoForHash(string basePath, string hash)
|
||||
{
|
||||
if (hash.Length != 40 || !hash.All(char.IsAsciiLetterOrDigit)) throw new InvalidOperationException();
|
||||
|
||||
FileInfo fi = new(Path.Join(basePath, hash[0].ToString(), hash));
|
||||
if (!fi.Exists)
|
||||
{
|
||||
fi = new FileInfo(Path.Join(basePath, hash));
|
||||
if (!fi.Exists)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
return fi;
|
||||
}
|
||||
|
||||
public static string GetFilePath(string basePath, string hash)
|
||||
{
|
||||
if (hash.Length != 40 || !hash.All(char.IsAsciiLetterOrDigit)) throw new InvalidOperationException();
|
||||
|
||||
var dirPath = Path.Join(basePath, hash[0].ToString());
|
||||
var path = Path.Join(dirPath, hash);
|
||||
if (!Directory.Exists(dirPath)) Directory.CreateDirectory(dirPath);
|
||||
return path;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
using LightlessSyncShared.Metrics;
|
||||
using LightlessSyncStaticFilesServer.Services;
|
||||
using Microsoft.AspNetCore.Mvc;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Utils;
|
||||
|
||||
public class RequestFileStreamResult : FileStreamResult
|
||||
{
|
||||
private readonly Guid _requestId;
|
||||
private readonly RequestQueueService _requestQueueService;
|
||||
private readonly LightlessMetrics _lightlessMetrics;
|
||||
|
||||
public RequestFileStreamResult(Guid requestId, RequestQueueService requestQueueService, LightlessMetrics lightlessMetrics,
|
||||
Stream fileStream, string contentType) : base(fileStream, contentType)
|
||||
{
|
||||
_requestId = requestId;
|
||||
_requestQueueService = requestQueueService;
|
||||
_lightlessMetrics = lightlessMetrics;
|
||||
_lightlessMetrics.IncGauge(MetricsAPI.GaugeCurrentDownloads);
|
||||
}
|
||||
|
||||
public override void ExecuteResult(ActionContext context)
|
||||
{
|
||||
try
|
||||
{
|
||||
base.ExecuteResult(context);
|
||||
}
|
||||
catch
|
||||
{
|
||||
throw;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_requestQueueService.FinishRequest(_requestId);
|
||||
|
||||
_lightlessMetrics.DecGauge(MetricsAPI.GaugeCurrentDownloads);
|
||||
FileStream?.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
public override async Task ExecuteResultAsync(ActionContext context)
|
||||
{
|
||||
try
|
||||
{
|
||||
await base.ExecuteResultAsync(context).ConfigureAwait(false);
|
||||
}
|
||||
catch
|
||||
{
|
||||
throw;
|
||||
}
|
||||
finally
|
||||
{
|
||||
_requestQueueService.FinishRequest(_requestId);
|
||||
_lightlessMetrics.DecGauge(MetricsAPI.GaugeCurrentDownloads);
|
||||
FileStream?.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
using LightlessSyncShared.Metrics;
|
||||
using LightlessSyncShared.Services;
|
||||
using LightlessSyncShared.Utils.Configuration;
|
||||
using LightlessSyncStaticFilesServer.Services;
|
||||
|
||||
namespace LightlessSyncStaticFilesServer.Utils;
|
||||
|
||||
public class RequestFileStreamResultFactory
|
||||
{
|
||||
private readonly LightlessMetrics _metrics;
|
||||
private readonly RequestQueueService _requestQueueService;
|
||||
private readonly IConfigurationService<StaticFilesServerConfiguration> _configurationService;
|
||||
|
||||
public RequestFileStreamResultFactory(LightlessMetrics metrics, RequestQueueService requestQueueService, IConfigurationService<StaticFilesServerConfiguration> configurationService)
|
||||
{
|
||||
_metrics = metrics;
|
||||
_requestQueueService = requestQueueService;
|
||||
_configurationService = configurationService;
|
||||
}
|
||||
|
||||
public RequestFileStreamResult Create(Guid requestId, Stream stream)
|
||||
{
|
||||
return new RequestFileStreamResult(requestId, _requestQueueService,
|
||||
_metrics, stream, "application/octet-stream");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
namespace LightlessSyncStaticFilesServer.Utils;
|
||||
|
||||
public class UserQueueEntry
|
||||
{
|
||||
public UserQueueEntry(UserRequest userRequest, DateTime expirationDate)
|
||||
{
|
||||
UserRequest = userRequest;
|
||||
ExpirationDate = expirationDate;
|
||||
}
|
||||
|
||||
public void MarkActive()
|
||||
{
|
||||
IsActive = true;
|
||||
ActivationDate = DateTime.UtcNow;
|
||||
}
|
||||
|
||||
public UserRequest UserRequest { get; }
|
||||
public DateTime ExpirationDate { get; }
|
||||
public bool IsActive { get; private set; } = false;
|
||||
public DateTime ActivationDate { get; private set; }
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
namespace LightlessSyncStaticFilesServer.Utils;
|
||||
|
||||
public record UserRequest(Guid RequestId, string User, List<string> FileIds)
|
||||
{
|
||||
public bool IsCancelled { get; set; } = false;
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"DetailedErrors": true,
|
||||
"Logging": {
|
||||
"LogLevel": {
|
||||
"Default": "Information",
|
||||
"Microsoft.AspNetCore": "Warning"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
{
|
||||
"ConnectionStrings": {
|
||||
"DefaultConnection": "Host=localhost;Port=5432;Database=;Username=;Password="
|
||||
},
|
||||
"Logging": {
|
||||
"LogLevel": {
|
||||
"Default": "Information",
|
||||
"Microsoft.AspNetCore": "Warning"
|
||||
}
|
||||
},
|
||||
"Kestrel": {
|
||||
"Endpoints": {
|
||||
"Http": {
|
||||
"Url": "http://+:5001"
|
||||
},
|
||||
"Gprc": {
|
||||
"Protocols": "Http2",
|
||||
"Url": "http://+:5003"
|
||||
}
|
||||
}
|
||||
},
|
||||
"LightlessSync": {
|
||||
"ForcedDeletionOfFilesAfterHours": -1,
|
||||
"CacheSizeHardLimitInGiB": -1,
|
||||
"UnusedFileRetentionPeriodInDays": 7,
|
||||
"CacheDirectory": "G:\\ServerTest",
|
||||
"ServiceAddress": "http://localhost:5002",
|
||||
"RemoteCacheSourceUri": ""
|
||||
},
|
||||
"AllowedHosts": "*"
|
||||
}
|
||||
Reference in New Issue
Block a user