|
|
@@ -3,6 +3,8 @@ package importer
|
|
|
import (
|
|
|
"encoding/json"
|
|
|
"fmt"
|
|
|
+ "regexp"
|
|
|
+ "strings"
|
|
|
"sync"
|
|
|
"sync/atomic"
|
|
|
"time"
|
|
|
@@ -151,7 +153,7 @@ func (odbci *ODBCImporter) masterlevel1data(classaliasname string, suid string,
|
|
|
level1data = entiredata
|
|
|
}
|
|
|
// 重新插入完整的 level1
|
|
|
- retrycount, e := odbci.insertData("level1", "", "", level1data)
|
|
|
+ retrycount, _, e := odbci.insertData("level1", "", "", level1data)
|
|
|
if e != nil {
|
|
|
return retrycount, e
|
|
|
}
|
|
|
@@ -175,7 +177,7 @@ func (odbci *ODBCImporter) masterlevel1data(classaliasname string, suid string,
|
|
|
data = entiredata
|
|
|
}
|
|
|
// 插入 level1 数据
|
|
|
- retrycount, e := odbci.insertData("level1", "", "", data)
|
|
|
+ retrycount, _, e := odbci.insertData("level1", "", "", data)
|
|
|
if e != nil {
|
|
|
return retrycount, e
|
|
|
}
|
|
|
@@ -320,7 +322,7 @@ func (odbci *ODBCImporter) InsertData(classaliasname string, data map[string]any
|
|
|
} else {
|
|
|
data["depend"] = referencedata(classaliasname, data)
|
|
|
}
|
|
|
- rc, e := odbci.insertData(classaliasname, oid, suid, data)
|
|
|
+ rc, _, e := odbci.insertData(classaliasname, oid, suid, data)
|
|
|
retrycount += rc
|
|
|
if e != nil {
|
|
|
return retrycount, e
|
|
|
@@ -382,13 +384,13 @@ func referencedata(classaliasname string, data map[string]any) (depend map[strin
|
|
|
return
|
|
|
}
|
|
|
|
|
|
-func (odbci *ODBCImporter) insertData(classaliasname string, oid, suid string, data map[string]any) (retrycount int, err error) {
|
|
|
+func (odbci *ODBCImporter) insertData(classaliasname string, oid, suid string, data map[string]any) (retrycount int, responsetime time.Duration, err error) {
|
|
|
cdi := classdatainfos.GetIFPresent(classaliasname)
|
|
|
if cdi == nil {
|
|
|
- return retrycount, merrs.NewError("class not defined " + classaliasname)
|
|
|
+ return retrycount, 0, merrs.NewError("class not defined " + classaliasname)
|
|
|
}
|
|
|
if cdi.Insertmql == "" {
|
|
|
- return retrycount, merrs.NewError("class no fields to insert " + classaliasname)
|
|
|
+ return retrycount, 0, merrs.NewError("class no fields to insert " + classaliasname)
|
|
|
}
|
|
|
values := []any{}
|
|
|
for _, fn := range cdi.Fieldslist {
|
|
|
@@ -400,7 +402,7 @@ func (odbci *ODBCImporter) insertData(classaliasname string, oid, suid string, d
|
|
|
// 合并扩展字段
|
|
|
if strset.New(fi.Datakey...).Has("*") {
|
|
|
if fi.Fieldtype != "map<varchar,varchar>" {
|
|
|
- return retrycount, merrs.NewError("fi.Fieldtype=" + fi.Fieldtype + " != map<varchar,varchar>")
|
|
|
+ return retrycount, 0, merrs.NewError("fi.Fieldtype=" + fi.Fieldtype + " != map<varchar,varchar>")
|
|
|
}
|
|
|
td := map[string]any{}
|
|
|
for k, v := range data {
|
|
|
@@ -422,7 +424,7 @@ func (odbci *ODBCImporter) insertData(classaliasname string, oid, suid string, d
|
|
|
case "timestamp":
|
|
|
tv, e := cast.ToDateTimeE(v, "2006-01-02-15.04.05.000000")
|
|
|
if e != nil {
|
|
|
- return retrycount, merrs.NewError(fmt.Sprint("can't parse datetime value '", v, "'"))
|
|
|
+ return retrycount, 0, merrs.NewError(fmt.Sprint("can't parse datetime value '", v, "'"))
|
|
|
}
|
|
|
v = tv.Format("2006-01-02 15:04:05.000000")
|
|
|
}
|
|
|
@@ -439,14 +441,14 @@ func (odbci *ODBCImporter) insertData(classaliasname string, oid, suid string, d
|
|
|
mql := "select id,uniqueid from " + classaliasname + " where id=?"
|
|
|
r, e := odbci.client.Query(mql, oid).Do()
|
|
|
if e != nil {
|
|
|
- return retrycount, e
|
|
|
+ return retrycount, 0, e
|
|
|
}
|
|
|
if r != nil && len(r.Data) != 0 {
|
|
|
logger.Debug(classaliasname, "exists id:", oid, ", uniqueid:", r.Data[0]["uniqueid"], ", new uniqueid:", suid)
|
|
|
}
|
|
|
}
|
|
|
// logger.Info(values...)
|
|
|
- retrycount, err = odbci.insertDo(cdi.Insertmql, values...)
|
|
|
+ retrycount, responsetime, err = odbci.insertDo(cdi.Insertmql, values...)
|
|
|
if err != nil {
|
|
|
databs, _ := json.MarshalIndent(data, "", " ")
|
|
|
err = merrs.NewError(err, merrs.SSMaps{{"mql": cdi.Insertmql}, {"values": fmt.Sprint(values)}, {"data": string(databs)}})
|
|
|
@@ -465,28 +467,78 @@ func (odbci *ODBCImporter) insertData(classaliasname string, oid, suid string, d
|
|
|
return
|
|
|
}
|
|
|
|
|
|
-var insertretrylimitcount = 0
|
|
|
+type ODBCRetryConfig struct {
|
|
|
+ retry int
|
|
|
+ contains string
|
|
|
+}
|
|
|
+
|
|
|
+var reodbcretry = regexp.MustCompile(`(?s)^\s*(-?[0-9]+)\s*(?:,\s*(.*)\s*)?$`)
|
|
|
+var odbcretry = ""
|
|
|
+var odbcretryconfig []*ODBCRetryConfig
|
|
|
|
|
|
func init() {
|
|
|
mcfg.OnChange(func() {
|
|
|
- insertretrylimitcount = mcfg.GetInt("odbc.insert.retry", 3)
|
|
|
+ _odbcretry := mcfg.GetStrings("odbc.retry", "-1, timed out")
|
|
|
+ if strings.Join(_odbcretry, "|") != odbcretry {
|
|
|
+ odbcretryconfig = RetryConfig(_odbcretry...)
|
|
|
+ odbcretry = strings.Join(_odbcretry, "|")
|
|
|
+ }
|
|
|
})
|
|
|
}
|
|
|
|
|
|
-func (odbci *ODBCImporter) insertDo(insertmql string, values ...any) (trycount int, err error) {
|
|
|
+func RetryConfig(retryconfig ...string) (orcs []*ODBCRetryConfig) {
|
|
|
+ defaultorc := &ODBCRetryConfig{
|
|
|
+ retry: 0,
|
|
|
+ contains: "",
|
|
|
+ }
|
|
|
+ for _, retrycfg := range retryconfig {
|
|
|
+ ss := reodbcretry.FindStringSubmatch(retrycfg)
|
|
|
+ if len(ss) > 1 {
|
|
|
+ orc := &ODBCRetryConfig{
|
|
|
+ retry: cast.ToInt(ss[0]),
|
|
|
+ contains: ss[1],
|
|
|
+ }
|
|
|
+ if orc.contains == "" {
|
|
|
+ defaultorc = orc
|
|
|
+ } else {
|
|
|
+ orcs = append(orcs, orc)
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ panic("odbc.retry config format error")
|
|
|
+ }
|
|
|
+ }
|
|
|
+ orcs = append(orcs, defaultorc)
|
|
|
+ return
|
|
|
+}
|
|
|
+
|
|
|
+func (odbci *ODBCImporter) insertDo(insertmql string, values ...any) (trycount int, responsetime time.Duration, err error) {
|
|
|
for {
|
|
|
+ st := time.Now()
|
|
|
_, e := odbci.client.Query(insertmql, values...).Do()
|
|
|
if e != nil {
|
|
|
+ maxtrycount := 0
|
|
|
+ for _, orc := range odbcretryconfig {
|
|
|
+ if orc.contains != "" {
|
|
|
+ if strings.Contains(e.Error(), orc.contains) {
|
|
|
+ maxtrycount = orc.retry
|
|
|
+ break
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ maxtrycount = orc.retry
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
trycount++
|
|
|
e = merrs.New(e, merrs.Map{"trycount": trycount})
|
|
|
- if trycount <= insertretrylimitcount {
|
|
|
- logger.Debug(e)
|
|
|
+ if maxtrycount < 0 || trycount <= maxtrycount {
|
|
|
+ logger.Debug(merrs.New(e, merrs.Map{"retrycount": trycount}))
|
|
|
time.Sleep(time.Duration(trycount) * time.Second)
|
|
|
continue
|
|
|
}
|
|
|
- return trycount, e
|
|
|
+ return trycount, 0, e
|
|
|
}
|
|
|
- return trycount, nil
|
|
|
+ responsetime = time.Since(st)
|
|
|
+ return trycount, responsetime, nil
|
|
|
}
|
|
|
}
|
|
|
|
