Add auth for naming rpc api and ix some bugs for 2.0.0 (#4675)

* Add auth for naming rpc api

* Fix Callback do not execute problem

* Fix client distro do not retry problem.

common/src/main/java/com/alibaba/nacos/common/remote/client/grpc/GrpcClient.java

@@ -241,7 +241,7 @@ TimeUnit.SECONDS, new SynchronousQueue(),
 
                 BiRequestStreamGrpc.BiRequestStreamStub biRequestStreamStub = BiRequestStreamGrpc
                         .newStub(newChannelStubTemp.getChannel());
-                GrpcConnection grpcConn = new GrpcConnection(serverInfo, super.executor);
+                GrpcConnection grpcConn = new GrpcConnection(serverInfo, executor);
 
                 //create stream request and bind connection event to this connection.
                 StreamObserver<Payload> payloadStreamObserver = bindRequestStream(biRequestStreamStub, grpcConn);

naming/src/main/java/com/alibaba/nacos/naming/consistency/ephemeral/distro/v2/DistroClientComponentRegistry.java

@@ -21,8 +21,10 @@
 import com.alibaba.nacos.core.distributed.distro.DistroProtocol;
 import com.alibaba.nacos.core.distributed.distro.component.DistroComponentHolder;
 import com.alibaba.nacos.core.distributed.distro.component.DistroTransportAgent;
+import com.alibaba.nacos.core.distributed.distro.task.DistroTaskEngineHolder;
 import com.alibaba.nacos.naming.core.v2.client.manager.ClientManager;
 import com.alibaba.nacos.naming.core.v2.client.manager.ClientManagerDelegate;
+import com.alibaba.nacos.naming.misc.GlobalConfig;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
@@ -41,16 +43,22 @@ public class DistroClientComponentRegistry {
 
     private final DistroComponentHolder componentHolder;
 
+    private final DistroTaskEngineHolder taskEngineHolder;
+
+    private final GlobalConfig globalConfig;
+
     private final ClientManager clientManager;
 
     private final ClusterRpcClientProxy clusterRpcClientProxy;
 
     public DistroClientComponentRegistry(ServerMemberManager serverMemberManager, DistroProtocol distroProtocol,
-            DistroComponentHolder componentHolder, ClientManagerDelegate clientManager,
-            ClusterRpcClientProxy clusterRpcClientProxy) {
+            DistroComponentHolder componentHolder, DistroTaskEngineHolder taskEngineHolder, GlobalConfig globalConfig,
+            ClientManagerDelegate clientManager, ClusterRpcClientProxy clusterRpcClientProxy) {
         this.serverMemberManager = serverMemberManager;
         this.distroProtocol = distroProtocol;
         this.componentHolder = componentHolder;
+        this.taskEngineHolder = taskEngineHolder;
+        this.globalConfig = globalConfig;
         this.clientManager = clientManager;
         this.clusterRpcClientProxy = clusterRpcClientProxy;
     }
@@ -64,8 +72,11 @@ public void doRegister() {
         DistroClientDataProcessor dataProcessor = new DistroClientDataProcessor(clientManager, distroProtocol);
         DistroTransportAgent transportAgent = new DistroClientTransportAgent(clusterRpcClientProxy,
                 serverMemberManager);
+        DistroClientTaskFailedHandler taskFailedHandler = new DistroClientTaskFailedHandler(globalConfig,
+                taskEngineHolder);
         componentHolder.registerDataStorage(DistroClientDataProcessor.TYPE, dataProcessor);
         componentHolder.registerDataProcessor(dataProcessor);
         componentHolder.registerTransportAgent(DistroClientDataProcessor.TYPE, transportAgent);
+        componentHolder.registerFailedTaskHandler(DistroClientDataProcessor.TYPE, taskFailedHandler);
     }
 }

naming/src/main/java/com/alibaba/nacos/naming/consistency/ephemeral/distro/v2/DistroClientTaskFailedHandler.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 1999-2020 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 com.alibaba.nacos.naming.consistency.ephemeral.distro.v2;
+
+import com.alibaba.nacos.consistency.DataOperation;
+import com.alibaba.nacos.core.distributed.distro.component.DistroFailedTaskHandler;
+import com.alibaba.nacos.core.distributed.distro.entity.DistroKey;
+import com.alibaba.nacos.core.distributed.distro.task.DistroTaskEngineHolder;
+import com.alibaba.nacos.core.distributed.distro.task.delay.DistroDelayTask;
+import com.alibaba.nacos.naming.misc.GlobalConfig;
+
+/**
+ * Distro client task failed handler.
+ *
+ * @author xiweng.yy
+ */
+public class DistroClientTaskFailedHandler implements DistroFailedTaskHandler {
+
+    private final GlobalConfig globalConfig;
+
+    private final DistroTaskEngineHolder distroTaskEngineHolder;
+
+    public DistroClientTaskFailedHandler(GlobalConfig globalConfig, DistroTaskEngineHolder distroTaskEngineHolder) {
+        this.globalConfig = globalConfig;
+        this.distroTaskEngineHolder = distroTaskEngineHolder;
+    }
+
+    @Override
+    public void retry(DistroKey distroKey, DataOperation action) {
+        DistroDelayTask retryTask = new DistroDelayTask(distroKey, action, globalConfig.getSyncRetryDelay());
+        distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKey, retryTask);
+    }
+}

naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/InstanceRequestHandler.java

@@ -21,9 +21,12 @@
 import com.alibaba.nacos.api.naming.remote.request.InstanceRequest;
 import com.alibaba.nacos.api.naming.remote.response.InstanceResponse;
 import com.alibaba.nacos.api.remote.request.RequestMeta;
+import com.alibaba.nacos.auth.annotation.Secured;
+import com.alibaba.nacos.auth.common.ActionTypes;
 import com.alibaba.nacos.core.remote.RequestHandler;
 import com.alibaba.nacos.naming.core.v2.pojo.Service;
 import com.alibaba.nacos.naming.core.v2.service.impl.EphemeralClientOperationServiceImpl;
+import com.alibaba.nacos.naming.web.NamingResourceParser;
 import org.springframework.stereotype.Component;
 
 /**
@@ -41,6 +44,7 @@ public InstanceRequestHandler(EphemeralClientOperationServiceImpl clientOperatio
     }
 
     @Override
+    @Secured(action = ActionTypes.WRITE, parser = NamingResourceParser.class)
     public InstanceResponse handle(InstanceRequest request, RequestMeta meta) throws NacosException {
         Service service = Service
                 .newService(request.getNamespace(), request.getGroupName(), request.getServiceName(), true);

naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/ServiceListRequestHandler.java

@@ -20,10 +20,13 @@
 import com.alibaba.nacos.api.naming.remote.request.ServiceListRequest;
 import com.alibaba.nacos.api.naming.remote.response.ServiceListResponse;
 import com.alibaba.nacos.api.remote.request.RequestMeta;
+import com.alibaba.nacos.auth.annotation.Secured;
+import com.alibaba.nacos.auth.common.ActionTypes;
 import com.alibaba.nacos.core.remote.RequestHandler;
 import com.alibaba.nacos.naming.core.v2.ServiceManager;
 import com.alibaba.nacos.naming.core.v2.pojo.Service;
 import com.alibaba.nacos.naming.utils.ServiceUtil;
+import com.alibaba.nacos.naming.web.NamingResourceParser;
 import org.springframework.stereotype.Component;
 
 import java.util.Collection;
@@ -41,6 +44,7 @@
 public class ServiceListRequestHandler extends RequestHandler<ServiceListRequest, ServiceListResponse> {
 
     @Override
+    @Secured(action = ActionTypes.READ, parser = NamingResourceParser.class)
     public ServiceListResponse handle(ServiceListRequest request, RequestMeta meta) throws NacosException {
         Collection<Service> serviceSet = ServiceManager.getInstance().getSingletons(request.getNamespace());
         ServiceListResponse result = ServiceListResponse.buildSuccessResponse(0, new LinkedList<>());

naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/ServiceQueryRequestHandler.java

@@ -21,10 +21,13 @@
 import com.alibaba.nacos.api.naming.remote.request.ServiceQueryRequest;
 import com.alibaba.nacos.api.naming.remote.response.QueryServiceResponse;
 import com.alibaba.nacos.api.remote.request.RequestMeta;
+import com.alibaba.nacos.auth.annotation.Secured;
+import com.alibaba.nacos.auth.common.ActionTypes;
 import com.alibaba.nacos.core.remote.RequestHandler;
 import com.alibaba.nacos.naming.core.v2.index.ServiceStorage;
 import com.alibaba.nacos.naming.core.v2.pojo.Service;
 import com.alibaba.nacos.naming.utils.ServiceUtil;
+import com.alibaba.nacos.naming.web.NamingResourceParser;
 import org.springframework.stereotype.Component;
 
 /**
@@ -42,6 +45,7 @@ public ServiceQueryRequestHandler(ServiceStorage serviceStorage) {
     }
 
     @Override
+    @Secured(action = ActionTypes.READ, parser = NamingResourceParser.class)
     public QueryServiceResponse handle(ServiceQueryRequest request, RequestMeta meta) throws NacosException {
         String namespaceId = request.getNamespace();
         String groupName = request.getGroupName();

naming/src/main/java/com/alibaba/nacos/naming/remote/rpc/handler/SubscribeServiceRequestHandler.java

@@ -23,13 +23,16 @@
 import com.alibaba.nacos.api.naming.utils.NamingUtils;
 import com.alibaba.nacos.api.remote.request.RequestMeta;
 import com.alibaba.nacos.api.remote.response.ResponseCode;
+import com.alibaba.nacos.auth.annotation.Secured;
+import com.alibaba.nacos.auth.common.ActionTypes;
 import com.alibaba.nacos.common.utils.StringUtils;
 import com.alibaba.nacos.core.remote.RequestHandler;
 import com.alibaba.nacos.naming.core.v2.index.ServiceStorage;
 import com.alibaba.nacos.naming.core.v2.pojo.Service;
 import com.alibaba.nacos.naming.core.v2.service.impl.EphemeralClientOperationServiceImpl;
 import com.alibaba.nacos.naming.pojo.Subscriber;
 import com.alibaba.nacos.naming.utils.ServiceUtil;
+import com.alibaba.nacos.naming.web.NamingResourceParser;
 import org.springframework.stereotype.Component;
 
 /**
@@ -52,6 +55,7 @@ public SubscribeServiceRequestHandler(ServiceStorage serviceStorage,
     }
 
     @Override
+    @Secured(action = ActionTypes.READ, parser = NamingResourceParser.class)
     public SubscribeServiceResponse handle(SubscribeServiceRequest request, RequestMeta meta) throws NacosException {
         String namespaceId = request.getNamespace();
         String serviceName = request.getServiceName();