Retour d'expérience sur Cassandra

MISC

n°

mars 2016

Par

Le but de cet article est de présenter et de donner un premier retour d'expérience sur la base de données distribuée Apache Cassandra (notée C*) à travers le cas pratique de TextMeUp, une solution de télécommunication (SMS/MMS/VoIP/vidéo).

Initialement, les applications de communication TextMe ne manipulaient qu’un « tampon » de messages effacés aussitôt qu’ils étaient relevés par le premier périphérique d’un utilisateur. Un couple de serveurs MySQL en mode maître-esclave a été provisionné dans ce but.

Or la nouvelle version, TextMeUp [1], permet dorénavant de gérer, stocker et synchroniser les messages et appels de plusieurs dizaines de millions de personnes, sur plusieurs mois glissants, entre périphériques mobiles et web. Ce contenu, qui ne représente que quelques dizaines de kilo-octets par personne (hors pièces jointes), pèse à l'échelle de notre parc utilisateurs plusieurs centaines de giga-octets dont l'intégralité doit pouvoir être interrogeable instantanément.

De plus, avec un volume de plusieurs millions de messages et d'appels par jours, une solution traditionnelle de type « base de données relationnelle » aurait vite forcé nos équipes d'exploitation à mettre en place des stratégies parfois complexes de partitionnement et de redistribution des données.

Enfin, si on considère à la fois la croissance de notre nombre d'utilisateurs et les fonctionnalités de plus en plus riches de l'application TextMeUp, nous devions dans l'idéal pouvoir stocker davantage de données ou augmenter le nombre de requêtes vers celles-ci juste en ajoutant des ressources matérielles.

Voici donc ce que nous souhaitions comme propriétés pour la base de données cible :

- pouvoir mettre à l'échelle facilement le stockage et la charge de travail ;

- être déployée sur plusieurs centres de données, voire sur plusieurs points de présence géographiques ;

- minimiser l'opérationnel (notamment les opérations de « sharding » ou de distribution des données) ;

- proposer une tolérance à la panne ;

- disposer d'un modèle de données assez souple.

Après évaluation de plusieurs solutions NoSQL ou distribuées, notre choix s'est porté sur Cassandra.

1. Concepts

1.1. Historique et écosystème

Cassandra a été initialement développée en Java par Facebook pour sa fonctionnalité de messagerie (avant Messenger) d'après les principes de haute disponibilité de Dynamo par Amazon [2] et de modèle de données de BigTable par Google [3]. Publiée en open source sur Google Code en juillet 2008, puis incubée par la fondation Apache en mars 2009, elle en est devenue un des projets de premier niveau en février 2010.

Sur la base d'une version majeure par an environ, Cassandra existe aujourd'hui en versions 2.1, 2.2 et 3.0 activement supportées. À ce jour, Datastax est l'entité commerciale la plus active de l'écosystème C* en fournissant documentation [4], environnement de développement et conseils. Elle emploie aussi les développeurs principaux de Cassandra (versions Community et Enterprise) et de ses connecteurs pour tous les langages majeurs (Java, .Net, Python, Ruby, PHP, Node).

Une version alternative à la version Java de Cassandra est en cours de développement sous le nom de ScyllaDB [5]. Visant essentiellement de meilleures performances grâce à une base de code en C++ permettant de s'affranchir des inconvénients d'une JVM, elle se substitue aisément à Cassandra, avec quelques limitations (ancienne version de CQL aux fonctionnalités partielles, pas de support SSL, protocole Gossip non compatible) [6].

1.2. Principes

Cassandra a été pensée dès le départ comme une base de données distribuée où chaque serveur du même cluster a la même configuration que ses pairs et discute avec eux via un protocole (Gossip) pour organiser le stockage et la charge de travail.

Ainsi, sachant qu'une même donnée peut être répliquée en permanence sur plusieurs nodes (via un facteur de réplication), on obtient une architecture sans point individuel de défaillance : en cas de crash ou d'indisponibilité d'un serveur, un autre node identique gère la même portion de donnée et peut assez aisément prendre en charge les traitements demandés. En cas de défaillance prolongée ou de modification de la topologie du cluster C*, la redistribution des données se fait de façon automatique et transparente.

De plus, une des spécificités de Cassandra est qu'elle « scale » de façon quasi-linéaire avec le nombre de machines allouées au cluster, que ce soit en capacité de stockage ou en charge de travail : passer de 6 à 12 machines identiques, par exemple, permettra de doubler l'espace disque disponible, ainsi que le nombre de requêtes pouvant être traitées.

Il est donc plus fréquent de déployer des clusters C* sur un nombre assez important de machines aux caractéristiques modestes (« commodity hardware ») plutôt que sur quelques monstres de puissance. À noter que dans ce cadre, l'utilisation de stockage centralisé de type NAS ne fait aucun sens (on distribue, on ne centralise pas) et qu'il est conseillé d'équiper ses machines de disques SSD locaux pour de meilleures performances.

Enfin, Cassandra s'appuie sur le principe de « eventual consistency », à savoir qu'à la cible tout accès à une donnée distribuée sur plusieurs nodes donnera le même résultat. Pour y arriver :

1. les différentes versions d'une même donnée, stockées sur plusieurs réplicas, sont comparées ;

2. puis une version réconciliée est retournée, en accord avec le niveau de consistance (ANY, ONE, LOCAL_QUORUM, ALL...) choisi par le client lors de sa lecture ou de son écriture.

Chaque écriture dans Cassandra est faite avec un horodatage à la micro-seconde près et de façon générale, c'est la dernière information inscrite qui l'emporte en cas de conflit (note : il est très important de synchroniser et de surveiller précisément les horloges de chaque machine via NTP [10][11]).

2. Distribution et stockage des données

Commençons tout d'abord par regarder comment déclarer une simple table (aussi appelée « columnfamily »).

Pour rendre la manipulation des données plus aisée, l'interface de programmation historique, Thrift, est dorénavant négligée au profit du langage CQL (accessible par exemple avec l'outil en ligne de commandes cqlsh), d'apparence très proche de SQL avec lequel il partage les concepts de table, enregistrement et colonne. Par exemple, voici une déclaration simplifiée de la table message utilisée par TextMeUp :

CREATE TABLE message (

user_id bigint,

message_id timeuuid,

content text,

status text,

PRIMARY KEY (user_id, message_id)

)

Outre les différentes colonnes et leurs types de données associés (int, text, timeuuid), c'est la déclaration de la clé primaire (PRIMARY KEY) de la table qui nous intéresse avant tout. Elle va définir comment les données vont être réparties et stockées physiquement sur le cluster. Elle est composée de deux parties :

- la clé de partitionnement (« partition key », user_id dans notre exemple) : c'est elle qui, faisant l'objet d'un hachage consistant, va déterminer sur quelle(s) node(s) du cluster l'enregistrement va être stocké. Dans notre cas, tous les messages d'un même user_id seront stockés sur la même node. On verra par la suite qu'il est possible de définir des clés de partitionnement composites, s'appuyant sur plusieurs colonnes.

- la clé d'agrégation (« clustering key », message_id dans notre exemple) : c'est celle qui va définir comment les données vont être inscrites sur le disque de la node cible, cela à des fins d'optimisation des entrées/sorties lors de leur lecture.

Pour finir ce survol de CQL, voici les types de données supportés :

- int, bigint, varint, double pour les entiers ;

- float et decimal pour les nombres réels ;

- ascii, text et varchar pour les données textuelles ;

- blob pour les données binaires ;

- boolean ;

- inet pour les adresses IP ;

- timestamp pour les dates ;

- uuid pour les identifiants uniques ;

- timeuuid : c'est un timestamp auquel on adjoint un identifiant unique pour garantir l'unicité d'un enregistrement ;

- list, map et set pour les collections d'éléments ;

- counter pour les compteurs distribués.

À noter enfin que les commandes SELECT, UPDATE et INSERT ont quasiment la même syntaxe que leurs pendants SQL, à ceci près qu'elles ne supportent ni jointure ni sous-requêtes. Je vous invite bien entendu à consulter la documentation du langage pour davantage de détails.

2.1. Distribution des données sur le cluster

2.1.1. Keyspace

Les données C* sont stockées dans un « key space » global au cluster, dont les propriétés sont :

- sa topologie, à base de nodes dans des racks répartis dans des datacenters ;

- son facteur de réplication : il est possible de définir sur combien de nodes une même donnée sera stockée, et ce par datacenter.

On peut par exemple imaginer un petit cluster gérant un keyspace textmeup réparti sur deux points géographiques :

- Côte Est des États-Unis (production) : 4 nodes sur 3 centres de données Amazon EC2, facteur de réplication : 3 ;

- Europe (site de secours) : hébergement traditionnel, sur un seul serveur, facteur de réplication : 1.

La déclaration CQL correspondante sera :

CREATE KEYSPACE textmeup WITH replication = {

'class': 'NetworkTopologyStrategy',

'us-east-1': '3',

'europe': '1'

}

2.1.2. Snitch

C'est ensuite un algorithme appelé « snitch » qui définit à quel rack et à quel datacenter appartient une node. À noter que Cassandra fait de son mieux pour ne pas stocker deux copies d'une même donnée dans le même rack pour une meilleure résilience à la panne. Voici une liste non-exhaustive des snitchs utilisables :

- SimpleSnitch: utilisé uniquement pour les déploiements dans un seul datacenter, déconseillé en production ;

- Ec2MultiRegionSnitch, GoogleCloudSnitch, CloudstackSnitch... : spécifiques aux déploiements sur des plateformes de Cloud Computing ;

- PropertyFileSnitch : s'appuie sur un fichier local qui référence toutes les nodes du cluster ;

- GossipingPropertyFileSnitch : s'appuie sur un fichier local à chaque node pour en déterminer l'emplacement, Gossip prenant le relais pour propager l'information à l'ensemble des nodes du cluster.

Gossip est le protocole utilisé par toutes les nodes du cluster pour échanger à intervalle régulier (toutes les secondes, par défaut) des informations à propos d'elles-mêmes ou des pairs qu'elles connaissent. De cette façon, tout changement dans le cluster est très rapidement propagé à l'intégralité de ses nodes.

2.1.3. Partitionnement

Maintenant que la topologie de notre cluster (et plus particulièrement de notre keyspace) est connue, nous pouvons y répartir l'ensemble de nos données.

Dans l'exemple ci-dessus, les enregistrements de notre table message seront distribués dans des partitions définies par la première partie de la clé primaire, à savoir la colonne user_id . Celle-ci va être traitée par un algorithme de hachage consistant, appelé « partitioner », qui va, quel que soit le type de donnée, retourner un token unique pour une clé donnée. Voici la liste des partitioners disponibles :

- ByteOrderedPartitioner : retourne une représentation hexadécimale des premiers caractères de la clé. S'il permet de conserver un ordre aux données, son exploitation est délicate et déconseillée ;

- RandomPartitioner : retourne un hash MD5 sur 128 bits (intervalle de 0 à 2127 - 1). Il est déprécié depuis Cassandra 1.2 ;

- Murmur3Partitioner : il retourne un entier 64 bits signé (intervalle de -263 à +263). Plus rapide que MD5, c'est cet algorithme qui est largement plébiscité et activé par défaut en production.

Ainsi, en utilisant Murmur3 distribué sur 4 nodes, on obtient les intervalles de tokens suivants :

Node	Début de l'intervalle Murmur3	Fin de l'intervalle Murmur3
N0	-9223372036854775808	-4611686018427387903
N1	-4611686018427387904	-1
N2	0	4611686018427387903
N3	4611686018427387904	9223372036854775807

Pour information, voici le code Python utilisé pour trouver les tokens Murmur3 associés à chacune des nodes, selon leur nombre :

python -c 'print [str(((2**64 / nombre_de_nodes) * i) - 2**63) for i in range(nombre_de_nodes)]'

Cette méthode de distribution force néanmoins à :

1. calculer les tokens de chaque node ;

2. éditer correctement le fichier de configuration spécifique à chaque node pour préciser quel token lui était assigné.

De plus, les opérations d'ajout ou de retrait de nodes dans le cluster ne sont pas des plus aisées, obligeant à propager des nouveaux fichiers de configuration et à bouger les tokens de node en node pour se conformer à la nouvelle topologie.

Ainsi, afin d'uniformiser la configuration de toutes les machines et de rendre la redistribution des données automatique et transparente, Cassandra s'appuie dorénavant sur un système de nodes virtuelles (vnodes), au nombre configurable de 256 par défaut. Chaque vnode porte ainsi 1/256ème des données et est assignée à une node physique ou à une autre selon leur nombre. Si la topologie du cluster change, Cassandra se charge de répartir les vnodes le plus équitablement possible sur les machines disponibles.

Mais simplifions et illustrons notre propos sur la répartition des partitions au sein d'un cluster C* et imaginons un cluster de 10 nodes gérant 100 tokens au total : le token 0 est assigné à la node N0, le token 10 à la node N1, et ainsi de suite jusqu'à N9.

Il est courant de représenter un cluster comme un anneau de nodes par datacenter (le « ring »), chaque node étant en charge d'un sous-ensemble équitable de l'intervalle de tokens retourné par le « partitioner ».

Fig. 1 : Un cluster de 10 nodes gérant 100 tokens.

En introduisant un facteur de réplication de 3, on s'assure que les 2 nodes consécutives à une node dans l'anneau répliquent également ses données. Ainsi, la node N3 finira par gérer les tokens 1 à 30. Et si on regarde où sont stockées les plages de tokens, l'intervalle 1 à 10 sera partagé entre N1, N2 et N3 ; 11 à 20 sera sur N2, N3 et N4 ; 21 à 30 sera sur N3, N4 et N5...

Sur cette base théorique, Cassandra propose une distribution et une réplication automatique des partitions sur toutes les nodes du cluster. La distribution équitable des enregistrements et des requêtes associées dépend ensuite directement du modèle de données associé aux tables, comme on le verra plus bas dans cet article avec l'exemple de TextMeUp.

Gardez bien à l'esprit que chaque machine du cluster, hormis sa localisation, a le même rôle et la même configuration que ses pairs. Il n'y a absolument aucune notion de node de contrôle/stockage/traitement comme on peut le rencontrer parfois dans d'autres solutions de bases de données distribuées.

2.2. Stockage des données sur disque

Un disque, mécanique ou même SSD, est de loin la source de données la plus lente d'une machine. Par conséquent, un des objectifs de Cassandra est d'être très efficace en termes d'entrées/sorties disque pour conserver un niveau de performance élevé. Une des façons d'y arriver est de privilégier les accès disque séquentiels (on lit l'intégralité des informations demandées d'une traite) plutôt que ceux aléatoires (allant piocher de-ci de-là sur le disque), que ce soit pour les lectures ou les écritures.

Attardons-nous un peu sur le chemin qu'emprunte un enregistrement lorsqu'il est inséré, modifié, ou supprimé de la node qui en a la charge.

Tout d'abord, l'enregistrement est modifié en mémoire et un journal de modification (« commitlog ») est inscrit sur disque de façon atomique afin de pouvoir reconstruire la donnée en cas de plantage de la node. L'espace mémoire dédié à la manipulation des données d'une table est appelé « Memtable » et est stocké dans la « heap » de la JVM.

Ensuite, au fur et à mesure que les modifications arrivent sur la node, les Memtables grossissent et le heap se remplit. Passé un seuil configurable, le contenu des Memtables est d'abord trié selon la clé d'agrégation de chaque table (message_id dans notre exemple précédent) avant d'être écrit sur disque via des écritures séquentielles, donc performantes. C'est ce même tri qui garantira les bonnes performances ultérieures en lecture, pour peu que la donnée soit lue (séquentiellement) dans l'ordre défini par cette même clé d'agrégation.

Chaque table Cassandra fait objet ses propres Memtables et SSTables : les données de deux tables distinctes ne sont pas mélangées dans une même structure en mémoire ni dans le même fichier sur disque.

Enfin, une fois écrites sur disque, les SSTables ne peuvent pas être modifiées, elles sont dites « immutables ». Par conséquent :

- une insertion ou une modification n'écrase pas la donnée présente dans une SSTable, elle stocke une nouvelle version de la donnée dans une autre SSTable ;

- une suppression n'efface pas la donnée précédemment écrite, elle marque une nouvelle version d'un marqueur de suppression appelé « tombstone » dans une autre SSTable.

Ce dispositif peut paraître étrange à première vue, mais il garantit une efficacité sans pareil en termes d'accès disque. Par contre, une partition est généralement stockée dans plusieurs SSTables. Ainsi, pour éviter la prolifération de ces dernières et d'avoir à lire systématiquement plusieurs fichiers pour une seule donnée, Cassandra déclenche régulièrement des opérations de compaction des SSTables.

Une compaction fusionne la donnée de chaque SSTable selon sa clé de partitionnement, en ne sélectionnant que la dernière version en date. Si cette version est un tombstone, la donnée est effacée. Encore une fois, le tri préalable selon la clé d'agrégation garantit une lecture et une réécriture séquentielles performantes des SSTables. Les nouvelles SSTables deviennent actives et les anciennes sont effacées à la fin de l'opération.

Les compactions se font de façon automatique et provoquent une augmentation temporaire des entrées/sorties disque et de l'espace utilisé. Plusieurs stratégies de compaction existent, selon le type de données et la charge de travail qui y est appliquée :

- SizeTieredCompactionStrategy : déclenchée lorsqu'un certain nombre de SSTables ont été écrites, elle est plutôt performante dans les environnements à écritures intensives ;

- LeveledCompactionStrategy : basée sur l'accumulation de SSTables de même taille, elle excelle quand les lectures sont majoritaires ;

- DateTieredCompactionStrategy : basée sur la date des données, elle est adaptée aux séries temporelles et aux données avec expiration.

Le sujet du stockage sur disque des données par Cassandra est complexe et ne peut pas être traité de façon exhaustive dans le cadre de cet article, je ne saurais que trop vous recommander de vous documenter davantage sur chacun de ses aspects (modèle de données, stratégies de compaction, configuration avancée).

3. Retour d'expérience

Voici maintenant notre retour après 8 mois d'exploitation de Cassandra en production pour TextMeUp.

3.1. Modèle de données

Détaillons maintenant les étapes de la modélisation de données de telle façon que Cassandra tire le meilleur profit de ses accès disque.

Pour rappel, TextMeUp est une application de communication capable de synchroniser le contenu d'un utilisateur entre plusieurs périphériques. Pour y arriver, nous avons décidé de stocker dans Cassandra tout l'historique des événements de l'utilisateur sur 60 jours, de telle façon qu'un périphérique puisse se synchroniser aisément :

- soit partiellement en remontant anti-chronologiquement jusqu'au dernier événement connu localement ;

- soit en intégralité, en itérant les 60 jours disponibles, dans le cas où aucune donnée locale n'est présente sur l'appareil.

La gestion de ce journal d'événements utilisateur peut s'apparenter au traitement d'une série temporelle (« time series »), discipline dans laquelle Cassandra est réputée pour délivrer de très bonnes performances [7][8].

En termes de charge de travail, nous ne considérons que les données d'un seul utilisateur à la fois, ce qui fait de user_id de notre exemple précédent un candidat parfait comme élément de clé de partitionnement :

CREATE TABLE message (

user_id bigint,

message_id timeuuid,

content text,

status text,

PRIMARY KEY (user_id, message_id)

)

Par contre, si on conserve ce modèle simple où tous les messages d'un utilisateur font partie de la même partition et sont par conséquent stockés sur une seule node (hors réplication), on risque d'avoir des « points chauds » dans le cluster, c'est-à-dire des nodes plus sollicitées que d'autres. En effet, certains utilisateurs sont beaucoup plus actifs (en nombre de messages, appels, requêtes ou périphériques) que d'autres.

Cassandra possède également une limite de deux milliards de cellules ( = enregistrements x colonnes) pour une même partition. Si ce plafond est hors propos dans notre cas, il peut toutefois poser problème dans des modèles plus avancés.

Pour mitiger ce problème, nous avons choisi de répartir les données d'un même utilisateur jour par jour sur des partitions différentes. Pour ce faire, nous allons ajouter une colonne date de type int à notre table. Elle sera de la forme suivante : AAAAMMJJ (exemples : 20160105 pour le 5 janvier 2016, 20150809 pour le 9 août 2015).

Nous allons l'inclure dans notre clé primaire, et plus particulièrement dans notre clé de partitionnement. On parle alors de clé composite, car elle inclut plus d'une colonne :

CREATE TABLE message (

user_id bigint,

date int,

message_id timeuuid,

content text,

status text,

PRIMARY KEY ((user_id, date), message_id)

)

Il est tout à fait possible d'ajouter ou de supprimer des colonnes à une table sans devoir réécrire les données (les SSTables sont immutables, après tout). À ceci près qu'il est interdit de le faire sur des éléments de la clé primaire : ça reviendrait à modifier à la fois la distribution des données sur le cluster et la façon dont les données sont écrites sur disque.

Nous avons donc dorénavant un modèle où les données d'un utilisateur, même très actif, se verront distribuées, jour par jour, sur l'ensemble des nodes du cluster, évitant des points chauds potentiels.

Par contre, lorsqu'un périphérique se synchronise, il le fait en considérant les événements de façon anti-chronologique de sa date courante jusqu'au dernier événement connu localement s'il existe. Il y a encore matière à optimiser le modèle ici.

En effet, message_id est de type timeuuid, c'est-à-dire un simple timestamp avec une précision d'une microseconde auquel on a adjoint un identifiant unique pour éviter que deux événements utilisateurs arrivant sur une même microseconde n'entrent en conflit.

Sans mention particulière dans notre modèle, l'écriture des données de notre table de la Memtable vers une SSTable se fait en ordonnant les données de façon chronologique. Ainsi au moment de lire la donnée sur le disque via une lecture séquentielle, Cassandra va considérer en premier les événements les plus anciens du jour, arriver au dernier événement connu par le périphérique client, puis commencer à retourner des enregistrements utiles.

Afin d'optimiser le parcours de la donnée sur le disque, nous tirons donc profit de l'option WITH CLUSTERING ORDER BY de CREATE TABLE pour écrire (et par conséquent lire) les données de façon anti-chronologique :

CREATE TABLE message (

user_id bigint,

date int,

message_id timeuuid,

content text,

status text,

PRIMARY KEY ((user_id, date), message_id)

) WITH CLUSTERING ORDER BY (message_id DESC)

Sur la base de ce modèle, nous pouvons maintenant écrire et lire les événements d'un utilisateur de façon satisfaisante. Mais afin de maîtriser l'espace disque utilisé, nous avons fait le choix de ne conserver que 60 jours d'historique utilisateur. Nous arrivons à ce résultat en écrivant nos données avec la clause CQL USING TTL 5184000 de façon systématique. Le temps de vie (TTL) de l'enregistrement se trouve alors repoussé à chaque modification.

Attention, dans ce cas, le TTL dans Cassandra se fait au niveau de la colonne. Ainsi, si vous changez seulement la valeur de status d'un événement via une requête de ce type :

UPDATE message USING TTL 5184000

SET status = 'read'

WHERE user_id = xxxxxx AND date = yyyyyyyy AND message_id = zzzzzzzzz

seul status est réécrit avec un TTL repoussé de 60 jours (5184000 secondes), les autres colonnes gardent leur TTL initial. Ainsi, sous 60 jours, vous pourriez voir apparaître des enregistrements avec les colonnes de la clé primaire et de status avec les bonnes valeurs, mais les autres champs totalement vides. Plutôt gênant, pensez à réécrire chaque colonne via SET à chaque mise à jour d'enregistrement pour l'éviter.

L'utilisation de TTL implique une multiplication des tombstones à expiration du TTL, ce qui met davantage de pression sur les compactions automatiques des SSTables. Après quelques tests d'endurance, nous avons trouvé que la compaction par niveaux (LeveledCompactionStrategy) était la plus adaptée à notre charge de travail.

À noter enfin qu'il est aussi possible d'activer la compression des données au niveau des tables, afin d'optimiser encore l'espace et les accès disque. D'où le modèle final pour notre table message :

CREATE TABLE message (

user_id bigint,

date int,

message_id timeuuid,

content text,

status text,

PRIMARY KEY ((user_id, date), message_id)

) WITH CLUSTERING ORDER BY (message_id DESC)

AND compression = {'sstable_compression': 'org.apache.cassandra.io.compress.LZ4Compressor'}

AND compaction = {'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'}

3.2. Déploiement

Le déploiement de l'architecture Cassandra de TextMeUp s'est fait avec la version communautaire 2.1.12. Nous envisagerons une migration à une version supérieure lors de projets futurs d'analyse temps réel de données.

Le tout a été installé sur un minimum de 3 instances Amazon EC2 i2.xlarge réparties sur 3 centres de données de la région us-east-1. Doté de 4 cœurs de processeurs et de 31 Go de mémoire vive, ce type d'instance est en adéquation avec le fonctionnement de C* et Java. Nous tirons notamment parti de ses 800 Go de SSD locaux pour stocker données et commit logs avec de bonnes performances.

Le système utilisé est une Debian 7 « Wheezy », avec un Java Runtime Environment 1.7 à jour et une taille de heap conseillée de 8 Go. Le tout est déployé via un cookbook Chef [14], notre outil de gestion de configuration.

Avec un facteur de réplication de 3 et compte tenu du snitch choisi (GossipingPropertyFileSnitch) avec de la correspondance suivante :

- datacenter C* = région AWS (us-east-1 dans notre cas) ;

- rack C* = datacenter AWS (us-east-1a, us-east-1b, us-east-1c…).

On s'assure qu'une même donnée est autant que possible localisée dans 3 centres de données différents au sein d'une même région AWS. De plus, nous gardons la possibilité d'étendre le cluster avec des datacenters hors de Amazon EC2, si besoin.

Nos serveurs d'application Python utilisent le connecteur Cassandra en version 2.5, configuré avec un niveau de consistance LOCAL_QUORUM pour les lectures et écritures. On s'assure ainsi que :

- le cluster est disponible si au moins 2 nodes sont disponibles ;

- le temps de réponse observé est celui des deux nodes les plus rapides. Dans un environnement réseau hétérogène comme celui de Amazon EC2, ça a son importance.

3.3. Exploitation

Avant même que notre cluster Cassandra soit mis en production, nous avons mis en place des tâches d'entretien, de sauvegarde et de surveillance afin d'en assurer la bonne marche.

Tout d'abord, sur chaque node de notre cluster, nous planifions (via des tâches cron) la réparation manuelle de chacune des tables de notre keyspace (via la commande nodetool repair) afin de forcer la réconciliation régulière de tous les réplicas de nos données [9]. Ces opérations provoquent des pics d'utilisation CPU, réseau et disque, mais sont absolument nécessaires pour assurer la consistance de notre cluster sur le long terme (notamment lorsque l'on utilise beaucoup de tombstones).

Au fur et à mesure que nous aurons davantage de nodes, il ne fait nul doute que nous aurons besoin d'outils de planification plus élaborés pour assurer les réparations. Nous sommes en train d'évaluer Cassandra reaper [12] (utilisé chez Spotify), mais OpsCenter de Datastax présente une bonne alternative pour les utilisateurs de leur offre Enterprise.

Nous avons ensuite assuré le backup de chaque node en scriptant la commande nodetool snapshot. Celle-ci va créer des liens durs vers les fichiers de SSTable, de façon à créer un instantané de nos données sans perturber la marche de chaque node. Nous exportons ensuite ces fichiers vers deux régions différentes de Amazon S3. Les procédures de restauration adéquates ont également été testées. À noter que DataStax OpsCenter permet aussi d'automatiser la sauvegarde des données d'un cluster Enterprise vers S3.

Enfin, concernant la surveillance du cluster, Cassandra offre à la fois une interface JMX, standard dans le monde Java, et un système modulaire d'export de métriques (Pluggable metrics [13]). C'est ce dernier que nous exploitons afin de remonter des données à notre service de monitoring Zabbix, qui se charge ensuite de produire rapports, graphiques et alertes nécessaires aux équipes d'exploitation.

Les métriques remontées sont très riches :

- compteurs, ratios, taux à 1/5/15 minutes, percentiles, moyennes...

- sur la JVM, les opérations (lectures, écritures, compaction) , les latences, les keyspaces, les columnfamilies...

Nous avons mis au point un modèle Zabbix qui nous permet de surveiller très précisément la bonne marche de nos serveurs C*.

Des solutions clés en main existent déjà certainement pour votre environnement de supervision habituel : DataStax OpsCenter, modèle pour Nagios, greffons pour NewRelic et autres services SaaS…

Conclusion

Voilà maintenant 8 mois que Cassandra est en production pour TextMeUp et nous n'avons à ce jour connu aucune interruption de service malgré :

- le remplacement d'une node devenue indisponible ;

- des latences périodiques accrues vers un des datacenters Amazon EC2 (le snitch dynamique utilisé est alors capable de router les requêtes CQL vers les nodes les plus performantes) ;

- des pics de fréquentation sur notre API (fêtes de fin d'année, promotions, bots).

Nous avons également fait des montées de version dans la branche 2.1 ou des changements de configuration sans incidence sur la production : en éteignant, mettant à jour et relançant chaque node tour à tour, le niveau de consistance choisi (LOCAL_QUORUM) est toujours respecté.

Notre utilisation de Cassandra reste plutôt modeste à ce jour, car TextMeUp est une nouvelle application à notre catalogue. Ainsi, si nous assurons sur 3 nodes l’intégralité du flux en écriture de tous nos utilisateurs, seuls 2% d'entre eux bénéficient pour l’instant des nouvelles fonctionnalités et donc provoquent une lecture de la donnée. Cela représente en pic :

- 1200 écritures par seconde ;

- 2500 lectures par seconde ;

- 200 Go de données compressées, répliquées sur nos 3 nodes (600 Go au total) pour stocker 60 jours d’événements utilisateurs.

Nous entamons une migration massive de nos utilisateurs dans les semaines à venir, ce qui devrait amener ce pourcentage de 2% à 90% sur le premier trimestre 2016.

Nous sommes pleinement satisfaits de Cassandra et envisageons :

- de l'utiliser dans nos futurs projets, notamment de l'analyse de données temps réel (les fonctions utilisateurs introduites dans Cassandra 2.2 nous seront d'une grande utilité)

- éventuellement d'étendre notre cluster sur d'autres zones géographiques (Europe, Asie) et d'autres prestataires afin de rapprocher la donnée de nos utilisateurs.