kemko/nomad
1
0
Fork 0
mirror of https://github.com/kemko/nomad.git synced 2026-01-07 19:05:42 +03:00
Code Issues Packages Projects Releases Wiki Activity

add create and modify timestamps to evaluations (#5881)

Browse Source
This commit is contained in:
Jasmine Dahilig
2019-08-07 09:50:35 -07:00
committed by GitHub
parent c5e026a816
commit 1bdb111127
15 changed files with 148 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
api/evaluations.go
View File

@@ -80,6 +80,8 @@ type Evaluation struct {
SnapshotIndex uint64
CreateIndex uint64
ModifyIndex uint64
CreateTime int64
ModifyTime int64
}
// EvalIndexSort is a wrapper to sort evaluations by CreateIndex.

13
command/eval_status.go
View File

@@ -4,6 +4,7 @@ import (
"fmt"
"sort"
"strings"
"time"
"github.com/hashicorp/nomad/api"
"github.com/hashicorp/nomad/api/contexts"
@@ -203,9 +204,21 @@ func (c *EvalStatusCommand) Run(args []string) int {
statusDesc = eval.Status
}
// Format eval timestamps
var formattedCreateTime, formattedModifyTime string
if verbose {
formattedCreateTime = formatUnixNanoTime(eval.CreateTime)
formattedModifyTime = formatUnixNanoTime(eval.ModifyTime)
} else {
formattedCreateTime = prettyTimeDiff(time.Unix(0, eval.CreateTime), time.Now())
formattedModifyTime = prettyTimeDiff(time.Unix(0, eval.ModifyTime), time.Now())
}
// Format the evaluation data
basic := []string{
fmt.Sprintf("ID|%s", limit(eval.ID, length)),
fmt.Sprintf("Create Time|%s", formattedCreateTime),
fmt.Sprintf("Modify Time|%s", formattedModifyTime),
fmt.Sprintf("Status|%s", eval.Status),
fmt.Sprintf("Status Description|%s", statusDesc),
fmt.Sprintf("Type|%s", eval.Type),

3
nomad/alloc_endpoint.go
View File

@@ -235,6 +235,7 @@ func (a *Alloc) Stop(args *structs.AllocStopRequest, reply *structs.AllocStopRes
return fmt.Errorf(structs.ErrUnknownAllocationPrefix)
}
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: alloc.Namespace,
@@ -244,6 +245,8 @@ func (a *Alloc) Stop(args *structs.AllocStopRequest, reply *structs.AllocStopRes
JobID: alloc.Job.ID,
JobModifyIndex: alloc.Job.ModifyIndex,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
transitionReq := &structs.AllocUpdateDesiredTransitionRequest{

3
nomad/deploymentwatcher/deployment_watcher.go
View File

@@ -791,6 +791,7 @@ func (w *deploymentWatcher) createBatchedUpdate(allowReplacements []string, forI
// getEval returns an evaluation suitable for the deployment
func (w *deploymentWatcher) getEval() *structs.Evaluation {
now := time.Now().UTC().UnixNano()
return &structs.Evaluation{
ID: uuid.Generate(),
Namespace: w.j.Namespace,
@@ -800,6 +801,8 @@ func (w *deploymentWatcher) getEval() *structs.Evaluation {
JobID: w.j.ID,
DeploymentID: w.deploymentID,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
}

3
nomad/drainer/drainer.go
View File

@@ -402,6 +402,7 @@ func (n *NodeDrainer) drainAllocs(future *structs.BatchFuture, allocs []*structs
}
evals := make([]*structs.Evaluation, 0, len(jobs))
now := time.Now().UTC().UnixNano()
for job, alloc := range jobs {
evals = append(evals, &structs.Evaluation{
ID: uuid.Generate(),
@@ -411,6 +412,8 @@ func (n *NodeDrainer) drainAllocs(future *structs.BatchFuture, allocs []*structs
TriggeredBy: structs.EvalTriggerNodeDrain,
JobID: job,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
})
}

19
nomad/job_endpoint.go
View File

@@ -208,6 +208,7 @@ func (j *Job) Register(args *structs.JobRegisterRequest, reply *structs.JobRegis
}
// Create a new evaluation
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: args.RequestNamespace(),
@@ -217,6 +218,8 @@ func (j *Job) Register(args *structs.JobRegisterRequest, reply *structs.JobRegis
JobID: args.Job.ID,
JobModifyIndex: reply.JobModifyIndex,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
update := &structs.EvalUpdateRequest{
Evals: []*structs.Evaluation{eval},
@@ -571,6 +574,7 @@ func (j *Job) Evaluate(args *structs.JobEvaluateRequest, reply *structs.JobRegis
}
// Create a new evaluation
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: args.RequestNamespace(),
@@ -580,6 +584,8 @@ func (j *Job) Evaluate(args *structs.JobEvaluateRequest, reply *structs.JobRegis
JobID: job.ID,
JobModifyIndex: job.ModifyIndex,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
// Create a AllocUpdateDesiredTransitionRequest request with the eval and any forced rescheduled allocs
@@ -650,6 +656,7 @@ func (j *Job) Deregister(args *structs.JobDeregisterRequest, reply *structs.JobD
// Create a new evaluation
// XXX: The job priority / type is strange for this, since it's not a high
// priority even if the job was.
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: args.RequestNamespace(),
@@ -659,6 +666,8 @@ func (j *Job) Deregister(args *structs.JobDeregisterRequest, reply *structs.JobD
JobID: args.JobID,
JobModifyIndex: index,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
update := &structs.EvalUpdateRequest{
Evals: []*structs.Evaluation{eval},
@@ -738,6 +747,7 @@ func (j *Job) BatchDeregister(args *structs.JobBatchDeregisterRequest, reply *st
}
// Create a new evaluation
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: jobNS.Namespace,
@@ -746,6 +756,8 @@ func (j *Job) BatchDeregister(args *structs.JobBatchDeregisterRequest, reply *st
TriggeredBy: structs.EvalTriggerJobDeregister,
JobID: jobNS.ID,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
args.Evals = append(args.Evals, eval)
}
@@ -1195,6 +1207,7 @@ func (j *Job) Plan(args *structs.JobPlanRequest, reply *structs.JobPlanResponse)
}
// Create an eval and mark it as requiring annotations and insert that as well
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: args.RequestNamespace(),
@@ -1205,6 +1218,9 @@ func (j *Job) Plan(args *structs.JobPlanRequest, reply *structs.JobPlanResponse)
JobModifyIndex: updatedIndex,
Status: structs.EvalStatusPending,
AnnotatePlan: true,
// Timestamps are added for consistency but this eval is never persisted
CreateTime: now,
ModifyTime: now,
}
snap.UpsertEvals(100, []*structs.Evaluation{eval})
@@ -1415,6 +1431,7 @@ func (j *Job) Dispatch(args *structs.JobDispatchRequest, reply *structs.JobDispa
// If the job is periodic, we don't create an eval.
if !dispatchJob.IsPeriodic() {
// Create a new evaluation
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: args.RequestNamespace(),
@@ -1424,6 +1441,8 @@ func (j *Job) Dispatch(args *structs.JobDispatchRequest, reply *structs.JobDispa
JobID: dispatchJob.ID,
JobModifyIndex: jobCreateIndex,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
update := &structs.EvalUpdateRequest{
Evals: []*structs.Evaluation{eval},

36
nomad/job_endpoint_test.go
View File

@@ -96,6 +96,12 @@ func TestJobEndpoint_Register(t *testing.T) {
if eval.Status != structs.EvalStatusPending {
t.Fatalf("bad: %#v", eval)
}
if eval.CreateTime == 0 {
t.Fatalf("eval CreateTime is unset: %#v", eval)
}
if eval.ModifyTime == 0 {
t.Fatalf("eval ModifyTime is unset: %#v", eval)
}
}
func TestJobEndpoint_Register_ACL(t *testing.T) {
@@ -302,6 +308,12 @@ func TestJobEndpoint_Register_Existing(t *testing.T) {
if eval.Status != structs.EvalStatusPending {
t.Fatalf("bad: %#v", eval)
}
if eval.CreateTime == 0 {
t.Fatalf("eval CreateTime is unset: %#v", eval)
}
if eval.ModifyTime == 0 {
t.Fatalf("eval ModifyTime is unset: %#v", eval)
}
if err := msgpackrpc.CallWithCodec(codec, "Job.Register", req, &resp); err != nil {
t.Fatalf("err: %v", err)
@@ -1500,6 +1512,12 @@ func TestJobEndpoint_Evaluate(t *testing.T) {
if eval.Status != structs.EvalStatusPending {
t.Fatalf("bad: %#v", eval)
}
if eval.CreateTime == 0 {
t.Fatalf("eval CreateTime is unset: %#v", eval)
}
if eval.ModifyTime == 0 {
t.Fatalf("eval ModifyTime is unset: %#v", eval)
}
}
func TestJobEndpoint_ForceRescheduleEvaluate(t *testing.T) {
@@ -1569,6 +1587,8 @@ func TestJobEndpoint_ForceRescheduleEvaluate(t *testing.T) {
require.Equal(eval.JobID, job.ID)
require.Equal(eval.JobModifyIndex, resp.JobModifyIndex)
require.Equal(eval.Status, structs.EvalStatusPending)
require.NotZero(eval.CreateTime)
require.NotZero(eval.ModifyTime)
// Lookup the alloc, verify DesiredTransition ForceReschedule
alloc, err = state.AllocByID(ws, alloc.ID)
@@ -1647,6 +1667,8 @@ func TestJobEndpoint_Evaluate_ACL(t *testing.T) {
require.Equal(eval.JobID, job.ID)
require.Equal(eval.JobModifyIndex, validResp2.JobModifyIndex)
require.Equal(eval.Status, structs.EvalStatusPending)
require.NotZero(eval.CreateTime)
require.NotZero(eval.ModifyTime)
}
func TestJobEndpoint_Evaluate_Periodic(t *testing.T) {
@@ -1790,6 +1812,8 @@ func TestJobEndpoint_Deregister(t *testing.T) {
require.Equal(structs.EvalTriggerJobDeregister, eval.TriggeredBy)
require.Equal(job.ID, eval.JobID)
require.Equal(structs.EvalStatusPending, eval.Status)
require.NotZero(eval.CreateTime)
require.NotZero(eval.ModifyTime)
// Deregister and purge
dereg2 := &structs.JobDeregisterRequest{
@@ -1820,6 +1844,8 @@ func TestJobEndpoint_Deregister(t *testing.T) {
require.Equal(structs.EvalTriggerJobDeregister, eval.TriggeredBy)
require.Equal(job.ID, eval.JobID)
require.Equal(structs.EvalStatusPending, eval.Status)
require.NotZero(eval.CreateTime)
require.NotZero(eval.ModifyTime)
}
func TestJobEndpoint_Deregister_ACL(t *testing.T) {
@@ -1899,6 +1925,8 @@ func TestJobEndpoint_Deregister_ACL(t *testing.T) {
require.Equal(eval.JobID, job.ID)
require.Equal(eval.JobModifyIndex, validResp2.JobModifyIndex)
require.Equal(eval.Status, structs.EvalStatusPending)
require.NotZero(eval.CreateTime)
require.NotZero(eval.ModifyTime)
}
func TestJobEndpoint_Deregister_Nonexistent(t *testing.T) {
@@ -1959,6 +1987,12 @@ func TestJobEndpoint_Deregister_Nonexistent(t *testing.T) {
if eval.Status != structs.EvalStatusPending {
t.Fatalf("bad: %#v", eval)
}
if eval.CreateTime == 0 {
t.Fatalf("eval CreateTime is unset: %#v", eval)
}
if eval.ModifyTime == 0 {
t.Fatalf("eval ModifyTime is unset: %#v", eval)
}
}
func TestJobEndpoint_Deregister_Periodic(t *testing.T) {
@@ -2165,6 +2199,8 @@ func TestJobEndpoint_BatchDeregister(t *testing.T) {
require.Equal(structs.EvalTriggerJobDeregister, eval.TriggeredBy)
require.Equal(expectedJob.ID, eval.JobID)
require.Equal(structs.EvalStatusPending, eval.Status)
require.NotZero(eval.CreateTime)
require.NotZero(eval.ModifyTime)
}
}

3
nomad/leader.go
View File

@@ -526,8 +526,10 @@ func (s *Server) reapFailedEvaluations(stopCh chan struct{}) {
// due to the fairly large backoff.
followupEvalWait := s.config.EvalFailedFollowupBaselineDelay +
time.Duration(rand.Int63n(int64(s.config.EvalFailedFollowupDelayRange)))
followupEval := eval.CreateFailedFollowUpEval(followupEvalWait)
updateEval.NextEval = followupEval.ID
updateEval.UpdateModifyTime()
// Update via Raft
req := structs.EvalUpdateRequest{
@@ -564,6 +566,7 @@ func (s *Server) reapDupBlockedEvaluations(stopCh chan struct{}) {
newEval := dup.Copy()
newEval.Status = structs.EvalStatusCancelled
newEval.StatusDescription = fmt.Sprintf("existing blocked evaluation exists for job %q", newEval.JobID)
newEval.UpdateModifyTime()
cancel[i] = newEval
}

15
nomad/mock/mock.go
View File

@@ -402,13 +402,16 @@ func PeriodicJob() *structs.Job {
}
func Eval() *structs.Evaluation {
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: structs.DefaultNamespace,
Priority: 50,
Type: structs.JobTypeService,
JobID: uuid.Generate(),
Status: structs.EvalStatusPending,
ID: uuid.Generate(),
Namespace: structs.DefaultNamespace,
Priority: 50,
Type: structs.JobTypeService,
JobID: uuid.Generate(),
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
return eval
}

10
nomad/node_endpoint.go
View File

@@ -1077,6 +1077,8 @@ func (n *Node) UpdateAlloc(args *structs.AllocUpdateRequest, reply *structs.Gene
Type: job.Type,
Priority: job.Priority,
Status: structs.EvalStatusPending,
CreateTime: now.UTC().UnixNano(),
ModifyTime: now.UTC().UnixNano(),
}
evals = append(evals, eval)
}
@@ -1134,6 +1136,9 @@ func (n *Node) batchUpdate(future *structs.BatchFuture, updates []*structs.Alloc
}
_, exists := evalsByJobId[namespacedID]
if !exists {
now := time.Now().UTC().UnixNano()
eval.CreateTime = now
eval.ModifyTime = now
trimmedEvals = append(trimmedEvals, eval)
evalsByJobId[namespacedID] = struct{}{}
}
@@ -1281,6 +1286,7 @@ func (n *Node) createNodeEvals(nodeID string, nodeIndex uint64) ([]string, uint6
var evals []*structs.Evaluation
var evalIDs []string
jobIDs := make(map[string]struct{})
now := time.Now().UTC().UnixNano()
for _, alloc := range allocs {
// Deduplicate on JobID
@@ -1300,6 +1306,8 @@ func (n *Node) createNodeEvals(nodeID string, nodeIndex uint64) ([]string, uint6
NodeID: nodeID,
NodeModifyIndex: nodeIndex,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
evals = append(evals, eval)
evalIDs = append(evalIDs, eval.ID)
@@ -1324,6 +1332,8 @@ func (n *Node) createNodeEvals(nodeID string, nodeIndex uint64) ([]string, uint6
NodeID: nodeID,
NodeModifyIndex: nodeIndex,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
evals = append(evals, eval)
evalIDs = append(evalIDs, eval.ID)

12
nomad/node_endpoint_test.go
View File

@@ -2352,6 +2352,12 @@ func TestClientEndpoint_CreateNodeEvals(t *testing.T) {
if eval.JobID != expJobID {
t.Fatalf("JobID incorrect on type %v: %#v", schedType, eval)
}
if eval.CreateTime == 0 {
t.Fatalf("CreateTime is unset on type %v: %#v", schedType, eval)
}
if eval.ModifyTime == 0 {
t.Fatalf("ModifyTime is unset on type %v: %#v", schedType, eval)
}
}
}
@@ -2434,6 +2440,12 @@ func TestClientEndpoint_Evaluate(t *testing.T) {
if eval.Status != structs.EvalStatusPending {
t.Fatalf("bad: %#v", eval)
}
if eval.CreateTime == 0 {
t.Fatalf("CreateTime is unset: %#v", eval)
}
if eval.ModifyTime == 0 {
t.Fatalf("ModifyTime is unset: %#v", eval)
}
}
func TestClientEndpoint_Evaluate_ACL(t *testing.T) {

3
nomad/periodic.go
View File

@@ -63,6 +63,7 @@ func (s *Server) DispatchJob(job *structs.Job) (*structs.Evaluation, error) {
}
// Create a new evaluation
now := time.Now().UTC().UnixNano()
eval := &structs.Evaluation{
ID: uuid.Generate(),
Namespace: job.Namespace,
@@ -72,6 +73,8 @@ func (s *Server) DispatchJob(job *structs.Job) (*structs.Evaluation, error) {
JobID: job.ID,
JobModifyIndex: index,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
update := &structs.EvalUpdateRequest{
Evals: []*structs.Evaluation{eval},

2
nomad/plan_apply.go
View File

@@ -293,6 +293,8 @@ func (p *planner) applyPlan(plan *structs.Plan, result *structs.PlanResult, snap
Type: job.Type,
Priority: job.Priority,
Status: structs.EvalStatusPending,
CreateTime: now,
ModifyTime: now,
}
evals = append(evals, eval)
}

25
nomad/structs/structs.go
View File

@@ -8181,6 +8181,9 @@ type Evaluation struct {
// Raft Indexes
CreateIndex uint64
ModifyIndex uint64
CreateTime int64
ModifyTime int64
}
// TerminalStatus returns if the current status is terminal and
@@ -8280,6 +8283,7 @@ func (e *Evaluation) MakePlan(j *Job) *Plan {
// NextRollingEval creates an evaluation to followup this eval for rolling updates
func (e *Evaluation) NextRollingEval(wait time.Duration) *Evaluation {
now := time.Now().UTC().UnixNano()
return &Evaluation{
ID: uuid.Generate(),
Namespace: e.Namespace,
@@ -8291,6 +8295,8 @@ func (e *Evaluation) NextRollingEval(wait time.Duration) *Evaluation {
Status: EvalStatusPending,
Wait: wait,
PreviousEval: e.ID,
CreateTime: now,
ModifyTime: now,
}
}
@@ -8300,7 +8306,7 @@ func (e *Evaluation) NextRollingEval(wait time.Duration) *Evaluation {
// quota limit was reached.
func (e *Evaluation) CreateBlockedEval(classEligibility map[string]bool,
escaped bool, quotaReached string) *Evaluation {
now := time.Now().UTC().UnixNano()
return &Evaluation{
ID: uuid.Generate(),
Namespace: e.Namespace,
@@ -8314,6 +8320,8 @@ func (e *Evaluation) CreateBlockedEval(classEligibility map[string]bool,
ClassEligibility: classEligibility,
EscapedComputedClass: escaped,
QuotaLimitReached: quotaReached,
CreateTime: now,
ModifyTime: now,
}
}
@@ -8322,6 +8330,7 @@ func (e *Evaluation) CreateBlockedEval(classEligibility map[string]bool,
// be retried by the eval_broker. Callers should copy the created eval's ID to
// into the old eval's NextEval field.
func (e *Evaluation) CreateFailedFollowUpEval(wait time.Duration) *Evaluation {
now := time.Now().UTC().UnixNano()
return &Evaluation{
ID: uuid.Generate(),
Namespace: e.Namespace,
@@ -8333,6 +8342,20 @@ func (e *Evaluation) CreateFailedFollowUpEval(wait time.Duration) *Evaluation {
Status: EvalStatusPending,
Wait: wait,
PreviousEval: e.ID,
CreateTime: now,
ModifyTime: now,
}
}
// UpdateModifyTime takes into account that clocks on different servers may be
// slightly out of sync. Even in case of a leader change, this method will
// guarantee that ModifyTime will always be after CreateTime.
func (e *Evaluation) UpdateModifyTime() {
now := time.Now().UTC().UnixNano()
if now <= e.CreateTime {
e.ModifyTime = e.CreateTime + 1
} else {
e.ModifyTime = now
}
}

6
nomad/worker.go
View File

@@ -353,6 +353,7 @@ func (w *Worker) UpdateEval(eval *structs.Evaluation) error {
// Store the snapshot index in the eval
eval.SnapshotIndex = w.snapshotIndex
eval.UpdateModifyTime()
// Setup the request
req := structs.EvalUpdateRequest{
@@ -391,6 +392,10 @@ func (w *Worker) CreateEval(eval *structs.Evaluation) error {
// Store the snapshot index in the eval
eval.SnapshotIndex = w.snapshotIndex
now := time.Now().UTC().UnixNano()
eval.CreateTime = now
eval.ModifyTime = now
// Setup the request
req := structs.EvalUpdateRequest{
Evals: []*structs.Evaluation{eval},
@@ -451,6 +456,7 @@ func (w *Worker) ReblockEval(eval *structs.Evaluation) error {
// Store the snapshot index in the eval
eval.SnapshotIndex = w.snapshotIndex
eval.UpdateModifyTime()
// Setup the request
req := structs.EvalUpdateRequest{