Enhance the error message (#113)

* Reporting clear msg when something wrong

* add error handle error message

* refactor error message base on feedback from Frank.

* lost one error message base on feedback

* update error handling for schemas

Author: lixingwang

definition/definition_ser.go

@@ -74,9 +74,8 @@ func NewDefinition(rep *DefinitionRep) (def *Definition, err error) {
 		for _, taskRep := range rep.Tasks {
 
 			task, err := createTask(def, taskRep, ef)
-
 			if err != nil {
-				return nil, fmt.Errorf("create task [%s] error: %s", taskRep.ID, err.Error())
+				return nil, fmt.Errorf("error creating task [%s] in flow [%s]: %s", taskRep.ID, rep.Name, err.Error())
 			}
 			def.tasks[task.id] = task
 		}
@@ -88,7 +87,7 @@ func NewDefinition(rep *DefinitionRep) (def *Definition, err error) {
 
 			link, err := createLink(def.tasks, linkRep, id, ef)
 			if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("error creating link [%s] in flow [%s]: %s", linkRep.Name, rep.Name, err.Error())
 			}
 
 			def.links[link.id] = link
@@ -107,9 +106,8 @@ func NewDefinition(rep *DefinitionRep) (def *Definition, err error) {
 			for _, taskRep := range rep.ErrorHandler.Tasks {
 
 				task, err := createTask(def, taskRep, ef)
-
 				if err != nil {
-					return nil, err
+					return nil, fmt.Errorf("error creating task [%s] in flow [%s]'s error handler:%s", taskRep.ID, rep.Name, err.Error())
 				}
 				errorHandler.tasks[task.id] = task
 			}
@@ -123,7 +121,7 @@ func NewDefinition(rep *DefinitionRep) (def *Definition, err error) {
 
 				link, err := createLink(errorHandler.tasks, linkRep, id+idOffset, ef)
 				if err != nil {
-					return nil, err
+					return nil, fmt.Errorf("error creating link [%s] in flow [%s]'s error handler:%s", linkRep.Name, rep.Name, err.Error())
 				}
 				errorHandler.links[link.id] = link
 			}
@@ -228,7 +226,7 @@ func createActivityConfig(task *Task, rep *activity.Config, ef expression.Factor
 		for name, value := range rep.Settings {
 			activityCfg.settings[name], err = metadata.ResolveSettingValue(name, value, mdSettings, ef)
 			if err != nil {
-				return nil, err
+				return nil, fmt.Errorf("unable to resolve setting [%s]'s value [%s]:%s", name, value, err.Error())
 			}
 		}
 	}
@@ -266,7 +264,7 @@ func createActivityConfig(task *Task, rep *activity.Config, ef expression.Factor
 							}
 						}
 					}
-					return nil, fmt.Errorf("convert value [%+v] to type [%s] error: %s", v, fieldMetaddata.Type(), err.Error())
+					return nil, fmt.Errorf("unable to convert input [%s]'s value [%s] to type [%s]:%s", k, v, fieldMetaddata.Type(), err.Error())
 				}
 				input[k] = newVal
 			} else {
@@ -291,7 +289,7 @@ func createActivityConfig(task *Task, rep *activity.Config, ef expression.Factor
 			if ok {
 				v, err = coerce.ToType(v, fieldMetaddata.Type())
 				if err != nil {
-					return nil, fmt.Errorf("convert value [%+v] to type [%s] error: %s", v, fieldMetaddata.Type(), err.Error())
+					return nil, fmt.Errorf("unable to convert output [%s]'s value [%s] to type [%s]:%s", k, v, fieldMetaddata.Type(), err.Error())
 				}
 				output[k] = v
 			} else {
@@ -329,7 +327,7 @@ func createActivityConfig(task *Task, rep *activity.Config, ef expression.Factor
 			for name, def := range in {
 				s, err := schema.FindOrCreate(def)
 				if err != nil {
-					return nil, err
+					return nil, fmt.Errorf("unable to find/create input [%s]'s schema [%+v]:%s", name, def, err.Error())
 				}
 				activityCfg.inputSchemas[name] = s
 			}
@@ -340,7 +338,7 @@ func createActivityConfig(task *Task, rep *activity.Config, ef expression.Factor
 			for name, def := range out {
 				s, err := schema.FindOrCreate(def)
 				if err != nil {
-					return nil, err
+					return nil, fmt.Errorf("unable to find/create output [%s]'s schema [%+v]:%s", name, def, err.Error())
 				}
 				activityCfg.outputSchemas[name] = s
 			}
@@ -476,7 +474,7 @@ func getLoopCfg(settings map[string]interface{}, taskType string, ef expression.
 		lcd, err = getLoopCfgDef(setting)
 	} else if setting, ok := settings["doWhile"]; ok {
 		lcd, err = getLoopCfgDef(setting)
-		if accum,ok := settings["accumulate"];  lcd != nil && ok {
+		if accum, ok := settings["accumulate"]; lcd != nil && ok {
 			lcd.Accumulate, _ = coerce.ToBool(accum)
 		}
 	} else {
@@ -511,7 +509,7 @@ func getLoopCfgDef(setting interface{}) (*loopCfgDef, error) {
 		lcd.IterateOn = lcd.Iterate
 	}
 
-	if lcd.Condition == "" && lcd.IterateOn == ""  {
+	if lcd.Condition == "" && lcd.IterateOn == "" {
 		return nil, nil
 	}
 

support/manager.go

@@ -181,7 +181,7 @@ func materializeFlow(flowRep *definition.DefinitionRep) (*definition.Definition,
 
 	def, err := definition.NewDefinition(flowRep)
 	if err != nil {
-		return nil, fmt.Errorf("error unmarshalling flow: %s", err.Error())
+		return nil, err
 	}
 
 	//todo validate flow

support/resource.go

@@ -23,7 +23,7 @@ func (*FlowLoader) LoadResource(config *resource.Config) (*resource.Resource, er
 	var defRep *definition.DefinitionRep
 	err := json.Unmarshal(flowDefBytes, &defRep)
 	if err != nil {
-		return nil, fmt.Errorf("error marshalling flow resource with id '%s', %s", config.ID, err.Error())
+		return nil, fmt.Errorf("error loading flow resource with id '%s': %s", config.ID, err.Error())
 	}
 
 	flow, err := materializeFlow(defRep)