Golang github.com-asiainfoLDP-datahub-utils-clog.Debug类(方法)实例源码

func saveEntryPoint(ep string) {
	log.Println("TODO save ep to db")
	count := `SELECT COUNT(*) FROM DH_DAEMON;`
	row, err := g_ds.QueryRow(count)
	if err != nil {
		l := log.Error(count, "error.", err)
		logq.LogPutqueue(l)
	}
	var c int
	row.Scan(&c)
	if c > 0 {
		Update := fmt.Sprintf(`UPDATE DH_DAEMON SET ENTRYPOINT='%s';`, ep)
		log.Debug(Update)
		if _, e := g_ds.Update(Update); e != nil {
			l := log.Error(Update, "error.", e)
			logq.LogPutqueue(l)
		}
	} else {
		Insert := fmt.Sprintf(`INSERT INTO DH_DAEMON (ENTRYPOINT) VALUES ('%s');`, ep)
		log.Debug(c, Insert)
		if _, e := g_ds.Insert(Insert); e != nil {
			l := log.Error(Insert, "error.", e)
			logq.LogPutqueue(l)
		}
	}
}

func saveDaemonID(id string) {
	log.Println("TODO save daemonid to db when srv returns code 0.")
	count := `SELECT COUNT(*) FROM DH_DAEMON;`
	row, err := g_ds.QueryRow(count)
	if err != nil {
		l := log.Error(count, "error.", err)
		logq.LogPutqueue(l)
	}
	var c int
	row.Scan(&c)
	if c > 0 {
		Update := fmt.Sprintf(`UPDATE DH_DAEMON SET DAEMONID='%s';`, id)
		log.Debug(Update)
		if _, e := g_ds.Update(Update); e != nil {
			l := log.Error(Update, "error.", e)
			logq.LogPutqueue(l)
		}
	} else {
		Insert := fmt.Sprintf(`INSERT INTO DH_DAEMON (DAEMONID) VALUES ('%s');`, id)
		log.Debug(c, Insert)
		if _, e := g_ds.Insert(Insert); e != nil {
			l := log.Error(Insert, "error.", e)
			logq.LogPutqueue(l)
		}
	}
}

func publishedOfDatapoolHandler(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	log.Debug(r.URL.Path, "published of a datapool", r)
	r.ParseForm()
	datapool := ps.ByName("dpname")
	status := "published"

	count := getRepoCountByDp(datapool, status)
	offset, limit := optionalOffsetAndSize(r, 10, 1, 100)
	log.Debug("offset, limit", offset, limit)
	validateOffsetAndLimit(count, &offset, &limit)

	repoInfos, err := GetRepoInfo(datapool, status, offset, limit)

	log.Debug(repoInfos, offset, limit)

	if err != nil {
		log.Error(err)
		JsonResult(w, http.StatusInternalServerError, cmd.InternalError, err.Error(), nil)
		return
	}

	if len(repoInfos) == 0 {
		msg := fmt.Sprintf("No published dataitem in %s.", datapool)
		JsonResult(w, http.StatusOK, cmd.ErrorPublishedItemEmpty, msg, nil)
	} else {
		msg := fmt.Sprintf("Dataitems have been published into %s.", datapool)
		JsonResult(w, http.StatusOK, cmd.ResultOK, msg, newQueryListResult(count, &repoInfos))
	}
}

func pulledOfRepoHandler(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	log.Debug(r.URL.Path, "item pulled of a repository")
	r.ParseForm()
	dpName := ps.ByName("dpname")
	repoName := ps.ByName("repo")

	isPublished := "N"
	count := getItemCountByDpRepo(dpName, repoName, isPublished)
	offset, limit := optionalOffsetAndSize(r, 10, 1, 100)
	validateOffsetAndLimit(count, &offset, &limit)

	pulledRepoItems, err := GetPulledRepoInfo(dpName, repoName, offset, limit)
	if err != nil {
		log.Debug(err)
		JsonResult(w, http.StatusInternalServerError, cmd.InternalError, err.Error(), nil)
		return
	}

	if len(pulledRepoItems) == 0 {
		msg := fmt.Sprintf("Pulled DataItem of %s is empty.", repoName)
		JsonResult(w, http.StatusOK, cmd.ErrorPublishedItemEmpty, msg, nil)
	} else {
		msg := fmt.Sprintf("All DataItems have been pulled of %s.", repoName)
		JsonResult(w, http.StatusOK, cmd.ResultOK, msg, newQueryListResult(count, pulledRepoItems))
	}
}

func deleteItemsAccordingToHeartbeat(body []byte) {
	log.Debug("deleteItemsAccordingToHeartbeat() BEGIN:", string(body))
	result := ds.Result{}
	itemEvent := &Event{}
	result.Data = itemEvent
	itemsdelete := []ItemDel{}
	itemEvent.Data = &itemsdelete

	if err := json.Unmarshal(body, &result); err == nil {
		log.Debug("items delete:", itemsdelete)
		for _, v := range itemsdelete {
			log.Debugf("delete item according to heartbeat: %v/%v\n", v.Repname, v.Itemname)
			err := delTagsForDelItem(v.Repname, v.Itemname)
			if err != nil {
				log.Error(err)
				return
			}

			err = delItem(v.Repname, v.Itemname)
			if err != nil {
				log.Error(err)
				return
			}
			log.Infof("Delete data item %v/%v according to heartbeat successfully.\n", v.Repname, v.Itemname)
		}
	} else {
		log.Warn("Unmarshal error:", err)
	}
}

func publishedTagOfItemHandler(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	log.Debug(r.URL.Path, "tags published of dataitem")
	r.ParseForm()

	dpname := ps.ByName("dpname")
	repo := ps.ByName("repo")
	item := ps.ByName("item")

	count, err := getPublishedTagCount(dpname, repo, item)
	if err != nil {
		log.Debug(err)
		JsonResult(w, http.StatusInternalServerError, cmd.InternalError, err.Error(), nil)
		return
	}
	offset, limit := optionalOffsetAndSize(r, 10, 1, 100)
	log.Debug("offset, limit", offset, limit)
	validateOffsetAndLimit(count, &offset, &limit)

	publishedTagsOfItem, err := GetPublishedTagsOfItemInfo(dpname, repo, item, offset, limit)
	if err != nil {
		log.Debug(err)
		JsonResult(w, http.StatusInternalServerError, cmd.InternalError, err.Error(), nil)
		return
	}

	if len(publishedTagsOfItem) == 0 {
		msg := fmt.Sprintf("Published tags of %s/%s is empty.", repo, item)
		JsonResult(w, http.StatusOK, cmd.ErrorPulledTagEmpty, msg, nil)
	} else {
		msg := fmt.Sprintf("All tags have been published of %s/%s", repo, item)
		JsonResult(w, http.StatusOK, cmd.ResultOK, msg, newQueryListResult(count, &publishedTagsOfItem))
	}
}

func optionalIntParamInQuery(r *http.Request, paramName string, defaultInt int64) int64 {
	if r.Form.Get(paramName) == "" {
		log.Debug("paramName nil", paramName, r.Form)
		return defaultInt
	}

	i, err := strconv.ParseInt(r.Form.Get(paramName), 10, 64)
	if err != nil {
		log.Debug("ParseInt", err)
		return defaultInt
	} else {
		return i
	}
}

func getRepoCountByDp(datapool, status string) int64 {
	if status == "published" {
		status = "Y"
	} else {
		status = "N"
	}

	sql := fmt.Sprintf(`SELECT COUNT(DISTINCT REPOSITORY) 
		FROM DH_DP_RPDM_MAP 
		WHERE DPID IN
		(SELECT DPID FROM DH_DP WHERE DPNAME = '%s' AND STATUS='A')
		AND PUBLISH= '%s' 
		AND STATUS = 'A';`, datapool, status)

	row, err := g_ds.QueryRow(sql)
	if err != nil {
		l := log.Error(err)
		logq.LogPutqueue(l)
		return 0
	}

	var count int64
	row.Scan(&count)
	log.Debug("Published repository count:", count)
	return count
}

func getPulledTagCount(datapool, repo, item string) (int64, error) {

	sql := fmt.Sprintf(`SELECT COUNT(*)
		FROM DH_RPDM_TAG_MAP
		WHERE RPDMID = (SELECT RPDMID FROM DH_DP_RPDM_MAP
					WHERE REPOSITORY  = '%s'
					AND DATAITEM = '%s'

					AND PUBLISH = 'N'
					AND STATUS = 'A'
					AND DPID = (SELECT DPID FROM DH_DP WHERE DPNAME = '%s' AND STATUS='A'))
		AND STATUS = 'A';`, repo, item, datapool)

	row, err := g_ds.QueryRow(sql)
	if err != nil {
		l := log.Error(err)
		logq.LogPutqueue(l)
		return 0, err
	}

	var count int64
	row.Scan(&count)
	log.Debug("Published repository count:", count)
	return count, err
}

func GetDaemonRoleByPubRecord() (role int) {
	sql := `SELECT COUNT(*) FROM DH_DP_RPDM_MAP WHERE PUBLISH='Y' AND STATUS='A' 
	        AND DPID IN (SELECT DPID FROM DH_DP WHERE STATUS='A');`
	row := g_ds.Db.QueryRow(sql)

	var count int
	row.Scan(&count)
	if count > 0 {
		role = PUBLISHER
		log.Debug("This datahub daemon is a publisher.")
	} else {
		role = PULLER
		log.Debug("This datahub daemon is a puller.")
	}
	return
}

func delTagsForDelItem(reponame, itemname string) error {
	log.Println("Begin to remove tags for remove item from db")
	sqlrpdmid := fmt.Sprintf(`SELECT RPDMID FROM DH_DP_RPDM_MAP WHERE REPOSITORY='%s' AND DATAITEM='%s' AND STATUS='A';`, reponame, itemname)

	row, err := g_ds.QueryRow(sqlrpdmid)
	if err != nil {
		l := log.Error("select rpdmid from DH_DP_RPDM_MAP error:", err)
		logq.LogPutqueue(l)
		return err
	}
	var rpdmId int
	row.Scan(&rpdmId)
	if rpdmId == 0 {
		log.Debug(reponame, itemname, "not exist.")
		return nil
	}
	sqldeltag := fmt.Sprintf(`UPDATE DH_RPDM_TAG_MAP SET STATUS='N' WHERE RPDMID=%d`, rpdmId)
	_, err = g_ds.Update(sqldeltag)
	log.Info("sqldeltag", sqldeltag)
	if err != nil {
		l := log.Error("delete tag error:", err)
		logq.LogPutqueue(l)
		return err
	}

	return nil
}

func CheckHealthClock() {
	log.Debug("--------->BEGIN")

	checkHealth(&Errortagsmap)

	timer := time.NewTicker(10 * time.Minute)
	for {
		select {
		case <-timer.C:
			now := time.Now()
			if now.Hour()%6 == 0 {
				log.Info("Time:", now)
				checkHealth(&Errortagsmap)
			}
		}
	}
	log.Debug("---------->END")
}

func removeAllJobDB() (e error) {
	log.Debug("TODO remove all jobs from db")
	sRmJobs := `DELETE FROM DH_JOB;`
	_, e = g_ds.Delete(sRmJobs)
	if e != nil {
		l := log.Error(e)
		logq.LogPutqueue(l)
	}
	return
}

func removeJobDB(job *ds.JobInfo) (e error) {
	log.Debug("TODO remove jobid from db")
	sRmJob := fmt.Sprintf(`DELETE FROM DH_JOB WHERE JOBID=%d;`, job.ID)
	_, e = g_ds.Delete(sRmJob)
	if e != nil {
		l := log.Error(e)
		logq.LogPutqueue(l)
	}
	return
}

func GetItemslocationInDatapool(itemslocation map[string]string, dpname string, dpid int, dpconn string) error {

	sql := fmt.Sprintf("SELECT DISTINCT ITEMDESC, REPOSITORY, DATAITEM FROM DH_DP_RPDM_MAP WHERE DPID=%v AND STATUS='A';", dpid)
	log.Debug(sql)
	rows, err := g_ds.QueryRows(sql)
	if err != nil {
		l := log.Errorf("datapool name %s, dpid %v, dpconn %v, error:%v", dpname, dpid, dpconn, err)
		logq.LogPutqueue(l)
		return err
	}

	var location, repo, item string
	for rows.Next() {
		rows.Scan(&location, &repo, &item)
		log.Debug(location, repo, item)
		itemslocation[location] = repo + "/" + item
	}
	log.Trace(itemslocation)
	return err
}

func updateJobStatus(job *ds.JobInfo) (e error) {
	log.Debug("TODO updata job stat to db.")
	sUpdateJob := fmt.Sprintf(`UPDATE DH_JOB SET STATUS='%s', STAT_TIME=datetime('now'), DOWNSIZE=%d
		WHERE JOBID='%s';`, job.Stat, job.Dlsize, job.ID)
	_, e = g_ds.Update(sUpdateJob)
	if e != nil {
		l := log.Error(e)
		logq.LogPutqueue(l)
	}
	return
}

func New(name string) (*Datapool, error) {
	datapooldriver, ok := datapooldrivers[name]
	for k, _ := range datapooldrivers {
		log.Debug(k, datapooldrivers[k], reflect.TypeOf(datapooldrivers[k]))
	}
	if !ok {
		s := fmt.Sprintf("Can't find datapooldriver %v", name)
		log.Error(s)
		return nil, errors.New(s)
	}
	return &Datapool{driver: datapooldriver}, nil
}

func saveJobDB(job *ds.JobInfo) (e error) {
	log.Debug("TODO save job info to db.")
	sInsertJob := fmt.Sprintf(`INSERT INTO DH_JOB (JOBID, TAG, FILEPATH, STATUS, CREATE_TIME, STAT_TIME, DOWNSIZE, SRCSIZE)
		VALUES ('%s','%s','%s','%s', datetime('now'), datetime('now'),%d, %d);`,
		job.ID, job.Tag, job.Path, job.Stat, job.Dlsize, job.Srcsize)
	_, e = g_ds.Insert(sInsertJob)
	if e != nil {
		l := log.Error(e)
		logq.LogPutqueue(l)
	}
	return
}

func GetPricePlan(path string) (plans []PricePlan) {
	config := path + "/" + PriceFile
	if isFileExists(config) == true {
		bytes, err := ioutil.ReadFile(config)
		if err != nil {
			log.Error(err)
			return
		}
		log.Debug(string(bytes))
		type LPrices struct {
			PricePlans []PricePlan `json:"price,omitempty"`
		}
		struPrices := LPrices{}
		if err = json.Unmarshal(bytes, &struPrices); err != nil {
			log.Error(err)
			return
		}
		log.Debug(struPrices)
		plans = struPrices.PricePlans
	}
	return
}

func itemPulledHandler(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
	log.Debug(r.URL.Path, "item pulled or not")
	repo := ps.ByName("repo")
	item := ps.ByName("item")

	itemInfo := ItemInDatapool{}
	itemInfo.Dpname, itemInfo.Dpconn, itemInfo.Dptype, itemInfo.ItemLocation = GetDpnameDpconnItemdesc(repo, item)

	if len(itemInfo.ItemLocation) == 0 {
		JsonResult(w, http.StatusOK, cmd.ErrorItemNotExist, "The DataItem hasn't been pulled.", nil)
	} else {
		JsonResult(w, http.StatusOK, cmd.ResultOK, "The DataItem has been pulled.", &itemInfo)
	}
}