Skip to content
This repository was archived by the owner on Nov 1, 2023. It is now read-only.

Commit bb71b73

Browse files
nharper285nharper285
authored
Merge branch 'main' into revert-3557-revert-7bcc41c67b74cd97668e169705d9af365f5c1297
2 parents 75ffcf0 + 909b130 commit bb71b73

File tree

7 files changed

+88
-106
lines changed

7 files changed

+88
-106
lines changed

‎src/ApiService/ApiService/Functions/QueueJobResult.cs‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,12 @@ public async Async.Task Run([QueueTrigger("job-result", Connection = "AzureWebJo
3131

3232
var job = await _jobs.Get(task.JobId);
3333
if (job == null) {
34-
_log.LogWarning("invalid {JobId}", task.JobId);
34+
_log.LogWarning("invalid message {JobId}", task.JobId);
35+
return;
36+
}
37+
38+
if (jr.CreatedAt == null) {
39+
_log.LogWarning("invalid message, no created_at field {JobId}", task.JobId);
3540
return;
3641
}
3742

@@ -52,7 +57,7 @@ public async Async.Task Run([QueueTrigger("job-result", Connection = "AzureWebJo
5257
return;
5358
}
5459

55-
var jobResult = await _context.JobResultOperations.CreateOrUpdate(job.JobId, jobResultType, value);
60+
var jobResult = await _context.JobResultOperations.CreateOrUpdate(job.JobId, jr.TaskId, jr.MachineId, jr.CreatedAt.Value, jr.Version, jobResultType, value);
5661
if (!jobResult.IsOk) {
5762
_log.LogError("failed to create or update with job result {JobId}", job.JobId);
5863
}

‎src/ApiService/ApiService/OneFuzzTypes/Model.cs‎

Lines changed: 18 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -34,19 +34,6 @@ public enum HeartbeatType {
3434
TaskAlive,
3535
}
3636

37-
[SkipRename]
38-
public enum JobResultType {
39-
NewCrashingInput,
40-
NoReproCrashingInput,
41-
NewReport,
42-
NewUniqueReport,
43-
NewRegressionReport,
44-
NewCoverage,
45-
NewCrashDump,
46-
CoverageData,
47-
RuntimeStats,
48-
}
49-
5037
public record HeartbeatData(HeartbeatType Type);
5138

5239
public record TaskHeartbeatEntry(
@@ -55,12 +42,14 @@ public record TaskHeartbeatEntry(
5542
[property: Required] Guid MachineId,
5643
HeartbeatData[] Data);
5744

58-
public record JobResultData(JobResultType Type);
45+
public record JobResultData(string Type);
5946

6047
public record TaskJobResultEntry(
6148
Guid TaskId,
6249
Guid? JobId,
6350
Guid MachineId,
51+
DateTime? CreatedAt,
52+
double Version,
6453
JobResultData Data,
6554
Dictionary<string, double> Value
6655
);
@@ -921,26 +910,24 @@ public record SecretAddress<T>(Uri Url) : ISecret<T> {
921910
public record SecretData<T>(ISecret<T> Secret) {
922911
}
923912

913+
[SkipRename]
914+
public enum JobResultType {
915+
CoverageData,
916+
RuntimeStats,
917+
}
918+
924919
public record JobResult(
925-
[PartitionKey][RowKey] Guid JobId,
920+
[PartitionKey] Guid JobId,
921+
[RowKey] string TaskIdMachineIdMetric,
922+
Guid TaskId,
923+
Guid MachineId,
924+
DateTime CreatedAt,
926925
string Project,
927926
string Name,
928-
double NewCrashingInput = 0,
929-
double NoReproCrashingInput = 0,
930-
double NewReport = 0,
931-
double NewUniqueReport = 0,
932-
double NewRegressionReport = 0,
933-
double NewCrashDump = 0,
934-
double InstructionsCovered = 0,
935-
double TotalInstructions = 0,
936-
double CoverageRate = 0,
937-
double IterationCount = 0
938-
) : EntityBase() {
939-
public JobResult(Guid JobId, string Project, string Name) : this(
940-
JobId: JobId,
941-
Project: Project,
942-
Name: Name, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) { }
943-
}
927+
string Type,
928+
double Version,
929+
Dictionary<string, double> MetricValue
930+
) : EntityBase();
944931

945932
public record JobConfig(
946933
string Project,

‎src/ApiService/ApiService/host.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
{
22
"version": "2.0",
3+
"functionTimeout": "12:00:00",
34
"logging": {
45
"applicationInsights": {
56
"samplingSettings": {

‎src/ApiService/ApiService/onefuzzlib/JobResultOperations.cs‎

Lines changed: 47 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -2,99 +2,75 @@
22
using Microsoft.Extensions.Logging;
33
using Polly;
44
namespace Microsoft.OneFuzz.Service;
5+
using System.Net;
56

67
public interface IJobResultOperations : IOrm<JobResult> {
78

8-
Async.Task<JobResult?> GetJobResult(Guid jobId);
9-
Async.Task<OneFuzzResultVoid> CreateOrUpdate(Guid jobId, JobResultType resultType, Dictionary<string, double> resultValue);
9+
Async.Task<JobResult?> GetJobResult(Guid jobId, Guid taskId, Guid machineId, string metricType);
10+
Async.Task<OneFuzzResultVoid> CreateOrUpdate(Guid jobId, Guid taskId, Guid machineId, DateTime createdAt, double version, string resultType, Dictionary<string, double> resultValue);
1011

1112
}
1213
public class JobResultOperations : Orm<JobResult>, IJobResultOperations {
1314

15+
const string COVERAGE_DATA = "CoverageData";
16+
const string RUNTIME_STATS = "RuntimeStats";
17+
1418
public JobResultOperations(ILogger<JobResultOperations> log, IOnefuzzContext context)
1519
: base(log, context) {
1620
}
1721

18-
public async Async.Task<JobResult?> GetJobResult(Guid jobId) {
19-
return await SearchByPartitionKeys(new[] { jobId.ToString() }).SingleOrDefaultAsync();
22+
public async Async.Task<JobResult?> GetJobResult(Guid jobId, Guid taskId, Guid machineId, string metricType) {
23+
var data = QueryAsync(Query.SingleEntity(jobId.ToString(), string.Concat(taskId, "-", machineId, "-", metricType)));
24+
return await data.FirstOrDefaultAsync();
2025
}
2126

22-
private JobResult UpdateResult(JobResult result, JobResultType type, Dictionary<string, double> resultValue) {
23-
24-
var newResult = result;
25-
double newValue;
26-
switch (type) {
27-
case JobResultType.NewCrashingInput:
28-
newValue = result.NewCrashingInput + resultValue["count"];
29-
newResult = result with { NewCrashingInput = newValue };
30-
break;
31-
case JobResultType.NewReport:
32-
newValue = result.NewReport + resultValue["count"];
33-
newResult = result with { NewReport = newValue };
34-
break;
35-
case JobResultType.NewUniqueReport:
36-
newValue = result.NewUniqueReport + resultValue["count"];
37-
newResult = result with { NewUniqueReport = newValue };
38-
break;
39-
case JobResultType.NewRegressionReport:
40-
newValue = result.NewRegressionReport + resultValue["count"];
41-
newResult = result with { NewRegressionReport = newValue };
42-
break;
43-
case JobResultType.NewCrashDump:
44-
newValue = result.NewCrashDump + resultValue["count"];
45-
newResult = result with { NewCrashDump = newValue };
46-
break;
47-
case JobResultType.CoverageData:
48-
double newCovered = resultValue["covered"];
49-
double newTotalCovered = resultValue["features"];
50-
double newCoverageRate = resultValue["rate"];
51-
newResult = result with { InstructionsCovered = newCovered, TotalInstructions = newTotalCovered, CoverageRate = newCoverageRate };
52-
break;
53-
case JobResultType.RuntimeStats:
54-
double newTotalIterations = resultValue["total_count"];
55-
newResult = result with { IterationCount = newTotalIterations };
56-
break;
57-
default:
58-
_logTracer.LogWarning($"Invalid Field {type}.");
59-
break;
60-
}
61-
_logTracer.LogInformation($"Attempting to log new result: {newResult}");
62-
return newResult;
63-
}
64-
65-
private async Async.Task<bool> TryUpdate(Job job, JobResultType resultType, Dictionary<string, double> resultValue) {
27+
private async Async.Task<bool> TryUpdate(Job job, Guid taskId, Guid machineId, DateTime createdAt, double version, string resultType, Dictionary<string, double> resultValue) {
6628
var jobId = job.JobId;
29+
var taskIdMachineIdMetric = string.Concat(taskId, "-", machineId, "-", resultType);
6730

68-
var jobResult = await GetJobResult(jobId);
69-
70-
if (jobResult == null) {
71-
_logTracer.LogInformation("Creating new JobResult for Job {JobId}", jobId);
72-
73-
var entry = new JobResult(JobId: jobId, Project: job.Config.Project, Name: job.Config.Name);
31+
var oldEntry = await GetJobResult(jobId, taskId, machineId, resultType);
7432

75-
jobResult = UpdateResult(entry, resultType, resultValue);
76-
77-
var r = await Insert(jobResult);
78-
if (!r.IsOk) {
79-
throw new InvalidOperationException($"failed to insert job result {jobResult.JobId}");
33+
if (oldEntry == null) {
34+
_logTracer.LogInformation($"attempt to insert new job result {taskId} and taskId+machineId+metricType {taskIdMachineIdMetric}");
35+
var newEntry = new JobResult(JobId: jobId, TaskIdMachineIdMetric: taskIdMachineIdMetric, TaskId: taskId, MachineId: machineId, CreatedAt: createdAt, Project: job.Config.Project, Name: job.Config.Name, resultType, Version: version, resultValue);
36+
var result = await Insert(newEntry);
37+
if (!result.IsOk) {
38+
throw new InvalidOperationException($"failed to insert job result with taskId {taskId} and taskId+machineId+metricType {taskIdMachineIdMetric}");
8039
}
81-
_logTracer.LogInformation("created job result {JobId}", jobResult.JobId);
82-
} else {
83-
_logTracer.LogInformation("Updating existing JobResult entry for Job {JobId}", jobId);
84-
85-
jobResult = UpdateResult(jobResult, resultType, resultValue);
40+
return true;
41+
}
8642

87-
var r = await Update(jobResult);
88-
if (!r.IsOk) {
89-
throw new InvalidOperationException($"failed to insert job result {jobResult.JobId}");
90-
}
91-
_logTracer.LogInformation("updated job result {JobId}", jobResult.JobId);
43+
ResultVoid<(HttpStatusCode Status, string Reason)> r;
44+
switch (resultType) {
45+
case COVERAGE_DATA:
46+
case RUNTIME_STATS:
47+
if (oldEntry.CreatedAt < createdAt) {
48+
oldEntry = oldEntry with { CreatedAt = createdAt, MetricValue = resultValue };
49+
r = await Update(oldEntry);
50+
if (!r.IsOk) {
51+
throw new InvalidOperationException($"failed to replace job result with taskId {taskId} and machineId+metricType {taskIdMachineIdMetric}");
52+
}
53+
} else {
54+
_logTracer.LogInformation($"received an out-of-date metric. skipping.");
55+
}
56+
break;
57+
default:
58+
_logTracer.LogInformation($"attempt to update job result {taskId} and taskId+machineId+metricType {taskIdMachineIdMetric}");
59+
oldEntry.MetricValue["count"]++;
60+
oldEntry = oldEntry with { MetricValue = oldEntry.MetricValue };
61+
r = await Update(oldEntry);
62+
if (!r.IsOk) {
63+
throw new InvalidOperationException($"failed to update job result with taskId {taskId} and machineId+metricType {taskIdMachineIdMetric}");
64+
}
65+
break;
9266
}
9367

68+
9469
return true;
70+
9571
}
9672

97-
public async Async.Task<OneFuzzResultVoid> CreateOrUpdate(Guid jobId, JobResultType resultType, Dictionary<string, double> resultValue) {
73+
public async Async.Task<OneFuzzResultVoid> CreateOrUpdate(Guid jobId, Guid taskId, Guid machineId, DateTime createdAt, double version, string resultType, Dictionary<string, double> resultValue) {
9874

9975
var job = await _context.JobOperations.Get(jobId);
10076
if (job == null) {
@@ -106,7 +82,7 @@ public async Async.Task<OneFuzzResultVoid> CreateOrUpdate(Guid jobId, JobResultT
10682
_logTracer.LogInformation("attempt to update job result {JobId}", job.JobId);
10783
var policy = Policy.Handle<InvalidOperationException>().WaitAndRetryAsync(50, _ => new TimeSpan(0, 0, 5));
10884
await policy.ExecuteAsync(async () => {
109-
success = await TryUpdate(job, resultType, resultValue);
85+
success = await TryUpdate(job, taskId, machineId, createdAt, version, resultType, resultValue);
11086
_logTracer.LogInformation("attempt {success}", success);
11187
});
11288
return OneFuzzResultVoid.Ok;

‎src/agent/Cargo.lock‎

Lines changed: 2 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎src/agent/onefuzz-result/Cargo.toml‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,10 +9,14 @@ license = "MIT"
99
[dependencies]
1010
anyhow = { version = "1.0", features = ["backtrace"] }
1111
async-trait = "0.1"
12+
chrono = { version = "0.4", default-features = false, features = [
13+
"clock",
14+
"std",
15+
"serde"
16+
] }
1217
reqwest = "0.11"
1318
serde = "1.0"
1419
storage-queue = { path = "../storage-queue" }
1520
uuid = { version = "1.4", features = ["serde", "v4"] }
1621
onefuzz-telemetry = { path = "../onefuzz-telemetry" }
1722
log = "0.4"
18-

‎src/agent/onefuzz-result/src/job_result.rs‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@
33

44
use anyhow::Result;
55
use async_trait::async_trait;
6+
use chrono::DateTime;
7+
pub use chrono::Utc;
68
use onefuzz_telemetry::warn;
79
use reqwest::Url;
810
use serde::{self, Deserialize, Serialize};
@@ -32,6 +34,8 @@ struct JobResult {
3234
job_id: Uuid,
3335
machine_id: Uuid,
3436
machine_name: String,
37+
created_at: DateTime<Utc>,
38+
version: f64,
3539
data: JobResultData,
3640
value: HashMap<String, f64>,
3741
}
@@ -103,7 +107,8 @@ impl JobResultSender for TaskJobResultClient {
103107
let job_id = self.context.state.job_id;
104108
let machine_id = self.context.state.machine_id;
105109
let machine_name = self.context.state.machine_name.clone();
106-
110+
let created_at = chrono::Utc::now();
111+
let version = 1.0;
107112
let _ = self
108113
.context
109114
.queue_client
@@ -112,6 +117,8 @@ impl JobResultSender for TaskJobResultClient {
112117
job_id,
113118
machine_id,
114119
machine_name,
120+
created_at,
121+
version,
115122
data,
116123
value,
117124
})

0 commit comments

Comments
 (0)