agent: compare unit hashes when reconciling

This incorporates hashes into the decision when agent reconciler is
calculating what tasks should be performed.

agent/agent.go

@@ -103,7 +103,11 @@ func (a *Agent) stopUnit(unitName string) {
 	a.um.TriggerStop(unitName)
 }
 
-type unitStates map[string]job.JobState
+type unitState struct {
+	state job.JobState
+	hash  string
+}
+type unitStates map[string]unitState
 
 // units returns a map representing the current state of units known by the agent.
 func (a *Agent) units() (unitStates, error) {
@@ -117,25 +121,34 @@ func (a *Agent) units() (unitStates, error) {
 		loaded.Add(jName)
 	}
 
-	units, err := a.um.Units()
+	unitFiles, err := a.um.Units()
 	if err != nil {
 		return nil, fmt.Errorf("failed fetching loaded units from UnitManager: %v", err)
 	}
 
 	filter := pkg.NewUnsafeSet()
-	for _, u := range units {
+	for _, u := range unitFiles {
 		filter.Add(u)
 	}
 
+	units, err := a.um.GetUnitStates(filter)
+	if err != nil {
+		return nil, fmt.Errorf("failed fetching unit states from UnitManager: %v", err)
+	}
+
 	states := make(unitStates)
-	for _, uName := range units {
+	for uName, state := range units {
 		js := job.JobStateInactive
 		if loaded.Contains(uName) {
 			js = job.JobStateLoaded
 		} else if launched.Contains(uName) {
 			js = job.JobStateLaunched
 		}
-		states[uName] = js
+		us := unitState{
+			state: js,
+			hash:  state.UnitHash,
+		}
+		states[uName] = us
 	}
 
 	return states, nil

agent/agent_test.go

@@ -64,7 +64,9 @@ func TestAgentLoadUnloadUnit(t *testing.T) {
 
 	jsLoaded := job.JobStateLoaded
 	expectUnits := unitStates{
-		"foo.service": jsLoaded,
+		"foo.service": unitState{
+			state: jsLoaded,
+		},
 	}
 
 	if !reflect.DeepEqual(expectUnits, units) {
@@ -107,7 +109,9 @@ func TestAgentLoadStartStopUnit(t *testing.T) {
 
 	jsLaunched := job.JobStateLaunched
 	expectUnits := unitStates{
-		"foo.service": jsLaunched,
+		"foo.service": unitState{
+			state: jsLaunched,
+		},
 	}
 
 	if !reflect.DeepEqual(expectUnits, units) {
@@ -123,7 +127,9 @@ func TestAgentLoadStartStopUnit(t *testing.T) {
 
 	jsLoaded := job.JobStateLoaded
 	expectUnits = unitStates{
-		"foo.service": jsLoaded,
+		"foo.service": unitState{
+			state: jsLoaded,
+		},
 	}
 
 	if !reflect.DeepEqual(expectUnits, units) {

agent/reconcile.go

@@ -65,12 +65,12 @@ func (ar *AgentReconciler) Reconcile(a *Agent) {
 		return
 	}
 
-	for tc := range ar.calculateTaskChainsForJobs(dAgentState, cAgentState) {
+	for tc := range ar.calculateTaskChainsForUnits(dAgentState, cAgentState) {
 		ar.launchTaskChain(tc, a)
 	}
 }
 
-// Purge attempts to unload all Jobs that have been loaded locally
+// Purge attempts to unload all Units that have been loaded locally
 func (ar *AgentReconciler) Purge(a *Agent) {
 	for {
 		cAgentState, err := a.units()
@@ -144,10 +144,10 @@ func desiredAgentState(a *Agent, reg registry.Registry) (*AgentState, error) {
 	return &as, nil
 }
 
-// calculateTaskChainsForJobs compares the desired and current state of an Agent.
+// calculateTaskChainsForUnits compares the desired and current state of an Agent.
 // The generated taskChains represent what should be done to make the desired
 // state match the current state.
-func (ar *AgentReconciler) calculateTaskChainsForJobs(dState *AgentState, cState unitStates) <-chan taskChain {
+func (ar *AgentReconciler) calculateTaskChainsForUnits(dState *AgentState, cState unitStates) <-chan taskChain {
 	tcChan := make(chan taskChain)
 	go func() {
 		jobs := pkg.NewUnsafeSet()
@@ -160,7 +160,7 @@ func (ar *AgentReconciler) calculateTaskChainsForJobs(dState *AgentState, cState
 		}
 
 		for _, name := range jobs.Values() {
-			tc := ar.calculateTaskChainForJob(dState, cState, name)
+			tc := ar.calculateTaskChainForUnit(dState, cState, name)
 			if tc == nil {
 				continue
 			}
@@ -173,14 +173,20 @@ func (ar *AgentReconciler) calculateTaskChainsForJobs(dState *AgentState, cState
 	return tcChan
 }
 
-func (ar *AgentReconciler) calculateTaskChainForJob(dState *AgentState, cState unitStates, jName string) *taskChain {
+func (ar *AgentReconciler) calculateTaskChainForUnit(dState *AgentState, cState unitStates, jName string) *taskChain {
 	var dJob *job.Unit
+	var dJHash string
 	if dState != nil {
 		dJob = dState.Units[jName]
+		if dJob != nil {
+			dJHash = dJob.Unit.Hash().String()
+		}
 	}
 	var cJState *job.JobState
-	if state, ok := cState[jName]; ok {
-		cJState = &state
+	var cJHash string
+	if us, ok := cState[jName]; ok {
+		cJState = &us.state
+		cJHash = us.hash
 	}
 	if dJob == nil && cJState == nil {
 		log.Errorf("Desired state and current state of Job(%s) nil, not sure what to do", jName)
@@ -225,6 +231,19 @@ func (ar *AgentReconciler) calculateTaskChainForJob(dState *AgentState, cState u
 		return &tc
 	}
 
+	if cJHash != dJHash {
+		log.V(1).Infof("Desired hash %q differs to current hash %s of Job(%s) - unloading", dJHash, cJHash, jName)
+		t := task{
+			typ:    taskTypeUnloadUnit,
+			reason: taskReasonLoadedButHashDiffers,
+		}
+		u := &job.Unit{
+			Name: jName,
+		}
+		tc := newTaskChain(u, t)
+		return &tc
+	}
+
 	if *cJState == dJob.TargetState {
 		log.V(1).Infof("Desired state %q matches current state of Job(%s), nothing to do", *cJState, jName)
 		return nil