Skip to content

Commit

Permalink
fix: 整理输出日志
Browse files Browse the repository at this point in the history
  • Loading branch information
王荣昌 committed Apr 7, 2024
1 parent c27e9d2 commit 233db23
Show file tree
Hide file tree
Showing 13 changed files with 44 additions and 44 deletions.
2 changes: 1 addition & 1 deletion config/kis_func_config.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ func NewFuncConfig(
Name: "unNamedSource",
}
source = &defaultSource
log.Logger().InfoF("funcName NewConfig source is nil, funcName = %s, use default unNamed Source.", funcName)
log.Logger().InfoF("funcName NewConfig source is nil, use default unNamed Source.", "funcName", funcName)
}
config.Source = *source

Expand Down
8 changes: 4 additions & 4 deletions flow/kis_flow.go
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,8 @@ func (flow *KisFlow) Fork(ctx context.Context) kis.Flow {
}
}

log.Logger().DebugFX(ctx, "=====>Flow Fork, oldFlow.funcParams = %+v\n", flow.funcParams)
log.Logger().DebugFX(ctx, "=====>Flow Fork, newFlow.funcParams = %+v\n", newFlow.GetFuncParamsAllFuncs())
log.Logger().DebugFX(ctx, "=====>Flow Fork, ", "oldFlow.funcParams", flow.funcParams)
log.Logger().DebugFX(ctx, "=====>Flow Fork, ", "newFlow.funcParams", newFlow.GetFuncParamsAllFuncs())

return newFlow
}
Expand Down Expand Up @@ -257,7 +257,7 @@ func (flow *KisFlow) Run(ctx context.Context) error {

// 得到当前Function要处理与的源数据
if inputData, err := flow.getCurData(); err != nil {
log.Logger().ErrorFX(ctx, "flow.Run(): getCurData err = %s\n", err.Error())
log.Logger().ErrorFX(ctx, "flow.Run(): getCurData", "err", err.Error())
return err
} else {
flow.inPut = inputData
Expand Down Expand Up @@ -346,7 +346,7 @@ func (flow *KisFlow) GetFuncConfigByName(funcName string) *config.KisFuncConfig
if f, ok := flow.Funcs[funcName]; ok {
return f.GetConfig()
} else {
log.Logger().ErrorF("GetFuncConfigByName(): Function %s not found", funcName)
log.Logger().ErrorF("GetFuncConfigByName(): Function not found", "FunctionName", funcName)
return nil
}
}
12 changes: 6 additions & 6 deletions flow/kis_flow_data.go
Original file line number Diff line number Diff line change
Expand Up @@ -69,11 +69,11 @@ func (flow *KisFlow) commitSrcData(ctx context.Context) error {
// 统计数据总量 Metrics.DataTota 指标累计加1
metrics.Metrics.DataTotal.Add(float64(dataCnt))

//统计当前Flow数量指标
// 统计当前Flow数量指标
metrics.Metrics.FlowDataTotal.WithLabelValues(flow.Name).Add(float64(dataCnt))
}

log.Logger().DebugFX(ctx, "====> After CommitSrcData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)
log.Logger().DebugFX(ctx, "====> After CommitSrcData", "flow_name", flow.Name, "flow_id", flow.Id, "All Level Data", flow.data)

return nil
}
Expand Down Expand Up @@ -106,7 +106,7 @@ func (flow *KisFlow) commitReuseData(ctx context.Context) error {
// 清空缓冲Buf (如果是ReuseData选项,那么提交的全部数据,都将不会携带到下一层)
flow.buffer = flow.buffer[0:0]

log.Logger().DebugFX(ctx, " ====> After commitReuseData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)
log.Logger().DebugFX(ctx, " ====> After commitReuseData", "flow_name", flow.Name, "flow_id", flow.Id, "All Level Data", flow.data)

return nil
}
Expand All @@ -122,12 +122,12 @@ func (flow *KisFlow) commitVoidData(ctx context.Context) error {
// 将本层计算的缓冲数据提交到本层结果数据中
flow.data[flow.ThisFunctionId] = batch

log.Logger().DebugFX(ctx, " ====> After commitVoidData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)
log.Logger().DebugFX(ctx, " ====> After commitVoidData", "flow_name", flow.Name, "flow_id", flow.Id, "All Level Data", flow.data)

return nil
}

//commitCurData 提交Flow当前执行Function的结果数据
// commitCurData 提交Flow当前执行Function的结果数据
func (flow *KisFlow) commitCurData(ctx context.Context) error {

// 判断本层计算是否有结果数据,如果没有则退出本次Flow Run循环
Expand All @@ -150,7 +150,7 @@ func (flow *KisFlow) commitCurData(ctx context.Context) error {
// 清空缓冲Buf
flow.buffer = flow.buffer[0:0]

log.Logger().DebugFX(ctx, " ====> After commitCurData, flow_name = %s, flow_id = %s\nAll Level Data =\n %+v\n", flow.Name, flow.Id, flow.data)
log.Logger().DebugFX(ctx, " ====> After commitCurData", "flow_name", flow.Name, "flow_id", flow.Id, "All Level Data", flow.data)

return nil
}
Expand Down
4 changes: 2 additions & 2 deletions function/kis_function_c.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@ func NewKisFunctionC() kis.Function {
}

func (f *KisFunctionC) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().DebugF("KisFunctionC, flow = %+v\n", flow)
log.Logger().DebugF("KisFunctionC", "flow", flow)

// 通过KisPool 路由到具体的执行计算Function中
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, "Function Called Error err = %s\n", err)
log.Logger().ErrorFX(ctx, "Function Called Error", "err", err)
return err
}

Expand Down
4 changes: 2 additions & 2 deletions function/kis_function_e.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@ func NewKisFunctionE() kis.Function {
}

func (f *KisFunctionE) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().DebugF("KisFunctionE, flow = %+v\n", flow)
log.Logger().DebugF("KisFunctionE", "flow", flow)

// 通过KisPool 路由到具体的执行计算Function中
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, "Function Called Error err = %s\n", err)
log.Logger().ErrorFX(ctx, "Function Called Error", "err", err)
return err
}

Expand Down
4 changes: 2 additions & 2 deletions function/kis_function_l.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@ func NewKisFunctionL() kis.Function {
}

func (f *KisFunctionL) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().DebugF("KisFunctionL, flow = %+v\n", flow)
log.Logger().DebugF("KisFunctionL", "flow", flow)

// 通过KisPool 路由到具体的执行计算Function中
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, "Function Called Error err = %s\n", err)
log.Logger().ErrorFX(ctx, "Function Called Error", "err", err)
return err
}

Expand Down
4 changes: 2 additions & 2 deletions function/kis_function_s.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@ func NewKisFunctionS() kis.Function {
}

func (f *KisFunctionS) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().DebugF("KisFunctionS, flow = %+v\n", flow)
log.Logger().DebugF("KisFunctionS", "flow", flow)

// 通过KisPool 路由到具体的执行计算Function中
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, "Function Called Error err = %s\n", err)
log.Logger().ErrorFX(ctx, "Function Called Error", "err", err)
return err
}

Expand Down
4 changes: 2 additions & 2 deletions function/kis_function_v.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,11 @@ func NewKisFunctionV() kis.Function {
}

func (f *KisFunctionV) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().DebugF("KisFunctionV, flow = %+v\n", flow)
log.Logger().DebugF("KisFunctionV", "flow", flow)

// 通过KisPool 路由到具体的执行计算Function中
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, "Function Called Error err = %s\n", err)
log.Logger().ErrorFX(ctx, "Function Called Error", "err", err)
return err
}

Expand Down
20 changes: 10 additions & 10 deletions kis/pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ var _pool *kisPool
// Pool 单例构造
func Pool() *kisPool {
_poolOnce.Do(func() {
//创建kisPool对象
// 创建kisPool对象
_pool = new(kisPool)

// fnRouter初始化
Expand Down Expand Up @@ -61,7 +61,7 @@ func (pool *kisPool) AddFlow(name string, flow Flow) {
panic(errString)
}

log.Logger().InfoF("Add FlowRouter FlowName=%s", name)
log.Logger().InfoF("Add FlowRouter", "FlowName", name)
}

func (pool *kisPool) GetFlow(name string) Flow {
Expand Down Expand Up @@ -95,7 +95,7 @@ func (pool *kisPool) FaaS(fnName string, f FaaS) {
panic(errString)
}

log.Logger().InfoF("Add KisPool FuncName=%s", fnName)
log.Logger().InfoF("Add KisPool", "FuncName", fnName)
}

// CallFunction 调度 Function
Expand Down Expand Up @@ -127,7 +127,7 @@ func (pool *kisPool) CallFunction(ctx context.Context, fnName string, flow Flow)
// 将flow.Input()中的原始数据,反序列化为argType类型的数据
value, err := funcDesc.Serialize.UnMarshal(flow.Input(), argType)
if err != nil {
log.Logger().ErrorFX(ctx, "funcDesc.Serialize.DecodeParam err=%v", err)
log.Logger().ErrorFX(ctx, "funcDesc.Serialize.DecodeParam", "err", err)
} else {
params = append(params, value)
continue
Expand All @@ -153,7 +153,7 @@ func (pool *kisPool) CallFunction(ctx context.Context, fnName string, flow Flow)

}

log.Logger().ErrorFX(ctx, "FuncName: %s Can not find in KisPool, Not Added.\n", fnName)
log.Logger().ErrorFX(ctx, "FuncName: Can not find in KisPool, Not Added.", "FuncName", fnName)

return errors.New("FuncName: " + fnName + " Can not find in NsPool, Not Added.")
}
Expand All @@ -170,7 +170,7 @@ func (pool *kisPool) CaaSInit(cname string, c ConnInit) {
panic(errString)
}

log.Logger().InfoF("Add KisPool CaaSInit CName=%s", cname)
log.Logger().InfoF("Add KisPool CaaSInit", "CName", cname)
}

// CallConnInit 调度 ConnInit
Expand All @@ -193,10 +193,10 @@ func (pool *kisPool) CaaS(cname string, fname string, mode common.KisMode, c Caa
defer pool.cLock.Unlock()

if _, ok := pool.cTree[cname]; !ok {
//cid 首次注册,不存在,创建二级树NsConnSL
// cid 首次注册,不存在,创建二级树NsConnSL
pool.cTree[cname] = make(connSL)

//初始化各类型FunctionMode
// 初始化各类型FunctionMode
pool.cTree[cname][common.S] = make(connFuncRouter)
pool.cTree[cname][common.L] = make(connFuncRouter)
}
Expand All @@ -208,7 +208,7 @@ func (pool *kisPool) CaaS(cname string, fname string, mode common.KisMode, c Caa
panic(errString)
}

log.Logger().InfoF("Add KisPool CaaS CName=%s, FName=%s, Mode =%s", cname, fname, mode)
log.Logger().InfoF("Add KisPool CaaS", "CName", cname, "FName", fname, "Mode", mode)
}

// CallConnector 调度 Connector
Expand All @@ -223,7 +223,7 @@ func (pool *kisPool) CallConnector(ctx context.Context, flow Flow, conn Connecto
return callback(ctx, conn, fn, flow, args)
}

log.Logger().ErrorFX(ctx, "CName:%s FName:%s mode:%s Can not find in KisPool, Not Added.\n", conn.GetName(), fnConf.FName, mode)
log.Logger().ErrorFX(ctx, "Can not find in KisPool, Not Added.", "CName", conn.GetName(), " FName", fnConf.FName, "mode", mode)

return nil, errors.New(fmt.Sprintf("CName:%s FName:%s mode:%s Can not find in KisPool, Not Added.", conn.GetName(), fnConf.FName, mode))
}
Expand Down
12 changes: 6 additions & 6 deletions metrics/kis_metrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,9 @@ func RunMetricsService(serverAddr string) error {
http.Handle(common.METRICS_ROUTE, promhttp.Handler())

// 启动HttpServer
err := http.ListenAndServe(serverAddr, nil) //多个进程不可监听同一个端口
err := http.ListenAndServe(serverAddr, nil) // 多个进程不可监听同一个端口
if err != nil {
log.Logger().ErrorF("RunMetricsService err = %s\n", err)
log.Logger().ErrorF("RunMetricsService", "err", err)
}

return err
Expand Down Expand Up @@ -67,7 +67,7 @@ func InitMetrics() {
Name: common.GANGE_FLOW_SCHE_CNTS_NAME,
Help: common.GANGE_FLOW_SCHE_CNTS_HELP,
},
//标签名称
// 标签名称
[]string{common.LABEL_FLOW_NAME},
)

Expand All @@ -77,15 +77,15 @@ func InitMetrics() {
Name: common.GANGE_FUNC_SCHE_CNTS_NAME,
Help: common.GANGE_FUNC_SCHE_CNTS_HELP,
},
//标签名称
// 标签名称
[]string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE},
)

// FunctionDuration初始化HistogramVec
Metrics.FunctionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
Name: common.HISTOGRAM_FUNCTION_DURATION_NAME,
Help: common.HISTOGRAM_FUNCTION_DURATION_HELP,
Buckets: []float64{0.005, 0.01, 0.03, 0.08, 0.1, 0.5, 1.0, 5.0, 10, 100, 1000, 5000, 30000}, //单位ms,最大半分钟
Buckets: []float64{0.005, 0.01, 0.03, 0.08, 0.1, 0.5, 1.0, 5.0, 10, 100, 1000, 5000, 30000}, // 单位ms,最大半分钟
},
[]string{common.LABEL_FUNCTION_NAME, common.LABEL_FUNCTION_MODE},
)
Expand All @@ -95,7 +95,7 @@ func InitMetrics() {
prometheus.HistogramOpts{
Name: common.HISTOGRAM_FLOW_DURATION_NAME,
Help: common.HISTOGRAM_FLOW_DURATION_HELP,
Buckets: []float64{0.005, 0.01, 0.03, 0.08, 0.1, 0.5, 1.0, 5.0, 10, 100, 1000, 5000, 30000, 60000}, //单位ms,最大1分钟
Buckets: []float64{0.005, 0.01, 0.03, 0.08, 0.1, 0.5, 1.0, 5.0, 10, 100, 1000, 5000, 30000, 60000}, // 单位ms,最大1分钟
},
[]string{common.LABEL_FLOW_NAME},
)
Expand Down
4 changes: 2 additions & 2 deletions test/faas/faas_demo2.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,12 @@ func FuncDemo2Handler(ctx context.Context, flow kis.Flow) error {

conn, err := flow.GetConnector()
if err != nil {
log.Logger().ErrorFX(ctx, "FuncDemo2Handler(): GetConnector err = %s\n", err.Error())
log.Logger().ErrorFX(ctx, "FuncDemo2Handler(): GetConnector", "err", err.Error())
return err
}

if _, err := conn.Call(ctx, flow, row); err != nil {
log.Logger().ErrorFX(ctx, "FuncDemo2Handler(): Call err = %s\n", err.Error())
log.Logger().ErrorFX(ctx, "FuncDemo2Handler(): Call", "err", err.Error())
return err
}

Expand Down
8 changes: 4 additions & 4 deletions test/kis_config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ func TestNewFuncConfig(t *testing.T) {

myFunc1 := config.NewFuncConfig("funcName1", common.S, &source, &option)

log.Logger().InfoF("funcName1: %+v\n", myFunc1)
log.Logger().InfoF("funcName1", myFunc1)
}

func TestNewFlowConfig(t *testing.T) {
Expand All @@ -50,7 +50,7 @@ func TestNewFlowConfig(t *testing.T) {
myFlow1.AppendFunctionConfig(flowFuncParams1)
myFlow1.AppendFunctionConfig(flowFuncParams2)

log.Logger().InfoF("myFlow1: %+v\n", myFlow1)
log.Logger().InfoF("myFlow1", myFlow1)
}

func TestNewConnConfig(t *testing.T) {
Expand Down Expand Up @@ -81,8 +81,8 @@ func TestNewConnConfig(t *testing.T) {
myConnector1 := config.NewConnConfig("connectorName1", "0.0.0.0:9987,0.0.0.0:9997", common.REDIS, "key", connParams)

if err := myConnector1.WithFunc(myFunc1); err != nil {
log.Logger().ErrorF("WithFunc err: %s\n", err.Error())
log.Logger().ErrorF("WithFunc", "err", err.Error())
}

log.Logger().InfoF("myConnector1: %+v\n", myConnector1)
log.Logger().InfoF("myConnector1", myConnector1)
}
2 changes: 1 addition & 1 deletion test/kis_flow_commit_batch_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ func TestForkFlowCommitBatch(t *testing.T) {

// 3. 提交原始数据
if err := flow1.CommitRowBatch(stringRows); err != nil {
log.Logger().ErrorF("CommitRowBatch Error, err = %+v", err)
log.Logger().ErrorF("CommitRowBatch Error", "err", err)
panic(err)
}

Expand Down

0 comments on commit 233db23

Please sign in to comment.