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

Merge pull request #5894 from hashicorp/f-remove-deprecated-code

Browse Source 
Remove deprecated code
This commit is contained in:
Preetha
2019-07-02 09:29:24 -05:00
committed by GitHub
parent e68709f9f1 02d0b9274c
commit 0819e4af18
10 changed files with 7 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
command/agent/node_endpoint.go
View File

@@ -108,7 +108,7 @@ func (s *HTTPServer) nodeToggleDrain(resp http.ResponseWriter, req *http.Request
var drainRequest api.NodeUpdateDrainRequest
// COMPAT: Remove in 0.9. Allow the old style enable query param.
// COMPAT: Remove in 0.10. Allow the old style enable query param.
// Get the enable parameter
enableRaw := req.URL.Query().Get("enable")
var enable bool

2
command/alloc_status.go
View File

@@ -235,7 +235,7 @@ func formatAllocBasicInfo(alloc *api.Allocation, client *api.Client, uuidLength
fmt.Sprintf("Node ID|%s", limit(alloc.NodeID, uuidLength)),
fmt.Sprintf("Node Name|%s", alloc.NodeName),
fmt.Sprintf("Job ID|%s", alloc.JobID),
fmt.Sprintf("Job Version|%d", getVersion(alloc.Job)),
fmt.Sprintf("Job Version|%d", alloc.Job.Version),
fmt.Sprintf("Client Status|%s", alloc.ClientStatus),
fmt.Sprintf("Client Description|%s", alloc.ClientDescription),
fmt.Sprintf("Desired Status|%s", alloc.DesiredStatus),

26
command/helpers.go
View File

@@ -438,32 +438,6 @@ func (j *JobGetter) ApiJob(jpath string) (*api.Job, error) {
return jobStruct, nil
}
// COMPAT: Remove in 0.7.0
// Nomad 0.6.0 introduces the submit time field so CLI's interacting with
// older versions of Nomad would SEGFAULT as reported here:
// https://github.com/hashicorp/nomad/issues/2918
// getSubmitTime returns a submit time of the job converting to time.Time
func getSubmitTime(job *api.Job) time.Time {
if job.SubmitTime != nil {
return time.Unix(0, *job.SubmitTime)
}
return time.Time{}
}
// COMPAT: Remove in 0.7.0
// Nomad 0.6.0 introduces job Versions so CLI's interacting with
// older versions of Nomad would SEGFAULT as reported here:
// https://github.com/hashicorp/nomad/issues/2918
// getVersion returns a version of the job in safely.
func getVersion(job *api.Job) uint64 {
if job.Version != nil {
return *job.Version
}
return 0
}
// mergeAutocompleteFlags is used to join multiple flag completion sets.
func mergeAutocompleteFlags(flags ...complete.Flags) complete.Flags {
merged := make(map[string]complete.Predictor, len(flags))

6
command/job_status.go
View File

@@ -169,7 +169,7 @@ func (c *JobStatusCommand) Run(args []string) int {
basic := []string{
fmt.Sprintf("ID|%s", *job.ID),
fmt.Sprintf("Name|%s", *job.Name),
fmt.Sprintf("Submit Date|%s", formatTime(getSubmitTime(job))),
fmt.Sprintf("Submit Date|%s", formatTime(time.Unix(0, *job.SubmitTime))),
fmt.Sprintf("Type|%s", *job.Type),
fmt.Sprintf("Priority|%d", *job.Priority),
fmt.Sprintf("Datacenters|%s", strings.Join(job.Datacenters, ",")),
@@ -462,7 +462,7 @@ func formatAllocList(allocations []*api.Allocation, verbose bool, uuidLength int
limit(alloc.EvalID, uuidLength),
limit(alloc.NodeID, uuidLength),
alloc.TaskGroup,
getVersion(alloc.Job),
alloc.Job.Version,
alloc.DesiredStatus,
alloc.ClientStatus,
formatUnixNanoTime(alloc.CreateTime),
@@ -478,7 +478,7 @@ func formatAllocList(allocations []*api.Allocation, verbose bool, uuidLength int
limit(alloc.ID, uuidLength),
limit(alloc.NodeID, uuidLength),
alloc.TaskGroup,
getVersion(alloc.Job),
alloc.Job.Version,
alloc.DesiredStatus,
alloc.ClientStatus,
createTimePretty,

54
nomad/fsm.go
View File

@@ -1135,11 +1135,6 @@ func (n *nomadFSM) Restore(old io.ReadCloser) error {
return err
}
// COMPAT: Handle upgrade to v0.7.0
if eval.Namespace == "" {
eval.Namespace = structs.DefaultNamespace
}
if err := restore.EvalRestore(eval); err != nil {
return err
}
@@ -1150,11 +1145,6 @@ func (n *nomadFSM) Restore(old io.ReadCloser) error {
return err
}
// COMPAT: Handle upgrade to v0.7.0
if alloc.Namespace == "" {
alloc.Namespace = structs.DefaultNamespace
}
if err := restore.AllocRestore(alloc); err != nil {
return err
}
@@ -1174,11 +1164,6 @@ func (n *nomadFSM) Restore(old io.ReadCloser) error {
return err
}
// COMPAT: Handle upgrade to v0.7.0
if launch.Namespace == "" {
launch.Namespace = structs.DefaultNamespace
}
if err := restore.PeriodicLaunchRestore(launch); err != nil {
return err
}
@@ -1189,11 +1174,6 @@ func (n *nomadFSM) Restore(old io.ReadCloser) error {
return err
}
// COMPAT: Handle upgrade to v0.7.0
if summary.Namespace == "" {
summary.Namespace = structs.DefaultNamespace
}
if err := restore.JobSummaryRestore(summary); err != nil {
return err
}
@@ -1213,11 +1193,6 @@ func (n *nomadFSM) Restore(old io.ReadCloser) error {
return err
}
// COMPAT: Handle upgrade to v0.7.0
if version.Namespace == "" {
version.Namespace = structs.DefaultNamespace
}
if err := restore.JobVersionRestore(version); err != nil {
return err
}
@@ -1228,11 +1203,6 @@ func (n *nomadFSM) Restore(old io.ReadCloser) error {
return err
}
// COMPAT: Handle upgrade to v0.7.0
if deployment.Namespace == "" {
deployment.Namespace = structs.DefaultNamespace
}
if err := restore.DeploymentRestore(deployment); err != nil {
return err
}
@@ -1280,30 +1250,6 @@ func (n *nomadFSM) Restore(old io.ReadCloser) error {
restore.Commit()
// Create Job Summaries
// COMPAT 0.4 -> 0.4.1
// We can remove this in 0.5. This exists so that the server creates job
// summaries if they were not present previously. When users upgrade to 0.5
// from 0.4.1, the snapshot will contain job summaries so it will be safe to
// remove this block.
index, err := newState.Index("job_summary")
if err != nil {
return fmt.Errorf("couldn't fetch index of job summary table: %v", err)
}
// If the index is 0 that means there is no job summary in the snapshot so
// we will have to create them
if index == 0 {
// query the latest index
latestIndex, err := newState.LatestIndex()
if err != nil {
return fmt.Errorf("unable to query latest index: %v", index)
}
if err := newState.ReconcileJobSummaries(latestIndex); err != nil {
return fmt.Errorf("error reconciling summaries: %v", err)
}
}
// COMPAT Remove in 0.10
// Clean up active deployments that do not have a job
if err := n.failLeakedDeployments(newState); err != nil {

41
nomad/fsm_test.go
View File

@@ -2701,47 +2701,6 @@ func TestFSM_SnapshotRestore_SchedulerConfiguration(t *testing.T) {
}
func TestFSM_SnapshotRestore_AddMissingSummary(t *testing.T) {
t.Parallel()
// Add some state
fsm := testFSM(t)
state := fsm.State()
// make an allocation
alloc := mock.Alloc()
state.UpsertJob(1010, alloc.Job)
state.UpsertAllocs(1011, []*structs.Allocation{alloc})
// Delete the summary
state.DeleteJobSummary(1040, alloc.Namespace, alloc.Job.ID)
// Delete the index
if err := state.RemoveIndex("job_summary"); err != nil {
t.Fatalf("err: %v", err)
}
fsm2 := testSnapshotRestore(t, fsm)
state2 := fsm2.State()
latestIndex, _ := state.LatestIndex()
ws := memdb.NewWatchSet()
out, _ := state2.JobSummaryByID(ws, alloc.Namespace, alloc.Job.ID)
expected := structs.JobSummary{
JobID: alloc.Job.ID,
Namespace: alloc.Job.Namespace,
Summary: map[string]structs.TaskGroupSummary{
"web": {
Starting: 1,
},
},
CreateIndex: 1010,
ModifyIndex: latestIndex,
}
if !reflect.DeepEqual(&expected, out) {
t.Fatalf("expected: %#v, actual: %#v", &expected, out)
}
}
func TestFSM_ReconcileSummaries(t *testing.T) {
t.Parallel()
// Add some state

28
nomad/leader.go
View File

@@ -268,15 +268,6 @@ func (s *Server) establishLeadership(stopCh chan struct{}) error {
return err
}
// COMPAT 0.4 - 0.4.1
// Reconcile the summaries of the registered jobs. We reconcile summaries
// only if the server is 0.4.1 since summaries are not present in 0.4 they
// might be incorrect after upgrading to 0.4.1 the summaries might not be
// correct
if err := s.reconcileJobSummaries(); err != nil {
return fmt.Errorf("unable to reconcile job summaries: %v", err)
}
// Start replication of ACLs and Policies if they are enabled,
// and we are not the authoritative region.
if s.config.ACLEnabled && s.config.Region != s.config.AuthoritativeRegion {
@@ -798,25 +789,6 @@ func (s *Server) reconcileMember(member serf.Member) error {
return nil
}
// reconcileJobSummaries reconciles the summaries of all the jobs registered in
// the system
// COMPAT 0.4 -> 0.4.1
func (s *Server) reconcileJobSummaries() error {
index, err := s.fsm.state.LatestIndex()
if err != nil {
return fmt.Errorf("unable to read latest index: %v", err)
}
s.logger.Debug("leader reconciling job summaries", "index", index)
args := &structs.GenericResponse{}
msg := structs.ReconcileJobSummariesRequestType | structs.IgnoreUnknownTypeFlag
if _, _, err = s.raftApply(msg, args); err != nil {
return fmt.Errorf("reconciliation of job summaries failed: %v", err)
}
return nil
}
// addRaftPeer is used to add a new Raft peer when a Nomad server joins
func (s *Server) addRaftPeer(m serf.Member, parts *serverParts) error {
// Check for possibility of multiple bootstrap nodes

206
nomad/state/state_store.go
View File

@@ -261,9 +261,6 @@ func (s *StateStore) UpsertPlanResults(index uint64, results *structs.ApplyPlanR
s.upsertDeploymentUpdates(index, results.DeploymentUpdates, txn)
}
// COMPAT: Nomad versions before 0.7.1 did not include the eval ID when
// applying the plan. Thus while we are upgrading, we ignore updating the
// modify index of evaluations from older plans.
if results.EvalID != "" {
// Update the modify index of the eval id
if err := s.updateEvalModifyIndex(txn, index, results.EvalID); err != nil {
@@ -353,11 +350,6 @@ func (s *StateStore) UpsertJobSummary(index uint64, jobSummary *structs.JobSumma
txn := s.db.Txn(true)
defer txn.Abort()
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if jobSummary.Namespace == "" {
jobSummary.Namespace = structs.DefaultNamespace
}
// Check if the job summary already exists
existing, err := txn.First("job_summary", "id", jobSummary.Namespace, jobSummary.JobID)
if err != nil {
@@ -393,11 +385,6 @@ func (s *StateStore) DeleteJobSummary(index uint64, namespace, id string) error
txn := s.db.Txn(true)
defer txn.Abort()
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Delete the job summary
if _, err := txn.DeleteAll("job_summary", "id", namespace, id); err != nil {
return fmt.Errorf("deleting job summary failed: %v", err)
@@ -428,11 +415,6 @@ func (s *StateStore) upsertDeploymentImpl(index uint64, deployment *structs.Depl
return fmt.Errorf("deployment lookup failed: %v", err)
}
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if deployment.Namespace == "" {
deployment.Namespace = structs.DefaultNamespace
}
// Setup the indexes correctly
if existing != nil {
deployment.CreateIndex = existing.(*structs.Deployment).CreateIndex
@@ -539,11 +521,6 @@ func (s *StateStore) deploymentByIDImpl(ws memdb.WatchSet, deploymentID string,
func (s *StateStore) DeploymentsByJobID(ws memdb.WatchSet, namespace, jobID string, all bool) ([]*structs.Deployment, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
var job *structs.Job
// Read job from state store
_, existing, err := txn.FirstWatch("jobs", "id", namespace, jobID)
@@ -587,11 +564,6 @@ func (s *StateStore) DeploymentsByJobID(ws memdb.WatchSet, namespace, jobID stri
func (s *StateStore) LatestDeploymentByJobID(ws memdb.WatchSet, namespace, jobID string) (*structs.Deployment, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Get an iterator over the deployments
iter, err := txn.Get("deployment", "job", namespace, jobID)
if err != nil {
@@ -820,7 +792,7 @@ func (s *StateStore) updateNodeDrainImpl(txn *memdb.Txn, index uint64, nodeID st
}
// Update the drain in the copy
copyNode.Drain = drain != nil // COMPAT: Remove in Nomad 0.9
copyNode.Drain = drain != nil // COMPAT: Remove in Nomad 0.10
copyNode.DrainStrategy = drain
if drain != nil {
copyNode.SchedulingEligibility = structs.NodeSchedulingIneligible
@@ -1025,11 +997,6 @@ func (s *StateStore) UpsertJobTxn(index uint64, job *structs.Job, txn Txn) error
// upsertJobImpl is the implementation for registering a job or updating a job definition
func (s *StateStore) upsertJobImpl(index uint64, job *structs.Job, keepVersion bool, txn *memdb.Txn) error {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if job.Namespace == "" {
job.Namespace = structs.DefaultNamespace
}
// Assert the namespace exists
if exists, err := s.namespaceExists(txn, job.Namespace); err != nil {
return err
@@ -1116,11 +1083,6 @@ func (s *StateStore) DeleteJob(index uint64, namespace, jobID string) error {
// DeleteJobTxn is used to deregister a job, like DeleteJob,
// but in a transaction. Useful for when making multiple modifications atomically
func (s *StateStore) DeleteJobTxn(index uint64, namespace, jobID string, txn Txn) error {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Lookup the node
existing, err := txn.First("jobs", "id", namespace, jobID)
if err != nil {
@@ -1164,11 +1126,6 @@ func (s *StateStore) DeleteJobTxn(index uint64, namespace, jobID string, txn Txn
// Update the modify index
pSummary.ModifyIndex = index
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if pSummary.Namespace == "" {
pSummary.Namespace = structs.DefaultNamespace
}
// Insert the summary
if err := txn.Insert("job_summary", pSummary); err != nil {
return fmt.Errorf("job summary insert failed: %v", err)
@@ -1207,11 +1164,6 @@ func (s *StateStore) DeleteJobTxn(index uint64, namespace, jobID string, txn Txn
// deleteJobVersions deletes all versions of the given job.
func (s *StateStore) deleteJobVersions(index uint64, job *structs.Job, txn *memdb.Txn) error {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if job.Namespace == "" {
job.Namespace = structs.DefaultNamespace
}
iter, err := txn.Get("job_version", "id_prefix", job.Namespace, job.ID)
if err != nil {
return err
@@ -1252,11 +1204,6 @@ func (s *StateStore) deleteJobVersions(index uint64, job *structs.Job, txn *memd
// upsertJobVersion inserts a job into its historic version table and limits the
// number of job versions that are tracked.
func (s *StateStore) upsertJobVersion(index uint64, job *structs.Job, txn *memdb.Txn) error {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if job.Namespace == "" {
job.Namespace = structs.DefaultNamespace
}
// Insert the job
if err := txn.Insert("job_version", job); err != nil {
return fmt.Errorf("failed to insert job into job_version table: %v", err)
@@ -1313,11 +1260,6 @@ func (s *StateStore) JobByID(ws memdb.WatchSet, namespace, id string) (*structs.
// JobByIDTxn is used to lookup a job by its ID, like JobByID. JobByID returns the job version
// accessible through in the transaction
func (s *StateStore) JobByIDTxn(ws memdb.WatchSet, namespace, id string, txn Txn) (*structs.Job, error) {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
watchCh, existing, err := txn.FirstWatch("jobs", "id", namespace, id)
if err != nil {
return nil, fmt.Errorf("job lookup failed: %v", err)
@@ -1334,11 +1276,6 @@ func (s *StateStore) JobByIDTxn(ws memdb.WatchSet, namespace, id string, txn Txn
func (s *StateStore) JobsByIDPrefix(ws memdb.WatchSet, namespace, id string) (memdb.ResultIterator, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
iter, err := txn.Get("jobs", "id_prefix", namespace, id)
if err != nil {
return nil, fmt.Errorf("job lookup failed: %v", err)
@@ -1353,11 +1290,6 @@ func (s *StateStore) JobsByIDPrefix(ws memdb.WatchSet, namespace, id string) (me
func (s *StateStore) JobVersionsByID(ws memdb.WatchSet, namespace, id string) ([]*structs.Job, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
return s.jobVersionByID(txn, &ws, namespace, id)
}
@@ -1365,11 +1297,6 @@ func (s *StateStore) JobVersionsByID(ws memdb.WatchSet, namespace, id string) ([
// versions of a job and is called under an existing transaction. A watch set
// can optionally be passed in to add the job histories to the watch set.
func (s *StateStore) jobVersionByID(txn *memdb.Txn, ws *memdb.WatchSet, namespace, id string) ([]*structs.Job, error) {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Get all the historic jobs for this ID
iter, err := txn.Get("job_version", "id_prefix", namespace, id)
if err != nil {
@@ -1407,10 +1334,6 @@ func (s *StateStore) jobVersionByID(txn *memdb.Txn, ws *memdb.WatchSet, namespac
// JobByIDAndVersion returns the job identified by its ID and Version. The
// passed watchset may be nil.
func (s *StateStore) JobByIDAndVersion(ws memdb.WatchSet, namespace, id string, version uint64) (*structs.Job, error) {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
txn := s.db.Txn(false)
return s.jobByIDAndVersionImpl(ws, namespace, id, version, txn)
}
@@ -1419,10 +1342,6 @@ func (s *StateStore) JobByIDAndVersion(ws memdb.WatchSet, namespace, id string,
// passed watchset may be nil.
func (s *StateStore) jobByIDAndVersionImpl(ws memdb.WatchSet, namespace, id string,
version uint64, txn *memdb.Txn) (*structs.Job, error) {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
watchCh, existing, err := txn.FirstWatch("job_version", "id", namespace, id, version)
if err != nil {
@@ -1537,11 +1456,6 @@ func (s *StateStore) JobsByGC(ws memdb.WatchSet, gc bool) (memdb.ResultIterator,
func (s *StateStore) JobSummaryByID(ws memdb.WatchSet, namespace, jobID string) (*structs.JobSummary, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
watchCh, existing, err := txn.FirstWatch("job_summary", "id", namespace, jobID)
if err != nil {
return nil, err
@@ -1576,11 +1490,6 @@ func (s *StateStore) JobSummaries(ws memdb.WatchSet) (memdb.ResultIterator, erro
func (s *StateStore) JobSummaryByPrefix(ws memdb.WatchSet, namespace, id string) (memdb.ResultIterator, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
iter, err := txn.Get("job_summary", "id_prefix", namespace, id)
if err != nil {
return nil, fmt.Errorf("eval lookup failed: %v", err)
@@ -1596,11 +1505,6 @@ func (s *StateStore) UpsertPeriodicLaunch(index uint64, launch *structs.Periodic
txn := s.db.Txn(true)
defer txn.Abort()
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if launch.Namespace == "" {
launch.Namespace = structs.DefaultNamespace
}
// Check if the job already exists
existing, err := txn.First("periodic_launch", "id", launch.Namespace, launch.ID)
if err != nil {
@@ -1643,11 +1547,6 @@ func (s *StateStore) DeletePeriodicLaunch(index uint64, namespace, jobID string)
// DeletePeriodicLaunchTxn is used to delete the periodic launch, like DeletePeriodicLaunch
// but in a transaction. Useful for when making multiple modifications atomically
func (s *StateStore) DeletePeriodicLaunchTxn(index uint64, namespace, jobID string, txn Txn) error {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Lookup the launch
existing, err := txn.First("periodic_launch", "id", namespace, jobID)
if err != nil {
@@ -1673,11 +1572,6 @@ func (s *StateStore) DeletePeriodicLaunchTxn(index uint64, namespace, jobID stri
func (s *StateStore) PeriodicLaunchByID(ws memdb.WatchSet, namespace, id string) (*structs.PeriodicLaunch, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
watchCh, existing, err := txn.FirstWatch("periodic_launch", "id", namespace, id)
if err != nil {
return nil, fmt.Errorf("periodic launch lookup failed: %v", err)
@@ -1751,11 +1645,6 @@ func (s *StateStore) nestedUpsertEval(txn *memdb.Txn, index uint64, eval *struct
return fmt.Errorf("eval lookup failed: %v", err)
}
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if eval.Namespace == "" {
eval.Namespace = structs.DefaultNamespace
}
// Update the indexes
if existing != nil {
eval.CreateIndex = existing.(*structs.Evaluation).CreateIndex
@@ -1787,11 +1676,6 @@ func (s *StateStore) nestedUpsertEval(txn *memdb.Txn, index uint64, eval *struct
// Insert the job summary
if hasSummaryChanged {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if js.Namespace == "" {
js.Namespace = structs.DefaultNamespace
}
js.ModifyIndex = index
if err := txn.Insert("job_summary", js); err != nil {
return fmt.Errorf("job summary insert failed: %v", err)
@@ -1826,11 +1710,6 @@ func (s *StateStore) nestedUpsertEval(txn *memdb.Txn, index uint64, eval *struct
newEval.StatusDescription = fmt.Sprintf("evaluation %q successful", newEval.ID)
newEval.ModifyIndex = index
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if newEval.Namespace == "" {
newEval.Namespace = structs.DefaultNamespace
}
if err := txn.Insert("evals", newEval); err != nil {
return fmt.Errorf("eval insert failed: %v", err)
}
@@ -1960,11 +1839,6 @@ func (s *StateStore) EvalsByIDPrefix(ws memdb.WatchSet, namespace, id string) (m
ws.Add(iter.WatchCh())
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Wrap the iterator in a filter
wrap := memdb.NewFilterIterator(iter, evalNamespaceFilter(namespace))
return wrap, nil
@@ -1987,11 +1861,6 @@ func evalNamespaceFilter(namespace string) func(interface{}) bool {
func (s *StateStore) EvalsByJob(ws memdb.WatchSet, namespace, jobID string) ([]*structs.Evaluation, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Get an iterator over the node allocations
iter, err := txn.Get("evals", "job_prefix", namespace, jobID)
if err != nil {
@@ -2092,11 +1961,6 @@ func (s *StateStore) nestedUpdateAllocFromClient(txn *memdb.Txn, index uint64, a
// Copy everything from the existing allocation
copyAlloc := exist.Copy()
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if copyAlloc.Namespace == "" {
copyAlloc.Namespace = structs.DefaultNamespace
}
// Pull in anything the client is the authority on
copyAlloc.ClientStatus = alloc.ClientStatus
copyAlloc.ClientDescription = alloc.ClientDescription
@@ -2230,11 +2094,6 @@ func (s *StateStore) upsertAllocsImpl(index uint64, allocs []*structs.Allocation
}
}
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if alloc.Namespace == "" {
alloc.Namespace = structs.DefaultNamespace
}
// OPTIMIZATION:
// These should be given a map of new to old allocation and the updates
// should be one on all changes. The current implementation causes O(n)
@@ -2459,11 +2318,6 @@ func (s *StateStore) AllocsByNodeTerminal(ws memdb.WatchSet, node string, termin
func (s *StateStore) AllocsByJob(ws memdb.WatchSet, namespace, jobID string, all bool) ([]*structs.Allocation, error) {
txn := s.db.Txn(false)
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Get the job
var job *structs.Job
rawJob, err := txn.First("jobs", "id", namespace, jobID)
@@ -2752,11 +2606,6 @@ func (s *StateStore) updateDeploymentStatusImpl(index uint64, u *structs.Deploym
copy.StatusDescription = u.StatusDescription
copy.ModifyIndex = index
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if copy.Namespace == "" {
copy.Namespace = structs.DefaultNamespace
}
// Insert the deployment
if err := txn.Insert("deployment", copy); err != nil {
return err
@@ -2783,11 +2632,6 @@ func (s *StateStore) UpdateJobStability(index uint64, namespace, jobID string, j
txn := s.db.Txn(true)
defer txn.Abort()
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
if err := s.updateJobStabilityImpl(index, namespace, jobID, jobVersion, stable, txn); err != nil {
return err
}
@@ -2798,11 +2642,6 @@ func (s *StateStore) UpdateJobStability(index uint64, namespace, jobID string, j
// updateJobStabilityImpl updates the stability of the given job and version
func (s *StateStore) updateJobStabilityImpl(index uint64, namespace, jobID string, jobVersion uint64, stable bool, txn *memdb.Txn) error {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if namespace == "" {
namespace = structs.DefaultNamespace
}
// Get the job that is referenced
job, err := s.jobByIDAndVersionImpl(nil, namespace, jobID, jobVersion, txn)
if err != nil {
@@ -3231,11 +3070,6 @@ func (s *StateStore) ReconcileJobSummaries(index uint64) error {
summary.Summary[tg.Name] = structs.TaskGroupSummary{}
}
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if job.Namespace == "" {
job.Namespace = structs.DefaultNamespace
}
// Find all the allocations for the jobs
iterAllocs, err := txn.Get("allocs", "job", job.Namespace, job.ID)
if err != nil {
@@ -3299,10 +3133,6 @@ func (s *StateStore) ReconcileJobSummaries(index uint64) error {
func (s *StateStore) setJobStatuses(index uint64, txn *memdb.Txn,
jobs map[structs.NamespacedID]string, evalDelete bool) error {
for tuple, forceStatus := range jobs {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if tuple.Namespace == "" {
tuple.Namespace = structs.DefaultNamespace
}
existing, err := txn.First("jobs", "id", tuple.Namespace, tuple.ID)
if err != nil {
@@ -3355,11 +3185,6 @@ func (s *StateStore) setJobStatus(index uint64, txn *memdb.Txn,
updated.Status = newStatus
updated.ModifyIndex = index
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if updated.Namespace == "" {
updated.Namespace = structs.DefaultNamespace
}
// Insert the job
if err := txn.Insert("jobs", updated); err != nil {
return fmt.Errorf("job insert failed: %v", err)
@@ -3385,11 +3210,6 @@ func (s *StateStore) setJobStatus(index uint64, txn *memdb.Txn,
pSummary.Children = new(structs.JobChildrenSummary)
}
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if pSummary.Namespace == "" {
pSummary.Namespace = structs.DefaultNamespace
}
// Determine the transition and update the correct fields
children := pSummary.Children
@@ -3436,11 +3256,6 @@ func (s *StateStore) setJobStatus(index uint64, txn *memdb.Txn,
}
func (s *StateStore) getJobStatus(txn *memdb.Txn, job *structs.Job, evalDelete bool) (string, error) {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if job.Namespace == "" {
job.Namespace = structs.DefaultNamespace
}
// System, Periodic and Parameterized jobs are running until explicitly
// stopped
if job.Type == structs.JobTypeSystem || job.IsParameterized() || job.IsPeriodic() {
@@ -3499,11 +3314,6 @@ func (s *StateStore) getJobStatus(txn *memdb.Txn, job *structs.Job, evalDelete b
func (s *StateStore) updateSummaryWithJob(index uint64, job *structs.Job,
txn *memdb.Txn) error {
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if job.Namespace == "" {
job.Namespace = structs.DefaultNamespace
}
// Update the job summary
summaryRaw, err := txn.First("job_summary", "id", job.Namespace, job.ID)
if err != nil {
@@ -3543,11 +3353,6 @@ func (s *StateStore) updateSummaryWithJob(index uint64, job *structs.Job,
if hasSummaryChanged {
summary.ModifyIndex = index
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if summary.Namespace == "" {
summary.Namespace = structs.DefaultNamespace
}
// Update the indexes table for job summary
if err := txn.Insert("index", &IndexEntry{"job_summary", index}); err != nil {
return fmt.Errorf("index update failed: %v", err)
@@ -3662,10 +3467,6 @@ func (s *StateStore) updateSummaryWithAlloc(index uint64, alloc *structs.Allocat
if alloc.Job == nil {
return nil
}
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if alloc.Namespace == "" {
alloc.Namespace = structs.DefaultNamespace
}
summaryRaw, err := txn.First("job_summary", "id", alloc.Namespace, alloc.JobID)
if err != nil {
@@ -3761,11 +3562,6 @@ func (s *StateStore) updateSummaryWithAlloc(index uint64, alloc *structs.Allocat
if summaryChanged {
jobSummary.ModifyIndex = index
// COMPAT 0.7: Upgrade old objects that do not have namespaces
if jobSummary.Namespace == "" {
jobSummary.Namespace = structs.DefaultNamespace
}
// Update the indexes table for job summary
if err := txn.Insert("index", &IndexEntry{"job_summary", index}); err != nil {
return fmt.Errorf("index update failed: %v", err)

12
scheduler/context.go
View File

@@ -282,10 +282,7 @@ func (e *EvalEligibility) GetClasses() map[string]bool {
// JobStatus returns the eligibility status of the job.
func (e *EvalEligibility) JobStatus(class string) ComputedClassFeasibility {
// COMPAT: Computed node class was introduced in 0.3. Clients running < 0.3
// will not have a computed class. The safest value to return is the escaped
// case, since it disables any optimization.
if e.jobEscaped || class == "" {
if e.jobEscaped {
return EvalComputedClassEscaped
}
@@ -307,13 +304,6 @@ func (e *EvalEligibility) SetJobEligibility(eligible bool, class string) {
// TaskGroupStatus returns the eligibility status of the task group.
func (e *EvalEligibility) TaskGroupStatus(tg, class string) ComputedClassFeasibility {
// COMPAT: Computed node class was introduced in 0.3. Clients running < 0.3
// will not have a computed class. The safest value to return is the escaped
// case, since it disables any optimization.
if class == "" {
return EvalComputedClassEscaped
}
if escaped, ok := e.tgEscapedConstraints[tg]; ok {
if escaped {
return EvalComputedClassEscaped

10
scheduler/context_test.go
View File

@@ -168,11 +168,6 @@ func TestEvalEligibility_JobStatus(t *testing.T) {
if status := e.JobStatus(cc); status != EvalComputedClassEligible {
t.Fatalf("JobStatus() returned %v; want %v", status, EvalComputedClassEligible)
}
// Check that if I pass an empty class it returns escaped
if status := e.JobStatus(""); status != EvalComputedClassEscaped {
t.Fatalf("JobStatus() returned %v; want %v", status, EvalComputedClassEscaped)
}
}
func TestEvalEligibility_TaskGroupStatus(t *testing.T) {
@@ -195,11 +190,6 @@ func TestEvalEligibility_TaskGroupStatus(t *testing.T) {
if status := e.TaskGroupStatus(tg, cc); status != EvalComputedClassEligible {
t.Fatalf("TaskGroupStatus() returned %v; want %v", status, EvalComputedClassEligible)
}
// Check that if I pass an empty class it returns escaped
if status := e.TaskGroupStatus(tg, ""); status != EvalComputedClassEscaped {
t.Fatalf("TaskGroupStatus() returned %v; want %v", status, EvalComputedClassEscaped)
}
}
func TestEvalEligibility_SetJob(t *testing.T) {