Please welcome Sanju, Sanju and Pranin for us and enjoy.
Thank you guys, thank you.
Good morning guys, I am Sanju and he is Pranin, we work at Foonpei, yeah today we are going
to discuss about the lessons that we learnt while we manage the cluster first cluster
at the scale and the some of the problems we have faced and the solutions that we have
came up with, yeah Foonpei is the leading Indian digital payments and technology company
headquartered in Bangalore, India and it uses unified payments interface which is introduced
by government of India, so in India if you are thinking of any payment you can do it
using Foonpei app, this is how our Foonpei app home screen looks like.
And we have like a we see 800 k rps on our edge layer every day and we do 130 million
daily transactions, so this will generate lots of records and this will generate lots
of records in the documents that we have to store and as per the regulations in India
we have to store all of them in India only, so Foonpei has a private cloud where we store
all these things and we need a service to store and retrieve the files from the cloud,
we have developed a service called darkstore which will write the data to Glustreface and
which will fetch the data from the Glustreface, so coming to the question why did we choose
the Glustreface, we didn't wanted to have a metadata server because like we have lots
of small files and storing all the metadata, we didn't wanted it, so Glustreface has no
metadata server, so we went ahead with it and our team had earlier success in the Glustreface
project, so they were confident that Glustreface will work for our use case, so we are here
and this is the data flow to and from the Glustreface, so all the traffic is fronted
up by CDM and the request is forwarded to nginx and nginx will send the request to the API
gateway and API gateway can choose to store or retrieve any file from the, any file or
it can choose to send the request to any back-end service, now if the back-end service wants
to store this file or if it wants a file it can be a post or get request I mean like it
can store or it can retrieve, it will send the request to darkstore, now the darkstore
will store the data or retrieve the data from Glustreface servers and darkstore also uses
elastic search to store some of the metadata and it uses aero spike to store the earth related
info and some of the rate limiting features, it uses RMQ for asynchronous jobs like deletions
and batch operations and this is our team, yeah today's our agenda is an introduction
to Glustreface and then we will discuss about different problems that I have faced and the
solutions that we are using and we have some proposals as a roadmap.
What is Glustreface?
Glustreface it is a distributed file system that means whenever you do some write the
data is distributed across multiple servers, these servers have some of the directories
we call them as BRICS and this is where the data is actually getting stored, yes so this
is a typical Glustreface server, each server can have multiple BRICS, the BRICS will have
underlying file system where the data will be stored and in the root partition we store
the Glustreface configuration, go ahead, yeah this is how a 3 by 3 Glustreface volume looks
like, when I say 3 by 3 whenever a write comes to Glustreface mount point, so how Mr. 1 point
like we can mount Glustreface volume on any mission over the network and you can read
and write from that mission, now from the client where the mount is happened if any write
comes, so it is distributed across 3 sub volumes based on the hash range allocation, we will
talk more about the hash range in a coming slides and another 3 is transfer the data
is replicated 3 times, so whenever a write comes the data will choose one of the sub
volume and in a sub volume, sub volume is a replica set, here it is a 3, so it is replicated
thrice, over to Pranip, hello, yeah so let us look at some numbers that we see at phone
pay for dock store service and then to Glustreface, in a day we see about 4.3 million uploads
and downloads are 9 million, with peak upload rps as 200 and download rps as 800, the aggregate
upload size per day is just 150 GB, not a lot but the download size is 2.5 TB, so it
is completely read heavy workload and this is after a Syrian is fronting it, that means
only when the file is not available in your CDN, the call will come to Glustreface which
will download the file onto the CDN and then it will be served and this is how the rps
is distributed throughout the day, rps is request per second, so the uploads actually
are reasonably uniform from 6 am to 5 in the evening, then it tapers off for the rest of
the day, whereas the downloads are in bimodal distribution with one peak at around 12 pm
and another at around 7 pm, the latencies are function of the size of the file, so we
have post upload latencies with mean of about 50 ms to the p99 at around 250 ms, similarly
for gates the mean is around 10 ms and p99 is around 100 ms, let us look at the configuration
that we use at phone pay for Glustreface, we have 30 nodes in the cluster, each node
contributes 2 bricks and one brick corresponds to 10 TB and that is a ZFS pool, so 30 into
20 that is 600 TB of available capacity and we use replica 3, so the available size is
200 TB out of which 130 TB is in use at the moment, let us now go to the problems that
we face and how we solved it, I will start off with the capacity expansion problem that
we solved, then Sanju will take over and talk about the data migration problem that we solved,
I will talk about how performance issues are debugged and how we solved the problems using
that method, then Sanju will finish it off with maintenance activities that we do to
prevent the problems, before we talk about the capacity expansion problem, let us try
to understand a bit about the distribution, so the data is distributed across the servers
based on hashes, in this diagram we have 3 distribute sub volumes, each sub volume is
a replica 3, so when you create a directory, each of the directory in these 3 replica sets
will get a hash range and whenever you create a file or try to read a file, it will actually
compute the hash of the name and it will figure out which of these directories in these 3
sub volumes has that hash range and tries to get that file or store that file in that
node, so for folks who are well versed with database, this is more like sharding but the
entity here that is getting sharded is the directory based on the file names, alright,
so the files actually can have varying sizes, for example in our setup, the minimum size
would be less than a kb but the maximum size is like 26gb, so you will run into this problem
where some of the shards or distributes of volumes that you have would fill up the space
before the others, so you need to handle that part as well, so there is a feature in Glouceref
is called min-free disk where if you hit that level, when you create the directory again,
the hash range will not be allocated for the ones that met the threshold, so for example
here, even though there are 3 distribute sub volumes, data is going to only 2 because
the middle one actually has met the threshold, so the hash range will only be distributed
between the 2, 50% and 50% instead of one third that you would expect normally, so let's
talk about the actual process of increasing the capacity and why it didn't work for us,
when you want to increase the capacity that is you bring in more distributes of volumes
or shards, the way that you do it is you first you do something called as cluster peer probe,
that will bring the new machines into the cluster, then you do another operation called
add brick that will add the bricks to your volume, then you have to do something called
as cluster volume rebalance to redistribute the data among the nodes equally, so what
are the problems that we faced, when we did the benchmark, the rebalance had this application
latency impact in some cases up to 25 seconds and as I mentioned most of the P99 latencies
were just in milliseconds, so this is this will be like a partial timeout partial outage
for us, so this is not going to work for us, the other thing that we notice is for large
volumes the rebalance may take up to months and at the moment cluster FS rebalance does
not have pause and resume, so we can't do the maintenance activity in off peak hours,
that is one more problem, the other one that we have seen is when you do the data migration
when it is going from one distribute sub volume or shard to two distribute sub volumes, you
would expect 50 percent of the data to be transferred that's all right, but when you
are going from 9 shards slash distributed sub volumes to 10, you want to only migrate
like 10 percent of the data, but less than FS is still like transferring about 30 percent
to 40 percent like irrespective of what is the number of sub volumes are, so the rebalance
itself may take so much time with our workload that by the time we want to do the next capacity
expansion the rebalance may not even complete, so that is also not going to work for us,
so these are the three main problems that we have seen, so this is the solution that
we are using now, then there is a proposal as well, since we know that the hash range
allocation is based on the based on both the number of sub volumes and number of free sub
volumes, what we are doing is in our doxor application every day in the night we create
directories with a new basically, so the directory structure will be something like the namespace
that the clients are going to use slash year slash month slash day, so each day you are
going to create new directories, so based on the size that is available only the ones
that have space will get the hash range allocation, so you will never run into the problem where
you will have to do rebalance that much, because we have seen that with our workloads reads
are distributed uniformly and as we have seen the it is read heavy workload and writes are
just a few, so we were okay with the solution in the interim, but long term the solution
that we are we have proposed and this is something that is yet to be accepted, but there are
some POC that we did very few use jump consistent hash instead of the one that we have when
you are going from 9 to 10 here it is only about 10 percent that is getting rebalanced,
so that is what we want to get to this is something that we are focusing on this year,
alright over to you Sanju, so let us look at the problems that we have faced while migrating
the data, so we had a use case where we wanted to move complete data which is present in
one server to another server, so in clusterface the standard way of doing this is to use a
rebalance operation, sorry replace brick operation, so when you do replace brick operation there
is a process called a self filled demon which will copy all the data which is present in
the old server to new server, so to copy 10 TB data it takes around 2 to 3 weeks, so
that is like a huge time we wanted to reduce this time so we came up with a new approach
so let us understand few aspects of clusterface before we jump to the solution, so that we
understand our approach better, so the right flow in clusterface is something like this
whenever a right comes based on the hash range allocation plan is just spoke it will choose
one of the sub volume, so the data will go to all the servers in that sub volume, now
let us say we have chosen replicas at 0 and the right will go to all the machines in
that sub volume, it is a client side replication so the client will send the right to all the
machines and it will wait for the success response to come, so client will assume the
right is successful only when quorum number of success responses has come, let us say
one of the node is down, in our case we see like a server 2 either it can be a node down
or the brick process is unhealthy this can be unresponsive at times, so something happened
the right came to one of the sub volume and it went to all the three replica servers,
but server 2 did not responded with the success response, now server 1 and server 3 has responded
with the success response, so client it assumes that the right is successful, now when the
server 2 is back up we to have the consistency of the data server 2 should get the data which
it has missed while it was down, so who will take care of the job of doing this it is SHD,
so SHD is a daemon process which will read the pending heal data like whatever the data
that was missing we call it as a pending heal, so it will read from one of the good copy
in our case server 1 and server 3 are the good copies and server 2 is a bad copy, so
SHD will read the data from one of the good copy and it will write to server 2, so server
2 will have all the data once the self heal is completed healing the data, we will use
this as part of our approach as well, our approach is we will kill the brick which we
want to migrate like we want to migrate from the server 3 to server 4, so we have to copy
all the data right, so self heal is taking 2 to 3 weeks, here in our case we will kill
the brick and we have a ZFS, we are using ZFS file system, so we will take a ZFS snapshot
and we will transfer this snapshot from the server 3 to server 4, it is like a old server
to the new server and now we will perform the replace brick operation, while we are
performing the replace brick operation server 4 that is a new server will already have all
the data which server 3 had, once the replace brick operation is performed server 4 is now
part of the sub volume and the heals will take place from server 1 and server 2 to server
4, so now we have reduced the amount of data that we are healing, previously we are copying
all the data that is like a 10 TB of data from server 3 to server 4, but here in our
case we are healing only the data which came after killing the brick before doing the rebalance
replace brick operation, so the data we heal is reduced hugely, with this approach now it
is taking only 50 hours to complete this, that is also if we are using the spinning
discs it will take 48 hours to transfer the snapshot of 10 TB and 2 hours for the healing
of data, but it is only 8 to 9 hours if we are using SSDs, if we are using SSD it takes
like a 8 hours to transfer the snapshot and it takes around 40 minutes to complete the
heals, so that is like we came from 2 to 3 weeks to 1 or 2 days or 9 hours we can say,
we are using netcat utility, it gave us very good performance, it is like a 60% performance
optimization and we have in flight checksum at both the ends in the old server and also
in the new server, so that it is like we are checking whether we are transferring the snapshot
perfectly or not, we are not using any data and yeah it is at the time, I have kept the
commands that we have exactly used in this link and we also have a rollback plan, so
let us say that we have started with this activity but we have not performed the replace
brick yet, because once the replace brick is performed it will be something like this,
the sub volume will already have the server 4 as a part of it, before we perform the replace
brick that means when we are here, we can we do not want to do this anymore, all we
need to do is start the volume with the force, so that the brick process that we have killed
will come up, once it is up the good copies that we have SSD will copy the data from good
copies to bad copy are the old server, so that we will have the consistent data across
all of our replicated servers, yeah that is so easy and we want to popularize this method
so that it helps the community, yeah over to Prenet, yeah so this we will now talk about
the performance issues that we faced and how we solved them, this is the graph that we
have seen in our prod setup, while doing this migration when something happened that we did
not account for, so the latencies have shot up to 1 minute here and I have said that it
is supposed to be only milliseconds, so this is horrible, there was like 2 hours of partial
voltage because of this, so let us see how these things can be debugged and how they
can be fixed, so we have a method called GlusterVolumeProfile in GlusterFS, so what
you do is you start profiling on the volume, then you run your benchmark or whatever is
your workload, then you keep executing GlusterVolumeProfile in for incremental and it will keep
giving you the stats of what is happening to the volume during that time, for each of
the bricks that are there in the volume you will get an output like this, where for that
interval in this case interval 9, for each of the block size you will see the number
of reads and writes that came and for all of the internal file operations that you see
on the volume, you will get the number of calls and the latency distribution, min max
average latency and what is the percentage latency that is taken by each of your file
operation internally.
So, what we have seen when this ZFS issue happened is the lookup call is taking more
than a second which is not what we generally see, so we knew something was happening during
lookup operation, so we did an stress on the brick and we have found that there is one
internal directory called GlusterFS indices XRTROP, to list three entries it is basically
taking 0.35 seconds, so we so imagine this, so you do LS it will just show you three entries,
but it will take like 0.35 seconds sometimes it even takes a second, so we after looking
at this we found that ZFS has this behavior where if you create a lot of files in one
directory like millions and then you delete most of them and then if you do LS it takes
up to a second, so this bug is open for more than like two years I think, so we did not
know whether ZFS would fix this issue anytime soon, so in GlusterFS we patched it by caching
this information, so that we do not have to keep doing this operation, so now you would
not see it if you are using any of the latest GlusterFS releases, but yeah this is one issue
that we found and fixed.
The second one is about increasing the RPS that we have on our volume, so the there was
a new application that was getting launched at the time and the RPS that they wanted was
not what we are giving, so basically they wanted something like 300, 360 RPS or something
like that, but when we did the benchmark we were getting only like 250 RPS, so we wanted
to figure out what is happening, so we ran benchmarks on Prod Gluster itself and we saw
that one of the threads is getting saturated, so there is a feature in GlusterFS called
client IO threads where multiple threads would take the responsibility of sending it over
the network, so we thought let us just enable it and it would solve all our problems, we
enabled it and it made it worse like from 250 it went down, so we realized that there
is a continuation problem in the client side that we are yet to fix, so for now what we
did is to on the containers of Dockstore where it was doing only one mount, we are now doing
three mounts and distributing the uploads and downloads over yes, so can you repeat
the, oh yeah, no I didn't, it is a fuse mount, yeah the thread that is saturating is fuse
thread, yeah so the question is which GlusterFS client we are using, the answer is fuse client
and the thread that is saturating is fuse thread, so what we are doing is we have created
multiple mounts on the container and we are distributing the load in the application itself
like the uploads will go to all three and even downloads will go to all three, that
is one thing that we did to solve the CPU saturation problem, the other thing that we
noticed this is like part of the Gluster volume profile output where it will tell you for
each block what is the number of reads and writes, we have seen that most of the writes
are coming as 8KB, so later when we looked at the Java application Dockstore we saw that
the IO block that Java is using the default size is 8KB, so we just increased it to 128KB,
so these two combined has given us 2X to 3X the number and we also increased the number
of VMs that we are using to mount the client, so put all together we got something like
10X performance improvement compared to the earlier one, so we are set for maybe 2, 3
KB all right, so let us now go on to health checks, so for any production cluster some
of the health checks are needed, so I will talk about the minimal health checks that
needed for GlusterFace cluster, so GlusterFace already provides POSIX health checks, so it
is a health checker thread which will do a write of 1KB for every 15 or 30 minutes,
I mean seconds, so there is one option to set the time interval in which you want to
do this, so if you set it as a 0 that means you are disabling the health check, so you
can set it as like a 10 seconds or something, so it sends a write and check if the disk
is responsive enough and brick is healthy or not, if it did not get a response in a
particular time, it will kill the brick process, so that like we will get to know that something
is wrong with the brick process, so the other one we have is the rest of the things are
we have a script and we have some config, these are the things we have kept externally
kind of thing, the POSIX health checks are the one which come with the GlusterFace project,
so the cluster health checks that we have are like we have a config where we will specify
number of nodes in the cluster, so that is like a expected number of nodes in the cluster
and using the Gluster peer status or GlusterPoorList command, we can check the number of nodes
that are present in the cluster and we will check if both of them are equal, if not we
will write an alert saying something unexpected is happening and we will also check whether
the node is in connected state or not, so in the GlusterFace cluster the nodes can be
in different state, so it can be connected or rejected or disconnected based on how the
GlusterFace management daemon is working, so now we will see whether, so the expected
is all the nodes should be in a connected state, we will check whether the nodes are
connected or not, if the nodes are not connected then we will get an alert saying okay one
of your node is not in a connected state and we have some of the health checks for the
BRICS as well, so we have number of BRICS that are present in each volume in the config
and in the GlusterVolume info output you will get how many number of volumes that are present
in that volume and you will check if they are equal, the another check we have on the
BRICS, if the BRICS is not online we will get to know it by checking the GlusterVolume
status command and if it is not online you will get an alert saying that one of your
BRICS is down and so whenever the server is down or the BRICS is down there will be some
of the pending heels and you can check the pending heels using the GlusterVolumeHealInfo
command and if there are any pending heels you will see an entry, so if the entry is
non-zero then you will get an alert saying that okay you have some pending heels in your
cluster that means something unexpected, unwanted is going on that can be like a BRICS down
or node is down anything and we always lock profile info incremental to our debug locks
using the health check so that whenever we see some issue like the Prandit just spoke
about some of the issues that we can solve by looking at the profile info output, so
in such cases this output will be helpful so we always log into our log backup servers
and the exact commands that we are using are listed in this link, so we have some of the
maintenance activities so things can go back sometimes, so we have a replica 3 setup in
our production, so at any point of time quorum number of BRICS process should be up so that
the reads and writes can go on smoothly, so whenever we are doing something which might
take some downtime of the BRICS process or which can have some load on particular server
at that time we do it only on one of the server from each replica set so that even if that
server goes down or the BRICS process running on that server goes down we won't be having
an issue because there are two other replica servers which can like do all the reads and
writes, so we are doing few activities in this way, one is ZFS scrubbing, ZFS scrubbing
is about doing the checksum of the data, it will see if the data is in a proper condition
or not and we do migrations in this way only, so we are doing it on one server from each
replica set so that even if it is down for some time or something didn't work out we
are in a good place and upgrades also we will do in the same manner, we have done some contributions
so the data migration part that I have spoke it's a production ready we have used it in
our production and Pranit has given some of the developer sessions which has many internals
of Glastrophase, they are very useful for any Glastrophase developers who wants to learn
about many translators that we have in Glastrophase and recently we have fixed one of the single
point of failure which was present in the geo-replication feature, it was merged into
the upstream very recently last week and this year we are looking at another thing the hashing
strategy that Pranit has proposed, once it is accepted at the community we will take
it and develop it, yeah that's all we had folks, thank you.
Just want to let you guys know that the production ready thing, we actually migrated like in
total 375 TB using the method that Sanju talked about so it is ready, so yeah you guys can
use it, I think it should work even with butter, basically any file system that has a snapshot
feature it should work, yeah thank you guys, yeah I think we have a few minutes for questions
if you have any otherwise you guys can catch us there, yeah so the question is how do you
handle a disk failure, so basically the problem that I showed you where we had the ZFS issue
where it was taking like minutes of latency that was the first time it happened on production
for us and initially we were waiting for the machine itself to be fixed so that it will
come back again and it went for like a week or so and the amount of data that needed to
be healed became too much that it coincided with our peak hours, so now the standard operating
procedure that we have come up with after this issue is if a machine goes down or disk
goes down we can just get it back online in 9 hours so why do we have to wait, so we just
consider that node dead, we get a new machine we do whatever Sanju mentioned using ZFS snapshot
migration and we just bring it up, so do you have the ZFS backup somewhere, do you have
the ZFS backup somewhere, the answer is no you have the ZFS data on the active bricks
so you take a snapshot on the active bricks and do the snapshot trend, yeah one of the
good ones yes, any other questions, I think that's it I think, thank you guys, thanks
a lot.
Thank you.