Category: Azure

3rd Party API Integration Pattern

~ Nish ~ Leave a comment

Most applications at the enterprise, mid-size, or even startup levels, would have one or more 3rd party APIs they integrate with. It may start off with just one component talking to the external API, but can very quickly grow into multiple systems talking to that 3rd party API. It’s not a good practice to have direct integrations for all those different systems. Instead, a good pattern is to have a 3rd Party API Integration Service, deployed as a company wide API, where this API acts as the gateway between the 3rd party system and systems within your application / enterprise.

The benefits of this pattern are mainly that the integration service insulates the rest of the ecosystem from the external API. This helps avoid negative coupling spread throughout the system as well as redundant codebases that could all very quickly go out of sync. All interactions are done in one system and one codebase, which makes it significantly simpler to maintain. Most such integration services often serve multiple functions.

  • Authorization – All of the auth code to the 3rd party API is managed in this one service.
  • Caching – If the 3rd party API is expensive or not as performant as you’d like it to be, you can add a fast caching layer here (and use something like Redis for recent repeat calls).
  • Schema Transform – You can transform the schema to a more company compliant standard.
  • Data Filtering – This would be a simple place to filter out certain results or certain fields within results.
  • Model Enrichment – You can enrich the 3d party response by adding internal system identifiers, which will make it far easier for downstream callers to resolve entities.
  • Local DB persistence – All fetched data can be persisted locally, meaning that in a future scenario where the 3rd party API may not be available to integrate with, you still have the original data.

And the fact that you have this integration service means that you can switch 3rd party services without the internal systems having to make any changes. A new version of the integration service will abstract away the new external API and convert the data to the same format expected by the downstream callers.

Using IConfiguration and appsettings.json in a .NET Core Azure Function project

~ Nish ~ Leave a comment

For very basic Web APIs, HTTP triggered Azure functions are a very efficient and cost effective way to implement the service. One inconvenience though is that by default it does not come hooked up with an appsettings.json file or an IConfiguration instance. Fortunately, it’s not very difficult to hook those up.

Step 1

Add the following NuGet package to your project.

  • Microsoft.Azure.Functions.Extensions

Step 2

Add a startup.cs file to your project and insert the following code.

[assembly: FunctionsStartup(typeof(Startup))]

public class Startup : FunctionsStartup
{
    public override void ConfigureAppConfiguration(
      IFunctionsConfigurationBuilder builder)
    {
        var context = builder.GetContext();

        builder.ConfigurationBuilder
            .AddJsonFile(
              Path.Combine(
                context.ApplicationRootPath, 
                "appsettings.json"), 
              optional: true, 
              reloadOnChange: false)
            .AddJsonFile(
              Path.Combine(
                context.ApplicationRootPath, 
                $"appsettings.{context.EnvironmentName}.json"), 
              optional: true, 
              reloadOnChange: false)
            .AddEnvironmentVariables();
    }
}

Step 3

Add your appsettings.json file and setup any configuration properties you need in there.

{
  "YourProperty": "some-value-here"
}

Step 4

Inject IConfiguration into your constructor. Note that the default function template won’t have a constructor as it uses a static class. Remove the static, and add the constructor.

private readonly IConfiguration _configuration;

public SomeFunction(IConfiguration configuration)
{
    _configuration = configuration;
}

And now you can use it in the function as you’d normally do.

[FunctionName("SomeFunction")]
public async Task<IActionResult> Run(
    [HttpTrigger(
      AuthorizationLevel.Anonymous, 
      "get", 
      "post", 
      Route = null)] 
    HttpRequest req,
    ILogger log)
{
  // Use _configuration["YourProperty"]
}

Exiting an Azure WebJob programmatically

~ Nish ~ Leave a comment

If you need your WebJob to run a task and exit, or if you need to conditionally exit it under certain scenarios, and you thought it’s just a matter of exiting the worker service’s ExecuteAsync method, you’d be wrong. Even after your worker code has stopped running, the parent host stays alive. The right way to exit your WebJob is to inject IHostApplicationLifetime into your worker, and then the last line of code in your ExecuteAsync would be a call to its StopApplication method. 

public Worker(ILogger<Worker> logger, IHostApplicationLifetime appLifeTime)
{
    _logger = logger;
    _appLifetime = appLifeTime;            
}

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    int runs = 0;

    while (!stoppingToken.IsCancellationRequested && runs++ < 3)
    {
        // do some task here   
        await Task.Delay(2000, stoppingToken);
    }

    _appLifetime.StopApplication();
}

Azure WebJobs not logging to Application Insights

~ Nish ~ Leave a comment

With regular App Services, if you enabled Application Insights, your ILogger logs will get written into Application Insights without having to do anything special. This is not the case with WebJobs though. After wasting a few hours trying to figure out why my logs were not showing up, here’s a solution I found that worked.

Step 1

Install the latest version of Microsoft.Azure.WebJobs.Logging.ApplicationInsights.

Step 2

Modify your startup code as follows.

public static void Main(string[] args)
{
    IHost host = Host.CreateDefaultBuilder(args)
        .ConfigureServices(services =>
        {
            services.AddHostedService<Worker>();
        })
        .ConfigureLogging((context, builder) =>
        {
            builder.AddConsole();

            var key = context.Configuration["APPINSIGHTS_INSTRUMENTATIONKEY"];
            if (!string.IsNullOrEmpty(key))
            {
                builder.AddApplicationInsightsWebJobs(
                    options => options.InstrumentationKey = key);
            }
        })
        .Build();

    host.Run();
}

And that’s it. Remember that there’s usually a 1-2 minute delay before your logs show up in App Insights, and it could be as high as 3-4 minutes if it’s the first time you deploy. So, don’t panic if you don’t see the logs right away. Happy debugging!

Implementing asynchronous endpoints in REST with callbacks

~ Nish ~ Leave a comment

(click on the image to view it in full resolution)

This is an extension of the asynchronous endpoint pattern where the consumer can register a callback URL, which is invoked when the task is completed. This only works if the consumer has the option to expose an externally hittable REST endpoint. The advantage is that the consumer does not have to keep polling the service to see if the task has completed.

Once the task completes, the service invokes the callback URL. A common approach is to send just a task id to the callback, and the consumer then invokes a call to the service to fetch the full payload. A less common approach (if sensitive info is not included in the payload) is to send the entire result-set to the callback endpoint.

On the service side, the implementation is done by reacting to the task completed event by invoking the registered callback for the completed task. There are several ways to do this. A simple approach is for the task worker itself to invoke the callback. The disadvantage there is that it’s complex to now support retries if the callback endpoint is down. With AWS, a common approach is to update the data in a DynamoDB table and for a Lambda to be triggered off that, and the Lambda would then invoke the callback. A similar approach for Azure is to have a Function triggered off Table storage or Cosmos DB, which would invoke the callback. Both AWS and Azure have mechanisms to add retry to the function/lambda, to handle scenarios where the callback URI is down.

Implementing asynchronous endpoints in REST

~ Nish ~ Leave a comment

When you have an API endpoint that takes too long to run, it can result in consumers facing time-outs or blocking scenarios. Now while consumers can use background workers to make these long running calls, it’s far more convenient if the service is implemented to support an asynchronous operation. A very common approach is to queue up the work on the service side and to return a task-id to the consumer. The consumer can then periodically poll the API with the task-id which returns a response indicating whether the task has been completed or not. Once the task completes, the consumer then makes a call to retrieve the completed task response. In some implementations, these two operations are combined, so it’s the same endpoint that both returns the pending status as well as the completed result.

On the service side, the implementation is typically done using a queue. The first call generates and persists a task-id which is returned, and the task is queued up in a queue (such as AWS SQS or Azure Queues). You can then trigger off a worker via AWS Lambdas or Azure Functions which performs the task and then marks the task as completed.

Azure queues – updating messages

~ Nish ~ Leave a comment

The default invisibility timeout of 30 seconds may not suffice on occasions. For those situations, you can change the timeout so that you get more time to do something with your message prior to deleting it. 

void DeQueueMessagesWithTimeOut(std::chrono::seconds timeout)
{
  auto storage_account = cloud_storage_account::parse(
      U("UseDevelopmentStorage=true"));
  auto queue_client = storage_account.create_cloud_queue_client();
  auto queue = queue_client.get_queue_reference(
      U("sample"));
  bool created = queue.create_if_not_exists();

  queue.download_attributes();
  int count = queue.approximate_message_count();
  ucout << U("Approx count = ") << count << endl;

  queue_request_options options;
  operation_context op_context;
  auto messages = queue.get_messages(
      count, timeout, options, op_context);
  for (auto message : messages)
  {
    // process the message here
    queue.delete_message(message);
    ucout << message.content_as_string() << endl;
  }

  queue.download_attributes();
  count = queue.approximate_message_count();
  ucout << U("Approx count = ") << count << endl;
}

You can also update an existing message. The code snippet below shows how that can be done.

void UpdateQueueMessage()
{
  auto storage_account = cloud_storage_account::parse(
      U("UseDevelopmentStorage=true"));
  auto queue_client = storage_account.create_cloud_queue_client();
  auto queue = queue_client.get_queue_reference(
      U("sample"));
  bool created = queue.create_if_not_exists(); 

  auto message = queue.get_message();
  message.set_content(U("Updated content"));
  queue.update_message(
      message, std::chrono::seconds(0), true);
}

One thing to be aware of is that the position of the message in the queue is lost. The updated content goes back to the back of the queue. This shouldn’t matter though for most practical scenarios as the order of items in the queue is not expected to be depended on by consumers. This might be a little different to how you used queues in your programming classes, where it was a guaranteed FIFO structure. But remember, those queues did not allow in-place edits either.

Using Azure queues from C++

~ Nish ~ Leave a comment

Azure queues are used to store a large number of items (referred to as messages). Each message can go up to 64 Kb. A typical use for queues is intra-app communication, for example your website might add messages to the queue for a background service to process later on. The C++ Azure storage SDK maintains its consistent API when it comes to queues as well (so if you know how to use blobs and tables, you are good). Here’s a code snippet that shows how to insert messages into a queue.

void InsertQueueMessages()
{
  auto storage_account = cloud_storage_account::parse(
      U("UseDevelopmentStorage=true"));
  auto queue_client = storage_account
      .create_cloud_queue_client();
  auto queue = queue_client.get_queue_reference(
      U("sample"));
  bool created = queue.create_if_not_exists();

  queue.add_message(cloud_queue_message(
      U("queue message 01")));
  queue.add_message(cloud_queue_message(
      U("queue message 02")));
  queue.add_message(cloud_queue_message(
      U("queue message 03")));
}

Like any decent queue, you can either peek or dequeue a message. Here’s an example snippet that shows how to peek into a queue. 

void PeekQueueMessage()
{
  auto storage_account = cloud_storage_account::parse(
      U("UseDevelopmentStorage=true"));
  auto queue_client = storage_account
      .create_cloud_queue_client();
  auto queue = queue_client.get_queue_reference(
      U("sample"));
  bool created = queue.create_if_not_exists();

  queue.download_attributes();
  int count = queue.approximate_message_count();
  ucout << U("Approx count = ") << count << endl;

  // Peek the next 2 messages (assumes at least 2 present)
  auto messages = queue.peek_messages(2);
  for (auto message : messages)
  {
    ucout << message.content_as_string() << endl;
  }
}

The count method has ‘approximate‘ in the name because the count returned is not refreshed until the next call to download_attributes.

Here’s a code snippet that shows how to dequeue items from a queue.

void DeQueueMessages()
{
  auto storage_account = cloud_storage_account::parse(
      U("UseDevelopmentStorage=true"));
  auto queue_client = storage_account
      .create_cloud_queue_client();
  auto queue = queue_client.get_queue_reference(
      U("sample"));
  bool created = queue.create_if_not_exists();

  queue.download_attributes();
  int count = queue.approximate_message_count();
  ucout << U("Approx count = ") << count << endl;

  for (size_t i = 0; i < count; i++)
  {
    auto message = queue.get_message();
    queue.delete_message(message);
    ucout << message.content_as_string() << endl;
  }

  queue.download_attributes();
  count = queue.approximate_message_count();
  ucout << U("Approx count = ") << count << endl;
}

Notice how I call delete_message after calling get_message. This is because get_message only hides the item from other callers for a default time out (30 seconds) after which the message is back. This allows you to rollback changes in case something goes wrong. So you’d get a message, process it, and when you are done processing it, you can call delete_message. This assumes you do your processing in under 30 seconds. If you want to increase (or decrease) that time out, you can do that too. I’ll cover that in the next blog entry along with showing how to update a message in the queue.

Downloading blob content

~ Nish ~ Leave a comment

I missed this in the previous blog entry. The SDK makes it really trivial to extract blob data. For text content you can directly get the text, and for non-text content you can either download to a file or to a stream. The code snippet below shows how it’s done.

void DownloadBlobData()
{
  auto storage_account = cloud_storage_account::parse(
      U("UseDevelopmentStorage=true"));
  auto blob_client = storage_account.create_cloud_blob_client();
  auto container = blob_client.get_container_reference(
      U("textdata"));
  bool created = container.create_if_not_exists();

  // Read the text content directly
  auto text_blob1 = container.get_block_blob_reference(
      U("texts/text1"));
  auto text = text_blob1.download_text();
  ucout << text << endl;

  // Download the blob data to a file
  auto text_blob2 = container.get_block_blob_reference(
      U("texts/text2"));
  text_blob2.download_to_file(
      U("d:\\tmp\\blobdata.txt"));

  // Download the blob data to an ostream
  stringstreambuf buffer;
  concurrency::streams::ostream output(buffer);
  text_blob2.download_to_stream(output);
  cout << buffer.collection() << endl;
}

Using Azure blob storage from C++

~ Nish ~ Leave a comment

Blob storage is for storing large amounts of semi-structured or unstructured data such as images, videos, documents etc. The blob service lets you create named containers that can then contain one or more named blobs which can be publicly (optional) accessed via an URI.

Creating blobs

void CreateTextBlobs()
{
  auto storage_account = cloud_storage_account::parse(
      U("UseDevelopmentStorage=true"));
  auto blob_client = storage_account.create_cloud_blob_client();
  auto container = blob_client.get_container_reference(
      U("textdata"));
  bool created = container.create_if_not_exists();

  blob_container_permissions permissions;
  permissions.set_public_access(
      blob_container_public_access_type::container);
  container.upload_permissions(permissions);

  auto text_blob1 = container.get_block_blob_reference(
      U("texts/text1"));
  text_blob1.upload_text(U("This is some text - modified"));

  auto text_blob2 = container.get_block_blob_reference(
      U("texts/text2"));
  text_blob2.upload_from_file(U("./stdafx.h"));
}

The classes/methods used are very similar to that used for table storage. Notice the use of blob_container_permissions to set the public access level for the container. It’s off by default, and you can optionally set it to blob (clients can read blob data) or container (clients can list blobs in the container and read blob data). You can simulate directories in the blob names. In the above example, both text1 and text2 are under the texts directory. This just affects the URI and are not physical directories.

Listing blobs

void ListTextBlobs()
{
  auto storage_account = cloud_storage_account::parse(
        U("UseDevelopmentStorage=true"));
  auto blob_client = storage_account.create_cloud_blob_client();
  auto container = blob_client.get_container_reference(
        U("textdata"));
  bool created = container.create_if_not_exists();

  continuation_token token;

  auto result = container.list_blobs_segmented(token);

  for (auto dir : result.directories())
  {
    ucout << U("Directory: ") << dir.uri().path() << endl;
    ucout << endl;

    continuation_token dir_token;
    auto resultInner = dir.list_blobs_segmented(dir_token);

    for (auto item : resultInner.blobs())
    {
      ucout << item.name() << endl;
      ucout << item.uri().path() << endl;
      ucout << item.properties().content_type() << endl;
      ucout << endl;
    }
  }
}

To list the blobs in a container, we need to use a continuation_token object. The container object supports a list_blobs_segmented method which takes this token. For each directory returned, we can then call list_blobs_segmented using a separate continuation_token object. Once we iterate through the blobs in the directory, we can access properties like the name, URI, content type etc. Here’s a sample output from calling the above fucntion.

Directory: /devstoreaccount1/textdata/texts/

texts/text1
/devstoreaccount1/textdata/texts/text1
text/plain; charset=utf-8

texts/text2
/devstoreaccount1/textdata/texts/text2
application/octet-stream

Notice how the text data we uploaded is of type text/plain whereas the file we uploaded is of type application/octet-stream. If you paste the first blob’s URI in a browser, the text is directly returned. Whereas in the second case, the file is offered for download – obviously this is browser specific behavior based on the content type.