Open Telemetry: Fixes Operation name to follow Otel convention (#4643)

* Changed Operation name according to open telemetry convention

* fix tests

* updated baseline

* fix tests

* renamed ops name

* fixed operatio name

* reverted few files

* append prefix

* fix tests

* remove commented code

* fix activity name

* code refatot for different operation name for diagnostics

* baseline files

* fix tests

* refcator code

* fix test

* remove stream

* feediterator changes

* fix bugs

* further fixes

* typo fix

* fix test

Microsoft.Azure.Cosmos/src/Batch/BatchCore.cs

@@ -10,6 +10,7 @@ namespace Microsoft.Azure.Cosmos
  using System.Linq;
  using System.Threading;
  using System.Threading.Tasks;
+ using Microsoft.Azure.Cosmos.Telemetry.OpenTelemetry;
  using Microsoft.Azure.Documents;
 
  internal class BatchCore : TransactionalBatchInternal
@@ -18,6 +19,11 @@ internal class BatchCore : TransactionalBatchInternal
 
  private readonly ContainerInternal container;
 
+ /// <summary>
+ /// The list of operations in the batch.
+ /// </summary>
+ protected List<ItemBatchOperation> operations;
+
  /// <summary>
  /// Initializes a new instance of the <see cref="BatchCore"/> class.
  /// </summary>
@@ -29,6 +35,8 @@ internal BatchCore(
  {
  this.container = container;
  this.partitionKey = partitionKey;
+
+ this.operations = new List<ItemBatchOperation>();
  }
 
  public override TransactionalBatch CreateItem<T>(
@@ -40,7 +48,7 @@ public override TransactionalBatch CreateItem<T>(
  throw new ArgumentNullException(nameof(item));
  }
 
- this.AddOperation(new ItemBatchOperation<T>(
+ this.operations.Add(new ItemBatchOperation<T>(
  operationType: OperationType.Create,
  operationIndex: this.operations.Count,
  resource: item,
@@ -59,7 +67,7 @@ public override TransactionalBatch CreateItemStream(
  throw new ArgumentNullException(nameof(streamPayload));
  }
 
- this.AddOperation(new ItemBatchOperation(
+ this.operations.Add(new ItemBatchOperation(
  operationType: OperationType.Create,
  operationIndex: this.operations.Count,
  resourceStream: streamPayload,
@@ -78,7 +86,7 @@ public override TransactionalBatch ReadItem(
  throw new ArgumentNullException(nameof(id));
  }
 
- this.AddOperation(new ItemBatchOperation(
+ this.operations.Add(new ItemBatchOperation(
  operationType: OperationType.Read,
  operationIndex: this.operations.Count,
  id: id,
@@ -97,7 +105,7 @@ public override TransactionalBatch UpsertItem<T>(
  throw new ArgumentNullException(nameof(item));
  }
 
- this.AddOperation(new ItemBatchOperation<T>(
+ this.operations.Add(new ItemBatchOperation<T>(
  operationType: OperationType.Upsert,
  operationIndex: this.operations.Count,
  resource: item,
@@ -116,7 +124,7 @@ public override TransactionalBatch UpsertItemStream(
  throw new ArgumentNullException(nameof(streamPayload));
  }
 
- this.AddOperation(new ItemBatchOperation(
+ this.operations.Add(new ItemBatchOperation(
  operationType: OperationType.Upsert,
  operationIndex: this.operations.Count,
  resourceStream: streamPayload,
@@ -141,7 +149,7 @@ public override TransactionalBatch ReplaceItem<T>(
  throw new ArgumentNullException(nameof(item));
  }
 
- this.AddOperation(new ItemBatchOperation<T>(
+ this.operations.Add(new ItemBatchOperation<T>(
  operationType: OperationType.Replace,
  operationIndex: this.operations.Count,
  id: id,
@@ -167,7 +175,7 @@ public override TransactionalBatch ReplaceItemStream(
  throw new ArgumentNullException(nameof(streamPayload));
  }
 
- this.AddOperation(new ItemBatchOperation(
+ this.operations.Add(new ItemBatchOperation(
  operationType: OperationType.Replace,
  operationIndex: this.operations.Count,
  id: id,
@@ -187,7 +195,7 @@ public override TransactionalBatch DeleteItem(
  throw new ArgumentNullException(nameof(id));
  }
 
- this.AddOperation(new ItemBatchOperation(
+ this.operations.Add(new ItemBatchOperation(
  operationType: OperationType.Delete,
  operationIndex: this.operations.Count,
  id: id,
@@ -232,10 +240,7 @@ public override Task<TransactionalBatchResponse> ExecuteAsync(
  this.operations = new List<ItemBatchOperation>();
  return executor.ExecuteAsync(trace, cancellationToken);
  },
- openTelemetry: (response) => new OpenTelemetryResponse(
- responseMessage: response, 
- isHomogenousOperations: this.isHomogenousOperations, 
- batchOperation: this.homogenousOperation));
+ openTelemetry: new (OpenTelemetryConstants.Operations.ExecuteBatch, (response) => new OpenTelemetryResponse(responseMessage: response)));
  }
 
  /// <summary>
@@ -250,7 +255,7 @@ public virtual TransactionalBatch PatchItemStream(
  Stream patchStream,
  TransactionalBatchPatchItemRequestOptions requestOptions = null)
  {
- this.AddOperation(new ItemBatchOperation(
+ this.operations.Add(new ItemBatchOperation(
  operationType: OperationType.Patch,
  operationIndex: this.operations.Count,
  id: id,
@@ -286,7 +291,7 @@ public override TransactionalBatch PatchItem(
 
  PatchSpec patchSpec = new PatchSpec(patchOperations, requestOptions);
 
- this.AddOperation(new ItemBatchOperation<PatchSpec>(
+ this.operations.Add(new ItemBatchOperation<PatchSpec>(
  operationType: OperationType.Patch,
  operationIndex: this.operations.Count,
  id: id,

Microsoft.Azure.Cosmos/src/Batch/TransactionalBatchInternal.cs

@@ -5,8 +5,6 @@
 namespace Microsoft.Azure.Cosmos
 {
  using System.Collections.Generic;
- using System.Linq;
- using Microsoft.Azure.Documents;
 
  /// <summary>
  /// Represents an internal abstract class for handling transactional batches of operations.
@@ -15,47 +13,5 @@ namespace Microsoft.Azure.Cosmos
  /// </summary>
  internal abstract class TransactionalBatchInternal : TransactionalBatch
  {
- /// <summary>
- /// The list of operations in the batch.
- /// </summary>
- protected List<ItemBatchOperation> operations;
-
- /// <summary>
- /// Initializes a new instance of the <see cref="TransactionalBatchInternal"/> class.
- /// </summary>
- public TransactionalBatchInternal()
- {
- this.operations = new List<ItemBatchOperation>();
- }
-
- /// <summary>
- /// Indicates whether all operations in the batch are of the same type.
- /// </summary>
- internal bool isHomogenousOperations = true;
-
- /// <summary>
- /// Stores the operation type if all operations in the batch are of the same type; otherwise, null.
- /// </summary>
- internal OperationType? homogenousOperation = null;
-
- /// <summary>
- /// Adds an operation to the batch.
- /// </summary>
- /// <param name="itemBatchOperation">The operation to add to the batch.</param>
- /// <remarks>
- /// This method performs the following actions:
- /// 1. Checks if the batch is homogeneous (all operations of the same type) and if the new operation's type matches the type of the existing operations.
- /// 2. Updates the <see cref="isHomogenousOperations"/> flag and the <see cref="homogenousOperation"/> property based on the check.
- /// 3. Adds the operation to the list of operations.
- /// </remarks>
- protected void AddOperation(ItemBatchOperation itemBatchOperation)
- {
- if (this.isHomogenousOperations && this.operations.Count > 0)
- {
- this.isHomogenousOperations = this.operations.First().OperationType == itemBatchOperation.OperationType;
- this.homogenousOperation = this.isHomogenousOperations ? itemBatchOperation.OperationType : null;
- }
- this.operations.Add(itemBatchOperation);
- }
  }
 }

Microsoft.Azure.Cosmos/src/ChangeFeed/ChangeFeedIteratorCore.cs

@@ -13,6 +13,7 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed
  using Microsoft.Azure.Cosmos.Pagination;
  using Microsoft.Azure.Cosmos.Query.Core;
  using Microsoft.Azure.Cosmos.Query.Core.Monads;
+ using Microsoft.Azure.Cosmos.Telemetry.OpenTelemetry;
  using Microsoft.Azure.Cosmos.Tracing;
  using Microsoft.Azure.Documents;
 
@@ -50,6 +51,8 @@ public ChangeFeedIteratorCore(
  this.changeFeedRequestOptions = changeFeedRequestOptions ?? new ChangeFeedRequestOptions();
  this.changeFeedQuerySpec = changeFeedQuerySpec;
 
+ this.operationName = OpenTelemetryConstants.Operations.QueryChangeFeed;
+
  this.lazyMonadicEnumerator = new AsyncLazy<TryCatch<CrossPartitionChangeFeedAsyncEnumerator>>(
  valueFactory: async (trace, cancellationToken) =>
  {
@@ -226,7 +229,7 @@ public override async Task<ResponseMessage> ReadNextAsync(CancellationToken canc
  operationType: OperationType.ReadFeed,
  requestOptions: this.changeFeedRequestOptions,
  task: (trace) => this.ReadNextInternalAsync(trace, cancellationToken),
- openTelemetry: (response) => new OpenTelemetryResponse(responseMessage: response),
+ openTelemetry: new (OpenTelemetryConstants.Operations.QueryChangeFeed, (response) => new OpenTelemetryResponse(responseMessage: response)),
  traceComponent: TraceComponent.ChangeFeed,
  traceLevel: TraceLevel.Info);
  }

Microsoft.Azure.Cosmos/src/ChangeFeed/ChangeFeedPartitionKeyResultSetIteratorCore.cs

@@ -9,7 +9,7 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed
  using System.Threading;
  using System.Threading.Tasks;
  using Microsoft.Azure.Cosmos.ChangeFeed.LeaseManagement;
- using Microsoft.Azure.Cosmos.CosmosElements;
+ using Microsoft.Azure.Cosmos.Telemetry.OpenTelemetry;
  using Microsoft.Azure.Cosmos.Tracing;
  using Microsoft.Azure.Documents;
 
@@ -81,6 +81,8 @@ private ChangeFeedPartitionKeyResultSetIteratorCore(
  this.changeFeedStartFrom = changeFeedStartFrom ?? throw new ArgumentNullException(nameof(changeFeedStartFrom));
  this.clientContext = this.container.ClientContext;
  this.changeFeedOptions = options;
+
+ this.operationName = OpenTelemetryConstants.Operations.QueryChangeFeed;
  }
 
  public override bool HasMoreResults => this.hasMoreResultsInternal;
@@ -99,8 +101,6 @@ public override Task<ResponseMessage> ReadNextAsync(CancellationToken cancellati
  operationType: Documents.OperationType.ReadFeed,
  requestOptions: this.changeFeedOptions,
  task: (trace) => this.ReadNextAsync(trace, cancellationToken),
- openTelemetry: (response) => new OpenTelemetryResponse(
- responseMessage: response),
  traceComponent: TraceComponent.ChangeFeed,
  traceLevel: TraceLevel.Info);
  }

Microsoft.Azure.Cosmos/src/ChangeFeed/StandByFeedIteratorCore.cs

@@ -11,6 +11,7 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed
  using System.Threading.Tasks;
  using Microsoft.Azure.Cosmos.CosmosElements;
  using Microsoft.Azure.Cosmos.Routing;
+ using Microsoft.Azure.Cosmos.Telemetry.OpenTelemetry;
  using Microsoft.Azure.Cosmos.Tracing;
  using Microsoft.Azure.Documents;
 
@@ -45,6 +46,8 @@ internal StandByFeedIteratorCore(
  this.changeFeedOptions = options;
  this.maxItemCount = maxItemCount;
  this.continuationToken = continuationToken;
+
+ this.operationName = OpenTelemetryConstants.Operations.QueryChangeFeed;
  }
 
  /// <summary>

Microsoft.Azure.Cosmos/src/ChangeFeedProcessor/ChangeFeedEstimatorIterator.cs

@@ -18,6 +18,7 @@ namespace Microsoft.Azure.Cosmos.ChangeFeed
  using Microsoft.Azure.Cosmos.Diagnostics;
  using Microsoft.Azure.Cosmos.Query.Core;
  using Microsoft.Azure.Cosmos.Query.Core.Monads;
+ using Microsoft.Azure.Cosmos.Telemetry.OpenTelemetry;
  using Microsoft.Azure.Cosmos.Tracing;
  using Newtonsoft.Json.Linq;
 
@@ -106,6 +107,8 @@ private ChangeFeedEstimatorIterator(
 
  this.monitoredContainerFeedCreator = monitoredContainerFeedCreator;
  this.documentServiceLeaseContainer = documentServiceLeaseContainer;
+
+ this.operationName = OpenTelemetryConstants.Operations.QueryChangeFeedEstimator;
  }
 
  public override bool HasMoreResults => this.hasMoreResults;
@@ -119,7 +122,7 @@ public override Task<FeedResponse<ChangeFeedProcessorState>> ReadNextAsync(Cance
  operationType: Documents.OperationType.ReadFeed,
  requestOptions: null,
  task: (trace) => this.ReadNextAsync(trace, cancellationToken),
- openTelemetry: (response) => new OpenTelemetryResponse<ChangeFeedProcessorState>(responseMessage: response),
+ openTelemetry: new (OpenTelemetryConstants.Operations.QueryChangeFeedEstimator, (response) => new OpenTelemetryResponse<ChangeFeedProcessorState>(responseMessage: response)),
  traceComponent: TraceComponent.ChangeFeed,
  traceLevel: TraceLevel.Info);
  }

Microsoft.Azure.Cosmos/src/CosmosClient.cs

@@ -18,6 +18,8 @@ namespace Microsoft.Azure.Cosmos
  using Microsoft.Azure.Cosmos.Handlers;
  using Microsoft.Azure.Cosmos.Query.Core.Monads;
  using Microsoft.Azure.Cosmos.Query.Core.QueryPlan;
+ using Microsoft.Azure.Cosmos.Telemetry;
+ using Microsoft.Azure.Cosmos.Telemetry.OpenTelemetry;
  using Microsoft.Azure.Cosmos.Tracing;
  using Microsoft.Azure.Cosmos.Tracing.TraceData;
  using Microsoft.Azure.Documents;
@@ -756,7 +758,7 @@ public virtual Task<DatabaseResponse> CreateDatabaseAsync(
  trace: trace,
  cancellationToken: cancellationToken);
  },
- openTelemetry: (response) => new OpenTelemetryResponse<DatabaseProperties>(responseMessage: response));
+ openTelemetry: new (OpenTelemetryConstants.Operations.CreateDatabase, (response) => new OpenTelemetryResponse<DatabaseProperties>(responseMessage: response)));
  }
 
  /// <summary>
@@ -804,7 +806,7 @@ public virtual Task<DatabaseResponse> CreateDatabaseAsync(
  trace: trace,
  cancellationToken: cancellationToken);
  },
- openTelemetry: (response) => new OpenTelemetryResponse<DatabaseProperties>(responseMessage: response));
+ openTelemetry: new (OpenTelemetryConstants.Operations.CreateDatabase, (response) => new OpenTelemetryResponse<DatabaseProperties>(responseMessage: response)));
  }
 
  /// <summary>
@@ -900,8 +902,7 @@ public virtual Task<DatabaseResponse> CreateDatabaseIfNotExistsAsync(
  return this.ClientContext.ResponseFactory.CreateDatabaseResponse(this.GetDatabase(databaseProperties.Id), readResponseAfterConflict);
  }
  },
- openTelemetry: (response) => new OpenTelemetryResponse<DatabaseProperties>(
- responseMessage: response));
+ openTelemetry: new (OpenTelemetryConstants.Operations.CreateDatabaseIfNotExists, (response) => new OpenTelemetryResponse<DatabaseProperties>(responseMessage: response)));
  }
 
  /// <summary>
@@ -1205,7 +1206,7 @@ public virtual Task<ResponseMessage> CreateDatabaseStreamAsync(
  trace,
  cancellationToken);
  },
- openTelemetry: (response) => new OpenTelemetryResponse(response));
+ openTelemetry: new (OpenTelemetryConstants.Operations.CreateDatabase, (response) => new OpenTelemetryResponse(response)));
  }
 
  /// <summary>
@@ -1288,7 +1289,7 @@ internal virtual Task<ResponseMessage> CreateDatabaseStreamAsync(
  }
 
  return this.ClientContext.OperationHelperAsync(
- operationName: nameof(CreateDatabaseIfNotExistsAsync),
+ operationName: nameof(CreateDatabaseStreamAsync),
  containerName: null,
  databaseName: databaseProperties.Id,
  operationType: OperationType.Create,
@@ -1303,7 +1304,7 @@ internal virtual Task<ResponseMessage> CreateDatabaseStreamAsync(
  trace,
  cancellationToken);
  },
- openTelemetry: (response) => new OpenTelemetryResponse(response));
+ openTelemetry: new (OpenTelemetryConstants.Operations.CreateDatabase, (response) => new OpenTelemetryResponse(response)));
  }
 
  private async Task<DatabaseResponse> CreateDatabaseInternalAsync(

Microsoft.Azure.Cosmos/src/Query/v3Query/QueryIterator.cs

@@ -19,6 +19,7 @@ namespace Microsoft.Azure.Cosmos.Query
  using Microsoft.Azure.Cosmos.Query.Core.Pipeline.Pagination;
  using Microsoft.Azure.Cosmos.Query.Core.QueryClient;
  using Microsoft.Azure.Cosmos.Query.Core.QueryPlan;
+ using Microsoft.Azure.Cosmos.Telemetry.OpenTelemetry;
  using Microsoft.Azure.Cosmos.Tracing;
 
  internal sealed class QueryIterator : FeedIteratorInternal
@@ -53,6 +54,8 @@ private QueryIterator(
  this.correlatedActivityId = correlatedActivityId;
 
  this.container = container;
+ this.operationName = OpenTelemetryConstants.Operations.QueryItems;
+ this.operationType = Documents.OperationType.Query;
  }
 
  public static QueryIterator Create(

Microsoft.Azure.Cosmos/src/ReadFeed/ReadFeedIteratorCore.cs

@@ -17,6 +17,7 @@ namespace Microsoft.Azure.Cosmos.ReadFeed
  using Microsoft.Azure.Cosmos.ReadFeed.Pagination;
  using Microsoft.Azure.Cosmos.Resource.CosmosExceptions;
  using Microsoft.Azure.Cosmos.Routing;
+ using Microsoft.Azure.Cosmos.Telemetry.OpenTelemetry;
  using Microsoft.Azure.Cosmos.Tracing;
  using Microsoft.Azure.Documents;
 
@@ -38,6 +39,7 @@ public ReadFeedIteratorCore(
  CancellationToken cancellationToken)
  {
  this.container = container;
+ this.operationName = OpenTelemetryConstants.Operations.ReadFeedRanges;
 
  this.queryRequestOptions = queryRequestOptions;
  readFeedPaginationOptions ??= ReadFeedExecutionOptions.Default;