rgw: support kafka auth mechanism parameter

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 <r.dahlen@gmail.com>

Documentation/CRDs/specification.md

@@ -7494,6 +7494,18 @@ string
 <p>The ack level required for this topic (none/broker)</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>mechanism</code><br/>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>The authentication mechanism for this topic (PLAIN/SCRAM-SHA-512/SCRAM-SHA-256/GSSAPI/OAUTHBEARER)</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="ceph.rook.io/v1.KerberosConfigFiles">KerberosConfigFiles

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.

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

deploy/examples/bucket-topic.yaml

@@ -25,3 +25,4 @@ spec:
     #   disableVerifySSL: true
     #   ackLevel: broker # none, broker (default)
     #   useSSL: false
+    #   mechanism: PLAIN

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

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",
 				},
 			},
 		},

pkg/apis/ceph.rook.io/v1/types.go

@@ -2349,6 +2349,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

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

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,
 					},
 				},
 			},