4. Server startup of source code analysis of Nacos configuration center

In the last article we used -Dnacos.standalone=true Started locally Nacos The server , And you can http://localhost:8848/nacos adopt nacos/nacos User name and password can access nacos Control page . Now let's take a general look Nacos What core things did you do when you started .

1、Nacos Certification services

Nacos Medium nacos-console The project relies on the following modules related to the configuration center ：

• nacos-config： Configure the central project module
• nacos-plugin-default-impl：Nacos Plug in module ： Mainly user authentication and authorization related operations , the Spring Security To do security certification . The related configuration classes are ：NacosAuthConfig

1.1 Memory database loading data

JVM Start parameter add -Dnacos.standalone=true, At this time in DynamicDataSource#getDataSource Will initialize LocalDataSourceServiceImpl.

DynamicDataSource#getDataSource()

    public synchronized DataSourceService getDataSource() {
    
        try {
    
            
            // Embedded storage is used by default in stand-alone mode
            // In cluster mode, external databases are used by default
            
            if (PropertyUtil.isEmbeddedStorage()) {
    
                if (localDataSourceService == null) {
    
                    localDataSourceService = new LocalDataSourceServiceImpl();
                    localDataSourceService.init();
                }
                return localDataSourceService;
            } else {
    
                if (basicDataSourceService == null) {
    
                    basicDataSourceService = new ExternalDataSourceServiceImpl();
                    basicDataSourceService.init();
                }
                return basicDataSourceService;
            }
        } catch (Exception e) {
    
            throw new RuntimeException(e);
        }
    }

LocalDataSourceServiceImpl#initialize Will use derby This memory database is used to store data . The address for saving data is ：${user.home}/nacos/data/derby-data .

And then call LocalDataSourceServiceImpl#reload load console/resources/META-INF/schema.sql The data in it is transferred to the memory database . Among them, we log in and use nacos/nacos User name, password

1.2 nacos plugin Expose user authentication

stay plugin-default-impl Module com.alibaba.nacos.plugin.auth.impl.controller The package contains the following Http service ：

• PermissionController： Permission Operation Service ： Get all permissions 、 Add role permissions 、 Delete role permissions and other interfaces
• RoleController： Role operation service ： Get all roles 、 Query roles 、 Adding roles 、 Delete interfaces such as roles
• UserController： User operation service ： Include creating users 、 Delete user 、 Modify the user 、 Get all users 、 User login and other interfaces

Nacos The use of Spring Seurity Do permission management , The default support Nacos Type permissions are database management permissions and Ldap Rights management . The database is used by default RBAC Rights management . Spring Security Web Configuration class is ：NacosAuthConfig.

1.3 User login sequence diagram

That's all Nacos Login authentication sequence diagram . The specific information of the user will be cached in ConcurrentHashMap among . adopt Spring Timer NacosUserDetailsServiceImpl#reload Periodically refresh the data in the cache from the database . Because we will call the following when initializing the database SQL Statement to save data to an in memory database derby among .

CREATE TABLE users (
	username varchar(50) NOT NULL PRIMARY KEY,
	password varchar(500) NOT NULL,
	enabled boolean NOT NULL DEFAULT true
);

INSERT INTO users (username, password, enabled) VALUES 
('nacos', '$2a$10$EuWPZHzz32dJN7jexM34MOeYirDdFAZm2kuWj7VEOJhhZkDrxfvUu', TRUE);

This data is loaded into NacosUserDetailsServiceImpl The cache of Map in , So we can use it nacos/nacos Login with user name and password .

2、 expose Http Service to the console

Let's talk about console What is exposed after the service is started http Service to the console , Here we exclude naming Services only discuss configuration services rest service .

2.1 console service

So let's see console In service ConsoleConfig This configuration class .

ConsoleConfig.java

@Component
@EnableScheduling
@PropertySource("/application.properties")
public class ConsoleConfig {
    
    
    @Autowired
    private ControllerMethodsCache methodsCache;
    
    /** * Init. */
    @PostConstruct
    public void init() {
    
        methodsCache.initClassMethod("com.alibaba.nacos.core.controller");
        methodsCache.initClassMethod("com.alibaba.nacos.naming.controllers");
        methodsCache.initClassMethod("com.alibaba.nacos.config.server.controller");
        methodsCache.initClassMethod("com.alibaba.nacos.console.controller");
    }
    
    @Bean
    public CorsFilter corsFilter() {
    
        CorsConfiguration config = new CorsConfiguration();
        config.setAllowCredentials(true);
        config.addAllowedOrigin("*");
        config.addAllowedHeader("*");
        config.setMaxAge(18000L);
        config.addAllowedMethod("*");
        UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
        source.registerCorsConfiguration("/**", config);
        return new CorsFilter(source);
    }
    
    @Bean
    public XssFilter xssFilter() {
    
        return new XssFilter();
    }
    
    @Bean
    public Jackson2ObjectMapperBuilderCustomizer jacksonObjectMapperCustomization() {
    
        return jacksonObjectMapperBuilder -> jacksonObjectMapperBuilder.timeZone(ZoneId.systemDefault().toString());
    }
}

First configure two Filter filter ：CorsFilter To solve the problem console The front and back end share interface calls cross domain problems ;XssFilter To solve the problem of cross domain attack .

ConsoleConfig#init The method is to parse the incoming package Controller in @RequestMapping Method cache http Ask for something to do with @ReqeustMapping Method mapping . In this way Filter It can filter out illegal requests .

Let's analyze console Under the module of Controller：

• HealthController：Nacos Healthy operation , such as ： Service health testing , see Config And Naming Whether to provide external services .
• NamespaceController：Nacos Namespace related operations , such as ： Get all namespace information 、 Get namespace details 、 Create a namespace 、 Modify namespace 、 Delete namespaces, etc .
• ServerStateController：Nacos Operation of service status , such as ： View service status

2.2 core service

Let's analyze core Under the module of Controller：

• NacosClusterController：Nacos Cluster related operations , such as ： Get your own node information , Get all node information 、 Get all node addresses 、 Check whether the current node is healthy .
• ServerLoaderController：Nacos Connection related operations , such as ： Get the of the current service Client Connection information 、 Get the service status of the current service 、 The current service reconnects the client 、 Get current service indicators, etc .
• CoreOpsController：Nacos Basic operation , such as ： perform raft command 、 obtain ID Generate rule related information 、 Set the log level
• CoreOpsV2Controller：Nacos Basic operation v2, such as ： perform raft command 、 obtain ID Generate rule related information 、 Set the log level
• NacosClusterV2Controller：Nacos Cluster related operations v2, such as ： Get your own node information , Get all node information 、 Get all node addresses, etc .

2.3 config service

Let's analyze config Under the module of Controller：

• CapacityController： Get the configuration capability of the tenant 、 Modify the configuration capability of the tenant
• ClientMetricsController： Get the indicator information of the configuration center cluster 、 Get the indicators of all clients of the current node
• CommunicationController： Save the specified configuration to a local file 、 obtain grpc Connection information 、 obtain http Long rotation training status
• ConfigController： Publish configuration information 、 Get configuration information 、 Delete configuration information 、 Query configuration information
• ConfigOpsController： Save all current configurations to local files 、 Set the log level 、 Use derby Memory database dynamic incoming SQL operation 、 Dynamically upload configuration through files
• HealthController： Judge the health of the current node configuration center
• HistoryController： Paging query configuration history 、ID Query the history details of the configuration 、ID Query the last history details of the configuration 、 Query the configured history according to the namespace
• ListenerController： according to IP Get all order information

3、 expose Grpc Services to clients

stay console When the module starts, it will start two GRPC Server side ： One is SDK Use ; One Cluster Use . The two exposed services are the same, but the exposed ports are different ：

• SDK service ： Exposed port 9848,Nacos Start port of 8848 + 1000(GrpcSdkServer#rpcPortOffset),
• Cluster service ： Exposed port 9849,Nacos Start port of 8848 + 1001(GrpcClusterServer#rpcPortOffset),

GRPC The class structure is as follows ：

• GrpcClusterServer：Cluster Offset port 、 obtain Cluster RPC Thread pool
• GrpcSdkServer：Sdk Offset port 、 obtain Sdk RPC Thread pool
• BaseGrpcServer： Specific implementation of service startup
• BaseRpcServer：RPC The base class to implement , register Payload(GRPC Request response class )、RPC Service startup

3.1 register Payload

Nacos exposed Grpc The request parameters and response parameters of the service are both base classes Payload, In the base class BaseRpcServer Pass through static Code block registration Payload.

BaseRpcServer.java

    static {
    
        PayloadRegistry.init();
    }

adopt Java Of SPI The mechanisms are respectively client Module loaded to ClientPayloadPackageProvider as well as config Load into ConfigPayloadPackageProvider.

client/META-INF/services/com.alibaba.nacos.common.remote.PayloadPackageProvider

public class ClientPayloadPackageProvider implements PayloadPackageProvider {
    
    
    private final Set<String> scanPackage = new HashSet<>();
    
    {
    
        scanPackage.add("com.alibaba.nacos.api.naming.remote.request");
        scanPackage.add("com.alibaba.nacos.api.remote.request");
        scanPackage.add("com.alibaba.nacos.api.config.remote.request");
        scanPackage.add("com.alibaba.nacos.api.naming.remote.response");
        scanPackage.add("com.alibaba.nacos.api.config.remote.response");
        scanPackage.add("com.alibaba.nacos.api.remote.response");
    }
    
    @Override
    public Set<String> getScanPackage() {
    
        return scanPackage;
    }
}

Here is config Configuration of ：

config/META-INF/services/com.alibaba.nacos.common.remote.PayloadPackageProvider

public class ConfigPayloadPackageProvider implements PayloadPackageProvider {
    
    
    private final Set<String> scanPackage = new HashSet<>();
    
    {
    
        scanPackage.add("com.alibaba.nacos.api.remote.request");
        scanPackage.add("com.alibaba.nacos.api.remote.response");
        scanPackage.add("com.alibaba.nacos.api.config.remote.request");
        scanPackage.add("com.alibaba.nacos.api.config.remote.response");
    }
    
    @Override
    public Set<String> getScanPackage() {
    
        return scanPackage;
    }
}

Then scan the package defined by these two classes by class name and class Class Sign up to PayloadRegistry.REGISTRY_REQUEST among . When doing GRPC It is provided to GrpcUtils Perform request response object conversion .

3 .2 start-up grpc service

BaseRpcServer.java

    @PostConstruct
    public void start() throws Exception {
    
        String serverName = getClass().getSimpleName();
        Loggers.REMOTE.info("Nacos {} Rpc server starting at port {}", serverName, getServicePort());
        
        startServer();
    
        Loggers.REMOTE.info("Nacos {} Rpc server started at port {}", serverName, getServicePort());
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    
            Loggers.REMOTE.info("Nacos {} Rpc server stopping", serverName);
            try {
    
                BaseRpcServer.this.stopServer();
                Loggers.REMOTE.info("Nacos {} Rpc server stopped successfully...", serverName);
            } catch (Exception e) {
    
                Loggers.REMOTE.error("Nacos {} Rpc server stopped fail...", serverName, e);
            }
        }));

    }

The logic in this class is relatively simple ：

• Defines the service startup interface startServer(), Implemented by subclasses
• Register a hook function called after the service stops stopServer()

BaseGrpcServer.java

    @Override
    public void startServer() throws Exception {
    
        final MutableHandlerRegistry handlerRegistry = new MutableHandlerRegistry();
        
        // #1
        ServerInterceptor serverInterceptor = new ServerInterceptor() {
    
            @Override
            public <T, S> ServerCall.Listener<T> interceptCall(ServerCall<T, S> call, Metadata headers,
                    ServerCallHandler<T, S> next) {
    
                Context ctx = Context.current()
                        .withValue(CONTEXT_KEY_CONN_ID, call.getAttributes().get(TRANS_KEY_CONN_ID))
                        .withValue(CONTEXT_KEY_CONN_REMOTE_IP, call.getAttributes().get(TRANS_KEY_REMOTE_IP))
                        .withValue(CONTEXT_KEY_CONN_REMOTE_PORT, call.getAttributes().get(TRANS_KEY_REMOTE_PORT))
                        .withValue(CONTEXT_KEY_CONN_LOCAL_PORT, call.getAttributes().get(TRANS_KEY_LOCAL_PORT));
                if (REQUEST_BI_STREAM_SERVICE_NAME.equals(call.getMethodDescriptor().getServiceName())) {
    
                    Channel internalChannel = getInternalChannel(call);
                    ctx = ctx.withValue(CONTEXT_KEY_CHANNEL, internalChannel);
                }
                return Contexts.interceptCall(ctx, call, headers, next);
            }
        };
        
        // #2
        addServices(handlerRegistry, serverInterceptor);
        
        // #3
        server = ServerBuilder.forPort(getServicePort()).executor(getRpcExecutor())
                .maxInboundMessageSize(getInboundMessageSize()).fallbackHandlerRegistry(handlerRegistry)
                .compressorRegistry(CompressorRegistry.getDefaultInstance())
                .decompressorRegistry(DecompressorRegistry.getDefaultInstance())
                .addTransportFilter(new ServerTransportFilter() {
    
                    @Override
                    public Attributes transportReady(Attributes transportAttrs) {
    
                        InetSocketAddress remoteAddress = (InetSocketAddress) transportAttrs
                                .get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
                        InetSocketAddress localAddress = (InetSocketAddress) transportAttrs
                                .get(Grpc.TRANSPORT_ATTR_LOCAL_ADDR);
                        int remotePort = remoteAddress.getPort();
                        int localPort = localAddress.getPort();
                        String remoteIp = remoteAddress.getAddress().getHostAddress();
                        Attributes attrWrapper = transportAttrs.toBuilder()
                                .set(TRANS_KEY_CONN_ID, System.currentTimeMillis() + "_" + remoteIp + "_" + remotePort)
                                .set(TRANS_KEY_REMOTE_IP, remoteIp).set(TRANS_KEY_REMOTE_PORT, remotePort)
                                .set(TRANS_KEY_LOCAL_PORT, localPort).build();
                        String connectionId = attrWrapper.get(TRANS_KEY_CONN_ID);
                        Loggers.REMOTE_DIGEST.info("Connection transportReady,connectionId = {} ", connectionId);
                        return attrWrapper;
                        
                    }
                    
                    @Override
                    public void transportTerminated(Attributes transportAttrs) {
    
                        String connectionId = null;
                        try {
    
                            connectionId = transportAttrs.get(TRANS_KEY_CONN_ID);
                        } catch (Exception e) {
    
                            // Ignore
                        }
                        if (StringUtils.isNotBlank(connectionId)) {
    
                            Loggers.REMOTE_DIGEST
                                    .info("Connection transportTerminated,connectionId = {} ", connectionId);
                            connectionManager.unregister(connectionId);
                        }
                    }
                }).build();
        
        // #4
        server.start();
    }

• #1 In fact, it is to define a server interceptor , Used for setting up connection id Etc
• #2 Go to MutableHandlerRegistry Dynamic addition GRPC Method definition
• #3 Build with the above parameters GRPC Of Server.
• #4 start-up GRPC service

So let's see GRPC It's a dynamic addition GRPC Method to Server Inside .

BaseGrpcServer.java

    private void addServices(MutableHandlerRegistry handlerRegistry, ServerInterceptor... serverInterceptor) {
    
        
        // unary common call register.
        final MethodDescriptor<Payload, Payload> unaryPayloadMethod = MethodDescriptor.<Payload, Payload>newBuilder()
                .setType(MethodDescriptor.MethodType.UNARY)
                .setFullMethodName(MethodDescriptor.generateFullMethodName(REQUEST_SERVICE_NAME, REQUEST_METHOD_NAME))
                .setRequestMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance()))
                .setResponseMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance())).build();
        
        final ServerCallHandler<Payload, Payload> payloadHandler = ServerCalls
                .asyncUnaryCall((request, responseObserver) -> grpcCommonRequestAcceptor.request(request, responseObserver));
        
        final ServerServiceDefinition serviceDefOfUnaryPayload = ServerServiceDefinition.builder(REQUEST_SERVICE_NAME)
                .addMethod(unaryPayloadMethod, payloadHandler).build();
        handlerRegistry.addService(ServerInterceptors.intercept(serviceDefOfUnaryPayload, serverInterceptor));
        
        // bi stream register.
        final ServerCallHandler<Payload, Payload> biStreamHandler = ServerCalls.asyncBidiStreamingCall(
                (responseObserver) -> grpcBiStreamRequestAcceptor.requestBiStream(responseObserver));
        
        final MethodDescriptor<Payload, Payload> biStreamMethod = MethodDescriptor.<Payload, Payload>newBuilder()
                .setType(MethodDescriptor.MethodType.BIDI_STREAMING).setFullMethodName(MethodDescriptor
                        .generateFullMethodName(REQUEST_BI_STREAM_SERVICE_NAME, REQUEST_BI_STREAM_METHOD_NAME))
                .setRequestMarshaller(ProtoUtils.marshaller(Payload.newBuilder().build()))
                .setResponseMarshaller(ProtoUtils.marshaller(Payload.getDefaultInstance())).build();
        
        final ServerServiceDefinition serviceDefOfBiStream = ServerServiceDefinition
                .builder(REQUEST_BI_STREAM_SERVICE_NAME).addMethod(biStreamMethod, biStreamHandler).build();
        handlerRegistry.addService(ServerInterceptors.intercept(serviceDefOfBiStream, serverInterceptor));
        
    }

In this way GRPC Two methods are defined ：

• MethodType.UNARY Method of type , Both request parameters and response parameters are Payload, Then the method is named Request/request. adopt GrpcRequestAcceptor Class to handle requests .
• MethodType.BIDI_STREAMING Method of type , Both request parameters and response parameters are Payload, Then the method is named BiRequestStream/requestBiStream. adopt GrpcBiStreamRequestAcceptor To process requests