Preventing Jobs From Running Simultaneously on Multiple Role Instances

There are times where we need to schedule maintenance jobs to maintain our Azure Cloud Services. Usually, it requires us to design our systems with an extra Role in order to host these jobs.

Adding this extra Role (Extra Small VM) costs about $178.56 a year!

But don’t worry! There’s another way to schedule jobs. You can use Blob leasing to control the number of concurrent execution of each job. You can also use Blob leasing to help distribute the jobs over multiple Role instances.

To limit the number of concurrent executions of a job, you need to acquire a Blob lease before you start to execute the job. If the acquisition fails, the job is already running elsewhere and the Role instance must wait for the lease to be released before trying to run the job.

To distribute jobs evenly over multiple Role instances, you can execute a single job per Role instance and have each role determine which job to execute based on the Blob lease that they successfully acquire.

In a previous post I mentioned that it’s better to acquire a short lease and to keep renewing it than to acquire an indefinite lease. This also applies to job scheduling, because if a Role is taken down for maintenance, other roles must be able to execute the job.

To help me manage this complexity, I created a Job Reservation Service that uses a Blob Lease Manager in order to regulate how jobs are executed throughout my Cloud Services. The rest of this post will demonstrate how to use the Job Reservation Service.

 

DEMO Code: Distributing Jobs over Multiple Worker Role Instances

The worker role below tries to distribute 2 jobs over an unknown number of instances. I have made the assumption, that there are always a minimum of 2 instances.

In order to limit concurrent executions of these jobs, I start by creating a JobReservationService instance that I preserve throughout Role’s lifetime. Then when the Role starts, I find and reserve an available job. Once the job is reserved, the Role is able to execute the job safely. Finally, when the Role is stopped, we release the job reservation so that other Roles may start executing it.

Code:

public class WorkerRole : RoleEntryPoint
{
    private readonly JobReservationService jobReservationService;

    readonly Dictionary<string, Func<IJob>> jobs = new Dictionary<string, Func<IJob>>
    {
        { MonitorCloudServiceJob.Name, () => new MonitorCloudServiceJob() },
        { CleanDatabaseJob.Name, () => new CleanDatabaseJob() },
    };

    public WorkerRole()
    {
        var reserverName = RoleEnvironment.CurrentRoleInstance.Id;
        jobReservationService = new JobReservationService("Storage", "jobs", reserverName);
    }

    string job;

    public override void Run()
    {
        Trace.TraceInformation("Worker entry point called", "Information");

        if (!string.IsNullOrWhiteSpace(job))
            Task.Run(() => jobs[job].Invoke().Execute());

        while (true)
        {
            Thread.Sleep(10000);

            TraceJobExecution();
        }
    }

    private void TraceJobExecution()
    {
        if (string.IsNullOrWhiteSpace(job))
            Trace.TraceInformation("Not Working on a Job", "Information");
        else
            Trace.TraceInformation("Working on Job |> " + job, "Information");
    }

    public override bool OnStart()
    {
        ServicePointManager.DefaultConnectionLimit = 12;

        job = SelectJob();

        return base.OnStart();
    }

    private string SelectJob()
    {
        foreach (var j in jobs)
        {
            if (!jobReservationService.TryReserveJob(j.Key, 60)) continue;
            return j.Key;
        }
        return string.Empty;
    }

    public override void OnStop()
    {
        if (!string.IsNullOrWhiteSpace(job))
            jobReservationService.CancelReservation(job);

        base.OnStop();
    }
}

Sample Jobs

These sample jobs are empty shells used to demonstrate how to distribute job execution over multiple roles.

internal class MonitorCloudServiceJob : IJob
{
    public void Execute()
    {

    }

    public static string Name
    {
        get
        {
            return “monitor-cloud-service.job”;
        }
    }
}

internal class CleanDatabaseJob : IJob
{
    public void Execute()
    {

    }

    public static string Name
    {
        get
        {
            return “clean-database.job”;
        }
    }
}

internal interface IJob
{
    void Execute();
}

The Job Reservation Service

I created the JobReservationService to encapsulate Blob leasing and job reservation logging.

The service will create a Blob for each job reservation and will use the Blobs content to keep track of the last 10 reservations.

• The connectionString parameter is the name of the setting that contains the actual connection string.
• The containerName parameter is the name of the container that is used to track jobs.
• The reserverName is the name of the Role instance which makes reservations using the JobReservationService instance.

Please notify me if something isn’t clear. As a last note, I would like to point out that the code for this service is Open Source and can be found on my GitHub project.

public class JobReservationService : IDisposable
{
    private readonly string reserverName;

    private readonly BlobLeaseManager manager;
       
    private CloudBlobContainer container;

    public JobReservationService(string connectionString, 
                                    string containerName, 
                                    string reserverName)
    {
        this.reserverName = reserverName;
        Init(connectionString, containerName);

        manager =  new BlobLeaseManager();
    }

    private void Init(string connectionString, string containerName)
    {
        var cs = CloudConfigurationManager.GetSetting(connectionString);

        var account = CloudStorageAccount.Parse(cs);

        var client = account.CreateCloudBlobClient();

        var deltaBackoff = new TimeSpan(0, 0, 0, 2);
        client.RetryPolicy = new ExponentialRetry(deltaBackoff, 10);

        container = client.GetContainerReference(containerName);
        container.CreateIfNotExists();
    }

    public bool TryReserveJob(string jobName, double jobReservationInSeconds)
    {
        var blobReference = GetJobReservationBlob(jobName);
            
        if(!blobReference.Exists())
            InitializeLeaseBlob(blobReference);

        var acquireLease = manager.TryAcquireLease(blobReference, jobReservationInSeconds);
            
        if(acquireLease)
            UpdateReservationLog(jobName);

        return acquireLease;
    }

    public bool HasReservation(string jobName)
    {
        return manager.HasLease(GetJobReservationBlob(jobName));
    }

    public void CancelReservation(string jobName)
    {
        manager.ReleaseLease(GetJobReservationBlob(jobName));
    }

    private CloudBlockBlob GetJobReservationBlob(string jobName)
    {
        var blobReference = container.GetBlockBlobReference(jobName);
        return blobReference;
    }

    private void InitializeLeaseBlob(CloudBlockBlob blobReference)
    {
        var log = new JobReservationLog();
        UpdateBlobContent(log, blobReference);
    }

    private void UpdateBlobContent(JobReservationLog jobReservationLog, 
                                    CloudBlockBlob jobReservationBlob)
    {
        jobReservationLog.Add(MakeJobReservation());

        string leaseId = manager.GetLeaseId(jobReservationBlob);

        AccessCondition accessCondition = string.IsNullOrWhiteSpace(leaseId)
            ? null
            : new AccessCondition
            {
                LeaseId = leaseId
            };

        jobReservationBlob.UploadText(jobReservationLog.ToJson(), 
                                        null, 
                                        accessCondition);
    }

    private void UpdateReservationLog(string jobName)
    {
        CloudBlockBlob blobReference = GetJobReservationBlob(jobName);

        JobReservationLog jobReservationLog = JobReservationLog.Make(blobReference.DownloadText());
        JobReservation lastReservation = jobReservationLog.LastReservation;

        if (lastReservation.Reserver == reserverName) 
            return;
            
        UpdateBlobContent(jobReservationLog, blobReference);
    }

    private JobReservation MakeJobReservation()
    {
        return new JobReservation
        {
            Obtained = DateTime.UtcNow,
            Reserver = reserverName,
        };
    }
        
    public void Dispose()
    {
        manager.Dispose();
    }

    public struct JobReservation
    {
        public string Reserver { get; set; }
        public DateTime Obtained { get; set; }
    }

    private class JobReservationLog
    {
        private readonly List<JobReservation> reservations = new List<JobReservation>();

        public JobReservationLog()
        {
        }

        private JobReservationLog(List<JobReservation> lockReservations)
        {
            reservations = lockReservations;
        }

        internal JobReservation LastReservation
        {
            get { return reservations.FirstOrDefault(); }
        }

        internal static JobReservationLog Make(string json)
        {
            var list = JsonConvert.DeserializeObject<List<JobReservation>>(json);
            return new JobReservationLog(list);
        }

        internal void Add(JobReservation jobReservation)
        {
            reservations.Insert(0, jobReservation);

            if (reservations.Count > 10)
                reservations.Remove(reservations.Last());
        }

        internal string ToJson()
        {
            return JsonConvert.SerializeObject(reservations);
        }
    }
}

Summary

This job reservation service can be quite handy in helping to reduce maintenance costs by redistributing maintenance jobs over many Role instances. Furthermore, distributing these jobs over many Role instances will make them more reliable.