enhancement(pulsar sink): Refactor to use StreamSink (#14345)

This commit heavily refactors the Pulsar Sink to use the StreamSink
interface and is modeled after the Kafka Sink. It also adds additional
features that bring it in line with Kafka Sink feature set.

This includes:
* Refactoring to use StreamSink instead of Sink interface.
* Supports dynamic topics using a topic template
* Refactor configurations in advance of adding Pulsar source
* Rework message parsing to support logs and metrics, with support for
  dynamic keys and properties

Co-authored-by: Spencer Gilbert <[email protected]>
Co-authored-by: Nathan Fox <[email protected]>
Co-authored-by: neuronull <[email protected]>

Author: 4 people

Cargo.toml

@@ -688,7 +688,7 @@ sinks-new_relic_logs = ["sinks-http"]
 sinks-new_relic = []
 sinks-papertrail = ["dep:syslog"]
 sinks-prometheus = ["aws-core", "dep:base64", "dep:prometheus-parser", "dep:snap"]
-sinks-pulsar = ["dep:apache-avro", "dep:pulsar"]
+sinks-pulsar = ["dep:apache-avro", "dep:pulsar", "dep:lru"]
 sinks-redis = ["dep:redis"]
 sinks-sematext = ["sinks-elasticsearch", "sinks-influxdb"]
 sinks-socket = ["sinks-utils-udp"]

scripts/integration/pulsar/test.yaml

@@ -1,7 +1,7 @@
 features:
 - pulsar-integration-tests
 
-test_filter: '::pulsar::'
+test_filter: '::pulsar::integration_tests::'
 
 env:
   PULSAR_ADDRESS: pulsar://pulsar:6650

src/internal_events/pulsar.rs

@@ -33,3 +33,26 @@ impl InternalEvent for PulsarSendingError {
         });
     }
 }
+
+pub struct PulsarPropertyExtractionError<F: std::fmt::Display> {
+    pub property_field: F,
+}
+
+impl<F: std::fmt::Display> InternalEvent for PulsarPropertyExtractionError<F> {
+    fn emit(self) {
+        error!(
+            message = "Failed to extract properties. Value should be a map of String -> Bytes.",
+            error_code = "extracting_property",
+            error_type = error_type::PARSER_FAILED,
+            stage = error_stage::PROCESSING,
+            property_field = %self.property_field,
+            internal_log_rate_limit = true,
+        );
+        counter!(
+            "component_errors_total", 1,
+            "error_code" => "extracting_property",
+            "error_type" => error_type::PARSER_FAILED,
+            "stage" => error_stage::PROCESSING,
+        );
+    }
+}

src/sinks/mod.rs

@@ -357,7 +357,7 @@ pub enum Sinks {
     /// Publish observability events to Apache Pulsar topics.
     #[cfg(feature = "sinks-pulsar")]
     #[configurable(metadata(docs::label = "Pulsar"))]
-    Pulsar(pulsar::PulsarSinkConfig),
+    Pulsar(pulsar::config::PulsarSinkConfig),
 
     /// Publish observability data to Redis.
     #[cfg(feature = "sinks-redis")]