Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Push to Production #465

Merged
merged 4 commits into from
Jan 27, 2024
Merged

Push to Production #465

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fae1d03
Configurable Taskengine initial pause
angrave Jan 26, 2024
75d38ab
Merge pull request #463 from classtranscribe/Configurable-Taskengine-…
angrave Jan 26, 2024
ac57d33
Shared Box Client
angrave Jan 27, 2024
9dd1712
Merge pull request #464 from classtranscribe/SharedBoxClient
angrave Jan 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ClassTranscribeDatabase/Globals.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,8 @@ public class AppSettings
public string BUILDNUMBER { get; set; }
public string ES_CONNECTION_ADDR { get; set; }

public string INITIAL_TASKENGINE_PAUSE_MINUTES { get; set; } = "2";

// We let the message expire - which is likely if the server is overloaded- the periodic check will rediscover things to do later
// The suggested value is PERIODIC_CHECK_EVERY_MINUTES -5
// i.e we assume that the periodic check takes no more than 5 minutes to enqueue a task
Expand Down
90 changes: 61 additions & 29 deletions ClassTranscribeDatabase/Services/BoxAPI.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,19 +1,28 @@
using Box.V2;
using System;
using System.Linq;
using System.Threading.Tasks;
using System.Threading;

using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;

using Box.V2;
using Box.V2.Auth;
using Box.V2.Config;

using ClassTranscribeDatabase.Models;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using System;
using System.Linq;
using System.Threading.Tasks;


namespace ClassTranscribeDatabase.Services
{
public class BoxAPI
{
private readonly SlackLogger _slack;
private readonly ILogger _logger;
private BoxClient? _boxClient;

Check warning on line 22 in ClassTranscribeDatabase/Services/BoxAPI.cs

View workflow job for this annotation

GitHub Actions / Build 

The annotation for nullable reference types should only be used in code within a '#nullable' annotations context.

Check warning on line 22 in ClassTranscribeDatabase/Services/BoxAPI.cs

View workflow job for this annotation

GitHub Actions / Build 

The annotation for nullable reference types should only be used in code within a '#nullable' annotations context.
private DateTimeOffset _lastRefreshed = DateTimeOffset.MinValue;
private SemaphoreSlim _RefreshSemaphore = new SemaphoreSlim(1, 1); // async-safe mutex to ensure only one thread is refreshing the token at a time

public BoxAPI(ILogger<BoxAPI> logger, SlackLogger slack)
{
_logger = logger;
Expand All @@ -30,6 +39,7 @@
// This implementation is overly chatty with the database, but we rarely create access tokens so it is not a problem
using (var _context = CTDbContext.CreateDbContext())
{

if (!await _context.Dictionaries.Where(d => d.Key == CommonUtils.BOX_ACCESS_TOKEN).AnyAsync())
{
_context.Dictionaries.Add(new Dictionary
Expand All @@ -47,13 +57,15 @@
await _context.SaveChangesAsync();
}


var accessToken = _context.Dictionaries.Where(d => d.Key == CommonUtils.BOX_ACCESS_TOKEN).First();
var refreshToken = _context.Dictionaries.Where(d => d.Key == CommonUtils.BOX_REFRESH_TOKEN).First();
var config = new BoxConfig(Globals.appSettings.BOX_CLIENT_ID, Globals.appSettings.BOX_CLIENT_SECRET, new Uri("http://locahost"));
var client = new Box.V2.BoxClient(config);
var auth = await client.Auth.AuthenticateAsync(authCode);
_logger.LogInformation("Created Box Tokens");
var tmpClient = new Box.V2.BoxClient(config);
var auth = await tmpClient.Auth.AuthenticateAsync(authCode);

_logger.LogInformation($"Created Box Tokens Access:({auth.AccessToken.Substring(0, 5)}) Refresh({auth.RefreshToken.Substring(0, 5)})");

accessToken.Value = auth.AccessToken;
refreshToken.Value = auth.RefreshToken;
await _context.SaveChangesAsync();
Expand All @@ -62,31 +74,37 @@
/// <summary>
/// Updates the accessToken and refreshToken. These keys must already exist in the Dictionary table.
/// </summary>
public async Task RefreshAccessTokenAsync()
private async Task<BoxClient> RefreshAccessTokenAsync()
{
// Only one thread should call this at a time (see semaphore in GetBoxClientAsync)
try
{
_logger.LogInformation($"RefreshAccessTokenAsync: Starting");
using (var _context = CTDbContext.CreateDbContext())
{
var accessToken = await _context.Dictionaries.Where(d => d.Key == CommonUtils.BOX_ACCESS_TOKEN).FirstAsync();
var refreshToken = await _context.Dictionaries.Where(d => d.Key == CommonUtils.BOX_REFRESH_TOKEN).FirstAsync();
var config = new BoxConfig(Globals.appSettings.BOX_CLIENT_ID, Globals.appSettings.BOX_CLIENT_SECRET, new Uri("http://locahost"));
var auth = new OAuthSession(accessToken.Value, refreshToken.Value, 3600, "bearer");
var client = new BoxClient(config, auth);
/// Try to refresh the access token
auth = await client.Auth.RefreshAccessTokenAsync(auth.AccessToken);
var initialAuth = new OAuthSession(accessToken.Value, refreshToken.Value, 3600, "bearer");
var initialClient = new BoxClient(config, initialAuth);
/// Refresh the access token
var auth = await initialClient.Auth.RefreshAccessTokenAsync(initialAuth.AccessToken);
/// Create the client again
client = new BoxClient(config, auth);
_logger.LogInformation("Refreshed Tokens");
_logger.LogInformation($"RefreshAccessTokenAsync: New Access Token ({auth.AccessToken.Substring(0, 5)}), New Refresh Token ({auth.RefreshToken.Substring(0, 5)})");

accessToken.Value = auth.AccessToken;
refreshToken.Value = auth.RefreshToken;
_lastRefreshed = DateTimeOffset.Now;
await _context.SaveChangesAsync();
_logger.LogInformation($"RefreshAccessTokenAsync: Creating New Box Client");
var client = new BoxClient(config, auth);
return client;
}
}
catch (Box.V2.Exceptions.BoxSessionInvalidatedException e)
{
_logger.LogError(e, "Box Token Failure.");
await _slack.PostErrorAsync(e, "Box Token Failure.");
_logger.LogError(e, "RefreshAccessTokenAsync: Box Token Failure.");
await _slack.PostErrorAsync(e, "RefreshAccessTokenAsync: Box Token Failure.");
throw;
}
}
Expand All @@ -95,18 +113,32 @@
/// </summary>
public async Task<BoxClient> GetBoxClientAsync()
{
// Todo RefreshAccessTokenAsync could return this information for us; and avoid another trip to the database
await RefreshAccessTokenAsync();
BoxClient boxClient;
using (var _context = CTDbContext.CreateDbContext())
try
{
var accessToken = await _context.Dictionaries.Where(d => d.Key == CommonUtils.BOX_ACCESS_TOKEN).FirstAsync();
var refreshToken = await _context.Dictionaries.Where(d => d.Key == CommonUtils.BOX_REFRESH_TOKEN).FirstAsync();
var config = new BoxConfig(Globals.appSettings.BOX_CLIENT_ID, Globals.appSettings.BOX_CLIENT_SECRET, new Uri("http://locahost"));
var auth = new OAuthSession(accessToken.Value, refreshToken.Value, 3600, "bearer");
boxClient = new Box.V2.BoxClient(config, auth);
await _RefreshSemaphore.WaitAsync(); // // critical section : implementation of an async-safe mutex
var MAX_AGE_MINUTES = 50;
var remain = DateTimeOffset.Now.Subtract(_lastRefreshed).TotalMinutes;
_logger.LogInformation($"GetBoxClientAsync: {remain} minutes since last refresh. Max age {MAX_AGE_MINUTES}.");
if (_boxClient != null && remain < MAX_AGE_MINUTES)
{
return _boxClient;
}
_boxClient = await RefreshAccessTokenAsync();
_logger.LogInformation($"GetBoxClientAsync: _boxClient updated");
}
catch (Exception e)
{
_logger.LogError(e, "GetBoxClientAsync: Box Refresh Failure.");
throw;
}
return boxClient;
finally
{
_logger.LogInformation($"GetBoxClientAsync: Releasing Semaphore and returning");
_RefreshSemaphore.Release(1);
}

return _boxClient;
}

}
}
11 changes: 6 additions & 5 deletions TaskEngine/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ public static void Main()
.AddSingleton<PythonCrawlerTask>()
.AddSingleton<DescribeVideoTask>()
.AddSingleton<DescribeImageTask>()
.AddSingleton<UpdateBoxTokenTask>()
// .AddSingleton<UpdateBoxTokenTask>()
.AddSingleton<CreateBoxTokenTask>()
.AddSingleton<BuildElasticIndexTask>()
.AddSingleton<ExampleTask>()
Expand Down Expand Up @@ -141,8 +141,8 @@ public static void Main()
// We dont want concurrency for these tasks
_logger.LogInformation("Creating QueueAwakerTask and Box token tasks consumers.");
serviceProvider.GetService<QueueAwakerTask>().Consume(NO_CONCURRENCY); //TODO TOREVIEW: NO_CONCURRENCY?
serviceProvider.GetService<UpdateBoxTokenTask>().Consume(NO_CONCURRENCY);
serviceProvider.GetService<CreateBoxTokenTask>().Consume(NO_CONCURRENCY);
// does nothing at the moment serviceProvider.GetService<UpdateBoxTokenTask>().Consume(NO_CONCURRENCY);
serviceProvider.GetService<CreateBoxTokenTask>().Consume(NO_CONCURRENCY); // calls _box.CreateAccessTokenAsync(authCode);

// Elastic Search index should be built after TranscriptionTask
serviceProvider.GetService<BuildElasticIndexTask>().Consume(NO_CONCURRENCY);
Expand Down Expand Up @@ -170,11 +170,12 @@ public static void Main()
QueueAwakerTask queueAwakerTask = serviceProvider.GetService<QueueAwakerTask>();

int periodicCheck = Math.Max(1,Convert.ToInt32(Globals.appSettings.PERIODIC_CHECK_EVERY_MINUTES));

int initialPauseMinutes = Math.Max(1, Convert.ToInt32(Globals.appSettings.INITIAL_TASKENGINE_PAUSE_MINUTES));

_logger.LogInformation("Periodic Check Every {0} minutes", periodicCheck);
var timeInterval = new TimeSpan(0, periodicCheck, 0);

var initialPauseInterval = new TimeSpan(0, 2, 0);
var initialPauseInterval = new TimeSpan(0, initialPauseMinutes, 0);
_logger.LogInformation("Pausing {0} minutes before first periodicCheck", initialPauseInterval);

Thread.Sleep(initialPauseInterval);
Expand Down
Loading