зеркало из https://github.com/Azure/azure-hpc.git
Allow splitting of sequence files
This commit is contained in:
Christian Smith
Родитель
5ec1a6f009
Коммит
11b89a76f3
|
|
@ -42,9 +42,9 @@ namespace Microsoft.Azure.Blast.Web.Controllers.Api
|
|||
}
|
||||
|
||||
[Route("searches/{searchId}/queries"), HttpGet]
|
||||
public IEnumerable<SearchQuery> GetSearchQueries(Guid searchId)
|
||||
public IEnumerable<SearchQueryEntity> GetSearchQueries(Guid searchId)
|
||||
{
|
||||
return _searchProvider.ListSearchQueries(searchId).OrderBy(q => q.InputFilename);
|
||||
return _searchProvider.ListSearchQueries(searchId).OrderBy(q => q.QueryFilename);
|
||||
}
|
||||
|
||||
[Route("searches/{searchId}"), HttpGet]
|
||||
|
|
@ -116,13 +116,89 @@ namespace Microsoft.Azure.Blast.Web.Controllers.Api
|
|||
|
||||
using (var reader = fileInfo.OpenText())
|
||||
{
|
||||
var sequenceText = reader.ReadToEnd();
|
||||
searchInputFiles.Add(new SearchInputFile
|
||||
var fullFilename = file.Headers.ContentDisposition.FileName.Replace("\"", "");
|
||||
|
||||
if (searchModel.SplitSequenceFile)
|
||||
{
|
||||
Filename = file.Headers.ContentDisposition.FileName.Replace("\"", ""),
|
||||
Length = fileInfo.Length,
|
||||
Content = new MemoryStream(Encoding.UTF8.GetBytes(sequenceText)),
|
||||
});
|
||||
var filename = Path.GetFileNameWithoutExtension(fullFilename);
|
||||
var extension = Path.GetExtension(fullFilename);
|
||||
|
||||
int currentSequenceCount = 0;
|
||||
int currentFileCount = 1;
|
||||
string currentSequenceContent = null;
|
||||
string sequenceFilename;
|
||||
|
||||
var line = reader.ReadLine();
|
||||
while (line != null)
|
||||
{
|
||||
if (string.IsNullOrEmpty(line) || line.Trim() == "")
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
if (line.StartsWith(">"))
|
||||
{
|
||||
if (string.IsNullOrEmpty(currentSequenceContent))
|
||||
{
|
||||
// We're the first sequence
|
||||
currentSequenceContent = line;
|
||||
}
|
||||
else
|
||||
{
|
||||
currentSequenceCount++;
|
||||
|
||||
if (currentSequenceCount % searchModel.SequencesPerQuery == 0)
|
||||
{
|
||||
// Flush previous sequence(s)
|
||||
sequenceFilename = string.Format("{0}_part{1}{2}",
|
||||
filename, currentFileCount++, extension);
|
||||
searchInputFiles.Add(new SearchInputFile
|
||||
{
|
||||
Filename = sequenceFilename,
|
||||
Length = Encoding.UTF8.GetByteCount(currentSequenceContent),
|
||||
Content = new MemoryStream(Encoding.UTF8.GetBytes(currentSequenceContent)),
|
||||
});
|
||||
|
||||
currentSequenceContent = line;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Keep appending
|
||||
currentSequenceContent += "\n" + line;
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
currentSequenceContent += "\n" + line;
|
||||
}
|
||||
|
||||
line = reader.ReadLine();
|
||||
}
|
||||
|
||||
if (!string.IsNullOrEmpty(currentSequenceContent))
|
||||
{
|
||||
// Flush the final one
|
||||
sequenceFilename = string.Format("{0}_part{1}{2}",
|
||||
filename, currentFileCount, extension);
|
||||
searchInputFiles.Add(new SearchInputFile
|
||||
{
|
||||
Filename = sequenceFilename,
|
||||
Length = Encoding.UTF8.GetByteCount(currentSequenceContent),
|
||||
Content = new MemoryStream(Encoding.UTF8.GetBytes(currentSequenceContent)),
|
||||
});
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
var sequenceText = reader.ReadToEnd();
|
||||
searchInputFiles.Add(new SearchInputFile
|
||||
{
|
||||
Filename = fullFilename,
|
||||
Length = fileInfo.Length,
|
||||
Content = new MemoryStream(Encoding.UTF8.GetBytes(sequenceText)),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
try
|
||||
|
|
@ -157,6 +233,12 @@ namespace Microsoft.Azure.Blast.Web.Controllers.Api
|
|||
"The program name must be provided with the search");
|
||||
}
|
||||
|
||||
if (searchModel.SplitSequenceFile && searchModel.SequencesPerQuery < 1)
|
||||
{
|
||||
return Request.CreateErrorResponse(HttpStatusCode.BadRequest,
|
||||
"Sequences per query must ne greater than 0 when splitting a sequence file.");
|
||||
}
|
||||
|
||||
if (!searchModel.SearchInputFiles.Any())
|
||||
{
|
||||
return Request.CreateErrorResponse(HttpStatusCode.BadRequest,
|
||||
|
|
@ -191,6 +273,8 @@ namespace Microsoft.Azure.Blast.Web.Controllers.Api
|
|||
Executable = formData["executable"],
|
||||
ExecutableArgs = formData["executableArgs"],
|
||||
SearchInputFiles = new List<SearchInputFile>(),
|
||||
SplitSequenceFile = ToBoolean(formData["splitSequenceFile"]),
|
||||
SequencesPerQuery = ToInt(formData["seqencesPerQuery"], 1),
|
||||
};
|
||||
|
||||
AddPoolSpecToSearch(formData, spec);
|
||||
|
|
@ -231,5 +315,25 @@ namespace Microsoft.Azure.Blast.Web.Controllers.Api
|
|||
spec.PoolDisplayName = formData["poolName"];
|
||||
}
|
||||
}
|
||||
|
||||
private bool ToBoolean(string formValue)
|
||||
{
|
||||
bool value;
|
||||
if (Boolean.TryParse(formValue, out value))
|
||||
{
|
||||
return value;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private int ToInt(string formValue, int defaultValue)
|
||||
{
|
||||
int value;
|
||||
if (int.TryParse(formValue, out value))
|
||||
{
|
||||
return value;
|
||||
}
|
||||
return defaultValue;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -54,7 +54,19 @@
|
|||
<div class="form-group">
|
||||
<label class="control-label col-sm-2" for="searchFile">Sequence file(s)</label>
|
||||
<div class="col-sm-10">
|
||||
<input id="searchFile" name="searchFile" type="file" multiple="multiple" accept="*" />
|
||||
<input id="searchFile" name="searchFile" type="file" multiple="multiple" accept="*"/>
|
||||
</div>
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label class="control-label col-sm-2" for="searchFile">Split sequence file</label>
|
||||
<div class="col-sm-10">
|
||||
<input id="splitSequenceFileCheckBox" type="checkbox" name="splitSequenceFile" value="0">
|
||||
</div>
|
||||
</div>
|
||||
<div class="form-group">
|
||||
<label class="control-label col-sm-2" for="seqencesPerQuery">Sequences per query</label>
|
||||
<div class="col-sm-10">
|
||||
<input id="seqencesPerQuery" type="text" class="form-control" placeholder="1" aria-describedby="basic-addon1" value="1">
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
|
@ -235,6 +247,9 @@
|
|||
|
||||
var searchName = $('#searchName').val().trim();
|
||||
var searchSequenceText = $('#searchSequenceText').val().trim();
|
||||
var splitSequenceFile = $('#splitSequenceFileCheckBox').is(':checked');
|
||||
var seqencesPerQuery = $('#seqencesPerQuery').val().trim();
|
||||
|
||||
|
||||
var poolId = $('#poolListBtn').text().trim();
|
||||
var vmSize = $('#vmSizeListBtn').text().trim();
|
||||
|
|
@ -245,6 +260,8 @@
|
|||
var searchFile = $("#searchFile")[0];
|
||||
formData.append("searchName", searchName);
|
||||
formData.append("searchSequenceText", searchSequenceText);
|
||||
formData.append("splitSequenceFile", splitSequenceFile);
|
||||
formData.append("seqencesPerQuery", seqencesPerQuery);
|
||||
formData.append("databaseName", database);
|
||||
formData.append("executable", executable);
|
||||
formData.append("executableArgs", executableArgs);
|
||||
|
|
|
|||
|
|
@ -116,7 +116,7 @@
|
|||
{
|
||||
<script id="query-template" type="text/x-handlebars-template">
|
||||
<tr>
|
||||
<td>{{inputFilename}}</td>
|
||||
<td>{{queryFilename}}</td>
|
||||
<td>{{state}}</td>
|
||||
<td>{{startTimeFormatted}}</td>
|
||||
<td>{{endTimeFormatted}}</td>
|
||||
|
|
|
|||
|
|
@ -157,10 +157,10 @@
|
|||
<Compile Include="Searches\QueryState.cs" />
|
||||
<Compile Include="Searches\Search.cs" />
|
||||
<Compile Include="Searches\SearchInputFile.cs" />
|
||||
<Compile Include="Searches\SearchQuery.cs" />
|
||||
<Compile Include="Searches\SearchSpecification.cs" />
|
||||
<Compile Include="Searches\SearchState.cs" />
|
||||
<Compile Include="Searches\TimeSpanExtensions.cs" />
|
||||
<Compile Include="Storage\Entities\SearchQueryEntity.cs" />
|
||||
<Compile Include="Storage\InputFileStager.cs" />
|
||||
<Compile Include="Storage\RenewableBlobLease.cs" />
|
||||
<Compile Include="Storage\AzureBlobStorageProvider.cs" />
|
||||
|
|
@ -189,6 +189,9 @@
|
|||
<None Include="packages.config" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<Content Include="BatchScripts\SearchJobManager.py">
|
||||
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
|
||||
</Content>
|
||||
<Content Include="BatchScripts\jobmanager.py">
|
||||
<CopyToOutputDirectory>Always</CopyToOutputDirectory>
|
||||
</Content>
|
||||
|
|
|
|||
|
|
@ -0,0 +1,119 @@
|
|||
import os
|
||||
import sys
|
||||
import datetime
|
||||
import time
|
||||
import azure.batch.batch_service_client as batch
|
||||
import azure.batch.batch_auth as batchauth
|
||||
import azure.batch.models as batchmodels
|
||||
from azure.storage.table import TableService, TableBatch
|
||||
from azure.storage.blob import BlockBlobService
|
||||
|
||||
|
||||
def get_search_state(all_tasks_complete, any_failures):
|
||||
if all_tasks_complete and any_failures:
|
||||
return 'Error'
|
||||
if all_tasks_complete:
|
||||
return 'Complete'
|
||||
return 'Running'
|
||||
|
||||
|
||||
def get_query_state(task):
|
||||
if task.state == batchmodels.TaskState.active:
|
||||
return 'Waiting'
|
||||
if task.state == batchmodels.TaskState.preparing:
|
||||
return 'Waiting'
|
||||
if task.state == batchmodels.TaskState.running:
|
||||
return 'Running'
|
||||
if task.state == batchmodels.TaskState.completed:
|
||||
if task.execution_info.exit_code == 0:
|
||||
return 'Success'
|
||||
return 'Error'
|
||||
|
||||
def wait_for_tasks_to_complete(
|
||||
table_service, batch_client, entity_pk, entity_rk, job_id):
|
||||
"""
|
||||
Returns when all tasks in the specified job reach the Completed state.
|
||||
"""
|
||||
|
||||
while True:
|
||||
entity = table_service.get_entity(
|
||||
'SearchEntity', entity_pk, entity_rk)
|
||||
|
||||
tasks = batch_client.task.list(job_id)
|
||||
|
||||
incomplete_tasks = [task for task in tasks if
|
||||
task.id != 'JobManager' and
|
||||
task.state != batchmodels.TaskState.completed]
|
||||
complete_tasks = [task for task in tasks if
|
||||
task.id != 'JobManager' and
|
||||
task.state == batchmodels.TaskState.completed]
|
||||
failed_tasks = [task for task in complete_tasks if
|
||||
task.execution_info.exit_code != 0 or
|
||||
task.execution_info.scheduling_error is not None]
|
||||
|
||||
queries = table_service.query_entities(
|
||||
'SearchQueryEntity',
|
||||
filter="PartitionKey eq '{}'".format(entity.RowKey))
|
||||
|
||||
updateBatch = TableBatch()
|
||||
|
||||
for task in tasks:
|
||||
matching_queries = [q for q in queries if q.RowKey == task.id]
|
||||
if not matching_queries:
|
||||
print('Could not find query {}'.format(task.id))
|
||||
continue
|
||||
query = matching_queries[0]
|
||||
update = False
|
||||
state = get_query_state(task)
|
||||
if query._State != state:
|
||||
query._State = state
|
||||
update = True
|
||||
|
||||
if task.state == batchmodels.TaskState.running:
|
||||
if not hasattr(query, 'StartTime'):
|
||||
query.StartTime = task.execution_info.start_time
|
||||
update = True
|
||||
|
||||
if task.state == batchmodels.TaskState.completed:
|
||||
if not hasattr(query, 'EndTime'):
|
||||
query.EndTime = task.execution_info.end_time
|
||||
update = True
|
||||
|
||||
if update:
|
||||
updateBatch.update_entity(query)
|
||||
|
||||
table_service.commit_batch('SearchQueryEntity', updateBatch)
|
||||
|
||||
all_tasks_complete = not incomplete_tasks
|
||||
any_failures = len(failed_tasks) > 0
|
||||
|
||||
entity.CompletedTasks = len(complete_tasks)
|
||||
entity._State = get_search_state(all_tasks_complete, any_failures)
|
||||
|
||||
if not incomplete_tasks:
|
||||
entity.EndTime = datetime.datetime.utcnow()
|
||||
table_service.update_entity('SearchEntity', entity)
|
||||
return
|
||||
else:
|
||||
table_service.update_entity('SearchEntity', entity)
|
||||
time.sleep(5)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
storage_account = sys.argv[1]
|
||||
storage_key = sys.argv[2]
|
||||
batch_account = sys.argv[3]
|
||||
batch_key = sys.argv[4]
|
||||
batch_url = sys.argv[5]
|
||||
job_id = sys.argv[6]
|
||||
entity_pk = sys.argv[7]
|
||||
entity_rk = sys.argv[8]
|
||||
|
||||
table_service = TableService(account_name=storage_account,
|
||||
account_key=storage_key)
|
||||
blob_service = BlockBlobService(account_name=storage_account,
|
||||
account_key=storage_key)
|
||||
credentials = batchauth.SharedKeyCredentials(batch_account, batch_key)
|
||||
batch_client = batch.BatchServiceClient(credentials, base_url=batch_url)
|
||||
wait_for_tasks_to_complete(table_service, batch_client, entity_pk, entity_rk, job_id)
|
||||
|
||||
|
|
@ -68,13 +68,25 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
_blobStorageProvider.UploadBlobFromStream(searchEntity.InputContainer, filename, queryFile.Content);
|
||||
});
|
||||
|
||||
// Keep record of the filenames
|
||||
searchEntity.Files =
|
||||
search.SearchInputFiles.Select(queryFile => Path.GetFileName(queryFile.Filename)).ToList();
|
||||
var queryIndex = 0;
|
||||
var searchQueries = new List<SearchQueryEntity>();
|
||||
foreach (var queryFile in search.SearchInputFiles)
|
||||
{
|
||||
var query = new SearchQueryEntity(searchEntity.Id, queryIndex.ToString());
|
||||
query.OutputContainer = searchEntity.OutputContainer;
|
||||
query.QueryFilename = Path.GetFileName(queryFile.Filename);
|
||||
query.State = QueryState.Waiting;
|
||||
query.QueryOutputFilename = GetQueryOutputFilename(searchEntity.OutputfileFormat, queryIndex.ToString());
|
||||
query.LogOutputFilename = GetLogFilename(searchEntity.OutputfileFormat, queryIndex.ToString());
|
||||
searchQueries.Add(query);
|
||||
queryIndex++;
|
||||
}
|
||||
|
||||
_tableStorageProvider.InsertEntities(searchQueries);
|
||||
|
||||
// Stage the generic batch scripts to storage
|
||||
var resourceFiles = InputFileStager.StageImportScripts(_blobStorageProvider);
|
||||
SubmitBatchJob(searchEntity, resourceFiles);
|
||||
SubmitBatchJob(searchEntity, searchQueries, resourceFiles);
|
||||
|
||||
searchEntity.State = SearchState.Running;
|
||||
_tableStorageProvider.UpdateEntity(searchEntity);
|
||||
|
|
@ -143,7 +155,7 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
return searchEntity;
|
||||
}
|
||||
|
||||
private void SubmitBatchJob(SearchEntity searchEntity, List<ResourceFile> resourceFiles)
|
||||
private void SubmitBatchJob(SearchEntity searchEntity, IEnumerable<SearchQueryEntity> queries, List<ResourceFile> resourceFiles)
|
||||
{
|
||||
PoolInformation poolInfo;
|
||||
|
||||
|
|
@ -188,32 +200,29 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
resourceFiles);
|
||||
job.Commit();
|
||||
|
||||
var tasks = GetTasks(searchEntity);
|
||||
var tasks = GetTasks(searchEntity, queries);
|
||||
job.Refresh();
|
||||
job.AddTask(tasks);
|
||||
|
||||
job.Refresh();
|
||||
job.OnAllTasksComplete = OnAllTasksComplete.TerminateJob;
|
||||
// job.Refresh();
|
||||
// job.OnAllTasksComplete = OnAllTasksComplete.TerminateJob;
|
||||
}
|
||||
|
||||
public IEnumerable<CloudTask> GetTasks(SearchEntity searchEntity)
|
||||
private IEnumerable<CloudTask> GetTasks(SearchEntity searchEntity, IEnumerable<SearchQueryEntity> queries)
|
||||
{
|
||||
var taskId = 0;
|
||||
foreach (var filename in searchEntity.Files)
|
||||
foreach (var query in queries)
|
||||
{
|
||||
var outputFilename = GetQueryFilename(searchEntity.OutputfileFormat, taskId.ToString());
|
||||
var blastOutputFilename = GetLogFilename(taskId.ToString());
|
||||
|
||||
var cmd = string.Format("/bin/bash -c '{0}; result=$?; {1}; exit $result'",
|
||||
GetBlastCommandLine(searchEntity.DatabaseId, searchEntity.Executable, searchEntity.ExecutableArgsSanitised, filename, outputFilename, blastOutputFilename),
|
||||
GetBlastCommandLine(searchEntity.DatabaseId, searchEntity.Executable, searchEntity.ExecutableArgsSanitised, query.QueryFilename, query.QueryOutputFilename, query.LogOutputFilename),
|
||||
GetUploadCommandLine(searchEntity.OutputContainer));
|
||||
|
||||
var task = new CloudTask(taskId.ToString(), cmd);
|
||||
task.DisplayName = filename;
|
||||
task.DisplayName = query.QueryFilename;
|
||||
task.UserIdentity = new UserIdentity(new AutoUserSpecification(elevationLevel: ElevationLevel.Admin));
|
||||
task.ResourceFiles = new List<ResourceFile>
|
||||
{
|
||||
new ResourceFile(_blobStorageProvider.GetBlobSAS(searchEntity.InputContainer, filename), filename)
|
||||
new ResourceFile(_blobStorageProvider.GetBlobSAS(searchEntity.InputContainer, query.QueryFilename), query.QueryFilename)
|
||||
};
|
||||
|
||||
task.EnvironmentSettings = new[]
|
||||
|
|
@ -227,18 +236,27 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
}
|
||||
}
|
||||
|
||||
private static string GetQueryFilename(string outputFilenameFormat, string taskId)
|
||||
private static string GetQueryOutputFilename(string outputFilenameFormat, string taskId)
|
||||
{
|
||||
// outputFilenameFormat is only used here to determine if we should use the old legacy names.
|
||||
if (string.IsNullOrEmpty(outputFilenameFormat))
|
||||
{
|
||||
return string.Format("queryoutput-{0}.xml", taskId);
|
||||
}
|
||||
return string.Format(outputFilenameFormat, taskId);
|
||||
}
|
||||
|
||||
private static string GetLogFilename(string taskId)
|
||||
private static string GetLogFilename(string outputFilenameFormat, string taskId)
|
||||
{
|
||||
if (string.IsNullOrEmpty(outputFilenameFormat))
|
||||
{
|
||||
return string.Format("blastoutput-{0}.log", taskId);
|
||||
}
|
||||
return string.Format("log-{0}.txt", taskId);
|
||||
}
|
||||
|
||||
private string GetBlastCommandLine(string databaseName, string executable, string executableArgs,
|
||||
string queryFilename, string queryOutputFilename, string blastOutputFilename)
|
||||
string queryFilename, string queryOutputFilename, string logOutputFilename)
|
||||
{
|
||||
var outputFormat = "-outfmt 5"; // XML
|
||||
if (!string.IsNullOrEmpty(executableArgs) && executableArgs.Contains(" -outfmt "))
|
||||
|
|
@ -254,7 +272,7 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
outputFormat,
|
||||
queryOutputFilename,
|
||||
executableArgs,
|
||||
blastOutputFilename);
|
||||
logOutputFilename);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -304,13 +322,12 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
public JobManagerTask GetJobManagerTask(string searchId, List<ResourceFile> resourceFiles)
|
||||
{
|
||||
var cmd =
|
||||
string.Format("/bin/bash -c 'python3 jobmanager.py {0} {1} {2} {3} {4} {5} {6} {7} {8}'",
|
||||
string.Format("/bin/bash -c 'python3 SearchJobManager.py {0} {1} {2} {3} {4} {5} {6} {7}'",
|
||||
_storageCredentials.Account,
|
||||
_storageCredentials.Key,
|
||||
_batchCredentials.Account,
|
||||
_batchCredentials.Key,
|
||||
_batchCredentials.Url,
|
||||
typeof(SearchEntity).Name, // Table name
|
||||
"$AZ_BATCH_JOB_ID",
|
||||
SearchEntity.AllUsersPk, // PK for JobMananger
|
||||
searchId); // RK for JobManager
|
||||
|
|
@ -322,7 +339,8 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
RunExclusive = false,
|
||||
UserIdentity = new UserIdentity(new AutoUserSpecification(elevationLevel: ElevationLevel.Admin)),
|
||||
ResourceFiles = resourceFiles,
|
||||
Constraints = new TaskConstraints(null, null, 3),
|
||||
Constraints = new TaskConstraints(null, null, 5),
|
||||
KillJobOnCompletion = true,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -430,7 +448,7 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
return _tableStorageProvider.ListEntities<SearchEntity>(SearchEntity.AllUsersPk);
|
||||
}
|
||||
|
||||
public IEnumerable<SearchQuery> ListSearchQueries(Guid searchId)
|
||||
public IEnumerable<SearchQueryEntity> ListSearchQueries(Guid searchId)
|
||||
{
|
||||
var entity = _tableStorageProvider.GetEntity<SearchEntity>(SearchEntity.AllUsersPk, searchId.ToString());
|
||||
|
||||
|
|
@ -439,9 +457,24 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
throw new Exception("No such search " + searchId);
|
||||
}
|
||||
|
||||
if (entity.Version == 0)
|
||||
{
|
||||
return ListLegacySearchQueries(entity);
|
||||
}
|
||||
|
||||
if (entity.Version == 1)
|
||||
{
|
||||
return ListV1SearchQueries(entity);
|
||||
}
|
||||
|
||||
throw new ArgumentException("Unknown search version: " + entity.Version);
|
||||
}
|
||||
|
||||
private IEnumerable<SearchQueryEntity> ListLegacySearchQueries(SearchEntity entity)
|
||||
{
|
||||
IEnumerable<QueryOutput> queryOutputs = GetAllQueryOutputs(entity).ToList();
|
||||
|
||||
List<SearchQuery> searchQueries = new List<SearchQuery>();
|
||||
List<SearchQueryEntity> searchQueries = new List<SearchQueryEntity>();
|
||||
|
||||
try
|
||||
{
|
||||
|
|
@ -450,16 +483,16 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
|
||||
foreach (var task in tasks)
|
||||
{
|
||||
var queryOutput = GetQueryFilename(entity.OutputfileFormat, task.Id);
|
||||
var logOutput = GetLogFilename(task.Id);
|
||||
var queryOutput = GetQueryOutputFilename(entity.OutputfileFormat, task.Id);
|
||||
var logOutput = GetLogFilename(entity.OutputfileFormat, task.Id);
|
||||
var outputs =
|
||||
queryOutputs.Where(output => output.Filename == queryOutput || output.Filename == logOutput)
|
||||
.ToList();
|
||||
|
||||
searchQueries.Add(new SearchQuery
|
||||
searchQueries.Add(new SearchQueryEntity
|
||||
{
|
||||
Id = task.Id,
|
||||
InputFilename = task.DisplayName,
|
||||
QueryFilename = task.DisplayName,
|
||||
Outputs = outputs,
|
||||
State = BatchToQueryState(task),
|
||||
StartTime = task.ExecutionInformation?.StartTime,
|
||||
|
|
@ -470,20 +503,20 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
catch (Exception)
|
||||
{
|
||||
var inputFiles = entity.Files;
|
||||
foreach (var queryNumber in Enumerable.Range(0, (int) entity.TotalTasks))
|
||||
foreach (var queryNumber in Enumerable.Range(0, (int)entity.TotalTasks))
|
||||
{
|
||||
var queryOutput = GetQueryFilename(entity.OutputfileFormat, queryNumber.ToString());
|
||||
var logOutput = GetLogFilename(queryNumber.ToString());
|
||||
var queryOutput = GetQueryOutputFilename(entity.OutputfileFormat, queryNumber.ToString());
|
||||
var logOutput = GetLogFilename(entity.OutputfileFormat, queryNumber.ToString());
|
||||
var outputs =
|
||||
queryOutputs.Where(output => output.Filename == queryOutput || output.Filename == logOutput)
|
||||
.ToList();
|
||||
|
||||
try
|
||||
{
|
||||
searchQueries.Add(new SearchQuery
|
||||
searchQueries.Add(new SearchQueryEntity
|
||||
{
|
||||
Id = queryNumber.ToString(),
|
||||
InputFilename = inputFiles[queryNumber],
|
||||
QueryFilename = inputFiles[queryNumber],
|
||||
Outputs = outputs,
|
||||
State = QueryState.Success,
|
||||
StartTime = null,
|
||||
|
|
@ -500,6 +533,25 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
return searchQueries;
|
||||
}
|
||||
|
||||
private IEnumerable<SearchQueryEntity> ListV1SearchQueries(SearchEntity entity)
|
||||
{
|
||||
IEnumerable<QueryOutput> queryOutputs = GetAllQueryOutputs(entity).ToList();
|
||||
|
||||
var queries = _tableStorageProvider.ListEntities<SearchQueryEntity>(entity.Id.ToString()).ToList();
|
||||
|
||||
foreach (var searchQueryEntity in queries)
|
||||
{
|
||||
var queryOutput = GetQueryOutputFilename(entity.OutputfileFormat, searchQueryEntity.Id);
|
||||
var logOutput = GetLogFilename(entity.OutputfileFormat, searchQueryEntity.Id);
|
||||
var outputs =
|
||||
queryOutputs.Where(output => output.Filename == queryOutput || output.Filename == logOutput)
|
||||
.ToList();
|
||||
searchQueryEntity.Outputs = outputs;
|
||||
}
|
||||
|
||||
return queries;
|
||||
}
|
||||
|
||||
private QueryState BatchToQueryState(CloudTask task)
|
||||
{
|
||||
switch (task.State)
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
|
||||
IEnumerable<SearchEntity> ListSearches();
|
||||
|
||||
IEnumerable<SearchQuery> ListSearchQueries(Guid searchId);
|
||||
IEnumerable<SearchQueryEntity> ListSearchQueries(Guid searchId);
|
||||
|
||||
string GetSearchQueryOutput(Guid searchId, string queryId, string filename);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,35 +0,0 @@
|
|||
// Copyright (c) Microsoft Corporation. All rights reserved.
|
||||
// Licensed under the MIT License. See License.txt in the project root for license information.
|
||||
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
|
||||
namespace Microsoft.Azure.Batch.Blast.Searches
|
||||
{
|
||||
public class SearchQuery
|
||||
{
|
||||
public string Id { get; set; }
|
||||
|
||||
public QueryState State { get; set; }
|
||||
|
||||
public DateTime? StartTime { get; set; }
|
||||
|
||||
public DateTime? EndTime { get; set; }
|
||||
|
||||
public string Duration
|
||||
{
|
||||
get
|
||||
{
|
||||
if (StartTime == null || EndTime == null)
|
||||
{
|
||||
return "";
|
||||
}
|
||||
return (EndTime.Value - StartTime.Value).GetFriendlyDuration();
|
||||
}
|
||||
}
|
||||
|
||||
public string InputFilename { get; set; }
|
||||
|
||||
public IEnumerable<QueryOutput> Outputs { get; set; }
|
||||
}
|
||||
}
|
||||
|
|
@ -17,6 +17,10 @@ namespace Microsoft.Azure.Batch.Blast.Searches
|
|||
|
||||
public IEnumerable<SearchInputFile> SearchInputFiles { get; set; }
|
||||
|
||||
public bool SplitSequenceFile { get; set; }
|
||||
|
||||
public int SequencesPerQuery { get; set; }
|
||||
|
||||
public string PoolId { get; set; }
|
||||
|
||||
public int? TargetDedicated { get; set; }
|
||||
|
|
|
|||
|
|
@ -27,6 +27,18 @@ namespace Microsoft.Azure.Batch.Blast.Storage
|
|||
table.Execute(insertOperation);
|
||||
}
|
||||
|
||||
public void InsertEntities<T>(IEnumerable<T> entities) where T : TableEntity
|
||||
{
|
||||
CloudTable table = _cloudTableClient.GetTableReference(typeof(T).Name);
|
||||
table.CreateIfNotExists();
|
||||
TableBatchOperation batchInsert = new TableBatchOperation();
|
||||
foreach (var entity in entities)
|
||||
{
|
||||
batchInsert.Insert(entity);
|
||||
}
|
||||
table.ExecuteBatch(batchInsert);
|
||||
}
|
||||
|
||||
public void UpdateEntity<T>(T entity) where T : TableEntity
|
||||
{
|
||||
CloudTable table = _cloudTableClient.GetTableReference(entity.GetType().Name);
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ namespace Microsoft.Azure.Batch.Blast.Storage.Entities
|
|||
{
|
||||
PartitionKey = AllUsersPk;
|
||||
RowKey = queryId.ToString();
|
||||
Version = 1;
|
||||
}
|
||||
|
||||
public Guid Id { get { return Guid.Parse(RowKey); } }
|
||||
|
|
@ -33,6 +34,8 @@ namespace Microsoft.Azure.Batch.Blast.Storage.Entities
|
|||
|
||||
public Int64 CompletedTasks { get; set; }
|
||||
|
||||
public int Version { get; set; }
|
||||
|
||||
public string InputContainer { get; set; }
|
||||
|
||||
public string OutputContainer { get; set; }
|
||||
|
|
|
|||
|
|
@ -0,0 +1,64 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Azure.Batch.Blast.Searches;
|
||||
using Microsoft.WindowsAzure.Storage.Table;
|
||||
|
||||
namespace Microsoft.Azure.Batch.Blast.Storage.Entities
|
||||
{
|
||||
public class SearchQueryEntity : TableEntity
|
||||
{
|
||||
public SearchQueryEntity(Guid searchId, string queryId)
|
||||
{
|
||||
PartitionKey = searchId.ToString();
|
||||
RowKey = queryId;
|
||||
CreationTime = DateTime.UtcNow;
|
||||
}
|
||||
|
||||
public SearchQueryEntity()
|
||||
{ }
|
||||
|
||||
public string Id { get { return RowKey; } set { RowKey = value; } }
|
||||
|
||||
public string OutputContainer { get; set; }
|
||||
|
||||
public string QueryFilename { get; set; }
|
||||
|
||||
public string LogOutputFilename { get; set; }
|
||||
|
||||
public string QueryOutputFilename { get; set; }
|
||||
|
||||
public string _State { get; set; }
|
||||
|
||||
[IgnoreProperty]
|
||||
public QueryState State
|
||||
{
|
||||
get { return (QueryState)Enum.Parse(typeof(QueryState), _State); }
|
||||
set { _State = value.ToString(); }
|
||||
}
|
||||
|
||||
public DateTime CreationTime { get; set; }
|
||||
|
||||
public DateTime? StartTime { get; set; }
|
||||
|
||||
public DateTime? EndTime { get; set; }
|
||||
|
||||
[IgnoreProperty]
|
||||
public string Duration
|
||||
{
|
||||
get
|
||||
{
|
||||
if (StartTime == null || EndTime == null)
|
||||
{
|
||||
return "";
|
||||
}
|
||||
return (EndTime.Value - StartTime.Value).GetFriendlyDuration();
|
||||
}
|
||||
}
|
||||
|
||||
[IgnoreProperty]
|
||||
public IEnumerable<QueryOutput> Outputs { get; set; }
|
||||
}
|
||||
}
|
||||
|
|
@ -10,6 +10,8 @@ namespace Microsoft.Azure.Batch.Blast.Storage
|
|||
{
|
||||
void InsertEntity<T>(T entity) where T : TableEntity;
|
||||
|
||||
void InsertEntities<T>(IEnumerable<T> entities) where T : TableEntity;
|
||||
|
||||
void UpdateEntity<T>(T entity) where T : TableEntity;
|
||||
|
||||
void UpsertEntity<T>(T entity) where T : TableEntity;
|
||||
|
|
|
|||
Загрузка…
Ссылка в новой задаче