Forem: Sateesh Madagoni

Orchestrate AWS Lambdas using MongoDB - Part 2

Sateesh Madagoni — Wed, 15 Nov 2023 01:02:08 +0000

Continuation to the first part. This post assumes you are a developer with working knowledge on AWS, Lambda, EventBridge, MongoDB, NodeJs.

Technical Implementation:

1. Design State-Machine
This step you need to figure out, how do you want to orchestrate your jobs and create an object like below.

const stateMachine = {
  id: '1',
  purposeId: 'catch_the_fish',
  currentPhase: 'thinking',
  phases: {
    thinking: {
      steps: {
        `thinking`: {
          jobs: ['buy_fishing_gear'],
        },
        buy_fishing_gear: {
          jobs: ['rent_a_boat'],
        },
        rent_a_boat: {
          jobs: ['go_to_fishing_ground'],
        },
        go_to_fishing_ground: {
          jobs: ['cast_fishing_pole', 'drink_beer', 'catch_fish'],
        },
        catch_fish: {
          jobs: ['go_home'],
        },
      },
      finished: [],
      next: 'end',
      waitFor: [
        'buy_fishing_gear',
        'rent_a_boat',
        'go_to_fishing_ground',
        'cast_fishing_pole',
        'drink_beer',
        'catch_fish',
        'go_home',
      ],
    },
  },
};

In the above example we are running buy_fishing_gear after thinking job is finished in a sequential manner, but we run cast_fishing_pole, drink_beer and catch_fish in parallel after go_to_fishing_ground is successful.

2. Start the Process:
Creation of the statemachine could be anything from API to a cron job. Lets take API as an example as POST /process/catch-the-fish.

router.post('/process/catch-the-fish', async (req, res, next) => {
  const state = {}; // above mentioned state
  // Insert into state-machines
  const createdState = await db.collection('state-machines').insertOne(state);
  // Start the process by sending first event. As thinking is success we start the process of catching fish
  await db.collection('statuses').insertOne({
    stateId: createdState._id,
    status: 'success',
    job: 'thinking',
    date: new Date(),
    // additional parameters as needed
  });
});

3. State Machine Job:

exports.handler = async (event, ctx, callback) => {
  // Mongodb Document from the trigger
  const document = event.detail.fullDocument;
  // Find the state machine
  const state = await db
    .collection('state-machines')
    .findOne({ _id: document.stateId });
  // if its ended do nothing
  if (state?.currentPhase === 'end') {
    callback();
  }
  // Update the job as finished for the state machine
  const currentPhase = state.phases[state.currentPhase];
  currentPhase.finished.push(document.job);
  await db
    .collection('state-machines')
    .findOneAndUpdate(
      { _id: document.stateId },
      { $set: { [`phases.${state.currentPhase}`]: currentPhase } }
    );
  const jobsRemaining = currentPhase.waitFor.filter(
    (s) => !currentPhase.finished.includes(s)
  );
  // If there are no remaining jobs in the current phase
  if (jobsRemaining.length === 0) {
    // Update the currentPhase to be the next Phase.
    await db.collection('state-machines').findOneAndUpdate(
      { _id: document.stateId },
      {
        $set: {
          currentPhase: currentPhase.next,
        },
      }
    );
    // If the next phase is not end trigger the phase.
    if (currentPhase.next !== 'end') {
      await db.collection('statuses').insertOne({
        ...document,
        job: `${currentPhase.next}Start`,
        status: 'success',
      });
    }
  }
};

Above, whenever a job succeeds this will update the finished jobs and check if the phase is finished and move onto next phase until it meets the end phase.
4. Orchestrator Job:

exports.handler = async (event, ctx, callback) => {
  ctx.callbackWaitsForEmptyEventLoop = false;
  const document = event.detail.fullDocument;
  const state = await db
    .collection('pipeline-state-machines')
    .findOne({ _id: document.stateId });
  const currentPhase = state[state.currentPhase];
  if (!document.stateId) {
    console.log('Cannot execute the pipeline without the state id:', state);
    return { success: false };
  }
  if (state.currentPhase === 'end') {
    console.log('Cannot execute the pipeline without the state id:', state);
    return { success: false };
  }
  if (!currentPhase) {
    console.log('Cannot execute the pipeline without the state id:', state);
    return { success: false };
  }
  const step = currentPhase.steps[document.job];
  if (step && step.jobs && step.jobs.length > 0) {
    await Promise.all(
      step.jobs.map(async (job) => {
        const payload = {}
        const event = {
          FunctionName: job,
          InvocationType: 'Event',
          LogType: 'Tail',
          Payload: JSON.stringify(payload),
        };
        return lambda.invoke(event);
      })
    );
  }
  callback();
};

The above job receives the success event and finds the next jobs to trigger and invoke them.
5. Notifications Job:
As above the job receives all the events so we can use the data and structure the message to send.

What if there are multiple phases:
Then we just have to add another phase to existing state machine configuration.

const stateMachine = {
  id: '1',
  purposeId: 'catch_the_fish',
  currentPhase: 'thinking',
  phases: {
    thinking: {
      // Prev things
    },
    cooking: {
      steps: {
        cookingStart: {
          jobs: ['clean'],
        },
        clean: {
          jobs: ['cook'],
        },
      },
      waitFor: ['cook', 'clean'],
      finished: [],
      next: 'end',
    },
  },
};

I hope the above code and explanation gives you a way to implement your own solutions. For anymore details please do comment, I would be happy to help. Thanks.

Orchestrate AWS Lambdas using MongoDB - Part 1

Sateesh Madagoni — Wed, 15 Nov 2023 01:01:30 +0000

This post assumes you are a developer with working knowledge on AWS, Lambda, EventBridge, MongoDB, NodeJs.

Usecase: Let's say you have several lambdas that you need to run some in parallel, in sequential and some in conditional, and their collective goal is to finish a task. To orchestrate these jobs we need a state-machine(https://statecharts.dev/what-is-a-state-machine.html), which ensures the jobs are run based on the given conditions and will terminated itself at the end.

Solution: For the similar problem, AWS Lambda provides a solution called step-functions where you could chain these lambdas, I felt its too difficult to establish and manage them, so I created a custom solution using MongoDB Triggers and AWS EventBridge. Now lets work on the process.

Components
_1. Statemachine Configuration: A configuration which consists of all the phases and jobs to run in order. This will be saved to a collection state-machines.

2. Status Event: Whenever a job status changes, i.e, success, failed, started we insert an event to database which will be consumed by other necessary jobs. This events will be saved statuses collections. This is the collection where MongoDB trigger is configured.

3. Statemachine Job: An additional job apart from the business jobs to update the state machine whenever a job completes or fails. This job only receives success events.

4. Orchestrator Job: An additional job apart from the business jobs to trigger the next jobs. This job only receives success events.

5. Notification Job: You could have another lambda to send messages to slack or any other messaging platform. This job only receives all events.

How it works:

A state machine will be created by either an API or any such equivalent method based on the requirements. And sends an status event as the start of the process. Now this event is received by statemachine_job, orchestrator_job and notifications_job
statemachine_job will update the corresponding statemachine with the finished job. Check if it has to either end the state machine or move onto next phase.
orchestrator_job will find the next job to run based on the given configuration, until nothing left to run.
notifications_job will send the current event message to configured messaging medium.
This cycle continues until all the phases finished & satisfied with finished jobs.

Set Up:
As mentioned we need to two collections state-machines, statuses. The model will be explained later. Create a trigger on statuses collection, for insert event such that whenever an item inserted into statuses it will be sent to the AWS Eventbridge. Set up the database trigger for a collection statuses. Follow this link to setup events for AWS Eventbridge https://www.mongodb.com/docs/atlas/triggers/.
The above created event bridge will invoke statemachine_job, orchestrator_job and notifications_job

Lets work on the technical implementation on the next part.

User Role Management in NodeJS, Express, MongoDB

Sateesh Madagoni — Fri, 28 Feb 2020 18:06:47 +0000

Problem: Multiple users in a system, allowed to do specific actions.

Solution: There are multiple user role management packages in npm, but I want something easier, quicker. So I started solving it myself.

Example: A blog with users - U, authors - A, admin - M

Create users with a field user_type.

Users.create({
name: 'User',
user_type: 'U'
})
Users.create({
name: 'Author',
user_type: 'A'
})
Users.create({
name: 'Author',
user_type: 'M'
})

Assuming user logins managed using a jwt token. And sign the token including user_type, add a middleware to decode and save user data to req.user

const decoded = await jwt.verify(token, process.env.JWT_SECRET);
req.user = {
    name: decoded.name,
    user_type: decoded.user_type
};

Write another middleware to authenticate role.

const authenticateRole = (roleArray) => (req, res, next) => {
  if(!req.user) {
    return res.status(401).json({
      success: false,
      message: 'Session expired',
      code: 'SESSION_EXPIRED'
    });
  }
  const authorized = false;
//if user has a role that is required to access any API
  rolesArray.forEach(role => {
   authorized = req.user.user_type === role;
  })
  if(authorized) {
    return next();
  }
  return res.status(401).json({
    success: false,
    message: 'Unauthorized',
  })
}

Finally use the authenticateRole middleware in the API access.

//This is accessed by only Admin user
route.get('/users', authenticateRole(['M']), handler)
//This is accessed by anyone
route.get('/posts', authenticateRole(['M','U','A']))

I am trying to enhance this idea as my needs.