Bull & NestJS = Achieving Scale in Node.js

node.js is single threaded which means it's only going to use up to a single core so how do we break up large tasks that require a ton of processing power beyond a single core the solution is distributed asynchronous processing that happens over multiple instances of your app that are scaled out horizontally in order to achieve this we need some sort of message broker that can coordinate messages on a queue like system and that is exactly what we're going to do in this video with an sjs and Bowl mq we're going to see how we can process messages in parallel scale out and even reprocess messages if they die halfway through this is a really powerful pattern to scale out in node.js and do a lot of work so let's go ahead and jump right in and see how we can do that in this video I'll see you there alright so we're going to start off by using the nest CLI to create a new Nest project and of course like all my other videos I'll include a link in the description to a GitHub repository that contains all of this code so we'll go ahead and init this project by typing in Nest new and we will call this Nest JS bull mq I'm going to go ahead and use pnpm as my package manager and wait for the installation to finish so after we finish installing the project we'll go ahead and CD into the nest.js bull mq directory then we'll go ahead and use pnpm to install nest.js bull as well as bull itself here so we're also going to install a development dependency here the types for bull as well once we have all of these dependencies we can run pmpm Stark Dev here to start our server up in development mode and watch for changes so I've gone ahead and opened up the project in vs code here and we can very easily set up our Bowl mq module by opening up our source directory and going into our app dot module and in our Imports array we're going to go ahead and import the bull module from nest.js slash Bull and we're going to call for root on this and then provide a configuration object that will tell it how to connect to redis which of course is the underlying data store that is going to be persisting our messages so we provide a key redis here that has the host name here for now we're going to hard code it to localhost however we will be injecting environment variables in here shortly and then we're going to provide the port number of the redis instance and for now again we'll hard code this but this will eventually be an environment variable so once we've actually initialized the bull module here we're then going to go ahead and also import bull module again here and then we're going to call register Q here and register Q is what we're going to use to tell the bowl module which cues to register and this will also provide the injection token for the queue which we will use later on so register Q takes any number of objects here depending on the number of cues you have and inside of each queue you provide the name of the queue so in this case we'll provide the name here and we'll call this Q name trans code you could think that perhaps we're doing some video or audio transcoding any really heavy CPU intensive task that we want to distribute over multiple instances of our app that we could put on this queue really makes sense to put here instead of blocking the node event Loop and slowing up request responses we can process this transcode queue in the background asynchronously so once we have our Bowl module set up I'm going to go inside of my app controller and by default we have our get route here to get hello we're going to add a new post route to our app controller here that will be called async transcode and here we're going to return this Dot appservice.transcode and now we're going to go inside of the app service and add that asynchronous transcode functions so now in our app service we want to actually initialize this transcoding and put a message onto our transcode queue so we're going to add a Constructor in our app service here and now we're going to use the inject keyword here to be able to inject our transcode Q now if you remember in our app module I told you that this but the name here will also be the injection token we're going to use so we can actually pull this out into its own constants we'll have a new constants dot TS file here and we can export a const called transcode q and we'll call this transcode so we have this in one place and I'll go back into my app module now and we can change the name out here to transcode Q from constants and in our app service we'll do the same thing we'll inject transcode q and we will say this is a private read-only transcode Q of type Q from bull so now all we have to do to add a job to this queue is to call await this dot transcode Q dot add and inside of add we can add any data we want to this queue so I'll add some sample information here where we perhaps specify the file name of the file that we're going to transcode so that our consumer will be able to find it so we'll just add a dummy file.mp3 here there's also a ton of other options that we can add to this job as an options object here there's also a number of other options that we can specify in this options object here including things like the priority for this job delay a timeout back off retry options and more however for now we just want to send off this job and see that we can consume it so the next step is to create a consumer for this job so I'll create a new Trans code dot consumer and this will be decorated with the processor decorator here and then we provide the name of the queue like always so we will have the transcode Q provided in here and I will export a class called transcode consumer and then with inside of this class we also need to add the process decorator here on the function that will actually process this job so I'll have an async transcode function here that takes in a job of type job from bowl of type unknown for now and then we actually have this job so for now all I want to do is just log it out and see that we actually have received this job and we're actually going to go ahead and quickly change this to use the nest logger so we'll have a new private read-only logger variable here that will create a new logger we'll provide the transcode consumer.name here for a little more context and then we'll use this Dot logger.log and log the job out here instead and then one other fix that we need to make is in the app service instead of using the inject keyword here we actually special a decorator inject Q from nest.js bull so make sure we use inject Q instead of inject so now if we go back to our application you can see we're still running with no errors now so we're ready to go ahead and start testing out our queuing mechanism okay so I've opened up Postman here and I have a route at localhost 3000 slash transcode it is a post request which corresponds to our app controller transcode route here so we'll go ahead and execute this request and we have a 404 can't post this route because we need to add a transcode prefix here to our route so now if we send this off you can see we're sending our request and it's actually hanging right now it's not completing and this is actually what we'd expect because right now bolem Q is trying to send a message to this transcode Q however it's not going to be able to of course because we don't have redis actually running in the background so I'll go ahead and cancel this request and we need to actually start up an instance of redis to be able to connect as we have specified in our app dot module here so we can easily do this I've opened up a new terminal shell here and of course make sure you have Docker installed and running on your machine and then we'll use Docker run and I'll expose a port on the container so by default in this redis image the default Port is 6379 which is the same port that we specified here 6379 and I want to map that to 6379 on my machine and then we can just type in redis which is the default Docker image here and you can see we're now ready to accept connections for this redis container so now if we send off a request we can see we got that 201 created so the request did succeed here however if we go to our terminal we don't see the processor or the consumer actually logging out the job and the reason for that is because we didn't actually add this transcode consumer to our application we need to actually add it to our providers array here so that Nest knows about it so make sure we add our trans code consumer to our providers array here so now if we go back to our terminal here we can see our trans code consumer picked up that message once Nest finally knew about it and you can see by default we actually don't see much about this message so let's actually go change our law logging here in our transcode consumer so in the Transco consumer instead of logging the object directly let's go ahead and stringify this object here so that we can see it a little bit better so I'll go ahead and send off this request and now we can see the trans code consumer actually stringifying out this log statement here so you can see the ID of the message and of course the data here so we actually get the data that we sent which was the file name MP3 as well as an options object with the default options so if we wanted to actually deal with the data directly in this transcode consumer we would do that by accessing the data field so we can have a debug statement here that says data and access the job dot data so this is mostly what you'll be working with in your consumers the actual data itself so now if we send a request off we should see in the logs we get the data here which is what mostly we will be concerned with the actual data for this consumer so you could imagine we get this data here we can do some async processing in the background not affect the request response life cycle and essentially distribute this work amongst a large number of instances instead of just running this on a single instance which is the biggest benefit of this bull mq approach so let's go ahead and actually take a look at how we can do that launch multiple instances with this running and see how jobs are distributed and processed amongst them so like I said before in our app service when we add a new job here there are a number of options that we can provide to change how this message is processed and I will leave a link in the description to the docs where you can take a look at all these options for now what I want to do is actually prepare our app to be able to run inside of a kubernetes cluster where we will be using it to actually distribute this amongst a lot of different running instances so let's go back into our transcode consumer and at the start here instead of stringifying the whole message here I want to change this and actually say the ID of the job that we are starting to process and then I want to introduce some fake delay some fake async processing so start here I'll say transcoding message and then I'll go into the job here dot ID so whenever a different transcode consumer picks up a message we can clearly see the ID of the message and then we will obviously have the data that gets sped out here too and then what I want to do is I want to introduce some delay here some fake processing so you could imagine if we're actually transcoding some audio file it'll take a significant amount of time and we want to actually emulate that so to do that we'll introduce an await new promise here that will have a resolve parameter and inside of our function here we will use a set timeout and the set timeout takes a callback where we will execute that resolve and additionally of course it takes the number of seconds that we want to wait before we do execute this resolve function so in our case what we'll do is we'll introduce an eight second delay so we'll say it takes eight seconds for this transcoding process to happen and then it's finished so after we do finish the transcoding I'll have another log statement saying transcoding complete for job and I'll specify the job.id here and change this to a template literal here and make sure we use logger.log here and don't forget to actually execute the resolve function here you can see it's complaining about the argument here so just make sure we add a void here for the promise and after this is complete we will go ahead and send off a new message we can see we are transcoding message ID 11 here and with the data object here and then after eight seconds we then see the transcode consumer say that it was complete for job 11. this looks great we can see our transcoding in action and now we are ready to test this out in a kubernetes cluster the last thing I want to do is back in our app. module right now we're hard coding figuration to redis however I want to read this in from an environment variable which will be much easier when we're running in kubernetes so I'll go ahead and PM install the nest.js config package here which uses dot m under the hood to read in environment variables we can go ahead and start our server back up and now our Imports array here we'll add the config module dot for root here and we of course import the config module up top we can do that manually config module from Nest JS slash config here so now instead of hard coding our configuration to Red is here instead of calling for root what we can do is we can call for root async now and instead this takes an async configuration which allows us to provide an options object here where we can specify a use Factory and inject any number of providers to get the environment variables we need of course in this case that will be the config service so we'll inject the config service of type of config service from nest.js config here and then we're going to export that same options object here so we have that redis object here and now we're going to specify again the host so we'll call configservice.get and we'll call this redis host as the environment variable and then the port here will be config service dot get redis Port now we'll have to also make sure that we actually inject the config service here so so add an inject property here and actually provide the config service to this asynchronous configuration as well as importing it directly at top in our Imports array we'll import the config module here so now of course we need to specify the actual environment variables here that we're reading in so I'll create a dot m for we have the redis host and of course this will just be localhost and the redisport is still going to be 6379 here so I'll go ahead and restart my development server up here to read in that dot m you can see we start off with no issues and if I send off a request to transcode you can see everything's still working fine here so we now have this asynchronous configuration we're reading it in from the config service and we can also do the same thing here for the bull module on the Queue itself we have the option to register the Q asynchronously as well and use a use Factory here if you wanted to read in your cues from a dot m for any other service you can do that but for now this will work fine for us okay so we're ready to dockerize our app and deploy it into a kubernetes cluster so I'm going to create a Docker file in the root of our app and I'm going to copy over the docker file that I've used in countless other videos to dockerizing sjs app if you'd like to see how we created this from scratch I'll leave a link in the description to one of my videos where we do that for now I'll paste this in here the only change we're going to make to this Docker file is the command at the bottom here we're going to start up the disk file which corresponds to this disk directory here and the main.js in there so essentially in this Docker file what we're doing is we're installing our dependencies all the dependencies including the development dependencies that we need to build the app in this development stage of the image here and then in the second stage the production stage we are just installing the production dependencies that we need to run the app we copy over the app from the first stage and we use node to start up the main JS one more thing we want to do is add a DOT Docker ignore as well here to make sure that we don't build in our node modules or our dist folder when we run a copy command of this directory so to actually build this image we can run Docker build specify the tag here so I'm going to push it to my own Docker Hub repository so you can create your own if you'd like or you can pull the one I'm going to push up to right now I'm going to call this Nest JS metrics and we'll specify the path the docker file here this is going to go ahead and build up the docker file and then I will just run Docker image push and push to that same repository here and this will go up and push to my repo which as I said before I've already created here in Docker Hub you can create your own or pull the one that I'm using lastly when we're building the image make sure that we actually specify the tag of the latest here so we'll go ahead and specify with tag latest and that we actually push up the latest tag here okay so once we've pushed our image up we have everything we need to start creating our kubernetes cluster for this app so I'll make a new directory in the root here called kh's CD into there and then I'm going to use Helm which is like a package manager for kubernetes to manage our deployment into our cluster here so I'm going to run Helm and if you don't have Helm I can leave a link in the description to see where you can grab that so we'll create a new album chart by running Helm create we'll call this Nest JS bull mq so this will create a new folder here or an sjs bollumq lives and we can see back in vs code here we have in the case director we have this new directory where we have the chart.yaml which is essentially the central manifest for this kubernetes chart that has metadata about it we have a default values file here so we can clear everything out in the default here as well as in the template section here I'll just go ahead and get rid of everything here because we will not be using any of these default templates and now in our chart.yaml I want to go ahead and make sure we have redis introduced into this app which is the only third-party app we'll need in our deployment so I'll go to this website artifact hub.io which is like a repository for kubernetes packages that we can add to this chart.yaml so we can find any third-party software here we want to look for redis and then we will go into the bit Nami one here epitome is a great open source provider for these Helm charts so I'll go to the install button to the right here and what we're looking for is the name of the repo here and in the chart now we can specify a new dependencies section here where we will list out the dependencies we want in this case we know this one will be called redis and the name here will correspond to values.yaml here where we can override settings if we want and of course all of the settings for this chart will be in the docs here where we can override any number of parameters to redis when it starts up like authentication other redis config we're going to use the default config so we won't really have any need to override any of those values here so let's just go ahead and specify the version firstly that we want to use so if we go up right now of the most current chart version is 17.5 5.1 so I'll use that one you can use whatever's the most recent and lastly we need to specify the repository and this is where we paste in that URL to the bitnami bow so now that we have the redis dependency we can run Helm penency update which we'll go ahead and get the latest dependencies and you can see we've downloaded the redis chart successfully here so we now have redis ready to deploy the only thing we need to add now is the deployment for our actual app so to make this easier we'll use Cube CTL command I have kubernetes running as Docker desktop so if you don't have a kubernetes cluster already running you can create one easily by downloading dock or desktop and going into the UI here there is a whole kubernetes section here where we can enable a kubernetes cluster so we can then use the cube CTL command use the create command here to create a new deployment the name will be nest.js bull mq we can specify the path to the image here which we know will be my repo of an sjs bull mq you can change this to yours if you'd like we can then specify the port that we want to expose on this container by using the dash dash report flag here which we of course know will be 3000 on this container so then to make sure that we actually output this to a yaml file and don't execute it I'll set dry run equal to client here and the output will be yaml and then we'll pipe this to a deployment dot yaml file so that will go ahead and actually create all of the metadata we need to actually Source control this deployment I'll just go ahead and move this directly into the templates folder so that when we install this Helm chart this deployment in our of our application will be included as well as redis and all the resources needed to run and connect to redis so the one change I'm going to make here is I'm actually going to run five instances of the nastrius bull mq so we can see how the message processing Works in a distributed fashion install this Helm sharp by running Helm install an sjs bolt mq and providing the path to the helm chart which is the current directory we can see that we successfully deployed it and if I run a cube CTL get pods here we can see that our pods have been created the 5 Bolt mq sjs pods as well as the redis pods here we have the redis master and a redis replica that is part of the bitnami helm chart so if we run on Cube CTL get pods again after a little bit we should see that our pods are all up running correctly we can run Cube CTL logs and paste the Pod name in here to get the logs for our app we can see they start up successfully as well as the redis master here if we want to take a look at what's going on here we can see it's synchronized with replicas and it has started up successfully so if you also run a cube CTL get service we can see that the bitnami redis helm chart created a service for us to be able to connect to redis a service in kubernetes is how pods can talk to one another through the underlying Network additionally we also need to create another servers for our app to be able to talk to it over over HTTP and actually produce a message we can do this by in cubectl exposed deployment sjs bull mq we'll specify the type here should be a type of node Port which is a different type of service in kubernetes which will essentially allow us to open a port on our local machine and communicate to our app via localhost we'll then specify the port here which will be of Port 3000 we know that's the port we opened on our next JS app and then when we created the deployment we'll set the dry run equal to client here and set the output to yaml and pipe this to a service.yaml file so now we can see we have a service.yaml file here which is everything we need to create this service that will expose our app so let's move this service into the templates folder and the other thing we want to do is in the deployment.yaml for our app we want to add a new environment variable section here so that we can provide the redis host and port in our cluster we can do this very easily in our containers section here we'll add a new m property here where we have each environment variable being a different key pair value so we have the redis host here that we firstly need to Define in this case the value so we need to find the hostname for the redis service that's running right now so if we're on cubesatel get service we want to just connect to this master here so since this is a cluster IP the hostname here will be the actual name of the service so I'll grab the name of the service here and the port will be 6379 so we'll add the redist hostname here and then the other environment variable will be the redis port and the value here will be 6379 so now that we have all this make sure we actually graph the report in quotes here we can go back to our terminal and run Helm upgrade nest.js bowl mq and provide the path here so now we can see our release has been upgraded if we run Cube CTL gets serviced now we can see our nest.js Bowl mq service with a node Port opened up on Port 30455 so let's go ahead and test this out we will get this port here this could be different for you I'll copy this port and we'll open up Postman and now we'll execute a post request for transcode on this port and send it off you can see if we send this request off we're getting a socket hang up error so let's investigate this by looking at our pods we can see they're actually crashing right now so there's some sort of issue let's grab a pod name here and take a look at the logs we can see that there's an authentication error on trying to connect to our redis cluster so let's go ahead and disable authentication I want to go look at the artifact Hub configuration and look for the common configuration parameters where we can see the auth.enabled property here to enable authentication I want to go ahead and disable this all together by default it is enabled so let's go back to our values.yaml in our chart we are going to override the redis dependency here so we take the name that we supplied of redis here and we know that the key we're overriding is auth enabled and we'll set this to false so now we'll go back to our terminal and upgrade our deployment again we can see that our new redis instances are starting up now let's wait until these are running okay so after we give our containers a bit of time to finish restarting we should see our deployments are successfully starting up and we have no more problems now if we look at the logs for our apps they look good I'm going to run keep still get service make sure my port is correct for my nest.js Bolin q and I'm going to launch a post request now localhost 3000 that port and hit the transcode route now we can see we have a 201 that came back so let's take a look at the log for running Cube CTL logs Dash L for label app equal nest.js boomq this is a label that we applied to our deployment so we can get the logs for all the Pods at once and we can tail them as well by using the follow so we can already see all the logs for each of the apps in this case we can see that we are trying transcoding message one on one of these pods and the data object is here which is great to see and lower down below you see the transcoding was completed for that same consumer so if I send another one we can see it in action here we have the second message being transcoded with the data object here and then after eight seconds we will see that it is complete so the beauty of this is that right now we have five instances of our application running however when we send a message and produce it to our queue only one of the pods is actually consuming it so we can see the distributed nature of this system we send off a message and only one of the consumers pick it up it is a controlled cue just like we'd expect and if we send several more messages you can see them all get picked up off the queue one after the other this is exactly what we'd expect in this distributed system so we can have any number of instances instances running kicking off messages off these cues processing them and dealing with these processing Peaks when they happen so thanks so much for watching and I'll see you in the next one

Channel: Michael Guay

Views: 25,742

Keywords:

Id: oy-t6V6aAZY

Length: 31min 40sec (1900 seconds)

Published: Sat Jan 21 2023