Add batch clean instances maintian interface

apiserver/httpserver/http/handler.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"strings"
 
@@ -243,3 +244,15 @@ func ParseQueryParams(req *restful.Request) map[string]string {
 
 	return queryParams
 }
+
+// ParseJsonBody parse http body as json object
+func ParseJsonBody(req *restful.Request, value interface{}) error {
+	body, err := ioutil.ReadAll(req.Request.Body)
+	if err != nil {
+		return err
+	}
+	if err := json.Unmarshal(body, value); err != nil {
+		return err
+	}
+	return nil
+}

apiserver/httpserver/http/handler_test.go

@@ -18,8 +18,10 @@
 package http
 
 import (
+	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/emicklei/go-restful/v3"
@@ -73,3 +75,27 @@ func Test_i18n(t *testing.T) {
 		}
 	}
 }
+
+func Test_ParseJsonBody(t *testing.T) {
+	type TestJsonObject struct {
+		Text string `json:"text"`
+	}
+
+	expectText := "this is a test"
+
+	httpReq, _ := http.NewRequest(
+		http.MethodPost,
+		"http://example.com",
+		strings.NewReader(fmt.Sprintf("{\"text\": \"%s\"}", expectText)))
+	req := restful.NewRequest(httpReq)
+
+	testResult := TestJsonObject{}
+	err := ParseJsonBody(req, &testResult)
+	if err != nil {
+		t.Errorf("ParseJsonBody err %v, want %v", err, expectText)
+	}
+	if testResult.Text != expectText {
+		t.Errorf("ParseJsonBody = %v, want %v", testResult.Text, expectText)
+	}
+
+}

apiserver/httpserver/maintain_access.go

@@ -20,7 +20,6 @@ package httpserver
 import (
 	"context"
 	"encoding/json"
-	"io/ioutil"
 	"net/http"
 	"strconv"
 
@@ -42,6 +41,7 @@ func (h *HTTPServer) GetMaintainAccessServer() *restful.WebService {
 	ws.Route(enrichCloseConnectionsApiDocs(ws.POST("apiserver/conn/close").To(h.CloseConnections)))
 	ws.Route(enrichFreeOSMemoryApiDocs(ws.POST("/memory/free").To(h.FreeOSMemory)))
 	ws.Route(enrichCleanInstanceApiDocs(ws.POST("/instance/clean").To(h.CleanInstance)))
+	ws.Route(enrichBatchCleanInstancesApiDocs(ws.POST("/instance/batchclean").To(h.BatchCleanInstances)))
 	ws.Route(enrichGetLastHeartbeatApiDocs(ws.GET("/instance/heartbeat").To(h.GetLastHeartbeat)))
 	ws.Route(enrichGetLogOutputLevelApiDocs(ws.GET("/log/outputlevel").To(h.GetLogOutputLevel)))
 	ws.Route(enrichSetLogOutputLevelApiDocs(ws.PUT("/log/outputlevel").To(h.SetLogOutputLevel)))
@@ -149,6 +149,30 @@ func (h *HTTPServer) CleanInstance(req *restful.Request, rsp *restful.Response)
 	handler.WriteHeaderAndProto(h.maintainServer.CleanInstance(ctx, instance))
 }
 
+func (h *HTTPServer) BatchCleanInstances(req *restful.Request, rsp *restful.Response) {
+	ctx := initContext(req)
+
+	var param struct {
+		BatchSize uint32 `json:"batch_size"`
+	}
+
+	if err := httpcommon.ParseJsonBody(req, &param); err != nil {
+		_ = rsp.WriteError(http.StatusBadRequest, err)
+		return
+	}
+
+	if count, err := h.maintainServer.BatchCleanInstances(ctx, param.BatchSize); err != nil {
+		_ = rsp.WriteError(http.StatusInternalServerError, err)
+	} else {
+		var ret struct {
+			RowsAffected uint32 `json:"rows_affected"`
+		}
+		ret.RowsAffected = count
+		_ = rsp.WriteAsJson(ret)
+	}
+
+}
+
 // GetLastHeartbeat 获取实例，上一次心跳的时间
 func (h *HTTPServer) GetLastHeartbeat(req *restful.Request, rsp *restful.Response) {
 	ctx := initContext(req)
@@ -193,12 +217,8 @@ func (h *HTTPServer) SetLogOutputLevel(req *restful.Request, rsp *restful.Respon
 		Scope string `json:"scope"`
 		Level string `json:"level"`
 	}
-	body, err := ioutil.ReadAll(req.Request.Body)
-	if err != nil {
-		_ = rsp.WriteErrorString(http.StatusBadRequest, err.Error())
-		return
-	}
-	if err := json.Unmarshal(body, &scopeLogLevel); err != nil {
+
+	if err := httpcommon.ParseJsonBody(req, &scopeLogLevel); err != nil {
 		_ = rsp.WriteErrorString(http.StatusBadRequest, err.Error())
 		return
 	}

apiserver/httpserver/maintain_apidoc.go

@@ -71,6 +71,13 @@ func enrichCleanInstanceApiDocs(r *restful.RouteBuilder) *restful.RouteBuilder {
 		Notes(enrichCleanInstanceApiNotes)
 }
 
+func enrichBatchCleanInstancesApiDocs(r *restful.RouteBuilder) *restful.RouteBuilder {
+	return r.
+		Doc("彻底清理flag=1的实例").
+		Metadata(restfulspec.KeyOpenAPITags, maintainApiTags).
+		Notes(enrichBatchCleanInstancesApiNotes)
+}
+
 func enrichGetLastHeartbeatApiDocs(r *restful.RouteBuilder) *restful.RouteBuilder {
 	return r.
 		Doc("获取上一次心跳的时间").

apiserver/httpserver/maintain_apinotes.go

@@ -107,6 +107,20 @@ Header Content-Type: application/json
 }
 ~~~
 
+`
+	enrichBatchCleanInstancesApiNotes = `
+请求示例：
+
+~~~
+POST /maintain/v1/instance/batchclean
+Header X-Polaris-Token: {访问凭据}
+Header Content-Type: application/json
+
+{
+    "batch_size": 100
+}
+~~~
+
 `
 	enrichGetLastHeartbeatApiNotes = `
 请求示例：

bootstrap/server.go

@@ -190,7 +190,7 @@ func StartComponents(ctx context.Context, cfg *boot_config.Config) error {
 	}
 
 	// 初始化运维操作模块
-	if err := maintain.Initialize(ctx, namingSvr, healthCheckServer); err != nil {
+	if err := maintain.Initialize(ctx, namingSvr, healthCheckServer, s); err != nil {
 		return err
 	}
 

maintain/api.go

@@ -63,6 +63,9 @@ type MaintainOperateServer interface {
 	// CleanInstance Clean deleted instance
 	CleanInstance(ctx context.Context, req *api.Instance) *api.Response
 
+	// BatchCleanInstances Batch clean deleted instances
+	BatchCleanInstances(ctx context.Context, batchSize uint32) (uint32, error)
+
 	// GetLastHeartbeat Get last heartbeat
 	GetLastHeartbeat(ctx context.Context, req *api.Instance) *api.Response
 

maintain/default.go

@@ -24,6 +24,7 @@ import (
 	"github.com/polarismesh/polaris/auth"
 	"github.com/polarismesh/polaris/service"
 	"github.com/polarismesh/polaris/service/healthcheck"
+	"github.com/polarismesh/polaris/store"
 )
 
 var (
@@ -33,12 +34,12 @@ var (
 )
 
 // Initialize 初始化
-func Initialize(ctx context.Context, namingService service.DiscoverServer, healthCheckServer *healthcheck.Server) error {
+func Initialize(ctx context.Context, namingService service.DiscoverServer, healthCheckServer *healthcheck.Server, storage store.Store) error {
 	if finishInit {
 		return nil
 	}
 
-	err := initialize(ctx, namingService, healthCheckServer)
+	err := initialize(ctx, namingService, healthCheckServer, storage)
 	if err != nil {
 		return err
 	}
@@ -47,14 +48,15 @@ func Initialize(ctx context.Context, namingService service.DiscoverServer, healt
 	return nil
 }
 
-func initialize(_ context.Context, namingService service.DiscoverServer, healthCheckServer *healthcheck.Server) error {
+func initialize(_ context.Context, namingService service.DiscoverServer, healthCheckServer *healthcheck.Server, storage store.Store) error {
 	authServer, err := auth.GetAuthServer()
 	if err != nil {
 		return err
 	}
 
 	maintainServer.namingServer = namingService
 	maintainServer.healthCheckServer = healthCheckServer
+	maintainServer.storage = storage
 
 	server = newServerAuthAbility(maintainServer, authServer)
 	return nil

maintain/maintain.go

@@ -135,6 +135,10 @@ func (s *Server) CleanInstance(ctx context.Context, req *api.Instance) *api.Resp
 	return s.namingServer.CleanInstance(ctx, req)
 }
 
+func (s *Server) BatchCleanInstances(ctx context.Context, batchSize uint32) (uint32, error) {
+	return s.storage.BatchCleanDeletedInstances(batchSize)
+}
+
 func (s *Server) GetLastHeartbeat(_ context.Context, req *api.Instance) *api.Response {
 	return s.healthCheckServer.GetLastHeartbeat(req)
 }

maintain/maintain_authability.go

@@ -76,6 +76,16 @@ func (svr *serverAuthAbility) CleanInstance(ctx context.Context, req *api.Instan
 	return svr.targetServer.CleanInstance(ctx, req)
 }
 
+func (svr *serverAuthAbility) BatchCleanInstances(ctx context.Context, batchSize uint32) (uint32, error) {
+	authCtx := svr.collectMaintainAuthContext(ctx, model.Delete, "BatchCleanInstances")
+	_, err := svr.authMgn.CheckConsolePermission(authCtx)
+	if err != nil {
+		return 0, err
+	}
+
+	return svr.targetServer.BatchCleanInstances(ctx, batchSize)
+}
+
 func (svr *serverAuthAbility) GetLastHeartbeat(ctx context.Context, req *api.Instance) *api.Response {
 	authCtx := svr.collectMaintainAuthContext(ctx, model.Read, "GetLastHeartbeat")
 	_, err := svr.authMgn.CheckConsolePermission(authCtx)

maintain/server.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/polarismesh/polaris/service"
 	"github.com/polarismesh/polaris/service/healthcheck"
+	"github.com/polarismesh/polaris/store"
 )
 
 var _ MaintainOperateServer = (*Server)(nil)
@@ -30,4 +31,5 @@ type Server struct {
 	mu                sync.Mutex
 	namingServer      service.DiscoverServer
 	healthCheckServer *healthcheck.Server
+	storage           store.Store
 }

store/api.go

@@ -51,6 +51,9 @@ type Store interface {
 
 	// ClientStore Client the central module storage interface
 	ClientStore
+
+	// MaintainStore Maintain inteface
+	MaintainStore
 }
 
 // NamespaceStore Namespace storage interface

store/boltdb/default.go

@@ -64,6 +64,9 @@ type boltStore struct {
 	// v2 存储
 	*routingStoreV2
 
+	// maintain store
+	*maintainStore
+
 	handler BoltHandler
 	start   bool
 }
@@ -261,6 +264,10 @@ func (m *boltStore) newStore() error {
 		return err
 	}
 
+	if err := m.newMaintainModuleStore(); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -326,6 +333,12 @@ func (m *boltStore) newConfigModuleStore() error {
 	return nil
 }
 
+func (m *boltStore) newMaintainModuleStore() error {
+	m.maintainStore = &maintainStore{handler: m.handler}
+
+	return nil
+}
+
 // Destroy store
 func (m *boltStore) Destroy() error {
 	m.start = false

store/boltdb/maintain.go

@@ -0,0 +1,27 @@
+/**
+ * Tencent is pleased to support the open source community by making Polaris available.
+ *
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ *
+ * Licensed under the BSD 3-Clause License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://opensource.org/licenses/BSD-3-Clause
+ *
+ * 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 boltdb
+
+type maintainStore struct {
+	handler BoltHandler
+}
+
+// BatchCleanDeletedInstances
+func (m *maintainStore) BatchCleanDeletedInstances(batchSize uint32) (uint32, error) {
+	return 0, nil
+}

store/maintain_api.go

@@ -0,0 +1,24 @@
+/**
+ * Tencent is pleased to support the open source community by making Polaris available.
+ *
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ *
+ * Licensed under the BSD 3-Clause License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://opensource.org/licenses/BSD-3-Clause
+ *
+ * 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 store
+
+type MaintainStore interface {
+
+	// BatchCleanDeletedInstances batch clean soft deleted instances
+	BatchCleanDeletedInstances(batchSize uint32) (uint32, error)
+}

store/mock/README.md

@@ -5,5 +5,5 @@
 在`./store`目录执行
 
 ```
-mockgen -source=api.go -aux_files github.com/polarismesh/polaris-server/store=config_file_api.go,github.com/polarismesh/polaris-server/store=discover_api.go,github.com/polarismesh/polaris-server/store=auth_api.go -destination=mock/api_mock.go -package=mock
-```
+mockgen -source=api.go -aux_files github.com/polarismesh/polaris/store=config_file_api.go,github.com/polarismesh/polaris/store=discover_api.go,github.com/polarismesh/polaris/store=auth_api.go,github.com/polarismesh/polaris/store=maintain_api.go -destination=mock/api_mock.go -package=mock
+```