Uniphar.Cosmos.Extensions 4.4.1

cosmos-extensions

NuGet package repository for Cosmos Extensions

Purpose

This library is intended for provisioning Az Cosmos Database(s) and container(s).

It also updates TTL and indexing policy for existing containers.

Setup your project to use Cosmos Extension package

When using this NuGet package you need to setup a few things in your project

appsettings.json

Add this to structure to your appsettings.json with the appropriate values needed in your solution. Note that this contains examples from PIMS API. They are here to illustrate example values.

  "StoreOptions": {
    "DatabaseName": "pims-api",
    "DatabaseThroughput": <provisioned database throughput in RUs>,
    "Containers": [
      {
        "InstanceName": "responsePayloadContainer",
        "PartitionKey": "/partition",
        "DefaultTimeToLive": "128.0:0:0.0"
      },
      {
        "InstanceName": "auditContainer",
        "PartitionKey": "/partition",
        "IncludedPaths": [
          {
            "Path": "/myProperty/*"
          }
        ]
        "ExcludedPaths": [
          {
            "Path": "/*"
          }
        ]
      },
      {
        "InstanceName": "sequenceContainer",
        "PartitionKey": "/id",
        "IndexingMode": "None"
      },
      //...more containers to be defined with potential settings 
      {
        "InstanceName": "<name of container>",
        "PartitionKey": "<name of partition key>",
        "ContainerThroughput": <provisioned database throughput in RUs>,
        "DefaultTimeToLive" :"128.0:0:0.0"
        "IndexingMode": "<indexing mode value>"
        "CompositeIndexes": [
          {
            "Paths": [
              {
                "Path": "/path1",
                "Order": "Ascending"
              },
              {
                "Path": "/path2",
                "Order": "Ascending"
              }
            ]
          }
        ]
      }
    ]
  }

NOTE 1: The PartitionKeyPath (unless it is also "/id") is not indexed and should be included in the index.

NOTE 2: The PartitionKey should always be specified in the request options object in Linux/MacOS clients

NOTE 3: Worth noting on the settings.

• DatabaseName: Name of the database to be checked/created. Mandatory.
• DatabaseThroughput: integer, if undefined (null) it will default to 400; it will provision database with auto scaling as Manual throughput has been dismissed.
• Containers:InstanceName: string, name of container. Mandatory.
• Containers:PartitionKey: string, if undefined (null), it will default to partitioning by /partition Be aware that partitionKey in appsettings will need to define a name preceded with a forwards slash
• Containers:ContainerThroughput: integer, if undefined (null) it will default to 400
• DefaultTimeToLiveAsTimeSpan: Optional TimeSpan defining for how long data should be retained; if undefined (null) we implement it to default to -1 (items don’t expire by default). The example shows TTL defined as 128 days, zero days, zero minutes and zero seconds
• IndexingMode: Optional If undefined (null), it will default to IndexingMode.Consistent. If required, then use one of these enumerated values (with capitalization!!!): None, Lazy, Consistent.
• CompositeIndexes: Optional The set of composite indexes to add to the parent container.

Range Indexes

https://learn.microsoft.com/en-us/azure/cosmos-db/index-policy

TLDR;

• /* must be in IncludedPaths or ExcludedPaths to enable range indexes.
• The path must end with ?, [], or * to enable range indexes.
  • ? - Single value
  • [] - Array. Numbers are also supported to index a specific array element. my-array/1/my-property/?
  • * - Anything

Composite Indexes

See here for more details about composite indexes.

Program.cs

Add the following to Program.cs

• Add this to bind the above configuration section to class StoreOptions. That will ensure we can DI it into the where ever we need to make the call.

builder.Services.Configure<StoreOptions>(
    builder.Configuration.GetSection(nameof(StoreOptions)));

• Register a singleton instance of a CosmosExtension

builder.Services.AddSingleton<CosmosExtension>();

• Register a singleton instance of CosmosClient (MS recommendation) by adding the following lines:

// System.Text.Json with JsonStringEnumConverter
var cosmosClientOptions = new CosmosClientOptions().DefaultOptions(); 

// Register the client
var connectionString = builder.Configuration["Cosmos:ConnectionString"]!;
var cosmosClient = new CosmosClient(connectionString, cosmosClientOptions);
builder.Services.AddSingleton(cosmosClient);

NOTE: Cosmos Client ConnectionString is defined in KeyVault.

CosmosInitialiser.cs (or other file responsible of Cosmos Initialization)

Whatever file responsible for initializing cosmos database and containers (called from StartupBackgroundService.cs) has a method ExecuteAsync(CancellationToken stoppingToken). That method can be simplified to this

public async Task ExecuteAsync(CancellationToken stoppingToken)
{
    _logger.LogInformation("Creating Cosmos Database And Containers");
    
    await _cosmosExtension.ExecuteAsync(stoppingToken);
}

Create extended class of StoreOptions

Some APIs may have additional logic that is specific to the solutions. You can then create a class extension that inherits from StoreOptions. This is particular valid for PIMS API. You would create something like this.

    public class ExtendedStoreOptions : StoreOptions
    {
        /// <summary>
        /// Disable the account processing check
        /// </summary>
        /// <remarks>
        /// If set to true all account processing checks will return true
        /// Used to disable routing of requests to legacy pims system
        /// </remarks>
        public bool DisableAccountCheck { get; set; } = false;
    }

Your appsettings.json would look like this then (based on PIMS API as an example)

  "ExtendedStoreOptions": {
    "DatabaseName": "<name of database>",
    "DatabaseThroughput": <provisioned database throughput in RUs>,
    "DisableAccountCheck": true, // < --- this is the extra field from the above example
    "Containers": [
      {
        "InstanceName": "<name of container>",
        "PartitionKey": "<name of partition key>",
        "ContainerThroughput": <provisioned database throughput in RUs>,
        "DefaultTimeToLiveInSeconds" : "128.0:0:0.0"
      },
      //...more containers to be defined
    ]
  }

Your program.cs would then need to bind the config section like this

builder.Services.Configure<ExtendedStoreOptions>(
    builder.Configuration.GetSection(nameof(ExtendedStoreOptions)));

Likewise you;ll need to update Store.cs to reflect using ExtendedStoreOptions instead.

Cosmos Repository

ICosmosRepository<T> is a wrapper around Container which handles exception handling, telemetry, and utility methods. Also, there is no need to specify the type anymore on each method.

The below code register CosmosClient, and ICosmosRepository<T> as singletons:

// Program.cs
var connectionString = builder.Configuration["Cosmos:ConnectionString"]!;
var databaseId = builder.Configuration["StoreOptions:DatabaseName"]!;
builder.AddCosmos(connectionString, databaseId)
    .AddRepository<EntityType1>("myEntityContainer1")
    .AddRepository<EntityType2>("myEntityContainer2")

TTelemetry has to extend IBaseTelemetry.

To use, just inject ICosmosRepository<T> into your class:

public class MyService(
    ICosmosRepository<MyEntity> entityRepository)
{
}

Handled exception status codes

When the following exceptions are thrown, the methods returns null.

Queries

There are a couple of helper methods to enumerate Cosmos queries from FeedIterator<T>:

• ToListAsync(cancellationToken)
• ToAsyncEnumerable(cancellationToken)
• FirstOrDefaultAsync(cancellationToken)

Telemetry

• Each method sends a CosmosDbMetrics that logs RU usage and other metrics.
• On exception, it sends a UnsuccessfulCosmosRequestEvent.

It is possible to disable telemetry at:

• Globally: in AddCosmos.
• Repository level: in AddRepository.
• In each method.

Keyed Repository

If two or more repositories use the same object type, it is possible to use a keyed repository.

// Program.cs
.AddRepository<Entity>("rightEntity", serviceKey: "right")
.AddRepository<Entity>("leftEntity", serviceKey: "left")

// Service.cs
[FromKeyedServices("right")] ICosmosRepository<Entity> rightRepository,

Cosmos Exception Handler

There is a global exception handler to log CosmosException methods. It is registered automatically with AddCosmos.

To register it manually:

builder.Services.AddSingleton<IExceptionHandler, CosmosExceptionHandler>(provider =>
      {
          var telemetry = provider.GetRequiredService<ExampleServiceTelemetry>();
          return new CosmosExceptionHandler(telemetry!);
      });

Cosmos Mutex

The CosmosMutex class is an implementation of a simple distributed mutex. The CosmosMutex utilizes a Cosmos container as a backing durable store for a give mutex state.

Initialization

An instance of a CosmosMutex must be initialized before it can be used by an application. This initialization can be performed during startup.

To initialize a CosmosMutex invoke the InitializeAsync during the startup initialization of the application.

The InitializeAsync method requires the following parameters:

• mutexInstance
  • The name of the mutex that is being acquired

public override async Task StartAsync(CancellationToken cancellationToken)
{
   _logger!.LogInformation("Creating Cosmos Database And Containers");

    // Supply the mutex instance name to the initialize method
    await _mutex.InitializeAsync(MutexName);

    _logger!.LogInformation("Health Check Startup Completed");
    _healthCheck!.StartupCompleted = true;
}

Acquire

In order to access some distributed data in an exclusive access context a named CosmosMutex must be acquired by a process.

The AcquireAsync method requires the following parameters:

• Owner
  • The Owner name of the process attempting to acquire the mutex instance
• mutexInstance
  • The name of the mutex that is being acquired
• leasExpiry
  • The maximum length of time the mutex can be held. An application must allow for enough time to complete its exclusive processing

Note: The mutex is released after the leaseExpiry time has elapsed.

// Attempt to acquire the mutex supplying the machine name as the acquiring owner of the mutex
if (await _mutex.AcquireAsync(Environment.MachineName, MutexName, _mutexLeaseTime))
{
    // Perform execution in a mutually exclusive context
    await ProcessAsync(cancellationToken);

    // Release the mutex
    await _mutex.ReleaseAsync(Environment.MachineName, MutexName, cancellationToken);
}

Release

After a process has completed executing its mutually exclusive operation an acquired CosmosMutex must be released.

The ReleaseAsync method requires the following parameters:

• owner
  • The Owner name of the process attempting to acquire the mutex instance
• mutexInstance
  • The name of the mutex that is being acquired

{
    // Release the mutex
    await _mutex.ReleaseAsync(Environment.MachineName, MutexName, cancellationToken);
}