migrate and protobuf (#1181)

* add list api * sign log api * enrich testcase * collector v2 * collector v2 * add protobuf * sign api with api/v1 * log store v2 * fix * enrich testcase * enrich log * log query with grpc * position migrate * enrich test case * migrate collector && use erda-proto-go * remove version two * adjust LOG_LEVEL * testcase * resolve fix * import fix
erda-project · Jul 28, 2021 · 4267747 · 4267747
1 parent df6e054
commit 4267747
Show file tree

Hide file tree

Showing 62 changed files with 2,516 additions and 639 deletions.
diff --git a/apistructs/access_key.go b/apistructs/access_key.go
@@ -29,3 +29,10 @@ type AccessKeyUpdateRequest struct {
 	Status      string `json:"status" validate:"eq=ACTIVE|DISABLED"`
 	Description string `json:"description"`
 }
+
+type AccessKeyListQueryRequest struct {
+	IsSystem    *bool  `json:"isSystem"`
+	Status      string `json:"status"`
+	SubjectType string `json:"subjectType"`
+	Subject     string `json:"subject"`
+}
diff --git a/bundle/access_key.go b/bundle/access_key.go
@@ -14,6 +14,9 @@
 package bundle
 
 import (
+	"strconv"
+
+	"github.com/erda-project/erda/apistructs"
 	"github.com/erda-project/erda/bundle/apierrors"
 	"github.com/erda-project/erda/modules/core-services/model"
 	"github.com/erda-project/erda/pkg/http/httpclient"
@@ -41,3 +44,37 @@ func (b *Bundle) GetAccessKeyByAccessKeyID(ak string) (model.AccessKey, error) {
 type AkSkResponse struct {
 	Data model.AccessKey `json:"data"`
 }
+
+func (b *Bundle) ListAccessKeys(req apistructs.AccessKeyListQueryRequest) ([]model.AccessKey, error) {
+	host, err := b.urls.CoreServices()
+	if err != nil {
+		return nil, err
+	}
+	hc := b.hc
+
+	var obj AccessKeysListResponse
+	r := hc.Get(host, httpclient.RetryErrResp).Path("/api/credential/access-keys").Header("Content-Type", "application/json")
+	if req.Subject != "" {
+		r = r.Param("subject", req.Subject)
+	}
+	if req.SubjectType != "" {
+		r = r.Param("subjectType", req.SubjectType)
+	}
+	if req.Status != "" {
+		r = r.Param("status", req.Status)
+	}
+	if req.IsSystem != nil {
+		r = r.Param("isSystem", strconv.FormatBool(*req.IsSystem))
+	}
+
+	resp, err := r.Do().JSON(&obj)
+	if err != nil || !resp.IsOK() {
+		return nil, apierrors.ErrInvoke.NotFound()
+	}
+
+	return obj.Data, nil
+}
+
+type AccessKeysListResponse struct {
+	Data []model.AccessKey `json:"data"`
+}
diff --git a/cmd/monitor/collector/main.go b/cmd/monitor/collector/main.go
@@ -24,7 +24,7 @@ import (
 	_ "github.com/erda-project/erda-infra/providers/pprof"
 
 	// providers
-	_ "github.com/erda-project/erda/modules/monitor/core/collector"
+	_ "github.com/erda-project/erda/modules/core/monitor/collector"
 )
 
 //go:generate sh -c "cd ${PROJ_PATH} && go generate -v -x github.com/erda-project/erda/modules/monitor/core/collector"

diff --git a/cmd/monitor/monitor/main.go b/cmd/monitor/monitor/main.go
@@ -20,6 +20,7 @@ import (
 	"github.com/erda-project/erda/pkg/common"
 
 	// modules
+	_ "github.com/erda-project/erda/modules/core/monitor/log/query"
 	_ "github.com/erda-project/erda/modules/core/monitor/metric/index"
 	_ "github.com/erda-project/erda/modules/core/monitor/metric/query"
 	_ "github.com/erda-project/erda/modules/core/monitor/metric/query-example"
@@ -32,7 +33,6 @@ import (
 	_ "github.com/erda-project/erda/modules/monitor/apm/report"
 	_ "github.com/erda-project/erda/modules/monitor/apm/runtime"
 	_ "github.com/erda-project/erda/modules/monitor/apm/topology"
-	_ "github.com/erda-project/erda/modules/monitor/core/logs/query"
 	_ "github.com/erda-project/erda/modules/monitor/dashboard/chart-block"
 	_ "github.com/erda-project/erda/modules/monitor/dashboard/node-topo"
 	_ "github.com/erda-project/erda/modules/monitor/dashboard/org-apis"

diff --git a/cmd/monitor/streaming/main.go b/cmd/monitor/streaming/main.go
@@ -19,9 +19,10 @@ import (
 	"github.com/erda-project/erda/pkg/common"
 
 	// modules
+	_ "github.com/erda-project/erda/modules/core/monitor/log/storage"
 	_ "github.com/erda-project/erda/modules/core/monitor/metric/storage"
 	_ "github.com/erda-project/erda/modules/monitor/alert/storage/alert-record"
-	_ "github.com/erda-project/erda/modules/monitor/core/logs/storage"
+	_ "github.com/erda-project/erda/modules/monitor/notify/storage/notify-record"
 	_ "github.com/erda-project/erda/modules/msp/apm/browser"
 	_ "github.com/erda-project/erda/modules/msp/apm/trace/storage"
 
@@ -32,10 +33,6 @@ import (
 	_ "github.com/erda-project/erda-infra/providers/kafka"
 	_ "github.com/erda-project/erda-infra/providers/mysql"
 	_ "github.com/erda-project/erda-infra/providers/pprof"
-
-	// storage record
-	_ "github.com/erda-project/erda/modules/monitor/alert/storage/alert-record"
-	_ "github.com/erda-project/erda/modules/monitor/notify/storage/notify-record"
 )
 
 func main() {

diff --git a/conf/monitor/collector/collector.yaml b/conf/monitor/collector/collector.yaml
@@ -9,6 +9,7 @@ kafka:
         options:
             go.produce.channel.size: ${COLLECTOR_KAFKA_PRODUCE_CHANNEL_SIZE:100}
             queue.buffering.max.kbytes: ${COLLECTOR_KAFKA_QUEUE_SIZE_KB:524288} # 500MB
+
 monitor-collector:
     _enable: ${COLLECTOR_ENABLE:true}
     ta_sampling_rate: ${COLLECTOR_BROWSER_SAMPLING_RATE:100}
@@ -22,6 +23,8 @@ monitor-collector:
         password: "${COLLECTOR_AUTH_PASSWORD:G$9767bP32drYFPWrK4XMLRMTatiM6cU}"
         force: ${COLLECTOR_AUTH_FORCE:false}
     sign_auth:
+        sync_interval: ${COLLECTOR_AK_SYNC_INTERVAL:3m}
+        expired_duration: ${COLLECTOR_AK_EXPIRED_DURATION:10m}
 
 pprof:
 http-server@admin:

diff --git a/conf/monitor/monitor/monitor.yaml b/conf/monitor/monitor/monitor.yaml
@@ -67,8 +67,8 @@ metrics-query-compatibility:
     path: ${CONFIG_PATH}/charts
     reload_interval: "30s"
 
-logs-query:
-  _enable: ${LOGS_QUERY_ENABLE:true}
+erda.core.monitor.log.query:
+  _enable: ${LOG_QUERY_ENABLE:true}
   cassandra:
     keyspace:
       name: "spot_prod"

diff --git a/conf/monitor/streaming/streaming.yaml b/conf/monitor/streaming/streaming.yaml
@@ -44,8 +44,8 @@ etcd:
 etcd-mutex:
   root_path: "/erda/streaming"
 
-logs-store:
-  _enable: ${LOGS_STORE_ENABLE:true}
+log-store:
+  _enable: ${LOG_STORE_ENABLE:true}
   input:
     topics: "${LOG_TOPICS:spot-container-log,spot-job-log}"
     group: "${LOG_GROUP_ID:spot-monitor-log-dev}"
@@ -73,7 +73,6 @@ logs-store:
       default_ttl: "${LOG_TTL:168h}"
       gc_grace_seconds: 86400
 
-
 browser-analytics:
   _enable: ${BROWSER_ENABLE:true}
   input:

diff --git a/erda.yml b/erda.yml
@@ -374,6 +374,7 @@ services:
         protocol: "TCP"
         l4_protocol: "TCP"
     envs:
+      LOG_LEVEL: INFO
     resources:
       cpu: ${request_cpu:1}
       mem: ${request_mem:512}
@@ -628,6 +629,7 @@ services:
       TRACE_ENABLE: "true"
       TRACE_GROUP_ID: "spot-monitor-trace"
       TRACE_TTL: "168h"
+      LOG_LEVEL: INFO
     resources:
       cpu: ${request_cpu:1.5}
       mem: ${request_mem:1024}

diff --git a/go.mod b/go.mod
@@ -5,6 +5,7 @@ go 1.16
 require (
 	bou.ke/monkey v1.0.2
 	github.com/360EntSecGroup-Skylar/excelize/v2 v2.3.2
+	github.com/DATA-DOG/go-sqlmock v1.5.0
 	github.com/GoogleCloudPlatform/spark-on-k8s-operator v0.0.0-20201215015655-2e8b733f5ad0
 	github.com/Masterminds/semver v1.5.0
 	github.com/alecthomas/assert v0.0.0-20170929043011-405dbfeb8e38
@@ -156,7 +157,6 @@ require (
 )
 
 replace (
-	github.com/coreos/bbolt => go.etcd.io/bbolt v1.3.5
 	github.com/google/gnostic => github.com/googleapis/gnostic v0.4.0
 	github.com/googlecloudplatform/flink-operator => github.com/johnlanni/flink-on-k8s-operator v0.0.0-20210712093304-4d24aba33511
 	github.com/influxdata/influxql => github.com/erda-project/influxql v1.1.0-ex

diff --git a/go.sum b/go.sum
@@ -304,6 +304,8 @@ github.com/containerd/typeurl v0.0.0-20180627222232-a93fcdb778cd/go.mod h1:Cm3kw
 github.com/containerd/typeurl v1.0.1/go.mod h1:TB1hUtrpaiO88KEK56ijojHS1+NeF0izUACaJW2mdXg=
 github.com/containernetworking/cni v0.8.0/go.mod h1:LGwApLUm2FpoOfxTDEeq8T9ipbpZ61X79hmU3w8FmsY=
 github.com/coredns/corefile-migration v1.0.11/go.mod h1:RMy/mXdeDlYwzt0vdMEJvT2hGJ2I86/eO0UdXmH9XNI=
+github.com/coreos/bbolt v1.3.2 h1:wZwiHHUieZCquLkDL0B8UhzreNWsPHooDAG3q34zk0s=
+github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/bbolt v1.3.5 h1:XFv7xaq7701j8ZSEzR28VohFYSlyakMyqNMU5FQH6Ac=
 github.com/coreos/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
@@ -1713,8 +1715,6 @@ go.elastic.co/apm/module/apmhttp v1.8.0/go.mod h1:9LPFlEON51/lRbnWDfqAWErihIiAFD
 go.elastic.co/fastjson v1.0.0/go.mod h1:PmeUOMMtLHQr9ZS9J9owrAVg0FkaZDRZJEFTTGHtchs=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
-go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0=
-go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 go.etcd.io/etcd v0.0.0-20191023171146-3cf2f69b5738/go.mod h1:dnLIgRNXwCJa5e+c6mIZCrds/GIG4ncV9HhK5PX7jPg=
 go.etcd.io/etcd v0.5.0-alpha.5.0.20191023171146-3cf2f69b5738/go.mod h1:dnLIgRNXwCJa5e+c6mIZCrds/GIG4ncV9HhK5PX7jPg=
 go.etcd.io/etcd v0.5.0-alpha.5.0.20200910180754-dd1b699fc489 h1:1JFLBqwIgdyHN1ZtgjTBwO+blA6gVOmZurpiMEsETKo=

diff --git a/modules/core-services/dao/access_key.go b/modules/core-services/dao/access_key.go
@@ -55,6 +55,25 @@ func (client *DBClient) GetByAccessKeyID(ak string) (model.AccessKey, error) {
 	return obj, nil
 }
 
+func (client *DBClient) ListAccessKey(req apistructs.AccessKeyListQueryRequest) ([]model.AccessKey, error) {
+	var objs []model.AccessKey
+	query := client.Where(&model.AccessKey{
+		Status:      req.Status,
+		Subject:     req.Subject,
+		SubjectType: req.SubjectType,
+	})
+	if req.IsSystem != nil {
+		query = query.Where(map[string]interface{}{
+			"is_system": req.IsSystem,
+		})
+	}
+	res := query.Find(&objs)
+	if res.Error != nil {
+		return nil, res.Error
+	}
+	return objs, nil
+}
+
 func (client *DBClient) DeleteByAccessKeyID(ak string) error {
 	return client.Where(&model.AccessKey{AccessKeyID: ak}).Delete(&model.AccessKey{}).Error
 }
diff --git a/modules/core-services/endpoints/access_key.go b/modules/core-services/endpoints/access_key.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"encoding/json"
 	"net/http"
+	"strconv"
 
 	"github.com/pkg/errors"
 
@@ -71,6 +72,43 @@ func (e *Endpoints) GetByAccessKeyID(ctx context.Context, r *http.Request, vars
 	return httpserver.OkResp(obj)
 }
 
+func getAccessKeyListParam(r *http.Request) (apistructs.AccessKeyListQueryRequest, error) {
+	q, res := r.URL.Query(), apistructs.AccessKeyListQueryRequest{}
+
+	if v := q.Get("isSystem"); v != "" {
+		val, err := strconv.ParseBool(v)
+		if err != nil {
+			return apistructs.AccessKeyListQueryRequest{}, err
+		}
+		res.IsSystem = &val
+	}
+
+	if v := q.Get("status"); v != "" {
+		res.Status = v
+	}
+
+	if v := q.Get("subjectType"); v != "" {
+		res.SubjectType = v
+	}
+
+	if v := q.Get("subject"); v != "" {
+		res.Subject = v
+	}
+	return res, nil
+}
+
+func (e *Endpoints) ListAccessKeys(ctx context.Context, r *http.Request, vars map[string]string) (httpserver.Responser, error) {
+	req, err := getAccessKeyListParam(r)
+	if err != nil {
+		return apierrors.ErrGetAccessKey.InvalidParameter(err).ToResp(), nil
+	}
+	obj, err := e.accesskey.ListAccessKey(ctx, req)
+	if err != nil {
+		return apierrors.ErrGetAccessKey.InternalError(err).ToResp(), nil
+	}
+	return httpserver.OkResp(obj)
+}
+
 func (e *Endpoints) DeleteByAccessKeyID(ctx context.Context, r *http.Request, vars map[string]string) (httpserver.Responser, error) {
 	ak, ok := vars["accessKeyId"]
 	if !ok {

diff --git a/modules/core-services/endpoints/endpoints.go b/modules/core-services/endpoints/endpoints.go
@@ -355,6 +355,7 @@ func (e *Endpoints) Routes() []httpserver.Endpoint {
 
 		// the interface of accesskey
 		{Path: "/api/credential/access-keys/{accessKeyId}", Method: http.MethodGet, Handler: e.GetByAccessKeyID},
+		{Path: "/api/credential/access-keys", Method: http.MethodGet, Handler: e.ListAccessKeys},
 		{Path: "/api/credential/access-keys", Method: http.MethodPost, Handler: e.CreateAccessKey},
 		{Path: "/api/credential/access-keys/{accessKeyId}", Method: http.MethodPut, Handler: e.UpdateAccessKey},
 		{Path: "/api/credential/access-keys/{accessKeyId}", Method: http.MethodDelete, Handler: e.DeleteByAccessKeyID},