[Internal] Autopilot: Add internal autopilot support

Microsoft.Azure.Cosmos/src/CosmosClient.cs

@@ -375,6 +375,47 @@ public virtual Task<DatabaseResponse> CreateDatabaseAsync(
                 cancellationToken: cancellationToken));
         }
 
+        /// <summary>
+        /// Sends a request for creating a database.
+        ///
+        /// A database manages users, permissions and a set of containers.
+        /// Each Azure Cosmos DB Database Account is able to support multiple independent named databases,
+        /// with the database being the logical container for data.
+        ///
+        /// Each Database consists of one or more containers, each of which in turn contain one or more
+        /// documents. Since databases are an administrative resource, the Service Master Key will be
+        /// required in order to access and successfully complete any action using the User APIs.
+        /// </summary>
+        /// <param name="id">The database id.</param>
+        /// <param name="throughputProperties">(Optional) The throughput provisioned for a database in measurement of Request Units per second in the Azure Cosmos DB service.</param>
+        /// <param name="requestOptions">(Optional) A set of options that can be set.</param>
+        /// <param name="cancellationToken">(Optional) <see cref="CancellationToken"/> representing request cancellation.</param>
+        /// <returns>A <see cref="Task"/> containing a <see cref="DatabaseResponse"/> which wraps a <see cref="DatabaseProperties"/> containing the resource record.</returns>
+        /// <seealso href="https://docs.microsoft.com/azure/cosmos-db/request-units">Request Units</seealso>
+#if INTERNAL
+        public
+#else
+        internal
+#endif
+        virtual Task<DatabaseResponse> CreateDatabaseAsync(
+                string id,
+                ThroughputProperties throughputProperties,
+                RequestOptions requestOptions = null,
+                CancellationToken cancellationToken = default(CancellationToken))
+        {
+            if (string.IsNullOrEmpty(id))
+            {
+                throw new ArgumentNullException(nameof(id));
+            }
+
+            DatabaseProperties databaseProperties = this.PrepareDatabaseProperties(id);
+            return TaskHelper.RunInlineIfNeededAsync(() => this.CreateDatabaseAsync(
+                databaseProperties: databaseProperties,
+                throughputProperties: throughputProperties,
+                requestOptions: requestOptions,
+                cancellationToken: cancellationToken));
+        }
+
         /// <summary>
         /// <para>Check if a database exists, and if it doesn't, create it.
         /// Only the database id is used to verify if there is an existing database. Other database properties 
@@ -709,6 +750,21 @@ internal DatabaseProperties PrepareDatabaseProperties(string id)
             return databaseProperties;
         }
 
+        internal Task<DatabaseResponse> CreateDatabaseAsync(
+                    DatabaseProperties databaseProperties,
+                    ThroughputProperties throughputProperties,
+                    RequestOptions requestOptions = null,
+                    CancellationToken cancellationToken = default(CancellationToken))
+        {
+            Task<ResponseMessage> response = this.CreateAutopilotDatabaseStreamInternalAsync(
+                streamPayload: this.ClientContext.SerializerCore.ToStream<DatabaseProperties>(databaseProperties),
+                throughputProperties: throughputProperties,
+                requestOptions: requestOptions,
+                cancellationToken: cancellationToken);
+
+            return this.ClientContext.ResponseFactory.CreateDatabaseResponseAsync(this.GetDatabase(databaseProperties.Id), response);
+        }
+
         internal Task<DatabaseResponse> CreateDatabaseAsync(
                     DatabaseProperties databaseProperties,
                     int? throughput = null,
@@ -724,6 +780,25 @@ internal Task<DatabaseResponse> CreateDatabaseAsync(
             return this.ClientContext.ResponseFactory.CreateDatabaseResponseAsync(this.GetDatabase(databaseProperties.Id), response);
         }
 
+        private Task<ResponseMessage> CreateAutopilotDatabaseStreamInternalAsync(
+                Stream streamPayload,
+                ThroughputProperties throughputProperties,
+                RequestOptions requestOptions = null,
+                CancellationToken cancellationToken = default(CancellationToken))
+        {
+            return this.ClientContext.ProcessResourceOperationStreamAsync(
+                resourceUri: this.DatabaseRootUri,
+                resourceType: ResourceType.Database,
+                operationType: OperationType.Create,
+                requestOptions: requestOptions,
+                cosmosContainerCore: null,
+                partitionKey: null,
+                streamPayload: streamPayload,
+                requestEnricher: (httpRequestMessage) => httpRequestMessage.AddThroughputPropertiesHeader(throughputProperties),
+                diagnosticsContext: null,
+                cancellationToken: cancellationToken);
+        }
+
         private Task<ResponseMessage> CreateDatabaseStreamInternalAsync(
                 Stream streamPayload,
                 int? throughput = null,

Microsoft.Azure.Cosmos/src/Handler/RequestMessage.cs

@@ -169,6 +169,24 @@ internal void AddThroughputHeader(int? throughputValue)
             }
         }
 
+        internal void AddThroughputPropertiesHeader(ThroughputProperties throughputProperties)
+        {
+            if (throughputProperties.Throughput.HasValue &&
+                (throughputProperties.MaxThroughput.HasValue || throughputProperties.AutoUpgradeMaxThroughputIncrementPercentage.HasValue))
+            {
+                throw new InvalidOperationException("Autoscale provisioned throughput can not be configured with fixed offer");
+            }
+
+            if (throughputProperties.Throughput.HasValue)
+            {
+                this.AddThroughputHeader(throughputProperties.Throughput);
+            }
+            else if (throughputProperties?.Content?.OfferAutopilotSettings != null)
+            {
+                this.Headers.Add(HttpConstants.HttpHeaders.OfferAutopilotSettings, throughputProperties.Content.OfferAutopilotSettings.GetJsonString());
+            }
+        }
+
         internal async Task AssertPartitioningDetailsAsync(CosmosClient client, CancellationToken cancellationToken)
         {
             if (this.IsMasterOperation())

Microsoft.Azure.Cosmos/src/Microsoft.Azure.Cosmos.csproj

@@ -73,7 +73,7 @@
   <ItemGroup Condition=" '$(ProjectRef)' != 'True' ">
     <PackageReference Include="Microsoft.Azure.Cosmos.Direct" Version="[$(DirectVersion)]" PrivateAssets="All" />
     <PackageReference Include="Microsoft.Azure.Cosmos.Serialization.HybridRow" Version="[$(HybridRowVersion)]" PrivateAssets="All" />
-    <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All"/>
+    <PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.0.0" PrivateAssets="All" />
   </ItemGroup>
 
   <ItemGroup>

Microsoft.Azure.Cosmos/src/Resource/Container/Container.cs

@@ -293,6 +293,23 @@ public abstract Task<ThroughputResponse> ReplaceThroughputAsync(
             RequestOptions requestOptions = null,
             CancellationToken cancellationToken = default(CancellationToken));
 
+#if INTERNAL
+        /// <summary>
+        /// Sets throughput provisioned for a container in measurement of request units per second in the Azure Cosmos service.
+        /// </summary>
+        /// <param name="throughputProperties">The Cosmos container throughput expressed in Request Units per second.</param>
+        /// <param name="requestOptions">(Optional) The options for the throughput request.</param>
+        /// <param name="cancellationToken">(Optional) <see cref="CancellationToken"/> representing request cancellation.</param>
+        /// <returns>The throughput response.</returns>
+        /// <remarks>
+        /// <seealso href="https://docs.microsoft.com/azure/cosmos-db/request-units">Request Units</seealso>
+        /// </remarks>
+        public abstract Task<ThroughputResponse> ReplaceThroughputPropertiesAsync(
+            ThroughputProperties throughputProperties,
+            RequestOptions requestOptions = null,
+            CancellationToken cancellationToken = default(CancellationToken));
+#endif
+
         /// <summary>
         /// Creates a Item as an asynchronous operation in the Azure Cosmos service.
         /// </summary>

Microsoft.Azure.Cosmos/src/Resource/Container/ContainerCore.cs

@@ -162,6 +162,25 @@ internal async Task<ThroughputResponse> ReplaceThroughputIfExistsAsync(
                 cancellationToken: cancellationToken);
         }
 
+#if INTERNAL
+        public override
+#else
+        internal
+#endif
+        async Task<ThroughputResponse> ReplaceThroughputPropertiesAsync(
+            ThroughputProperties throughputProperties,
+            RequestOptions requestOptions = null,
+            CancellationToken cancellationToken = default)
+        {
+            string rid = await this.GetRIDAsync(cancellationToken);
+            CosmosOffers cosmosOffers = new CosmosOffers(this.ClientContext);
+            return await cosmosOffers.ReplaceThroughputPropertiesAsync(
+                rid,
+                throughputProperties,
+                requestOptions,
+                cancellationToken);
+        }
+
         public override Task<ResponseMessage> DeleteContainerStreamAsync(
             ContainerRequestOptions requestOptions = null,
             CancellationToken cancellationToken = default(CancellationToken))

Microsoft.Azure.Cosmos/src/Resource/Container/ContainerInlineCore.cs

@@ -100,6 +100,16 @@ public override Task<ThroughputResponse> ReplaceThroughputAsync(
             return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReplaceThroughputAsync(throughput, requestOptions, cancellationToken));
         }
 
+#if INTERNAL
+        public override
+#else
+        internal
+#endif
+        Task<ThroughputResponse> ReplaceThroughputPropertiesAsync(ThroughputProperties throughputProperties, RequestOptions requestOptions = null, CancellationToken cancellationToken = default)
+        {
+            return TaskHelper.RunInlineIfNeededAsync(() => this.container.ReplaceThroughputPropertiesAsync(throughputProperties, requestOptions, cancellationToken));
+        }
+
         public override Task<ResponseMessage> CreateItemStreamAsync(
             Stream streamPayload,
             PartitionKey partitionKey,
@@ -262,6 +272,7 @@ public override TransactionalBatch CreateTransactionalBatch(PartitionKey partiti
         {
             return this.container.CreateTransactionalBatch(partitionKey);
         }
+
 #if PREVIEW
         public override Task<IReadOnlyList<FeedToken>> GetFeedTokensAsync(CancellationToken cancellationToken = default(CancellationToken))
         {

Microsoft.Azure.Cosmos/src/Resource/Database/Database.cs

@@ -158,6 +158,67 @@ public abstract Task<ThroughputResponse> ReadThroughputAsync(
             RequestOptions requestOptions,
             CancellationToken cancellationToken = default(CancellationToken));
 
+#if INTERNAL
+        /// <summary>
+        /// Sets throughput provisioned for a database in measurement of request units per second in the Azure Cosmos service.
+        /// </summary>
+        /// <param name="throughputProperties">The Cosmos database throughput expressed in Request Units per second.</param>
+        /// <param name="requestOptions">(Optional) The options for the throughput request.</param>
+        /// <param name="cancellationToken">(Optional) <see cref="CancellationToken"/> representing request cancellation.</param>
+        /// <returns>The throughput response.</returns>
+        /// <value>
+        /// The provisioned throughput for this database.
+        /// </value>
+        /// <example>
+        /// The following example shows how to get the throughput.
+        /// <code language="c#">
+        /// <![CDATA[
+        /// ThroughputProperties throughput = await this.cosmosDatabase.ReplaceThroughputPropertiesAsync(10000);
+        /// ]]>
+        /// </code>
+        /// </example>
+        /// <remarks>
+        /// <seealso href="https://docs.microsoft.com/azure/cosmos-db/request-units">Request Units</seealso>
+        /// </remarks>
+        public abstract Task<ThroughputResponse> ReplaceThroughputPropertiesAsync(
+            ThroughputProperties throughputProperties,
+            RequestOptions requestOptions = null,
+            CancellationToken cancellationToken = default(CancellationToken));
+
+        /// <summary>
+        /// Creates a container as an asynchronous operation in the Azure Cosmos service.
+        /// </summary>
+        /// <param name="containerProperties">The <see cref="ContainerProperties"/> object.</param>
+        /// <param name="throughputProperties">(Optional) The throughput provisioned for a container in measurement of Requests Units per second in the Azure Cosmos DB service.</param>
+        /// <param name="requestOptions">(Optional) The options for the request.</param>
+        /// <param name="cancellationToken">(Optional) <see cref="CancellationToken"/> representing request cancellation.</param>
+        /// <returns>A <see cref="Task"/> containing a <see cref="ContainerResponse"/> which wraps a <see cref="ContainerProperties"/> containing the read resource record.</returns>
+        /// <exception cref="ArgumentNullException">If <paramref name="containerProperties"/> is not set.</exception>
+        /// <exception cref="System.AggregateException">Represents a consolidation of failures that occurred during async processing. Look within InnerExceptions to find the actual exception(s).</exception>
+        /// <exception cref="CosmosException">This exception can encapsulate many different types of errors. To determine the specific error always look at the StatusCode property. Some common codes you may get when creating a container are:
+        /// <list type="table">
+        ///     <listheader>
+        ///         <term>StatusCode</term><description>Reason for exception</description>
+        ///     </listheader>
+        ///     <item>
+        ///         <term>400</term><description>BadRequest - This means something was wrong with the request supplied. It is likely that an id was not supplied for the new container.</description>
+        ///     </item>
+        ///     <item>
+        ///         <term>403</term><description>Forbidden - This means you attempted to exceed your quota for containers. Contact support to have this quota increased.</description>
+        ///     </item>
+        ///     <item>
+        ///         <term>409</term><description>Conflict - This means a <see cref="ContainerProperties"/> with an id matching the id you supplied already existed.</description>
+        ///     </item>
+        /// </list>
+        /// </exception>
+        /// <seealso href="https://docs.microsoft.com/azure/cosmos-db/request-units">Request Units</seealso>
+        public abstract Task<ContainerResponse> CreateContainerAsync(
+                    ContainerProperties containerProperties,
+                    ThroughputProperties throughputProperties,
+                    RequestOptions requestOptions = null,
+                    CancellationToken cancellationToken = default(CancellationToken));
+#endif
+
         /// <summary>
         /// Sets throughput provisioned for a database in measurement of request units per second in the Azure Cosmos service.
         /// </summary>

Microsoft.Azure.Cosmos/src/Resource/Database/DatabaseCore.cs

@@ -122,6 +122,65 @@ internal async Task<ThroughputResponse> ReplaceThroughputIfExistsAsync(
                 requestOptions: requestOptions,
                 cancellationToken: cancellationToken);
         }
+#if INTERNAL
+        public override
+#else
+        internal
+#endif
+        Task<ContainerResponse> CreateContainerAsync(
+            ContainerProperties containerProperties,
+            ThroughputProperties throughputProperties,
+            RequestOptions requestOptions = null,
+            CancellationToken cancellationToken = default)
+        {
+            if (containerProperties == null)
+            {
+                throw new ArgumentNullException(nameof(containerProperties));
+            }
+
+            this.ValidateContainerProperties(containerProperties);
+
+            Task<ResponseMessage> response = this.ProcessCollectionCreateAsync(
+                streamPayload: this.ClientContext.SerializerCore.ToStream(containerProperties),
+                throughputProperties: throughputProperties,
+                requestOptions: requestOptions,
+                cancellationToken: cancellationToken);
+
+            return this.ClientContext.ResponseFactory.CreateContainerResponseAsync(this.GetContainer(containerProperties.Id), response);
+        }
+
+#if INTERNAL
+        public override
+#else
+        internal
+#endif
+        async Task<ThroughputResponse> ReplaceThroughputPropertiesAsync(
+            ThroughputProperties throughputProperties,
+            RequestOptions requestOptions = null,
+            CancellationToken cancellationToken = default)
+        {
+            string rid = await this.GetRIDAsync(cancellationToken);
+            CosmosOffers cosmosOffers = new CosmosOffers(this.ClientContext);
+            return await cosmosOffers.ReplaceThroughputPropertiesAsync(
+                targetRID: rid,
+                throughputProperties: throughputProperties,
+                requestOptions: requestOptions,
+                cancellationToken: cancellationToken);
+        }
+
+        internal async Task<ThroughputResponse> ReplaceThroughputPropertiesIfExistsAsync(
+            ThroughputProperties throughputProperties,
+            RequestOptions requestOptions = null,
+            CancellationToken cancellationToken = default(CancellationToken))
+        {
+            string rid = await this.GetRIDAsync(cancellationToken);
+            CosmosOffers cosmosOffers = new CosmosOffers(this.ClientContext);
+            return await cosmosOffers.ReplaceThroughputPropertiesIfExistsAsync(
+                targetRID: rid,
+                throughputProperties: throughputProperties,
+                requestOptions: requestOptions,
+                cancellationToken: cancellationToken);
+        }
 
         public override Task<ResponseMessage> ReadStreamAsync(
             RequestOptions requestOptions = null,
@@ -675,6 +734,25 @@ internal void ValidateContainerProperties(ContainerProperties containerPropertie
             this.ClientContext.ValidateResource(containerProperties.Id);
         }
 
+        internal Task<ResponseMessage> ProcessCollectionCreateAsync(
+            Stream streamPayload,
+            ThroughputProperties throughputProperties,
+            RequestOptions requestOptions,
+            CancellationToken cancellationToken)
+        {
+            return this.ClientContext.ProcessResourceOperationStreamAsync(
+               resourceUri: this.LinkUri,
+               resourceType: ResourceType.Collection,
+               operationType: OperationType.Create,
+               cosmosContainerCore: null,
+               partitionKey: null,
+               streamPayload: streamPayload,
+               requestOptions: requestOptions,
+               requestEnricher: (httpRequestMessage) => httpRequestMessage.AddThroughputPropertiesHeader(throughputProperties),
+               diagnosticsContext: null,
+               cancellationToken: cancellationToken);
+        }
+
         internal Task<ResponseMessage> ProcessCollectionCreateAsync(
             Stream streamPayload,
             int? throughput,