feat(new sink): Add AppSignal sink

.github/actions/spelling/allow.txt

@@ -29,6 +29,9 @@ Apanda
 apikey
 apimachinery
 apiserver
+appsignal
+Appsignal
+APPSIGNAL
 archlinux
 Archos
 Arival

.github/semantic.yml

@@ -185,6 +185,7 @@ scopes:
 
   # sinks
   - amqp sink # Anything `amqp` sink related
+  - appsignal sink # Anything `appsignal` sink related
   - aws_cloudwatch_logs sink # Anything `aws_cloudwatch_logs` sink related
   - aws_cloudwatch_metrics sink # Anything `aws_cloudwatch_metrics` sink related
   - aws_kinesis_firehose sink # Anything `aws_kinesis_firehose` sink related

.github/workflows/integration-test.yml

@@ -57,6 +57,7 @@ jobs:
       matrix:
         include:
           - test: 'amqp'
+          - test: 'appsignal'
           - test: 'aws'
           - test: 'axiom'
           - test: 'azure'

Cargo.toml

@@ -594,6 +594,7 @@ transforms-throttle = ["dep:governor"]
 sinks = ["sinks-logs", "sinks-metrics"]
 sinks-logs = [
   "sinks-amqp",
+  "sinks-appsignal",
   "sinks-aws_cloudwatch_logs",
   "sinks-aws_kinesis_firehose",
   "sinks-aws_kinesis_streams",
@@ -635,6 +636,7 @@ sinks-logs = [
   "sinks-websocket",
 ]
 sinks-metrics = [
+  "sinks-appsignal",
   "sinks-aws_cloudwatch_metrics",
   "sinks-blackhole",
   "sinks-console",
@@ -650,6 +652,7 @@ sinks-metrics = [
 ]
 
 sinks-amqp = ["lapin"]
+sinks-appsignal = []
 sinks-aws_cloudwatch_logs = ["aws-core", "dep:aws-sdk-cloudwatchlogs"]
 sinks-aws_cloudwatch_metrics = ["aws-core", "dep:aws-sdk-cloudwatch"]
 sinks-aws_kinesis_firehose = ["aws-core", "dep:aws-sdk-firehose"]
@@ -714,6 +717,7 @@ nightly = []
 # Testing-related features
 all-integration-tests = [
   "amqp-integration-tests",
+  "appsignal-integration-tests",
   "aws-integration-tests",
   "axiom-integration-tests",
   "azure-integration-tests",
@@ -751,6 +755,7 @@ all-integration-tests = [
 ]
 
 amqp-integration-tests = ["sources-amqp", "sinks-amqp"]
+appsignal-integration-tests = ["sinks-appsignal"]
 
 aws-integration-tests = [
   "aws-cloudwatch-logs-integration-tests",

Makefile

@@ -330,7 +330,7 @@ test-behavior: test-behavior-transforms test-behavior-formats test-behavior-conf
 
 .PHONY: test-integration
 test-integration: ## Runs all integration tests
-test-integration: test-integration-amqp test-integration-aws test-integration-axiom test-integration-azure test-integration-chronicle test-integration-clickhouse
+test-integration: test-integration-amqp test-integration-appsignal test-integration-aws test-integration-axiom test-integration-azure test-integration-chronicle test-integration-clickhouse
 test-integration: test-integration-databend test-integration-docker-logs test-integration-elasticsearch
 test-integration: test-integration-eventstoredb test-integration-fluent test-integration-gcp test-integration-humio test-integration-http-client test-integration-influxdb
 test-integration: test-integration-kafka test-integration-logstash test-integration-loki test-integration-mongodb test-integration-nats

scripts/integration/appsignal/test.yaml

@@ -0,0 +1,11 @@
+features:
+- appsignal-integration-tests
+
+test_filter: '::appsignal::integration_tests::'
+
+runner:
+  env:
+    TEST_APPSIGNAL_PUSH_API_KEY:
+
+matrix:
+  version: [latest]

src/sinks/appsignal/integration_tests.rs

@@ -0,0 +1,93 @@
+use futures::stream;
+use indoc::indoc;
+use vector_core::event::{BatchNotifier, BatchStatus, Event, Metric, MetricKind, MetricValue};
+
+use crate::{
+    config::SinkConfig,
+    sinks::appsignal::AppsignalSinkConfig,
+    sinks::util::test::load_sink,
+    test_util::{
+        components::{
+            assert_sink_compliance, assert_sink_error, run_and_assert_sink_compliance,
+            COMPONENT_ERROR_TAGS, SINK_TAGS,
+        },
+        generate_lines_with_stream, map_event_batch_stream,
+    },
+};
+
+#[tokio::test]
+async fn logs_real_endpoint() {
+    let config = indoc! {r#"
+        push_api_key = "${TEST_APPSIGNAL_PUSH_API_KEY}"
+    "#};
+    let api_key = std::env::var("TEST_APPSIGNAL_PUSH_API_KEY")
+        .expect("couldn't find the AppSignal push API key in environment variables");
+    assert!(!api_key.is_empty(), "$TEST_APPSIGNAL_PUSH_API_KEY required");
+    let config = config.replace("${TEST_APPSIGNAL_PUSH_API_KEY}", &api_key);
+    let (config, cx) = load_sink::<AppsignalSinkConfig>(config.as_str()).unwrap();
+
+    let (sink, _) = config.build(cx).await.unwrap();
+    let (batch, receiver) = BatchNotifier::new_with_receiver();
+    let generator = |index| format!("this is a log with index {}", index);
+    let (_, events) = generate_lines_with_stream(generator, 10, Some(batch));
+
+    run_and_assert_sink_compliance(sink, events, &SINK_TAGS).await;
+
+    assert_eq!(receiver.await, BatchStatus::Delivered);
+}
+
+#[tokio::test]
+async fn metrics_real_endpoint() {
+    assert_sink_compliance(&SINK_TAGS, async {
+        let config = indoc! {r#"
+            push_api_key = "${TEST_APPSIGNAL_PUSH_API_KEY}"
+        "#};
+        let api_key = std::env::var("TEST_APPSIGNAL_PUSH_API_KEY")
+            .expect("couldn't find the AppSignal push API key in environment variables");
+        assert!(!api_key.is_empty(), "$TEST_APPSIGNAL_PUSH_API_KEY required");
+        let config = config.replace("${TEST_APPSIGNAL_PUSH_API_KEY}", &api_key);
+        let (config, cx) = load_sink::<AppsignalSinkConfig>(config.as_str()).unwrap();
+
+        let (sink, _) = config.build(cx).await.unwrap();
+        let (batch, receiver) = BatchNotifier::new_with_receiver();
+        let events: Vec<_> = (0..10)
+            .map(|index| {
+                Event::Metric(Metric::new(
+                    "counter",
+                    MetricKind::Absolute,
+                    MetricValue::Counter {
+                        value: index as f64,
+                    },
+                ))
+            })
+            .collect();
+        let stream = map_event_batch_stream(stream::iter(events.clone()), Some(batch));
+
+        sink.run(stream).await.unwrap();
+        assert_eq!(receiver.await, BatchStatus::Delivered);
+    })
+    .await;
+}
+
+#[tokio::test]
+async fn error_scenario_real_endpoint() {
+    assert_sink_error(&COMPONENT_ERROR_TAGS, async {
+        let config = indoc! {r#"
+            push_api_key = "invalid key"
+        "#};
+        let (config, cx) = load_sink::<AppsignalSinkConfig>(config).unwrap();
+
+        let (sink, _) = config.build(cx).await.unwrap();
+        let (batch, receiver) = BatchNotifier::new_with_receiver();
+        let events = vec![Event::Metric(Metric::new(
+            "counter",
+            MetricKind::Absolute,
+            MetricValue::Counter { value: 1.0 },
+        ))];
+        let stream = map_event_batch_stream(stream::iter(events.clone()), Some(batch));
+
+        sink.run(stream).await.unwrap();
+        assert_eq!(receiver.await, BatchStatus::Rejected);
+    })
+    .await;
+}