go_routine_writing_file.go

gistfile1.txt

func doKickoff2(opts Option) {
	scriptName := opts.Kickoff2.ScriptName
	configFile := opts.Kickoff2.ConfigFile
	contentByte, err := ioutil.ReadFile(opts.Kickoff2.ConfigFile)
	if err != nil {
		log.Error("%v", err)
		return
	}
	var config TomlConfig
	_, err = toml.Decode(string(contentByte), &config)
	if err != nil {
		log.Error("%v", err)
		return
	}
	if config.LogDirectory == "" {
		log.Error("Please specify log directory in .toml file")
		return
	}
	log.Info("Kickoff " + scriptName + " following config in " + configFile)
	scriptAllInfo, ok := config.Scripts2[scriptName]
	if !ok {
		log.Error("No " + scriptName + " script configurations in " + configFile)
		return
	}
	log.Info("[Kick off]\nscript: %s \ninventory_model_url: %s\nmodel_field_name: %s\nmatch_inventory_value: %v\nother_values: %v\narguments: %v\nnumber_of_execution: %d\n",
		scriptAllInfo.ScriptPath,
		scriptAllInfo.ModelUrl,
		scriptAllInfo.ModelFieldName,
		scriptAllInfo.MatchValues,
		scriptAllInfo.OtherValues,
		scriptAllInfo.Arguments,
		scriptAllInfo.NumberOfExecution)
	if opts.TargetInventory == "" {
		log.Info("Please specify inventory url.")
		return
	}
	request := gorequest.New()
	targetUrl := opts.TargetInventory + scriptAllInfo.ModelUrl
	request.Get(targetUrl)
	for _, matchValueArr := range scriptAllInfo.MatchValues {
		if len(matchValueArr) != 2 {
			log.Error("Provided match_values in config does not have field and value")
		}
		field := matchValueArr[0]
		value := matchValueArr[1]
		request.Query(field + "=" + value)
	}
	request.Query("limit=1000")
	resp, body, errs := request.End()
	if errs != nil {
		log.Error("%v", errs)
		return
	} else if resp.StatusCode != 200 {
		log.Error("%v", resp.Status)
		return
	}
	var data []map[string]interface{}
	if err := json.Unmarshal([]byte(body), &data); err != nil {
		log.Error("Response cannot be unmarshalled")
		log.Debug(body)
		return
	}
	limit := scriptAllInfo.NumberOfExecution
	argsNum := len(scriptAllInfo.Arguments)
	log.Info("Found %d results but limit to %d", len(data), scriptAllInfo.NumberOfExecution)
	var wg sync.WaitGroup
	for i := 0; i < len(data) && i < limit; i++ {
		item := data[i]
		itemName := item[scriptAllInfo.ModelFieldName].(string)
		log.Info("Processing item(" + itemName + ")")

		args := make([]string, argsNum)
		cancel := false
		for argIdx, argName := range scriptAllInfo.Arguments {
			if argName == "inventory_url" {
				args[argIdx] = opts.TargetInventory
			} else {
				found := false
				for _, otherValueArr := range scriptAllInfo.OtherValues {
					if otherValueArr[0] == argName {
						args[argIdx] = otherValueArr[1]
						found = true
						break
					}
				}
				if !found {
					switch item[argName].(type) {
					case float64:
						args[argIdx] = strconv.Itoa(int(item[argName].(float64)))
					case string:
						args[argIdx] = item[argName].(string)
					default:
						log.Error("No data in argument - " + argName)
						log.Error("Will not kickoff for this item - " + itemName)
						log.Debug("Showing response values -  %v", item)
						cancel = true
					}
					if cancel {
						break
					}
				}
			}
		}
		if cancel {
			continue
		}
		wg.Add(1)
		go func(gScriptAllInfo script2, gScriptName string, gItemName string, gArgs []string) {
			defer wg.Done()
			layout := "20060102_1504"
			err := os.MkdirAll(config.LogDirectory+"/"+gScriptName, 0755)
			if err != nil {
				log.Error("Error making folder")
				log.Error("%v", err)
				return
			}
			logFile, err := os.Create(config.LogDirectory + "/" + gScriptName + "/" + gItemName + "." + time.Now().Format(layout) + ".log")
			if err != nil {
				log.Error("Cannot create logfile")
				log.Error("%v", err)
				return
			}

			// declare writer for logfile
			wLogFile := bufio.NewWriter(logFile)

			log.Info("Starting to run - %s "+"%v", gScriptAllInfo.ScriptPath, gArgs)
			cmd := exec.Command(gScriptAllInfo.ScriptPath, gArgs...)
			// create pipe to use reader to write to stdout and logfile in another goroutine
			r, w := io.Pipe()
			cmd.Stderr = w
			cmd.Stdout = w

			if err := cmd.Start(); err != nil {
				log.Error("Unsuccessful running script for " + gItemName)
				log.Error("%v", err)
				return
			}
			var wgWriting sync.WaitGroup
			wgWriting.Add(1)
			go func(reader io.Reader) {
				scanner := bufio.NewScanner(reader)
				for scanner.Scan() {
					// writing to stdout
					log.Info("%s", scanner.Text())
					// writing to logfile
					if _, err := wLogFile.WriteString(scanner.Text() + "\n"); err != nil {
						log.Error("%v", err)
					}
					if err = wLogFile.Flush(); err != nil {
						log.Error("%v", err)
					}

				}
				if err := scanner.Err(); err != nil {
					log.Error("%v", err)
				}
				logFile.Close()
				wgWriting.Done()
				log.Info("END")
			}(r)
			cmd.Wait()
			wgWriting.Wait()
			log.Info("Finished running - %s "+"%v", gScriptAllInfo.ScriptPath, gArgs)
		}(scriptAllInfo, scriptName, itemName, args)
	}
	wg.Wait()
	return

}