mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
test(e2e): add S3 event-notification webhook delivery regression net (#4821)
Cover the previously untested configure-target -> put-notification-config -> object operation -> webhook delivery chain (backlog#1154 peri-1): PUT/multipart-complete/DELETE event fields, prefix/suffix filter negatives, and store-queue redelivery after target recovery. Wire both tests into the e2e-smoke nextest profile (ci-4 mechanism).
This commit is contained in:
Zhengchao An
@@ -191,6 +191,12 @@ mod stale_multipart_cleanup_cluster_test;
|
||||
#[cfg(test)]
|
||||
mod object_lambda_test;
|
||||
|
||||
// S3 event-notification webhook delivery end-to-end (backlog#1154 peri-1):
|
||||
// configure webhook target -> PutBucketNotificationConfiguration -> object
|
||||
// operation -> event delivered, plus filter negatives and store-queue redelivery.
|
||||
#[cfg(test)]
|
||||
mod notification_webhook_test;
|
||||
|
||||
// Replication extension end-to-end regression tests
|
||||
#[cfg(test)]
|
||||
mod replication_extension_test;
|
||||
|
||||
@@ -0,0 +1,576 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! End-to-end regression net for the S3 event-notification pipeline
|
||||
//! (backlog#1154 peri-1). Proves the full "configure webhook target ->
|
||||
//! PutBucketNotificationConfiguration -> object operation -> event delivered"
|
||||
//! chain against a real rustfs binary and a real HTTP receiver, which no other
|
||||
//! test covers: the target-plugin suites stop at broker integration and the
|
||||
//! object-lambda suite only exercises the webhook target as a transform
|
||||
//! function, never as an S3 event sink.
|
||||
//!
|
||||
//! Coverage:
|
||||
//! * PUT / multipart-complete / DELETE each deliver one event with the correct
|
||||
//! eventName, bucket, key, versionId and eTag.
|
||||
//! * prefix/suffix filters drop non-matching keys (rule-engine gate).
|
||||
//! * an event queued while the target endpoint is unreachable is redelivered
|
||||
//! from the on-disk store once the endpoint recovers (store-and-forward).
|
||||
|
||||
use crate::common::{RustFSTestEnvironment, init_logging};
|
||||
use aws_sdk_s3::Client;
|
||||
use aws_sdk_s3::primitives::ByteStream;
|
||||
use aws_sdk_s3::types::{
|
||||
BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, Event, FilterRule, FilterRuleName,
|
||||
NotificationConfiguration, NotificationConfigurationFilter, QueueConfiguration, S3KeyFilter, VersioningConfiguration,
|
||||
};
|
||||
use http::header::{CONTENT_TYPE, HOST};
|
||||
use local_ip_address::local_ip;
|
||||
use reqwest::StatusCode;
|
||||
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
|
||||
use rustfs_signer::sign_v4;
|
||||
use s3s::Body;
|
||||
use serde_json::Value;
|
||||
use serial_test::serial;
|
||||
use std::error::Error;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::task::JoinHandle;
|
||||
use tokio::time::{Duration, Instant, timeout};
|
||||
|
||||
type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
|
||||
type BoxError = Box<dyn Error + Send + Sync>;
|
||||
|
||||
/// Region embedded in the target ARN. Only the `id:name` tail of the ARN is used
|
||||
/// for target lookup (see `process_queue_configurations`), so the region is
|
||||
/// nominal, but it must be present for `ARN::parse` to succeed.
|
||||
const NOTIFY_REGION: &str = "us-east-1";
|
||||
|
||||
/// Webhook targets are registered as `TargetID { id: <name>, name: "webhook" }`,
|
||||
/// so the ARN a notification rule references is
|
||||
/// `arn:rustfs:sqs:<region>:<name>:webhook`.
|
||||
fn target_arn(target_name: &str) -> String {
|
||||
format!("arn:rustfs:sqs:{NOTIFY_REGION}:{target_name}:webhook")
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// In-test HTTP event receiver
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Reads one HTTP request off the stream, returning its method and body. Handles
|
||||
/// the target's HEAD reachability probe (no body) and POST event deliveries.
|
||||
async fn read_http_message(stream: &mut tokio::net::TcpStream) -> Result<(String, Vec<u8>), BoxError> {
|
||||
let mut buffer = Vec::new();
|
||||
let mut chunk = [0_u8; 4096];
|
||||
|
||||
let header_end = loop {
|
||||
let read = stream.read(&mut chunk).await?;
|
||||
if read == 0 {
|
||||
return Err("connection closed before request headers were complete".into());
|
||||
}
|
||||
buffer.extend_from_slice(&chunk[..read]);
|
||||
if let Some(pos) = buffer.windows(4).position(|w| w == b"\r\n\r\n") {
|
||||
break pos;
|
||||
}
|
||||
};
|
||||
|
||||
let header_text = std::str::from_utf8(&buffer[..header_end])?;
|
||||
let mut lines = header_text.split("\r\n");
|
||||
let request_line = lines.next().ok_or("missing request line")?;
|
||||
let method = request_line.split_whitespace().next().ok_or("missing method")?.to_string();
|
||||
|
||||
let mut content_length = 0usize;
|
||||
for line in lines {
|
||||
if let Some((name, value)) = line.split_once(':')
|
||||
&& name.trim().eq_ignore_ascii_case("content-length")
|
||||
{
|
||||
content_length = value.trim().parse::<usize>().unwrap_or(0);
|
||||
}
|
||||
}
|
||||
|
||||
let body_offset = header_end + 4;
|
||||
while buffer.len().saturating_sub(body_offset) < content_length {
|
||||
let read = stream.read(&mut chunk).await?;
|
||||
if read == 0 {
|
||||
return Err("connection closed before request body was complete".into());
|
||||
}
|
||||
buffer.extend_from_slice(&chunk[..read]);
|
||||
}
|
||||
|
||||
Ok((method, buffer[body_offset..body_offset + content_length].to_vec()))
|
||||
}
|
||||
|
||||
/// Drives an accepted listener: answers every request 200 OK (so the target's
|
||||
/// HEAD reachability probe reports it online) and forwards each non-empty POST
|
||||
/// body, parsed as the S3 event envelope JSON, to `tx`.
|
||||
fn serve_event_collector(listener: TcpListener, tx: mpsc::UnboundedSender<Value>) -> JoinHandle<()> {
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let Ok((mut stream, _)) = listener.accept().await else {
|
||||
return;
|
||||
};
|
||||
let tx = tx.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Ok(Ok((method, body))) = timeout(Duration::from_secs(5), read_http_message(&mut stream)).await {
|
||||
let _ = stream
|
||||
.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
|
||||
.await;
|
||||
let _ = stream.shutdown().await;
|
||||
if method == "POST"
|
||||
&& !body.is_empty()
|
||||
&& let Ok(value) = serde_json::from_slice::<Value>(&body)
|
||||
{
|
||||
let _ = tx.send(value);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Binds a fresh collector on a random port. Returns its `/events`
|
||||
/// endpoint URL, a receiver of parsed event envelopes, and the serving task.
|
||||
async fn spawn_event_collector() -> Result<(String, mpsc::UnboundedReceiver<Value>, JoinHandle<()>), BoxError> {
|
||||
let listener = TcpListener::bind("0.0.0.0:0").await?;
|
||||
let port = listener.local_addr()?.port();
|
||||
let endpoint_ip = local_ip()?;
|
||||
let (tx, rx) = mpsc::unbounded_channel();
|
||||
let handle = serve_event_collector(listener, tx);
|
||||
Ok((format!("http://{endpoint_ip}.nip.io:{port}/events"), rx, handle))
|
||||
}
|
||||
|
||||
/// Decoded object key of the first record in an event envelope.
|
||||
fn event_key(envelope: &Value) -> Option<String> {
|
||||
let raw = envelope["Records"][0]["s3"]["object"]["key"].as_str()?;
|
||||
Some(
|
||||
urlencoding::decode(raw)
|
||||
.map(|d| d.into_owned())
|
||||
.unwrap_or_else(|_| raw.to_string()),
|
||||
)
|
||||
}
|
||||
|
||||
/// Waits until an event targeting `key` whose `EventName` starts with
|
||||
/// `event_prefix` arrives, returning the full envelope. Other envelopes are
|
||||
/// discarded, so a lingering duplicate of an earlier event (delivery is
|
||||
/// at-least-once) or the create event of a key that is later deleted never
|
||||
/// satisfies a wait for the opposite event type.
|
||||
async fn wait_for_event(
|
||||
rx: &mut mpsc::UnboundedReceiver<Value>,
|
||||
key: &str,
|
||||
event_prefix: &str,
|
||||
within: Duration,
|
||||
) -> Result<Value, BoxError> {
|
||||
let deadline = Instant::now() + within;
|
||||
loop {
|
||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||
if remaining.is_zero() {
|
||||
return Err(format!("no {event_prefix}* event for key {key} within {within:?}").into());
|
||||
}
|
||||
match timeout(remaining, rx.recv()).await {
|
||||
Ok(Some(envelope)) => {
|
||||
let key_matches = event_key(&envelope).as_deref() == Some(key);
|
||||
let name_matches = envelope["EventName"].as_str().is_some_and(|n| n.starts_with(event_prefix));
|
||||
if key_matches && name_matches {
|
||||
return Ok(envelope);
|
||||
}
|
||||
}
|
||||
Ok(None) => return Err("event collector channel closed".into()),
|
||||
Err(_) => return Err(format!("no {event_prefix}* event for key {key} within {within:?}").into()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Collects every event until `stop_key` is seen (plus a short grace window to
|
||||
/// catch stragglers), or until `max_wait` elapses. Used to prove a negative:
|
||||
/// non-matching keys must never appear even though a matching sentinel does.
|
||||
async fn collect_until(
|
||||
rx: &mut mpsc::UnboundedReceiver<Value>,
|
||||
stop_key: &str,
|
||||
max_wait: Duration,
|
||||
grace: Duration,
|
||||
) -> Vec<Value> {
|
||||
let hard_deadline = Instant::now() + max_wait;
|
||||
let mut soft_deadline: Option<Instant> = None;
|
||||
let mut collected = Vec::new();
|
||||
loop {
|
||||
let deadline = soft_deadline.unwrap_or(hard_deadline).min(hard_deadline);
|
||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||
if remaining.is_zero() {
|
||||
break;
|
||||
}
|
||||
match timeout(remaining, rx.recv()).await {
|
||||
Ok(Some(envelope)) => {
|
||||
let matched = event_key(&envelope).as_deref() == Some(stop_key);
|
||||
collected.push(envelope);
|
||||
if matched && soft_deadline.is_none() {
|
||||
soft_deadline = Some(Instant::now() + grace);
|
||||
}
|
||||
}
|
||||
Ok(None) => break,
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
collected
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Admin target configuration (signed admin HTTP)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
async fn signed_admin_request(
|
||||
env: &RustFSTestEnvironment,
|
||||
method: http::Method,
|
||||
url: &str,
|
||||
body: Option<Vec<u8>>,
|
||||
) -> Result<reqwest::Response, BoxError> {
|
||||
let uri = url.parse::<http::Uri>()?;
|
||||
let authority = uri.authority().ok_or("admin URL missing authority")?.to_string();
|
||||
let mut builder = http::Request::builder()
|
||||
.method(method.clone())
|
||||
.uri(uri)
|
||||
.header(HOST, authority)
|
||||
.header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
|
||||
if body.is_some() {
|
||||
builder = builder.header(CONTENT_TYPE, "application/json");
|
||||
}
|
||||
|
||||
let content_len = body.as_ref().map(|b| b.len() as i64).unwrap_or_default();
|
||||
let signed = sign_v4(
|
||||
builder.body(Body::empty())?,
|
||||
content_len,
|
||||
&env.access_key,
|
||||
&env.secret_key,
|
||||
"",
|
||||
"us-east-1",
|
||||
);
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?;
|
||||
let mut request = crate::common::local_http_client().request(reqwest_method, url);
|
||||
for (name, value) in signed.headers() {
|
||||
request = request.header(name, value);
|
||||
}
|
||||
if let Some(body) = body {
|
||||
request = request.body(body);
|
||||
}
|
||||
Ok(request.send().await?)
|
||||
}
|
||||
|
||||
async fn enable_notify_module(env: &RustFSTestEnvironment) -> TestResult {
|
||||
let payload = serde_json::json!({
|
||||
"notify_enabled": true,
|
||||
"audit_enabled": false,
|
||||
});
|
||||
let url = format!("{}/rustfs/admin/v3/module-switches", env.url);
|
||||
let response = signed_admin_request(env, http::Method::PUT, &url, Some(payload.to_string().into_bytes())).await?;
|
||||
if response.status() != StatusCode::OK {
|
||||
let status = response.status();
|
||||
let text = response.text().await.unwrap_or_default();
|
||||
return Err(format!("failed to enable notify module: {status} {text}").into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Registers a webhook notification target with a persistent queue directory, so
|
||||
/// delivery goes through the durable store-and-forward path.
|
||||
async fn configure_webhook_target(env: &RustFSTestEnvironment, target_name: &str, endpoint: &str) -> TestResult {
|
||||
let queue_dir = format!("{}/notify-queue-{target_name}", env.temp_dir);
|
||||
tokio::fs::create_dir_all(&queue_dir).await?;
|
||||
let payload = serde_json::json!({
|
||||
"key_values": [
|
||||
{ "key": "endpoint", "value": endpoint },
|
||||
{ "key": "queue_dir", "value": queue_dir },
|
||||
]
|
||||
});
|
||||
let url = format!("{}/rustfs/admin/v3/target/notify_webhook/{target_name}", env.url);
|
||||
let response = signed_admin_request(env, http::Method::PUT, &url, Some(payload.to_string().into_bytes())).await?;
|
||||
let status = response.status();
|
||||
if status != StatusCode::OK {
|
||||
let text = response.text().await.unwrap_or_default();
|
||||
return Err(format!("failed to configure webhook target {target_name}: {status} {text}").into());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Polls the admin target ARN list until the freshly configured target is
|
||||
/// registered in the runtime, so notification rules and object events do not
|
||||
/// race target activation.
|
||||
async fn wait_for_target_registered(env: &RustFSTestEnvironment, target_name: &str) -> TestResult {
|
||||
let suffix = format!(":{target_name}:webhook");
|
||||
let url = format!("{}/rustfs/admin/v3/target/arns", env.url);
|
||||
for _ in 0..40 {
|
||||
let response = signed_admin_request(env, http::Method::GET, &url, None).await?;
|
||||
if response.status() == StatusCode::OK {
|
||||
let arns: Vec<String> = serde_json::from_slice(&response.bytes().await?)?;
|
||||
if arns.iter().any(|arn| arn.ends_with(&suffix)) {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
Err(format!("target {target_name} was not registered in admin ARNs").into())
|
||||
}
|
||||
|
||||
/// Binds a bucket to a webhook target for ObjectCreated:*/ObjectRemoved:* events,
|
||||
/// filtered to `prefix` + `suffix`.
|
||||
async fn put_notification_config(client: &Client, bucket: &str, target_name: &str, prefix: &str, suffix: &str) -> TestResult {
|
||||
let key_filter = S3KeyFilter::builder()
|
||||
.filter_rules(FilterRule::builder().name(FilterRuleName::Prefix).value(prefix).build())
|
||||
.filter_rules(FilterRule::builder().name(FilterRuleName::Suffix).value(suffix).build())
|
||||
.build();
|
||||
let queue = QueueConfiguration::builder()
|
||||
.id(format!("{target_name}-rule"))
|
||||
.queue_arn(target_arn(target_name))
|
||||
.events(Event::from("s3:ObjectCreated:*"))
|
||||
.events(Event::from("s3:ObjectRemoved:*"))
|
||||
.filter(NotificationConfigurationFilter::builder().key(key_filter).build())
|
||||
.build()?;
|
||||
let config = NotificationConfiguration::builder().queue_configurations(queue).build();
|
||||
client
|
||||
.put_bucket_notification_configuration()
|
||||
.bucket(bucket)
|
||||
.notification_configuration(config)
|
||||
.send()
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn enable_versioning(client: &Client, bucket: &str) -> TestResult {
|
||||
client
|
||||
.put_bucket_versioning()
|
||||
.bucket(bucket)
|
||||
.versioning_configuration(
|
||||
VersioningConfiguration::builder()
|
||||
.status(BucketVersioningStatus::Enabled)
|
||||
.build(),
|
||||
)
|
||||
.send()
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn trimmed_etag(value: Option<&str>) -> Option<String> {
|
||||
value.map(|e| e.trim_matches('"').to_string())
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// PUT / multipart-complete / DELETE each deliver one event with correct fields,
|
||||
/// and the prefix/suffix filter drops non-matching keys.
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_webhook_event_delivery_and_filtering() -> TestResult {
|
||||
init_logging();
|
||||
|
||||
let (endpoint, mut rx, handle) = spawn_event_collector().await?;
|
||||
|
||||
let mut env = RustFSTestEnvironment::new().await?;
|
||||
env.start_rustfs_server(vec![]).await?;
|
||||
enable_notify_module(&env).await?;
|
||||
|
||||
let bucket = "peri1-events";
|
||||
let target = "peri1events";
|
||||
let client = env.create_s3_client();
|
||||
client.create_bucket().bucket(bucket).send().await?;
|
||||
enable_versioning(&client, bucket).await?;
|
||||
|
||||
configure_webhook_target(&env, target, &endpoint).await?;
|
||||
wait_for_target_registered(&env, target).await?;
|
||||
put_notification_config(&client, bucket, target, "uploads/", ".dat").await?;
|
||||
|
||||
// --- PUT: ObjectCreated:Put with exact eTag + versionId ------------------
|
||||
let put_key = "uploads/report.dat";
|
||||
let put = client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(put_key)
|
||||
.body(ByteStream::from_static(b"peri-1 put body"))
|
||||
.send()
|
||||
.await?;
|
||||
let put_version = put
|
||||
.version_id()
|
||||
.ok_or("PUT response missing versionId (versioning not enabled?)")?;
|
||||
|
||||
let created = wait_for_event(&mut rx, put_key, "s3:ObjectCreated:", Duration::from_secs(20)).await?;
|
||||
let record = &created["Records"][0];
|
||||
let object = &record["s3"]["object"];
|
||||
assert_eq!(
|
||||
created["EventName"].as_str(),
|
||||
Some("s3:ObjectCreated:Put"),
|
||||
"envelope EventName: {created}"
|
||||
);
|
||||
assert!(
|
||||
record["eventName"]
|
||||
.as_str()
|
||||
.is_some_and(|n| n.starts_with("s3:ObjectCreated:")),
|
||||
"record eventName: {record}"
|
||||
);
|
||||
assert_eq!(record["s3"]["bucket"]["name"].as_str(), Some(bucket), "bucket in event: {record}");
|
||||
assert_eq!(object["versionId"].as_str(), Some(put_version), "versionId in event: {object}");
|
||||
assert_eq!(
|
||||
trimmed_etag(object["eTag"].as_str()),
|
||||
trimmed_etag(put.e_tag()),
|
||||
"eTag in event: {object}"
|
||||
);
|
||||
|
||||
// --- Multipart complete: ObjectCreated:CompleteMultipartUpload -----------
|
||||
let mp_key = "uploads/multi.dat";
|
||||
let created_mp = client.create_multipart_upload().bucket(bucket).key(mp_key).send().await?;
|
||||
let upload_id = created_mp.upload_id().ok_or("missing upload id")?;
|
||||
let part = client
|
||||
.upload_part()
|
||||
.bucket(bucket)
|
||||
.key(mp_key)
|
||||
.upload_id(upload_id)
|
||||
.part_number(1)
|
||||
.body(ByteStream::from_static(b"peri-1 multipart part body"))
|
||||
.send()
|
||||
.await?;
|
||||
let complete = client
|
||||
.complete_multipart_upload()
|
||||
.bucket(bucket)
|
||||
.key(mp_key)
|
||||
.upload_id(upload_id)
|
||||
.multipart_upload(
|
||||
CompletedMultipartUpload::builder()
|
||||
.parts(
|
||||
CompletedPart::builder()
|
||||
.part_number(1)
|
||||
.e_tag(part.e_tag().unwrap_or_default())
|
||||
.build(),
|
||||
)
|
||||
.build(),
|
||||
)
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
let mp_event = wait_for_event(&mut rx, mp_key, "s3:ObjectCreated:", Duration::from_secs(20)).await?;
|
||||
let mp_record = &mp_event["Records"][0];
|
||||
assert_eq!(
|
||||
mp_event["EventName"].as_str(),
|
||||
Some("s3:ObjectCreated:CompleteMultipartUpload"),
|
||||
"multipart EventName: {mp_event}"
|
||||
);
|
||||
assert_eq!(mp_record["s3"]["bucket"]["name"].as_str(), Some(bucket));
|
||||
assert_eq!(
|
||||
trimmed_etag(mp_record["s3"]["object"]["eTag"].as_str()),
|
||||
trimmed_etag(complete.e_tag()),
|
||||
"multipart eTag in event: {mp_record}"
|
||||
);
|
||||
|
||||
// --- Filter: non-matching prefix and suffix must never be delivered ------
|
||||
let wrong_prefix = "logs/report.dat"; // right suffix, wrong prefix
|
||||
let wrong_suffix = "uploads/report.txt"; // right prefix, wrong suffix
|
||||
let sentinel = "uploads/keep.dat";
|
||||
for key in [wrong_prefix, wrong_suffix, sentinel] {
|
||||
client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.body(ByteStream::from_static(b"filter probe"))
|
||||
.send()
|
||||
.await?;
|
||||
}
|
||||
let collected = collect_until(&mut rx, sentinel, Duration::from_secs(20), Duration::from_secs(2)).await;
|
||||
let seen: Vec<String> = collected.iter().filter_map(event_key).collect();
|
||||
assert!(
|
||||
seen.iter().any(|k| k == sentinel),
|
||||
"matching sentinel {sentinel} was not delivered; saw {seen:?}"
|
||||
);
|
||||
assert!(
|
||||
!seen.iter().any(|k| k == wrong_prefix),
|
||||
"wrong-prefix key {wrong_prefix} bypassed the filter; saw {seen:?}"
|
||||
);
|
||||
assert!(
|
||||
!seen.iter().any(|k| k == wrong_suffix),
|
||||
"wrong-suffix key {wrong_suffix} bypassed the filter; saw {seen:?}"
|
||||
);
|
||||
|
||||
// --- DELETE on a versioned bucket: ObjectRemoved:* with delete-marker version
|
||||
let delete = client.delete_object().bucket(bucket).key(put_key).send().await?;
|
||||
let removed = wait_for_event(&mut rx, put_key, "s3:ObjectRemoved:", Duration::from_secs(20)).await?;
|
||||
let removed_record = &removed["Records"][0];
|
||||
assert!(
|
||||
removed["EventName"]
|
||||
.as_str()
|
||||
.is_some_and(|n| n.starts_with("s3:ObjectRemoved:")),
|
||||
"delete EventName: {removed}"
|
||||
);
|
||||
if let Some(marker_version) = delete.version_id() {
|
||||
assert_eq!(
|
||||
removed_record["s3"]["object"]["versionId"].as_str(),
|
||||
Some(marker_version),
|
||||
"delete-marker versionId in event: {removed_record}"
|
||||
);
|
||||
}
|
||||
|
||||
env.stop_server();
|
||||
handle.abort();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// An event queued while the target endpoint is unreachable survives on the
|
||||
/// durable store and is redelivered once the endpoint comes back.
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_webhook_redelivers_event_after_target_recovers() -> TestResult {
|
||||
init_logging();
|
||||
|
||||
let mut env = RustFSTestEnvironment::new().await?;
|
||||
env.start_rustfs_server(vec![]).await?;
|
||||
enable_notify_module(&env).await?;
|
||||
|
||||
let bucket = "peri1-redeliver";
|
||||
let target = "peri1redeliver";
|
||||
let client = env.create_s3_client();
|
||||
client.create_bucket().bucket(bucket).send().await?;
|
||||
|
||||
// Reserve a port but leave it unbound: the webhook endpoint is
|
||||
// unreachable (connection refused -> retryable NotConnected), so the first
|
||||
// delivery attempts fail and the event is persisted to the queue store.
|
||||
let port = RustFSTestEnvironment::find_available_port().await?;
|
||||
let endpoint_ip = local_ip()?;
|
||||
let endpoint = format!("http://{endpoint_ip}.nip.io:{port}/events");
|
||||
configure_webhook_target(&env, target, &endpoint).await?;
|
||||
wait_for_target_registered(&env, target).await?;
|
||||
put_notification_config(&client, bucket, target, "uploads/", ".dat").await?;
|
||||
|
||||
let key = "uploads/redeliver.dat";
|
||||
client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.body(ByteStream::from_static(b"queued while target down"))
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
// Bring the endpoint up on the reserved port; the replay worker retries with
|
||||
// exponential backoff and delivers the queued event.
|
||||
let listener = TcpListener::bind(("0.0.0.0", port)).await?;
|
||||
let (tx, mut rx) = mpsc::unbounded_channel();
|
||||
let handle = serve_event_collector(listener, tx);
|
||||
|
||||
let redelivered = wait_for_event(&mut rx, key, "s3:ObjectCreated:", Duration::from_secs(45)).await?;
|
||||
assert_eq!(
|
||||
redelivered["EventName"].as_str(),
|
||||
Some("s3:ObjectCreated:Put"),
|
||||
"redelivered EventName: {redelivered}"
|
||||
);
|
||||
assert_eq!(redelivered["Records"][0]["s3"]["bucket"]["name"].as_str(), Some(bucket));
|
||||
|
||||
env.stop_server();
|
||||
handle.abort();
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user