functional: make replace tests wait for N active units

api/units.go

@@ -76,6 +76,7 @@ func (ur *unitsResource) set(rw http.ResponseWriter, req *http.Request, item str
 	}
 
 	var su schema.Unit
+	newUnit := false
 	dec := json.NewDecoder(req.Body)
 	err := dec.Decode(&su)
 	if err != nil {
@@ -105,11 +106,28 @@ func (ur *unitsResource) set(rw http.ResponseWriter, req *http.Request, item str
 		if len(su.Options) == 0 {
 			err := errors.New("unit does not exist and options field empty")
 			sendError(rw, http.StatusConflict, err)
+			return
 		} else if err := ValidateOptions(su.Options); err != nil {
 			sendError(rw, http.StatusBadRequest, err)
+			return
 		} else {
-			ur.create(rw, su.Name, &su)
+			newUnit = true
 		}
+	} else if eu.Name == su.Name && len(su.Options) > 0 {
+		// There is already a unit with the same name
+		// check the hashes if they do not match then we will
+		// create a new unit with the same name and later
+		// the job will be updated to this new unit.
+		// if su.Options == 0 then probably we don't want to update
+		// the Unit Options but only its target job state, in
+		// this case just ignore.
+		a := schema.MapSchemaUnitOptionsToUnitFile(su.Options)
+		b := schema.MapSchemaUnitOptionsToUnitFile(eu.Options)
+		newUnit = !unit.MatchUnitFiles(a, b)
+	}
+
+	if newUnit {
+		ur.create(rw, su.Name, &su)
 		return
 	}
 

fleetctl/fleetctl.go

@@ -75,9 +75,10 @@ var (
 
 	// flags used by all commands
 	globalFlags = struct {
-		Debug   bool
-		Version bool
-		Help    bool
+		Debug          bool
+		Version        bool
+		Help           bool
+		currentCommand string
 
 		ClientDriver    string
 		ExperimentalAPI bool
@@ -103,6 +104,7 @@ var (
 		Full          bool
 		NoLegend      bool
 		NoBlock       bool
+		Replace       bool
 		BlockAttempts int
 		Fields        string
 		SSHPort       int
@@ -287,6 +289,10 @@ func main() {
 		}
 	}
 
+	// We use this to know in which context are we:
+	// submit, load or start
+	globalFlags.currentCommand = cmd.Name
+
 	os.Exit(cmd.Run(cmd.Flags.Args()))
 
 }
@@ -555,7 +561,7 @@ func getUnitFileFromTemplate(uni *unit.UnitNameInfo, fileName string) (*unit.Uni
 	}
 
 	if tmpl != nil {
-		warnOnDifferentLocalUnit(fileName, tmpl)
+		isLocalUnitDifferent(fileName, tmpl, true, false)
 		uf = schema.MapSchemaUnitOptionsToUnitFile(tmpl.Options)
 		log.Debugf("Template Unit(%s) found in registry", uni.Template)
 	} else {
@@ -663,6 +669,81 @@ func createUnit(name string, uf *unit.UnitFile) (*schema.Unit, error) {
 	return &u, nil
 }
 
+// checkReplaceUnitState checks if the unit should be replaced
+// It takes a Unit object as a parameter
+// It returns 0 on success and if the unit should be replaced, 1 if the
+// unit should not be replaced; and any error acountered
+func checkReplaceUnitState(unit *schema.Unit) (int, error) {
+	allowedReplace := map[string][]job.JobState{
+		"submit": []job.JobState{
+			job.JobStateInactive,
+		},
+		"load": []job.JobState{
+			job.JobStateInactive,
+			job.JobStateLoaded,
+		},
+		"start": []job.JobState{
+			job.JobStateInactive,
+			job.JobStateLoaded,
+			job.JobStateLaunched,
+		},
+	}
+
+	if allowedJobs, ok := allowedReplace[globalFlags.currentCommand]; ok {
+		for _, j := range allowedJobs {
+			if job.JobState(unit.DesiredState) == j {
+				return 0, nil
+			}
+		}
+		stderr("Warning: can not replace Unit(%s) in state '%s', use the appropriate command", unit.Name, unit.DesiredState)
+	} else {
+		return 1, fmt.Errorf("error 'replace' is not supported for this command")
+	}
+
+	return 1, nil
+}
+
+// checkUnitCreation checks if the unit should be created
+// It takes a unit file path as a parameter.
+// It returns 0 on success and if the unit should be created, 1 if the
+// unit should not be created; and any error acountered
+func checkUnitCreation(arg string) (int, error) {
+	name := unitNameMangle(arg)
+
+	// First, check if there already exists a Unit by the given name in the Registry
+	unit, err := cAPI.Unit(name)
+	if err != nil {
+		return 1, fmt.Errorf("error retrieving Unit(%s) from Registry: %v", name, err)
+	}
+
+	// check if the unit is running
+	if unit == nil {
+		if sharedFlags.Replace {
+			log.Debugf("Unit(%s) was not found in Registry", name)
+		}
+		// Create a new unit
+		return 0, nil
+	}
+
+	// if sharedFlags.Replace is not set then we warn
+	different, err := isLocalUnitDifferent(arg, unit, !sharedFlags.Replace, false)
+
+	// if sharedFlags.Replace is set then we fail for errors
+	if sharedFlags.Replace {
+		if err != nil {
+			return 1, err
+		} else if different {
+			return checkReplaceUnitState(unit)
+		} else {
+			stdout("Found same Unit(%s) in Registry, nothing to do", unit.Name)
+		}
+	} else if different == false {
+		log.Debugf("Found same Unit(%s) in Registry, no need to recreate it", name)
+	}
+
+	return 1, nil
+}
+
 // lazyCreateUnits iterates over a set of unit names and, for each, attempts to
 // ensure that a unit by that name exists in the Registry, by checking a number
 // of conditions and acting on the first one that succeeds, in order of:
@@ -680,14 +761,10 @@ func lazyCreateUnits(args []string) error {
 		arg = maybeAppendDefaultUnitType(arg)
 		name := unitNameMangle(arg)
 
-		// First, check if there already exists a Unit by the given name in the Registry
-		u, err := cAPI.Unit(name)
+		ret, err := checkUnitCreation(arg)
 		if err != nil {
-			return fmt.Errorf("error retrieving Unit(%s) from Registry: %v", name, err)
-		}
-		if u != nil {
-			log.Debugf("Found Unit(%s) in Registry, no need to recreate it", name)
-			warnOnDifferentLocalUnit(arg, u)
+			return err
+		} else if ret != 0 {
 			continue
 		}
 
@@ -726,24 +803,52 @@ func lazyCreateUnits(args []string) error {
 	return nil
 }
 
-func warnOnDifferentLocalUnit(loc string, su *schema.Unit) {
-	suf := schema.MapSchemaUnitOptionsToUnitFile(su.Options)
-	if _, err := os.Stat(loc); !os.IsNotExist(err) {
-		luf, err := getUnitFromFile(loc)
-		if err == nil && luf.Hash() != suf.Hash() {
-			stderr("WARNING: Unit %s in registry differs from local unit file %s", su.Name, loc)
-			return
+// matchLocalFileAndUnit compares a file with a Unit
+// Returns true if the contents of the file matches the unit one, false
+// otherwise; and any error ocountered
+func matchLocalFileAndUnit(file string, su *schema.Unit) (bool, error) {
+	result := false
+	a := schema.MapSchemaUnitOptionsToUnitFile(su.Options)
+
+	_, err := os.Stat(file)
+	if err == nil {
+		b, err := getUnitFromFile(file)
+		if err == nil {
+			result = unit.MatchUnitFiles(a, b)
 		}
 	}
-	if uni := unit.NewUnitNameInfo(path.Base(loc)); uni != nil && uni.IsInstance() {
-		file := path.Join(path.Dir(loc), uni.Template)
-		if _, err := os.Stat(file); !os.IsNotExist(err) {
-			tmpl, err := getUnitFromFile(file)
-			if err == nil && tmpl.Hash() != suf.Hash() {
-				stderr("WARNING: Unit %s in registry differs from local template unit file %s", su.Name, uni.Template)
-			}
+
+	return result, err
+}
+
+func isLocalUnitDifferent(file string, su *schema.Unit, warnIfDifferent bool, fatal bool) (bool, error) {
+	result, err := matchLocalFileAndUnit(file, su)
+	if err == nil {
+		if result == false && warnIfDifferent {
+			stderr("WARNING: Unit %s in registry differs from local unit file %s", su.Name, file)
+		}
+		return !result, nil
+	} else if fatal {
+		return false, err
+	}
+
+	info := unit.NewUnitNameInfo(path.Base(file))
+	if info == nil {
+		return false, fmt.Errorf("error extracting information from unit name %s", file)
+	} else if !info.IsInstance() {
+		return false, fmt.Errorf("error Unit %s does not seem to be a template unit", file)
+	}
+
+	templFile := path.Join(path.Dir(file), info.Template)
+	result, err = matchLocalFileAndUnit(templFile, su)
+	if err == nil {
+		if result == false && warnIfDifferent {
+			stderr("WARNING: Unit %s in registry differs from local template unit file %s", su.Name, info.Template)
 		}
+		return !result, nil
 	}
+
+	return false, err
 }
 
 func lazyLoadUnits(args []string) ([]*schema.Unit, error) {

fleetctl/load.go

@@ -43,6 +43,7 @@ func init() {
 	cmdLoadUnits.Flags.BoolVar(&sharedFlags.Sign, "sign", false, "DEPRECATED - this option cannot be used")
 	cmdLoadUnits.Flags.IntVar(&sharedFlags.BlockAttempts, "block-attempts", 0, "Wait until the jobs are loaded, performing up to N attempts before giving up. A value of 0 indicates no limit. Does not apply to global units.")
 	cmdLoadUnits.Flags.BoolVar(&sharedFlags.NoBlock, "no-block", false, "Do not wait until the jobs have been loaded before exiting. Always the case for global units.")
+	cmdLoadUnits.Flags.BoolVar(&sharedFlags.Replace, "replace", false, "Replace the old loaded unit with a new one.")
 }
 
 func runLoadUnits(args []string) (exit int) {

fleetctl/start.go

@@ -51,6 +51,7 @@ func init() {
 	cmdStartUnit.Flags.BoolVar(&sharedFlags.Sign, "sign", false, "DEPRECATED - this option cannot be used")
 	cmdStartUnit.Flags.IntVar(&sharedFlags.BlockAttempts, "block-attempts", 0, "Wait until the units are launched, performing up to N attempts before giving up. A value of 0 indicates no limit. Does not apply to global units.")
 	cmdStartUnit.Flags.BoolVar(&sharedFlags.NoBlock, "no-block", false, "Do not wait until the units have launched before exiting. Always the case for global units.")
+	cmdStartUnit.Flags.BoolVar(&sharedFlags.Replace, "replace", false, "Replace the old started unit with a new one.")
 }
 
 func runStartUnit(args []string) (exit int) {

fleetctl/submit.go

@@ -33,6 +33,7 @@ Submit a directory of units with glob matching:
 
 func init() {
 	cmdSubmitUnit.Flags.BoolVar(&sharedFlags.Sign, "sign", false, "DEPRECATED - this option cannot be used")
+	cmdSubmitUnit.Flags.BoolVar(&sharedFlags.Replace, "replace", false, "Replace the old submitted unit with a new one.")
 }
 
 func runSubmitUnits(args []string) (exit int) {