Add AMQP client Getter for RabbitMQ

go.mod

@@ -17,14 +17,15 @@ require (
 	github.com/opensearch-project/opensearch-go v1.1.0
 	github.com/opensearch-project/opensearch-go/v2 v2.3.0
 	github.com/pkg/errors v0.9.1
+	github.com/rabbitmq/amqp091-go v1.9.0
 	github.com/redis/go-redis/v9 v9.5.1
 	go.mongodb.org/mongo-driver v1.14.0
 	k8s.io/api v0.29.2
 	k8s.io/apimachinery v0.29.2
 	k8s.io/klog/v2 v2.120.1
 	kmodules.xyz/client-go v0.29.13
 	kmodules.xyz/custom-resources v0.29.1
-	kubedb.dev/apimachinery v0.44.0
+	kubedb.dev/apimachinery v0.44.1-0.20240418111017-ee6ba81797f6
 	sigs.k8s.io/controller-runtime v0.17.2
 	xorm.io/xorm v1.3.6
 )

go.sum

@@ -305,6 +305,8 @@ github.com/prometheus/common v0.45.0 h1:2BGz0eBc2hdMDLnO/8n0jeB3oPrt2D08CekT0lne
 github.com/prometheus/common v0.45.0/go.mod h1:YJmSTw9BoKxJplESWWxlbyttQR4uaEcGyv9MZjVOJsY=
 github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
 github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
+github.com/rabbitmq/amqp091-go v1.9.0 h1:qrQtyzB4H8BQgEuJwhmVQqVHB9O4+MNDJCCAcpc3Aoo=
+github.com/rabbitmq/amqp091-go v1.9.0/go.mod h1:+jPrT9iY2eLjRaMSRHUhc3z14E/l85kv/f+6luSD3pc=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/redis/go-redis/v9 v9.5.1 h1:H1X4D3yHPaYrkL5X06Wh6xNVM/pX0Ft4RV0vMGvLBh8=
@@ -375,6 +377,7 @@ go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
 go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
+go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
@@ -582,8 +585,8 @@ kmodules.xyz/monitoring-agent-api v0.29.0 h1:gpFl6OZrlMLb/ySMHdREI9EwGtnJ91oZBn9
 kmodules.xyz/monitoring-agent-api v0.29.0/go.mod h1:iNbvaMTgVFOI5q2LJtGK91j4Dmjv4ZRiRdasGmWLKQI=
 kmodules.xyz/offshoot-api v0.29.0 h1:GHLhxxT9jU1N8+FvOCCeJNyU5g0duYS46UGrs6AHNLY=
 kmodules.xyz/offshoot-api v0.29.0/go.mod h1:5NxhBblXoDHWStx9HCDJR2KFTwYjEZ7i1Id3jelIunw=
-kubedb.dev/apimachinery v0.44.0 h1:R7tiR35tmjJxNlpHUS8PVPmSrnFMfLycDE0c9XEU1/A=
-kubedb.dev/apimachinery v0.44.0/go.mod h1:7daaaWragCFLV38plrrJtsOuzinBSX3enMpliqlm3Uo=
+kubedb.dev/apimachinery v0.44.1-0.20240418111017-ee6ba81797f6 h1:VrQhjNOGtqmMiT7lMxIydlfhuL8Ya+uitzyvBbvToQ4=
+kubedb.dev/apimachinery v0.44.1-0.20240418111017-ee6ba81797f6/go.mod h1:7daaaWragCFLV38plrrJtsOuzinBSX3enMpliqlm3Uo=
 kubeops.dev/petset v0.0.5 h1:VVXi39JhjondlbHyZ98z0MLp6VCmiCMinL59K48Y2zA=
 kubeops.dev/petset v0.0.5/go.mod h1:ijtKT1HlAht2vBEZj5LW7C00XEs3B0d1VdCQgd5V4cA=
 lukechampine.com/uint128 v1.1.1/go.mod h1:c4eWIwlEGaxC/+H1VguhU4PHXNWDCDMUlWdIWl2j1gk=

rabbitmq/client.go

@@ -0,0 +1,25 @@
+/*
+Copyright AppsCode Inc. and Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package rabbitmq
+
+import (
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+type Client struct {
+	*amqp.Connection
+}

rabbitmq/kubedb_client_builder.go

@@ -0,0 +1,102 @@
+/*
+Copyright AppsCode Inc. and Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package rabbitmq
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+	core "k8s.io/api/core/v1"
+	kerr "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/klog/v2"
+	api "kubedb.dev/apimachinery/apis/kubedb/v1alpha2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type KubeDBClientBuilder struct {
+	kc      client.Client
+	db      *api.RabbitMQ
+	url     string
+	podName string
+	ctx     context.Context
+}
+
+func NewKubeDBClientBuilder(kc client.Client, db *api.RabbitMQ) *KubeDBClientBuilder {
+	return &KubeDBClientBuilder{
+		kc: kc,
+		db: db,
+	}
+}
+
+func (o *KubeDBClientBuilder) WithPod(podName string) *KubeDBClientBuilder {
+	o.podName = podName
+	return o
+}
+
+func (o *KubeDBClientBuilder) WithURL(url string) *KubeDBClientBuilder {
+	o.url = url
+	return o
+}
+
+func (o *KubeDBClientBuilder) WithContext(ctx context.Context) *KubeDBClientBuilder {
+	o.ctx = ctx
+	return o
+}
+
+func (o *KubeDBClientBuilder) GetAMQPconnURL(username string, password string) string {
+	return fmt.Sprintf("amqp://%s:%s@%s.%s.svc.cluster.local:%d/", username, password, o.db.OffshootName(), o.db.Namespace, api.RabbitMQAMQPPort)
+}
+
+func (o *KubeDBClientBuilder) GetRabbitMQClient() (*Client, error) {
+	authSecret := &core.Secret{}
+	if !o.db.Spec.DisableSecurity {
+		if o.db.Spec.AuthSecret == nil {
+			klog.Info("Auth-secret not set")
+			return nil, errors.New("auth-secret is not set")
+		}
+		err := o.kc.Get(o.ctx, types.NamespacedName{
+			Namespace: o.db.Namespace,
+			Name:      o.db.Spec.AuthSecret.Name,
+		}, authSecret)
+		if err != nil {
+			if kerr.IsNotFound(err) {
+				klog.Error(err, "Auth-secret not found")
+				return nil, errors.New("auth-secret is not found")
+			}
+			klog.Error(err, "Failed to get auth-secret")
+			return nil, err
+		}
+	}
+	dbConnURL := o.GetAMQPconnURL(string(authSecret.Data[core.BasicAuthUsernameKey]), string(authSecret.Data[core.BasicAuthPasswordKey]))
+
+	rabbitConnection, err := amqp.DialConfig(dbConnURL, amqp.Config{
+		Vhost:  "/",
+		Locale: "en_US",
+	})
+	if err != nil {
+		klog.Error(err, "Failed to connect to rabbitmq")
+		return nil, err
+	}
+	klog.Info("Successfully created client for db")
+
+	return &Client{
+		Connection: rabbitConnection,
+	}, nil
+}