First cut at EH perf tests (#2602)

* First cut at EH perf tests

* Skip EH benchmarks if we're not running live tests

Author: LarryOsterman

Cargo.lock

@@ -39,12 +39,17 @@ azure_identity.workspace = true
 azure_messaging_eventhubs = { path = ".", features = [
   "test",
 ] }
+criterion.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
 
 [features]
 test = []
 blob_checkpoint_store = []
 
+[[bench]]
+name = "benchmarks"
+harness = false
+
 [lints]
 workspace = true

sdk/eventhubs/azure_messaging_eventhubs/Cargo.toml

@@ -0,0 +1,139 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+use azure_identity::DefaultAzureCredential;
+use azure_messaging_eventhubs::{models::EventData, ProducerClient, SendEventOptions};
+use criterion::{criterion_group, criterion_main, Criterion};
+use std::{env, sync::Arc};
+use tokio::runtime::Runtime;
+
+//static INIT_LOGGING: std::sync::Once = std::sync::Once::new();
+
+fn setup() {
+    // INIT_LOGGING.call_once(|| {
+    //     println!("Setting up test logger...");
+
+    //     use tracing_subscriber::{fmt::format::FmtSpan, EnvFilter};
+    //     tracing_subscriber::fmt()
+    //         .with_env_filter(EnvFilter::from_default_env())
+    //         .with_span_events(FmtSpan::NEW | FmtSpan::CLOSE)
+    //         .with_ansi(std::env::var("NO_COLOR").map_or(true, |v| v.is_empty()))
+    //         .with_writer(std::io::stderr)
+    //         .init();
+    // });
+}
+
+fn send_batch_benchmark(c: &mut Criterion) {
+    setup();
+
+    // Check if the environment variable is set thus allowing the benchmarks to run
+    if azure_core_test::TestMode::current().unwrap() != azure_core_test::TestMode::Live {
+        println!("Skipping benchmarks. Set AZURE_TEST_MODE to run.");
+        return;
+    }
+
+    let rt = Runtime::new().unwrap();
+
+    let namespace = env::var("EVENTHUBS_HOST")
+        .expect("Set EVENTHUBS_HOST in env (e.g. my-namespace.servicebus.windows.net)");
+    let event_hub = env::var("EVENTHUB_NAME").expect("Set EVENTHUB_NAME in env");
+
+    let credential =
+        DefaultAzureCredential::new().expect("Failed to create DefaultAzureCredential");
+
+    let client = rt.block_on(async move {
+        Arc::new(
+            ProducerClient::builder()
+                .open(namespace.as_str(), event_hub.as_str(), credential)
+                .await
+                .expect("Failed to create ProducerClient"),
+        )
+    });
+
+    let num_events = 1000;
+    let payload = b"perf-test-payload";
+
+    let client = client.clone();
+    c.bench_function("eventhubs_send_batch", |b| {
+        let client = client.clone();
+        b.to_async(&rt).iter(move || {
+            let client = client.clone();
+            async move {
+                let batch = client.create_batch(None).await.unwrap();
+                for _ in 0..num_events {
+                    batch
+                        .try_add_event_data(EventData::from(payload.as_ref()), None)
+                        .unwrap();
+                }
+                let _ = client.send_batch(&batch, None).await;
+            }
+        });
+    });
+}
+
+criterion_group!(
+    name = send_batch_benchmarks;
+    config = Criterion::default().sample_size(100).warm_up_time(std::time::Duration::new(1, 0));
+    targets = send_batch_benchmark
+);
+
+fn send_benchmark(c: &mut Criterion) {
+    setup();
+
+    // Check if the environment variable is set thus allowing the benchmarks to run
+    if azure_core_test::TestMode::current().unwrap() != azure_core_test::TestMode::Live {
+        println!("Skipping benchmarks. Set AZURE_TEST_MODE to run.");
+        return;
+    }
+
+    let rt = Runtime::new().unwrap();
+
+    let namespace = env::var("EVENTHUBS_HOST")
+        .expect("Set EVENTHUBS_HOST in env (e.g. my-namespace.servicebus.windows.net)");
+    let event_hub = env::var("EVENTHUB_NAME").expect("Set EVENTHUB_NAME in env");
+
+    let credential =
+        DefaultAzureCredential::new().expect("Failed to create DefaultAzureCredential");
+
+    let client = rt.block_on(async move {
+        Arc::new(
+            ProducerClient::builder()
+                .open(namespace.as_str(), event_hub.as_str(), credential)
+                .await
+                .expect("Failed to create ProducerClient"),
+        )
+    });
+
+    c.bench_function("eventhubs_send", |b| {
+        let client = client.clone();
+        let mut index: u64 = 0;
+        b.to_async(&rt).iter(move || {
+            let client = client.clone();
+            async move {
+                index += 1;
+                let payload = b"a";
+                let event_data = EventData::builder()
+                    .with_body(payload)
+                    .add_property("Number".to_string(), index)
+                    .add_property("PartitionId".to_string(), "0")
+                    .build();
+                client
+                    .send_event(
+                        event_data,
+                        Some(SendEventOptions {
+                            partition_id: Some("0".to_string()),
+                        }),
+                    )
+                    .await
+            }
+        });
+    });
+}
+
+criterion_group!(
+    name = send_benchmarks;
+    config = Criterion::default().sample_size(1000).warm_up_time(std::time::Duration::new(1, 0)).measurement_time(std::time::Duration::new(205, 0));
+    targets =  send_benchmark
+);
+
+criterion_main!(send_batch_benchmarks, send_benchmarks);

sdk/eventhubs/azure_messaging_eventhubs/benches/benchmarks.rs

@@ -0,0 +1,17 @@
+Service: azure_messaging_eventhubs
+
+Project: azure_messaging_eventhubs
+
+PackageVersions:
+  - azure_messaging_eventhubs: source
+
+Tests:
+  - Test: eventhubs_send_batch
+    Class: eventhubs_send_batch
+    Arguments:
+      - --sync
+
+  - Test: eventhubs_send
+    Class: eventhubs_send
+    Arguments:
+      - --sync

sdk/eventhubs/azure_messaging_eventhubs/perf-tests.yml

@@ -0,0 +1,35 @@
+parameters:
+- name: PackageVersions
+  displayName: PackageVersions (regex of package versions to run)
+  type: string
+  default: 'source'
+- name: Tests
+  displayName: Tests (regex of tests to run)
+  type: string
+  default: '.*'
+- name: Arguments
+  displayName: Arguments (regex of arguments to run)
+  type: string
+  default: '.*'
+- name: Iterations
+  displayName: Iterations (times to run each test)
+  type: number
+  default: 1
+- name: Profile
+  type: boolean
+  default: false
+- name: AdditionalArguments
+  displayName: AdditionalArguments (passed to PerfAutomation)
+  type: string
+  default: ''
+
+extends:
+  template: /eng/pipelines/templates/jobs/perf.yml@self
+  parameters:
+    ServiceDirectory: eventhubs
+    PackageVersions: ${{ parameters.PackageVersions }}
+    Tests: ${{ parameters.Tests }}
+    Arguments: ${{ parameters.Arguments }}
+    Iterations: ${{ parameters.Iterations }}
+    AdditionalArguments: ${{ parameters.AdditionalArguments }}
+    Profile: ${{ parameters.Profile }}