Compare commits

..
Author SHA1 Message Date
benbierens aa58a7aee4 add instructions link QR 2024-11-09 02:54:48 +01:00
benbierens 7b45a1171e Ads QR codes for CIDs 2024-11-08 16:09:47 +01:00
benbierens b1cb61b51c link to devcon setup instructions 2024-11-01 10:46:34 +01:00
benbierens 768bfcc8eb working upload both local and cloud-9 node 2024-11-01 10:04:19 +01:00
benbierens 8af91ea74b working local 2024-11-01 09:51:20 +01:00
benbierens c785df3adf implementing 2024-11-01 09:24:46 +01:00
benbierens bb3f0fccd0 set up plan 2024-11-01 08:48:42 +01:00
benbierens 5c1ffbb8af Updates codex contracts 2024-10-31 08:52:07 +01:00
Ben ff4711e802 sets content-type correctly 2024-10-30 08:40:46 +01:00
Ben b8d6ac929b Update to new codex image 2024-10-30 08:34:41 +01:00
Ben acb0bf4f29 Updates to latest 2024-10-29 14:02:00 +01:00
Ben 8fe0bd6307 Applies http-timeout to download stream 2024-10-25 12:39:10 +02:00
benbierens e6a5838b05 Loops in metrics from Giuliano's async profiler 2024-10-24 16:10:07 +02:00
Ben 5c65d1d74e Fixes message length issue for admin-channel messages. 2024-10-22 13:55:52 +02:00
Ben 292b4b9b06 bumps to 0.1.7 2024-10-22 12:57:07 +02:00
Ben ddbe5b111a Merge branch 'feature/block-retransmit' 2024-10-22 12:54:33 +02:00
Ben a0abea4432 Merge branch 'master' into feature/block-retransmit
# Conflicts:
#	ProjectPlugins/CodexPlugin/ApiChecker.cs
2024-10-22 12:54:22 +02:00
Ben 2f39327db2 Adds multiswarm test 2024-10-22 10:19:00 +02:00
Ben d452293cac makes autoclient delete old data files 2024-10-21 11:10:39 +02:00
benbierens 3e245b707c Fix updating of moment buckets 2024-10-21 11:05:48 +02:00
benbierens 9e842207ab improves error reporting for discord bot 2024-10-18 11:03:05 +02:00
benbierens e29ffe4f9c blockTimeFinder now uses inclusive time ranges 2024-10-18 10:35:49 +02:00
benbierens d8a6df5845 attempt to fix infinite loop in moment-reference-builder 2024-10-18 09:03:19 +02:00
benbierens 1859994ec6 nice csv outputs 2024-10-18 08:45:54 +02:00
Ben Bierens 60b489ced1 Merge pull request #104 from codex-storage/feature/bot-checks-cids
Feature/bot checks cids
2024-10-18 08:23:58 +02:00
benbierens b72b4a850b writes csv 2024-10-17 09:37:15 +02:00
benbierens fc942b11f8 Implements /check command 2024-10-16 14:05:02 +02:00
benbierens 2c88ddfb6b brings in api with new calls 2024-10-16 12:42:52 +02:00
benbierens e352e5c65c Moves code to file 2024-10-14 11:30:07 +02:00
benbierens 2cbe030cff Implements logical operations for indexSets 2024-10-14 11:19:40 +02:00
benbierens a38e93a607 set up logical operations for indexset 2024-10-14 10:10:18 +02:00
benbierens 269365e101 Orders chain events by blocknumber 2024-10-14 09:43:05 +02:00
benbierens 02ca9db001 wip: blockreceives to csv 2024-10-14 09:11:29 +02:00
benbierens 672092b232 handles codex events without node or peer 2024-10-14 09:11:09 +02:00
benbierens 1bd84a4892 combinatorial swarm tests 2024-10-11 11:20:59 +02:00
benbierens 5896735884 set up swarm test 2024-10-11 10:30:07 +02:00
Ben bebeb3766e Fixes unstable transcript test. 2024-10-10 15:31:38 +02:00
Ben e0cbf8c84d Updates emoji maps 2024-10-10 14:03:42 +02:00
Ben 3ed91d7310 Bumps containers to dotnet 8.0 2024-10-10 11:54:42 +02:00
Ben 9573814574 Adds emoji icons to chain events log 2024-10-10 11:45:04 +02:00
Ben 5313f8a7ac Adds SlotReservationsFull event to chain state tracker 2024-10-10 11:15:07 +02:00
Ben c239d555dc More reliable test 2024-10-09 11:37:16 +02:00
Ben 2e9d7641a3 Adds test to capture upload-interrupt node crash. 2024-10-09 09:11:04 +02:00
Ben fc9249da20 updates to 0.1.6 2024-10-08 14:49:19 +02:00
Ben 978e085219 updates contracts plugin 2024-10-08 10:38:49 +02:00
Ben 960b0c3788 Automatic updating of CodexPlugin openapi.yaml. 2024-10-08 09:29:26 +02:00
benbierens 9ca4bf8afc Updates marketplace plugin 2024-10-07 15:59:05 +02:00
Slava cfdc25335c Add .gitattributes for line endings in .bat, .cmd and .sh scripts (#103) 2024-10-05 23:08:49 +03:00
benbierens 04f087efe4 Bump to dotnet 8 2024-10-03 14:02:28 +02:00
Ben f6aa122245 Fixes issue where timesegments is locked up for max 50x segmentDuration when catching up to realtime. 2024-09-30 16:23:55 +02:00
Ben db4c4a87e0 Rewarderbot: Speeds up time segments when catching up to realtime. 2024-09-30 15:34:40 +02:00
Ben ffb5eb294a Fix kubernetes port name issue 2024-09-30 15:23:14 +02:00
Ben 6a8c74e02a Adds short-format UTC to block header for each chain-events channel entry. 2024-09-30 11:09:44 +02:00
Ben 200de1d7f7 Update: Bot reports interactions in admin channel 2024-09-30 10:59:22 +02:00
Ben f801cb082e update 2024-09-24 15:21:33 +02:00
Ben 2a61dad556 updates codex to 0.1.4 2024-09-24 13:06:00 +02:00
Ben 2d90349b7b Updates autobootstrap fixture for parallel running. 2024-09-24 10:10:59 +02:00
Ben c4b6d01530 Adds check for usage of availabilities. 2024-09-23 11:31:57 +02:00
Ben c9fedac592 Adds call to get availabilities 2024-09-23 10:52:12 +02:00
benbierens cedec0d4cc defaults for EC params that actually work 2024-09-12 15:29:50 +02:00
benbierens 769b9c3aca div by zero in time calculation 2024-09-12 15:14:15 +02:00
benbierens 88c675adf9 implements performance logger for autoclient 2024-09-12 15:03:06 +02:00
benbierens 75fcc68caf Merge branch 'feature/better-autoclient' 2024-09-12 14:42:42 +02:00
benbierens a41272f160 Removes rest server 2024-09-12 14:42:19 +02:00
benbierens 8e018cbae9 setup 2024-09-12 14:38:15 +02:00
Ben 3c447eb4c5 multiple nodes on one autoclient 2024-09-12 12:05:42 +02:00
benbierens d53b760731 fixes docker path 2024-09-11 14:26:28 +02:00
benbierens fcadceb009 References autoclientcenter from autoclient 2024-09-11 14:24:12 +02:00
benbierens a02d9558e5 Merge branch 'feature/better-autoclient' 2024-09-11 14:08:29 +02:00
benbierens b3013a9b65 moves project 2024-09-11 14:08:21 +02:00
benbierens 6b0a16b627 Merge branch 'feature/better-autoclient' 2024-09-11 14:07:04 +02:00
benbierens eac06e8b3a builds docker image 2024-09-11 14:06:48 +02:00
benbierens f7fa35c7ba Implements center service 2024-09-11 14:00:22 +02:00
Ben a7526aaed1 setup center api 2024-09-11 12:08:06 +02:00
benbierens e7d9e833f1 Merge branch 'feature/self-updating-contracts-code' 2024-08-30 12:42:24 +02:00
benbierens 13dd0a649c Moves self-updater call to starter class. 2024-08-30 12:25:41 +02:00
benbierens bf5bd8d726 Successful automatic update 2024-08-30 11:17:36 +02:00
benbierens 8dcf9ff15e first implementation of contracts self-updater 2024-08-30 10:58:27 +02:00
benbierens 489209d549 generates deployment base class 2024-08-30 10:27:03 +02:00
Ben 017cee43c0 Merge branch 'index-encoding' 2024-08-27 13:32:46 +02:00
Ben 8fbef6ff51 working runlength implementation and tests 2024-08-27 13:32:33 +02:00
Ben 38ee2e4eb5 run-length encoding wins 2024-08-27 10:16:51 +02:00
Ben e4f7249f1e Update marketplace bytecode 2024-08-26 11:04:23 +02:00
benbierens e3ca516b97 Readme on how to run auto-client 2024-08-23 13:37:25 +02:00
benbierens 88daab379f Allows autoclient to generate files of any size with random data 2024-08-23 13:21:31 +02:00
Ben 00ed3caafe Sets up CI for building marketinsights api 2024-08-21 15:19:11 +02:00
Ben daad6468c6 Removes marketaverages from discordbot and rewarderbot 2024-08-21 15:08:18 +02:00
Ben 3d347a936d Implements marketinsights api 2024-08-21 15:03:20 +02:00
Ben 87bda475df Set up contract-failed event. 2024-08-21 13:59:54 +02:00
Ben 44dfa39737 api plus description 2024-08-21 11:34:54 +02:00
Ben d38f6da26f setup 2024-08-21 11:00:47 +02:00
Ben 5667cfc054 Pyramid test 2024-08-21 10:53:38 +02:00
Ben ecd0e70261 fixes serialization issue of containerAdditionals 2024-08-21 10:45:17 +02:00
Ben 5bbb95f1ff Option to exclude blockReceivedEvents from transcript 2024-08-21 09:53:20 +02:00
Ben 5aa3edbda2 Json formatting for transcript: omit null objects 2024-08-21 09:51:22 +02:00
Ben 9bdebb963b Fixes printing of disttest status 2024-08-20 15:31:45 +02:00
Ben c690598868 Merge branch 'master' into feature/viewing-network-topology
# Conflicts:
#	Tests/CodexTests/BasicTests/MarketplaceTests.cs
2024-08-20 15:26:54 +02:00
Ben ba59ac91d7 Updates marketplace contract 2024-08-20 15:26:19 +02:00
Ben 5170699ae4 implements writers for codex marketplace transcript events 2024-08-20 12:13:42 +02:00
Ben 09d2f418eb Setting up marketplace hooks 2024-08-20 11:44:15 +02:00
Ben cc9d04acd7 Adds normalized kademlia positions 2024-08-14 15:10:33 +02:00
Ben b5c8c55c72 wires up header node identity 2024-08-14 14:13:00 +02:00
benbierens 38b987144e wip: rework identity map 2024-08-13 14:21:15 +02:00
benbierens 8136bf1c92 Includes nodeId as part of node identity in transcript 2024-08-13 13:48:54 +02:00
benbierens 5ddec99114 ensures long-ids for transcript events 2024-08-08 10:24:16 +02:00
benbierens 58cebd43ce Identified issue with nodesDegree overview 2024-08-08 09:39:20 +02:00
benbierens 11a015e0b6 node degree overview 2024-08-07 11:18:40 +02:00
benbierens 745b7e43b4 simple analyser console tool 2024-08-06 15:56:42 +02:00
Ben a72866c725 Fixes issue where full path is included in moment-reference objects 2024-08-05 14:16:44 +02:00
Ben 2ec7f26387 finishes: fast millions of events support 2024-08-05 12:24:57 +02:00
benbierens 72865e7f96 works for 10,000,000 events, within 40 minutes 2024-08-02 17:52:44 +02:00
benbierens 6a0d2830d6 Speeds up transcript building but not by much 2024-08-02 14:25:38 +02:00
benbierens b3710f26ae working but very slow moment aggregation 2024-08-02 10:44:15 +02:00
benbierens f6cd9db408 test passes 2024-08-02 08:56:49 +02:00
benbierens 7a3a2b558b wip 2024-08-01 16:25:28 +02:00
benbierens 53aad5cb37 Sets up event bucket 2024-08-01 14:50:25 +02:00
benbierens fd65e1f022 fixes concurrency issue in http, fixes peer-dropped event 2024-08-01 11:19:05 +02:00
benbierens d16b8cb011 Fixes identity issue for runningpod/runningcontainer and log saving for stopped containers 2024-08-01 10:39:06 +02:00
benbierens 37611bdc66 peer dropped and starting/started events for upload and download 2024-08-01 09:09:30 +02:00
benbierens 7352cdf3fe Adds successful peer-dial event 2024-08-01 08:29:12 +02:00
benbierens 8e084d6ca0 prevents filename clashes 2024-07-31 14:49:42 +02:00
benbierens 017c71a6e8 typo in model 2024-07-31 11:25:59 +02:00
benbierens 488fbb383b Upgrade reader with moment-handler 2024-07-31 11:02:09 +02:00
benbierens cfd635146b Test attribute for creating transcripts 2024-07-31 09:50:02 +02:00
benbierens 7d737a5534 Moves to framework folder 2024-07-30 15:46:16 +02:00
benbierens 0d21246b06 Updates from viewer project 2024-07-30 15:42:51 +02:00
benbierens ec99f5f8aa Ensure peerId is known for all node events 2024-07-29 14:17:54 +02:00
benbierens 9d9f65c5a3 Fixes missing name and null events 2024-07-29 11:02:24 +02:00
benbierens 87271f4f37 Sets up starting event and bootstrap event 2024-07-29 10:16:37 +02:00
benbierens 9bebc23f6a Adds common header 2024-07-29 08:08:17 +02:00
benbierens 682c267b10 wip 2024-07-26 15:19:41 +02:00
benbierens 4de35b579f lock required for transcript writer 2024-07-26 13:21:29 +02:00
benbierens fb4d520ba4 Debugging transcripts for all tests 2024-07-26 10:56:22 +02:00
benbierens 1f72d4f37d Fixes compile of continuous tests 2024-07-26 10:14:54 +02:00
benbierens ecada92dc5 can generate transcript for two-client test 2024-07-26 10:11:29 +02:00
benbierens 410d62849a setting up log converter 2024-07-26 09:14:46 +02:00
benbierens 927ccaa119 successful transcript for two-client test 2024-07-26 08:39:27 +02:00
benbierens d58d65b751 wip connecting to codex plugin 2024-07-25 16:00:51 +02:00
benbierens 802b18e990 working transcript writer and reader 2024-07-25 15:12:25 +02:00
benbierens 8064051f2f Setting up basic idea of overwatch transcript 2024-07-25 10:30:20 +02:00
benbierens 485b2387e4 fixes multipeer download test 2024-07-25 10:10:11 +02:00
Ben ae4d782566 wip 2024-07-24 16:06:45 +02:00
Ben 8f16699bab CodexPluginPrebuild directory travel 2024-07-23 10:59:07 +02:00
Ben 5ba919a638 argument order error 2024-07-23 10:29:01 +02:00
Ben af7aeb2bba update 2024-07-23 09:58:29 +02:00
benbierens 1f11c5d6b5 Fixes custom replacements 2024-07-03 19:11:21 +02:00
Ben 8694bfe9be Merge branch 'feature/auto-client' 2024-07-02 10:44:31 +02:00
Ben 637b5a4736 Very interesting marketplace test 2024-07-02 10:44:15 +02:00
Ben e0cc46260b Adds presistence to log replaces 2024-07-02 10:37:32 +02:00
Ben c1ad97d26f Fixes exception in app-command creation 2024-07-02 10:22:00 +02:00
Ben 1e8f8106b0 Attempt to fix bot commands 2024-07-02 10:08:13 +02:00
Ben 926e323569 Better file manager 2024-07-01 15:59:08 +02:00
benbierens 8d32758918 Add option to delete app commands on startup to discord bot 2024-06-28 13:11:31 +02:00
235 changed files with 9088 additions and 1307 deletions
+30
View File
@@ -0,0 +1,30 @@
**/.classpath
**/.dockerignore
**/.env
**/.git
**/.gitignore
**/.project
**/.settings
**/.toolstarget
**/.vs
**/.vscode
**/*.*proj.user
**/*.dbmdl
**/*.jfm
**/azds.yaml
**/bin
**/charts
**/docker-compose*
**/Dockerfile*
**/node_modules
**/npm-debug.log
**/obj
**/secrets.dev.yaml
**/values.dev.yaml
LICENSE
README.md
!**/.gitignore
!.git/HEAD
!.git/config
!.git/packed-refs
!.git/refs/heads/**
+10
View File
@@ -0,0 +1,10 @@
# Set default behavior to automatically normalize line endings.
* text=auto
# Force bash scripts to always use lf line endings so that if a repo is accessed
# in Unix via a file share from Windows, the scripts will work.
*.sh text eol=lf
# Likewise, force cmd and batch scripts to always use crlf
*.cmd text eol=crlf
*.bat text eol=crlf
@@ -0,0 +1,27 @@
name: Docker - MarketInsights API
on:
push:
branches:
- master
tags:
- 'v*.*.*'
paths:
- 'Tools/MarketInsights/**'
- 'Framework/**'
- 'ProjectPlugins/**'
- .github/workflows/docker-marketinsights.yml
- .github/workflows/docker-reusable.yml
workflow_dispatch:
jobs:
build-and-push:
name: Build and Push
uses: ./.github/workflows/docker-reusable.yml
with:
docker_file: Tools/MarketInsights/Dockerfile
docker_repo: codexstorage/codex-marketinsights
secrets: inherit
+9
View File
@@ -0,0 +1,9 @@
<Application x:Class="DevconBoothImages.App"
xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
xmlns:local="clr-namespace:DevconBoothImages"
StartupUri="MainWindow.xaml">
<Application.Resources>
</Application.Resources>
</Application>
+14
View File
@@ -0,0 +1,14 @@
using System.Configuration;
using System.Data;
using System.Windows;
namespace DevconBoothImages
{
/// <summary>
/// Interaction logic for App.xaml
/// </summary>
public partial class App : Application
{
}
}
+10
View File
@@ -0,0 +1,10 @@
using System.Windows;
[assembly: ThemeInfo(
ResourceDictionaryLocation.None, //where theme specific resource dictionaries are located
//(used if a resource is not found in the page,
// or application resource dictionaries)
ResourceDictionaryLocation.SourceAssembly //where the generic resource dictionary is located
//(used if a resource is not found in the page,
// app, or any theme specific resource dictionaries)
)]
+77
View File
@@ -0,0 +1,77 @@
using CodexOpenApi;
using IdentityModel.Client;
using System.Net.Http;
using System.Windows;
using Utils;
namespace DevconBoothImages
{
public class Codexes
{
public Codexes(CodexApi local, CodexApi testnet)
{
Local = local;
Testnet = testnet;
}
public CodexApi Local { get; }
public CodexApi Testnet { get; }
}
public class CodexWrapper
{
public async Task<Codexes> GetCodexes()
{
var config = new Configuration();
return new Codexes(
await GetCodexWithPort(config.CodexLocalEndpoint),
await GetCodexWithoutPort(config.CodexPublicEndpoint, config.AuthUser, config.AuthPw)
);
}
private async Task<CodexApi> GetCodexWithPort(string endpoint)
{
var splitIndex = endpoint.LastIndexOf(':');
var host = endpoint.Substring(0, splitIndex);
var port = Convert.ToInt32(endpoint.Substring(splitIndex + 1));
var address = new Address(
host: host,
port: port
);
var client = new HttpClient();
var codex = new CodexApi(client);
codex.BaseUrl = $"{address.Host}:{address.Port}/api/codex/v1";
await CheckCodex(codex, endpoint);
return codex;
}
private async Task<CodexApi> GetCodexWithoutPort(string endpoint, string user, string pw)
{
var client = new HttpClient();
client.SetBasicAuthentication(user, pw);
var codex = new CodexApi(client);
codex.BaseUrl = $"{endpoint}/api/codex/v1";
await CheckCodex(codex, endpoint);
return codex;
}
private async Task CheckCodex(CodexApi codex, string endpoint)
{
try
{
var info = await codex.GetDebugInfoAsync();
if (string.IsNullOrEmpty(info.Id)) throw new Exception("Failed to fetch Codex node id");
}
catch (Exception ex)
{
MessageBox.Show($"Failed to connect to codex '{endpoint}': {ex}");
throw;
}
}
}
}
+13
View File
@@ -0,0 +1,13 @@
namespace DevconBoothImages
{
public class Configuration
{
public string CodexLocalEndpoint { get; } = "http://localhost:8080";
public string CodexPublicEndpoint { get; } = "https://api.testnet.codex.storage/storage/node-9";
public string AuthUser { get; } = "";
public string AuthPw { get; } = "";
public string LocalNodeBootstrapInfo { get; } = "";
public string WorkingDir { get; } = "D:\\DevconBoothApp";
}
}
@@ -0,0 +1,20 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>WinExe</OutputType>
<TargetFramework>net8.0-windows</TargetFramework>
<Nullable>enable</Nullable>
<ImplicitUsings>enable</ImplicitUsings>
<UseWPF>true</UseWPF>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="QRCoder" Version="1.6.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\ProjectPlugins\CodexPlugin\CodexPlugin.csproj" />
<ProjectReference Include="..\Tools\AutoClient\AutoClient.csproj" />
</ItemGroup>
</Project>
@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="utf-8"?>
<Project ToolsVersion="Current" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
<PropertyGroup />
<ItemGroup>
<ApplicationDefinition Update="App.xaml">
<SubType>Designer</SubType>
</ApplicationDefinition>
</ItemGroup>
<ItemGroup>
<Page Update="MainWindow.xaml">
<SubType>Designer</SubType>
</Page>
</ItemGroup>
</Project>
+38
View File
@@ -0,0 +1,38 @@
<Window x:Class="DevconBoothImages.MainWindow"
xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
xmlns:local="clr-namespace:DevconBoothImages"
mc:Ignorable="d"
Title="CodexBoothImages" Height="450" Width="800" WindowState="Maximized">
<Grid>
<Grid.RowDefinitions>
<RowDefinition Height="*"/>
<RowDefinition Height="Auto"/>
</Grid.RowDefinitions>
<Grid Grid.Row="0">
<Grid.ColumnDefinitions>
<ColumnDefinition Width="*"/>
<ColumnDefinition Width="Auto"/>
</Grid.ColumnDefinitions>
<Image Grid.Column="0" Name="Img" />
<StackPanel Grid.Column="1">
<TextBlock Text="Instructions:"/>
<Image Name="ImgInstructions" />
<TextBlock Text="Local CID:"/>
<Image Name="ImgLocalCid" />
<TextBlock Text="TestNet CID:"/>
<Image Name="ImgTestnetCid" />
</StackPanel>
</Grid>
<StackPanel Grid.Row="1">
<TextBlock Name="Txt" HorizontalAlignment="Center" />
<Button Content="Check Codex connections" Click="Button_Click_2" />
<Button Content="Generate image -> Upload to Codex -> Put CID info in clipboard" Padding="10" Click="Button_Click"/>
<Button Content="Put last CID info in clipboard" Padding="10" Click="Button_Click_1"/>
</StackPanel>
</Grid>
</Window>
+160
View File
@@ -0,0 +1,160 @@
using AutoClient;
using CodexOpenApi;
using Logging;
using QRCoder;
using System.Drawing;
using System.Drawing.Imaging;
using System.IO;
using System.Windows;
using System.Windows.Media;
using System.Windows.Media.Imaging;
namespace DevconBoothImages
{
public partial class MainWindow : Window
{
private readonly Configuration config = new Configuration();
private readonly CodexWrapper codexWrapper = new CodexWrapper();
private readonly ImageGenerator imageGenerator = new ImageGenerator(new NullLog());
private string currentLocalCid = string.Empty;
private string currentPublicCid = string.Empty;
public MainWindow()
{
InitializeComponent();
Application.Current.DispatcherUnhandledException += Current_DispatcherUnhandledException;
}
private void Current_DispatcherUnhandledException(object sender, System.Windows.Threading.DispatcherUnhandledExceptionEventArgs e)
{
MessageBox.Show("Unhandled exception: " + e.Exception);
}
private async void Button_Click(object sender, RoutedEventArgs e)
{
// image
Log("Getting image...");
var file = await imageGenerator.Generate();
var filename = Path.Combine(config.WorkingDir, file);
File.Copy(file, filename);
var bmp = new BitmapImage();
bmp.BeginInit();
bmp.UriSource = new Uri(filename);
bmp.EndInit();
Img.Source = bmp;
Log("Uploading...");
// upload
await UploadToCodexes(filename, file);
// clipboard info
InfoToClipboard();
}
private BitmapImage GenerateQr(string text)
{
using (QRCodeGenerator qrGenerator = new QRCodeGenerator())
using (QRCodeData qrCodeData = qrGenerator.CreateQrCode(text, QRCodeGenerator.ECCLevel.Default))
using (PngByteQRCode qrCode = new PngByteQRCode(qrCodeData))
{
byte[] qrCodeImage = qrCode.GetGraphic(7);
using (var ms = new MemoryStream(qrCodeImage))
{
var img = Image.FromStream(ms);
using (var ms2 = new MemoryStream())
{
img.Save(ms2, ImageFormat.Png);
ms.Seek(0, SeekOrigin.Begin);
var bitmapImage = new BitmapImage();
bitmapImage.BeginInit();
bitmapImage.CacheOption = BitmapCacheOption.OnLoad;
bitmapImage.StreamSource = ms2;
bitmapImage.EndInit();
return bitmapImage;
}
}
}
}
private async void Button_Click_1(object sender, RoutedEventArgs e)
{
InfoToClipboard();
}
private async void Button_Click_2(object sender, RoutedEventArgs e)
{
// check codexes
Log("Checking Codex connections...");
await codexWrapper.GetCodexes();
Log("Connections OK");
}
private async Task UploadToCodexes(string filename, string shortName)
{
var codexes = await codexWrapper.GetCodexes();
try
{
currentLocalCid = await UploadFile(filename, shortName, codexes.Local);
currentPublicCid = await UploadFile(filename, shortName, codexes.Testnet);
}
catch (Exception ex)
{
MessageBox.Show("Upload failed: " + ex);
}
Log($"Generated CIDs");
}
private async Task<string> UploadFile(string filename, string shortName, CodexApi codex)
{
using (var fileStream = File.OpenRead(filename))
{
var response = await codex.UploadAsync(
"image/jpeg",
$"attachment; filename=\"{shortName}\"",
fileStream);
if (string.IsNullOrEmpty(response) ||
response.ToLowerInvariant().Contains("unable to store block"))
{
throw new Exception("Unable to upload image. Response empty or error message.");
}
return response;
}
}
private void InfoToClipboard()
{
Clipboard.Clear();
if (string.IsNullOrEmpty(currentLocalCid) || string.IsNullOrEmpty(currentPublicCid))
{
Log("No CIDs were generated! Clipboard cleared.");
return;
}
var nl = Environment.NewLine;
var msg =
$"** Codex@Devcon 💻 Raspberry Pi Challenge **{nl}" +
$"📢 A new image is available. Download it and bring it to the booth!{nl}" +
$"Public Testnet CID: `{currentPublicCid}`{nl}" +
$"Local Devcon network CID: `{currentLocalCid}`{nl}" +
$"Setup instructions: [Here](https://docs.codex.storage){nl}" +
$"Local Devcon network information: [Here](https://github.com/codex-storage/codex-testnet-starter/blob/master/SETUP_DEVCONNET.md)";
Clipboard.SetText(msg);
Log("CID info copied to clipboard. Paste it in Discord plz!");
ImgLocalCid.Source = GenerateQr(currentLocalCid);
ImgTestnetCid.Source = GenerateQr(currentPublicCid);
ImgInstructions.Source = GenerateQr("https://github.com/codex-storage/codex-testnet-starter/blob/master/SETUP_DEVCONNET.md");
}
private void Log(string v)
{
Txt.Text = v;
}
}
}
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
+1 -5
View File
@@ -30,11 +30,7 @@ namespace Core
public IDownloadedLog DownloadLog(RunningContainer container, int? tailLines = null)
{
var workflow = entryPoint.Tools.CreateWorkflow();
var msg = $"Downloading container log for '{container.Name}'";
entryPoint.Tools.GetLog().Log(msg);
var logHandler = new WriteToFileLogHandler(entryPoint.Tools.GetLog(), msg);
workflow.DownloadContainerLog(container, logHandler, tailLines);
return new DownloadedLog(logHandler);
return workflow.DownloadContainerLog(container, tailLines);
}
public string ExecuteContainerCommand(IHasContainer containerSource, string command, params string[] args)
+17 -4
View File
@@ -13,18 +13,21 @@ namespace Core
internal class Http : IHttp
{
private static readonly object httpLock = new object();
private static object lockLock = new object();
private static readonly Dictionary<string, object> httpLocks = new Dictionary<string, object>();
private readonly ILog log;
private readonly ITimeSet timeSet;
private readonly Action<HttpClient> onClientCreated;
private readonly string id;
internal Http(ILog log, ITimeSet timeSet)
: this(log, timeSet, DoNothing)
internal Http(string id, ILog log, ITimeSet timeSet)
: this(id, log, timeSet, DoNothing)
{
}
internal Http(ILog log, ITimeSet timeSet, Action<HttpClient> onClientCreated)
internal Http(string id, ILog log, ITimeSet timeSet, Action<HttpClient> onClientCreated)
{
this.id = id;
this.log = log;
this.timeSet = timeSet;
this.onClientCreated = onClientCreated;
@@ -63,12 +66,22 @@ namespace Core
private T LockRetry<T>(Func<T> operation, Retry retry)
{
var httpLock = GetLock();
lock (httpLock)
{
return retry.Run(operation);
}
}
private object GetLock()
{
lock (lockLock) // I had to.
{
if (!httpLocks.ContainsKey(id)) httpLocks.Add(id, new object());
return httpLocks[id];
}
}
private HttpClient GetClient()
{
var client = new HttpClient();
+9 -9
View File
@@ -27,9 +27,9 @@ namespace Core
public interface IHttpFactoryTool
{
IHttp CreateHttp(Action<HttpClient> onClientCreated);
IHttp CreateHttp(Action<HttpClient> onClientCreated, ITimeSet timeSet);
IHttp CreateHttp();
IHttp CreateHttp(string id, Action<HttpClient> onClientCreated);
IHttp CreateHttp(string id, Action<HttpClient> onClientCreated, ITimeSet timeSet);
IHttp CreateHttp(string id);
}
public interface IFileTool
@@ -58,19 +58,19 @@ namespace Core
log.Prefix = prefix;
}
public IHttp CreateHttp(Action<HttpClient> onClientCreated)
public IHttp CreateHttp(string id, Action<HttpClient> onClientCreated)
{
return CreateHttp(onClientCreated, TimeSet);
return CreateHttp(id, onClientCreated, TimeSet);
}
public IHttp CreateHttp(Action<HttpClient> onClientCreated, ITimeSet ts)
public IHttp CreateHttp(string id, Action<HttpClient> onClientCreated, ITimeSet ts)
{
return new Http(log, ts, onClientCreated);
return new Http(id, log, ts, onClientCreated);
}
public IHttp CreateHttp()
public IHttp CreateHttp(string id)
{
return new Http(log, TimeSet);
return new Http(id, log, TimeSet);
}
public IStartupWorkflow CreateWorkflow(string? namespaceOverride = null)
+2 -2
View File
@@ -34,12 +34,12 @@
{
public TimeSpan HttpCallTimeout()
{
return TimeSpan.FromMinutes(3);
return TimeSpan.FromMinutes(2);
}
public TimeSpan HttpRetryTimeout()
{
return TimeSpan.FromMinutes(10);
return TimeSpan.FromMinutes(5);
}
public TimeSpan HttpCallRetryDelay()
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
+6 -11
View File
@@ -3,12 +3,12 @@
public class GiveRewardsCommand
{
public RewardUsersCommand[] Rewards { get; set; } = Array.Empty<RewardUsersCommand>();
public MarketAverage[] Averages { get; set; } = Array.Empty<MarketAverage>();
public string[] EventsOverview { get; set; } = Array.Empty<string>();
public ChainEventMessage[] EventsOverview { get; set; } = Array.Empty<ChainEventMessage>();
public string[] Errors { get; set; } = Array.Empty<string>();
public bool HasAny()
{
return Rewards.Any() || Averages.Any() || EventsOverview.Any();
return Rewards.Any() || EventsOverview.Any();
}
}
@@ -18,14 +18,9 @@
public string[] UserAddresses { get; set; } = Array.Empty<string>();
}
public class MarketAverage
public class ChainEventMessage
{
public int NumberOfFinished { get; set; }
public int TimeRangeSeconds { get; set; }
public float Price { get; set; }
public float Size { get; set; }
public float Duration { get; set; }
public float Collateral { get; set; }
public float ProofProbability { get; set; }
public ulong BlockNumber { get; set; }
public string Message { get; set; } = string.Empty;
}
}
+35 -41
View File
@@ -7,27 +7,35 @@ namespace FileUtils
{
TrackedFile CreateEmptyFile(string label = "");
TrackedFile GenerateFile(ByteSize size, string label = "");
TrackedFile GenerateFile(Action<IGenerateOption> options, string label = "");
void DeleteAllFiles();
void ScopedFiles(Action action);
T ScopedFiles<T>(Func<T> action);
}
public interface IGenerateOption
{
IGenerateOption Random(ByteSize size);
IGenerateOption StringRepeat(string str, ByteSize size);
IGenerateOption StringRepeat(string str, int times);
IGenerateOption ByteRepeat(byte[] bytes, ByteSize size);
IGenerateOption ByteRepeat(byte[] bytes, int times);
}
public class FileManager : IFileManager
{
public const int ChunkSize = 1024 * 1024 * 100;
private static NumberSource folderNumberSource = new NumberSource(0);
private readonly Random random = new Random();
private static readonly NumberSource folderNumberSource = new NumberSource(0);
private readonly ILog log;
private readonly string rootFolder;
private readonly string folder;
private readonly List<List<TrackedFile>> fileSetStack = new List<List<TrackedFile>>();
public const int ChunkSize = 1024 * 1024 * 100;
public FileManager(ILog log, string rootFolder)
{
folder = Path.Combine(rootFolder, folderNumberSource.GetNextNumber().ToString("D5"));
this.log = log;
this.rootFolder = rootFolder;
}
public TrackedFile CreateEmptyFile(string label = "")
@@ -41,10 +49,15 @@ namespace FileUtils
return result;
}
public TrackedFile GenerateFile(ByteSize size, string label)
public TrackedFile GenerateFile(ByteSize size, string label = "")
{
return GenerateFile(o => o.Random(size), label);
}
public TrackedFile GenerateFile(Action<IGenerateOption> options, string label = "")
{
var sw = Stopwatch.Begin(log);
var result = GenerateRandomFile(size, label);
var result = RunGenerators(options, label);
sw.End($"Generated file {result.Describe()}.");
return result;
}
@@ -100,26 +113,35 @@ namespace FileUtils
if (!Directory.GetFiles(folder).Any()) DeleteDirectory();
}
private TrackedFile GenerateRandomFile(ByteSize size, string label)
private TrackedFile RunGenerators(Action<IGenerateOption> options, string label)
{
var result = CreateEmptyFile(label);
CheckSpaceAvailable(result, size);
var generators = GetGenerators(options);
CheckSpaceAvailable(result, generators.GetRequiredSpace());
GenerateFileBytes(result, size);
using var stream = new FileStream(result.Filename, FileMode.Append);
generators.Run(stream);
return result;
}
private void CheckSpaceAvailable(TrackedFile testFile, ByteSize size)
private GeneratorCollection GetGenerators(Action<IGenerateOption> options)
{
var result = new GeneratorCollection();
options(result);
return result;
}
private void CheckSpaceAvailable(TrackedFile testFile, long requiredSize)
{
var file = new FileInfo(testFile.Filename);
var drive = new DriveInfo(file.Directory!.Root.FullName);
var spaceAvailable = drive.TotalFreeSpace;
if (spaceAvailable < size.SizeInBytes)
if (spaceAvailable < requiredSize)
{
var msg = $"Not enough disk space. " +
$"{Formatter.FormatByteSize(size.SizeInBytes)} required. " +
$"{Formatter.FormatByteSize(requiredSize)} required. " +
$"{Formatter.FormatByteSize(spaceAvailable)} available.";
log.Log(msg);
@@ -127,34 +149,6 @@ namespace FileUtils
}
}
private void GenerateFileBytes(TrackedFile result, ByteSize size)
{
long bytesLeft = size.SizeInBytes;
int chunkSize = ChunkSize;
while (bytesLeft > 0)
{
try
{
var length = Math.Min(bytesLeft, chunkSize);
AppendRandomBytesToFile(result, length);
bytesLeft -= length;
}
catch
{
chunkSize = chunkSize / 2;
if (chunkSize < 1024) throw;
}
}
}
private void AppendRandomBytesToFile(TrackedFile result, long length)
{
var bytes = new byte[length];
random.NextBytes(bytes);
using var stream = new FileStream(result.Filename, FileMode.Append);
stream.Write(bytes, 0, bytes.Length);
}
private void EnsureDirectory()
{
Directory.CreateDirectory(folder);
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
+141
View File
@@ -0,0 +1,141 @@
using System.Text;
using Utils;
namespace FileUtils
{
public class GeneratorCollection : IGenerateOption
{
private readonly List<IGenerator> generators = new List<IGenerator>();
public IGenerateOption ByteRepeat(byte[] bytes, ByteSize size)
{
var times = size.SizeInBytes / bytes.Length;
generators.Add(new ByteRepeater(bytes, times));
return this;
}
public IGenerateOption ByteRepeat(byte[] bytes, int times)
{
generators.Add(new ByteRepeater(bytes, times));
return this;
}
public IGenerateOption Random(ByteSize size)
{
generators.Add(new RandomGenerator(size));
return this;
}
public IGenerateOption StringRepeat(string str, ByteSize size)
{
var times = size.SizeInBytes / str.Length;
generators.Add(new StringRepeater(str, times));
return this;
}
public IGenerateOption StringRepeat(string str, int times)
{
generators.Add(new StringRepeater(str, times));
return this;
}
public void Run(FileStream file)
{
foreach (var generator in generators)
{
generator.Generate(file);
}
}
public long GetRequiredSpace()
{
return generators.Sum(g => g.GetRequiredSpace());
}
}
public interface IGenerator
{
void Generate(FileStream file);
long GetRequiredSpace();
}
public class ByteRepeater : IGenerator
{
private readonly byte[] bytes;
private readonly long times;
public ByteRepeater(byte[] bytes, long times)
{
this.bytes = bytes;
this.times = times;
}
public void Generate(FileStream file)
{
for (var i = 0; i < times; i++)
{
file.Write(bytes, 0, bytes.Length);
}
}
public long GetRequiredSpace()
{
return bytes.Length * times;
}
}
public class StringRepeater : IGenerator
{
private readonly string str;
private readonly long times;
public StringRepeater(string str, long times)
{
this.str = str;
this.times = times;
}
public void Generate(FileStream file)
{
using var writer = new StreamWriter(file);
for (var i = 0; i < times; i++)
{
writer.Write(str);
}
}
public long GetRequiredSpace()
{
return Encoding.ASCII.GetBytes(str).Length * times;
}
}
public class RandomGenerator : IGenerator
{
private readonly Random random = new Random();
private readonly ByteSize size;
public RandomGenerator(ByteSize size)
{
this.size = size;
}
public void Generate(FileStream file)
{
var bytesLeft = size.SizeInBytes;
while (bytesLeft > 0)
{
var size = Math.Min(bytesLeft, FileManager.ChunkSize);
var bytes = new byte[size];
random.NextBytes(bytes);
file.Write(bytes, 0, bytes.Length);
bytesLeft -= size;
}
}
public long GetRequiredSpace()
{
return size.SizeInBytes;
}
}
}
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -1,13 +1,15 @@
using KubernetesWorkflow;
using Logging;
using Logging;
namespace Core
namespace KubernetesWorkflow
{
public interface IDownloadedLog
{
void IterateLines(Action<string> action);
string ContainerName { get; }
void IterateLines(Action<string> action, params string[] thatContain);
string[] GetLinesContaining(string expectedString);
string[] FindLinesThatContain(params string[] tags);
string GetFilepath();
void DeleteFile();
}
@@ -15,12 +17,15 @@ namespace Core
{
private readonly LogFile logFile;
internal DownloadedLog(WriteToFileLogHandler logHandler)
internal DownloadedLog(WriteToFileLogHandler logHandler, string containerName)
{
logFile = logHandler.LogFile;
ContainerName = containerName;
}
public void IterateLines(Action<string> action)
public string ContainerName { get; }
public void IterateLines(Action<string> action, params string[] thatContain)
{
using var file = File.OpenRead(logFile.FullFilename);
using var streamReader = new StreamReader(file);
@@ -28,7 +33,10 @@ namespace Core
var line = streamReader.ReadLine();
while (line != null)
{
action(line);
if (thatContain.All(line.Contains))
{
action(line);
}
line = streamReader.ReadLine();
}
}
@@ -72,6 +80,11 @@ namespace Core
return result.ToArray();
}
public string GetFilepath()
{
return logFile.FullFilename;
}
public void DeleteFile()
{
File.Delete(logFile.FullFilename);
+1 -1
View File
@@ -24,8 +24,8 @@
.Replace("]", "-")
.Replace(",", "-");
result = result.Trim('-');
if (result.Length > maxLength) result = result.Substring(0, maxLength);
result = result.Trim('-');
return result;
}
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<RootNamespace>KubernetesWorkflow</RootNamespace>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
@@ -1,4 +1,5 @@
using Newtonsoft.Json.Linq;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
namespace KubernetesWorkflow.Recipe
{
@@ -21,14 +22,13 @@ namespace KubernetesWorkflow.Recipe
var typeName = GetTypeName(typeof(T));
var userData = Additionals.SingleOrDefault(a => a.Type == typeName);
if (userData == null) return default;
var jobject = (JObject)userData.UserData;
return jobject.ToObject<T>();
return JsonConvert.DeserializeObject<T>(userData.UserData);
}
private static Additional ConvertToAdditional(object userData)
{
var typeName = GetTypeName(userData.GetType());
return new Additional(typeName, userData);
return new Additional(typeName, JsonConvert.SerializeObject(userData));
}
private static string GetTypeName(Type type)
@@ -41,13 +41,13 @@ namespace KubernetesWorkflow.Recipe
public class Additional
{
public Additional(string type, object userData)
public Additional(string type, string userData)
{
Type = type;
UserData = userData;
}
public string Type { get; }
public object UserData { get; }
public string UserData { get; }
}
}
@@ -2,8 +2,9 @@
{
public class ContainerRecipe
{
public ContainerRecipe(int number, string? nameOverride, string image, ContainerResources resources, SchedulingAffinity schedulingAffinity, CommandOverride commandOverride, bool setCriticalPriority, Port[] exposedPorts, Port[] internalPorts, EnvVar[] envVars, PodLabels podLabels, PodAnnotations podAnnotations, VolumeMount[] volumes, ContainerAdditionals additionals)
public ContainerRecipe(DateTime recipeCreatedUtc, int number, string? nameOverride, string image, ContainerResources resources, SchedulingAffinity schedulingAffinity, CommandOverride commandOverride, bool setCriticalPriority, Port[] exposedPorts, Port[] internalPorts, EnvVar[] envVars, PodLabels podLabels, PodAnnotations podAnnotations, VolumeMount[] volumes, ContainerAdditionals additionals)
{
RecipeCreatedUtc = recipeCreatedUtc;
Number = number;
NameOverride = nameOverride;
Image = image;
@@ -31,6 +32,7 @@
if (exposedPorts.Any(p => string.IsNullOrEmpty(p.Tag))) throw new Exception("Port tags are required for all exposed ports.");
}
public DateTime RecipeCreatedUtc { get; }
public string Name { get; }
public int Number { get; }
public string? NameOverride { get; }
@@ -25,7 +25,7 @@ namespace KubernetesWorkflow.Recipe
Initialize(config);
var recipe = new ContainerRecipe(containerNumber, config.NameOverride, Image, resources, schedulingAffinity, commandOverride, setCriticalPriority,
var recipe = new ContainerRecipe(DateTime.UtcNow, containerNumber, config.NameOverride, Image, resources, schedulingAffinity, commandOverride, setCriticalPriority,
exposedPorts.ToArray(),
internalPorts.ToArray(),
envVars.ToArray(),
@@ -16,6 +16,7 @@ namespace KubernetesWorkflow
CrashWatcher CreateCrashWatcher(RunningContainer container);
void Stop(RunningPod pod, bool waitTillStopped);
void DownloadContainerLog(RunningContainer container, ILogHandler logHandler, int? tailLines = null, bool? previous = null);
IDownloadedLog DownloadContainerLog(RunningContainer container, int? tailLines = null, bool? previous = null);
string ExecuteCommand(RunningContainer container, string command, params string[] args);
void DeleteNamespace(bool wait);
void DeleteNamespacesStartingWith(string namespacePrefix, bool wait);
@@ -60,7 +61,7 @@ namespace KubernetesWorkflow
var startResult = controller.BringOnline(recipes, location);
var containers = CreateContainers(startResult, recipes, startupConfig);
var rc = new RunningPod(startupConfig, startResult, containers);
var rc = new RunningPod(Guid.NewGuid().ToString(), startupConfig, startResult, containers);
cluster.Configuration.Hooks.OnContainersStarted(rc);
if (startResult.ExternalService != null)
@@ -99,11 +100,19 @@ namespace KubernetesWorkflow
public void Stop(RunningPod runningPod, bool waitTillStopped)
{
if (runningPod.IsStopped) return;
foreach (var c in runningPod.Containers)
{
c.StopLog = DownloadContainerLog(c);
}
runningPod.IsStopped = true;
K8s(controller =>
{
controller.Stop(runningPod.StartResult, waitTillStopped);
cluster.Configuration.Hooks.OnContainersStopped(runningPod);
});
cluster.Configuration.Hooks.OnContainersStopped(runningPod);
}
public void DownloadContainerLog(RunningContainer container, ILogHandler logHandler, int? tailLines = null, bool? previous = null)
@@ -114,6 +123,20 @@ namespace KubernetesWorkflow
});
}
public IDownloadedLog DownloadContainerLog(RunningContainer container, int? tailLines = null, bool? previous = null)
{
var msg = $"Downloading container log for '{container.Name}'";
log.Log(msg);
var logHandler = new WriteToFileLogHandler(log, msg);
K8s(controller =>
{
controller.DownloadPodLog(container, logHandler, tailLines, previous);
});
return new DownloadedLog(logHandler, container.Name);
}
public string ExecuteCommand(RunningContainer container, string command, params string[] args)
{
return K8s(controller =>
@@ -147,7 +170,7 @@ namespace KubernetesWorkflow
var addresses = CreateContainerAddresses(startResult, r);
log.Debug($"{r}={name} -> container addresses: {string.Join(Environment.NewLine, addresses.Select(a => a.ToString()))}");
return new RunningContainer(name, r, addresses);
return new RunningContainer(Guid.NewGuid().ToString(), name, r, addresses);
}).ToArray();
}
@@ -7,27 +7,29 @@ namespace KubernetesWorkflow.Types
{
public class RunningContainer
{
public RunningContainer(string name, ContainerRecipe recipe, ContainerAddress[] addresses)
public RunningContainer(string id, string name, ContainerRecipe recipe, ContainerAddress[] addresses)
{
Id = id;
Name = name;
Recipe = recipe;
Addresses = addresses;
}
public string Id { get; }
public string Name { get; }
public ContainerRecipe Recipe { get; }
public ContainerAddress[] Addresses { get; }
public IDownloadedLog? StopLog { get; internal set; }
[JsonIgnore]
public RunningPod RunningPod { get; internal set; } = null!;
public Address GetAddress(ILog log, string portTag)
public Address GetAddress(string portTag)
{
var addresses = Addresses.Where(a => a.PortTag == portTag).ToArray();
if (!addresses.Any()) throw new Exception("No addresses found for portTag: " + portTag);
var select = SelectAddress(addresses);
log.Debug($"Container '{Name}' selected for tag '{portTag}' address: '{select}'");
return select.Address;
}
@@ -50,5 +52,21 @@ namespace KubernetesWorkflow.Types
}
throw new Exception("Running location not known.");
}
public override string ToString()
{
return Name;
}
public override bool Equals(object? obj)
{
return obj is RunningContainer container &&
Id == container.Id;
}
public override int GetHashCode()
{
return HashCode.Combine(Id);
}
}
}
@@ -4,8 +4,9 @@ namespace KubernetesWorkflow.Types
{
public class RunningPod
{
public RunningPod(StartupConfig startupConfig, StartResult startResult, RunningContainer[] containers)
public RunningPod(string id, StartupConfig startupConfig, StartResult startResult, RunningContainer[] containers)
{
Id = id;
StartupConfig = startupConfig;
StartResult = startResult;
Containers = containers;
@@ -13,6 +14,7 @@ namespace KubernetesWorkflow.Types
foreach (var c in containers) c.RunningPod = this;
}
public string Id { get; }
public StartupConfig StartupConfig { get; }
public StartResult StartResult { get; }
public RunningContainer[] Containers { get; }
@@ -23,10 +25,30 @@ namespace KubernetesWorkflow.Types
get { return $"'{string.Join("&", Containers.Select(c => c.Name).ToArray())}'"; }
}
[JsonIgnore]
public bool IsStopped { get; internal set; }
public string Describe()
{
return string.Join(",", Containers.Select(c => c.Name));
}
public override bool Equals(object? obj)
{
return obj is RunningPod pod &&
Id == pod.Id;
}
public override int GetHashCode()
{
return HashCode.Combine(Id);
}
public override string ToString()
{
if (IsStopped) return Name + " (*)";
return Name;
}
}
public static class RunningContainersExtensions
+1 -1
View File
@@ -77,7 +77,7 @@ namespace Logging
return new LogFile($"{GetFullName()}_{GetSubfileNumber()}", ext);
}
private string ApplyReplacements(string str)
protected string ApplyReplacements(string str)
{
if (IsDebug) return str;
foreach (var replacement in replacements)
+1 -1
View File
@@ -9,7 +9,7 @@
public override void Log(string message)
{
Console.WriteLine(message);
Console.WriteLine(ApplyReplacements(message));
}
}
}
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<RootNamespace>Logging</RootNamespace>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
@@ -29,7 +29,8 @@ namespace NethereumWorkflow.BlockUtils
public ulong? GetHighestBlockNumberBefore(DateTime moment)
{
bounds.Initialize();
if (moment <= bounds.Genesis.Utc) return null;
if (moment < bounds.Genesis.Utc) return null;
if (moment == bounds.Genesis.Utc) return bounds.Genesis.BlockNumber;
if (moment >= bounds.Current.Utc) return bounds.Current.BlockNumber;
return Log(() => Search(bounds.Genesis, bounds.Current, moment, HighestBeforeSelector));
@@ -38,7 +39,8 @@ namespace NethereumWorkflow.BlockUtils
public ulong? GetLowestBlockNumberAfter(DateTime moment)
{
bounds.Initialize();
if (moment >= bounds.Current.Utc) return null;
if (moment > bounds.Current.Utc) return null;
if (moment == bounds.Current.Utc) return bounds.Current.BlockNumber;
if (moment <= bounds.Genesis.Utc) return bounds.Genesis.BlockNumber;
return Log(()=> Search(bounds.Genesis, bounds.Current, moment, LowestAfterSelector)); ;
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<RootNamespace>NethereumWorkflow</RootNamespace>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
@@ -0,0 +1,61 @@
namespace OverwatchTranscript
{
public class ActionQueue
{
// Using ConcurrentQueue<> here would make this process slower.
private readonly object queueLock = new object();
private readonly AutoResetEvent signal = new AutoResetEvent(false);
private List<Action> queue = new List<Action>();
private Task queueWorker = null!;
private bool stopping = false;
public void Start()
{
queueWorker = Task.Run(QueueWorker);
}
public int Count { get; private set; }
public void StopAndJoin()
{
stopping = true;
queueWorker.Wait();
if (queue.Count > 0) throw new Exception("not all acions handled");
queueWorker.Dispose();
}
public void Add(Action action)
{
if (stopping) throw new Exception("queue stopping");
lock (queueLock)
{
queue.Add(action);
Count = queue.Count;
}
signal.Set();
}
private void QueueWorker()
{
while (true)
{
signal.WaitOne(10);
List<Action> work = null!;
lock (queueLock)
{
work = queue;
queue = new List<Action>();
Count = 0;
}
if (stopping && !work.Any()) return;
foreach (var action in work)
{
action();
}
}
}
}
}
+110
View File
@@ -0,0 +1,110 @@
using Logging;
using System.Collections.Concurrent;
namespace OverwatchTranscript
{
public class BucketSet
{
private const int numberOfActiveBuckets = 10;
private readonly ILog log;
private readonly string workingDir;
private readonly object _bucketLock = new object();
private readonly List<EventBucketWriter> fullBuckets = new List<EventBucketWriter>();
private readonly List<EventBucketWriter> activeBuckets = new List<EventBucketWriter>();
private readonly ActionQueue queue = new ActionQueue();
private int activeBucketIndex = 0;
private bool closed = false;
private string internalErrors = string.Empty;
public BucketSet(ILog log, string workingDir)
{
this.log = log;
this.workingDir = workingDir;
for (var i = 0; i < numberOfActiveBuckets;i++)
{
AddNewBucket();
}
queue.Start();
}
public void Add(DateTime utc, object payload)
{
if (closed) throw new Exception("Buckets already closed!");
queue.Add(() => AddInternal(utc, payload));
if (queue.Count > 1000)
{
Thread.Sleep(1);
}
}
public IFinalizedBucket[] FinalizeBuckets()
{
closed = true;
queue.StopAndJoin();
if (IsEmpty()) throw new Exception("No entries have been added.");
if (!string.IsNullOrEmpty(internalErrors)) throw new Exception(internalErrors);
var buckets = fullBuckets.Concat(activeBuckets).ToArray();
log.Debug($"Finalizing {buckets.Length} buckets...");
var finalized = new ConcurrentBag<IFinalizedBucket>();
var tasks = Parallel.ForEach(buckets, b => finalized.Add(b.FinalizeBucket()));
if (!tasks.IsCompleted) throw new Exception("Failed to finalize buckets: " + tasks);
return finalized.ToArray();
}
private bool IsEmpty()
{
return fullBuckets.All(b => b.Count == 0) && activeBuckets.All(b => b.Count == 0);
}
private void AddInternal(DateTime utc, object payload)
{
try
{
lock (_bucketLock)
{
var current = activeBuckets[activeBucketIndex];
current.Add(utc, payload);
activeBucketIndex = (activeBucketIndex + 1) % numberOfActiveBuckets;
if (current.IsFull)
{
log.Debug("Bucket is full. New bucket...");
fullBuckets.Add(current);
activeBuckets.Remove(current);
AddNewBucket();
}
}
}
catch (Exception ex)
{
internalErrors += ex.ToString();
log.Error(ex.ToString());
}
}
private static int bucketSizeIndex = 0;
private static int[] bucketSizes = new[]
{
10000,
15000,
20000,
};
private void AddNewBucket()
{
lock (_bucketLock)
{
var size = bucketSizes[bucketSizeIndex];
bucketSizeIndex = (bucketSizeIndex + 1) % bucketSizes.Length;
activeBuckets.Add(new EventBucketWriter(log, Path.Combine(workingDir, Guid.NewGuid().ToString()), size));
}
}
}
}
@@ -0,0 +1,151 @@
using Logging;
using Newtonsoft.Json;
using System.Collections.Concurrent;
namespace OverwatchTranscript
{
public interface IFinalizedBucket
{
bool IsEmpty { get; }
void Update();
DateTime? SeeTopUtc();
BucketTop? TakeTop();
}
public class BucketTop
{
public BucketTop(DateTime utc, OverwatchEvent[] events)
{
Utc = utc;
Events = events;
}
public DateTime Utc { get; }
public OverwatchEvent[] Events { get; }
}
public class EventBucketReader : IFinalizedBucket
{
private readonly string bucketFile;
private readonly ConcurrentQueue<BucketTop> topQueue = new ConcurrentQueue<BucketTop>();
private readonly AutoResetEvent itemDequeued = new AutoResetEvent(false);
private readonly AutoResetEvent itemEnqueued = new AutoResetEvent(false);
private bool sourceIsEmpty;
public EventBucketReader(ILog log, string bucketFile)
{
this.bucketFile = bucketFile;
if (!File.Exists(bucketFile)) throw new Exception("Doesn't exist: " + bucketFile);
log.Debug("Read Bucket open: " + bucketFile);
Task.Run(ReadBucket);
}
public bool IsEmpty { get; private set; }
public void Update()
{
if (IsEmpty) return;
while (topQueue.Count == 0)
{
UpdateIsEmpty();
if (IsEmpty) return;
itemDequeued.Set();
itemEnqueued.WaitOne(200);
}
}
public DateTime? SeeTopUtc()
{
if (IsEmpty) return null;
if (topQueue.TryPeek(out BucketTop? top))
{
return top.Utc;
}
return null;
}
public BucketTop? TakeTop()
{
if (IsEmpty) return null;
if (topQueue.TryDequeue(out BucketTop? top))
{
itemDequeued.Set();
return top;
}
return null;
}
private void ReadBucket()
{
using var file = File.OpenRead(bucketFile);
using var reader = new StreamReader(file);
while (true)
{
while (topQueue.Count < 5)
{
var top = CreateNewTop(reader);
if (top != null)
{
topQueue.Enqueue(top);
itemEnqueued.Set();
}
else
{
sourceIsEmpty = true;
UpdateIsEmpty();
return;
}
}
itemDequeued.Reset();
itemDequeued.WaitOne(5000);
}
}
private void UpdateIsEmpty()
{
var allEmpty = sourceIsEmpty && topQueue.IsEmpty;
if (!IsEmpty && allEmpty)
{
File.Delete(bucketFile);
IsEmpty = true;
}
}
private EventBucketEntry? nextEntry = null;
private BucketTop? CreateNewTop(StreamReader reader)
{
if (nextEntry == null)
{
nextEntry = ReadEntry(reader);
if (nextEntry == null) return null;
}
var topEntry = nextEntry;
var entries = new List<EventBucketEntry>
{
topEntry
};
nextEntry = ReadEntry(reader);
while (nextEntry != null && nextEntry.Utc == topEntry.Utc)
{
entries.Add(nextEntry);
nextEntry = ReadEntry(reader);
}
return new BucketTop(topEntry.Utc, entries.Select(e => e.Event).ToArray());
}
private EventBucketEntry? ReadEntry(StreamReader reader)
{
var line = reader.ReadLine();
if (string.IsNullOrEmpty(line)) return null;
return JsonConvert.DeserializeObject<EventBucketEntry>(line);
}
}
}
@@ -0,0 +1,113 @@
using Logging;
using Newtonsoft.Json;
namespace OverwatchTranscript
{
public class EventBucketWriter
{
private const int MaxBuffer = 1000;
private readonly object _lock = new object();
private bool closed = false;
private readonly ILog log;
private readonly string bucketFile;
private readonly int maxCount;
private readonly List<EventBucketEntry> buffer = new List<EventBucketEntry>();
public EventBucketWriter(ILog log, string bucketFile, int maxCount)
{
this.log = log;
this.bucketFile = bucketFile;
this.maxCount = maxCount;
if (File.Exists(bucketFile)) throw new Exception("Already exists");
log.Debug("Write Bucket open: " + bucketFile);
}
public int Count { get; private set; }
public bool IsFull { get; private set; }
public void Add(DateTime utc, object payload)
{
lock (_lock)
{
if (closed) throw new Exception("Already closed");
AddToBuffer(utc, payload);
BufferToFile(emptyBuffer: false);
}
}
public IFinalizedBucket FinalizeBucket()
{
lock (_lock)
{
closed = true;
BufferToFile(emptyBuffer: true);
SortFileByTimestamps();
}
log.Debug($"Finalized bucket with {Count} entries");
return new EventBucketReader(log, bucketFile);
}
public override string ToString()
{
return $"EventBucket: " + Count;
}
private void AddToBuffer(DateTime utc, object payload)
{
var typeName = payload.GetType().FullName;
if (string.IsNullOrEmpty(typeName)) throw new Exception("Empty typename for payload");
if (utc == default) throw new Exception("DateTimeUtc not set");
var entry = new EventBucketEntry
{
Utc = utc,
Event = new OverwatchEvent
{
Type = typeName,
Payload = Json.Serialize(payload)
}
};
Count++;
IsFull = Count > maxCount;
buffer.Add(entry);
}
private void BufferToFile(bool emptyBuffer)
{
if (emptyBuffer || buffer.Count > MaxBuffer)
{
using var file = File.Open(bucketFile, FileMode.Append);
using var writer = new StreamWriter(file);
foreach (var entry in buffer)
{
writer.WriteLine(Json.Serialize(entry));
}
log.Debug($"Bucket wrote {buffer.Count} entries to file.");
buffer.Clear();
}
}
private void SortFileByTimestamps()
{
var lines = File.ReadAllLines(bucketFile);
var entries = lines.Select(Json.Deserialize<EventBucketEntry>)
.Cast<EventBucketEntry>()
.OrderBy(e => e.Utc)
.ToArray();
File.Delete(bucketFile);
File.WriteAllLines(bucketFile, entries.Select(e => Json.Serialize(e)));
}
}
[Serializable]
public class EventBucketEntry
{
public DateTime Utc { get; set; }
public OverwatchEvent Event { get; set; } = new();
}
}
+27
View File
@@ -0,0 +1,27 @@
using Newtonsoft.Json;
using System.Globalization;
namespace OverwatchTranscript
{
public static class Json
{
private static JsonSerializerSettings settings = new JsonSerializerSettings
{
Formatting = Formatting.None,
NullValueHandling = NullValueHandling.Ignore,
Culture = CultureInfo.InvariantCulture,
DateFormatHandling = DateFormatHandling.IsoDateFormat,
FloatFormatHandling = FloatFormatHandling.Symbol
};
public static string Serialize(object obj, Formatting formatting = Formatting.None)
{
return JsonConvert.SerializeObject(obj, formatting, settings);
}
public static T Deserialize<T>(string json)
{
return JsonConvert.DeserializeObject<T>(json)!;
}
}
}
+56
View File
@@ -0,0 +1,56 @@
namespace OverwatchTranscript
{
[Serializable]
public class OverwatchTranscript
{
public OverwatchHeader Header { get; set; } = new();
public OverwatchMomentReference[] MomentReferences { get; set; } = Array.Empty<OverwatchMomentReference>();
}
[Serializable]
public class OverwatchMomentReference
{
public string MomentsFile { get; set; } = string.Empty;
public int NumberOfMoments { get; set; }
public int NumberOfEvents { get; set; }
public DateTime EarliestUtc { get; set; }
public DateTime LatestUtc { get; set; }
}
[Serializable]
public class OverwatchHeader
{
public OverwatchCommonHeader Common { get; set; } = new();
public OverwatchHeaderEntry[] Entries { get; set; } = Array.Empty<OverwatchHeaderEntry>();
}
[Serializable]
public class OverwatchCommonHeader
{
public long NumberOfMoments { get; set; }
public long NumberOfEvents { get; set; }
public DateTime EarliestUtc { get; set; }
public DateTime LatestUtc { get; set; }
}
[Serializable]
public class OverwatchHeaderEntry
{
public string Key { get; set; } = string.Empty;
public string Value { get; set; } = string.Empty;
}
[Serializable]
public class OverwatchMoment
{
public DateTime Utc { get; set; }
public OverwatchEvent[] Events { get; set; } = Array.Empty<OverwatchEvent>();
}
[Serializable]
public class OverwatchEvent
{
public string Type { get; set; } = string.Empty;
public string Payload { get; set; } = string.Empty;
}
}
@@ -0,0 +1,104 @@
using Newtonsoft.Json;
namespace OverwatchTranscript
{
public class MomentReader
{
private readonly OverwatchTranscript model;
private readonly string workingDir;
private int referenceIndex = 0;
private int momentsRead = 0;
private OpenReference currentRef;
public MomentReader(OverwatchTranscript model, string workingDir)
{
this.model = model;
this.workingDir = workingDir;
currentRef = CreateOpenReference();
}
public OverwatchMoment? Next()
{
if (referenceIndex >= model.MomentReferences.Length) return null;
var moment = currentRef.ReadNext();
if (moment == null)
{
Close();
// This reference file ran out.
// The number of moments read should match exactly the number of moments
// describe in the reference. If not, error:
var expected = model.MomentReferences[referenceIndex].NumberOfMoments;
if (momentsRead != expected)
{
throw new Exception("Number of moments read from referenced file does not match number of moments value in model. " +
$"Reads: { momentsRead} - model.MomentReferences[{referenceIndex}].NumberOfMoment: {expected}");
}
referenceIndex++;
if (referenceIndex < model.MomentReferences.Length)
{
// Proceed to next reference file.
currentRef = CreateOpenReference();
momentsRead = 0;
return Next();
}
else
{
// That was the last one.
return null;
}
}
else
{
momentsRead++;
return moment;
}
}
public void Close()
{
if (currentRef != null)
{
currentRef.Close();
currentRef = null!;
}
}
private OpenReference CreateOpenReference()
{
var filepath = Path.Combine(workingDir, model.MomentReferences[referenceIndex].MomentsFile);
return new OpenReference(filepath);
}
private class OpenReference
{
private readonly FileStream file;
private readonly StreamReader reader;
public OpenReference(string filePath)
{
file = File.OpenRead(filePath);
reader = new StreamReader(file);
}
public OverwatchMoment? ReadNext()
{
var line = reader.ReadLine();
if (string.IsNullOrEmpty(line)) return null;
return JsonConvert.DeserializeObject<OverwatchMoment>(line);
}
public void Close()
{
reader.Close();
file.Close();
reader.Dispose();
file.Dispose();
}
}
}
}
@@ -0,0 +1,153 @@
using Logging;
using Newtonsoft.Json;
namespace OverwatchTranscript
{
public class MomentReferenceBuilder
{
private const int MaxMomentsPerReference = 10000;
private readonly ILog log;
private readonly string workingDir;
public MomentReferenceBuilder(ILog log, string workingDir)
{
this.log = log;
this.workingDir = workingDir;
}
public OverwatchMomentReference[] Build(IFinalizedBucket[] finalizedBuckets)
{
var result = new List<OverwatchMomentReference>();
var currentBuilder = new Builder(log, workingDir);
var buckets = finalizedBuckets.ToList();
log.Debug($"Building references for {buckets.Count} buckets.");
while (buckets.Any())
{
foreach (var b in buckets) b.Update();
buckets.RemoveAll(b => b.IsEmpty);
if (!buckets.Any()) break;
var earliestUtc = GetEarliestUtc(buckets);
if (earliestUtc == null) continue;
var tops = CollectAllTopsForUtc(earliestUtc.Value, buckets);
var moment = ConvertTopsToMoment(tops);
currentBuilder.Add(moment);
if (currentBuilder.NumberOfMoments == MaxMomentsPerReference)
{
result.Add(currentBuilder.Build());
currentBuilder = new Builder(log, workingDir);
}
}
if (currentBuilder.NumberOfMoments > 0)
{
result.Add(currentBuilder.Build());
}
return result.ToArray();
}
private OverwatchMoment ConvertTopsToMoment(List<BucketTop> tops)
{
var discintUtc = tops.Select(e => e.Utc).Distinct().ToArray();
if (discintUtc.Length != 1) throw new Exception("UTC mixing in moment construction.");
return new OverwatchMoment
{
Utc = tops[0].Utc,
Events = tops.SelectMany(e => e.Events).ToArray()
};
}
private List<BucketTop> CollectAllTopsForUtc(DateTime earliestUtc, List<IFinalizedBucket> buckets)
{
var result = new List<BucketTop>();
foreach (var bucket in buckets)
{
if (bucket.IsEmpty) continue;
var utc = bucket.SeeTopUtc();
if (utc == null) continue;
if (utc.Value == earliestUtc)
{
var top = bucket.TakeTop();
if (top == null) throw new Exception("top was null after top utc was not");
result.Add(top);
}
}
return result;
}
private DateTime? GetEarliestUtc(List<IFinalizedBucket> buckets)
{
var earliest = DateTime.MaxValue;
foreach (var bucket in buckets)
{
var utc = bucket.SeeTopUtc();
if (utc == null) return null;
if (utc.Value < earliest) earliest = utc.Value;
}
return earliest;
}
public class Builder
{
private readonly ILog log;
private readonly string workingDir;
private OverwatchMomentReference reference;
private readonly ActionQueue queue = new ActionQueue();
public Builder(ILog log, string workingDir)
{
reference = new OverwatchMomentReference
{
MomentsFile = Guid.NewGuid().ToString(),
EarliestUtc = DateTime.MaxValue,
LatestUtc = DateTime.MinValue,
NumberOfEvents = 0,
NumberOfMoments = 0,
};
this.log = log;
this.workingDir = workingDir;
queue.Start();
}
public int NumberOfMoments => reference.NumberOfMoments;
public void Add(OverwatchMoment moment)
{
if (moment.Utc < reference.EarliestUtc) reference.EarliestUtc = moment.Utc;
if (moment.Utc > reference.LatestUtc) reference.LatestUtc = moment.Utc;
reference.NumberOfMoments++;
reference.NumberOfEvents += moment.Events.Length;
var filePath = Path.Combine(workingDir, reference.MomentsFile);
queue.Add(() =>
{
File.AppendAllLines(filePath, new[]
{
Json.Serialize(moment)
});
});
}
public OverwatchMomentReference Build()
{
queue.StopAndJoin();
log.Debug($"Created reference with {reference.NumberOfMoments} moments and {reference.NumberOfEvents} events...");
var result = reference;
reference = null!;
return result;
}
}
}
}
@@ -0,0 +1,17 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Logging\Logging.csproj" />
</ItemGroup>
</Project>
@@ -0,0 +1,23 @@
using Logging;
namespace OverwatchTranscript
{
public static class Transcript
{
public static ITranscriptWriter NewWriter(ILog log)
{
log = new LogPrefixer(log, "(TranscriptWriter) ");
return new TranscriptWriter(log, NewWorkDir());
}
public static ITranscriptReader NewReader(string transcriptFile)
{
return new TranscriptReader(NewWorkDir(), transcriptFile);
}
private static string NewWorkDir()
{
return Path.Combine(Path.GetTempPath(), Guid.NewGuid().ToString());
}
}
}
@@ -0,0 +1,8 @@
namespace OverwatchTranscript
{
public static class TranscriptConstants
{
public const string TranscriptFilename = "transcript.json";
public const string ArtifactFolderName = "artifacts";
}
}
@@ -0,0 +1,264 @@
using Newtonsoft.Json;
using System.IO;
using System;
using System.IO.Compression;
using System.Linq;
using System.Collections.Generic;
using System.Collections.Concurrent;
namespace OverwatchTranscript
{
public interface ITranscriptReader
{
OverwatchCommonHeader Header { get; }
T GetHeader<T>(string key);
void AddMomentHandler(Action<ActivateMoment> handler);
void AddEventHandler<T>(Action<ActivateEvent<T>> handler);
bool Next();
void Close();
}
public class TranscriptReader : ITranscriptReader
{
private readonly object handlersLock = new object();
private readonly string transcriptFile;
private readonly string artifactsFolder;
private readonly List<Action<ActivateMoment>> momentHandlers = new List<Action<ActivateMoment>>();
private readonly Dictionary<string, List<Action<ActivateMoment, string>>> eventHandlers = new Dictionary<string, List<Action<ActivateMoment, string>>>();
private readonly string workingDir;
private readonly OverwatchTranscript model;
private bool closed;
private long momentCounter;
private readonly ConcurrentQueue<OverwatchMoment> queue = new ConcurrentQueue<OverwatchMoment>();
private readonly Task queueFiller;
public TranscriptReader(string workingDir, string inputFilename)
{
closed = false;
this.workingDir = workingDir;
transcriptFile = Path.Combine(workingDir, TranscriptConstants.TranscriptFilename);
artifactsFolder = Path.Combine(workingDir, TranscriptConstants.ArtifactFolderName);
if (!Directory.Exists(workingDir)) Directory.CreateDirectory(workingDir);
if (File.Exists(transcriptFile) || Directory.Exists(artifactsFolder)) throw new Exception("workingdir not clean");
model = LoadModel(inputFilename);
queueFiller = Task.Run(() => FillQueue(model, workingDir));
}
public OverwatchCommonHeader Header
{
get
{
CheckClosed();
return model.Header.Common;
}
}
public T GetHeader<T>(string key)
{
CheckClosed();
var value = model.Header.Entries.First(e => e.Key == key).Value;
return JsonConvert.DeserializeObject<T>(value)!;
}
public void AddMomentHandler(Action<ActivateMoment> handler)
{
CheckClosed();
lock (handlersLock)
{
momentHandlers.Add(handler);
}
}
public void AddEventHandler<T>(Action<ActivateEvent<T>> handler)
{
CheckClosed();
var typeName = typeof(T).FullName;
if (string.IsNullOrEmpty(typeName)) throw new Exception("Empty typename for payload");
lock (handlersLock)
{
if (eventHandlers.ContainsKey(typeName))
{
eventHandlers[typeName].Add(CreateEventAction(handler));
}
else
{
eventHandlers.Add(typeName, new List<Action<ActivateMoment, string>>
{
CreateEventAction(handler)
});
}
}
}
private readonly object nextLock = new object();
private OverwatchMoment? moment = null;
private OverwatchMoment? next = null;
public bool Next()
{
CheckClosed();
OverwatchMoment? m = null;
TimeSpan? duration = null;
lock (nextLock)
{
if (next == null)
{
if (!queue.TryDequeue(out moment)) return false;
queue.TryDequeue(out next);
}
else
{
moment = next;
next = null;
queue.TryDequeue(out next);
}
m = moment;
duration = GetMomentDuration();
}
ActivateMoment(moment, duration);
return true;
}
public void Close()
{
CheckClosed();
closed = true;
queueFiller.Wait();
Directory.Delete(workingDir, true);
}
private Action<ActivateMoment, string> CreateEventAction<T>(Action<ActivateEvent<T>> handler)
{
return (m, s) =>
{
handler(new ActivateEvent<T>(m, JsonConvert.DeserializeObject<T>(s)!));
};
}
private void FillQueue(OverwatchTranscript model, string workingDir)
{
var reader = new MomentReader(model, workingDir);
while (true)
{
if (closed)
{
reader.Close();
return;
}
while (queue.Count < 10)
{
var moment = reader.Next();
if (moment == null)
{
reader.Close();
return;
}
queue.Enqueue(moment);
}
Thread.Sleep(1);
}
}
private TimeSpan? GetMomentDuration()
{
if (moment == null) return null;
if (next == null) return null;
return next.Utc - moment.Utc;
}
private void ActivateMoment(OverwatchMoment moment, TimeSpan? duration)
{
var m = new ActivateMoment(moment.Utc, duration, momentCounter);
lock (handlersLock)
{
ActivateMomentHandlers(m);
foreach (var @event in moment.Events)
{
ActivateEventHandlers(m, @event);
}
}
momentCounter++;
}
private void ActivateMomentHandlers(ActivateMoment m)
{
foreach (var handler in momentHandlers)
{
handler(m);
}
}
private void ActivateEventHandlers(ActivateMoment m, OverwatchEvent @event)
{
if (!eventHandlers.ContainsKey(@event.Type)) return;
var handlers = eventHandlers[@event.Type];
foreach (var handler in handlers)
{
handler(m, @event.Payload);
}
}
private OverwatchTranscript LoadModel(string inputFilename)
{
ZipFile.ExtractToDirectory(inputFilename, workingDir);
if (!File.Exists(transcriptFile))
{
closed = true;
throw new Exception("Is not a transcript file. Unzipped to: " + workingDir);
}
return JsonConvert.DeserializeObject<OverwatchTranscript>(File.ReadAllText(transcriptFile))!;
}
private void CheckClosed()
{
if (closed) throw new Exception("Transcript has already been closed.");
}
}
public class ActivateMoment
{
public ActivateMoment(DateTime utc, TimeSpan? duration, long index)
{
Utc = utc;
Duration = duration;
Index = index;
}
public DateTime Utc { get; }
public TimeSpan? Duration { get; }
public long Index { get; }
}
public class ActivateEvent<T>
{
public ActivateEvent(ActivateMoment moment, T payload)
{
Moment = moment;
Payload = payload;
}
public ActivateMoment Moment { get; }
public T Payload { get; }
}
}
@@ -0,0 +1,130 @@
using Logging;
using Newtonsoft.Json;
using System.IO.Compression;
namespace OverwatchTranscript
{
public interface ITranscriptWriter
{
void AddHeader(string key, object value);
void Add(DateTime utc, object payload);
void IncludeArtifact(string filePath);
void Write(string outputFilename);
}
public class TranscriptWriter : ITranscriptWriter
{
private readonly object _lock = new object();
private readonly MomentReferenceBuilder builder;
private readonly string transcriptFile;
private readonly string artifactsFolder;
private readonly Dictionary<string, string> header = new Dictionary<string, string>();
private readonly BucketSet bucketSet;
private readonly ILog log;
private readonly string workingDir;
private bool closed;
public TranscriptWriter(ILog log, string workingDir)
{
closed = false;
this.log = log;
this.workingDir = workingDir;
bucketSet = new BucketSet(log, workingDir);
builder = new MomentReferenceBuilder(log, workingDir);
transcriptFile = Path.Combine(workingDir, TranscriptConstants.TranscriptFilename);
artifactsFolder = Path.Combine(workingDir, TranscriptConstants.ArtifactFolderName);
if (!Directory.Exists(workingDir)) Directory.CreateDirectory(workingDir);
if (File.Exists(transcriptFile) || Directory.Exists(artifactsFolder)) throw new Exception("workingdir not clean");
}
public void Add(DateTime utc, object payload)
{
CheckClosed();
bucketSet.Add(utc, payload);
}
public void AddHeader(string key, object value)
{
CheckClosed();
lock (_lock)
{
header.Add(key, Json.Serialize(value));
}
}
public void IncludeArtifact(string filePath)
{
CheckClosed();
if (!File.Exists(filePath)) throw new Exception("File not found: " + filePath);
if (!Directory.Exists(artifactsFolder)) Directory.CreateDirectory(artifactsFolder);
var name = Path.GetFileName(filePath);
File.Copy(filePath, Path.Combine(artifactsFolder, name), overwrite: false);
}
public void Write(string outputFilename)
{
CheckClosed();
closed = true;
var momentReferences = builder.Build(bucketSet.FinalizeBuckets());
var model = CreateModel(momentReferences);
File.WriteAllText(transcriptFile, Json.Serialize(model, Formatting.Indented));
ZipFile.CreateFromDirectory(workingDir, outputFilename);
log.Debug($"Transcript written to {outputFilename}");
log.Debug($"Common header: {Json.Serialize(model.Header.Common, Formatting.Indented)}");
Directory.Delete(workingDir, true);
log.Debug($"Workdir {workingDir} deleted");
}
private OverwatchTranscript CreateModel(OverwatchMomentReference[] momentReferences)
{
lock (_lock)
{
var model = new OverwatchTranscript
{
Header = new OverwatchHeader
{
Common = CreateCommonHeader(momentReferences),
Entries = header.Select(h =>
{
return new OverwatchHeaderEntry
{
Key = h.Key,
Value = h.Value
};
}).ToArray()
},
MomentReferences = momentReferences
};
header.Clear();
return model;
}
}
private OverwatchCommonHeader CreateCommonHeader(OverwatchMomentReference[] momentReferences)
{
var moments = momentReferences.Sum(m => m.NumberOfMoments);
var events = momentReferences.Sum(m => m.NumberOfEvents);
var earliest = momentReferences.Min(m => m.EarliestUtc);
var latest = momentReferences.Max(m => m.LatestUtc);
return new OverwatchCommonHeader
{
NumberOfMoments = moments,
NumberOfEvents = events,
EarliestUtc = earliest,
LatestUtc = latest
};
}
private void CheckClosed()
{
if (closed) throw new Exception("Transcript has already been written. Cannot modify or write again.");
}
}
}
-1
View File
@@ -13,7 +13,6 @@
public long SizeInBytes { get; }
public long ToMB()
{
return SizeInBytes / (1024 * 1024);
+41
View File
@@ -0,0 +1,41 @@
namespace Utils
{
public static class PluginPathUtils
{
private const string ProjectPluginsFolderName = "ProjectPlugins";
private static string projectPluginsDir = string.Empty;
public static string ProjectPluginsDir
{
get
{
if (string.IsNullOrEmpty(projectPluginsDir)) projectPluginsDir = FindProjectPluginsDir();
return projectPluginsDir;
}
}
private static string FindProjectPluginsDir()
{
var current = Directory.GetCurrentDirectory();
while (true)
{
var localFolders = Directory.GetDirectories(current);
var projectPluginsFolders = localFolders.Where(l => l.EndsWith(ProjectPluginsFolderName)).ToArray();
if (projectPluginsFolders.Length == 1)
{
return projectPluginsFolders.Single();
}
var parent = Directory.GetParent(current);
if (parent == null)
{
var msg = $"Unable to locate '{ProjectPluginsFolderName}' folder. Travelled up from: '{Directory.GetCurrentDirectory()}'";
Console.WriteLine(msg);
throw new Exception(msg);
}
current = parent.FullName;
}
}
}
}
+32 -4
View File
@@ -3,13 +3,41 @@
public static class RandomUtils
{
private static readonly Random random = new Random();
private static readonly object @lock = new object();
public static T GetOneRandom<T>(this T[] items)
{
lock (@lock)
{
var i = random.Next(0, items.Length);
var result = items[i];
return result;
}
}
public static T PickOneRandom<T>(this List<T> remainingItems)
{
var i = random.Next(0, remainingItems.Count);
var result = remainingItems[i];
remainingItems.RemoveAt(i);
return result;
lock (@lock)
{
var i = random.Next(0, remainingItems.Count);
var result = remainingItems[i];
remainingItems.RemoveAt(i);
return result;
}
}
public static T[] Shuffled<T>(T[] items)
{
lock (@lock)
{
var result = new List<T>();
var source = items.ToList();
while (source.Any())
{
result.Add(RandomUtils.PickOneRandom(source));
}
return result.ToArray();
}
}
}
}
+4
View File
@@ -71,6 +71,10 @@
task();
return;
}
catch (OperationCanceledException)
{
return;
}
catch (Exception ex)
{
var failure = CaptureFailure(ex);
+28
View File
@@ -0,0 +1,28 @@
namespace Utils
{
public static class RollingAverage
{
/// <param name="currentAverage">Value of average before new value is added.</param>
/// <param name="newNumberOfValues">Number of values in average after new value is added.</param>
/// <param name="newValue">New value to be added.</param>
/// <returns>New average value.</returns>
/// <exception cref="Exception">newNumberOfValues must be 1 or greater.</exception>
public static float GetNewAverage(float currentAverage, int newNumberOfValues, float newValue)
{
if (newNumberOfValues < 1) throw new Exception("Should be at least 1 value.");
float n = newNumberOfValues;
var originalValue = currentAverage;
var originalValueWeight = ((n - 1.0f) / n);
var newValueWeight = (1.0f / n);
return GetWeightedAverage(originalValue, originalValueWeight, newValue, newValueWeight);
}
public static float GetWeightedAverage(float value1, float weight1, float value2, float weight2)
{
float totalWeight = weight1 + weight2;
if (totalWeight == 0.0f) return 0.0f;
return ((value1 * weight1) + (value2 * weight2)) / totalWeight;
}
}
}
+13
View File
@@ -0,0 +1,13 @@
namespace Utils
{
public static class Str
{
public static string Between(string input, string open, string close)
{
var openIndex = input.IndexOf(open) + open.Length;
var closeIndex = input.LastIndexOf(close);
return input.Substring(openIndex, closeIndex - openIndex);
}
}
}
+1 -1
View File
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<RootNamespace>Utils</RootNamespace>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
@@ -1,4 +1,5 @@
using CodexContractsPlugin.Marketplace;
using System.Collections.Generic;
using Utils;
namespace CodexContractsPlugin.ChainMonitor
@@ -10,38 +11,33 @@ namespace CodexContractsPlugin.ChainMonitor
Request[] requests,
RequestFulfilledEventDTO[] fulfilled,
RequestCancelledEventDTO[] cancelled,
RequestFailedEventDTO[] failed,
SlotFilledEventDTO[] slotFilled,
SlotFreedEventDTO[] slotFreed
SlotFreedEventDTO[] slotFreed,
SlotReservationsFullEventDTO[] slotReservationsFull
)
{
BlockInterval = blockInterval;
Requests = requests;
Fulfilled = fulfilled;
Cancelled = cancelled;
Failed = failed;
SlotFilled = slotFilled;
SlotFreed = slotFreed;
SlotReservationsFull = slotReservationsFull;
All = ConcatAll<IHasBlock>(requests, fulfilled, cancelled, failed, slotFilled, SlotFreed, SlotReservationsFull);
}
public BlockInterval BlockInterval { get; }
public Request[] Requests { get; }
public RequestFulfilledEventDTO[] Fulfilled { get; }
public RequestCancelledEventDTO[] Cancelled { get; }
public RequestFailedEventDTO[] Failed { get; }
public SlotFilledEventDTO[] SlotFilled { get; }
public SlotFreedEventDTO[] SlotFreed { get; }
public IHasBlock[] All
{
get
{
var all = new List<IHasBlock>();
all.AddRange(Requests);
all.AddRange(Fulfilled);
all.AddRange(Cancelled);
all.AddRange(SlotFilled);
all.AddRange(SlotFreed);
return all.ToArray();
}
}
public SlotReservationsFullEventDTO[] SlotReservationsFull { get; }
public IHasBlock[] All { get; }
public static ChainEvents FromBlockInterval(ICodexContracts contracts, BlockInterval blockInterval)
{
@@ -60,9 +56,21 @@ namespace CodexContractsPlugin.ChainMonitor
events.GetStorageRequests(),
events.GetRequestFulfilledEvents(),
events.GetRequestCancelledEvents(),
events.GetRequestFailedEvents(),
events.GetSlotFilledEvents(),
events.GetSlotFreedEvents()
events.GetSlotFreedEvents(),
events.GetSlotReservationsFull()
);
}
private T[] ConcatAll<T>(params T[][] arrays)
{
var result = Array.Empty<T>();
foreach (var array in arrays)
{
result = result.Concat(array).ToArray();
}
return result;
}
}
}
@@ -13,8 +13,12 @@ namespace CodexContractsPlugin.ChainMonitor
void OnRequestFinished(RequestEvent requestEvent);
void OnRequestFulfilled(RequestEvent requestEvent);
void OnRequestCancelled(RequestEvent requestEvent);
void OnRequestFailed(RequestEvent requestEvent);
void OnSlotFilled(RequestEvent requestEvent, EthAddress host, BigInteger slotIndex);
void OnSlotFreed(RequestEvent requestEvent, BigInteger slotIndex);
void OnSlotReservationsFull(RequestEvent requestEvent, BigInteger slotIndex);
void OnError(string msg);
}
public class RequestEvent
@@ -41,33 +45,35 @@ namespace CodexContractsPlugin.ChainMonitor
this.log = new LogPrefixer(log, "(ChainState) ");
this.contracts = contracts;
handler = changeHandler;
StartUtc = startUtc;
TotalSpan = new TimeRange(startUtc, startUtc);
}
public TimeRange TotalSpan { get; private set; }
public IChainStateRequest[] Requests => requests.ToArray();
public DateTime StartUtc { get; }
public void Update()
public int Update()
{
Update(DateTime.UtcNow);
return Update(DateTime.UtcNow);
}
public void Update(DateTime toUtc)
public int Update(DateTime toUtc)
{
var span = new TimeRange(TotalSpan.To, toUtc);
var events = ChainEvents.FromTimeRange(contracts, span);
Apply(events);
TotalSpan = new TimeRange(TotalSpan.From, span.To);
return events.All.Length;
}
private void Apply(ChainEvents events)
{
if (events.BlockInterval.TimeRange.From < TotalSpan.From)
throw new Exception("Attempt to update ChainState with set of events from before its current record.");
{
var msg = "Attempt to update ChainState with set of events from before its current record.";
handler.OnError(msg);
throw new Exception(msg);
}
log.Log($"ChainState updating: {events.BlockInterval}");
@@ -110,7 +116,7 @@ namespace CodexContractsPlugin.ChainMonitor
private void ApplyEvent(RequestFulfilledEventDTO @event)
{
var r = FindRequest(@event.RequestId);
var r = FindRequest(@event);
if (r == null) return;
r.UpdateState(@event.Block.BlockNumber, RequestState.Started);
handler.OnRequestFulfilled(new RequestEvent(@event.Block, r));
@@ -118,15 +124,23 @@ namespace CodexContractsPlugin.ChainMonitor
private void ApplyEvent(RequestCancelledEventDTO @event)
{
var r = FindRequest(@event.RequestId);
var r = FindRequest(@event);
if (r == null) return;
r.UpdateState(@event.Block.BlockNumber, RequestState.Cancelled);
handler.OnRequestCancelled(new RequestEvent(@event.Block, r));
}
private void ApplyEvent(RequestFailedEventDTO @event)
{
var r = FindRequest(@event);
if (r == null) return;
r.UpdateState(@event.Block.BlockNumber, RequestState.Failed);
handler.OnRequestFailed(new RequestEvent(@event.Block, r));
}
private void ApplyEvent(SlotFilledEventDTO @event)
{
var r = FindRequest(@event.RequestId);
var r = FindRequest(@event);
if (r == null) return;
r.Hosts.Add(@event.Host, (int)@event.SlotIndex);
r.Log($"[{@event.Block.BlockNumber}] SlotFilled (host:'{@event.Host}', slotIndex:{@event.SlotIndex})");
@@ -135,13 +149,21 @@ namespace CodexContractsPlugin.ChainMonitor
private void ApplyEvent(SlotFreedEventDTO @event)
{
var r = FindRequest(@event.RequestId);
var r = FindRequest(@event);
if (r == null) return;
r.Hosts.RemoveHost((int)@event.SlotIndex);
r.Log($"[{@event.Block.BlockNumber}] SlotFreed (slotIndex:{@event.SlotIndex})");
handler.OnSlotFreed(new RequestEvent(@event.Block, r), @event.SlotIndex);
}
private void ApplyEvent(SlotReservationsFullEventDTO @event)
{
var r = FindRequest(@event);
if (r == null) return;
r.Log($"[{@event.Block.BlockNumber}] SlotReservationsFull (slotIndex:{@event.SlotIndex})");
handler.OnSlotReservationsFull(new RequestEvent(@event.Block, r), @event.SlotIndex);
}
private void ApplyTimeImplicitEvents(ulong blockNumber, DateTime eventsUtc)
{
foreach (var r in requests)
@@ -155,10 +177,23 @@ namespace CodexContractsPlugin.ChainMonitor
}
}
private ChainStateRequest? FindRequest(byte[] requestId)
private ChainStateRequest? FindRequest(IHasRequestId request)
{
var r = requests.SingleOrDefault(r => Equal(r.Request.RequestId, requestId));
if (r == null) log.Log("Unable to find request by ID!");
var r = requests.SingleOrDefault(r => Equal(r.Request.RequestId, request.RequestId));
if (r == null)
{
var blockNumber = "unknown";
if (request is IHasBlock blk)
{
blockNumber = blk.Block.BlockNumber.ToString();
}
var msg = $"Received event of type '{request.GetType()}' in block '{blockNumber}' for request by Id: '{request.RequestId}'. " +
$"Failed to find request. Request creation event not seen! (Tracker start time: {TotalSpan.From})";
log.Error(msg);
handler.OnError(msg);
}
return r;
}
@@ -0,0 +1,60 @@
using GethPlugin;
using System.Numerics;
namespace CodexContractsPlugin.ChainMonitor
{
public class ChainStateChangeHandlerMux : IChainStateChangeHandler
{
public ChainStateChangeHandlerMux(params IChainStateChangeHandler[] handlers)
{
Handlers = handlers.ToList();
}
public List<IChainStateChangeHandler> Handlers { get; } = new List<IChainStateChangeHandler>();
public void OnNewRequest(RequestEvent requestEvent)
{
foreach (var handler in Handlers) handler.OnNewRequest(requestEvent);
}
public void OnRequestCancelled(RequestEvent requestEvent)
{
foreach (var handler in Handlers) handler.OnRequestCancelled(requestEvent);
}
public void OnRequestFailed(RequestEvent requestEvent)
{
foreach (var handler in Handlers) handler.OnRequestFailed(requestEvent);
}
public void OnRequestFinished(RequestEvent requestEvent)
{
foreach (var handler in Handlers) handler.OnRequestFinished(requestEvent);
}
public void OnRequestFulfilled(RequestEvent requestEvent)
{
foreach (var handler in Handlers) handler.OnRequestFulfilled(requestEvent);
}
public void OnSlotFilled(RequestEvent requestEvent, EthAddress host, BigInteger slotIndex)
{
foreach (var handler in Handlers) handler.OnSlotFilled(requestEvent, host, slotIndex);
}
public void OnSlotFreed(RequestEvent requestEvent, BigInteger slotIndex)
{
foreach (var handler in Handlers) handler.OnSlotFreed(requestEvent, slotIndex);
}
public void OnSlotReservationsFull(RequestEvent requestEvent, BigInteger slotIndex)
{
foreach (var handler in Handlers) handler.OnSlotReservationsFull(requestEvent, slotIndex);
}
public void OnError(string msg)
{
foreach (var handler in Handlers) handler.OnError(msg);
}
}
}
@@ -13,6 +13,10 @@ namespace CodexContractsPlugin.ChainMonitor
{
}
public void OnRequestFailed(RequestEvent requestEvent)
{
}
public void OnRequestFinished(RequestEvent requestEvent)
{
}
@@ -28,5 +32,13 @@ namespace CodexContractsPlugin.ChainMonitor
public void OnSlotFreed(RequestEvent requestEvent, BigInteger slotIndex)
{
}
public void OnSlotReservationsFull(RequestEvent requestEvent, BigInteger slotIndex)
{
}
public void OnError(string msg)
{
}
}
}
@@ -19,7 +19,7 @@ namespace CodexContractsPlugin
{
var config = startupConfig.Get<CodexContractsContainerConfig>();
var address = config.GethNode.StartResult.Container.GetAddress(new NullLog(), GethContainerRecipe.HttpPortTag);
var address = config.GethNode.StartResult.Container.GetAddress(GethContainerRecipe.HttpPortTag);
SetSchedulingAffinity(notIn: "false");
@@ -1,6 +1,7 @@
using CodexContractsPlugin.Marketplace;
using GethPlugin;
using Logging;
using Nethereum.Contracts;
using Nethereum.Hex.HexTypes;
using NethereumWorkflow.BlockUtils;
using Utils;
@@ -13,8 +14,10 @@ namespace CodexContractsPlugin
Request[] GetStorageRequests();
RequestFulfilledEventDTO[] GetRequestFulfilledEvents();
RequestCancelledEventDTO[] GetRequestCancelledEvents();
RequestFailedEventDTO[] GetRequestFailedEvents();
SlotFilledEventDTO[] GetSlotFilledEvents();
SlotFreedEventDTO[] GetSlotFreedEvents();
SlotReservationsFullEventDTO[] GetSlotReservationsFull();
}
public class CodexContractsEvents : ICodexContractsEvents
@@ -37,38 +40,32 @@ namespace CodexContractsPlugin
{
var events = gethNode.GetEvents<StorageRequestedEventDTO>(deployment.MarketplaceAddress, BlockInterval);
var i = new ContractInteractions(log, gethNode);
return events
.Select(e =>
return events.Select(e =>
{
var requestEvent = i.GetRequest(deployment.MarketplaceAddress, e.Event.RequestId);
var result = requestEvent.ReturnValue1;
result.Block = GetBlock(e.Log.BlockNumber.ToUlong());
result.RequestId = e.Event.RequestId;
return result;
})
.ToArray();
var requestEvent = i.GetRequest(deployment.MarketplaceAddress, e.Event.RequestId);
var result = requestEvent.ReturnValue1;
result.Block = GetBlock(e.Log.BlockNumber.ToUlong());
result.RequestId = e.Event.RequestId;
return result;
}).ToArray();
}
public RequestFulfilledEventDTO[] GetRequestFulfilledEvents()
{
var events = gethNode.GetEvents<RequestFulfilledEventDTO>(deployment.MarketplaceAddress, BlockInterval);
return events.Select(e =>
{
var result = e.Event;
result.Block = GetBlock(e.Log.BlockNumber.ToUlong());
return result;
}).ToArray();
return events.Select(SetBlockOnEvent).ToArray();
}
public RequestCancelledEventDTO[] GetRequestCancelledEvents()
{
var events = gethNode.GetEvents<RequestCancelledEventDTO>(deployment.MarketplaceAddress, BlockInterval);
return events.Select(e =>
{
var result = e.Event;
result.Block = GetBlock(e.Log.BlockNumber.ToUlong());
return result;
}).ToArray();
return events.Select(SetBlockOnEvent).ToArray();
}
public RequestFailedEventDTO[] GetRequestFailedEvents()
{
var events = gethNode.GetEvents<RequestFailedEventDTO>(deployment.MarketplaceAddress, BlockInterval);
return events.Select(SetBlockOnEvent).ToArray();
}
public SlotFilledEventDTO[] GetSlotFilledEvents()
@@ -86,12 +83,20 @@ namespace CodexContractsPlugin
public SlotFreedEventDTO[] GetSlotFreedEvents()
{
var events = gethNode.GetEvents<SlotFreedEventDTO>(deployment.MarketplaceAddress, BlockInterval);
return events.Select(e =>
{
var result = e.Event;
result.Block = GetBlock(e.Log.BlockNumber.ToUlong());
return result;
}).ToArray();
return events.Select(SetBlockOnEvent).ToArray();
}
public SlotReservationsFullEventDTO[] GetSlotReservationsFull()
{
var events = gethNode.GetEvents<SlotReservationsFullEventDTO>(deployment.MarketplaceAddress, BlockInterval);
return events.Select(SetBlockOnEvent).ToArray();
}
private T SetBlockOnEvent<T>(EventLog<T> e) where T : IHasBlock
{
var result = e.Event;
result.Block = GetBlock(e.Log.BlockNumber.ToUlong());
return result;
}
private BlockTimeEntry GetBlock(ulong number)
@@ -1,11 +1,16 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Nethereum.Generators" Version="4.21.4" />
<PackageReference Include="Nethereum.Generators.Net" Version="4.21.4" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Framework\Core\Core.csproj" />
<ProjectReference Include="..\GethPlugin\GethPlugin.csproj" />
@@ -1,4 +1,5 @@
using Core;
using CodexContractsPlugin.Marketplace;
using Core;
using GethPlugin;
using KubernetesWorkflow;
using KubernetesWorkflow.Types;
@@ -64,7 +65,8 @@ namespace CodexContractsPlugin
var extractor = new ContractsContainerInfoExtractor(tools.GetLog(), workflow, container);
var marketplaceAddress = extractor.ExtractMarketplaceAddress();
var abi = extractor.ExtractMarketplaceAbi();
var (abi, bytecode) = extractor.ExtractMarketplaceAbiAndByteCode();
EnsureCompatbility(abi, bytecode);
var interaction = new ContractInteractions(tools.GetLog(), gethNode);
var tokenAddress = interaction.GetTokenAddress(marketplaceAddress);
@@ -78,6 +80,18 @@ namespace CodexContractsPlugin
return new CodexContractsDeployment(marketplaceAddress, abi, tokenAddress);
}
private void EnsureCompatbility(string abi, string bytecode)
{
var expectedByteCode = MarketplaceDeploymentBase.BYTECODE.ToLowerInvariant();
if (bytecode != expectedByteCode)
{
Log("Deployed contract is incompatible with current build of CodexContracts plugin. Running self-updater...");
var selfUpdater = new SelfUpdater();
selfUpdater.Update(abi, bytecode);
}
}
private void Log(string msg)
{
tools.GetLog().Log(msg);
@@ -31,14 +31,14 @@ namespace CodexContractsPlugin
return marketplaceAddress;
}
public string ExtractMarketplaceAbi()
public (string, string) ExtractMarketplaceAbiAndByteCode()
{
log.Debug();
var marketplaceAbi = Retry(FetchMarketplaceAbi);
if (string.IsNullOrEmpty(marketplaceAbi)) throw new InvalidOperationException("Unable to fetch marketplace artifacts from codex-contracts node. Test infra failure.");
var (abi, bytecode) = Retry(FetchMarketplaceAbiAndByteCode);
if (string.IsNullOrEmpty(abi)) throw new InvalidOperationException("Unable to fetch marketplace artifacts from codex-contracts node. Test infra failure.");
log.Debug("Got Marketplace ABI: " + marketplaceAbi);
return marketplaceAbi;
log.Debug("Got Marketplace ABI: " + abi);
return (abi, bytecode);
}
private string FetchMarketplaceAddress()
@@ -48,7 +48,7 @@ namespace CodexContractsPlugin
return marketplace!.address;
}
private string FetchMarketplaceAbi()
private (string, string) FetchMarketplaceAbiAndByteCode()
{
var json = workflow.ExecuteCommand(container, "cat", CodexContractsContainerRecipe.MarketplaceArtifactFilename);
@@ -56,19 +56,12 @@ namespace CodexContractsPlugin
var abi = artifact["abi"];
var byteCode = artifact["bytecode"];
var abiResult = abi!.ToString(Formatting.None);
var byteCodeResult = byteCode!.ToString(Formatting.None);
if (byteCodeResult
.ToLowerInvariant()
.Replace("\"", "") != MarketplaceDeploymentBase.BYTECODE.ToLowerInvariant())
{
throw new Exception("BYTECODE in CodexContractsPlugin does not match BYTECODE deployed by container. Update Marketplace.cs generated code?");
}
return abiResult;
var byteCodeResult = byteCode!.ToString(Formatting.None).ToLowerInvariant().Replace("\"", "");
return (abiResult, byteCodeResult);
}
private static string Retry(Func<string> fetch)
private static T Retry<T>(Func<T> fetch)
{
return Time.Retry(fetch, nameof(ContractsContainerInfoExtractor));
}
@@ -10,7 +10,12 @@ namespace CodexContractsPlugin.Marketplace
BlockTimeEntry Block { get; set; }
}
public partial class Request : RequestBase, IHasBlock
public interface IHasRequestId
{
byte[] RequestId { get; set; }
}
public partial class Request : RequestBase, IHasBlock, IHasRequestId
{
[JsonIgnore]
public BlockTimeEntry Block { get; set; }
@@ -28,26 +33,38 @@ namespace CodexContractsPlugin.Marketplace
}
}
public partial class RequestFulfilledEventDTO : IHasBlock
public partial class RequestFulfilledEventDTO : IHasBlock, IHasRequestId
{
[JsonIgnore]
public BlockTimeEntry Block { get; set; }
}
public partial class RequestCancelledEventDTO : IHasBlock
public partial class RequestCancelledEventDTO : IHasBlock, IHasRequestId
{
[JsonIgnore]
public BlockTimeEntry Block { get; set; }
}
public partial class SlotFilledEventDTO : IHasBlock
public partial class RequestFailedEventDTO : IHasBlock, IHasRequestId
{
[JsonIgnore]
public BlockTimeEntry Block { get; set; }
}
public partial class SlotFilledEventDTO : IHasBlock, IHasRequestId
{
[JsonIgnore]
public BlockTimeEntry Block { get; set; }
public EthAddress Host { get; set; }
}
public partial class SlotFreedEventDTO : IHasBlock
public partial class SlotFreedEventDTO : IHasBlock, IHasRequestId
{
[JsonIgnore]
public BlockTimeEntry Block { get; set; }
}
public partial class SlotReservationsFullEventDTO : IHasBlock, IHasRequestId
{
[JsonIgnore]
public BlockTimeEntry Block { get; set; }
File diff suppressed because one or more lines are too long
@@ -0,0 +1,96 @@
using Utils;
namespace CodexContractsPlugin
{
public class SelfUpdater
{
public void Update(string abi, string bytecode)
{
var filePath = GetMarketplaceFilePath();
var content = GenerateContent(abi, bytecode);
var contentLines = content.Split("\r\n");
var beginWith = new string[]
{
"using Nethereum.ABI.FunctionEncoding.Attributes;",
"using Nethereum.Contracts;",
"using System.Numerics;",
"",
"// Generated code, do not modify.",
"",
"#pragma warning disable CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider declaring as nullable.",
"namespace CodexContractsPlugin.Marketplace",
"{"
};
var endWith = new string[]
{
"}",
"#pragma warning restore CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider declaring as nullable."
};
File.Delete(filePath);
File.WriteAllLines(filePath,
beginWith.Concat(
contentLines.Concat(
endWith))
);
throw new Exception("Oh no! CodexContracts were updated. Current build of CodexContractsPlugin is incompatible. " +
"But fear not! SelfUpdater.cs has automatically updated the plugin. Just rebuild and rerun and it should work. " +
"Just in case, manual update instructions are found here: 'CodexContractsPlugin/Marketplace/README.md'.");
}
private string GetMarketplaceFilePath()
{
var projectPluginDir = PluginPathUtils.ProjectPluginsDir;
var path = Path.Combine(projectPluginDir, "CodexContractsPlugin", "Marketplace", "Marketplace.cs");
if (!File.Exists(path)) throw new Exception("Marketplace file not found. Expected: " + path);
return path;
}
private string GenerateContent(string abi, string bytecode)
{
var deserializer = new Nethereum.Generators.Net.GeneratorModelABIDeserialiser();
var abiModel = deserializer.DeserialiseABI(abi);
var abiCtor = abiModel.Constructor;
var c = new Nethereum.Generators.CQS.ContractDeploymentCQSMessageGenerator(abiCtor, "namespace", bytecode, "Marketplace", Nethereum.Generators.Core.CodeGenLanguage.CSharp);
var lines = "";
lines += c.GenerateClass();
lines += "\r\n";
foreach (var eventAbi in abiModel.Events)
{
var d = new Nethereum.Generators.DTOs.EventDTOGenerator(eventAbi, "namespace", Nethereum.Generators.Core.CodeGenLanguage.CSharp);
lines += d.GenerateClass();
lines += "\r\n";
}
foreach (var errorAbi in abiModel.Errors)
{
var e = new Nethereum.Generators.DTOs.ErrorDTOGenerator(errorAbi, "namespace", Nethereum.Generators.Core.CodeGenLanguage.CSharp);
lines += e.GenerateClass();
lines += "\r\n";
}
foreach (var funcAbi in abiModel.Functions)
{
var f = new Nethereum.Generators.DTOs.FunctionOutputDTOGenerator(funcAbi, "namespace", Nethereum.Generators.Core.CodeGenLanguage.CSharp);
var ff = new Nethereum.Generators.CQS.FunctionCQSMessageGenerator(funcAbi, "namespace", "funcoutput", Nethereum.Generators.Core.CodeGenLanguage.CSharp);
lines += f.GenerateClass();
lines += "\r\n";
lines += ff.GenerateClass();
lines += "\r\n";
}
foreach (var structAbi in abiModel.Structs)
{
var g = new Nethereum.Generators.DTOs.StructTypeGenerator(structAbi, "namespace", Nethereum.Generators.Core.CodeGenLanguage.CSharp);
lines += g.GenerateClass();
lines += "\r\n";
}
return lines;
}
}
}
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
+18 -3
View File
@@ -3,13 +3,14 @@ using KubernetesWorkflow.Types;
using Logging;
using System.Security.Cryptography;
using System.Text;
using Utils;
namespace CodexPlugin
{
public class ApiChecker
{
// <INSERT-OPENAPI-YAML-HASH>
private const string OpenApiYamlHash = "67-76-AB-FC-54-4F-EB-81-F5-E4-F8-27-DF-82-92-41-63-A5-EA-1B-17-14-0C-BE-20-9C-B3-DF-CE-E4-AA-38";
private const string OpenApiYamlHash = "09-53-C3-A6-31-A5-0C-8B-53-1C-3D-C7-2B-1E-85-C7-17-60-54-43-01-C4-49-4E-D9-68-35-7D-F7-41-13-B5";
private const string OpenApiFilePath = "/codex/openapi.yaml";
private const string DisableEnvironmentVariable = "CODEXPLUGIN_DISABLE_APICHECK";
@@ -21,8 +22,9 @@ namespace CodexPlugin
private const string Failure =
"Codex API compatibility check failed! " +
"openapi.yaml used by CodexPlugin does not match openapi.yaml in Codex container. Please update the openapi.yaml in " +
"'ProjectPlugins/CodexPlugin' and rebuild this project. If you wish to disable API compatibility checking, please set " +
"openapi.yaml used by CodexPlugin does not match openapi.yaml in Codex container. The openapi.yaml in " +
"'ProjectPlugins/CodexPlugin' has been overwritten with the container one. " +
"Please and rebuild this project. If you wish to disable API compatibility checking, please set " +
$"the environment variable '{DisableEnvironmentVariable}' or set the disable bool in 'ProjectPlugins/CodexPlugin/ApiChecker.cs'.";
private static bool checkPassed = false;
@@ -71,10 +73,23 @@ namespace CodexPlugin
return;
}
OverwriteOpenApiYaml(containerApi);
log.Error(Failure);
throw new Exception(Failure);
}
private void OverwriteOpenApiYaml(string containerApi)
{
Log("API compatibility check failed. Updating CodexPlugin...");
var openApiFilePath = Path.Combine(PluginPathUtils.ProjectPluginsDir, "CodexPlugin", "openapi.yaml");
if (!File.Exists(openApiFilePath)) throw new Exception("Unable to locate CodexPlugin/openapi.yaml. Expected: " + openApiFilePath);
File.Delete(openApiFilePath);
File.WriteAllText(openApiFilePath, containerApi);
Log("CodexPlugin/openapi.yaml has been updated.");
}
private string Hash(string file)
{
var fileBytes = Encoding.ASCII.GetBytes(file
+35 -25
View File
@@ -63,17 +63,17 @@ namespace CodexPlugin
});
}
public string UploadFile(FileStream fileStream, Action<Failure> onFailure)
public string UploadFile(UploadInput uploadInput, Action<Failure> onFailure)
{
return OnCodex(
api => api.UploadAsync(fileStream),
api => api.UploadAsync(uploadInput.ContentType, uploadInput.ContentDisposition, uploadInput.FileStream),
CreateRetryConfig(nameof(UploadFile), onFailure));
}
public Stream DownloadFile(string contentId, Action<Failure> onFailure)
{
var fileResponse = OnCodex(
api => api.DownloadNetworkAsync(contentId),
api => api.DownloadNetworkStreamAsync(contentId),
CreateRetryConfig(nameof(DownloadFile), onFailure));
if (fileResponse.StatusCode != 200) throw new Exception("Download failed with StatusCode: " + fileResponse.StatusCode);
@@ -88,19 +88,25 @@ namespace CodexPlugin
public StorageAvailability SalesAvailability(StorageAvailability request)
{
var body = mapper.Map(request);
var read = OnCodex<SalesAvailabilityREAD>(api => api.OfferStorageAsync(body));
var read = OnCodex(api => api.OfferStorageAsync(body));
return mapper.Map(read);
}
public StorageAvailability[] GetAvailabilities()
{
var collection = OnCodex(api => api.GetAvailabilitiesAsync());
return mapper.Map(collection);
}
public string RequestStorage(StoragePurchaseRequest request)
{
var body = mapper.Map(request);
return OnCodex<string>(api => api.CreateStorageRequestAsync(request.ContentId.Id, body));
return OnCodex(api => api.CreateStorageRequestAsync(request.ContentId.Id, body));
}
public CodexSpace Space()
{
var space = OnCodex<Space>(api => api.SpaceAsync());
var space = OnCodex(api => api.SpaceAsync());
return mapper.Map(space);
}
@@ -132,19 +138,6 @@ namespace CodexPlugin
return workflow.GetPodInfo(Container);
}
public void LogDiskSpace(string msg)
{
try
{
var diskInfo = tools.CreateWorkflow().ExecuteCommand(Container.Containers.Single(), "df", "--sync");
Log($"{msg} - Disk info: {diskInfo}");
}
catch (Exception e)
{
Log("Failed to get disk info: " + e);
}
}
public void DeleteRepoFolder()
{
try
@@ -163,13 +156,13 @@ namespace CodexPlugin
private T OnCodex<T>(Func<CodexApi, Task<T>> action)
{
var result = tools.CreateHttp(CheckContainerCrashed).OnClient(client => CallCodex(client, action));
var result = tools.CreateHttp(GetHttpId(), CheckContainerCrashed).OnClient(client => CallCodex(client, action));
return result;
}
private T OnCodex<T>(Func<CodexApi, Task<T>> action, Retry retry)
{
var result = tools.CreateHttp(CheckContainerCrashed).OnClient(client => CallCodex(client, action), retry);
var result = tools.CreateHttp(GetHttpId(), CheckContainerCrashed).OnClient(client => CallCodex(client, action), retry);
return result;
}
@@ -196,13 +189,18 @@ namespace CodexPlugin
private IEndpoint GetEndpoint()
{
return tools
.CreateHttp(CheckContainerCrashed)
.CreateHttp(GetHttpId(), CheckContainerCrashed)
.CreateEndpoint(GetAddress(), "/api/codex/v1/", Container.Name);
}
private Address GetAddress()
{
return Container.Containers.Single().GetAddress(log, CodexContainerRecipe.ApiPortTag);
return Container.Containers.Single().GetAddress(CodexContainerRecipe.ApiPortTag);
}
private string GetHttpId()
{
return GetAddress().ToString();
}
private void CheckContainerCrashed(HttpClient client)
@@ -227,8 +225,6 @@ namespace CodexPlugin
$"(HTTP timeout = {Time.FormatDuration(timeSet.HttpCallTimeout())}) " +
$"Checking if node responds to debug/info...");
LogDiskSpace("After retry failure");
try
{
var debugInfo = GetDebugInfo();
@@ -265,4 +261,18 @@ namespace CodexPlugin
log.Log($"{GetName()} {msg}");
}
}
public class UploadInput
{
public UploadInput(string contentType, string contentDisposition, FileStream fileStream)
{
ContentType = contentType;
ContentDisposition = contentDisposition;
FileStream = fileStream;
}
public string ContentType { get; }
public string ContentDisposition { get; }
public FileStream FileStream { get; }
}
}
@@ -7,8 +7,7 @@ namespace CodexPlugin
{
public class CodexContainerRecipe : ContainerRecipeFactory
{
private const string DefaultDockerImage = "codexstorage/nim-codex:sha-471ebb2-dist-tests";
private const string DefaultDockerImage = "codexstorage/nim-codex:latest-dist-tests";
public const string ApiPortTag = "codex_api_port";
public const string ListenPortTag = "codex_listen_port";
public const string MetricsPortTag = "codex_metrics_port";
+86 -26
View File
@@ -1,4 +1,5 @@
using Core;
using CodexPlugin.Hooks;
using Core;
using FileUtils;
using GethPlugin;
using KubernetesWorkflow;
@@ -12,10 +13,12 @@ namespace CodexPlugin
public interface ICodexNode : IHasContainer, IHasMetricsScrapeTarget, IHasEthAddress
{
string GetName();
string GetPeerId();
DebugInfo GetDebugInfo();
DebugPeer GetDebugPeer(string peerId);
ContentId UploadFile(TrackedFile file);
ContentId UploadFile(TrackedFile file, Action<Failure> onFailure);
ContentId UploadFile(TrackedFile file, string contentType, string contentDisposition, Action<Failure> onFailure);
TrackedFile? DownloadContent(ContentId contentId, string fileLabel = "");
TrackedFile? DownloadContent(ContentId contentId, Action<Failure> onFailure, string fileLabel = "");
LocalDatasetList LocalFiles();
@@ -39,19 +42,36 @@ namespace CodexPlugin
public class CodexNode : ICodexNode
{
private const string UploadFailedMessage = "Unable to store block";
private readonly ILog log;
private readonly IPluginTools tools;
private readonly ICodexNodeHooks hooks;
private readonly EthAccount? ethAccount;
private readonly TransferSpeeds transferSpeeds;
private string peerId = string.Empty;
private string nodeId = string.Empty;
public CodexNode(IPluginTools tools, CodexAccess codexAccess, CodexNodeGroup group, IMarketplaceAccess marketplaceAccess, EthAccount? ethAccount)
public CodexNode(IPluginTools tools, CodexAccess codexAccess, CodexNodeGroup group, IMarketplaceAccess marketplaceAccess, ICodexNodeHooks hooks, EthAccount? ethAccount)
{
this.tools = tools;
this.ethAccount = ethAccount;
CodexAccess = codexAccess;
Group = group;
Marketplace = marketplaceAccess;
this.hooks = hooks;
Version = new DebugInfoVersion();
transferSpeeds = new TransferSpeeds();
log = new LogPrefixer(tools.GetLog(), $"{GetName()} ");
}
public void Awake()
{
hooks.OnNodeStarting(Container.Recipe.RecipeCreatedUtc, Container.Recipe.Image, ethAccount);
}
public void Initialize()
{
hooks.OnNodeStarted(peerId, nodeId);
}
public RunningPod Pod { get { return CodexAccess.Container; } }
@@ -95,6 +115,11 @@ namespace CodexPlugin
return Container.Name;
}
public string GetPeerId()
{
return peerId;
}
public DebugInfo GetDebugInfo()
{
var debugInfo = CodexAccess.GetDebugInfo();
@@ -115,27 +140,35 @@ namespace CodexPlugin
public ContentId UploadFile(TrackedFile file, Action<Failure> onFailure)
{
CodexAccess.LogDiskSpace("Before upload");
return UploadFile(file, "application/x-binary", $"attachment; filename=\"{file.Filename}\"", onFailure);
}
public ContentId UploadFile(TrackedFile file, string contentType, string contentDisposition, Action<Failure> onFailure)
{
using var fileStream = File.OpenRead(file.Filename);
var uniqueId = Guid.NewGuid().ToString();
var size = file.GetFilesize();
var logMessage = $"Uploading file {file.Describe()}...";
Log(logMessage);
var measurement = Stopwatch.Measure(tools.GetLog(), logMessage, () =>
hooks.OnFileUploading(uniqueId, size);
var input = new UploadInput(contentType, contentDisposition, fileStream);
var logMessage = $"Uploading file {file.Describe()} with contentType: '{input.ContentType}' and disposition: '{input.ContentDisposition}'...";
var measurement = Stopwatch.Measure(log, logMessage, () =>
{
return CodexAccess.UploadFile(fileStream, onFailure);
return CodexAccess.UploadFile(input, onFailure);
});
var response = measurement.Value;
transferSpeeds.AddUploadSample(file.GetFilesize(), measurement.Duration);
transferSpeeds.AddUploadSample(size, measurement.Duration);
if (string.IsNullOrEmpty(response)) FrameworkAssert.Fail("Received empty response.");
if (response.StartsWith(UploadFailedMessage)) FrameworkAssert.Fail("Node failed to store block.");
Log($"Uploaded file. Received contentId: '{response}'.");
CodexAccess.LogDiskSpace("After upload");
Log($"Uploaded file {file.Describe()}. Received contentId: '{response}'.");
return new ContentId(response);
var cid = new ContentId(response);
hooks.OnFileUploaded(uniqueId, size, cid);
return cid;
}
public TrackedFile? DownloadContent(ContentId contentId, string fileLabel = "")
@@ -145,12 +178,17 @@ namespace CodexPlugin
public TrackedFile? DownloadContent(ContentId contentId, Action<Failure> onFailure, string fileLabel = "")
{
var logMessage = $"Downloading for contentId: '{contentId.Id}'...";
Log(logMessage);
var file = tools.GetFileManager().CreateEmptyFile(fileLabel);
var measurement = Stopwatch.Measure(tools.GetLog(), logMessage, () => DownloadToFile(contentId.Id, file, onFailure));
transferSpeeds.AddDownloadSample(file.GetFilesize(), measurement);
Log($"Downloaded file {file.Describe()} to '{file.Filename}'.");
hooks.OnFileDownloading(contentId);
Log($"Downloading '{contentId}'...");
var logMessage = $"Downloaded '{contentId}' to '{file.Filename}'";
var measurement = Stopwatch.Measure(log, logMessage, () => DownloadToFile(contentId.Id, file, onFailure));
var size = file.GetFilesize();
transferSpeeds.AddDownloadSample(size, measurement);
hooks.OnFileDownloaded(size, contentId);
return file;
}
@@ -188,6 +226,8 @@ namespace CodexPlugin
public void Stop(bool waitTillStopped)
{
Log("Stopping...");
hooks.OnNodeStopping();
CrashWatcher.Stop();
Group.Stop(this, waitTillStopped);
}
@@ -195,7 +235,8 @@ namespace CodexPlugin
public void EnsureOnlineGetVersionResponse()
{
var debugInfo = Time.Retry(CodexAccess.GetDebugInfo, "ensure online");
var nodePeerId = debugInfo.Id;
peerId = debugInfo.Id;
nodeId = debugInfo.Table.LocalNode.NodeId;
var nodeName = CodexAccess.Container.Name;
if (!debugInfo.Version.IsValid())
@@ -203,12 +244,18 @@ namespace CodexPlugin
throw new Exception($"Invalid version information received from Codex node {GetName()}: {debugInfo.Version}");
}
var log = tools.GetLog();
log.AddStringReplace(nodePeerId, nodeName);
log.AddStringReplace(peerId, nodeName);
log.AddStringReplace(CodexUtils.ToShortId(peerId), nodeName);
log.AddStringReplace(debugInfo.Table.LocalNode.NodeId, nodeName);
log.AddStringReplace(CodexUtils.ToShortId(debugInfo.Table.LocalNode.NodeId), nodeName);
Version = debugInfo.Version;
}
public override string ToString()
{
return $"CodexNode:{GetName()}";
}
private string[] GetPeerMultiAddresses(CodexNode peer, DebugInfo peerInfo)
{
// The peer we want to connect is in a different pod.
@@ -223,21 +270,34 @@ namespace CodexPlugin
private void DownloadToFile(string contentId, TrackedFile file, Action<Failure> onFailure)
{
CodexAccess.LogDiskSpace("Before download");
using var fileStream = File.OpenWrite(file.Filename);
var timeout = tools.TimeSet.HttpCallTimeout();
try
{
using var downloadStream = CodexAccess.DownloadFile(contentId, onFailure);
downloadStream.CopyTo(fileStream);
// Type of stream generated by openAPI client does not support timeouts.
var start = DateTime.UtcNow;
var cts = new CancellationTokenSource();
var downloadTask = Task.Run(() =>
{
using var downloadStream = CodexAccess.DownloadFile(contentId, onFailure);
downloadStream.CopyTo(fileStream);
}, cts.Token);
while (DateTime.UtcNow - start < timeout)
{
if (downloadTask.IsFaulted) throw downloadTask.Exception;
if (downloadTask.IsCompletedSuccessfully) return;
Thread.Sleep(100);
}
cts.Cancel();
throw new TimeoutException($"Download of '{contentId}' timed out after {Time.FormatDuration(timeout)}");
}
catch
{
Log($"Failed to download file '{contentId}'.");
throw;
}
CodexAccess.LogDiskSpace("After download");
}
private void EnsureMarketplace()
@@ -247,7 +307,7 @@ namespace CodexPlugin
private void Log(string msg)
{
tools.GetLog().Log($"{GetName()}: {msg}");
log.Log(msg);
}
private void DoNothing(Failure failure)
+11 -6
View File
@@ -1,4 +1,5 @@
using Core;
using CodexPlugin.Hooks;
using Core;
using GethPlugin;
using KubernetesWorkflow;
using KubernetesWorkflow.Types;
@@ -14,23 +15,27 @@ namespace CodexPlugin
public class CodexNodeFactory : ICodexNodeFactory
{
private readonly IPluginTools tools;
private readonly CodexHooksFactory codexHooksFactory;
public CodexNodeFactory(IPluginTools tools)
public CodexNodeFactory(IPluginTools tools, CodexHooksFactory codexHooksFactory)
{
this.tools = tools;
this.codexHooksFactory = codexHooksFactory;
}
public CodexNode CreateOnlineCodexNode(CodexAccess access, CodexNodeGroup group)
{
var ethAccount = GetEthAccount(access);
var marketplaceAccess = GetMarketplaceAccess(access, ethAccount);
return new CodexNode(tools, access, group, marketplaceAccess, ethAccount);
var hooks = codexHooksFactory.CreateHooks(access.Container.Name);
var marketplaceAccess = GetMarketplaceAccess(access, ethAccount, hooks);
return new CodexNode(tools, access, group, marketplaceAccess, hooks, ethAccount);
}
private IMarketplaceAccess GetMarketplaceAccess(CodexAccess codexAccess, EthAccount? ethAccount)
private IMarketplaceAccess GetMarketplaceAccess(CodexAccess codexAccess, EthAccount? ethAccount, ICodexNodeHooks hooks)
{
if (ethAccount == null) return new MarketplaceUnavailable();
return new MarketplaceAccess(tools.GetLog(), codexAccess);
return new MarketplaceAccess(tools.GetLog(), codexAccess, hooks);
}
private EthAccount? GetEthAccount(CodexAccess access)
+4 -1
View File
@@ -79,13 +79,16 @@ namespace CodexPlugin
}
Version = first;
foreach (var node in Nodes) node.Initialize();
}
private CodexNode CreateOnlineCodexNode(RunningPod c, IPluginTools tools, ICodexNodeFactory factory)
{
var watcher = factory.CreateCrashWatcher(c.Containers.Single());
var access = new CodexAccess(tools, c, watcher);
return factory.CreateOnlineCodexNode(access, this);
var node = factory.CreateOnlineCodexNode(access, this);
node.Awake();
return node;
}
}
}
@@ -1,3 +1,4 @@
using CodexPlugin.Hooks;
using Core;
using KubernetesWorkflow.Types;
@@ -57,6 +58,11 @@ namespace CodexPlugin
}
}
public void SetCodexHooksProvider(ICodexHooksProvider hooksProvider)
{
codexStarter.HooksFactory.Provider = hooksProvider;
}
private CodexSetup GetSetup(int numberOfNodes, Action<ICodexSetup> setup)
{
var codexSetup = new CodexSetup(numberOfNodes);
@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net7.0</TargetFramework>
<TargetFramework>net8.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -29,6 +29,7 @@
<ItemGroup>
<ProjectReference Include="..\..\Framework\Core\Core.csproj" />
<ProjectReference Include="..\..\Framework\KubernetesWorkflow\KubernetesWorkflow.csproj" />
<ProjectReference Include="..\..\Framework\OverwatchTranscript\OverwatchTranscript.csproj" />
<ProjectReference Include="..\CodexContractsPlugin\CodexContractsPlugin.csproj" />
<ProjectReference Include="..\GethPlugin\GethPlugin.csproj" />
<ProjectReference Include="..\MetricsPlugin\MetricsPlugin.csproj" />
+4 -3
View File
@@ -169,7 +169,7 @@ namespace CodexPlugin
public bool IsValidator { get; private set; }
public Ether InitialEth { get; private set; } = 0.Eth();
public TestToken InitialTestTokens { get; private set; } = 0.Tst();
public EthAccountSetup EthAccountSetup { get; private set; } = new EthAccountSetup();
public EthAccountSetup EthAccountSetup { get; } = new EthAccountSetup();
public IMarketplaceSetup AsStorageNode()
{
@@ -201,8 +201,8 @@ namespace CodexPlugin
var result = "[(clientNode)"; // When marketplace is enabled, being a clientNode is implicit.
result += IsStorageNode ? "(storageNode)" : "()";
result += IsValidator ? "(validator)" : "() ";
result += $"Address: '{EthAccountSetup}' ";
result += $"{InitialEth.Eth} / {InitialTestTokens}";
result += $"Pinned address: '{EthAccountSetup}' ";
result += $"{InitialEth} / {InitialTestTokens}";
result += "] ";
return result;
}
@@ -235,6 +235,7 @@ namespace CodexPlugin
public override string ToString()
{
if (!accounts.Any()) return "NoEthAccounts";
return string.Join(",", accounts.Select(a => a.ToString()).ToArray());
}
}
+15 -3
View File
@@ -1,4 +1,6 @@
using Core;
using CodexPlugin.Hooks;
using Core;
using GethPlugin;
using KubernetesWorkflow;
using KubernetesWorkflow.Types;
using Logging;
@@ -19,6 +21,8 @@ namespace CodexPlugin
apiChecker = new ApiChecker(pluginTools);
}
public CodexHooksFactory HooksFactory { get; } = new CodexHooksFactory();
public RunningPod[] BringOnline(CodexSetup codexSetup)
{
LogSeparator();
@@ -34,7 +38,8 @@ namespace CodexPlugin
{
var podInfo = GetPodInfo(rc);
var podInfos = string.Join(", ", rc.Containers.Select(c => $"Container: '{c.Name}' PodLabel: '{c.RunningPod.StartResult.Deployment.PodLabel}' runs at '{podInfo.K8SNodeName}'={podInfo.Ip}"));
Log($"Started {codexSetup.NumberOfNodes} nodes of image '{containers.First().Containers.First().Recipe.Image}'. ({podInfos})");
Log($"Started node with image '{containers.First().Containers.First().Recipe.Image}'. ({podInfos})");
LogEthAddress(rc);
}
LogSeparator();
@@ -43,7 +48,7 @@ namespace CodexPlugin
public ICodexNodeGroup WrapCodexContainers(CoreInterface coreInterface, RunningPod[] containers)
{
var codexNodeFactory = new CodexNodeFactory(pluginTools);
var codexNodeFactory = new CodexNodeFactory(pluginTools, HooksFactory);
var group = CreateCodexGroup(coreInterface, containers, codexNodeFactory);
@@ -141,6 +146,13 @@ namespace CodexPlugin
Log("----------------------------------------------------------------------------");
}
private void LogEthAddress(RunningPod rc)
{
var account = rc.Containers.First().Recipe.Additionals.Get<EthAccount>();
if (account == null) return;
Log($"{rc.Name} = {account}");
}
private void Log(string message)
{
pluginTools.GetLog().Log(message);
@@ -8,7 +8,7 @@ namespace CodexPlugin
public string? NameOverride { get; set; }
public ILocation Location { get; set; } = KnownLocations.UnspecifiedLocation;
public CodexLogLevel LogLevel { get; set; }
public CodexLogCustomTopics? CustomTopics { get; set; } = new CodexLogCustomTopics(CodexLogLevel.Warn, CodexLogLevel.Warn);
public CodexLogCustomTopics? CustomTopics { get; set; } = new CodexLogCustomTopics(CodexLogLevel.Info, CodexLogLevel.Warn);
public ByteSize? StorageQuota { get; set; }
public bool MetricsEnabled { get; set; }
public MarketplaceInitialConfig? MarketplaceConfig { get; set; }
@@ -50,10 +50,10 @@ namespace CodexPlugin
"secure",
"chronosstream",
"connection",
"connmanager",
// Removed: "connmanager", is used for transcript peer-dropped event.
"websock",
"ws-session",
"dialer",
// Removed: "dialer", is used for transcript successful-dial event.
"muxedupgrade",
"upgrade",
"identify"
+5
View File
@@ -95,6 +95,11 @@ namespace CodexPlugin
public string Id { get; }
public override string ToString()
{
return Id;
}
public override bool Equals(object? obj)
{
return obj is ContentId id && Id == id.Id;
+24
View File
@@ -0,0 +1,24 @@
namespace CodexPlugin
{
public static class CodexUtils
{
public static string ToShortId(string id)
{
if (id.Length > 10)
{
return $"{id[..3]}*{id[^6..]}";
}
return id;
}
// after update of codex-dht, shortID should be consistent!
public static string ToNodeIdShortId(string id)
{
if (id.Length > 10)
{
return $"{id[..2]}*{id[^6..]}";
}
return id;
}
}
}
@@ -1,4 +1,5 @@
using Core;
using CodexPlugin.Hooks;
using Core;
using KubernetesWorkflow.Types;
namespace CodexPlugin
@@ -38,6 +39,11 @@ namespace CodexPlugin
return ci.StartCodexNodes(number, s => { });
}
public static void SetCodexHooksProvider(this CoreInterface ci, ICodexHooksProvider hooksProvider)
{
Plugin(ci).SetCodexHooksProvider(hooksProvider);
}
private static CodexPlugin Plugin(CoreInterface ci)
{
return ci.GetPlugin<CodexPlugin>();
@@ -0,0 +1,71 @@
using GethPlugin;
using Utils;
namespace CodexPlugin.Hooks
{
public interface ICodexHooksProvider
{
ICodexNodeHooks CreateHooks(string nodeName);
}
public class CodexHooksFactory
{
public ICodexHooksProvider Provider { get; set; } = new DoNothingHooksProvider();
public ICodexNodeHooks CreateHooks(string nodeName)
{
return Provider.CreateHooks(nodeName);
}
}
public class DoNothingHooksProvider : ICodexHooksProvider
{
public ICodexNodeHooks CreateHooks(string nodeName)
{
return new DoNothingCodexHooks();
}
}
public class DoNothingCodexHooks : ICodexNodeHooks
{
public void OnFileDownloaded(ByteSize size, ContentId cid)
{
}
public void OnFileDownloading(ContentId cid)
{
}
public void OnFileUploaded(string uid, ByteSize size, ContentId cid)
{
}
public void OnFileUploading(string uid, ByteSize size)
{
}
public void OnNodeStarted(string peerId, string nodeId)
{
}
public void OnNodeStarting(DateTime startUtc, string image, EthAccount? ethAccount)
{
}
public void OnNodeStopping()
{
}
public void OnStorageAvailabilityCreated(StorageAvailability response)
{
}
public void OnStorageContractSubmitted(StoragePurchaseContract storagePurchaseContract)
{
}
public void OnStorageContractUpdated(StoragePurchase purchaseStatus)
{
}
}
}
@@ -0,0 +1,19 @@
using GethPlugin;
using Utils;
namespace CodexPlugin.Hooks
{
public interface ICodexNodeHooks
{
void OnNodeStarting(DateTime startUtc, string image, EthAccount? ethAccount);
void OnNodeStarted(string peerId, string nodeId);
void OnNodeStopping();
void OnFileUploading(string uid, ByteSize size);
void OnFileUploaded(string uid, ByteSize size, ContentId cid);
void OnFileDownloading(ContentId cid);
void OnFileDownloaded(ByteSize size, ContentId cid);
void OnStorageContractSubmitted(StoragePurchaseContract storagePurchaseContract);
void OnStorageContractUpdated(StoragePurchase purchaseStatus);
void OnStorageAvailabilityCreated(StorageAvailability response);
}
}
+52 -32
View File
@@ -16,8 +16,8 @@ namespace CodexPlugin
Spr = debugInfo.Spr,
Addrs = debugInfo.Addrs.ToArray(),
AnnounceAddresses = JArray(debugInfo.AdditionalProperties, "announceAddresses").Select(x => x.ToString()).ToArray(),
Version = MapDebugInfoVersion(JObject(debugInfo.AdditionalProperties, "codex")),
Table = MapDebugInfoTable(JObject(debugInfo.AdditionalProperties, "table"))
Version = Map(debugInfo.Codex),
Table = Map(debugInfo.Table)
};
}
@@ -63,6 +63,26 @@ namespace CodexPlugin
};
}
public StorageAvailability[] Map(ICollection<SalesAvailabilityREAD> availabilities)
{
return availabilities.Select(a => Map(a)).ToArray();
}
public StorageAvailability Map(SalesAvailabilityREAD availability)
{
return new StorageAvailability
(
ToByteSize(availability.TotalSize),
ToTimespan(availability.Duration),
new TestToken(ToBigIng(availability.MinPrice)),
new TestToken(ToBigIng(availability.MaxCollateral))
)
{
Id = availability.Id,
FreeSpace = ToByteSize(availability.FreeSize),
};
}
// TODO: Fix openapi spec for this call.
//public StoragePurchase Map(CodexOpenApi.Purchase purchase)
//{
@@ -105,19 +125,6 @@ namespace CodexPlugin
// };
//}
public StorageAvailability Map(CodexOpenApi.SalesAvailabilityREAD read)
{
return new StorageAvailability(
totalSpace: new ByteSize(Convert.ToInt64(read.TotalSize)),
maxDuration: TimeSpan.FromSeconds(Convert.ToDouble(read.Duration)),
minPriceForTotalSpace: new TestToken(BigInteger.Parse(read.MinPrice)),
maxCollateral: new TestToken(BigInteger.Parse(read.MaxCollateral))
)
{
Id = read.Id
};
}
public CodexSpace Map(Space space)
{
return new CodexSpace
@@ -129,47 +136,45 @@ namespace CodexPlugin
};
}
private DebugInfoVersion MapDebugInfoVersion(JObject obj)
private DebugInfoVersion Map(CodexVersion obj)
{
return new DebugInfoVersion
{
Version = StringOrEmpty(obj, "version"),
Revision = StringOrEmpty(obj, "revision")
Version = obj.Version,
Revision = obj.Revision
};
}
private DebugInfoTable MapDebugInfoTable(JObject obj)
private DebugInfoTable Map(PeersTable obj)
{
return new DebugInfoTable
{
LocalNode = MapDebugInfoTableNode(obj.GetValue("localNode")),
Nodes = MapDebugInfoTableNodeArray(obj.GetValue("nodes") as JArray)
LocalNode = Map(obj.LocalNode),
Nodes = Map(obj.Nodes)
};
}
private DebugInfoTableNode MapDebugInfoTableNode(JToken? token)
private DebugInfoTableNode Map(Node? token)
{
var obj = token as JObject;
if (obj == null) return new DebugInfoTableNode();
if (token == null) return new DebugInfoTableNode();
return new DebugInfoTableNode
{
Address = StringOrEmpty(obj, "address"),
NodeId = StringOrEmpty(obj, "nodeId"),
PeerId = StringOrEmpty(obj, "peerId"),
Record = StringOrEmpty(obj, "record"),
Seen = Bool(obj, "seen")
Address = token.Address,
NodeId = token.NodeId,
PeerId = token.PeerId,
Record = token.Record,
Seen = token.Seen
};
}
private DebugInfoTableNode[] MapDebugInfoTableNodeArray(JArray? nodes)
private DebugInfoTableNode[] Map(ICollection<Node> nodes)
{
if (nodes == null || nodes.Count == 0)
{
return new DebugInfoTableNode[0];
}
return nodes.Select(MapDebugInfoTableNode).ToArray();
return nodes.Select(Map).ToArray();
}
private Manifest MapManifest(CodexOpenApi.ManifestItem manifest)
@@ -222,5 +227,20 @@ namespace CodexPlugin
{
return t.TstWei.ToString("D");
}
private BigInteger ToBigIng(string tokens)
{
return BigInteger.Parse(tokens);
}
private TimeSpan ToTimespan(string duration)
{
return TimeSpan.FromSeconds(Convert.ToInt32(duration));
}
private ByteSize ToByteSize(string size)
{
return new ByteSize(Convert.ToInt64(size));
}
}
}
@@ -1,4 +1,5 @@
using Logging;
using CodexPlugin.Hooks;
using Logging;
using Utils;
namespace CodexPlugin
@@ -6,6 +7,7 @@ namespace CodexPlugin
public interface IMarketplaceAccess
{
string MakeStorageAvailable(StorageAvailability availability);
StorageAvailability[] GetAvailabilities();
IStoragePurchaseContract RequestStorage(StoragePurchaseRequest purchase);
}
@@ -13,11 +15,13 @@ namespace CodexPlugin
{
private readonly ILog log;
private readonly CodexAccess codexAccess;
private readonly ICodexNodeHooks hooks;
public MarketplaceAccess(ILog log, CodexAccess codexAccess)
public MarketplaceAccess(ILog log, CodexAccess codexAccess, ICodexNodeHooks hooks)
{
this.log = log;
this.codexAccess = codexAccess;
this.hooks = hooks;
}
public IStoragePurchaseContract RequestStorage(StoragePurchaseRequest purchase)
@@ -27,6 +31,7 @@ namespace CodexPlugin
var response = codexAccess.RequestStorage(purchase);
if (string.IsNullOrEmpty(response) ||
response == "Unable to encode manifest" ||
response == "Purchasing not available" ||
response == "Expiry required" ||
response == "Expiry needs to be in future" ||
@@ -37,8 +42,11 @@ namespace CodexPlugin
Log($"Storage requested successfully. PurchaseId: '{response}'.");
var contract = new StoragePurchaseContract(log, codexAccess, response, purchase);
var contract = new StoragePurchaseContract(log, codexAccess, response, purchase, hooks);
contract.WaitForStorageContractSubmitted();
hooks.OnStorageContractSubmitted(contract);
return contract;
}
@@ -49,10 +57,19 @@ namespace CodexPlugin
var response = codexAccess.SalesAvailability(availability);
Log($"Storage successfully made available. Id: {response.Id}");
hooks.OnStorageAvailabilityCreated(response);
return response.Id;
}
public StorageAvailability[] GetAvailabilities()
{
var result = codexAccess.GetAvailabilities();
Log($"Got {result.Length} availabilities:");
foreach (var a in result) a.Log(log);
return result;
}
private void Log(string msg)
{
log.Log($"{codexAccess.Container.Containers.Single().Name} {msg}");
@@ -73,6 +90,12 @@ namespace CodexPlugin
throw new NotImplementedException();
}
public StorageAvailability[] GetAvailabilities()
{
Unavailable();
throw new NotImplementedException();
}
private void Unavailable()
{
FrameworkAssert.Fail("Incorrect test setup: Marketplace was not enabled for this group of Codex nodes. Add 'EnableMarketplace(...)' after 'SetupCodexNodes()' to enable it.");
@@ -84,10 +84,11 @@ namespace CodexPlugin
public TimeSpan MaxDuration { get; }
public TestToken MinPriceForTotalSpace { get; }
public TestToken MaxCollateral { get; }
public ByteSize FreeSpace { get; set; } = ByteSize.Zero;
public void Log(ILog log)
{
log.Log($"Making storage available... (" +
log.Log($"Storage Availability: (" +
$"totalSize: {TotalSpace}, " +
$"maxDuration: {Time.FormatDuration(MaxDuration)}, " +
$"minPriceForTotalSpace: {MinPriceForTotalSpace}, " +
@@ -0,0 +1,128 @@
using CodexPlugin.OverwatchSupport.LineConverters;
using KubernetesWorkflow;
using OverwatchTranscript;
using Utils;
namespace CodexPlugin.OverwatchSupport
{
public class CodexLogConverter
{
private readonly ITranscriptWriter writer;
private readonly CodexTranscriptWriterConfig config;
private readonly IdentityMap identityMap;
public CodexLogConverter(ITranscriptWriter writer, CodexTranscriptWriterConfig config, IdentityMap identityMap)
{
this.writer = writer;
this.config = config;
this.identityMap = identityMap;
}
public void ProcessLog(IDownloadedLog log)
{
var name = DetermineName(log);
var identityIndex = identityMap.GetIndex(name);
var runner = new ConversionRunner(writer, config, identityMap, identityIndex);
runner.Run(log);
}
private string DetermineName(IDownloadedLog log)
{
// Expected string:
// Downloading container log for '<Downloader1>'
var nameLine = log.FindLinesThatContain("Downloading container log for").First();
return Str.Between(nameLine, "'", "'");
}
}
public class ConversionRunner
{
private readonly ITranscriptWriter writer;
private readonly IdentityMap nameIdMap;
private readonly int nodeIdentityIndex;
private readonly ILineConverter[] converters;
public ConversionRunner(ITranscriptWriter writer, CodexTranscriptWriterConfig config, IdentityMap nameIdMap, int nodeIdentityIndex)
{
this.nodeIdentityIndex = nodeIdentityIndex;
this.writer = writer;
this.nameIdMap = nameIdMap;
converters = CreateConverters(config).ToArray();
}
private IEnumerable<ILineConverter> CreateConverters(CodexTranscriptWriterConfig config)
{
if (config.IncludeBlockReceivedEvents)
{
yield return new BlockReceivedLineConverter();
}
yield return new BootstrapLineConverter();
yield return new DialSuccessfulLineConverter();
yield return new PeerDroppedLineConverter();
}
public void Run(IDownloadedLog log)
{
log.IterateLines(line =>
{
foreach (var converter in converters)
{
ProcessLine(line, converter);
}
});
}
private void AddEvent(DateTime utc, Action<OverwatchCodexEvent> action)
{
var e = new OverwatchCodexEvent
{
NodeIdentity = nodeIdentityIndex,
};
action(e);
e.Write(utc, writer);
}
private void ProcessLine(string line, ILineConverter converter)
{
if (!line.Contains(converter.Interest)) return;
var codexLine = CodexLogLine.Parse(line);
if (codexLine == null) throw new Exception("Unable to parse required line");
EnsureFullIds(codexLine);
converter.Process(codexLine, (action) =>
{
AddEvent(codexLine.TimestampUtc, action);
});
}
private void EnsureFullIds(CodexLogLine codexLine)
{
// The issue is: node IDs occure both in full and short version.
// Downstream tools will assume that a node ID string-equals its own ID.
// So we replace all shortened IDs we can find with their full ones.
// Usually, the shortID appears as the entire string of an attribute:
// "peerId=123*567890"
// But sometimes, it is part of a larger string:
// "thing=abc:123*567890,def"
foreach (var pair in codexLine.Attributes)
{
if (pair.Value.Contains("*"))
{
codexLine.Attributes[pair.Key] = nameIdMap.ReplaceShortIds(pair.Value);
}
}
}
}
public interface ILineConverter
{
string Interest { get; }
void Process(CodexLogLine line, Action<Action<OverwatchCodexEvent>> addEvent);
}
}
@@ -0,0 +1,191 @@
using CodexPlugin.Hooks;
using GethPlugin;
using OverwatchTranscript;
using Utils;
namespace CodexPlugin.OverwatchSupport
{
public class CodexNodeTranscriptWriter : ICodexNodeHooks
{
private readonly ITranscriptWriter writer;
private readonly IdentityMap identityMap;
private readonly string name;
private int identityIndex = -1;
private readonly List<(DateTime, OverwatchCodexEvent)> pendingEvents = new List<(DateTime, OverwatchCodexEvent)>();
public CodexNodeTranscriptWriter(ITranscriptWriter writer, IdentityMap identityMap, string name)
{
this.writer = writer;
this.identityMap = identityMap;
this.name = name;
}
public void OnNodeStarting(DateTime startUtc, string image, EthAccount? ethAccount)
{
WriteCodexEvent(startUtc, e =>
{
e.NodeStarting = new NodeStartingEvent
{
Image = image,
EthAddress = ethAccount != null ? ethAccount.ToString() : ""
};
});
}
public void OnNodeStarted(string peerId, string nodeId)
{
if (string.IsNullOrEmpty(peerId) || string.IsNullOrEmpty(nodeId))
{
throw new Exception("Node started - peerId and/or nodeId unknown.");
}
identityMap.Add(name, peerId, nodeId);
identityIndex = identityMap.GetIndex(name);
WriteCodexEvent(e =>
{
e.NodeStarted = new NodeStartedEvent
{
};
});
}
public void OnNodeStopping()
{
WriteCodexEvent(e =>
{
e.NodeStopping = new NodeStoppingEvent
{
};
});
}
public void OnFileDownloading(ContentId cid)
{
WriteCodexEvent(e =>
{
e.FileDownloading = new FileDownloadingEvent
{
Cid = cid.Id
};
});
}
public void OnFileDownloaded(ByteSize size, ContentId cid)
{
WriteCodexEvent(e =>
{
e.FileDownloaded = new FileDownloadedEvent
{
Cid = cid.Id,
ByteSize = size.SizeInBytes
};
});
}
public void OnFileUploading(string uid, ByteSize size)
{
WriteCodexEvent(e =>
{
e.FileUploading = new FileUploadingEvent
{
UniqueId = uid,
ByteSize = size.SizeInBytes
};
});
}
public void OnFileUploaded(string uid, ByteSize size, ContentId cid)
{
WriteCodexEvent(e =>
{
e.FileUploaded = new FileUploadedEvent
{
UniqueId = uid,
Cid = cid.Id,
ByteSize = size.SizeInBytes
};
});
}
public void OnStorageContractSubmitted(StoragePurchaseContract storagePurchaseContract)
{
WriteCodexEvent(e =>
{
e.StorageContractSubmitted = new StorageContractSubmittedEvent
{
PurchaseId = storagePurchaseContract.PurchaseId,
PurchaseRequest = storagePurchaseContract.Purchase
};
});
}
public void OnStorageContractUpdated(StoragePurchase purchaseStatus)
{
WriteCodexEvent(e =>
{
e.StorageContractUpdated = new StorageContractUpdatedEvent
{
StoragePurchase = purchaseStatus
};
});
}
public void OnStorageAvailabilityCreated(StorageAvailability response)
{
WriteCodexEvent(e =>
{
e.StorageAvailabilityCreated = new StorageAvailabilityCreatedEvent
{
StorageAvailability = response
};
});
}
private void WriteCodexEvent(Action<OverwatchCodexEvent> action)
{
WriteCodexEvent(DateTime.UtcNow, action);
}
private void WriteCodexEvent(DateTime utc, Action<OverwatchCodexEvent> action)
{
var e = new OverwatchCodexEvent
{
NodeIdentity = identityIndex
};
action(e);
if (identityIndex < 0)
{
// If we don't know our id, don't write the events yet.
AddToCache(utc, e);
}
else
{
e.Write(utc, writer);
// Write any events that we cached when we didn't have our id yet.
WriteAndClearCache();
}
}
private void AddToCache(DateTime utc, OverwatchCodexEvent e)
{
pendingEvents.Add((utc, e));
}
private void WriteAndClearCache()
{
if (pendingEvents.Any())
{
foreach (var pair in pendingEvents)
{
pair.Item2.NodeIdentity = identityIndex;
pair.Item2.Write(pair.Item1, writer);
}
pendingEvents.Clear();
}
}
}
}
@@ -0,0 +1,91 @@
using CodexPlugin.Hooks;
using KubernetesWorkflow;
using Logging;
using OverwatchTranscript;
using Utils;
namespace CodexPlugin.OverwatchSupport
{
public class CodexTranscriptWriter : ICodexHooksProvider
{
private const string CodexHeaderKey = "cdx_h";
private readonly ILog log;
private readonly CodexTranscriptWriterConfig config;
private readonly ITranscriptWriter writer;
private readonly CodexLogConverter converter;
private readonly IdentityMap identityMap = new IdentityMap();
private readonly KademliaPositionFinder positionFinder = new KademliaPositionFinder();
public CodexTranscriptWriter(ILog log, CodexTranscriptWriterConfig config, ITranscriptWriter transcriptWriter)
{
this.log = log;
this.config = config;
writer = transcriptWriter;
converter = new CodexLogConverter(writer, config, identityMap);
}
public void Finalize(string outputFilepath)
{
log.Log("Finalizing Codex transcript...");
writer.AddHeader(CodexHeaderKey, CreateCodexHeader());
writer.Write(outputFilepath);
log.Log("Done");
}
public ICodexNodeHooks CreateHooks(string nodeName)
{
nodeName = Str.Between(nodeName, "'", "'");
return new CodexNodeTranscriptWriter(writer, identityMap, nodeName);
}
public void IncludeFile(string filepath)
{
writer.IncludeArtifact(filepath);
}
public void ProcessLogs(IDownloadedLog[] downloadedLogs)
{
foreach (var l in downloadedLogs)
{
log.Log("Include artifact: " + l.GetFilepath());
writer.IncludeArtifact(l.GetFilepath());
// Not all of these logs are necessarily Codex logs.
// Check, and process only the Codex ones.
if (IsCodexLog(l))
{
log.Log("Processing Codex log: " + l.GetFilepath());
converter.ProcessLog(l);
}
}
}
public void AddResult(bool success, string result)
{
writer.Add(DateTime.UtcNow, new OverwatchCodexEvent
{
NodeIdentity = -1,
ScenarioFinished = new ScenarioFinishedEvent
{
Success = success,
Result = result
}
});
}
private OverwatchCodexHeader CreateCodexHeader()
{
return new OverwatchCodexHeader
{
Nodes = positionFinder.DeterminePositions(identityMap.Get())
};
}
private bool IsCodexLog(IDownloadedLog log)
{
return log.GetLinesContaining("Run Codex node").Any();
}
}
}
@@ -0,0 +1,12 @@
namespace CodexPlugin.OverwatchSupport
{
public class CodexTranscriptWriterConfig
{
public CodexTranscriptWriterConfig(bool includeBlockReceivedEvents)
{
IncludeBlockReceivedEvents = includeBlockReceivedEvents;
}
public bool IncludeBlockReceivedEvents { get; }
}
}
@@ -0,0 +1,63 @@
namespace CodexPlugin.OverwatchSupport
{
public class IdentityMap
{
private readonly List<CodexNodeIdentity> nodes = new List<CodexNodeIdentity>();
private readonly Dictionary<string, int> nameIndexMap = new Dictionary<string, int>();
private readonly Dictionary<string, string> shortToLong = new Dictionary<string, string>();
public void Add(string name, string peerId, string nodeId)
{
Add(new CodexNodeIdentity
{
Name = name,
PeerId = peerId,
NodeId = nodeId
});
nameIndexMap.Add(name, nameIndexMap.Count);
}
public void Add(CodexNodeIdentity identity)
{
if (string.IsNullOrWhiteSpace(identity.Name)) throw new Exception("Name required");
if (string.IsNullOrWhiteSpace(identity.PeerId) || identity.PeerId.Length < 11) throw new Exception("PeerId invalid");
if (string.IsNullOrWhiteSpace(identity.NodeId) || identity.NodeId.Length < 11) throw new Exception("NodeId invalid");
nodes.Add(identity);
shortToLong.Add(CodexUtils.ToShortId(identity.PeerId), identity.PeerId);
shortToLong.Add(CodexUtils.ToNodeIdShortId(identity.NodeId), identity.NodeId);
}
public CodexNodeIdentity[] Get()
{
return nodes.ToArray();
}
public int GetIndex(string name)
{
return nameIndexMap[name];
}
public CodexNodeIdentity GetId(string name)
{
return nodes.Single(n => n.Name == name);
}
public string ReplaceShortIds(string value)
{
var result = value;
foreach (var pair in shortToLong)
{
result = result.Replace(pair.Key, pair.Value);
}
return result;
}
public int Size
{
get { return nodes.Count; }
}
}
}
@@ -0,0 +1,68 @@
using System.Numerics;
using Utils;
using YamlDotNet.Core.Tokens;
namespace CodexPlugin.OverwatchSupport
{
public class KademliaPositionFinder
{
public CodexNodeIdentity[] DeterminePositions(CodexNodeIdentity[] identities)
{
var zero = identities.First();
var distances = CalculateDistances(zero, identities);
var maxDistance = distances.Values.Max();
CalculateNormalizedPositions(distances, maxDistance);
return identities;
}
private Dictionary<CodexNodeIdentity, BigInteger> CalculateDistances(CodexNodeIdentity zero, CodexNodeIdentity[] identities)
{
var result = new Dictionary<CodexNodeIdentity, BigInteger>();
foreach (var id in identities.Skip(1))
{
result.Add(id, GetDistance(zero.NodeId, id.NodeId));
}
return result;
}
private BigInteger GetDistance(string id1, string id2)
{
var one = BigInteger.Parse(id1, System.Globalization.NumberStyles.HexNumber).ToByteArray();
var two = BigInteger.Parse(id2, System.Globalization.NumberStyles.HexNumber).ToByteArray();
var x = Xor(one, two);
return new BigInteger(x, isUnsigned: true);
}
private byte[] Xor(byte[] one, byte[] two)
{
if (one.Length != two.Length) throw new Exception("Not equal length");
var result = new byte[one.Length];
for (int i = 0; i < one.Length; i++)
{
uint a = one[i];
uint b = two[i];
uint c = (a ^ b);
result[i] = (byte)c;
}
return result;
}
private void CalculateNormalizedPositions(Dictionary<CodexNodeIdentity, BigInteger> distances, BigInteger maxDistance)
{
foreach (var pair in distances)
{
pair.Key.KademliaNormalizedPosition = DeterminePosition(pair.Value, maxDistance);
}
}
private float DeterminePosition(BigInteger value, BigInteger maxDistance)
{
var f = (value * 10000) / maxDistance;
return ((float)f) / 10000.0f;
}
}
}
@@ -0,0 +1,44 @@
namespace CodexPlugin.OverwatchSupport.LineConverters
{
public class BlockReceivedLineConverter : ILineConverter
{
public string Interest => "Received blocks from peer";
public void Process(CodexLogLine line, Action<Action<OverwatchCodexEvent>> addEvent)
{
var peer = line.Attributes["peer"];
var blockAddresses = line.Attributes["blocks"];
SplitBlockAddresses(blockAddresses, address =>
{
addEvent(e =>
{
e.BlockReceived = new BlockReceivedEvent
{
SenderPeerId = peer,
BlockAddress = address
};
});
});
}
private void SplitBlockAddresses(string blockAddresses, Action<string> onBlockAddress)
{
// Single line can contain multiple block addresses.
var tokens = blockAddresses.Split(",", StringSplitOptions.RemoveEmptyEntries).ToList();
while (tokens.Count > 0)
{
if (tokens.Count == 1)
{
onBlockAddress(tokens[0]);
return;
}
var blockAddress = $"{tokens[0]}, {tokens[1]}";
tokens.RemoveRange(0, 2);
onBlockAddress(blockAddress);
}
}
}
}
@@ -0,0 +1,71 @@
namespace CodexPlugin.OverwatchSupport.LineConverters
{
public class BootstrapLineConverter : ILineConverter
{
private const string peerIdTag = "peerId: ";
public string Interest => "Starting codex node";
public void Process(CodexLogLine line, Action<Action<OverwatchCodexEvent>> addEvent)
{
// "(
// configFile: none(InputFile),
// logLevel: \"TRACE;warn:discv5,providers,manager,cache;warn:libp2p,multistream,switch,transport,tcptransport,semaphore,asyncstreamwrapper,lpstream,mplex,mplexchannel,noise,bufferstream,mplexcoder,secure,chronosstream,connection,connmanager,websock,ws-session,dialer,muxedupgrade,upgrade,identify;warn:contracts,clock;warn:serde,json,serialization\",
// logFormat: auto,
// metricsEnabled: false,
// metricsAddress: 127.0.0.1,
// metricsPort: 8008,
// dataDir: datadir5,
// circuitDir: /root/.cache/codex/circuits,
// listenAddrs: @[/ip4/0.0.0.0/tcp/8081],
// nat: 10.1.0.214,
// discoveryIp: 0.0.0.0,
// discoveryPort: 8080,
// netPrivKeyFile: \"key\",
// bootstrapNodes:
// @[(envelope: (publicKey: secp256k1 key (0414380858330307a4a59e8a1643512f80680dedb8c541674f40382be71c26556cd65b7e1775ec9fabfaf6d58d562b88a3c8afb969f8cc256db20e4c4c9e1f70a6),
// domain: \"libp2p-peer-record\",
// payloadType: @[3, 1],
// payload: @[10, 39, 0, 37, 8, 2, 18, 33, 2, 20, 56, 8, 88, 51, 3, 7, 164, 165, 158, 138, 22, 67, 81, 47, 128, 104, 13, 237, 184, 197, 65, 103, 79, 64, 56, 43, 231, 28, 38, 85, 108, 16, 254, 211, 141, 181, 6, 26, 11, 10, 9, 4, 10, 1, 0, 210, 145, 2, 31, 144],
// signature: 3045022100FA846871D96EDCA579990244B1C590E16B57A7BDB817908A2B580043F8A0B0280220342825DBE577E83C22CD59AA9099DE8F8DC86A38C659F60E8B9255126CF37FDE),
// data:
// (peerId: 16Uiu2HAkvnbgwdB2NmNe1uWGJxE3Ep3sDHxNW95W2rTPs2vfxvou,
// seqNo: 1721985534,
// addresses: @[(address: /ip4/10.1.0.210/udp/8080)]
// )
// )],
// maxPeers: 160,
// agentString: \"Codex\",
// apiBindAddress: \"0.0.0.0\",
// apiPort: 30035,
// apiCorsAllowedOrigin: none(string),
// repoKind: fs,
// storageQuota: 8589934592\'NByte,
// blockTtl: 1d,
// blockMaintenanceInterval: 10m,
// blockMaintenanceNumberOfBlocks: 1000,
// cacheSize: 0\'NByte,
// logFile: none(string),
// cmd: noCmd
// )"
var config = line.Attributes["config"];
while (config.Contains(peerIdTag))
{
var openIndex = config.IndexOf(peerIdTag) + peerIdTag.Length;
var closeIndex = config.IndexOf(",", openIndex);
var bootPeerId = config.Substring(openIndex, closeIndex - openIndex);
config = config.Substring(closeIndex);
addEvent(e =>
{
e.BootstrapConfig = new BootstrapConfigEvent
{
BootstrapPeerId = bootPeerId
};
});
}
}
}
}
@@ -0,0 +1,20 @@
namespace CodexPlugin.OverwatchSupport.LineConverters
{
public class DialSuccessfulLineConverter : ILineConverter
{
public string Interest => "Dial successful";
public void Process(CodexLogLine line, Action<Action<OverwatchCodexEvent>> addEvent)
{
var peerId = line.Attributes["peerId"];
addEvent(e =>
{
e.DialSuccessful = new PeerDialSuccessfulEvent
{
TargetPeerId = peerId
};
});
}
}
}
@@ -0,0 +1,20 @@
namespace CodexPlugin.OverwatchSupport.LineConverters
{
public class PeerDroppedLineConverter : ILineConverter
{
public string Interest => "Dropping peer";
public void Process(CodexLogLine line, Action<Action<OverwatchCodexEvent>> addEvent)
{
var peerId = line.Attributes["peer"];
addEvent(e =>
{
e.PeerDropped = new PeerDroppedEvent
{
DroppedPeerId = peerId
};
});
}
}
}

Some files were not shown because too many files have changed in this diff Show More