Skip to content

feat: support multibuckets by recieving bucket from task parameters #124

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Draft
asafMasa wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

feat: support multibuckets by recieving bucket from task parameters #124

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f59ebec
feat: add multi bucket support
asafmas-rnd Nov 1, 2023
d1ab66b
feat: get bucket from task parameter
asafmas-rnd Nov 2, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 5 additions & 9 deletions MergerCli/Properties/launchSettings.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,10 @@
"GPKG to S3": {
"workingDirectory": "",
"commandName": "Project",
"commandLineArgs": "Program.cs 1000 jpeg s3 area2-copy gpkg area2.gpkg",
"commandLineArgs": "Program.cs 1000 jpeg s3 bucket area2-copy gpkg area2.gpkg",
"environmentVariables": {
"AWS_ACCESS_KEY_ID": "minioadmin",
"AWS_SECRET_ACCESS_KEY": "minioadmin",
"S3__bucket": "dtm"
"AWS_SECRET_ACCESS_KEY": "minioadmin"
}
},
"s3 to gpkg": {
Expand All @@ -16,8 +15,7 @@
"commandLineArgs": "Program.cs 1000 jpeg gpkg area2-copy.gpkg -180,-90,180,90 s3 area3Cli",
"environmentVariables": {
"AWS_ACCESS_KEY_ID": "minio123",
"AWS_SECRET_ACCESS_KEY": "minio123",
"S3__bucket": "tiles"
"AWS_SECRET_ACCESS_KEY": "minio123"
}
},
"resume": {
Expand All @@ -27,8 +25,7 @@
"environmentVariables": {
"AWS_ACCESS_KEY_ID": "minio123",
"AWS_SECRET_ACCESS_KEY": "minio123",
"S3__url": "http://custom.localhost:9000",
"S3__bucket": "tiles"
"S3__url": "http://custom.localhost:9000"
}
},
"Test SSL": {
Expand All @@ -39,8 +36,7 @@
"ASPNETCORE_ENVIRONMENT": "Development",
"AWS_ACCESS_KEY_ID": "Q3AM3UQ867SPQQA43P2F",
"AWS_SECRET_ACCESS_KEY": "zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
"S3__url": "https://play.min.io:9000",
"S3__bucket": "tiles"
"S3__url": "https://play.min.io:9000"
}
},
"Docker": {
Expand Down
59 changes: 50 additions & 9 deletions MergerCli/SourceParser.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,8 @@ public SourceParser(IDataFactory dataFactory, ILogger<SourceParser> logger)
this._logger = logger;
}

private void LogDataErrorAndExit(bool isBase, Exception e) {
private void LogDataErrorAndExit(bool isBase, Exception e)
{
string source = isBase ? "base" : "new";
this._logger.LogError($"{source} data does not exist, error: {e.Message}");
Environment.Exit(1);
Expand Down Expand Up @@ -53,7 +54,6 @@ public List<IData> ParseSources(string[] args, int batchSize, out TileFormat for

break;
case "fs":
case "s3":
try
{
sources.Add(this.ParseFileSource(args, ref idx, batchSize, isBase));
Expand All @@ -62,7 +62,16 @@ public List<IData> ParseSources(string[] args, int batchSize, out TileFormat for
{
this.LogDataErrorAndExit(isBase, e);
}

break;
case "s3":
try
{
sources.Add(this.ParseS3Source(args, ref idx, batchSize, isBase));
}
catch (Exception e)
{
this.LogDataErrorAndExit(isBase, e);
}
break;
case "wmts":
case "xyz":
Expand Down Expand Up @@ -115,7 +124,29 @@ private IData ParseFileSource(string[] args, ref int idx, int batchSize, bool is

idx += paramCount;
Grid? grid = GetGrid(isOneXOne);
return this._dataFactory.CreateDataSource(sourceType, sourcePath, batchSize, grid, origin, null, isBase);
return this._dataFactory.CreateDataSource(sourceType, sourcePath, string.Empty, batchSize, grid, origin, null, isBase);
}

private IData ParseS3Source(string[] args, ref int idx, int batchSize, bool isBase)
{
const int requiredParamCount = 3;
const int optionalParamCount = 2;
int paramCount = this.ValidateAndGetSourceLength(args, idx, requiredParamCount, optionalParamCount);
string sourceType = args[idx];
string bucket = args[idx + 1];
string sourcePath = args[idx + 2];
bool? isOneXOne = null;
GridOrigin? origin = null;
if (paramCount > requiredParamCount)
{
// not using set as it allows optional prams with dynamic values aka. --minZoom 3
var optionalParams = args.Skip(idx + requiredParamCount).Take(optionalParamCount).ToArray();
this.ParseOptionalParameters(sourceType, sourcePath, ref isOneXOne, ref origin, optionalParams);
}

idx += paramCount;
Grid? grid = GetGrid(isOneXOne);
return this._dataFactory.CreateDataSource(sourceType, sourcePath, bucket, batchSize, grid, origin, null, isBase);
}

private IData ParseGpkgSource(string[] args, ref int idx, int batchSize, bool isBase)
Expand All @@ -142,7 +173,7 @@ private IData ParseGpkgSource(string[] args, ref int idx, int batchSize, bool is

idx += paramCount;
Grid? grid = GetGrid(isOneXOne);
return this._dataFactory.CreateDataSource(sourceType, sourcePath, batchSize, grid, origin, extent, isBase);
return this._dataFactory.CreateDataSource(sourceType, sourcePath, string.Empty, batchSize, grid, origin, extent, isBase);
}

private IData ParseHttpSource(string[] args, ref int idx, int batchSize, bool isBase)
Expand All @@ -166,7 +197,7 @@ private IData ParseHttpSource(string[] args, ref int idx, int batchSize, bool is

idx += paramCount;
Grid? grid = GetGrid(isOneXOne);
return this._dataFactory.CreateDataSource(sourceType, sourcePath, batchSize, isBase, extent, maxZoom,
return this._dataFactory.CreateDataSource(sourceType, sourcePath, string.Empty, batchSize, isBase, extent, maxZoom,
minZoom, grid, origin);
}

Expand All @@ -183,7 +214,7 @@ private Extent parseExtent(string extentString)
return extent;
}

private int ParseOptionalParameters(string sourceType, string sourcePath, ref bool? isOneXOne,
private int ParseOptionalParameters(string sourceType, string sourcePath, ref bool? isOneXOne,
ref GridOrigin? origin, string[] optionalParams)
{
int parsed = 0;
Expand All @@ -210,11 +241,21 @@ private int ParseOptionalParameters(string sourceType, string sourcePath, ref bo
parsed++;
}

/*if (optionalParams.Contains("--Bucket"))
Copy link
Collaborator

@shimoncohen shimoncohen Dec 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is this commented?

{
if (bucket != null)
{
throw new Exception($"layer {sourceType} {sourcePath} can't have 2 buckets");
}

bucket = GridOrigin.LOWER_LEFT;
parsed++;
}*/

return parsed;
}

private int ValidateAndGetSourceLength(string[] args, int startIdx, int minExpectedParamCount,
int optionalParamCount)
private int ValidateAndGetSourceLength(string[] args, int startIdx, int minExpectedParamCount, int optionalParamCount)
{
int i = startIdx + 1;
// check required parameters
Expand Down
1 change: 0 additions & 1 deletion MergerCli/appsettings.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
},
"S3": {
"url": "http://localhost:9000",
"bucket": "tiles",
"request": {
"timeoutSec": 5,
"retries": 3
Expand Down
5 changes: 4 additions & 1 deletion MergerLogic/Clients/IS3Client.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
using MergerLogic.Batching;
using Amazon.S3.Model;
using MergerLogic.Batching;
using MergerLogic.Utils;

namespace MergerLogic.Clients
Expand All @@ -7,5 +8,7 @@ public interface IS3Client : IDataUtils
{
void UpdateTile(Tile tile);
Tile? GetTile(string key);
string Bucket { get; }
Copy link
Collaborator

@shimoncohen shimoncohen Dec 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is this needed?

ListObjectsV2Response ListObject(ref string? continuationToken, string prefix, string startAfter, int? maxKeys = null);
}
}
49 changes: 42 additions & 7 deletions MergerLogic/Clients/S3Client.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,19 +11,22 @@ namespace MergerLogic.Clients
{
public class S3Client : DataUtils, IS3Client
{
private readonly string _bucket;

private string _bucket;
private readonly IAmazonS3 _client;
private readonly ILogger _logger;
private readonly IPathUtils _pathUtils;

public S3Client(IAmazonS3 client, IPathUtils pathUtils, IGeoUtils geoUtils, IImageFormatter formatter, ILogger<S3Client> logger,
string bucket, string path) : base(path, geoUtils, formatter)
public S3Client(IAmazonS3 client, IPathUtils pathUtils, IGeoUtils geoUtils, IImageFormatter formatter, ILogger<S3Client> logger, string bucket, string path) : base(path, geoUtils, formatter)
{
this._client = client;
this._bucket = bucket;
this._client = client ?? throw new Exception("s3 configuration is required");
this._pathUtils = pathUtils;
this._logger = logger;
this._bucket = bucket;
}

public string Bucket
Copy link
Collaborator

@shimoncohen shimoncohen Dec 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why is this needed?

{
get { return this._bucket; }
}

private byte[]? GetImageBytes(string key)
Expand Down Expand Up @@ -104,7 +107,9 @@ public void UpdateTile(Tile tile)

var request = new PutObjectRequest()
{
BucketName = this._bucket, CannedACL = S3CannedACL.PublicRead, Key = String.Format(key)
BucketName = this._bucket,
CannedACL = S3CannedACL.PublicRead,
Key = String.Format(key)
};

byte[] buffer = tile.GetImageBytes();
Expand All @@ -130,5 +135,35 @@ public void UpdateTile(Tile tile)
this._logger.LogDebug($"[{methodName}] end z: {z}, x: {x}, y: {y}");
return result;
}

public ListObjectsV2Response ListObject(ref string? continuationToken, string prefix, string startAfter, int? maxKeys = null)
{
ListObjectsV2Request listRequests = null;
if (maxKeys == null)
{
listRequests = new ListObjectsV2Request
{
BucketName = this._bucket,
Prefix = prefix,
StartAfter = startAfter,
ContinuationToken = continuationToken
};
}
else
{
listRequests = new ListObjectsV2Request
{
BucketName = this._bucket,
Prefix = prefix,
StartAfter = startAfter,
ContinuationToken = continuationToken,
MaxKeys = maxKeys.Value
Copy link
Collaborator

@shimoncohen shimoncohen Dec 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Only add this parameter it maxKeys is not null.

};
}

var task = this._client.ListObjectsV2Async(listRequests);
var response = task.Result;
return response;
}
}
}
4 changes: 2 additions & 2 deletions MergerLogic/DataTypes/Data.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ public abstract class Data<TUtilsType> : IData where TUtilsType : IDataUtils
protected ValFromCoordFunction ConvertOriginCoord;

protected Data(IServiceProvider container, DataType type, string path, int batchSize, Grid? grid,
GridOrigin? origin, bool isBase, Extent? extent = null)
GridOrigin? origin, bool isBase, Extent? extent = null, string? bucket = null)
Copy link
Collaborator

@shimoncohen shimoncohen Dec 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, data should not contain a bucket value.

{
string methodName = MethodBase.GetCurrentMethod().Name;
this._container = container;
Expand All @@ -88,7 +88,7 @@ protected Data(IServiceProvider container, DataType type, string path, int batch
this.Path = path;
this.BatchSize = batchSize;
var utilsFactory = container.GetRequiredService<IUtilsFactory>();
this.Utils = utilsFactory.GetDataUtils<TUtilsType>(path);
this.Utils = utilsFactory.GetDataUtils<TUtilsType>(path, bucket);
this.GeoUtils = container.GetRequiredService<IGeoUtils>();
this.Grid = grid ?? this.DefaultGrid();
this.Origin = origin ?? this.DefaultOrigin();
Expand Down
60 changes: 10 additions & 50 deletions MergerLogic/DataTypes/S3.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,18 +1,14 @@
using Amazon.S3;
using Amazon.S3.Model;
using MergerLogic.Batching;
using MergerLogic.Clients;
using MergerLogic.Utils;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using System.Reflection;

namespace MergerLogic.DataTypes
{
public class S3 : Data<IS3Client>
{
private IAmazonS3 _client;
private string _bucket;
private readonly List<int> _zoomLevels;
private IEnumerator<int> _zoomEnumerator;
private string? _continuationToken;
Expand All @@ -22,8 +18,8 @@ public class S3 : Data<IS3Client>

private readonly IPathUtils _pathUtils;

public S3(IPathUtils pathUtils, IServiceProvider container, string path, int batchSize, Grid? grid, GridOrigin? origin, bool isBase)
: base(container, DataType.S3, path, batchSize, grid, origin, isBase)
public S3(IPathUtils pathUtils, IServiceProvider container, string bucket, string path, int batchSize, Grid? grid, GridOrigin? origin, bool isBase)
: base(container, DataType.S3, path, batchSize, grid, origin, isBase, null, bucket)
{
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] Ctor started");
this._pathUtils = pathUtils;
Expand All @@ -40,12 +36,6 @@ public S3(IPathUtils pathUtils, IServiceProvider container, string path, int bat

protected override void Initialize()
Copy link
Collaborator

@shimoncohen shimoncohen Dec 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove completly if not needed...

{
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start");
var configurationManager = this._container.GetRequiredService<IConfigurationManager>();
var client = this._container.GetService<IAmazonS3>();
this._client = client ?? throw new Exception("s3 configuration is required");
this._bucket = configurationManager.GetConfiguration("S3", "bucket");
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] ended");
}

protected override GridOrigin DefaultOrigin()
Expand Down Expand Up @@ -97,18 +87,7 @@ public override List<Tile> GetNextBatch(out string currentBatchIdentifier, out s
}

string path = $"{this.Path}/{this._zoomEnumerator.Current}/";
var listRequests = new ListObjectsV2Request
{
BucketName = this._bucket,
Prefix = path,
StartAfter = path,
MaxKeys = missingTiles,
ContinuationToken = this._continuationToken
};

var listObjectsTask = this._client.ListObjectsV2Async(listRequests);
var response = listObjectsTask.Result;

ListObjectsV2Response response = this.Utils.ListObject(ref this._continuationToken, path, path, missingTiles);
Copy link
Collaborator

@shimoncohen shimoncohen Dec 9, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I understand the thought, but because of the generic use this is a mistake.
You could move this function to this class instead.

foreach (S3Object item in response.S3Objects)
{
Tile? tile = this.Utils.GetTile(item.Key);
Expand Down Expand Up @@ -145,27 +124,19 @@ public override void setBatchIdentifier(string batchIdentifier)

private bool FolderExists(string directory)
{
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start");
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start, bucket: {this.Utils.Bucket}, directory: {directory}");
directory = $"{this.Path}/{directory}";

var listRequests = new ListObjectsV2Request
{
BucketName = this._bucket,
Prefix = directory,
StartAfter = directory,
MaxKeys = 1
};
var task = this._client.ListObjectsV2Async(listRequests);
var response = task.Result;
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] end");
return response.KeyCount > 0;
string? continuationToken = null;
ListObjectsV2Response response = this.Utils.ListObject(ref continuationToken, directory, directory, 1);
bool exists = response.KeyCount > 0;
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] end, bucket: {this.Utils.Bucket}, directory: {directory}, exists: {exists}");
return exists;
}

public override bool Exists()
{
this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] bucket: {this._bucket}, path: {this.Path}");
bool exists = FolderExists("");
this._logger.LogInformation($"[{MethodBase.GetCurrentMethod().Name}] ended");
return exists;
}

Expand All @@ -174,20 +145,9 @@ public override long TileCount()
this._logger.LogDebug($"[{MethodBase.GetCurrentMethod().Name}] start");
long tileCount = 0;
string? continuationToken = null;

do
{
var listRequests = new ListObjectsV2Request
{
BucketName = this._bucket,
Prefix = this.Path,
StartAfter = this.Path,
ContinuationToken = continuationToken
};

var task = this._client.ListObjectsV2Async(listRequests);
var response = task.Result;

ListObjectsV2Response response = this.Utils.ListObject(ref continuationToken, this.Path, this.Path);
tileCount += response.KeyCount;
continuationToken = response.NextContinuationToken;
} while (continuationToken != null);
Expand Down
Loading