Merge branch 'master' of https://github.com/TencentBlueKing/bk-bcs in…

…to example-1014

bcs-scenarios/bcs-gitops-manager/cmd/gitgenerator-webhook/webhook/webhook.go

@@ -364,8 +364,8 @@ func (s *AdmissionWebhookServer) interceptAppSyncWithForbid(ctx context.Context,
 		blog.Errorf("not need intercept with forbidden flag because unmarshal application request error")
 		return nil
 	}
-	state := reqApp.Status.OperationState
-	if state == nil || state.Phase != synccommon.OperationRunning {
+	// 如果没有设置app的operation,就忽略
+	if reqApp.Operation == nil {
 		return nil
 	}
 	app, err := s.argoStore.GetApplication(ctx, req.Name)

bcs-services/bcs-bscp/cmd/data-service/app/app.go

@@ -14,6 +14,7 @@
 package app
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"net"
@@ -34,6 +35,7 @@ import (
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/cc"
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/criteria/uuid"
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/dal/dao"
+	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/dal/repository"
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/dal/vault"
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/logs"
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/metrics"
@@ -42,6 +44,8 @@ import (
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/runtime/ctl"
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/runtime/shutdown"
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/serviced"
+	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/space"
+	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/thirdparty/esb/client"
 	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/tools"
 )
 
@@ -80,12 +84,15 @@ func Run(opt *options.Option) error {
 }
 
 type dataService struct {
-	serve   *grpc.Server
-	gwServe *http.Server
-	service *service.Service
-	sd      serviced.Service
-	daoSet  dao.Set
-	vault   vault.Set
+	serve    *grpc.Server
+	gwServe  *http.Server
+	service  *service.Service
+	sd       serviced.Service
+	daoSet   dao.Set
+	vault    vault.Set
+	esb      client.Client
+	spaceMgr *space.Manager
+	repo     repository.Provider
 }
 
 // prepare do prepare jobs before run data service.
@@ -139,28 +146,61 @@ func (ds *dataService) prepare(opt *options.Option) error {
 	state := crontab.NewSyncClientOnlineState(ds.daoSet, ds.sd)
 	state.Run()
 
-	// initial Vault set
+	// initialize vault
+	if ds.vault, err = initVault(); err != nil {
+		return err
+	}
+
+	// initialize esb client
+	settings := cc.DataService().Esb
+	esbCli, err := client.NewClient(&settings, metrics.Register())
+	if err != nil {
+		return fmt.Errorf("new esb client failed, err: %v", err)
+	}
+	ds.esb = esbCli
+
+	// initialize space manager
+	spaceMgr, err := space.NewSpaceMgr(context.Background(), esbCli)
+	if err != nil {
+		return fmt.Errorf("init space manager failed, err: %v", err)
+	}
+	ds.spaceMgr = spaceMgr
+
+	// initialize repo provider
+	repo, err := repository.NewProvider(cc.DataService().Repo)
+	if err != nil {
+		return fmt.Errorf("new repo provider failed, err: %v", err)
+	}
+	ds.repo = repo
+
+	// sync files from master to slave repo
+	if cc.DataService().Repo.EnableHA {
+		repoSyncer := service.NewRepoSyncer(ds.daoSet, ds.repo, ds.spaceMgr, ds.sd)
+		repoSyncer.Run()
+	}
+
+	return nil
+}
+
+func initVault() (vault.Set, error) {
 	vaultSet, err := vault.NewSet(cc.DataService().Vault)
 	if err != nil {
-		return fmt.Errorf("initial vault set failed, err: %v", err)
+		return nil, fmt.Errorf("initial vault set failed, err: %v", err)
 	}
 	// 挂载目录
 	exists, err := vaultSet.IsMountPathExists(vault.MountPath)
 	if err != nil {
-		return fmt.Errorf("error checking mount path: %v", err)
+		return nil, fmt.Errorf("error checking mount path: %v", err)
 	}
 	if !exists {
 		mountConfig := &api.MountInput{
 			Type: "kv-v2",
 		}
 		if err = vaultSet.CreateMountPath(vault.MountPath, mountConfig); err != nil {
-			return fmt.Errorf("initial vault mount path failed, err: %v", err)
+			return nil, fmt.Errorf("initial vault mount path failed, err: %v", err)
 		}
 	}
-
-	ds.vault = vaultSet
-
-	return nil
+	return vaultSet, nil
 }
 
 // listenAndServe listen the grpc serve and set up the shutdown gracefully job.
@@ -197,7 +237,7 @@ func (ds *dataService) listenAndServe() error {
 	}
 
 	serve := grpc.NewServer(opts...)
-	svc, err := service.NewService(ds.sd, ds.daoSet, ds.vault)
+	svc, err := service.NewService(ds.sd, ds.daoSet, ds.vault, ds.esb, ds.repo)
 	if err != nil {
 		return err
 	}

bcs-services/bcs-bscp/cmd/data-service/service/metric.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 service
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/TencentBlueKing/bk-bcs/bcs-services/bcs-bscp/pkg/metrics"
+)
+
+var (
+	metricInstance *metric
+	once           sync.Once
+)
+
+func initMetric() *metric {
+	once.Do(func() {
+		m := new(metric)
+		labels := prometheus.Labels{}
+		m.syncQueueLen = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace:   metrics.Namespace,
+			Subsystem:   metrics.RepoSyncSubSys,
+			Name:        "sync_queue_len",
+			Help:        "the length of sync queue for repo sync",
+			ConstLabels: labels,
+		})
+		metrics.Register().MustRegister(m.syncQueueLen)
+
+		m.ackQueueLen = prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace:   metrics.Namespace,
+			Subsystem:   metrics.RepoSyncSubSys,
+			Name:        "ack_queue_len",
+			Help:        "the length of ack queue for repo sync",
+			ConstLabels: labels,
+		})
+		metrics.Register().MustRegister(m.ackQueueLen)
+
+		metricInstance = m
+
+	})
+	return metricInstance
+}
+
+type metric struct {
+	// syncQueueLen records the length of sync queue for repo sync
+	syncQueueLen prometheus.Gauge
+
+	// ackQueueLen records the length of ack queue for repo sync
+	ackQueueLen prometheus.Gauge
+}