Event-sourcing avancé

avec ScalaPSUG - Jonathan Winandy &

Valentin Kasas

1

Agenda● Eventsourcing : qu’es aquò ?● Patterns et anti-patterns● Streaming et métadonnées● Typage et sérialisation

2

Qui sommes-nous ? Jonathan Winandy @ahoy_jon

● Data engineer / Hardware destroyer● Eventsourcer since before it was cool● Scala/Java/Clojure● CYM● hipster

Valentin Kasas @ValentinKasas

● Web engineer / homme-orchestre● Eventsourcer since 2013● Scala :: scala :: HNil● Freelance● dandy

3

Eventsourcing : définitions

4

EventSourcing : définition

● Une idée maîtresse : prendre comme source de vérité la succession des événements qui surviennent dans un système

● Plutôt que s’appuyer sur un représentation muable du présent, on s’appuie sur l’enregistrement des événements passés et immuables.

5

RDBMS vs EventSourcing

Source http://docs.geteventstore.com/introduction/img/transactional-model-with-delete.png

VS

Source https://martin.kleppmann.com/2015/03/insideout-06.png

6

RDBMS vs EventSourcing

7

EventSourcing, au niveau des API 1/2 trait Stream[M[_]] {

val name: String

def append(event: Event, events: Event*): M[WriteCompleted]

def read(position: Int): M[Event]

}

case class Event(evenType: String, id: UUID, body: Content, metadata: Content)

case class WriteCompleted(start: Int, end: Int)

8

EventSourcing, au niveau des API 2/2 trait Stream[M[_]] {

def readPage(fromPosition: Int, pageSize: Int,

readDirection: ReadDirection = Forward): M[Page]

def subscribeTo(observer: SubscriptionObserver): Closeable

def subscribeToFrom(observer: SubscriptionObserver, position: Int): Closeable

}

case class Page(events: List[Event], fromPosition: Int,

nextPosition: Int, readDirection: ReadDirection)

trait SubscriptionObserver {

def onLiveProcessingStart(subscription: Closeable)

def onEvent(event: Event, subscription: Closeable)

def onError(e: Throwable)

def onClose()

}9

Modèle basique

95 96 97 98

append

appendappend

read(95)

read(96)read(97)

10

Invariant(s)

∀p ∈ ℕ, et pour une définition de ‘=‘

x := read(p)y := read(p)

‘x ≠ ⊥’ ⇒ x = y

Plus simplement :

La cacheabilité est infinie.

Une fois qu’un élément est disponible à une position (p), cet élément sera identique (=) pour toujours.

11

EventSourcing : les avantagesConserver l’intégralité de l’historique d’un système offre de nombreux avantages :

● Plus proche de la réalité (perception),● Auditabilité,● Observabilité de la nouveauté,● Simplicité de consommation,● Réinterprêtation de l’historique (BI for “Free”).● Facilite (beaucoup) CQRS

Decision

Apply

commande

état

event

Read Model

Read Model

Read Model 12

EventSourcing : les inconvénientsL’eventsourcing présente aussi quelques inconvénients :

● Consommation d’espace disque plus importante (et en constante augmentation),

● Démarrage potentiellement plus long (il faut relire tout l’historique),● Difficile de changer le schéma des événements (<= immuabilité).● Découpage en streams, et opérations sur plusieurs streams.● Multiplication des états (read models, Monoid[Event]).● Gestion asynchrone.

13

Patterns et anti-patterns

14

Patterns● Algébre d’événement (List[Event] à la place de A * List[Event]),● Génération d’id non coordonnée => UUID,● Quantification de la donnée,● Capture des réglès métier complexes, et décisions peu reproductibles.● Corrélation et causation● Sagas

15

Anti-Patterns● CRUD Sourcing● Command Sourcing● Make-My-Sum-Type Grow

16

Streaming, métadonnées et metastreams

17

Pourquoi ajouter des métadonnées ?● Les events contiennent la donnée métier, mais on a aussi besoin de

données techniques pour :○ Expliciter le contexte d’écriture (instance, saga, commande, position dans la commande).○ Périniser la donnée (schéma à l’écriture).○ Tracer la provenance (de qui, de la part de).○ Séparer la donnée entre différent environnement partageant la même infrastructure.

18

Quels types de métadonnées ?● Métadonnées des événements

○ Corrélation & causation○ Informations de type (schéma, on y reviendra)○ etc...

● Méta-streams (ou streams de méta-events) d’informations techniques sur le système○ Instances○ Schémas

19

Types et sérialisation

20

Pourquoi/comment typer les streams ?● Sans typage, pas d’albèbre, pas moyen d’écrire

○ state = events.foldLeft(initial)(_ apply _)● Problème : comment conserver le type des “vieux” events ?● Quand le schéma des events change, on doit pouvoir

○ Upcaster les anciens events dans le nouveaux schéma (pour pouvoir démarrer une nouvelle instance sur les anciennes données)

○ Downcaster les nouveaux events (pour que les anciennes instances continuent de fonctionner pendant la mise à jour)

21

API typée : définitioncase class Event[E](event:E, id:UUID, metadata:Metadata)

trait TypedStream[M[_], E] {

implicit val coproductEvidence: Generic[E]

implicit val serializableEvidence:AkiraSerializable[E]

val name: String @@ StreamId

def append(event: Event[E], events: Event[E]*): M[WriteCompleted]

def read(position: Int): M[Event[E]]

}

22

API typée : usagesealed trait MQTTEvent

case class Disconnected(clientID: String) extends MQTTEvent

case class Published(clientID: String,

payload: Seq[Byte],

topicName: String) extends MQTTEvent

val mqttStream:TypedStream[Id,MQTTEvent] = ___

mqttStream.append(___)

23

Comment sérialiser les événements ?L’EventSourcing nécessite une bonne sérialisation, car les événements vont être écrits et consommés par :

● Beaucoup de programmes (dans des langages différents).● Beaucoup d’intermédiaires.● Pendant plusieurs années.

Une bonne sérialisation est :

● Extensible.● Auto descriptible.● Peut être lue et écrite de manière générique et indépendante.

24

Comment sérialiser les messages ?

Pour l’instant, nous utilisons Avro 1.8.x avec l’encodeur JSON, mais c’est un détail d’implémentation.

25

● Conserver les schémas (passés et actuels) de tous les types d’événements ○ Dans les métadonnées de chaque événement, placer une référence vers le schéma dudit

événement

○ Dans un métastream dédié, stocker une représentation générique de chaque schéma (sous la forme d’événements SchemaCreated

● Il faut aussi conserver le schéma des métadonnées● Au démarrage, on utilise le méta stream des schémas pour initialiser le

Schéma Registry● Problème de l’oeuf et de la poule

○ quel schéma pour les événements du stream schémas ?○ sur quel stream les enregistre-t-on ?

Schema registry

26

Séquence de démarrage

schemas

meta

stream-xyz

SchemaCreated<Metadata>

hash : “ab12ce9”Schema : { “typ..

SchemaCreated<InstanceUp>

hash : “56f3e05”Schema : { “typ..

InstanceUp

data : nodeXYZmeta : 56f3e05

SchemaCreated<SmthHappened>hash : “76b4ed0”Schema : { “typ..

SmthHappened

data : {‘d’:”123”}meta : 76b4ed0

SmthHappened

data : {‘d’:”abc”} meta : 76b4ed0

SchemaCreated<SchemaCreated>

hash : “2d163b9”Schema : { “typ..

Application démarrée27

Séquence de démarrage● Au démarrage, on initialise le Schema Registry● Comme le méta stream schémas est vide on enregistre les schémas “par

defaut” :○ Le schéma de SchemaCreated○ Le schéma des métadonnées

● On veut ensuite enregistrer le démarrage d’une nouvelle instance (InstanceUp)○ Le schéma d’InstanceUp n’existe pas, il faut l’enregistrer○ On émet un InstanceUp dans le méta schéma

● L’application est démarrée● Avant d’enregistrer un nouvel event, on vérifie que son schéma est

présent dans le registry, et on l’enregistre si besoin28

Evolution des types

A + BA A + B + Cauto auto

BA f: A → BRecord with aliased fieldsRecord auto

wideningnarrowing

A × BA × B × C Aauto auto

projectionadd field

29

Evolution des types

A × BA × B × C Aauto auto

projectionadd field

A × B × Czero auto

A × B × (C + Czero)

auto

auto

auto

add field with default value

auto

Czero <: C

A @@ B auto A

auto

A as logical type A @@ B

Type level

Serialization level

A @@ B <: A

30

Evolution des types dans le tempsA1 A2 A3

autoauto auto× A4?

fB1

stream-xyz : A1 + B1A2+ B1 A3 + B1

autoauto A4? + B1

A’4 + A3 + B1

Gérer la compatibilité entre A3 et A4.

Poser A4 à coté de A3.

Deux solutions :

auto

31

Conclusion● L’EventSourcing c’est bon, mangez-en !● Les principaux problèmes pénibles ont des solutions● Coming soon : Akira (https://github.com/vil1/akira)

32

MerciPour les questions, demandez Sam

33