QCon London 2014, day Three

Lessons learned at scaling Twitter

Twitter nous a parlé de leur scaling, qui dans leur cas consiste à découper leur archi en modules de plus en plus petits et de redonder ceux-ci.

A l'origine, ils avaient une seule application rails, surnommée en interne "monorail". Une base de donnée mysql derrière, et toute la stack rails pour gérer le routing, la logique et la présentation. L'archi était faite selon les règles de l'art Rails, parfait pour une startup, le code était parfaitement connu par l'ensemble de l'équipe.

Néanmoins, avec la montée en charge du trafic, mysql est devenu un bottleneck qui ne pouvait pas scaler correctement. L'appli étant un unique bloc, la moindre modif de présentation voulait dire redeployer l'ensemble de la stack sur tous les serveurs. Ca les a forcé à faire du déploiement continu. Malgré tout, ils ont remarqué des problèmes de concurrence, et de performances.

Ils ont donc décidé de découper leur application en 4 applications, une pour chaque partie de leur métier : tweets, users, timelines et social graph.

Sur chacun de ces modules, ils ont séparé la présentation de la logique, en l'exposant sous forme d'API. Ils ont gardé l'appli monorail pour certaines pages, comme les FAQs.

Globalement, il ont un "TFE" (pas vraiment compris si c'était un acronyme général, le nom d'un composant ou un truc développé en interne), qui s'occupe de prendre les requetes qui arrivent et de les renvoyer sur l'appli correspondante, en gérant lui-même un buffer pour éviter d'overloader une appli. Dans les fait, c'est bien plus complexe que ça, mais le speaker est resté générique sur cette brique.

A chaque fois qu'un tweet est posté, il est renvoyé vers les 4 modules, qui vont chacun le gérer en fonction de leurs besoins. Au final, on a un même élément (le tweet) qui est stocké à 4 endroits différents.

La timeline twitter fonctionne comme une inbox. Chaque utilisateur possède la sienne. Ces inbox sont stockées dans une immense base Redis distribuée. Chaque clée est le twitter id, et chaque valeur est la liste des X derniers tweets. Si un utilisateur ne vient pas souvent sur twitter, il n'aura pas d'entrée dans la base Redis (mais un service de fallback peut lui générer), par contre les utilisateurs réguliers ont leur inbox très facilement accessible.

Bien sur, le Redis ne stocke pas la totalité des tweets, mais juste leurs ids. Une autre base s'occupe de faire la correlation entre un id et son contenu. Le service de timeline va donc ensuite s'occuper d'aller récupérer le contenu des tweets pour constituer la timeline.

Quand un tweet est posté, il est aussi passé dans Ingester, qui est la brique qui s'occupe d'alimenter la base EarlyBird (shardée) qui s'occupe de la recherche.

Ils ont aussi un système de push. Quand un tweet est posté, il passe dans leur application de push, qui va ensuite pusher ce tweet à tous les abonnées. L'appli à un excellent débit de 30Mo/s en input.

Pour réussir à faire communique tout ce beau monde, ils ont développé twitter-server, un template opensource de serveur, qu'ils utilisent en interne. Créer une archi composée de ces serveurs leur permet d'obtenir des metrics générique sur l'ensemble du réseau mais aussi sur chaque noeud. Ca leur simplifie le discovery, le deploiement et le loadbalancing de tout ca.

Tous leurs serveurs fonctionnent sous forme de fonction. Ils acceptent un input et retournent un output, et peuvent donc être chainés. L'output est une Future (c'est le terme dans le monde Scala dont l'équivalent est une Promise en Javascript). C'est un objet qui représente le résultat du traitement dans le futur; il n'est pas nécessaire d'attendre le résultat pour pouvoir déjà commencer à travailler sur celui-ci, il se mettra à jour une fois le traitement effectué. En utilisant une telle architecture, ils peuvent manipuler des objets simples représentant des traitement et décider de l'optimisation de ceux-ci s'ils souhaitent les effectuer en parallele ou en séquence.

Ils ont en interne deux équipes. L'une qui s'occupe du besoin et des appels qui vont être nécessaire pour faire une telle requete, et une autre équipe qui s'occupe d'optimiser la concurrence et/ou la parallelisation des Future qui la compose.

Le simple fait de séparer leur métier en plusieurs modules fait que leurs équipes sont aussi séparées verticalement. Un peu comme à Spotify, chaque team est chargée d'une feature et donc touche à tous les aspects de cette feature, depuis la présentation jusque la configuration de la JVM.

Twitter traite énormément de données, et ils ne peuvent pas facilement investiguer la suite de requetes d'un utilisateur en particulier si celui-ci leur remonte un bug. A l'inverse, ils travaillent sur les statistiques de la totalité du parc, et l'aggrégation d'erreurs. Si un event d'erreur commence à se produire de manière plus fréquente, ils vont investiguer, mais ils ne perdent pas de temps à débugguer chaque erreur, ils se contentent de traiter les errreurs les plus importantes en priorité. Chaque équipe possède un dashboard temps-réel des métriques des services qu'ils utilisent (pas forcément uniquement ceux qu'ils ont écrit, mais aussi ceux qu'ils consomment). Ainsi, si quelque chose déconne, ils peuvent le voir facilement.

Ils ne font pas de tests de performances en interne. Leur masse d'utilisateur est telle que leurs tests ne pourraient pas réussir à les simuler correctement. Ils sont aussi tellement dépendant de l'actualité, qui est random, qu'il n'y a rien de mieux pour eux que de tester directement en prod les performances. Mais ils font du déploiement progressif, comme Etsy, en n'ouvrant que petit à petit à de plus en plus d'utilisateurs.

Une question lui a été posée sur ce qui leur posait encore des soucis de scaling en prod. Il s'avère qu'un si un compte avec des millions de followers envoie un tweet à un autre compte aussi populaire, tous ceux qui suivent les deux vont le voir dans leur timeline. Ca fait une sacrée jointure de de plusieurs millions d'utilisateurs, c'est costaud, mais ça passe.

Il lui a été demandé si le tweet du selfie des Oscards avait posé problème à leur DB. La réponse est non, mais ce tweet a tellement fait parler de lui que plein de monde qui ne venaient plus sur Twitter sont revenus, et il a donc fallu remettre à jour leur timeline oubliée depuis des années, pour tout le monde en même temps, ce qui a ralentit le service.

Finalement, il nous a expliqué qu'au Japon, il y avait un mot magique dans un film de Miyazaki qui permettait de détruire la technologie environnante, et le grand jeu au Japon est que tout le monde tweet ce mot en même temps dès qu'il est prononcé à la télévision. Heureusement, chacun de ces tweets étant un objet distinct, ils sont correctement distribués dans leur shards et ça ne détruit pas Twitter.

What's Beyond Virtualization

Aujourd'hui, on va très loin dans la virtualisation. On peut instancier des VMS en quelques secondes sur le cloud, sur lesquelles on fait popper des containers Docker, à l'intérieur desquelles on configure tout à base de Chef et de Puppet.

Mais qu'elles sont donc les zones que l'on ne maitrise pas encore complétement, les zones dans lesquelles on peut encore s'améliorer ?

On a toujours les limites géographiques, le fait que nos instances doivent rester géographiquement proches pour limiter la latence. Ou alors pour des questions légales, il est nécessaire que nos données soient dans certains pays et pas dans d'autres. Si jamais un de nos noeuds tombe, pour une raison hardware ou software, il ne faut pas que le reste de l'architecture tombe ensemble. Il faut que toutes ces briques s'imbriquent correctement pour fonctionner toutes ensembles, mais que l'une ou l'autre puisse tomber sans tout emporter avec elle.

Il y a aussi des questions de rapidité. Il faut que nous soyons capable d'ajouter ou de supprimer des noeuds à notre archi rapidement, que les VMs bootent vite, et qu'elles puissent s'ajouter dans la cartographie existante. Il faut pour cela qu'elles se fassent connaitre des autres machines, que leur IP soit connue, et que les configurations (load-balancers, etc) de toute l'archi soient mises à jour directement.

Et si une brique tombe, comment réagir ? Est-ce qu'on monitore l'ensemble pour s'adapter quand un node fail, ou bien on attends de le découvrir manuellement ? Est-ce qu'on fait confiance à notre système de monitoring pour découvrir et/ou corriger ce genre de problèmes ? Et que faire si lui-même est en panne ? Est-ce qu'il faut monitorer les monitor ?

Et c'est sans compter les cas plus subtils, quand un node n'est ni complétement mort, ni vivant, qu'il est malade, lent, peu performant. Un système de Chaos Monkey est très bon pour tuer des services et s'assurer que tout fonctionne correctement avec des nodes en moins, mais comment simuler un système malade, avec des nodes présents, mais qui agissent bizaremment ?

Après toute ces questions, tout ces points sur lesquels des outils sont en cours de développement mais aucun n'est encore complétement mature, notre speaker à proposé sa solution (générique) d'un OS pour MultiDatacenter. Quelque chose qui gère la liste des serveurs, leurs versions, leurs connections, comment ils communiquent, comment ils scalent, et qui permette d'ajouter de nouveaux nodes facilement et de modifier des configurations à un niveau global plutot que node par node. Plutot que d'utiliser des tas de petits outils pour gérer les différents aspects de l'archi, plutot avoir une base solide, fondée sur ce qu'on a appris de chacun de ces outils, mais les packager en un OS pour permettre une meilleur interopérabilité.

Docker in the cloud

Je n'arrete pas d'entendre parler de Docker en ce moment. Et j'ai encore du mal à réussir à savoir exactement comment il s'intercale avec des outils comme Vagrant, Chef, Puppet ou Capistrano. Cette conférence m'a expliqué clairement ce que fait Docker, avec des exemples très simples, sans doute directement tirés du Tutorial, mais il n'y a rien de mieux qu'un peu de pratique pour se rendre compte de la puissance d'un outil.

Tout a commencé par un sudo docker run -i -t ubuntu:12.04 'echo Hello World' pour initialiser un container docker avec un ubuntu 12.04 et lui faire afficher Hello World.

Dans le jargon Docker, une image est un container statique, une template qu'on peut lancer. Une fois lancé, on possède une instance de cette image. Chaque instance sur le système possède son propre id, et on communique avec cette instance à partir de son id. On peut donc avoir plusieurs instances de la même image en parallele.

De manière inverse, il est possible de créer une image à partir d'une instance, si par exemple on vient de la configurer aux petits oignons et qu'on veut pouvoir repartir de ce point là plus tard. Le Dockerfile quand à lui est un simple fichier texte qui indique la configuration de base du container : OS et version à charger, packets à installer, etc. Il existe une liste collaborative de Dockerfile pour les taches les plus communes déjà disponible sur github.

On peut déplacer des fichiers depuis le host vers un container, ou lancer des commandes directement depuis le host à l'intérieur d'un container. On peut aussi mapper des ports du host vers des ports du container, ce qui est très pratique pour hébérger par exemple plusieurs sites sur le même host, mais qui ont des stacks techniques complétement différentes (différentes versions de ruby, node, apache, nginx, etc). Ca permet par exemple de n'avoir qu'une seule VM chez Amazon, mais de multipler les stacks techniques dessus.

L'avantage de docker est que la création d'instance est très rapide et cheap en ressource. Si j'ai foiré ma config dans mon container, je peux juste le tuer et le relancer sans avoir à me soucier de régler de potentiels problèmes que mon erreur de config aurait pu avoir créé. Docker utiliser un système de cache intéressant sur les commandes. Si j'ai déjà lancé une certaine commande dans une instance, alors je peux relancer la même plus tard et il connait déjà l'outpur de la commande, donc il peut me la rejouer depuis le cache. C'est particulièrement utile pour l'installation de paquets depuis les repositories.

Il reste quelques edge cases. Par exemple, si le gestionnaire de packet de votre distribution (ubuntu par exemple) contient des versions outdatées de l'app dont vous avez besoin, il va falloir passer par d'autres mécanismes que l'install classique. Si les paquets ont aussi été mis à jour entre deux commandes, le systeme de cache de docker risque d'intérférer et de vous installer une version trop ancienne.

La gestion des quotas de disque entre les containers et leurs hosts ne semble pas encore complétement terminée non plus. Idem pour les logs des containers qui semblent s'étaler sur les logs des hosts.

Finalement, le mapping des ports du host vers les containers reste encore manuel à grand échelle (dans une appli shardée) et est complexe à maintenir quand on souhaite faire communique des containers d'un host A vers un host B.

Netflix

Netflix nous a fait un petit topo de leurs problèmatiques de scaling et les solutions apportées. C'est difficile de se rendre compte de l'apport d'une telle conférence, leurs problématiques étant tellement éloignée de celles du commun des mortels.

Netflix streament du contenu vidéo à plus de 44 millions d'utilisateurs à travers le monde. Si quelque chose fail chez eux à un moment crucial, leur taille fait qu'on en parle sur Twitter, aux infos, et que leur service consommateur explose. Ils prennent donc le fail très au sérieux et... essaient de failer le plus souvent possible. Ils savent qu'avec un système de la taille de celui qu'ils ont (33% de la bandwith mondiale) il va forcément y avoir des fails, mais ils préférent que ce soit eux qui les déclenchent plutot que ça arrive de manière aléatoire au mauvais moment.

Ils ont trois niveaux d'erreurs. Les plus importantes sont celles qui affectent tout le monde, et qui ressortent dans les journaux ensuite. Viennent en second les erreurs qui remontent directement aux utilisateurs, mais qui sont isolés. Enfin, il y a les erreurs "normales", celles que les utilisateurs ne voient pas et qui impactent les analytics, l'A/B testing, etc et qui se résolvent automatiquement.

Ils testent la résilience de leur système avec leurs Chaos Monkeys, en production. Le premier Chaos Monkey était un script qui tuait aléatoirement un node dans leur infrastucture, et qui devait permettre de prouver que leur système savait s'adapter et trouver des routes alternatives, sans qu'aucun node ne soit un SPOF.

Ils l'ont ensuite upgradé en Chaos Gorilla, qui va tuer un serveur DNS. Chose rare, mais pas impossible. Ils ont même un Chaos Kong, qui va complétement détruire une région (ensemble de zones Amazon).

Leur motto est "Not if, but when. Everything will fail." Ils cherchent à réduire au maximum le time to detect et le time to recover. Leurs deux axes principaux pour réussir à avoir une infrastructure qui tienne la route sont l'isolation et la redondance. L'isolation permet qu'un node puisse tomber sans qu'il impacte les autres. Pas de SPOF. La redondance permet que si un node d'un ensemble tombe, sa charge peut se répartir sur les autres le temps qu'un nouveau node soit remonté et puisse reprendre sa charge. Ils ont poussé ce système tellement loin que si une zone ou région tombe, les autres de leur infra peuvent assurer la charge en attendant.

La redondance est obligatoire, mais elle ne doit pas être considérée comme un élément primordial du système. C'est à dire qu'il leur est inconcevable d'avoir des serveurs en un seul exemplaire. Chaque élément doit au moins être redondé une fois. Mais il ne faut pas compte sur ce second élement dans les cartographie, il est là en fantome, si le premier tombe, mais on ne lui envoie pas de charge normalement. Dans les faits cela signifie qu'ils doublent leur archi en prévision, dès le début.

Ils font tourner leur Symian Army en production, qui casse des trucs dans tous les sens, et ils comptent sur l'intelligence de leur système pour que la qualité de service n'en soit pas impactée. Si jamais malgré tout des indicateurs restent au rouge trop longtemps, cela signifie qu'une partie du système ne sait pas sa guérir, et ils vont alors travailler là dessus.

Quand un serveur commence à accuser des signes de faiblesse (latence, performances en chute), ils instancient un nouveau serveur identique et routes les requetes dessus petit à petit jusqu'à ce que le serveur initial soit libre de toute requete. Alors ils le tuent.

Comme je disais en introduction, leurs problèmatiques sont à des années-lumières de ce à quoi on est confrontés normalement. Pouvoir peter des zones completes Amazon et se permettre d'avoir tous ses serveurs en double, ce n'est pas donné à tout le monde.

La majorité des outils dont ils ont parlé dans la présentation sont disponibles sur leur Github.

Scaling Continuous Deployment at Etsy

J'en étais déjà convaincu après la première journée de conférence, mais cette dernière présentation d'Etsy n'a fait que le confirmer : les gars d'Etsy sont des brutes.

Chez Etsy, ils insistent fortement pour que chaque employé, dès son premier jour de travail, puisse déployer en production ses changements. Le déploiement ne doit pas faire peur, c'est quelque chose que tout le monde doit pouvoir faire. Ils font en moyenne 50 déploiements par jour.

Pour cela ils utilisent un outil interne (open-sourcé désormais), nommé Deployinator, qui est une page web accesible à tout Etsy et qui permet de faire un "One-click deploy" de ses modifications. Ils possèdent aussi un environnement, nommé Princess, qui est une copie-conforme de la production, qui permet de faire des tests à l'échelle. Si ca passe sur Princess, ça passera en production. Ils ont aussi un sharding de 200 jenkins dans des containers Dockers sur des disques SSD pour faire tourner leurs tests en parallelle avant chaque deploy.

Un deploy leur prenait initialement 15mn, et ils trouvaient ça trop long. Ils ne déploient pas à chaque commit, mais attendent d'avoir 8 commits pour faire un déploiement. Pourquoi 8 ? Empiriquement c'est ce qui marche le mieux, ils ont essayés avec 6 ou 10, mais ça ne leur convenait pas.

Quand tu veux déployer tes modifs à Etsy, tu t'inscris au "Push train". Le train prends des wagons de 8 personnes. Le premier du groupe de 8 est considéré le conducteur et c'est lui qui s'occupe des opérations. Généralement les modifications des 8 personnes touchent à des parties différentes du code, il n'y a donc pas de conflit, mais s'il y en a , c'est le conducteur qui est chargé de coordonner leur résolution. Vu qu'Etsy fonctionne énormément avec le feature flipping, ils n'ont pas plusieurs branches dans leur repo.

Ils se sont aussi fixés comme règle de ne jamais pusher dans le même train le code d'une feature ainsi que la config qui active cette feature. On sépare toujours le code de feature et le code de config. Il y a souvent des push de config, et les mettre dans le même train que les pushs de feature ralentissait le déploiement.

L'appli d'Etsy est un énorme monolithe en PHP qui contient la totalité du site. Pour déployer sur leurs serveurs, ils utilisent une version de rsync qu'ils ont patchés qui peut faire copies de fichier en parallele. La totalité du code de leurs serveurs n'est pas stocké sur disque sur leurs serveurs, mais directement en RAM. Comme ça, les temps de déploiement sont beaucoup plus rapide, et cela leur assure que si un serveur plante pendant un déploiement, il n'est pas dans un état instable avec des fichiers d'une version et d'une autre en même temps. Si ça plante, la RAM est vide, et on y réinstancie la dernière version directement.

Pour pouvoir faire des déploiement atomiques, ils déploient les nouvelles versions dans un nouveau dossier du serveur. Une fois qu'il est complétement copié, ils modifient la configuration Apache pour lui dire d'aller chercher les fichier dans ce nouveau dossier. Ils ont quelques soucis avec Apache et PHP qui continuent d'aller lire les fichiers dans l'ancien dossier pour toutes les requetes qui sont encore ouvertes, mais ils ont créé des modules Apache et PHP pour règler ces problèmes. En faisant ainsi, ils n'ont pas besoin de recharger ni Apache ni APC/Opcode.

Dans leur ancienne architecture, chaque serveur allait puller directement le nouveay code depuis leur serveur de deploy. Cela créait trop de bottleneck. Aujourd'hui ils déploient vers 2 serveurs de déploiement master, et les serveurs de prod vont puller depuis ces deux serveurs.

Ils ne suppriment pas les fichiers de la version n-1 quand ils déploient la version n, pour éviter qu'un utilisateur qui est sur le site à ce moment et qui a initié une requete vers un fichier qui n'existe plus se retrouve avec des erreurs 404. A la place, ils attendent le déploiement n+1 pour supprimer les fichiers de n-1.

Il leur a été demandé comment il géraient les changement de schéma de leur base de donnée. Pour cela il ont un process un peu plus long qui passe par l'aval d'un dev DB. Si un dev a besoin d'une modification de DB, il doit définir la liste des tables et des champs dont il a besoin et les fournir à un DB Master. Chaque mercredi, les DB Master prennent les demandes de modifs, discutent avec le dev pour voir exactement ce dont il a besoin et font les modifs nécessaires. Ils mettent les modifs online et redirigent alors 50% du traffic vers ces nouvelles structures, et gardent les autres 50% sur les anciennes versions. Si tout passe correctement pendant une semaine, ils passent alors les 100% sur les nouvelles versions et suppriment les anciennes.

Quand il est nécessaire de changer de la config au niveau de Apache ou autre service, ils utilisent Chef. Mais Chef est configuré pour ne jamais relancer les services lui-même. Il va juste mettre à jour sa config si le master a changé. Ce sont les admins qui vont ensuite relancer manuellement une instance avec les nouvelles configs. Si ca passe, ils en relancent d'autres, et ainsi de suite avec de plus en plus d'instances pour ne pas impacter tout le parc en une seule fois en cas d'erreur.

Ils cherchent encore à aller plus vite. Leur but est de réussir à faire des déploiements en moins de 2mn. Ils savent que leur vitesse de déploiement est un avantage compétitif. Si des concurrents déploients plus vite qu'eux, c'est dangereux pour eux, ils veulent donc avoir une longueur d'avance. Leurs prochaines améliorations porteront sur les tests automatisés de leurs déploiement, et sur l'aggrégation d'encore plus de métriques de vitesse pour identifier les bottlenecks des déploiements.

Quand on leur demande "Pourquoi PHP ?" ils répondent qu'à un moment ils avaient plein de technos différentes, du node, du ruby, du python, etc. Mais que finalement c'était trop compliqué à gérer, et qu'en se limitant au PHP le partage de compétence est bien plus facile. Ils peuvent aussi optimiser encore plus un langage parce qu'ils le connaissent très bien et se permettre de recruter les meilleurs dans ce domaine.

Tags : #qcon

Want to add something ? Feel free to get in touch on Twitter : @pixelastic