diff --git a/bcs-services/bcs-bkcmdb-synchronizer/cmd/synchronizer/main.go b/bcs-services/bcs-bkcmdb-synchronizer/cmd/synchronizer/main.go index d14a7cc521..3f08f7c110 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/cmd/synchronizer/main.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/cmd/synchronizer/main.go @@ -63,6 +63,16 @@ func init() { blog.Warnf("parse custom resource types from env failed: %s, using default", err) } + // Parse and set sql log level from environment variable + if err := common.SetSqlLogLevelFromEnv(BkcmdbSynchronizerOption); err != nil { + blog.Warnf("parse sql log level from env failed: %s, using default", err) + } + + // Parse and set clean local cache from environment variable + if err := common.SetCleanLocalCacheFromEnv(BkcmdbSynchronizerOption); err != nil { + blog.Warnf("parse clean local cache from env failed: %s, using default", err) + } + if err := common.DecryptCMOption(BkcmdbSynchronizerOption); err != nil { blog.Fatalf("load config failed, err: %s", err.Error()) } diff --git a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client/cmdb/cmdb.go b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client/cmdb/cmdb.go index 21214be8a4..756fc0f82f 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client/cmdb/cmdb.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client/cmdb/cmdb.go @@ -30,9 +30,9 @@ import ( "github.com/parnurzeal/gorequest" "google.golang.org/grpc" "gorm.io/gorm" - "gorm.io/gorm/logger" "github.com/Tencent/bk-bcs/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client" + "github.com/Tencent/bk-bcs/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/common" "github.com/Tencent/bk-bcs/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/metrics" "github.com/Tencent/bk-bcs/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/store/model" ) @@ -261,16 +261,16 @@ func (c *cmdbClient) GetBS2IDByBizID(bizID int64) (int, error) { EndStruct(&respData) }) if len(errs) > 0 { - blog.Errorf("call api GetBS2IDByBizID failed: %v", errs[0]) + blog.Errorf("call api GetBS2IDByBizID failed: %v, BizID: %d", errs[0], bizID) return 0, errs[0] } if !respData.Result { - blog.Errorf("call api GetBS2IDByBizID failed: %v, rid: %s", respData.Message, respData.RequestID) + blog.Errorf("call api GetBS2IDByBizID failed: %v, rid: %s, BizID: %d", respData.Message, respData.RequestID, bizID) return 0, fmt.Errorf(respData.Message) } // successfully request - blog.Infof("call api GetBS2IDByBizID with url(%s) successfully", reqURL) + blog.Infof("call api GetBS2IDByBizID with url(%s) successfully, BizID: %d", reqURL, bizID) if len(respData.Data.Info) > 0 { return respData.Data.Info[0].BS2NameID, nil @@ -310,16 +310,16 @@ func (c *cmdbClient) GetBizInfo(bizID int64) (*client.Business, error) { EndStruct(&respData) }) if len(errs) > 0 { - blog.Errorf("call api GetBizInfo failed: %v", errs[0]) + blog.Errorf("call api GetBizInfo failed: %v, BizID: %d", errs[0], bizID) return nil, errs[0] } if !respData.Result { - blog.Errorf("call api GetBizInfo failed: %v, rid: %s", respData.Message, respData.RequestID) + blog.Errorf("call api GetBizInfo failed: %v, rid: %s, BizID: %d", respData.Message, respData.RequestID, bizID) return nil, fmt.Errorf(respData.Message) } // successfully request - blog.Infof("call api GetBizInfo with url(%s) successfully", reqURL) + blog.Infof("call api GetBizInfo with url(%s) successfully, BizID: %d", reqURL, bizID) if len(respData.Data.Data) > 0 { return &respData.Data.Data[0], nil @@ -401,17 +401,17 @@ func (c *cmdbClient) GetHostInfo(hostIP []string) (*[]client.HostData, error) { }) // 检查是否有错误发生 if len(errs) > 0 { - blog.Errorf("call api QueryHost failed: %v", errs[0]) + blog.Errorf("call api QueryHost failed: %v, HostIP: %v", errs[0], hostIP[from:to]) return nil, errs[0] } // 检查API响应是否成功 if !respData.Result { - blog.Errorf("call api QueryHost failed: %v, rid: %s", respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api QueryHost failed: %v, rid: %s, HostIP: %v", respData.Message, resp.Header.Get("X-Request-Id"), hostIP[from:to]) return nil, fmt.Errorf(respData.Message) } // 请求成功，记录日志 - blog.Infof("call api QueryHost with url(%s) successfully, X-Request-Id: %s", reqURL, resp.Header.Get("X-Request-Id")) + blog.Infof("call api QueryHost with url(%s) successfully, X-Request-Id: %s, HostIP: %v", reqURL, resp.Header.Get("X-Request-Id"), hostIP[from:to]) // 将响应数据添加到主机数据切片中 hostData = append(hostData, respData.Data.Info...) @@ -506,13 +506,13 @@ func (c *cmdbClient) GetHostsByBiz(bkBizID int64, hostIP []string) (*[]client.Ho // 检查响应结果是否成功 if !respData.Result { - blog.Errorf("call api QueryHost failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api QueryHost failed: %v, rid: %s, BkBizID: %d, HostIP: %v", + respData.Message, resp.Header.Get("X-Request-Id"), bkBizID, hostIP[from:to]) return nil, fmt.Errorf(respData.Message) } // 成功获取数据，记录日志 - blog.Infof("call api QueryHost with url(%s) successfully, X-Request-Id: %s", - reqURL, resp.Header.Get("X-Request-Id")) + blog.Infof("call api QueryHost with url(%s) successfully, X-Request-Id: %s, BkBizID: %d, HostIP: %v", + reqURL, resp.Header.Get("X-Request-Id"), bkBizID, hostIP[from:to]) // 将当前分页的数据添加到总数据列表中 hostData = append(hostData, respData.Data.Info...) @@ -592,17 +592,17 @@ func (c *cmdbClient) GetBcsCluster( }) // 检查是否有错误发生 if len(errs) > 0 { - blog.Errorf("call api list_kube_cluster failed: %v", errs[0]) + blog.Errorf("call api list_kube_cluster failed: %v, bkBizID: %d", errs[0], request.BKBizID) return nil, errs[0] } // 检查API响应是否成功 if !respData.Result { - blog.Errorf("call api list_kube_cluster failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api list_kube_cluster failed: %v, rid: %s, bkBizID: %d", + respData.Message, resp.Header.Get("X-Request-Id"), request.BKBizID) return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api list_kube_cluster with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api list_kube_cluster with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d", + reqURL, request, resp.Header.Get("X-Request-Id"), request.BKBizID) // 如果db不为空，则将API获取的集群信息同步到数据库 if db != nil { @@ -685,14 +685,14 @@ func (c *cmdbClient) CreateBcsCluster( // 检查响应结果是否成功 if !respData.Result { - blog.Errorf("call api create_kube_cluster failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api create_kube_cluster failed: %v, rid: %s, bkBizID: %d, name: %s", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.Name)) return bkClusterID, fmt.Errorf(respData.Message) } // 记录成功的日志信息 - blog.Infof("call api create_kube_cluster with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api create_kube_cluster with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, name: %s", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.Name)) // 获取创建的集群ID bkClusterID = respData.Data.ID @@ -700,7 +700,7 @@ func (c *cmdbClient) CreateBcsCluster( // 根据集群ID查询集群信息，验证创建是否成功 _, err = c.GetBcsCluster(&client.GetBcsClusterRequest{ CommonRequest: client.CommonRequest{ - BKBizID: *request.BKBizID, + BKBizID: common.Deref(request.BKBizID), Page: client.Page{ Limit: 100, Start: 0, @@ -769,19 +769,19 @@ func (c *cmdbClient) UpdateBcsCluster(request *client.UpdateBcsClusterRequest, d // 如果响应结果为false，记录错误日志并返回错误信息 if !respData.Result { - blog.Errorf("call api batch_update_kube_cluster failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_update_kube_cluster failed: %v, rid: %s, bkBizID: %d, ids: %v", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) return fmt.Errorf(respData.Message) } // 记录成功调用API的日志信息 - blog.Infof("call api batch_update_kube_cluster with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_update_kube_cluster with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, ids: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) // 调用GetBcsCluster方法获取更新后的集群信息 _, err := c.GetBcsCluster(&client.GetBcsClusterRequest{ CommonRequest: client.CommonRequest{ - BKBizID: *request.BKBizID, + BKBizID: common.Deref(request.BKBizID), Page: client.Page{ Limit: 100, Start: 0, @@ -852,14 +852,14 @@ func (c *cmdbClient) UpdateBcsClusterType(request *client.UpdateBcsClusterTypeRe // 检查响应数据中的Result字段，如果不为true则表示API调用失败 if !respData.Result { - blog.Errorf("call api update_kube_cluster_type failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api update_kube_cluster_type failed: %v, rid: %s, bkBizID: %d, id: %v, type: %v", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.ID), common.Deref(request.Type)) return fmt.Errorf(respData.Message) } // 记录API调用成功的日志，包括请求URL、请求体和请求ID - blog.Infof("call api update_kube_cluster_type with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api update_kube_cluster_type with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, id: %v, type: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.ID), common.Deref(request.Type)) // 调用GetBcsCluster函数获取更新后的集群信息 _, err := c.GetBcsCluster(&client.GetBcsClusterRequest{ @@ -875,7 +875,7 @@ func (c *cmdbClient) UpdateBcsClusterType(request *client.UpdateBcsClusterTypeRe { Field: "id", Operator: "in", - Value: []int64{*request.ID}, + Value: []int64{common.Deref(request.ID)}, }, }, }, @@ -929,14 +929,14 @@ func (c *cmdbClient) DeleteBcsCluster(request *client.DeleteBcsClusterRequest, d // 如果响应结果指示操作失败，则记录错误信息并返回错误 if !respData.Result { - blog.Errorf("call api batch_delete_kube_cluster failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_delete_kube_cluster failed: %v, rid: %s, bkBizID: %d, ids: %v", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) return fmt.Errorf(respData.Message) } // 如果操作成功，则记录成功的日志信息 - blog.Infof("call api batch_delete_kube_cluster with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_delete_kube_cluster with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, ids: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) // 如果提供了数据库连接，则尝试从数据库中删除对应的集群记录 if db != nil { @@ -963,7 +963,7 @@ func (c *cmdbClient) GetBcsNamespace( return nil, ErrServerNotInit } if withDB && db != nil { - query := db.Session(&gorm.Session{NewDB: true, Logger: db.Logger.LogMode(logger.Info)}) + query := db.Session(&gorm.Session{NewDB: true}) for _, rule := range request.Filter.Rules { if request.Filter.Condition == And { query = query.Where(fmt.Sprintf("%s %s ?", rule.Field, rule.Operator), rule.Value) @@ -973,7 +973,7 @@ func (c *cmdbClient) GetBcsNamespace( } } var ns []model.Namespace - if err := query.Debug().Find(&ns).Error; err != nil { + if err := query.Find(&ns).Error; err != nil { blog.Errorf("query namespace withDB failed: %v", err) } else { if namespaceMarshal, err := json.Marshal(ns); err != nil { @@ -1008,12 +1008,12 @@ func (c *cmdbClient) GetBcsNamespace( return nil, errs[0] } if !respData.Result { - blog.Errorf("call api list_namespace failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api list_namespace failed: %v, rid: %s, bkBizID: %d", + respData.Message, resp.Header.Get("X-Request-Id"), request.BKBizID) return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api list_namespace with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api list_namespace with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d", + reqURL, request, resp.Header.Get("X-Request-Id"), request.BKBizID) if db != nil { if namespaceMarshal, err := json.Marshal(respData.Data.Info); err != nil { blog.Errorf("marshal namespace failed: %v", err) @@ -1077,13 +1077,13 @@ func (c *cmdbClient) CreateBcsNamespace(request *client.CreateBcsNamespaceReques } if !respData.Result { - blog.Errorf("call api create_kube_namespace failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api create_kube_namespace failed: %v, rid: %s, bkBizID: %d", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID)) if respData.Code == 1199014 && db != nil { for _, d := range *(request.Data) { if _, err := c.GetBcsNamespace(&client.GetBcsNamespaceRequest{ CommonRequest: client.CommonRequest{ - BKBizID: *request.BKBizID, + BKBizID: common.Deref(request.BKBizID), Page: client.Page{ Limit: 100, Start: 0, @@ -1112,12 +1112,12 @@ func (c *cmdbClient) CreateBcsNamespace(request *client.CreateBcsNamespaceReques return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api create_kube_namespace with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api create_kube_namespace with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID)) if _, err := c.GetBcsNamespace(&client.GetBcsNamespaceRequest{ CommonRequest: client.CommonRequest{ - BKBizID: *request.BKBizID, + BKBizID: common.Deref(request.BKBizID), Page: client.Page{ Limit: 100, Start: 0, @@ -1169,17 +1169,17 @@ func (c *cmdbClient) UpdateBcsNamespace(request *client.UpdateBcsNamespaceReques } if !respData.Result { - blog.Errorf("call api batch_update_namespace failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_update_namespace failed: %v, rid: %s, bkBizID: %d, ids: %v", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) return fmt.Errorf(respData.Message) } - blog.Infof("call api batch_update_namespace with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_update_namespace with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, ids: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) if _, err := c.GetBcsNamespace(&client.GetBcsNamespaceRequest{ CommonRequest: client.CommonRequest{ - BKBizID: *request.BKBizID, + BKBizID: common.Deref(request.BKBizID), Page: client.Page{ Limit: 100, Start: 0, @@ -1231,13 +1231,13 @@ func (c *cmdbClient) DeleteBcsNamespace(request *client.DeleteBcsNamespaceReques } if !respData.Result { - blog.Errorf("call api batch_delete_namespace failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_delete_namespace failed: %v, rid: %s, bkBizID: %d, ids: %v", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) return fmt.Errorf(respData.Message) } - blog.Infof("call api batch_delete_namespace with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_delete_namespace with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, ids: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) if db != nil { query := db.Session(&gorm.Session{NewDB: true}) @@ -1262,7 +1262,7 @@ func (c *cmdbClient) GetBcsWorkload( } if withDB && db != nil { - query := db.Session(&gorm.Session{NewDB: true, Logger: db.Logger.LogMode(logger.Info)}) + query := db.Session(&gorm.Session{NewDB: true}) for _, rule := range request.Filter.Rules { if request.Filter.Condition == And { query = query.Where(fmt.Sprintf("%s %s ?", rule.Field, rule.Operator), rule.Value) @@ -1275,7 +1275,7 @@ func (c *cmdbClient) GetBcsWorkload( switch request.Kind { case Deployment: var deployment []model.Deployment - if err := query.Debug().Find(&deployment).Error; err != nil { + if err := query.Find(&deployment).Error; err != nil { blog.Errorf("query deployment withDB failed: %v", err) } else { if deploymentMarshal, errM := json.Marshal(deployment); errM != nil { @@ -1292,7 +1292,7 @@ func (c *cmdbClient) GetBcsWorkload( } case StatefulSet: var statefulSet []model.StatefulSet - if err := query.Debug().Find(&statefulSet).Error; err != nil { + if err := query.Find(&statefulSet).Error; err != nil { blog.Errorf("query statefulSet withDB failed: %v", err) } else { if statefulSetMarshal, errM := json.Marshal(statefulSet); errM != nil { @@ -1309,7 +1309,7 @@ func (c *cmdbClient) GetBcsWorkload( } case DaemonSet: var daemonSet []model.DaemonSet - if err := query.Debug().Find(&daemonSet).Error; err != nil { + if err := query.Find(&daemonSet).Error; err != nil { blog.Errorf("query daemonSet withDB failed: %v", err) } else { if daemonSetMarshal, errM := json.Marshal(daemonSet); errM != nil { @@ -1326,7 +1326,7 @@ func (c *cmdbClient) GetBcsWorkload( } case GameDeployment: var gameDeployment []model.GameDeployment - if err := query.Debug().Find(&gameDeployment).Error; err != nil { + if err := query.Find(&gameDeployment).Error; err != nil { blog.Errorf("query gameDeployment withDB failed: %v", err) } else { if gameDeploymentMarshal, errM := json.Marshal(gameDeployment); errM != nil { @@ -1343,7 +1343,7 @@ func (c *cmdbClient) GetBcsWorkload( } case GameStatefulSet: var gameStatefulSet []model.GameStatefulSet - if err := query.Debug().Find(&gameStatefulSet).Error; err != nil { + if err := query.Find(&gameStatefulSet).Error; err != nil { blog.Errorf("query gameStatefulSet withDB failed: %v", err) } else { if gameStatefulSetMarshal, errM := json.Marshal(gameStatefulSet); errM != nil { @@ -1360,7 +1360,7 @@ func (c *cmdbClient) GetBcsWorkload( } case Pods: var podsWorkload []model.PodsWorkload - if err := query.Debug().Find(&podsWorkload).Error; err != nil { + if err := query.Find(&podsWorkload).Error; err != nil { blog.Errorf("query podsWorkload withDB failed: %v", err) } else { if podsWorkloadMarshal, errM := json.Marshal(podsWorkload); errM != nil { @@ -1377,7 +1377,7 @@ func (c *cmdbClient) GetBcsWorkload( } case CustomResource: var customResource []model.CustomResource - if err := query.Debug().Find(&customResource).Error; err != nil { + if err := query.Find(&customResource).Error; err != nil { blog.Errorf("query customResource withDB failed: %v", err) } else { if customResourceMarshal, errM := json.Marshal(customResource); errM != nil { @@ -1416,13 +1416,13 @@ func (c *cmdbClient) GetBcsWorkload( } if !respData.Result { - blog.Errorf("call api list_workload failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api list_workload failed: %v, rid: %s, bkBizID: %d, kind: %s", + respData.Message, resp.Header.Get("X-Request-Id"), request.BKBizID, request.Kind) return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api list_workload with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api list_workload with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, kind: %s", + reqURL, request, resp.Header.Get("X-Request-Id"), request.BKBizID, request.Kind) if db != nil { if workloadMarshal, err := json.Marshal(respData.Data.Info); err != nil { @@ -1660,14 +1660,14 @@ func (c *cmdbClient) CreateBcsWorkload(request *client.CreateBcsWorkloadRequest, } if !respData.Result { - blog.Errorf("call api batch_create_workload failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_create_workload failed: %v, rid: %s, bkBizID: %d, kind: %s", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.Kind)) if respData.Code == 1199014 && db != nil { for _, d := range *(request.Data) { if _, err := c.GetBcsWorkload(&client.GetBcsWorkloadRequest{ - Kind: *(request.Kind), + Kind: common.Deref(request.Kind), CommonRequest: client.CommonRequest{ - BKBizID: *request.BKBizID, + BKBizID: common.Deref(request.BKBizID), Page: client.Page{ Limit: 100, Start: 0, @@ -1696,13 +1696,13 @@ func (c *cmdbClient) CreateBcsWorkload(request *client.CreateBcsWorkloadRequest, return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api batch_create_workload with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_create_workload with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, kind: %s", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.Kind)) if _, err := c.GetBcsWorkload(&client.GetBcsWorkloadRequest{ - Kind: *(request.Kind), + Kind: common.Deref(request.Kind), CommonRequest: client.CommonRequest{ - BKBizID: *request.BKBizID, + BKBizID: common.Deref(request.BKBizID), Page: client.Page{ Limit: 100, Start: 0, @@ -1754,18 +1754,18 @@ func (c *cmdbClient) UpdateBcsWorkload(request *client.UpdateBcsWorkloadRequest, } if !respData.Result { - blog.Errorf("call api batch_update_workload failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_update_workload failed: %v, rid: %s, bkBizID: %d, kind: %s, ids: %v", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.Kind), common.Deref(request.IDs)) return fmt.Errorf(respData.Message) } - blog.Infof("call api batch_update_workload with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_update_workload with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, kind: %s, ids: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.Kind), common.Deref(request.IDs)) if _, err := c.GetBcsWorkload(&client.GetBcsWorkloadRequest{ - Kind: *request.Kind, + Kind: common.Deref(request.Kind), CommonRequest: client.CommonRequest{ - BKBizID: *request.BKBizID, + BKBizID: common.Deref(request.BKBizID), Page: client.Page{ Limit: 100, Start: 0, @@ -1819,12 +1819,12 @@ func (c *cmdbClient) DeleteBcsWorkload(request *client.DeleteBcsWorkloadRequest, if !respData.Result { blog.Errorf( "call api batch_delete_workload failed: %v, rid: %s, request: bkbizid: %d, kind: %s, ids: %v", - respData.Message, resp.Header.Get("X-Request-Id"), *request.BKBizID, *request.Kind, *request.IDs) + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.Kind), common.Deref(request.IDs)) return fmt.Errorf(respData.Message) } - blog.Infof("call api batch_delete_workload with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_delete_workload with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, kind: %s, ids: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.Kind), common.Deref(request.IDs)) deleteBcsWorkloadDB(request, db) @@ -1908,7 +1908,7 @@ func (c *cmdbClient) GetBcsNode( return nil, ErrServerNotInit } if withDB && db != nil { - query := db.Session(&gorm.Session{NewDB: true, Logger: db.Logger.LogMode(logger.Info)}) + query := db.Session(&gorm.Session{NewDB: true}) for _, rule := range request.Filter.Rules { if request.Filter.Condition == And { query = query.Where(fmt.Sprintf("%s %s ?", rule.Field, rule.Operator), rule.Value) @@ -1918,7 +1918,7 @@ func (c *cmdbClient) GetBcsNode( } } var node []model.Node - if err := query.Debug().Find(&node).Error; err != nil { + if err := query.Find(&node).Error; err != nil { blog.Errorf("query node withDB failed: %v", err) } else { if nodeMarshal, errM := json.Marshal(node); errM != nil { @@ -1953,12 +1953,12 @@ func (c *cmdbClient) GetBcsNode( return nil, errs[0] } if !respData.Result { - blog.Errorf("call api list_kube_node failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api list_kube_node failed: %v, rid: %s, bkBizID: %d", + respData.Message, resp.Header.Get("X-Request-Id"), request.BKBizID) return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api list_kube_node with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api list_kube_node with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d", + reqURL, request, resp.Header.Get("X-Request-Id"), request.BKBizID) if db != nil { if nodeMarshal, err := json.Marshal(respData.Data.Info); err != nil { blog.Errorf("marshal node failed: %v", err) @@ -2117,13 +2117,13 @@ func (c *cmdbClient) UpdateBcsNode(request *client.UpdateBcsNodeRequest, db *gor } if !respData.Result { - blog.Errorf("call api batch_update_kube_node failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_update_kube_node failed: %v, rid: %s, bkBizID: %d, ids: %v", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) return fmt.Errorf(respData.Message) } - blog.Infof("call api batch_update_kube_node with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_update_kube_node with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, ids: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) _, err := c.GetBcsNode(&client.GetBcsNodeRequest{ CommonRequest: client.CommonRequest{ @@ -2190,13 +2190,13 @@ func (c *cmdbClient) DeleteBcsNode(request *client.DeleteBcsNodeRequest, db *gor } if !respData.Result { - blog.Errorf("call api batch_delete_kube_node failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_delete_kube_node failed: %v, rid: %s, bkBizID: %d, ids: %v", + respData.Message, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) return fmt.Errorf(respData.Message) } - blog.Infof("call api batch_delete_kube_node with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_delete_kube_node with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, ids: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), common.Deref(request.BKBizID), common.Deref(request.IDs)) return nil } @@ -2211,7 +2211,7 @@ func (c *cmdbClient) GetBcsPod(request *client.GetBcsPodRequest, db *gorm.DB, wi } if withDB && db != nil { - query := db.Session(&gorm.Session{NewDB: true, Logger: db.Logger.LogMode(logger.Info)}) + query := db.Session(&gorm.Session{NewDB: true}) for _, rule := range request.Filter.Rules { if request.Filter.Condition == And { query = query.Where(fmt.Sprintf("%s %s ?", rule.Field, rule.Operator), rule.Value) @@ -2223,7 +2223,7 @@ func (c *cmdbClient) GetBcsPod(request *client.GetBcsPodRequest, db *gorm.DB, wi var pod []model.Pod - if err := query.Debug().Find(&pod).Error; err != nil { + if err := query.Find(&pod).Error; err != nil { blog.Errorf("query pod withDB failed: %v", err) } else { if podMarshal, errM := json.Marshal(pod); errM != nil { @@ -2262,13 +2262,13 @@ func (c *cmdbClient) GetBcsPod(request *client.GetBcsPodRequest, db *gorm.DB, wi } if !respData.Result { - blog.Errorf("call api list_pod failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api list_pod failed: %v, rid: %s, bkBizID: %d, filter: %v", + respData.Message, resp.Header.Get("X-Request-Id"), request.BKBizID, common.Deref(request.Filter)) return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api list_pod with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api list_pod with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, filter: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), request.BKBizID, common.Deref(request.Filter)) if db != nil { if podMarshal, err := json.Marshal(respData.Data.Info); err != nil { @@ -2310,7 +2310,7 @@ func (c *cmdbClient) GetBcsContainer( } if withDB && db != nil { - query := db.Session(&gorm.Session{NewDB: true, Logger: db.Logger.LogMode(logger.Info)}) + query := db.Session(&gorm.Session{NewDB: true}) for _, rule := range request.Filter.Rules { if request.Filter.Condition == And { query = query.Where(fmt.Sprintf("%s %s ?", rule.Field, rule.Operator), rule.Value) @@ -2322,7 +2322,7 @@ func (c *cmdbClient) GetBcsContainer( var container []model.Container - if err := query.Debug().Find(&container).Error; err != nil { + if err := query.Find(&container).Error; err != nil { blog.Errorf("query container withDB failed: %v", err) } else { if containerMarshal, errM := json.Marshal(container); errM != nil { @@ -2361,13 +2361,13 @@ func (c *cmdbClient) GetBcsContainer( } if !respData.Result { - blog.Errorf("call api list_kube_container failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api list_kube_container failed: %v, rid: %s, bkBizID: %d, filter: %v", + respData.Message, resp.Header.Get("X-Request-Id"), request.BKBizID, common.Deref(request.Filter)) return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api list_kube_container with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api list_kube_container with url(%s) (%v) successfully, X-Request-Id: %s, bkBizID: %d, filter: %v", + reqURL, request, resp.Header.Get("X-Request-Id"), request.BKBizID, common.Deref(request.Filter)) if db != nil { if containerMarshal, err := json.Marshal(respData.Data.Info); err != nil { @@ -2430,8 +2430,8 @@ func (c *cmdbClient) CreateBcsPod(request *client.CreateBcsPodRequest, db *gorm. } if !respData.Result { - blog.Errorf("call api batch_create_kube_pod failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_create_kube_pod failed: %v, rid: %s, pod_nums: %d", + respData.Message, resp.Header.Get("X-Request-Id"), len(common.Deref(request.Data))) if respData.Code == 1199014 && db != nil { data := (*(request.Data))[0] for _, d := range *data.Pods { @@ -2472,8 +2472,8 @@ func (c *cmdbClient) CreateBcsPod(request *client.CreateBcsPodRequest, db *gorm. return nil, fmt.Errorf(respData.Message) } - blog.Infof("call api batch_create_kube_pod with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_create_kube_pod with url(%s) (%v) successfully, X-Request-Id: %s, pod_nums: %d", + reqURL, request, resp.Header.Get("X-Request-Id"), len(common.Deref(request.Data))) if db != nil { _, err := c.GetBcsPod(&client.GetBcsPodRequest{ @@ -2558,13 +2558,13 @@ func (c *cmdbClient) DeleteBcsPod(request *client.DeleteBcsPodRequest, db *gorm. } if !respData.Result { - blog.Errorf("call api batch_delete_kube_pod failed: %v, rid: %s", - respData.Message, resp.Header.Get("X-Request-Id")) + blog.Errorf("call api batch_delete_kube_pod failed: %v, rid: %s, data_nums: %d", + respData.Message, resp.Header.Get("X-Request-Id"), len(common.Deref(request.Data))) return fmt.Errorf(respData.Message) } - blog.Infof("call api batch_delete_kube_pod with url(%s) (%v) successfully, X-Request-Id: %s", - reqURL, request, resp.Header.Get("X-Request-Id")) + blog.Infof("call api batch_delete_kube_pod with url(%s) (%v) successfully, X-Request-Id: %s, data_nums: %d", + reqURL, request, resp.Header.Get("X-Request-Id"), len(common.Deref(request.Data))) if db != nil { query := db.Session(&gorm.Session{NewDB: true}) diff --git a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client/cmdb/cmdb_test.go b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client/cmdb/cmdb_test.go index b2a79869e6..42986291f2 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client/cmdb/cmdb_test.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/client/cmdb/cmdb_test.go @@ -2443,7 +2443,7 @@ func Test_cmdbClient_GetHostsByBiz(t *testing.T) { } func Test_gorm_container(t *testing.T) { - sq := mySqlite.New("./test1.db") + sq := mySqlite.New("./test1.db", 2) if sq == nil { t.Fatal("failed to open database") } @@ -2499,7 +2499,7 @@ func Test_gorm_container(t *testing.T) { } func Test_gorm_pod(t *testing.T) { - sq := mySqlite.New("./test1.db") + sq := mySqlite.New("./test1.db", 2) if sq == nil { t.Fatal("failed to open database") } @@ -2564,7 +2564,7 @@ func Test_gorm_pod(t *testing.T) { } func Test_gorm_node(t *testing.T) { - sq := mySqlite.New("./test1.db") + sq := mySqlite.New("./test1.db", 2) if sq == nil { t.Fatal("failed to open database") } @@ -2642,7 +2642,7 @@ func Test_gorm_node(t *testing.T) { } func Test_gorm_deployment(t *testing.T) { - sq := mySqlite.New("./test1.db") + sq := mySqlite.New("./test1.db", 2) if sq == nil { t.Fatal("failed to open database") } @@ -2711,7 +2711,7 @@ func Test_gorm_deployment(t *testing.T) { } func Test_gorm_cluster(t *testing.T) { - sq := mySqlite.New("./test1.db") + sq := mySqlite.New("./test1.db", 2) if sq == nil { t.Fatal("failed to open database") } @@ -2778,7 +2778,7 @@ func Test_gorm_cluster(t *testing.T) { // nolint:golint func Test_cmdbClient_GetBcsCluster_withDB(t *testing.T) { - sq := mySqlite.New("./test1.db") + sq := mySqlite.New("./test1.db", 2) if sq == nil { t.Fatal("failed to open database") } @@ -2817,7 +2817,7 @@ func Test_cmdbClient_GetBcsCluster_withDB(t *testing.T) { } func Test_gorm_namespace(t *testing.T) { - sq := mySqlite.New("./test1.db") + sq := mySqlite.New("./test1.db", 2) if sq == nil { t.Fatal("failed to open database") } diff --git a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/common/common.go b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/common/common.go index 946269e91d..b93763f644 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/common/common.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/common/common.go @@ -19,6 +19,7 @@ import ( "hash/fnv" "os" "reflect" + "strconv" "strings" "time" @@ -135,6 +136,36 @@ func SetCustomResourceTypesFromEnv(opt *option.BkcmdbSynchronizerOption) error { return nil } +// SetSqlLogLevelFromEnv parses synchronizer_sqlLogLevel environment variable +// and sets it to the option. +func SetSqlLogLevelFromEnv(opt *option.BkcmdbSynchronizerOption) error { + envValue := os.Getenv("synchronizer_sqlLogLevel") + if envValue == "" { + return nil + } + logLevel, err := strconv.Atoi(envValue) + if err != nil { + return fmt.Errorf("parse sql log level failed: %w", err) + } + opt.Synchronizer.SqlLogLevel = logLevel + return nil +} + +// SetCleanLocalCacheFromEnv parses synchronizer_cleanLocalCache environment variable +// and sets it to the option. +func SetCleanLocalCacheFromEnv(opt *option.BkcmdbSynchronizerOption) error { + envValue := os.Getenv("synchronizer_cleanLocalCache") + if envValue == "" { + return nil + } + cleanLocalCache, err := strconv.ParseBool(envValue) + if err != nil { + return fmt.Errorf("parse clean local cache failed: %w", err) + } + opt.Synchronizer.CleanLocalCache = cleanLocalCache + return nil +} + // IsKindInSlice checks if a kind exists in the whitelist func IsKindInSlice(kind string, whitelist []string) bool { for _, s := range whitelist { @@ -144,3 +175,18 @@ func IsKindInSlice(kind string, whitelist []string) bool { } return false } + +// Deref dereferences a pointer and returns its value. If the pointer is nil, it returns the zero value of the type. +// This is a generic tool to prevent panics when dereferencing nil pointers. +func Deref[T any](p *T) T { + if p == nil { + var v T + return v + } + return *p +} + +// Ptr returns a pointer to the given value. +func Ptr[T any](v T) *T { + return &v +} diff --git a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/handler/handler.go b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/handler/handler.go index 7390447061..11cd42eaad 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/handler/handler.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/handler/handler.go @@ -126,7 +126,7 @@ func (b *BcsBkcmdbSynchronizerHandler) HandleMsg( path := "/data/bcs/bcs-bkcmdb-synchronizer/db/" + clusterId + ".db" - db := sqlite.Open(path) + db := sqlite.Open(path, b.Syncer.BkcmdbSynchronizerOption.Synchronizer.SqlLogLevel) if db == nil { blog.Errorf("open db failed, path: %s", path) return @@ -197,7 +197,7 @@ func (b *BcsBkcmdbSynchronizerHandler) HandleMsg( if v, ok := header["resourceType"]; ok { var errH error - blog.Infof("resourceType: %v", v) + blog.Infof("resourceType: %v, clusterUid: %s, bkBizID: %d", v, bkCluster.Uid, bkCluster.BizID) switch v.(string) { case "Pod": m := podMsg.M @@ -292,8 +292,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleCluster( } // 打印白名单和黑名单信息 - blog.Infof("whiteList: %v, len: %d", whiteList, len(whiteList)) - blog.Infof("blackList: %v, len: %d", blackList, len(blackList)) + blog.Infof("whiteList: %v, len: %d; blackList: %v, len: %d", whiteList, len(whiteList), blackList, len(blackList)) // 遍历所有集群 for _, cluster := range clusters { @@ -415,7 +414,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handlePods(podMsg *msgBuffer, bkCluster * // blog.Errorf("handlePod: Unable to unmarshal") // return fmt.Errorf("handlePod: Unable to unmarshal") // } - blog.Infof("podMsg: %d", len(podMsg.M)) + blog.Infof("podMsg: %d, clusterUid: %s, bkBizID: %d", len(podMsg.M), bkCluster.Uid, bkCluster.BizID) if time.Since(podMsg.T) < 10*time.Second { // blog.Infof("podMsg.T: %s, %s", podMsg.T, time.Now().Sub(podMsg.T)) if len(podMsg.M) < 100 { @@ -817,7 +816,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handlePodsDelete( } // 打印日志，显示将要处理的Pod名称 - blog.Infof("handlePodsDelete podNames: %v", nsPod) + blog.Infof("handlePodsDelete podNames: %v, clusterUid: %s, bkBizID: %d", nsPod, bkCluster.Uid, bkCluster.BizID) // 创建一个切片，用于存储要删除的BkPod的ID bkPodIDs := make([]int64, 0) @@ -1403,7 +1402,8 @@ func (b *BcsBkcmdbSynchronizerHandler) handlePodCreate(pod *corev1.Pod, bkCluste }, }, nil) - blog.Infof("podToAdd: %s+%s+%s", bkCluster.Uid, &pod.Namespace, &pod.Name) + blog.Infof("podToAdd, clusterUid: %s, bizID: %d, clusterID: %d, namespaceID: %d, namespace: %s, workloadKind: %s, workloadName: %s, workloadID: %d, podName: %s, podIP: %s", + bkCluster.Uid, bkNamespace.BizID, bkCluster.ID, bkNamespace.ID, pod.Namespace, workloadKind, workloadName, workloadID, pod.Name, pod.Status.PodIP) return nil } @@ -1418,7 +1418,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handlePodsCreate(podsCreate map[string]*c for _, v := range podsCreate { podNames = append(podNames, v.Name) } - blog.Infof("handlePodsCreate podNames: %v", podNames) + blog.Infof("handlePodsCreate podNames: %v, clusterUid: %s, bkBizID: %d", podNames, bkCluster.Uid, bkCluster.BizID) lcReq := cmp.ListClusterReq{ ClusterID: bkCluster.Uid, @@ -1823,7 +1823,8 @@ func (b *BcsBkcmdbSynchronizerHandler) handlePodsCreate(podsCreate map[string]*c }, }, }, db) - blog.Infof("podToAdd: %s+%s+%s", bkCluster.Uid, pod.Namespace, pod.Name) + blog.Infof("podToAdd, clusterUid: %s, bizID: %d, clusterID: %d, namespaceID: %d, namespace: %s, workloadKind: %s, workloadName: %s, workloadID: %d, podName: %s, podIP: %s", + bkCluster.Uid, bkNamespace.BizID, bkCluster.ID, bkNamespace.ID, pod.Namespace, workloadKind, workloadName, workloadID, pod.Name, pod.Status.PodIP) } return nil @@ -1833,7 +1834,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handlePodsCreate(podsCreate map[string]*c func (b *BcsBkcmdbSynchronizerHandler) handleDeployment( msg amqp.Delivery, bkCluster *bkcmdbkube.Cluster, db *gorm.DB) error { // 记录接收到的消息头信息 - blog.Infof("handleDeployment Message: %v", msg.Headers) + blog.Infof("handleDeployment Message: %v, clusterUid: %s, bkBizID: %d", msg.Headers, bkCluster.Uid, bkCluster.BizID) // 尝试获取消息头信息 msgHeader, err := getMsgHeader(&msg.Headers) @@ -2102,7 +2103,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleDeploymentCreate( func (b *BcsBkcmdbSynchronizerHandler) handleStatefulSet( msg amqp.Delivery, bkCluster *bkcmdbkube.Cluster, db *gorm.DB) error { - blog.Infof("handleStatefulSet Message: %v", msg.Headers) + blog.Infof("handleStatefulSet Message: %v, clusterUid: %s, bkBizID: %d", msg.Headers, bkCluster.Uid, bkCluster.BizID) msgHeader, err := getMsgHeader(&msg.Headers) if err != nil { blog.Errorf("handleStatefulSet unable to get headers, err: %s", err.Error()) @@ -2330,7 +2331,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleStatefulSetCreate( func (b *BcsBkcmdbSynchronizerHandler) handleDaemonSet( msg amqp.Delivery, bkCluster *bkcmdbkube.Cluster, db *gorm.DB) error { - blog.Infof("handleDaemonSet Message: %v", msg.Headers) + blog.Infof("handleDaemonSet Message: %v, clusterUid: %s, bkBizID: %d", msg.Headers, bkCluster.Uid, bkCluster.BizID) msgHeader, err := getMsgHeader(&msg.Headers) if err != nil { blog.Errorf("handleDaemonSet unable to get headers, err: %s", err.Error()) @@ -2521,7 +2522,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleDaemonSetCreate( func (b *BcsBkcmdbSynchronizerHandler) handleGameDeployment(msg amqp.Delivery, bkCluster *bkcmdbkube.Cluster, db *gorm.DB) error { - blog.Infof("handleGameDeployment Message: %v", msg.Headers) + blog.Infof("handleGameDeployment Message: %v, clusterUid: %s, bkBizID: %d", msg.Headers, bkCluster.Uid, bkCluster.BizID) msgHeader, err := getMsgHeader(&msg.Headers) if err != nil { blog.Errorf("handleGameDeployment unable to get headers, err: %s", err.Error()) @@ -2719,7 +2720,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleGameDeploymentCreate( // handle GameStateful Set func (b *BcsBkcmdbSynchronizerHandler) handleGameStatefulSet( msg amqp.Delivery, bkCluster *bkcmdbkube.Cluster, db *gorm.DB) error { - blog.Infof("handleGameStatefulSet Message: %v", msg.Headers) + blog.Infof("handleGameStatefulSet Message: %v, clusterUid: %s, bkBizID: %d", msg.Headers, bkCluster.Uid, bkCluster.BizID) msgHeader, err := getMsgHeader(&msg.Headers) if err != nil { blog.Errorf("handleGameStatefulSet unable to get headers, err: %s", err.Error()) @@ -2916,7 +2917,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleGameStatefulSetCreate( func (b *BcsBkcmdbSynchronizerHandler) handleNamespace( msg amqp.Delivery, bkCluster *bkcmdbkube.Cluster, db *gorm.DB) error { - blog.Infof("handleNamespace Message: %v", msg.Headers) + blog.Infof("handleNamespace Message: %v, clusterUid: %s, bkBizID: %d", msg.Headers, bkCluster.Uid, bkCluster.BizID) msgHeader, err := getMsgHeader(&msg.Headers) if err != nil { blog.Errorf("handleNamespace unable to get headers, err: %s", err.Error()) @@ -3151,7 +3152,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleEvent(msg amqp.Delivery, bkCluster // Node handle func (b *BcsBkcmdbSynchronizerHandler) handleNode(msg amqp.Delivery, bkCluster *bkcmdbkube.Cluster) error { // nolint - blog.Infof("handleNode Message: %v", msg.Headers) + blog.Infof("handleNode Message: %v, clusterUid: %s, bkBizID: %d", msg.Headers, bkCluster.Uid, bkCluster.BizID) msgHeader, err := getMsgHeader(&msg.Headers) if err != nil { blog.Errorf("handleNode unable to get headers, err: %s", err.Error()) @@ -3202,7 +3203,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleNodes( // return fmt.Errorf("handleNode: Unable to unmarshal") // } - blog.Infof("nodeMsg: %d", len(nodeMsg.M)) + blog.Infof("nodeMsg: %d, clusterUid: %s, bkBizID: %d", len(nodeMsg.M), bkCluster.Uid, bkCluster.BizID) if time.Since(nodeMsg.T) < 10*time.Second { // blog.Infof("podMsg.T: %s, %s", podMsg.T, time.Now().Sub(podMsg.T)) if len(nodeMsg.M) < 100 { @@ -3583,7 +3584,7 @@ func (b *BcsBkcmdbSynchronizerHandler) handleCustomResource( clusterID := msgHeader.ClusterId crKinds, ok := b.Syncer.BkcmdbSynchronizerOption.Synchronizer.CustomResourceTypes[clusterID] if !ok || len(crKinds) == 0 { - blog.Infof("cluster %s not configured for custom resource sync, skip", clusterID) + blog.Infof("cluster %s not configured for custom resource sync, bkBizID: %d, skip", clusterID, bkCluster.BizID) return nil } diff --git a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/option/option.go b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/option/option.go index 91a12839f9..4b1fce1c13 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/option/option.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/option/option.go @@ -41,6 +41,10 @@ type SynchronizerConfig struct { // Only clusters configured in this map will have custom resources synced // Example: {"cluster-id-1": ["BkApp", "CronJob"], "cluster-id-2": ["BkApp"]} CustomResourceTypes map[string][]string `json:"customResourceTypes"` + // SqlLogLevel define sql log level + SqlLogLevel int `json:"sqlLogLevel"` + // CleanLocalCache define whether to clean local cache + CleanLocalCache bool `json:"cleanLocalCache"` } // CustomResourceType defines parsed custom resource type configuration. diff --git a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/store/db/sqlite/sqlite.go b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/store/db/sqlite/sqlite.go index 9a7fe33df7..3565f37f1f 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/store/db/sqlite/sqlite.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/store/db/sqlite/sqlite.go @@ -17,6 +17,7 @@ import ( "github.com/Tencent/bk-bcs/bcs-common/common/blog" "gorm.io/driver/sqlite" "gorm.io/gorm" + "gorm.io/gorm/logger" ) // SQLite 结构体包含一个 GORM 数据库实例和数据库文件路径 @@ -27,9 +28,9 @@ type SQLite struct { // New 是 SQLite 的构造函数，它接受一个数据库文件路径作为参数 // 并尝试打开数据库连接，如果成功则返回一个初始化的 SQLite 实例 -func New(path string) *SQLite { - db := Open(path) // 尝试打开数据库连接 - if db == nil { // 如果打开失败，返回 nil +func New(path string, logLevel int) *SQLite { + db := Open(path, logLevel) // 尝试打开数据库连接 + if db == nil { // 如果打开失败，返回 nil return nil } return &SQLite{ // 返回初始化的 SQLite 实例 @@ -40,9 +41,14 @@ func New(path string) *SQLite { // Open 函数尝试使用给定的路径打开一个 SQLite 数据库连接 // 如果成功，它返回一个 GORM 数据库实例；如果失败，它会记录错误并返回 nil -func Open(path string) *gorm.DB { - db, err := gorm.Open(sqlite.Open(path), &gorm.Config{}) // 尝试打开数据库连接 - if err != nil { // 如果发生错误 +func Open(path string, logLevel int) *gorm.DB { + blog.Infof("sqlite.Open called with path: %s, logLevel: %d", path, logLevel) + config := &gorm.Config{ + Logger: logger.Default.LogMode(logger.LogLevel(logLevel)), + } + + db, err := gorm.Open(sqlite.Open(path), config) // 尝试打开数据库连接 + if err != nil { // 如果发生错误 blog.Errorf("Failed to open the SQLite database: %v", err) // 记录错误 return nil // 返回 nil } diff --git a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/syncer/syncer.go b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/syncer/syncer.go index 1c0909b397..1f0a3258f5 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/syncer/syncer.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/internal/pkg/syncer/syncer.go @@ -83,6 +83,9 @@ func (s *Syncer) Init() { } bkBizID = s.BkcmdbSynchronizerOption.Synchronizer.BkBizID hostID = s.BkcmdbSynchronizerOption.Synchronizer.HostID + if s.BkcmdbSynchronizerOption.Synchronizer.SqlLogLevel == 0 { + s.BkcmdbSynchronizerOption.Synchronizer.SqlLogLevel = 2 + } } // init Tls Config @@ -144,7 +147,7 @@ func (s *Syncer) SyncCluster(cluster *cmp.Cluster) error { clusterType = "SHARE_CLUSTER" } path := "/data/bcs/bcs-bkcmdb-synchronizer/db/" + cluster.ClusterID + ".db" - db := sqlite.Open(path) + db := sqlite.Open(path, s.BkcmdbSynchronizerOption.Synchronizer.SqlLogLevel) if db == nil { blog.Errorf("open db failed, path: %s", path) return fmt.Errorf("open db failed, path: %s", path) @@ -157,8 +160,6 @@ func (s *Syncer) SyncCluster(cluster *cmp.Cluster) error { clusterBkBizID, err = strconv.ParseInt(cluster.BusinessID, 10, 64) if err != nil { blog.Errorf("An error occurred: %s\n", err) - } else { - blog.Infof("Successfully converted string to int64: %d\n", clusterBkBizID) } } else { clusterBkBizID = bkBizID @@ -2055,7 +2056,8 @@ func (s *Syncer) syncPodsCheck(podMap map[string]*storage.Pod, bkPodMap map[stri Containers: &containers, Operator: &operator, }) - blog.Infof("podToAdd: %s+%s+%s", bkCluster.Uid, v.Data.Namespace, v.Data.Name) + blog.Infof("podToAdd, clusterUid: %s, bizID: %d, clusterID: %d, namespaceID: %d, namespace: %s, workloadKind: %s, workloadName: %s, workloadID: %d, podName: %s, podIP: %s", + bkCluster.Uid, bkNsMap[v.Data.Namespace].BizID, bkCluster.ID, bkNsMap[v.Data.Namespace].ID, v.Data.Namespace, workloadKind, workloadName, workloadID, v.Data.Name, v.Data.Status.PodIP) } else { podToAdd[bkNsMap[v.Data.Namespace].BizID] = []client.CreateBcsPodRequestDataPod{ client.CreateBcsPodRequestDataPod{ @@ -2082,7 +2084,8 @@ func (s *Syncer) syncPodsCheck(podMap map[string]*storage.Pod, bkPodMap map[stri Operator: &operator, }, } - blog.Infof("podToAdd: %s+%s+%s", bkCluster.Uid, v.Data.Namespace, v.Data.Name) + blog.Infof("podToAdd, clusterUid: %s, bizID: %d, clusterID: %d, namespaceID: %d, namespace: %s, workloadKind: %s, workloadName: %s, workloadID: %d, podName: %s, podIP: %s", + bkCluster.Uid, bkNsMap[v.Data.Namespace].BizID, bkCluster.ID, bkNsMap[v.Data.Namespace].ID, v.Data.Namespace, workloadKind, workloadName, workloadID, v.Data.Name, v.Data.Status.PodIP) } } else { @@ -2098,7 +2101,7 @@ func (s *Syncer) syncPodsCheck(podMap map[string]*storage.Pod, bkPodMap map[stri func (s *Syncer) SyncStore(bkCluster *bkcmdbkube.Cluster, force bool) error { path := "/data/bcs/bcs-bkcmdb-synchronizer/db/" + bkCluster.Uid + ".db" - db := sqlite.Open(path) + db := sqlite.Open(path, s.BkcmdbSynchronizerOption.Synchronizer.SqlLogLevel) if db == nil { blog.Errorf("open db failed, path: %s", path) return fmt.Errorf("open db failed, path: %s", path) @@ -3016,8 +3019,6 @@ func (s *Syncer) GetBkCluster(cluster *cmp.Cluster, db *gorm.DB, withDB bool) (* bizid, err := strconv.ParseInt(cluster.BusinessID, 10, 64) if err != nil { blog.Errorf("An error occurred: %s\n", err) - } else { - blog.Infof("Successfully converted string to int64: %d\n", bizid) } clusterBkBizID = bizid } else { diff --git a/bcs-services/bcs-bkcmdb-synchronizer/internal/synchronizer/synchronizer.go b/bcs-services/bcs-bkcmdb-synchronizer/internal/synchronizer/synchronizer.go index a48173156c..026640a929 100644 --- a/bcs-services/bcs-bkcmdb-synchronizer/internal/synchronizer/synchronizer.go +++ b/bcs-services/bcs-bkcmdb-synchronizer/internal/synchronizer/synchronizer.go @@ -240,8 +240,7 @@ func (s *Synchronizer) Run() { blackList = strings.Split(s.BkcmdbSynchronizerOption.Synchronizer.BlackList, ",") } - blog.Infof("whiteList: %v, len: %d", whiteList, len(whiteList)) - blog.Infof("blackList: %v, len: %d", blackList, len(blackList)) + blog.Infof("whiteList: %v, len: %d; blackList: %v, len: %d", whiteList, len(whiteList), blackList, len(blackList)) hostname, err := os.Hostname() if err != nil { @@ -297,6 +296,11 @@ func (s *Synchronizer) Run() { gm.Start(w, clusterMap[w]) } + // 初始清理本地缓存 + if s.BkcmdbSynchronizerOption.Synchronizer.CleanLocalCache { + s.cleanupLocalCaches(workList) + } + go func() { time.Sleep(time.Second * 10) ticker := time.NewTicker(10 * time.Minute) @@ -332,6 +336,12 @@ func (s *Synchronizer) Run() { blog.Errorf("cleanup orphaned clusters failed: %s", err.Error()) } } + + // 清理本地不再负责的集群缓存 + if s.BkcmdbSynchronizerOption.Synchronizer.CleanLocalCache { + s.cleanupLocalCaches(workListT) + } + prevWorkList, prevClusterMap = workListT, clusterMapT } }() @@ -737,7 +747,7 @@ func (s *Synchronizer) syncStoreHandler(podIndex int, whiteList, blackList []str func (s *Synchronizer) syncStorage(cluster *cmp.Cluster, bkCluster *bkcmdbkube.Cluster, force bool) { path := "/data/bcs/bcs-bkcmdb-synchronizer/db/" + bkCluster.Uid + ".db" - db := sqlite.Open(path) + db := sqlite.Open(path, s.BkcmdbSynchronizerOption.Synchronizer.SqlLogLevel) if db == nil { blog.Errorf("open db failed, path: %s", path) } @@ -1027,3 +1037,37 @@ func (s *Synchronizer) cleanupOrphanedClustersInCMDB(clusterMapT map[string]*cmp blog.Infof("cleanup orphaned clusters completed, deleted %d clusters total across all businesses", totalOrphanedCount) return nil } + +// cleanupLocalCaches 清理本地不再负责的集群缓存文件 +func (s *Synchronizer) cleanupLocalCaches(currentWorkList ClusterList) { + dbDir := "/data/bcs/bcs-bkcmdb-synchronizer/db/" + files, err := ioutil.ReadDir(dbDir) + if err != nil { + blog.Errorf("read db dir %s failed: %v", dbDir, err) + return + } + + workListMap := make(map[string]bool) + for _, clusterID := range currentWorkList { + workListMap[clusterID+".db"] = true + } + + for _, file := range files { + if file.IsDir() { + continue + } + + fileName := file.Name() + if !strings.HasSuffix(fileName, ".db") { + continue + } + + if _, ok := workListMap[fileName]; !ok { + blog.Infof("cleaning up orphaned local cache: %s", fileName) + err := os.Remove(dbDir + fileName) + if err != nil { + blog.Errorf("remove orphaned local cache %s failed: %v", fileName, err) + } + } + } +}