Dapr实现分布式有状态服务的细节

Dapr是为云上环境设计的跨语言, 事件驱动, 可以便捷的构建微服务的系统. balabala一堆, 有兴趣的小伙伴可以去了解一下.

Dapr提供有状态和无状态的微服务. 大部分人都是做无状态服务(微服务)的, 只是某些领域无状态并不好使, 因为开销实在是太大了; 有状态服务有固定的场景, 就是要求开销小, 延迟和吞吐都比较高. 废话少说, 直接来看Dapr是怎么实现有状态服务的.

先来了解一下有状态服务:

1. 稳定的路由

   发送给A服务器的请求, 不能发给B服务器, 否则就是无状态的

2. 状态

   状态保存在自己服务器内部, 而不是远程存储, 这一点和无状态有很明显的区别, 所以无状态服务需要用redis这种东西加速, 有状态不需要

3. 处理是单线程

   状态一般来讲比较复杂, 想要对一个比较复杂的东西进行并行的计算是比较困难的; 当然A和B的逻辑之间没有关系, 其实是可以并行的, 但是A自己本身的逻辑执行需要串行执行.

对于一个有状态服务来讲(dapr), 实现23实际上是很轻松的, 甚至有一些是用户需要实现的东西, 所以1才是关键, 当前这个消息(请求)需要被发送到哪个服务器上面处理才是最关键的, 甚至决定了他是什么系统.

决定哪个请求的目标地址, 这个东西在分布式系统里面叫Placement, 有时候也叫Naming. TiDB里面有一个Server叫PlacementDriver, 简称PD, 其实就是在干同样的事情.

好了, 开始研究Dapr的Placement是怎么实现的.

有一个Placement的进程, 2333, 目录cmd/placement, 就看他了

func main() {
	log.Infof("starting Dapr Placement Service -- version %s -- commit %s", version.Version(), version.Commit())

	cfg := newConfig()

	// Apply options to all loggers.
	if err := logger.ApplyOptionsToLoggers(&cfg.loggerOptions); err != nil {
		log.Fatal(err)
	}
	log.Infof("log level set to: %s", cfg.loggerOptions.OutputLevel)

	// Initialize dapr metrics for placement.
	if err := cfg.metricsExporter.Init(); err != nil {
		log.Fatal(err)
	}

	if err := monitoring.InitMetrics(); err != nil {
		log.Fatal(err)
	}

	// Start Raft cluster.
	raftServer := raft.New(cfg.raftID, cfg.raftInMemEnabled, cfg.raftBootStrap, cfg.raftPeers)
	if raftServer == nil {
		log.Fatal("failed to create raft server.")
	}

	if err := raftServer.StartRaft(nil); err != nil {
		log.Fatalf("failed to start Raft Server: %v", err)
	}

	// Start Placement gRPC server.
	hashing.SetReplicationFactor(cfg.replicationFactor)
	apiServer := placement.NewPlacementService(raftServer)

可以看到main函数里面启动了一个raft server, 一般这样的话, 就说明在某些能力方面做到了强一致性.

raft库用的是consul实现的raft, 而不是etcd, 因为etcd的raft不是库, 只能是一个服务器(包括etcd embed), 你不能定制里面的协议, 你只能使用etcd提供给你的client来访问他. 这一点etcd做的非常不友好.

如果用raft库来做placement, 那么协议可以定制, 可以找Apply相关的函数, 因为raft状态机只是负责log的一致性, log即消息, 消息的处理则表现出来状态, Apply函数就是需要用户做消息处理的地方. 幸亏之前有做过MIT 6.824的lab, 对这个稍微有一点了解.

// Apply log is invoked once a log entry is committed.
func (c *FSM) Apply(log *raft.Log) interface{} {
	buf := log.Data
	cmdType := CommandType(buf[0])

	if log.Index < c.state.Index {
		logging.Warnf("old: %d, new index: %d. skip apply", c.state.Index, log.Index)
		return nil
	}

	var err error
	var updated bool
	switch cmdType {
	case MemberUpsert:
		updated, err = c.upsertMember(buf[1:])
	case MemberRemove:
		updated, err = c.removeMember(buf[1:])
	default:
		err = errors.New("unimplemented command")
	}

	if err != nil {
		return err
	}

	return updated
}

在pkg/placement/raft文件夹下面找到raft相关的代码, fsm.go里面有对消息的处理函数.

可以看到, 消息的处理非常简单, 里面只有MemberUpsert, 和MemberRemove两个消息.  FSM状态机内保存的状态只有:

// DaprHostMemberState is the state to store Dapr runtime host and
// consistent hashing tables.
type DaprHostMemberState struct {
	// Index is the index number of raft log.
	Index uint64
	// Members includes Dapr runtime hosts.
	Members map[string]*DaprHostMember

	// TableGeneration is the generation of hashingTableMap.
	// This is increased whenever hashingTableMap is updated.
	TableGeneration uint64

	// hashingTableMap is the map for storing consistent hashing data
	// per Actor types.
	hashingTableMap map[string]*hashing.Consistent
}

很明显, 这里面只有DaprHostMember这个有用的信息, 而DaprHostMember就是集群内的节点.

这里可以分析出来, Dapr通过Raft协议来维护了一个强一致性的Membership, 除此之外什么也没干....据我的朋友说, 跟Orleans是有一点类似的, 只是Orleans是AP系统.

再通过对一致性Hash的分析, 可以看到:

func (a *actorsRuntime) lookupActorAddress(actorType, actorID string) (string, string) {
	if a.placementTables == nil {
		return "", ""
	}

	t := a.placementTables.Entries[actorType]
	if t == nil {
		return "", ""
	}
	host, err := t.GetHost(actorID)
	if err != nil || host == nil {
		return "", ""
	}
	return host.Name, host.AppID
}

通过　ActorType和ActorID到一致性的Hash表中去找host, 那个GetHost实现就是一致性Hash表实现的.

Actor RPC Call的实现:

func (a *actorsRuntime) Call(ctx context.Context, req *invokev1.InvokeMethodRequest) (*invokev1.InvokeMethodResponse, error) {
	if a.placementBlock {
		<-a.placementSignal
	}

	actor := req.Actor()
	targetActorAddress, appID := a.lookupActorAddress(actor.GetActorType(), actor.GetActorId())
	if targetActorAddress == "" {
		return nil, errors.Errorf("error finding address for actor type %s with id %s", actor.GetActorType(), actor.GetActorId())
	}

	var resp *invokev1.InvokeMethodResponse
	var err error

	if a.isActorLocal(targetActorAddress, a.config.HostAddress, a.config.Port) {
		resp, err = a.callLocalActor(ctx, req)
	} else {
		resp, err = a.callRemoteActorWithRetry(ctx, retry.DefaultLinearRetryCount, retry.DefaultLinearBackoffInterval, a.callRemoteActor, targetActorAddress, appID, req)
	}

	if err != nil {
		return nil, err
	}
	return resp, nil
}

通过刚才我们看到loopupActorAddress函数找到的Host, 然后判断是否是在当前Host宿主内, 否则就发送到远程, 对当前宿主做了优化, 实际上没鸡儿用, 因为分布式系统里面, 一般都会有很多个host, 在当前host内的概率实际上是非常低的.

从这边, 我们大概就能分析到全貌, 即Dapr实现分布式有状态服务的细节:

1. 通过Consul Raft库维护Membership

2. 集群和Placement组件通讯, 获取到Membership

3. 寻找Actor的算法实现在Host内, 而不是Placement组件. 通过ActorType找到可以提供某种服务的Host, 然后组成一个一致性Hash表, 到该表内查找Host, 进而转发请求

对Host内一致性Hash表的查找引用, 找到了修改内容的地方:

func (a *actorsRuntime) updatePlacements(in *placementv1pb.PlacementTables) {
	a.placementTableLock.Lock()
	defer a.placementTableLock.Unlock()

	if in.Version != a.placementTables.Version {
		for k, v := range in.Entries {
			loadMap := map[string]*hashing.Host{}
			for lk, lv := range v.LoadMap {
				loadMap[lk] = hashing.NewHost(lv.Name, lv.Id, lv.Load, lv.Port)
			}
			c := hashing.NewFromExisting(v.Hosts, v.SortedSet, loadMap)
			a.placementTables.Entries[k] = c
		}

		a.placementTables.Version = in.Version
		a.drainRebalancedActors()

		log.Infof("placement tables updated, version: %s", in.GetVersion())

		a.evaluateReminders()
	}
}

从这几行代码可以看出, 版本不不一样, 就会全更新, 而且还会进行rehash, 就是a.drainRebalanceActors. 

如果学过数据结构, 那么肯定学到过一种东西叫HashTable, HashTable在扩容的时候需要rehash, 需要构建一个更大的table, 然后把所有元素重新放进去, 位置会和原先的大不一样. 而一致性Hash可以解决全rehash的情况, 只让部分内容rehash, 失效的内容会比较少.

但是, 凡事都有一个但是, 所有的节点都同时rehash还好, 可一个分布式系统怎么做到所有node都同时rehash, 很显然是做不到的, 所以Dapr维护的Actor Address目录, 是最终一致的, 也就是系统里面会存在多个ID相同的Actor(短暂的), 还是会导致不一致.

对dapr/proto/placement/v1/placement.proto查看, 验证了我的猜想

// Placement service is used to report Dapr runtime host status.
service Placement {
  rpc ReportDaprStatus(stream Host) returns (stream PlacementOrder) {}
}

message PlacementOrder {
  PlacementTables tables = 1;
  string operation = 2;
}

Host启动, 就去placement那边通过gRPC Stream订阅了集群的变动. 懒到极点了, 居然是把整个membership发送过来, 而不是发送的diff.

总结一下, 从上面的源码分析我们可以知道, Dapr的Membership是CP系统, 但是Actor的Placement不是, 是一个最终一致的AP系统. 而TiDB的PD是一个CP系统, 只不过是通过etcd embed做的. 希望对大家有一点帮助.

对我有帮助的, 可能就是Dapr对于Consul raft的使用.

参考:

1. Dapr

2. Etcd Embed

3. Consul Raft