Тестирование выбора лидера (в Raft) с использованием Akka, без Thread.sleep () ;-)

В настоящее время я живу в Лондоне и очень скучаю по собраниям Краковского  Клуба чтения SCKRK  ( в принципе, это лучшее, что есть  нарезанный хлеб!) . Несмотря на то, что я посещаю их в любом случае в настоящее время, я все еще скучаю по тематике научных работ. Такое же чувство поразило меня, и некоторые друзья (Анджей из  GeeCON , чтобы назвать одного) решили начать аналогичную инициативу в Лондоне. Он называется  Paper Cup , Клуб чтения , и является «клубом чтения», что означает, что мы выбираем «Информатику» для чтения, чтения их дома, а затем во время встречи обсуждаем их ожесточенно ?

На этой неделе мы решили прочитать «Плот» (« Плот — в поисках понятного алгоритма консенсуса» ), который сейчас «очень модный» (и относительно свежий; статья написана с октября 2013 года!). Но если подвести итог, то это алгоритм, пытающийся решить ту же проблему, что и  Paxos  (достижение консенсуса в распределенных системах), но в то же время понятный. Мы читали «Паксос» несколько раз (например , документ «Лакпорт« Сделано просто »  ) Лампорта  ), но всегда было непросто сделать все правильно. С другой стороны, авторы Raft действительно хотели, чтобы алгоритм был понятным, а также все еще «достаточно хорошим» с точки зрения производительности.

Поскольку алгоритм действительно очень приятный и описан в терминах простого  конечного автомата , я решил реализовать его с использованием Akka (и FSM) для следующего собрания, а другой друг реализует его в Clojure. Справедливости ради, похоже, я не первый, кто понял, что FSM и актеры Akka в целом сделают этот супер забавный процесс реализации — уже есть несколько репозиториев, которые делают именно это (ну, некоторые из них являются «начальным коммитом»). ;-)). В этом посте я не буду фокусироваться на самом алгоритме, а скорее поделюсь полезной техникой, когда вы реализуете такую ​​вещь, и хотел бы написать «большой тест», который утверждает с высокой точки убедитесь, что система работает так, как вы ожидаете.

Тестовый случай, о котором мы поговорим, очень прост:

1. Плот должен выбрать лидера сразу после запуска узлов
2. Плот должен переизбрать лидера, если мы убьем текущего

Как я уже сказал, мы не будем углубляться в алгоритм здесь — если вам интересно, вы можете просто прочитать технический документ (и, возможно, даже заглянуть на PaperCup? :-)).
Давайте прыгнем прямо в тесты:

it should "elect initial Leader" in {
  // given
  info("Before election: ")
  infoMemberStates()
  // when
  awaitElectedLeader()
  info("After election: ")
  infoMemberStates()
  // then
  members.count(_.stateName == Leader) should equal (1)
  members.count(_.stateName == Candidate) should equal (0)
  members.count(_.stateName == Follower) should equal (4)
}
it should "elect replacement Leader if current Leader dies" in {
  // given
  infoMemberStates()
  // when
  val leaderToStop = leader.get
  leaderToStop.stop()
  info(s"Stopped leader: ${simpleName(leaderToStop)}")
  // then
  awaitElectedLeader()
  info("New leader elected: ")
  infoMemberStates()
  members.count(_.stateName == Leader) should equal (1)
  members.count(_.stateName == Candidate) should equal (0)
  members.count(_.stateName == Follower) should equal (3)
}

У каждого участника есть экземпляр Raft, и один из них (по выбору) через некоторое время станет Лидером. Один из способов проверить это — запланировать TestProbe в «элементах для уведомления» или использовать датчик в качестве промежуточного элемента между элементами (A -> Probe -> B). Но если честно, это большая работа. Решение, которое я предлагаю здесь, также используется akka-persistence  (ну, и мой  плагин akka-persistence-hbase  к нему тоже) — назовем его « Ожидание подписки на поток событий ». Теперь, когда у нас есть причудливое название, давайте продолжим его реализацию.

Давайте начнем с конца истории. Я хочу заблокировать, пока не произойдет какое-либо событие в системе (  awaitElectedLeader() метод делает это). Где я могу ловить рыбу для этих событий? Оказывается, в Akka есть встроенный eventBus, готовый к отправке сообщений (и он доступен для любой ActorSystem без какой-либо дополнительной настройки). Давайте сначала реализуем наши вспомогательные методы для ожидания  ElectedAsLeader сообщения:

implicit val probe = TestProbe() // somewhere
def subscribeElectedLeader()(implicit probe: TestProbe): Unit =
  system.eventStream.subscribe(probe.ref, classOf[ElectedAsLeader])
def awaitElectedLeader()(implicit probe: TestProbe): Unit =
  probe.expectMsgClass(max = 3.seconds, classOf[ElectedAsLeader])

Здесь (в тестовом классе, расширяющем  TestKit ) мы можем получить доступ к EventStream и подписаться на определенные типы сообщений на нем. Мы будем использовать TestProbe () для получения этих событий, потому что это позволяет нам  ожидать MSG *,  который именно то, что нам нужно. Поэтому в awaitElectedLeader () мы просто ждем использования зонда, пока не появится сообщение «в кластере появился какой-то лидер». Пока это не так уж отличается от того, что я описал ранее, а затем назвал его «много работы». Теперь мы вернемся к настоящему трюку в этом методе. EventStream, который вы видите здесь, определен в  ActorSystem , и, как мы знаем, он также доступен из самого Actor.

Например, в akka-persistence актеры публикуют события о ходе воспроизведения / подтверждений и т. Д. Но это только для тестирования, поэтому вы можете включить флаг (например, persistence.publish-Подтверждения  ), чтобы он публиковал события в eventStream. Это отличная идея, которая делает тестирование очень простым, и его, безусловно, можно будет внедрить в мою реализацию Raft (и, вероятно, все равно получится так). Однако давайте теперь подумаем, как мы могли бы расширить  прием  Actor, чтобы автоматически отправлять входящие сообщения также в  EventStream .

На самом деле это очень просто, если вы знаете о методе актера  roundReceive  . Так же , как это следует из названия (звуки очень похожи на АОП, кстати, не так ли? ;-)), он позволяет обернуть  получить  вызов актера с вашим кодом. Наша реализация просто отправит все сообщения в eventStream:

/**
 * Use for testing.
 * 
 * Forwards all messages received to the system's EventStream.
 * Use this to deterministically `awaitForLeaderElection` etc.
 */
trait EventStreamAllMessages {
  this: Actor =>
  override def aroundReceive(receive: Actor.Receive, msg: Any) = {
    context.system.eventStream.publish(msg.asInstanceOf[AnyRef])
    receive.applyOrElse(msg, unhandled)
  }
}

Легко! Теперь мы не хотим изменять наш производственный код (или, может быть, это не наш код, и т. Д.), Поэтому во время создания Actor мы можем использовать уточнение типа, чтобы смешать эту черту с нашим RaftActor (во время установки, как я кратко упомянул, но не показывать явно):

system.actorOf(Props(new RaftActor with EventStreamAllMessages))

И этот Actor теперь будет автоматически отправлять все сообщения в EventStream. В потоке событий мы подписываемся на события LeaderElected с помощью нашего зонда, а также   части ожидаемых сообщений, о которых мы уже говорили. Итак, это все! Очень простой способ проверить актеров «со стороны». Теперь вы можете оглянуться назад на первый фрагмент кода в этом посте, чтобы увидеть, как все это сочетается.

В общем, если вы создаете что-либо с использованием Actors или, может быть, даже библиотеки, которую кто-то может захотеть протестировать, это отличная идея, чтобы предоставить настройку, позволяющую вашим Actors отправлять события EventStream, потому что тогда даже извне это легко отслеживать, что происходит в глубине системы. Я почти уверен, что буду включать такие «включаемые» события для библиотек, которые я сейчас собираю, чтобы быть приятным для пользователей библиотеки. ?