feat: 集群项目数据实时订阅机制

Author: LidolLxf

bcs-services/bcs-storage/Makefile

@@ -1,16 +1,18 @@
 
 .PHONY: init
 init:
-	go install github.com/golang/protobuf/protoc-gen-go@latest
-	go install go install go-micro.dev/v4/cmd/[email protected]
+	go install google.golang.org/protobuf/cmd/[email protected]
+	go install google.golang.org/grpc/cmd/[email protected]
+	go install github.com/go-micro/generator/cmd/protoc-gen-micro@latest
 	go install github.com/micro/micro/v3/cmd/protoc-gen-openapi@latest
 	go install github.com/grpc-ecosystem/grpc-gateway/v2/[email protected]
 	go install github.com/grpc-ecosystem/grpc-gateway/v2/[email protected]
+	go install github.com/envoyproxy/[email protected]
 
 .PHONY: api
 api:
-	protoc -I../../bcs-common/pkg/third_party/ --proto_path=. --openapi_out=./pkg/proto/ pkg/proto/bcs-storage.proto
+	protoc -I./third_party/ --proto_path=. --openapi_out=./pkg/proto/ pkg/proto/bcs-storage.proto
 
 .PHONY: proto
 proto:
-	protoc -I../../bcs-common/pkg/third_party/ --proto_path=. --validate_out=paths=source_relative,lang=go:. --grpc-gateway_out=logtostderr=true,register_func_suffix=Gw:./pkg/proto/ --swagger_out=logtostderr=true:. --go-grpc_out=./pkg/proto/  --micro_out=./pkg/proto/ --go_out=:./pkg/proto/ ./pkg/proto/bcs-storage.proto
+	protoc -I./third_party/ --proto_path=. --validate_out=paths=source_relative,lang=go:. --grpc-gateway_out=logtostderr=true,register_func_suffix=Gw:./pkg/proto/ --swagger_out=logtostderr=true:. --go-grpc_out=./pkg/proto/  --micro_out=./pkg/proto/ --go_out=:./pkg/proto/ ./pkg/proto/bcs-storage.proto

bcs-services/bcs-storage/go.mod

@@ -40,6 +40,7 @@ require (
 	github.com/envoyproxy/protoc-gen-validate v1.0.4
 	github.com/go-micro/plugins/v4/client/grpc v1.1.0
 	github.com/go-micro/plugins/v4/server/grpc v1.1.1
+	github.com/golang/glog v1.2.0
 	github.com/golang/protobuf v1.5.4
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
 	github.com/pkg/errors v0.9.1
@@ -64,6 +65,7 @@ require (
 	github.com/evanphx/json-patch/v5 v5.5.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.5.4 // indirect
+	github.com/ghodss/yaml v1.0.0 // indirect
 	github.com/go-acme/lego/v4 v4.4.0 // indirect
 	github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect
 	github.com/go-git/go-billy/v5 v5.6.0 // indirect

bcs-services/bcs-storage/pkg/constants/constants.go

@@ -34,6 +34,10 @@ const (
 	TypeTag = "type"
 	// ClusterIDTag clusterId
 	ClusterIDTag = "clusterId"
+	// ClusterIDUpperTag clusterID
+	ClusterIDUpperTag = "clusterID"
+	// ProjectIDTag projectID
+	ProjectIDTag = "projectID"
 	// NamespaceTag namespace
 	NamespaceTag = "namespace"
 	// MessageTag message

bcs-services/bcs-storage/pkg/handler/bcs-storage.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/constants"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/alarm"
+	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/cluster"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/clusterconfig"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/dynamic"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/dynamicquery"
@@ -32,11 +33,14 @@ import (
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/hostconfig"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/metric"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/metricwatch"
+	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/project"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/handler/internal/watchk8smesos"
 	storage "github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/proto"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/util"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/storage/actions/lib"
+	v1httpclu "github.com/Tencent/bk-bcs/bcs-services/bcs-storage/storage/actions/v1http/cluster"
 	dynamic2 "github.com/Tencent/bk-bcs/bcs-services/bcs-storage/storage/actions/v1http/dynamic"
+	v1httppro "github.com/Tencent/bk-bcs/bcs-services/bcs-storage/storage/actions/v1http/project"
 )
 
 // Storage storage struct, which include operations of alarm,
@@ -126,6 +130,110 @@ func (s *Storage) ListAlarm(ctx context.Context, req *storage.ListAlarmRequest,
 	return nil
 }
 
+// PutProjectInfo 订阅项目数据
+func (s *Storage) PutProjectInfo(ctx context.Context, req *storage.PutProjectInfoRequest,
+	rsp *storage.PutProjectInfoResponse) error {
+	// 如果请求合法,则执行函数请求
+	if err := req.Validate(); err != nil {
+		rsp.Result = false
+		rsp.Message = err.Error()
+		rsp.Code = common.AdditionErrorCode + 500
+		return nil
+	}
+	// 打印请求体
+	blog.Infof("PutProjectInfo req: %v", util.PrettyStruct(req))
+	data, err := project.HandlerCreateProjectInfoReq(ctx, req)
+	if err != nil {
+		// 如果处理过程中出错，将错误信息直接返回
+		rsp.Result = false
+		rsp.Code = common.BcsErrStoragePutResourceFail
+		rsp.Message = common.BcsErrStoragePutResourceFailStr
+		blog.Errorf("PutProjectInfo | put project info failed.err: %v", err)
+		return nil
+	}
+
+	v1httppro.PushCreateProjectInfoToQueue(data)
+
+	// 将查询结果添加至响应体
+	rsp.Result = true
+	rsp.Code = common.BcsSuccess
+	rsp.Message = common.BcsSuccessStr
+
+	// 打印响应体
+	blog.Infof("PutProject rsp: %v", util.PrettyStruct(rsp))
+	return nil
+}
+
+// PutClusterInfo 订阅集群数据
+func (s *Storage) PutClusterInfo(ctx context.Context, req *storage.PutClusterInfoRequest,
+	rsp *storage.PutClusterInfoResponse) error {
+	// 如果请求合法,则执行函数请求
+	if err := req.Validate(); err != nil {
+		rsp.Result = false
+		rsp.Message = err.Error()
+		rsp.Code = common.AdditionErrorCode + 500
+		return nil
+	}
+	// 打印请求体
+	blog.Infof("PutClusterInfo req: %v", util.PrettyStruct(req))
+	data, err := cluster.HandlerCreateClusterInfoReq(ctx, req)
+	if err != nil {
+		// 如果处理过程中出错，将错误信息直接返回
+		rsp.Result = false
+		rsp.Code = common.BcsErrStoragePutResourceFail
+		rsp.Message = common.BcsErrStoragePutResourceFailStr
+		blog.Errorf("PutClusterInfo | put cluster info failed.err: %v", err)
+		return nil
+	}
+
+	v1httpclu.PushCreateClusterInfoToQueue(data)
+
+	// 将查询结果添加至响应体
+	rsp.Result = true
+	rsp.Code = common.BcsSuccess
+	rsp.Message = common.BcsSuccessStr
+
+	// 打印响应体
+	blog.Infof("PutProject rsp: %v", util.PrettyStruct(rsp))
+	return nil
+}
+
+// DeleteClusterInfo 删除集群数据
+func (s *Storage) DeleteClusterInfo(ctx context.Context, req *storage.DeleteClusterInfoRequest,
+	rsp *storage.PutClusterInfoResponse) error {
+	// 如果请求合法,则执行函数请求
+	if err := req.Validate(); err != nil {
+		rsp.Result = false
+		rsp.Message = err.Error()
+		rsp.Code = common.AdditionErrorCode + 500
+		return nil
+	}
+	// 打印请求体
+	blog.Infof("DeleteClusterInfo req: %v", util.PrettyStruct(req))
+	data, err := cluster.HandlerDeleteClusterInfoReq(ctx, req)
+	if err != nil {
+		// 如果处理过程中出错，将错误信息直接返回
+		rsp.Result = false
+		rsp.Code = common.BcsErrStorageDeleteResourceFail
+		rsp.Message = common.BcsErrStorageDeleteResourceFailStr
+		blog.Errorf("DeleteClusterInfo | delete cluster info failed.err: %v", err)
+		return nil
+	}
+
+	if len(data) == 1 {
+		v1httpclu.PushDeleteClusterInfoToQueue(data[0])
+	}
+
+	// 将查询结果添加至响应体
+	rsp.Result = true
+	rsp.Code = common.BcsSuccess
+	rsp.Message = common.BcsSuccessStr
+
+	// 打印响应体
+	blog.Infof("DeleteClusterInfo rsp: %v", util.PrettyStruct(rsp))
+	return nil
+}
+
 // ****  Cluster Config(集群配置) ****
 
 // GetClusterConfig 获取集群配置 get config of cluster

bcs-services/bcs-storage/pkg/handler/internal/cluster/cluster.go

@@ -0,0 +1,94 @@
+/*
+ * Tencent is pleased to support the open source community by making Blueking Container Service available.
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ * Licensed under the MIT License (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ * http://opensource.org/licenses/MIT
+ * Unless required by applicable law or agreed to in writing, software distributed under
+ * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package cluster xxx
+package cluster
+
+import (
+	"context"
+
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/odm/operator"
+
+	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/constants"
+	storage "github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/proto"
+	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/storage/actions/lib"
+	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/storage/actions/v1http/cluster"
+)
+
+var (
+	clusterFeatTags = []string{constants.ClusterIDUpperTag}
+)
+
+type general struct {
+	ctx              context.Context
+	clusterID        string
+	resourceFeatList []string
+	data             map[string]interface{}
+}
+
+func (g *general) getCluCondition() *operator.Condition {
+	features := operator.M{constants.ClusterIDUpperTag: g.clusterID}
+	return operator.NewLeafCondition(operator.Eq, features)
+}
+
+func (g *general) getFeatures() operator.M {
+	features := make(operator.M)
+	features[constants.ClusterIDUpperTag] = g.clusterID
+	return features
+}
+
+func (g *general) putResources() error {
+	return cluster.PutData(g.ctx, g.data, g.getFeatures(), g.resourceFeatList)
+}
+
+func (g *general) deleteResources() ([]operator.M, error) {
+	condition := g.getCluCondition()
+	getOption := &lib.StoreGetOption{
+		Cond:           condition,
+		IsAllDocuments: true,
+	}
+	rmOption := &lib.StoreRemoveOption{
+		Cond: condition,
+		// when resource to be deleted not found, do not return error
+		IgnoreNotFound: true,
+	}
+	return cluster.DeleteBatchData(g.ctx, getOption, rmOption)
+}
+
+// HandlerCreateClusterInfoReq  CreateClusterInfoReq业务方法
+func HandlerCreateClusterInfoReq(ctx context.Context, req *storage.PutClusterInfoRequest) (operator.M, error) {
+
+	data := map[string]interface{}{
+		"clusterID": req.ClusterID,
+		"data":      req.Data.AsMap(),
+	}
+	r := &general{
+		ctx:              ctx,
+		clusterID:        req.ClusterID,
+		resourceFeatList: clusterFeatTags,
+		data:             data,
+	}
+
+	return data, r.putResources()
+}
+
+// HandlerDeleteClusterInfoReq  DeleteClusterInfoReq业务方法
+func HandlerDeleteClusterInfoReq(ctx context.Context, req *storage.DeleteClusterInfoRequest) ([]operator.M, error) {
+
+	r := &general{
+		ctx:              ctx,
+		clusterID:        req.ClusterID,
+		resourceFeatList: clusterFeatTags,
+	}
+
+	return r.deleteResources()
+}

bcs-services/bcs-storage/pkg/handler/internal/project/project.go

@@ -0,0 +1,62 @@
+/*
+ * Tencent is pleased to support the open source community by making Blueking Container Service available.
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ * Licensed under the MIT License (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ * http://opensource.org/licenses/MIT
+ * Unless required by applicable law or agreed to in writing, software distributed under
+ * the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package project xxx
+package project
+
+import (
+	"context"
+
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/odm/operator"
+
+	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/constants"
+	storage "github.com/Tencent/bk-bcs/bcs-services/bcs-storage/pkg/proto"
+	"github.com/Tencent/bk-bcs/bcs-services/bcs-storage/storage/actions/v1http/project"
+)
+
+var (
+	projectFeatTags = []string{constants.ProjectIDTag}
+)
+
+type general struct {
+	ctx              context.Context
+	projectID        string
+	resourceFeatList []string
+	data             map[string]interface{}
+}
+
+func (g *general) getFeatures() operator.M {
+	features := make(operator.M)
+	features[constants.ProjectIDTag] = g.projectID
+	return features
+}
+
+func (g *general) putResources() error {
+	return project.PutData(g.ctx, g.data, g.getFeatures(), g.resourceFeatList)
+}
+
+// HandlerCreateProjectInfoReq  CreateProjectInfoReq业务方法
+func HandlerCreateProjectInfoReq(ctx context.Context, req *storage.PutProjectInfoRequest) (operator.M, error) {
+
+	data := map[string]interface{}{
+		"projectID": req.ProjectID,
+		"data":      req.Data.AsMap(),
+	}
+	r := &general{
+		ctx:              ctx,
+		projectID:        req.ProjectID,
+		resourceFeatList: projectFeatTags,
+		data:             data,
+	}
+
+	return data, r.putResources()
+}