Dynamic Scheduling Under The Hood

Feb 12

As promised, in this week’s edition of Cloud Musings, I thought I would do a deep dive with code into dynamic scheduling and explain how we solve this challenge. Don’t forget to subscribe here on Substack or Linkedin. Thanks for reading!

Last week, I wrote about this on a very high level. Here is a demo of how it works. I've also started to open source some of our code base so people can get an idea how our platform works.

We have a pretty solid architecture:

Python, Flask API
MongoDB database
Celery, Redis task queue
React, Typescript Frontend
Docker on AWS ECS and Digital Ocean

Essentially, we have a distributed system. One is the Python, Flask API that captures user inputs, saves them in MongoDB, and returns values. The second one is the scheduler. For the scheduler, we are using a library called celery-redbeat. This library provides a mechanism to insert tasks into Redis and hooks up into Celery's beat scheduler. The scheduler keeps track of tasks that need to be executed when they are due to be executed. The third piece is the Celery task queue. The task queue receives a task and executes it.

The UI:

To schedule a deployment, a user specifies a cloud infrastructure, the date, time, and dependency. Dependency is optional, but we could imagine the case of deploying the API before a change in the UI or a database change before deploying the API. When a user specifies a dependency, it runs 15 minutes before the actual deployment.

For the frontend, I’m using React, Typescript with a tool called umijs and ant design from Ant financial:

The scheduled deployment request function just calls the backend API passing the user inputs along with the access token for security and to know who the user is.

import { request } from 'umi';

export async function schedule(data: API.ScheduledDataType) {
  return request('/v1/scheduled_deployments', {
    method: 'POST',
    data,
    headers: {
      Authorization: 'Bearer oatfin_access_token'
    },
  });
}

The user specifies a date (year, month, day) and time (hour, minute). We use the moment-timezone npm package to guess the timezone: moment.tz.guess()

  const handleSubmit = async (values: API.DateTimeDependency) => {
    const data: API.ScheduledDataType = {
      app: current?.id,
      year: values.date.year(),
      month: values.date.month() + 1,
      day: values.date.date(),

      hour: values.time.hour(),
      minute: values.time.minute(),
      timezone: moment.tz.guess(),
      dependency: values.dependency,
    };

    try {
      ...
      const res = await schedule(data);
      ...
    } catch (error) {
      message.error('Error scheduling deployment.');
    }
  };

Inside the React functional component, we specify the onSubmit function in the form where we capture the user inputs. We get a lot of React components out of the box from antd. For example, Modal, Form, Form.Item, DatePicker, and TimePicker are all components from ant design.

 export const SchedComponent: FC<BasicListProps> = (props) => {
  ...
  const getModalContent = () => {
    return (
      <Form onFinish={handleFinish}>
        <Form.Item name="name" label="Application">
          <Input disabled value={current?.name} />
        </Form.Item>
        <Form.Item name="date" label="Date">
          <DatePicker
            disabledDate={(currentDate) => disabled(currentDate)}
          />
        </Form.Item>
        <Form.Item name="time" label="Time">
          <TimePicker use12Hours format="h:mm A" showNow={false}/>
        </Form.Item>
        <Form.Item name="dependency" label="Dependency">
            <Select>
              <Select.Option key={app.id} value={app.id}>
                {app.name}
              </Select.Option>
            </Select>
          </Form.Item>
      </Form>
    );
  };

  return(
    <Modal
      title="Schedule Deployment"
      onCancel={onCancel}
      onsubmit={handleSubmit}
    >
      {getModalContent()}
    </Modal>
  )
}

The Python, Flask API:

First, we capture the parameters that the UI sends, then call the service to create the schedule. If the user specifies a dependency, then we create a scheduled entry in MongoDB and Redis for the dependency also.

...
@api.route('/scheduled_deployments', methods=['POST'])
@jwt_required()
def schedule_deployment():
    user_id = get_jwt_identity()['user_id']
    req_data = flask.request.get_json()

    app_id = req_data.get('app')
    dependency = req_data.get('dependency')
    year = req_data.get('year')
    month = req_data.get('month')
    day = req_data.get('day')
    hour = req_data.get('hour')
    minute = req_data.get('minute')
    timezone = req_data.get('timezone')

    sd = ScheduledDeploymentService().create(...)

    args = [...]
    SchedulerService().create_entry(sd, args=args, app=app)

    if sd.dependency is not None:
        dep = sd.dependency
        args = [...]
        SchedulerService().create_entry(dep, args=args, app=app)

    return flask.jsonify(
        result=sd.json(),
    ), 200

Inside ScheduledDeploymentService, we create an entry in MongoDB for the deployment and any dependency the user specified. The dependency is then scheduled 15 minutes before the deployment.

def create(app_id, dep, user_id, year, month, day, hour, minute, tz):
    if dependency:
        dep_date = datetime.datetime(year, ++month, day, hour, \          
                  minute) - datetime.timedelta(minutes=15)
        sched = ScheduledDeployment(
            app=dependency_app,
            team=user.team,
            year=dep_date.year,
            month=dep_date.month,
            day=dep_date.day,
            hour=dep_date.hour,
            minute=dep_date.minute,
            original_timezone=tz
        ).save()

        return ScheduledDeployment(
            app=app,
            dependency=sched,
            team=user.team,
            year=year,
            month=month,
            day=day,
            hour=hour,
            minute=minute,
            original_tz=tz
        ).save()

In the MongoDB document, we store the deployment schedule as well as the key for the entry in Redis. The interesting thing here is that ScheduledDeployment references itself for the dependency.

from mongoengine import Document, IntField, ReferenceField, etc.

class ScheduledDeployment(Document):
    year = IntField()
    month = IntField()
    day = IntField()
    hour = IntField()
    minute = IntField()
    original_timezone = StringField()
    entry_key = StringField()
    dependency = ReferenceField('self', required=False)

    def json(self):
        return {
            'year': self['year'],
            'month': self['month'],
            'day': self['day'],
            'hour': self['hour'],
            'minute': self['minute'],
            'original_timezone': self['original_timezone'],
            'entry_key': self['entry_key']
        }

The SchedulerService is used to first translate the user date and time from their timezone to UTC. Then it creates an entry in Redis using RedbeatSchedulerEntry from celery-redbeat. This provides the mechanism to save a schedule in Redis. Once an entry is added to Redis, then the scheduler picks it up when it's due, then sends it to the task queue.

We’re using crontab from celery to create the actual schedule. The challenge here is that we can’t specify a year. We handle this by deleting the entry from Redis once the scheduled deployment is successful. After the entry is saved in Redis, we store the entry key in the MongoDB schedule deployment so we can delete it later.

class SchedulerService(object):
    def create_entry(self, sd, args, app):
        scheduled_date = self.to_utc(...)
        entry = RedBeatSchedulerEntry(
            name=str(sd.id),
            schedule=crontab(
                month_of_year=scheduled_date.month,
                day_of_month=scheduled_date.day,
                hour=scheduled_date.hour,
                minute=scheduled_date.minute,
            ),
            task='tasks.deploy',
            args=args,
            app=app
        )
        entry.save()
        sd.update(set__entry_key=entry.key)

    def to_utc(self, timezone, year, month, day, hour, minute):
        tz = pytz.timezone(timezone)
        user_dtz = tz.localize(datetime.datetime(...))
        return tz.normalize(user_dtz).astimezone(pytz.utc)

Finally the task queue looks like this:

...
app = Celery(__name__)
app.conf.broker_url = app_config.REDIS_BROKER
app.conf.result_backend = app_config.REDIS_BACKEND
app.conf.redbeat_redis_url = app_config.REDIS_BACKEND

app.conf.update()


@app.task(name='tasks.deploy')
def deploy(user_id, app_id, key, secret, region, deployment_id=None, scheduled_id=None):
    connect(
        db=app_config.DB_NAME,
        username=app_config.DB_USERNAME,
        password=app_config.DB_PASSWORD,
        host=app_config.DB_HOST,
        port=app_config.DB_PORT,
        authentication_source=app_config.DB_AUTH_SOURCE
    )

    ECSDeploymentService().deploy(
        user_id=user_id,
        app_id=app_id,
        oatfin_key=key,
        oatfin_secret=secret,
        oatfin_region=region,
        deployment_id=deployment_id,
        scheduled_id=scheduled_id
    )

Thanks for reading!

Jay