存储桶通知提供了一种从 radosgw 发送信息的机制 当存储桶上发生某些事件时。通知可以发送到 HTTP 端点、AMQP0.9.1 端点和 Kafka 端点。
用户可以创建主题。主题实体由其名称定义,并且是“每 租户”。用户只能关联其主题(通过通知配置) 它拥有的水桶。
必须创建通知实体才能发送事件通知 特定存储桶。可以为子集创建通知实体 事件类型或所有事件类型(默认值)。这 通知还可以根据前缀的匹配项过滤掉事件,并且 (1) 键的后缀,(2) 附加到对象的元数据属性, 或 (3) 对象标签。正则表达式匹配也可用于这些 以创建筛选器。任何特定主题都可以有多个通知, 并且同一主题可用于多个通知。
已定义 REST API,以便提供配置和控制接口 的存储桶通知机制。
注意
要启用存储桶通知 API,rgw_enable_apis配置参数应包含:“通知”。
通知可靠性
通知可以同步或异步发送。本节 描述同步的延迟和可靠性 和异步通知。
同步通知
通知可以同步发送,作为操作的一部分 触发了它们。在此模式下,操作仅在以下之后确认(确认) 通知将发送到主题的已配置终端节点。这意味着 通知的往返时间(发送 通知主题的终端节点以及接收 确认)被添加到操作本身的延迟中。
注意
即使原始触发操作被视为成功,即使 通知失败并显示错误、无法传递或超时。
异步通知
通知可以异步发送。他们致力于坚持不懈 存储,然后异步发送到主题的已配置终端节点。在 在这种情况下,添加到原始操作的唯一延迟是延迟 在将通知提交到持久存储时添加。
注意
如果通知失败并显示错误、无法传递或 超时,将重试,直到成功确认为止。
提示
为了最大限度地减少异步通知增加的延迟,我们 建议将“日志”池放在快速媒体上。
通过 CLI 进行主题管理
通过运行 以下命令:
radosgw-admin topic list [--tenant={
tenant}]
通过运行以下命令获取特定主题的配置:
radosgw-admin topic get --topic={
topic-name} [--tenant={
tenant}]
通过运行以下命令删除主题:
radosgw-admin topic rm --topic={
topic-name} [--tenant={
tenant}]
通知性能统计
-
pubsub_event_triggered
:至少有一个主题与之关联的事件的运行计数器 -
pubsub_event_lost
:具有与其关联的主题但未推送到任何终结点的事件的运行计数器 -
pubsub_push_ok
:成功推送到其终结点的事件的所有通知的运行计数器 -
pubsub_push_fail
:运行计数器,用于所有通知,用于推送到其终结点的事件 -
pubsub_push_pending
:推送到终结点但尚未确认或裸露的事件的仪表值
注意
pubsub_event_triggered
并按 事件,但每个通知上的每个推送操作都会递增。pubsub_event_lost
pubsub_push_ok
pubsub_push_fail
存储桶通知 REST API
主题
注意
在所有主题操作中,参数都经过 URL 编码并在 使用此内容类型的邮件正文:。application/x-www-form-urlencoded
创建主题
这将创建一个新主题。为主题提供推送终端节点参数, 稍后将在创建通知时使用。响应是 生成。成功的响应包括主题的 ARN(“Amazon 资源名称”,用于引用主题的唯一标识符)。 要更新主题,请使用用于创建主题的相同命令(但在 更新时,使用现有主题的名称和不同的终端节点值)。
提示
必须重新创建已与主题关联的任何通知 以便更新主题。
POST
Action=CreateTopic
&Name=<topic-name>
[&Attributes.entry.1.key=amqp-exchange&Attributes.entry.1.value=<exchange>]
[&Attributes.entry.2.key=amqp-ack-level&Attributes.entry.2.value=none|broker|routable]
[&Attributes.entry.3.key=verify-ssl&Attributes.entry.3.value=true|false]
[&Attributes.entry.4.key=kafka-ack-level&Attributes.entry.4.value=none|broker]
[&Attributes.entry.5.key=use-ssl&Attributes.entry.5.value=true|false]
[&Attributes.entry.6.key=ca-location&Attributes.entry.6.value=<file path>]
[&Attributes.entry.7.key=OpaqueData&Attributes.entry.7.value=<opaque data>]
[&Attributes.entry.8.key=push-endpoint&Attributes.entry.8.value=<endpoint>]
[&Attributes.entry.9.key=persistent&Attributes.entry.9.value=true|false]
[&Attributes.entry.10.key=cloudevents&Attributes.entry.10.value=true|false]
[&Attributes.entry.11.key=mechanism&Attributes.entry.11.value=<mechanism>]
请求参数:
-
推送终结点:这是要向其发送推送通知的终结点的 URI。
-
不透明数据:不透明数据在主题配置中设置并添加到所有 由主题触发的通知。
-
持久:这指示发送到此终结点的通知是否为 持久(=异步)或不持久。(默认情况下,这是“假的”。
-
HTTP 终结点
URI:
http[s]://<fqdn>[:<port]
port:对于 HTTP,默认为 80,对于 HTTPS,默认为 443。
verify-ssl:这指示服务器证书是否由 客户端。(默认情况下,这是“true”。
cloudevents:这指示 HTTP 标头是否应包含 属性,具体取决于 S3 云事件规范。(这是“错误的” 默认值。
-
AMQP0.9.1 端点
URI:
amqp[s]://[<user>:<password>@]<fqdn>[:<port>][/<vhost>]
用户/密码:默认为“访客/访客”。
用户/密码:这只能通过 HTTPS 提供。主题创建 否则,请求将被拒绝。
端口:对于未加密的连接,默认为 5672,对于未加密的连接,默认为 5671 SSL 加密连接。
虚拟主机:默认为“/”。
verify-ssl:这指示服务器证书是否由 客户端。(默认情况下,这是“true”。
如果提供并使用安全连接,则 指定的 CA 将用于对代理进行身份验证。默认 CA 将 不使用。
ca-location
amqp-exchange:交换必须存在,并且必须能够路由消息 基于主题。此参数是必需的。
amqp-ack-level:不需要end2end acking。消息可能会保留在 经纪人在交付到最终目的地之前。三种确认方法 存在:
“none”:如果消息被发送到代理,则被视为“已送达”。
“代理”:如果消息被代理确认,则消息被视为“已传递”(默认)。
“routable”: The message is considered “delivered” if the broker can route to a consumer.
Tip
The topic-name (see Create a Topic) is used for the AMQP topic (“routing key” for a topic exchange).
-
Kafka endpoint
URI:
kafka://[<user>:<password>@]<fqdn>[:<port]
use-ssl
: If this is set to “true”, a secure connection is used to connect to the broker. (This is “false” by default.)
ca-location
: If this is provided and a secure connection is used, the specified CA will be used instead of the default CA to authenticate the broker.user/password: This should be provided over HTTPS. If not, the config parameter rgw_allow_notification_secrets_in_cleartext must be true in order to create topics.
user/password: This should be provided together with . If not, the broker credentials will be sent over insecure transport.
use-ssl
mechanism: may be provided together with user/password (default: ). The supported SASL mechanisms are:
PLAIN
PLAIN
SCRAM-SHA-256
SCRAM-SHA-512
GSSAPI
OAUTHBEARER
port: This defaults to 9092.
kafka-ack-level: No end2end acking is required. Messages may persist in the broker before being delivered to their final destinations. Two ack methods exist:
“none”:如果消息发送到代理,则被视为“已送达”。
“代理”:如果消息被代理确认,则被视为“已传递”。(这 是默认值。
注意
-
特定参数的键值对不必驻留在同一参数中 行作为参数,并且不需要以任何特定顺序出现,但它 必须使用相同的索引。
-
属性索引不必是顺序的,也不需要从任何 特定值。
-
AWS 创建主题提供了终端节点的详细说明 属性格式。但是,在我们的例子中,不同的键和值是 使用。
响应具有以下格式:
<CreateTopicResponse xmlns="https://sns.amazonaws.com/doc/2010-03-31/">
<CreateTopicResult>
<TopicArn></TopicArn>
</CreateTopicResult>
<ResponseMetadata>
<RequestId></RequestId>
</ResponseMetadata>
</CreateTopicResponse>
响应中的主题 ARN 具有以下格式:
arn:aws:sns:<zone-group>:<tenant>:<topic>
获取主题属性
这将返回有关特定主题的信息。这包括推送端点 信息(如果提供)。
POST
Action=GetTopicAttributes
&TopicArn=<topic-arn>
响应具有以下格式:
<GetTopicAttributesResponse>
<GetTopicAttributesResult>
<Attributes>
<entry>
<key>User</key>
<value></value>
</entry>
<entry>
<key>Name</key>
<value></value>
</entry>
<entry>
<key>EndPoint</key>
<value></value>
</entry>
<entry>
<key>TopicArn</key>
<value></value>
</entry>
<entry>
<key>OpaqueData</key>
<value></value>
</entry>
</Attributes>
</GetTopicAttributesResult>
<ResponseMetadata>
<RequestId></RequestId>
</ResponseMetadata>
</GetTopicAttributesResponse>
-
用户:创建主题的用户的名称。
-
名称:主题的名称。
-
终结点:JSON 格式的终结点参数,包括:
-
终端节点地址:推送终端节点 URL。
-
端点参数:推送端点参数。
-
端点主题:要发送到端点的主题名称(可以不同 比上述主题名称)。
-
HasStorageSecret:如果端点URL包含用户/密码,则为“true” 信息。在这种情况下,必须通过 HTTPS 发出请求。“主题 否则将拒绝 get“请求。
-
持久:如果主题是持久,则为“true”。
-
-
TopicArn: topic ARN.
-
OpaqueData: The opaque data set on the topic.
GET TOPIC INFORMATION
This returns information about a specific topic. This includes push-endpoint information, if provided. Note that this API is now deprecated in favor of the AWS compliant GetTopicAttributes API.
POST
Action=GetTopic
&TopicArn=<topic-arn>
The response has the following format:
<GetTopicResponse>
<GetTopicResult>
<Topic>
<User></User>
<Name></Name>
<EndPoint>
<EndpointAddress></EndpointAddress>
<EndpointArgs></EndpointArgs>
<EndpointTopic></EndpointTopic>
<HasStoredSecret></HasStoredSecret>
<Persistent></Persistent>
</EndPoint>
<TopicArn></TopicArn>
<OpaqueData></OpaqueData>
</Topic>
</GetTopicResult>
<ResponseMetadata>
<RequestId></RequestId>
</ResponseMetadata>
</GetTopicResponse>
-
User: The name of the user that created the topic.
-
Name: The name of the topic.
-
EndpointAddress: The push-endpoint URL.
-
EndpointArgs: The push-endpoint args.
-
EndpointTopic: The topic name to be sent to the endpoint (which can be different than the above topic name).
-
HasStoredSecret: This is “true” if the endpoint URL contains user/password information. In this case, the request must be made over HTTPS. The “topic get” request will otherwise be rejected.
-
Persistent: “true” if topic is persistent.
-
TopicArn: topic ARN.
-
OpaqueData: the opaque data set on the topic.
DELETE TOPIC
POST
Action=DeleteTopic
&TopicArn=<topic-arn>
This deletes the specified topic.
Note
-
Deleting an unknown notification (for example, double delete) is not considered an error.
-
Deleting a topic does not automatically delete all notifications associated with it.
响应具有以下格式:
<DeleteTopicResponse xmlns="https://sns.amazonaws.com/doc/2010-03-31/">
<ResponseMetadata>
<RequestId></RequestId>
</ResponseMetadata>
</DeleteTopicResponse>
列出主题
列出与租户关联的所有主题。
POST
Action=ListTopics
响应具有以下格式:
<ListTopicsResponse xmlns="https://sns.amazonaws.com/doc/2010-03-31/">
<ListTopicsResult>
<Topics>
<member>
<User></User>
<Name></Name>
<EndPoint>
<EndpointAddress></EndpointAddress>
<EndpointArgs></EndpointArgs>
<EndpointTopic></EndpointTopic>
</EndPoint>
<TopicArn></TopicArn>
<OpaqueData></OpaqueData>
</member>
</Topics>
</ListTopicsResult>
<ResponseMetadata>
<RequestId></RequestId>
</ResponseMetadata>
</ListTopicsResponse>
-
如果终结点 URL 在 主题,则必须通过 HTTPS 发出请求。“主题列表”请求将 否则被拒绝。
通知
详见:存储桶操作。
注意
-
“中止分段上传”请求不发出通知
-
“启动分段上传”和“POST 对象”请求都将发出通知
s3:ObjectCreated:Post
事件
事件采用 JSON 格式(无论实际终端节点如何),并且与 S3 兼容。 例如:
{
"Records":[
{
"eventVersion":"2.1",
"eventSource":"ceph:s3",
"awsRegion":"zonegroup1",
"eventTime":"2019-11-22T13:47:35.124724Z",
"eventName":"ObjectCreated:Put",
"userIdentity":{
"principalId":"tester"
},
"requestParameters":{
"sourceIPAddress":""
},
"responseElements":{
"x-amz-request-id":"503a4c37-85eb-47cd-8681-2817e80b4281.5330.903595",
"x-amz-id-2":"14d2-zone1-zonegroup1"
},
"s3":{
"s3SchemaVersion":"1.0",
"configurationId":"mynotif1",
"bucket":{
"name":"mybucket1",
"ownerIdentity":{
"principalId":"tester"
},
"arn":"arn:aws:s3:zonegroup1::mybucket1",
"id":"503a4c37-85eb-47cd-8681-2817e80b4281.5332.38"
},
"object":{
"key":"myimage1.jpg",
"size":"1024",
"eTag":"37b51d194a7513e45b56f6524f2d51f2",
"versionId":"",
"sequencer": "F7E6D75DC742D108",
"metadata":[],
"tags":[]
}
},
"eventId":"",
"opaqueData":"[email protected]"
}
]}
-
awsRegion:区域组。
-
eventTime:时间戳,指示触发事件的时间。
-
eventName:有关支持的事件列表,请参阅: S3 通知 兼容性。请注意,事件名称值不以 s3: 前缀开头。
-
userIdentity.principalId:触发更改的用户。
-
requestParameters.sourceIPAddress: 不支持
-
responseElements.x-amz-request-id:原始更改的请求 ID。
-
responseElements.x_amz_id_2:进行更改的 RGW。
-
s3.configurationId:创建事件的通知 ID。
-
s3.bucket.name:存储桶的名称。
-
s3.bucket.ownerIdentity.principalId:存储桶的拥有者。
-
s3.bucket.arn:存储桶的 ARN。
-
s3.bucket.id:存储桶的 ID。(这是 S3 的扩展 通知 API。
-
s3.object.key:对象键。
-
s3.object.size:对象大小。
-
s3.object.eTag:对象 etag。
-
s3.object.versionId:对象版本(如果存储桶已版本控制)。当一个 进行复制,它包括目标对象的版本。当删除 标记已创建,它包括删除标记的版本。
-
s3.object.sequencer:“变化”的单调递增标识符 每个对象“(十六进制格式)。
-
s3.object.metadata:在作为 S3 通知 API 的扩展)。
x-amz-meta-
-
s3.object.tags:在对象上设置的任何标签。(这是 S3 的扩展 通知 API。
-
s3.eventId:事件的唯一 ID,可用于确认。(这 是 S3 通知 API 的扩展。
-
s3.opaqueData:这意味着在主题配置中设置了“不透明数据” 并添加到主题触发的所有通知中。(这是一个 S3 通知 API 的扩展。