This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kamelets.git
The following commit(s) were added to refs/heads/main by this push:
new ae02160d8 Fix #2303: surface the Ceph extension fields as headers
(#3030)
ae02160d8 is described below
commit ae02160d87e163a427a29f27d155224f26d733f9
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Sep 14 17:47:15 2026 +0200
Fix #2303: surface the Ceph extension fields as headers (#3030)
Re-lands work that missed the merge of #3026: that PR was merged at its
first
commit, so the header change never reached main.
Follows @yuvalif's guidance on which extensions matter: metadata and tags
because Ceph can filter notifications on them and they are useful feeding a
catalog or database, eventId for spotting duplicates after a Kafka
retransmit,
and opaqueData because users put their own information there. s3.bucket.id
is
left out -- an internal id useful only for debugging.
Named the way this catalog names source output headers, a plain form plus a
CloudEvents one, following infinispan-source and ftp-source.
Verified against a notification captured from a real Ceph deployment, which
@yuvalif posted on the pull request. All four headers populate, and metadata
and tags arrive as structured lists rather than flattened strings. That
capture
also settled the question the change originally left open: eventId is
populated
in practice, <timestamp>.<counter>.<etag> and unique per event, so it does
work
for deduplication. The empty eventId in the Ceph documentation is a
placeholder
in the sample rather than what RGW sends, and the partial documents the real
format instead of warning about an empty one.
A Ceph notification always carries these fields, empty when nothing is set,
so
the headers are always populated. A payload missing them is not a Ceph
event and
fails the same way any foreign message on the topic does, which the partial
states.
Claude-Session: https://claude.ai/code/session_01TvXfz2pbDbHPz33TcpxhPm
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../ceph-event-based-source-description.adoc | 30 ++++++++++++++++++++++
kamelets/ceph-event-based-source.kamelet.yaml | 29 +++++++++++++++++++++
2 files changed, 59 insertions(+)
diff --git
a/docs/modules/ROOT/partials/ceph-event-based-source-description.adoc
b/docs/modules/ROOT/partials/ceph-event-based-source-description.adoc
index 7a45743da..3dc18c944 100644
--- a/docs/modules/ROOT/partials/ceph-event-based-source-description.adoc
+++ b/docs/modules/ROOT/partials/ceph-event-based-source-description.adoc
@@ -16,3 +16,33 @@ JSON is emitted unchanged otherwise.
When `getObject` is enabled, `cephUrl`, `accessKey`, `secretKey` and
`zoneGroup` are
required so the object can be read back from the RGW endpoint.
+
+=== Ceph Extension Fields
+
+The fields Ceph adds beyond the S3 notification specification are surfaced as
headers, so a
+consumer does not have to re-parse the body for them. Each has a `ce-`
prefixed CloudEvents
+counterpart as well.
+
+- `ceph-event-id` from `eventId` - identifies the event, for spotting
duplicates after a
+ transport retransmit. RGW populates it as `<timestamp>.<counter>.<etag>`,
unique per
+ event. The event structure in the Ceph documentation shows it empty, but
that is a
+ placeholder in the sample rather than what RGW emits.
+- `ceph-opaque-data` from `opaqueData` - free-form information attached to the
topic by the
+ user.
+- `ceph-object-metadata` from `s3.object.metadata` - user attributes on the
object, sent as
+ `x-amz-meta-` headers. Ceph can also filter notifications on these.
+- `ceph-object-tags` from `s3.object.tags` - object tags. Both this and the
metadata arrive
+ as a list of key/val entries rather than a flattened string.
+
+`s3.bucket.id` is not surfaced. It is an internal Ceph identifier, useful for
debugging but
+not for routing.
+
+=== Expected Payload
+
+The topic is expected to carry only Ceph notifications. A message without a
`Records` array
+fails the exchange rather than being passed through, so that a foreign message
is not
+mistaken for an event with empty Ceph headers.
+
+A Ceph notification always carries the extension fields, empty when nothing is
set, so the
+headers above are always populated. A payload missing them is not a Ceph event
and fails the
+same way.
diff --git a/kamelets/ceph-event-based-source.kamelet.yaml
b/kamelets/ceph-event-based-source.kamelet.yaml
index 3db5f2067..f4e65acd7 100644
--- a/kamelets/ceph-event-based-source.kamelet.yaml
+++ b/kamelets/ceph-event-based-source.kamelet.yaml
@@ -122,6 +122,35 @@ spec:
saslUsername: "{{?saslUsername}}"
saslPassword: "{{?saslPassword}}"
steps:
+ # Ceph extension fields, surfaced as headers so a consumer does not
have
+ # to re-parse the body for them. Named the way the catalog names source
+ # output headers -- a plain form plus a CloudEvents one, as
+ # infinispan-source and ftp-source do. s3.bucket.id is deliberately not
+ # surfaced: it is an internal Ceph id useful only for debugging.
+ - setHeader:
+ name: ceph-event-id
+ jsonpath: $.Records[0].eventId
+ - setHeader:
+ name: ce-cepheventid
+ simple: "${header[ceph-event-id]}"
+ - setHeader:
+ name: ceph-opaque-data
+ jsonpath: $.Records[0].opaqueData
+ - setHeader:
+ name: ce-cephopaquedata
+ simple: "${header[ceph-opaque-data]}"
+ - setHeader:
+ name: ceph-object-metadata
+ jsonpath: $.Records[0].s3.object.metadata
+ - setHeader:
+ name: ce-cephobjectmetadata
+ simple: "${header[ceph-object-metadata]}"
+ - setHeader:
+ name: ceph-object-tags
+ jsonpath: $.Records[0].s3.object.tags
+ - setHeader:
+ name: ce-cephobjecttags
+ simple: "${header[ceph-object-tags]}"
# Ceph emits the classic S3 notification envelope -- a Records array --
# not the EventBridge shape that aws-s3-event-based-source parses, so
the
# paths below are Records based on purpose.
