From 98777ece70abfb46b62fedf0d7842bbaf16b2f5a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ragnar=20Dahl=C3=A9n?= Date: Thu, 20 Mar 2025 21:27:31 +0100 Subject: [PATCH] rgw: support kafka auth mechanism parameter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ceph has support [1,2] for configuring which authentication mechanism to use for bucket notifications using Kafka topics. This commit adds the corresponding support to rook. [1]: https://github.com/ceph/ceph/pull/48181/commits/d5dce601f65974a21c83c4b1add036030a75c3c4 [2]: https://tracker.ceph.com/issues/57608 Signed-off-by: Ragnar Dahlén (cherry picked from commit 8bf7c679bb4ecd7aa36742e75c4dd9c23c844c43) --- Documentation/CRDs/specification.md | 12 ++++++++++++ .../ceph-object-bucket-notifications.md | 7 +++++++ deploy/charts/rook-ceph/templates/resources.yaml | 10 ++++++++++ deploy/examples/bucket-topic.yaml | 1 + deploy/examples/crds.yaml | 10 ++++++++++ pkg/apis/ceph.rook.io/v1/topic_test.go | 1 + pkg/apis/ceph.rook.io/v1/types.go | 5 +++++ pkg/operator/ceph/object/topic/provisioner.go | 1 + pkg/operator/ceph/object/topic/provisioner_test.go | 9 ++++++--- 9 files changed, 53 insertions(+), 3 deletions(-) diff --git a/Documentation/CRDs/specification.md b/Documentation/CRDs/specification.md index 26afc868f..6b9ce2cde 100644 --- a/Documentation/CRDs/specification.md +++ b/Documentation/CRDs/specification.md @@ -7506,6 +7506,18 @@ string

The ack level required for this topic (none/broker)

+ + +mechanism
+ +string + + + +(Optional) +

The authentication mechanism for this topic (PLAIN/SCRAM-SHA-512/SCRAM-SHA-256/GSSAPI/OAUTHBEARER)

+ +

KerberosConfigFiles diff --git a/Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-bucket-notifications.md b/Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-bucket-notifications.md index 4df51db96..7653b0947 100644 --- a/Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-bucket-notifications.md +++ b/Documentation/Storage-Configuration/Object-Storage-RGW/ceph-object-bucket-notifications.md @@ -65,6 +65,7 @@ spec: # disableVerifySSL: true [16] # ackLevel: broker [17] # useSSL: false [18] +# mechanism: SCRAM-SHA-512 [19] ``` 1. `name` of the `CephBucketTopic` @@ -100,6 +101,12 @@ spec: + “none”: message is considered “delivered” if sent to broker + “broker”: message is considered “delivered” if acked by broker (default) 18. `useSSL` (optional) indicates that secure connection will be used for connecting with the broker (“false” by default) +19. `mechanism` (optional) select which authentication mechanism to use, one of: + + “PLAIN” (default) + + “SCRAM-SHA-512” + + “SCRAM-SHA-256” + + “GSSAPI” + + “OAUTHBEARER” !!! note In case of Kafka and AMQP, the consumer of the notifications is not required to ack the notifications, since the broker persists the messages before delivering them to their final destinations. diff --git a/deploy/charts/rook-ceph/templates/resources.yaml b/deploy/charts/rook-ceph/templates/resources.yaml index 538968872..ee25880a2 100644 --- a/deploy/charts/rook-ceph/templates/resources.yaml +++ b/deploy/charts/rook-ceph/templates/resources.yaml @@ -960,6 +960,16 @@ spec: disableVerifySSL: description: Indicate whether the server certificate is validated by the client or not type: boolean + mechanism: + default: PLAIN + description: The authentication mechanism for this topic (PLAIN/SCRAM-SHA-512/SCRAM-SHA-256/GSSAPI/OAUTHBEARER) + enum: + - PLAIN + - SCRAM-SHA-512 + - SCRAM-SHA-256 + - GSSAPI + - OAUTHBEARER + type: string uri: description: The URI of the Kafka endpoint to push notification to minLength: 1 diff --git a/deploy/examples/bucket-topic.yaml b/deploy/examples/bucket-topic.yaml index 8fd295745..0174891bb 100644 --- a/deploy/examples/bucket-topic.yaml +++ b/deploy/examples/bucket-topic.yaml @@ -25,3 +25,4 @@ spec: # disableVerifySSL: true # ackLevel: broker # none, broker (default) # useSSL: false + # mechanism: PLAIN diff --git a/deploy/examples/crds.yaml b/deploy/examples/crds.yaml index 655076e4c..f7779e4c0 100644 --- a/deploy/examples/crds.yaml +++ b/deploy/examples/crds.yaml @@ -960,6 +960,16 @@ spec: disableVerifySSL: description: Indicate whether the server certificate is validated by the client or not type: boolean + mechanism: + default: PLAIN + description: The authentication mechanism for this topic (PLAIN/SCRAM-SHA-512/SCRAM-SHA-256/GSSAPI/OAUTHBEARER) + enum: + - PLAIN + - SCRAM-SHA-512 + - SCRAM-SHA-256 + - GSSAPI + - OAUTHBEARER + type: string uri: description: The URI of the Kafka endpoint to push notification to minLength: 1 diff --git a/pkg/apis/ceph.rook.io/v1/topic_test.go b/pkg/apis/ceph.rook.io/v1/topic_test.go index eeb66a5ae..fee59a696 100644 --- a/pkg/apis/ceph.rook.io/v1/topic_test.go +++ b/pkg/apis/ceph.rook.io/v1/topic_test.go @@ -111,6 +111,7 @@ func TestValidateKafkaTopicSpec(t *testing.T) { UseSSL: true, DisableVerifySSL: true, AckLevel: "broker", + Mechanism: "SCRAM-SHA-512", }, }, }, diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go index de3c4800c..3722f3d3c 100755 --- a/pkg/apis/ceph.rook.io/v1/types.go +++ b/pkg/apis/ceph.rook.io/v1/types.go @@ -2364,6 +2364,11 @@ type KafkaEndpointSpec struct { // +kubebuilder:default=broker // +optional AckLevel string `json:"ackLevel,omitempty"` + // The authentication mechanism for this topic (PLAIN/SCRAM-SHA-512/SCRAM-SHA-256/GSSAPI/OAUTHBEARER) + // +kubebuilder:validation:Enum=PLAIN;SCRAM-SHA-512;SCRAM-SHA-256;GSSAPI;OAUTHBEARER + // +kubebuilder:default=PLAIN + // +optional + Mechanism string `json:"mechanism,omitempty"` } // +genclient diff --git a/pkg/operator/ceph/object/topic/provisioner.go b/pkg/operator/ceph/object/topic/provisioner.go index f46f4eb18..1f6c844f9 100644 --- a/pkg/operator/ceph/object/topic/provisioner.go +++ b/pkg/operator/ceph/object/topic/provisioner.go @@ -176,6 +176,7 @@ func createTopicAttributes(topic *cephv1.CephBucketTopic) map[string]*string { attr["kafka-ack-level"] = &topic.Spec.Endpoint.Kafka.AckLevel verifySSL = strconv.FormatBool(!topic.Spec.Endpoint.Kafka.DisableVerifySSL) attr["verify-ssl"] = &verifySSL + attr["mechanism"] = &topic.Spec.Endpoint.Kafka.Mechanism } return attr diff --git a/pkg/operator/ceph/object/topic/provisioner_test.go b/pkg/operator/ceph/object/topic/provisioner_test.go index 23ee17257..798a7b6c9 100644 --- a/pkg/operator/ceph/object/topic/provisioner_test.go +++ b/pkg/operator/ceph/object/topic/provisioner_test.go @@ -101,6 +101,7 @@ func TestTopicAttributesCreation(t *testing.T) { t.Run("test Kafka attributes", func(t *testing.T) { uri := "kafka://my-kafka-service:9092" ackLevel := "broker" + mechanism := "SCRAM-SHA-512" expectedAttrs := map[string]*string{ "OpaqueData": &emptyString, "persistent": &falseString, @@ -108,6 +109,7 @@ func TestTopicAttributesCreation(t *testing.T) { "verify-ssl": &trueString, "kafka-ack-level": &ackLevel, "use-ssl": &trueString, + "mechanism": &mechanism, } bucketTopic := &cephv1.CephBucketTopic{ ObjectMeta: metav1.ObjectMeta{ @@ -122,9 +124,10 @@ func TestTopicAttributesCreation(t *testing.T) { ObjectStoreNamespace: namespace, Endpoint: cephv1.TopicEndpointSpec{ Kafka: &cephv1.KafkaEndpointSpec{ - URI: uri, - AckLevel: ackLevel, - UseSSL: true, + URI: uri, + AckLevel: ackLevel, + UseSSL: true, + Mechanism: mechanism, }, }, },