PolarisMesh源码系列——服务如何注册

前话

PolarisMesh（北极星）是腾讯开源的服务治理平台，致力于解决分布式和微服务架构中的服务管理、流量管理、配置管理、故障容错和可观测性问题，针对不同的技术栈和环境提供服务治理的标准方案和最佳实践。

PolarisMesh 官网：

https://polarismesh.cn/#/

PolarisMesh Github：

https://github.com/polarismesh/polaris

Polaris-server 作为 PolarisMesh 的控制面，该进程主要负责服务数据、配置数据、治理规则的管理以及下发至北极星 SDK 以及实现了 xDS 的客户端。

Polaris-server 是如何处理客户端的服务注册请求的呢？服务数据是怎么存储的呢？带着这个疑问，我们来探究看下 Polaris-server 的启动流程，看看北极星是实现的。

前期准备

• golang 环境，需要1.17.x +

• 准备 vscode 或者 goland

• 从 github 中 clone 一份 polaris-server 的源码，这里推荐从 release-vx.y.z 分支中选择一个分支进行学习，以下文章将基于 release-v1.12.0 分支进行研究。

• 从 github 中 clone 一份 polaris-java 的源码，这里推荐从 release-vx.y.z 分支中选择一个分支进行学习，以下文章将基于release-v1.10.0分支进行研究。

正题

polaris-java

InstanceRegisterRequest request = new InstanceRegisterRequest();// 设置实例所属服务信息request.setService(service);// 设置实例所属服务的命名空间信息request.setNamespace(namespace);// 设置实例的 host 信息request.setHost(host);// 设置实例的端口信息request.setPort(port);// 可选，资源访问Token，即用户/用户组访问凭据，仅当服务端开启客户端鉴权时才需配置request.setToken(token);// 设置实例版本request.setVersion(version);// 设置实例的协议request.setProtocol(protocol);// 设置实例权重request.setWeight(weight);// 设置实例的标签request.setMetadata(metadata);// 设置实例地理位置 zone 信息request.setZone(zone);// 设置实例地理位置 region 信息request.setRegion(region);// 设置实例地理位置 campus 信息request.setCampus(campus);// ttl超时时间，如果节点要调用heartbeat上报，则必须填写，否则会400141错误码，单位：秒request.setTtl(ttl);

polaris-go

// InstanceRegisterRequest 注册服务请求type InstanceRegisterRequest struct {  // 必选，服务名  Service string  // 必选，命名空间  Namespace string  // 必选，服务监听host，支持IPv6地址  Host string  // 必选，服务实例监听port  Port int  // 可选，资源访问Token，即用户/用户组访问凭据，仅当服务端开启客户端鉴权时才需配置  ServiceToken string  // 以下字段可选，默认nil表示客户端不配置，使用服务端配置  // 服务协议  Protocol *string  // 服务权重，默认100，范围0-10000  Weight *int  // 实例提供服务版本号  Version *string  // 用户自定义metadata信息  Metadata map[string]string  // 该服务实例是否健康，默认健康  Healthy *bool  // 该服务实例是否隔离，默认不隔离  Isolate *bool  // ttl超时时间，如果节点要调用heartbeat上报，则必须填写，否则会400141错误码，单位：秒  TTL *int    // Location 当前注册实例的地理位置信息，主要用于就近路由  Location *Location  // 可选，单次查询超时时间，默认直接获取全局的超时配置  // 用户总最大超时时间为(1+RetryCount) * Timeout  Timeout *time.Duration  // 可选，重试次数，默认直接获取全局的超时配置  RetryCount *int}

客户端发起注册请求

可以先通过官方的 SDK 使用手册来看看是如何使用SDK的服务注册。

• polaris-java 服务注册功能使用

https://polarismesh.cn/docs/%E4%BD%BF%E7%94%A8%E6%8C%87%E5%8D%97/java%E5%BA%94%E7%94%A8%E5%BC%80%E5%8F%91/sdk/%E6%9C%8D%E5%8A%A1%E6%B3%A8%E5%86%8C%E5%8F%91%E7%8E%B0/

• polaris-go 服务注册功能使用

https://polarismesh.cn/docs/%E4%BD%BF%E7%94%A8%E6%8C%87%E5%8D%97/go%E5%BA%94%E7%94%A8%E5%BC%80%E5%8F%91/sdk/%E6%9C%8D%E5%8A%A1%E6%B3%A8%E5%86%8C%E5%8F%91%E7%8E%B0/

这里我们已 polaris-java 为例，看看 polaris-java 如何将服务实例注册请求发送至北极星服务端。

ProviderAPI providerAPI = DiscoveryAPIFactory.createProviderAPI();InstanceRegisterRequest registerRequest = new InstanceRegisterRequest();// 初始化服务实例注册信息...InstanceRegisterResponse registerResp = providerAPI.registerInstance(registerRequest);

通过这个简单示例代码可知，服务实例的注册动作是通过 polaris-java 中 ProviderAPI（负责服务实例注册相关方法的调用） 的 registerInstance 方法完成。

通过 IDEA 或者 vscode，查看这个方法的具体实现：

@Overridepublic InstanceRegisterResponse registerInstance(InstanceRegisterRequest req) throws PolarisException {    if (req.getTtl() == null) {        req.setTtl(DEFAULT_INSTANCE_TTL);    }    return registerFlow.registerInstance(req, this::doRegister, this::heartbeat);}

当调用 providerAPI.registerInstance 后，SDK 内部会自动设置实例的 TTL 周期，然后交由 RegisterFlow 这个负责注册动作的流程编排者执行。因此接着看看这个 RegisterFlow 的定义。

public class RegisterFlow {
    // 异步注册header key    private static final String HEADER_KEY_ASYNC_REGIS = "async-regis";    // 最大连续心跳失败阈值    private static final int HEARTBEAT_FAIL_COUNT_THRESHOLD = 2;    // SDK 上下文，包括SDK配置、SDK插件实例管理、内部任务流程编排等等    private final SDKContext sdkContext;    // 发送实例心跳    private final ScheduledThreadPoolExecutor asyncRegisterExecutor;
    ...}

其实 RegisterFlow 就干两件事件：

• 发起实例注册动作

• 内部维护每个实例的心跳上报

public InstanceRegisterResponse registerInstance(InstanceRegisterRequest request, RegisterFunction registerFunction,        HeartbeatFunction heartbeatFunction) {    // 将注册请求发送至北极星服务端    InstanceRegisterResponse instanceRegisterResponse = registerFunction.doRegister(request,            createRegisterV2Header());    // 当前实例的注册管理状态进行本地保存    RegisterState registerState = RegisterStateManager.putRegisterState(sdkContext, request);    if (registerState != null) {        // 首次存入实例的注册状态时，为该实例注册创建定期心跳上报动作任务        registerState.setTaskFuture(asyncRegisterExecutor.scheduleWithFixedDelay(                () -> doRunHeartbeat(registerState, registerFunction, heartbeatFunction), request.getTtl(),                request.getTtl(), TimeUnit.SECONDS));    }    return instanceRegisterResponse;}

来看看 registerFunction.doRegister 的主要流程以及如何将请求发送到服务端。

// com.tencent.polaris.discovery.client.flow.RegisterFlow#registerInstanceprivate InstanceRegisterResponse doRegister(InstanceRegisterRequest req, Map<String, String> customHeader) {    checkAvailable("ProviderAPI");    Validator.validateInstanceRegisterRequest(req);    // 填充注册实例的地理位置信息    enrichInstanceLocation(req);
    ...    // 调用协议插件，发起网络调用    CommonProviderResponse response = serverConnector.registerInstance(request, customHeader);    ...}
// com.tencent.polaris.plugins.connector.grpc.GrpcConnector#registerInstancepublic CommonProviderResponse registerInstance(CommonProviderRequest req, Map<String, String> customHeader)        throws PolarisException {    ...    try {        waitDiscoverReady();        // 从连接池中获取一个链接        connection = connectionManager                .getConnection(GrpcUtil.OP_KEY_REGISTER_INSTANCE, ClusterType.SERVICE_DISCOVER_CLUSTER);        req.setTargetServer(connectionToTargetNode(connection));        // 根据 Connection 创建一个 gRPC Stub        PolarisGRPCGrpc.PolarisGRPCBlockingStub stub = PolarisGRPCGrpc.newBlockingStub(connection.getChannel());        ...        // 向服务端发起 gRPC 请求，完成服务实例的注册        ResponseProto.Response registerInstanceResponse = stub.registerInstance(buildRegisterInstanceRequest(req));    ...}

服务端处理注册请求

当实例注册请求从北极星 SDK 发出之后，数据流在服务端主要经历这几个流程:

• apiserver 层接受 SDK 的注册请求，将其转为对应的服务端数据结构。

• apiserver 层将请求传递到 resource auth filter 层，进行资源鉴权。

• resource auth filter 完成资源鉴权后，数据流转到 service 层。

• service 层完成服务注册请求的合法性检查以及资源存在性检查比对。

• 如果资源存在，则直接返回已存在实例的唯一 ID。

• 如果资源不存在，则根据 namespace、service、host、port 计算出实例的唯一 ID。

• 将注册请求扔进 BatchController 中。

• batch controller 将一批注册请求转为 store 层的数据结构。

• batch controller 将请求传递到 store 层，store 层选择具体的 store plugin 实现完成写操作。

• mysql 插件：将服务实例信息写入到对应的数据库表中。

• bolt 插件：将服务实例信息写入到本地文件中。

apiserver 层

北极星的 apiserver 层进行接收并处理，由于北极星 SDK 和服务端的数据通信走的是 gRPC 协议，因此这里请求就会在基于 gRPC 实现的 apiserver 插件中进行处理。

// RegisterInstance 注册服务实例func (g *DiscoverServer) RegisterInstance(ctx context.Context, in *apiservice.Instance) (*apiservice.Response, error) {  // 需要记录操作来源，提高效率，只针对特殊接口添加operator  rCtx := grpcserver.ConvertContext(ctx)  rCtx = context.WithValue(rCtx, utils.StringContext("operator"), ParseGrpcOperator(ctx))
  // 客户端请求中带了 token 的，优先已请求中的为准  if in.GetServiceToken().GetValue() != "" {    rCtx = context.WithValue(rCtx, utils.ContextAuthTokenKey, in.GetServiceToken().GetValue())  }
  grpcHeader := rCtx.Value(utils.ContextGrpcHeader).(metadata.MD)
  if _, ok := grpcHeader["async-regis"]; ok {    rCtx = context.WithValue(rCtx, utils.ContextOpenAsyncRegis, true)  }
  out := g.namingServer.RegisterInstance(rCtx, in)  return out, nil}

  // 需要返回包装代理的 DiscoverServer  order := namingOpt.Interceptors  for i := range order {    factory, exist := serverProxyFactories[order[i]]    if !exist {      return fmt.Errorf("name(%s) not exist in serverProxyFactories", order[i])    }
    proxySvr, err := factory(namingServer, server)    if err != nil {      return err    }    server = proxySvr  }  return nil

// RegisterInstance create one instancefunc (svr *ServerAuthAbility) RegisterInstance(ctx context.Context, req *apiservice.Instance) *apiservice.Response {    // 收集本次客户端发起的服务实例注册请求所涉及的操作资源  authCtx := svr.collectClientInstanceAuthContext(    ctx, []*apiservice.Instance{req}, model.Create, "RegisterInstance")    // 调用对应的鉴权插件，对本次操作进行权限检查  _, err := svr.strategyMgn.GetAuthChecker().CheckClientPermission(authCtx)  if err != nil {    resp := api.NewResponseWithMsg(convertToErrCode(err), err.Error())    return resp  }
  ctx = authCtx.GetRequestContext()  ctx = context.WithValue(ctx, utils.ContextAuthContextKey, authCtx)    // 权限检查通过之后，将请求转发给 service 层  return svr.targetServer.RegisterInstance(ctx, req)}

// CreateInstance create a single service instancefunc (s *Server) CreateInstance(ctx context.Context, req *apiservice.Instance) *apiservice.Response {  ...  data, resp := s.createInstance(ctx, req, &ins)  ...        // 发布实例上线事件 & 记录实例注册操作记录  s.sendDiscoverEvent(*event)  s.RecordHistory(ctx, instanceRecordEntry(ctx, req, svc, data, model.OCreate))  ...}
// createInstance store operatefunc (s *Server) createInstance(ctx context.Context, req *apiservice.Instance, ins *apiservice.Instance) (  *model.Instance, *apiservice.Response) {  // 自动创建实例所在的服务信息  svcId, errResp := s.createWrapServiceIfAbsent(ctx, req)  if errResp != nil {    log.Errorf("[Instance] create service if absent fail : %+v, req : %+v", errResp.String(), req)    return nil, errResp  }  if len(svcId) == 0 {    log.Errorf("[Instance] create service if absent return service id is empty : %+v", req)    return nil, api.NewResponseWithMsg(apimodel.Code_BadRequest, "service id is empty")  }
  // 根据 CMDB 插件填充实例的地域信息  s.packCmdb(ins)        // 如果没有开启实例批量注册，则同步调用存储层接口将实例数据进行持久化  if namingServer.bc == nil || !namingServer.bc.CreateInstanceOpen() {    return s.serialCreateInstance(ctx, svcId, req, ins) // 单个同步  }        // 如果开启了实例批量注册，则会将注册请求丢入一个异步队列进行处理  return s.asyncCreateInstance(ctx, svcId, req, ins) // 批量异步}

func (s *Server) serialCreateInstance(  ctx context.Context, svcId string, req *apiservice.Instance, ins *apiservice.Instance) (*model.Instance, *apiservice.Response) {  ...  instance, err := s.storage.GetInstance(ins.GetId().GetValue())  ...  // 如果存在，则替换实例的属性数据，但是需要保留用户设置的隔离状态，以免出现关键状态丢失  if instance != nil && ins.Isolate == nil {    ins.Isolate = instance.Proto.Isolate  }  // 直接同步创建服务实例  data := instancecommon.CreateInstanceModel(svcId, ins)
  // 创建服务实例时，需要先锁住服务，避免在创建实例的时候把服务信息删除导致出现错误的数据  _, releaseFunc, errCode := s.lockService(ctx, req.GetNamespace().GetValue(),    req.GetService().GetValue())  if errCode != apimodel.Code_ExecuteSuccess {    return nil, api.NewInstanceResponse(errCode, req)  }
  defer releaseFunc()        // 调用存储层直接写实例信息  if err := s.storage.AddInstance(data); err != nil {    log.Error(err.Error(), utils.ZapRequestID(rid), utils.ZapPlatformID(pid))    return nil, wrapperInstanceStoreResponse(req, err)  }  return data, nil}

func (s *Server) asyncCreateInstance(  ctx context.Context, svcId string, req *apiservice.Instance, ins *apiservice.Instance) (  *model.Instance, *apiservice.Response) {  allowAsyncRegis, _ := ctx.Value(utils.ContextOpenAsyncRegis).(bool)        // 将实例注册请求放入异步任务池中  future := s.bc.AsyncCreateInstance(svcId, ins, !allowAsyncRegis)        // 等待任务完成  if err := future.Wait(); err != nil {    if future.Code() == apimodel.Code_ExistedResource {      req.Id = utils.NewStringValue(ins.GetId().GetValue())    }    return nil, api.NewInstanceResponse(future.Code(), req)  }
  return instancecommon.CreateInstanceModel(svcId, req), nil}

存储层处理注册数据

北极星的存储层是插件化设计，单机模式下是采用 boltdb，集群模式则是依赖 MySQL，这里只说集群模式下的存储层处理。

依赖 MySQL 的存储层实现中，针对实例信息，北极星将其拆分成了三个表。

实例主要信息

CREATE TABLE `instance`(    # 实例 ID，主键，唯一约束    `id`                  varchar(128) NOT NULL,    # 实例所属的服务 ID    `service_id`          varchar(32)  NOT NULL,    # 实例对外可访问的 IP 地址信息    `host`                varchar(128) NOT NULL,    # 实例对外可访问的端口地址信息    `port`                int(11)      NOT NULL,    # 实例的端口协议信息，用户自定义，比如 dubbo、gRPC、http 等    `protocol`            varchar(32)           DEFAULT NULL,    # 实例的版本信息，用户自定义，默认为空    `version`             varchar(32)           DEFAULT NULL,    # 实例的健康状态，1 为健康，0 为不健康    `health_status`       tinyint(4)   NOT NULL DEFAULT '1',    # 实例的隔离状态，1 为打开，0 为关闭    `isolate`             tinyint(4)   NOT NULL DEFAULT '0',    # 实例的权重信息，主要用于负载均衡    `weight`              smallint(6)  NOT NULL DEFAULT '100',    # 是否开启健康检查，0 为不开启，1 为开启    `enable_health_check` tinyint(4)   NOT NULL DEFAULT '0',    # 实例的地域信息，记录 region, zone, idc 信息    `cmdb_region`         varchar(128)          DEFAULT NULL comment 'The region information of the instance is mainly used to close the route',    `cmdb_zone`           varchar(128)          DEFAULT NULL comment 'The ZONE information of the instance is mainly used to close the route.',    `cmdb_idc`            varchar(128)          DEFAULT NULL comment 'The IDC information of the instance is mainly used to close the route',    # 实例的版本信息，仅北极星内部使用    `revision`            varchar(32)  NOT NULL comment 'Instance version information',    # 实例的逻辑删除标记，0 表示表示正常，1 表示已逻辑删除    `flag`                tinyint(4)   NOT NULL DEFAULT '0',    `ctime`               timestamp    NOT NULL DEFAULT CURRENT_TIMESTAMP comment 'Create time',    `mtime`               timestamp    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP comment 'Last updated time',    ...) ENGINE = InnoDB;

实例健康检查的类型

CREATE TABLE `health_check`(    `id`   varchar(128) NOT NULL comment 'Instance ID',    `type` tinyint(4)   NOT NULL DEFAULT '0' comment 'Instance health check type',    `ttl`  int(11)      NOT NULL comment 'TTL time jumping',    PRIMARY KEY (`id`),    CONSTRAINT `health_check_ibfk_1` FOREIGN KEY (`id`) REFERENCES `instance` (`id`) ON DELETE CASCADE ON UPDATE CASCADE) ENGINE = InnoDB;

实例的元数据

CREATE TABLE `instance_metadata`(    `id`     varchar(128)  NOT NULL comment 'Instance ID',    `mkey`   varchar(128)  NOT NULL comment 'instance label of Key',    `mvalue` varchar(4096) NOT NULL comment 'instance label Value',    `ctime`  timestamp     NOT NULL DEFAULT CURRENT_TIMESTAMP comment 'Create time',    `mtime`  timestamp     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP comment 'Last updated time',    PRIMARY KEY (`id`, `mkey`),    KEY `mkey` (`mkey`),    CONSTRAINT `instance_metadata_ibfk_1` FOREIGN KEY (`id`) REFERENCES `instance` (`id`) ON DELETE CASCADE ON UPDATE CASCADE) ENGINE = InnoDB;

因此在操作存储层持久化服务实例信息时，需要做以下几个操作：

// 添加实例主信息if err := batchAddMainInstances(tx, instances); err != nil {  log.Errorf("[Store][database] batch add main instances err: %s", err.Error())  return err}// 添加实例的健康检查信息if err := batchAddInstanceCheck(tx, instances); err != nil {  log.Errorf("[Store][database] batch add instance check err: %s", err.Error())  return err}// 先清理实例原先的 metadata 信息数据，确保不会遗留脏数据if err := batchDeleteInstanceMeta(tx, instances); err != nil {  log.Errorf("[Store][database] batch delete instance metadata err: %s", err.Error())  return err}// 添加实例的 metadata 信息if err := batchAddInstanceMeta(tx, instances); err != nil {  log.Errorf("[Store][database] batch add instance metadata err: %s", err.Error())  return err}

扫描下方二维码关注本公众号，

了解更多微服务、消息队列的相关信息！

解锁超多鹅厂周边！