Add Kafka RestProxy APIs (#1262)

* Add Kafka RestProxy APIs Signed-off-by: obaydullahmhs <[email protected]>
kubedb · Jul 26, 2024 · 88b6087 · 88b6087
1 parent 4af05fd
commit 88b6087
Show file tree

Hide file tree

Showing 22 changed files with 4,800 additions and 10 deletions.
diff --git a/apis/kafka/v1alpha1/connect_cluster_webhook.go b/apis/kafka/v1alpha1/connect_cluster_webhook.go
@@ -59,7 +59,7 @@ func (k *ConnectCluster) ValidateCreate() (admission.Warnings, error) {
 	if len(allErr) == 0 {
 		return nil, nil
 	}
-	return nil, apierrors.NewInvalid(schema.GroupKind{Group: "kafka.kubedb.com", Kind: "Kafka"}, k.Name, allErr)
+	return nil, apierrors.NewInvalid(schema.GroupKind{Group: "kafka.kubedb.com", Kind: "ConnectCluster"}, k.Name, allErr)
 }
 
 // ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
@@ -87,9 +87,9 @@ func (k *ConnectCluster) ValidateDelete() (admission.Warnings, error) {
 
 	var allErr field.ErrorList
 	if k.Spec.DeletionPolicy == dbapi.DeletionPolicyDoNotTerminate {
-		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("terminationPolicy"),
+		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("deletionPolicy"),
 			k.Name,
-			"Can not delete as terminationPolicy is set to \"DoNotTerminate\""))
+			"Can not delete as deletionPolicy is set to \"DoNotTerminate\""))
 		return nil, apierrors.NewInvalid(schema.GroupKind{Group: "kafka.kubedb.com", Kind: "ConnectCluster"}, k.Name, allErr)
 	}
 	return nil, nil
@@ -112,7 +112,7 @@ func (k *ConnectCluster) ValidateCreateOrUpdate() field.ErrorList {
 	}
 
 	if k.Spec.DeletionPolicy == dbapi.DeletionPolicyHalt {
-		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("terminationPolicy"),
+		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("deletionPolicy"),
 			k.Name,
 			"DeletionPolicyHalt is not supported for ConnectCluster"))
 	}

diff --git a/apis/kafka/v1alpha1/connector_webhook.go b/apis/kafka/v1alpha1/connector_webhook.go
@@ -64,9 +64,9 @@ func (k *Connector) ValidateDelete() (admission.Warnings, error) {
 
 	var allErr field.ErrorList
 	if k.Spec.DeletionPolicy == dbapi.DeletionPolicyDoNotTerminate {
-		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("terminationPolicy"),
+		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("deletionPolicy"),
 			k.Name,
-			"Can not delete as terminationPolicy is set to \"DoNotTerminate\""))
+			"Can not delete as deletionPolicy is set to \"DoNotTerminate\""))
 		return nil, apierrors.NewInvalid(schema.GroupKind{Group: "kafka.kubedb.com", Kind: "Connector"}, k.Name, allErr)
 	}
 	return nil, nil
@@ -75,7 +75,7 @@ func (k *Connector) ValidateDelete() (admission.Warnings, error) {
 func (k *Connector) ValidateCreateOrUpdate() error {
 	var allErr field.ErrorList
 	if k.Spec.DeletionPolicy == dbapi.DeletionPolicyHalt {
-		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("terminationPolicy"),
+		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("deletionPolicy"),
 			k.Name,
 			"DeletionPolicyHalt isn't supported for Connector"))
 	}

diff --git a/apis/kafka/v1alpha1/constants.go b/apis/kafka/v1alpha1/constants.go
@@ -111,3 +111,33 @@ const (
 	SchemaRegistryOperatorVolumeConfig = "registry-operator-config"
 	SchemaRegistryOperatorConfigPath   = "/deployments/config"
 )
+
+// RestProxy constants
+
+const (
+	RestProxyPrimaryPortName      = "primary"
+	RestProxyPortName             = "restproxy"
+	RestProxyRESTPort             = 8082
+	RestProxyContainerName        = "rest-proxy"
+	RestProxyOperatorVolumeConfig = "rest-proxy-operator-config"
+	RestProxyOperatorConfigPath   = "/opt/karapace/config"
+
+	RestProxyKarapaceLogLevel          = "log_level"
+	RestProxyKarapaceLogLevelWarning   = "WARNING"
+	RestProxyKarapaceLogLevelInfo      = "INFO"
+	RestProxyKarapaceLogLevelDebug     = "DEBUG"
+	RestProxyKarapaceHostName          = "host"
+	RestProxyKarapacePortName          = "port"
+	RestProxyKafkaBootstrapURI         = "bootstrap_uri"
+	RestProxyKafkaSecurityProtocolName = "security_protocol"
+	RestProxyKafkaSASLMechanismName    = "sasl_mechanism"
+	RestProxyKafkaSASLUsername         = "sasl_plain_username"
+	RestProxyKafkaSASLPassword         = "sasl_plain_password"
+	RestProxyKafkaSSLCAFile            = "ssl_cafile"
+	RestProxyKafkaSSLCertFile          = "ssl_certfile"
+	RestProxyKafkaSSLKeyFile           = "ssl_keyfile"
+	RestProxyKafkaSSLCAFilePath        = "/var/private/kafka-client-ssl/ca.crt"
+	RestProxyKafkaSSLCertFilePath      = "/var/private/kafka-client-ssl/tls.crt"
+	RestProxyKafkaSSLKeyFilePath       = "/var/private/kafka-client-ssl/tls.key"
+	RestProxyConfigFileName            = "rest.config.json"
+)
diff --git a/apis/kafka/v1alpha1/openapi_generated.go b/apis/kafka/v1alpha1/openapi_generated.go
diff --git a/apis/kafka/v1alpha1/register.go b/apis/kafka/v1alpha1/register.go
@@ -58,6 +58,8 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 		&ConnectClusterList{},
 		&Connector{},
 		&ConnectorList{},
+		&RestProxy{},
+		&RestProxyList{},
 		&SchemaRegistry{},
 		&SchemaRegistryList{},
 	)