Skip to content

Commit db011c5

Browse files
tapojit047tapojit047
authored
Add TLS config to druid client (#139)
* Add TLS config to druid client Signed-off-by: Tapajit Chandra Paul <tapajit@appscode.com>
1 parent fbbd601 commit db011c5

File tree

64 files changed

+2107
-437
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

64 files changed

+2107
-437
lines changed

‎druid/client.go

Lines changed: 24 additions & 102 deletions
Original file line numberDiff line numberDiff line change
@@ -17,14 +17,11 @@ limitations under the License.
1717
package druid
1818

1919
import (
20-
"bytes"
2120
"encoding/json"
2221
"fmt"
23-
"log"
2422
"time"
2523

2624
druidgo "github.com/grafadruid/go-druid"
27-
"github.com/hashicorp/go-retryablehttp"
2825
"github.com/pkg/errors"
2926
"k8s.io/klog/v2"
3027
health "kmodules.xyz/client-go/tools/healthchecker"
@@ -117,55 +114,15 @@ func (c *Client) CheckDataSourceExistence() (bool, error) {
117114
return false, errors.Wrap(err, "failed to marshal json response")
118115
}
119116
rawMessage := json.RawMessage(jsonData)
120-
response, err := c.SubmitRequest(method, path, rawMessage)
121-
if err != nil {
122-
return false, err
123-
}
124-
125-
exists, err := parseDatasourceExistenceQueryResponse(response)
126-
if err != nil {
127-
return false, errors.Wrap(err, "Failed to parse response of datasource existence request")
128-
}
129-
130-
if err := closeResponse(response); err != nil {
131-
return exists, err
132-
}
133-
return exists, nil
134-
}
135-
136-
func (c *Client) SubmitRequest(method, path string, opts interface{}) (*druidgo.Response, error) {
137-
res, err := c.NewRequest(method, path, opts)
138-
if err != nil {
139-
return nil, errors.Wrap(err, "failed to submit API request")
140-
}
141-
http := retryablehttp.NewClient()
142-
143-
var b []byte
144-
buf := bytes.NewBuffer(b)
145-
http.Logger = log.New(buf, "", 0)
146117

147-
resp, err := http.Do(res)
118+
var result []map[string]interface{}
119+
_, err = c.ExecuteRequest(method, path, rawMessage, &result)
148120
if err != nil {
149-
return nil, err
150-
}
151-
response := &druidgo.Response{Response: resp}
152-
return response, nil
153-
}
154-
155-
func parseDatasourceExistenceQueryResponse(res *druidgo.Response) (bool, error) {
156-
var responseBody []map[string]interface{}
157-
if err := json.NewDecoder(res.Body).Decode(&responseBody); err != nil {
158-
return false, errors.Wrap(err, "failed to deserialize the response")
121+
klog.Error("Failed to execute request", err)
122+
return false, err
159123
}
160-
return len(responseBody) != 0, nil
161-
}
162124

163-
func closeResponse(response *druidgo.Response) error {
164-
err := response.Body.Close()
165-
if err != nil {
166-
return errors.Wrap(err, "Failed to close the response body")
167-
}
168-
return nil
125+
return len(result) > 0, nil
169126
}
170127

171128
// CheckDBReadWriteAccess checks read and write access in the DB
@@ -238,41 +195,25 @@ func (c *Client) GetData() (string, error) {
238195
func (c *Client) runSelectQuery() (string, error) {
239196
method := "POST"
240197
path := "druid/v2/sql"
241-
242198
data := map[string]interface{}{
243199
"query": "SELECT * FROM \"kubedb-datasource\"",
244200
}
201+
245202
jsonData, err := json.Marshal(data)
246203
if err != nil {
247204
return "", errors.Wrap(err, "failed to marshal query json data")
248205
}
249206
rawMessage := json.RawMessage(jsonData)
250-
response, err := c.SubmitRequest(method, path, rawMessage)
251-
if err != nil {
252-
return "", err
253-
}
254-
if response == nil {
255-
return "", errors.New("response body is empty")
256-
}
257207

258-
id, err := parseSelectQueryResponse(response, "id")
208+
var result []map[string]interface{}
209+
_, err = c.ExecuteRequest(method, path, rawMessage, &result)
259210
if err != nil {
260-
return "", errors.Wrap(err, "failed to parse the response body")
261-
}
262-
263-
if err := closeResponse(response); err != nil {
211+
klog.Error("Failed to execute POST query request", err)
264212
return "", err
265213
}
266-
return id.(string), nil
267-
}
214+
id := result[0]["id"]
268215

269-
func parseSelectQueryResponse(res *druidgo.Response, key string) (interface{}, error) {
270-
var responseBody []map[string]interface{}
271-
if err := json.NewDecoder(res.Body).Decode(&responseBody); err != nil {
272-
return "", errors.Wrap(err, "failed to deserialize the response")
273-
}
274-
value := responseBody[0][key]
275-
return value, nil
216+
return id.(string), nil
276217
}
277218

278219
func (c *Client) updateCoordinatorsWaitBeforeDeletingConfig(value int32) error {
@@ -296,11 +237,9 @@ func (c *Client) updateCoordinatorDynamicConfig(data map[string]interface{}) err
296237
}
297238
rawMessage := json.RawMessage(jsonData)
298239

299-
response, err := c.SubmitRequest(method, path, rawMessage)
240+
_, err = c.ExecuteRequest(method, path, rawMessage, nil)
300241
if err != nil {
301-
return err
302-
}
303-
if err := closeResponse(response); err != nil {
242+
klog.Error("Failed to execute coordinator config update request", err)
304243
return err
305244
}
306245
return nil
@@ -336,33 +275,19 @@ func (c *Client) submitTask(taskType DruidTaskType, dataSource string, data stri
336275
} else {
337276
task = GetKillTaskDefinition()
338277
}
339-
340278
rawMessage := json.RawMessage(task)
341279
method := "POST"
342280
path := "druid/indexer/v1/task"
343281

344-
response, err := c.SubmitRequest(method, path, rawMessage)
345-
if err != nil {
346-
return "", err
347-
}
348-
349-
taskID, err := GetValueFromClusterResponse(response, "task")
282+
var result map[string]interface{}
283+
_, err := c.ExecuteRequest(method, path, rawMessage, &result)
350284
if err != nil {
351-
return "", errors.Wrap(err, "failed to parse response of task api request")
352-
}
353-
if err = closeResponse(response); err != nil {
285+
klog.Error("Failed to execute POST ingestion or kill task request", err)
354286
return "", err
355287
}
356-
return fmt.Sprintf("%v", taskID), nil
357-
}
358288

359-
func GetValueFromClusterResponse(res *druidgo.Response, key string) (interface{}, error) {
360-
responseBody := make(map[string]interface{})
361-
if err := json.NewDecoder(res.Body).Decode(&responseBody); err != nil {
362-
return "", errors.Wrap(err, "failed to deserialize the response")
363-
}
364-
value := responseBody[key]
365-
return value, nil
289+
taskID := result["task"]
290+
return taskID.(string), nil
366291
}
367292

368293
func GetIngestionTaskDefinition(dataSource string, data string) string {
@@ -419,21 +344,18 @@ func GetKillTaskDefinition() string {
419344
func (c *Client) CheckTaskStatus(taskID string) (bool, error) {
420345
method := "GET"
421346
path := fmt.Sprintf("druid/indexer/v1/task/%s/status", taskID)
422-
response, err := c.SubmitRequest(method, path, nil)
423-
if err != nil {
424-
return false, errors.Wrap(err, "failed to check task status")
425-
}
426347

427-
statusRes, err := GetValueFromClusterResponse(response, "status")
348+
var result map[string]interface{}
349+
_, err := c.ExecuteRequest(method, path, nil, &result)
428350
if err != nil {
429-
return false, errors.Wrap(err, "failed to parse respons of task ingestion request")
351+
klog.Error("Failed to execute GET task status request", err)
352+
return false, err
430353
}
354+
355+
statusRes := result["status"]
431356
statusMap := statusRes.(map[string]interface{})
432357
status := statusMap["status"].(string)
433358

434-
if err = closeResponse(response); err != nil {
435-
return false, err
436-
}
437359
return status == "SUCCESS", nil
438360
}
439361

‎druid/kubedb_client_builder.go

Lines changed: 92 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -18,15 +18,19 @@ package druid
1818

1919
import (
2020
"context"
21+
"crypto/tls"
22+
"crypto/x509"
2123
"errors"
2224
"fmt"
25+
"net/http"
2326

2427
druidgo "github.com/grafadruid/go-druid"
2528
_ "github.com/lib/pq"
2629
core "k8s.io/api/core/v1"
2730
kerr "k8s.io/apimachinery/pkg/api/errors"
2831
"k8s.io/apimachinery/pkg/types"
2932
"k8s.io/klog/v2"
33+
dbapi "kubedb.dev/apimachinery/apis/kubedb/v1"
3034
olddbapi "kubedb.dev/apimachinery/apis/kubedb/v1alpha2"
3135
"sigs.k8s.io/controller-runtime/pkg/client"
3236
)
@@ -74,28 +78,24 @@ func (o *KubeDBClientBuilder) WithNodeRole(nodeRole olddbapi.DruidNodeRoleType)
7478

7579
func (o *KubeDBClientBuilder) GetDruidClient() (*Client, error) {
7680
var druidOpts []druidgo.ClientOption
77-
if !*o.db.Spec.DisableSecurity {
78-
if o.db.Spec.AuthSecret == nil {
79-
klog.Error("AuthSecret not set")
80-
return nil, errors.New("auth-secret is not set")
81+
// Add druid auth credential to the client
82+
if !o.db.Spec.DisableSecurity {
83+
authOpts, err := o.getClientAuthOpts()
84+
if err != nil {
85+
klog.Error(err, "failed to get client auth options")
86+
return nil, err
8187
}
88+
druidOpts = append(druidOpts, *authOpts)
89+
}
8290

83-
authSecret := &core.Secret{}
84-
err := o.kc.Get(o.ctx, types.NamespacedName{
85-
Namespace: o.db.Namespace,
86-
Name: o.db.Spec.AuthSecret.Name,
87-
}, authSecret)
91+
// Add druid ssl configs to the client
92+
if o.db.Spec.EnableSSL {
93+
sslOpts, err := o.getClientSSLConfig()
8894
if err != nil {
89-
if kerr.IsNotFound(err) {
90-
klog.Error(err, "AuthSecret not found")
91-
return nil, errors.New("auth-secret not found")
92-
}
95+
klog.Error(err, "failed to get client ssl options")
9396
return nil, err
9497
}
95-
userName := string(authSecret.Data[core.BasicAuthUsernameKey])
96-
password := string(authSecret.Data[core.BasicAuthPasswordKey])
97-
98-
druidOpts = append(druidOpts, druidgo.WithBasicAuth(userName, password))
98+
druidOpts = append(druidOpts, *sslOpts)
9999
}
100100

101101
druidClient, err := druidgo.NewClient(o.url, druidOpts...)
@@ -107,8 +107,82 @@ func (o *KubeDBClientBuilder) GetDruidClient() (*Client, error) {
107107
}, nil
108108
}
109109

110+
func (o *KubeDBClientBuilder) getClientAuthOpts() (*druidgo.ClientOption, error) {
111+
if o.db.Spec.AuthSecret == nil {
112+
klog.Error("AuthSecret not set")
113+
return nil, errors.New("auth-secret is not set")
114+
}
115+
116+
authSecret := &core.Secret{}
117+
err := o.kc.Get(o.ctx, types.NamespacedName{
118+
Namespace: o.db.Namespace,
119+
Name: o.db.Spec.AuthSecret.Name,
120+
}, authSecret)
121+
if err != nil {
122+
if kerr.IsNotFound(err) {
123+
klog.Error(err, "AuthSecret not found")
124+
return nil, errors.New("auth-secret not found")
125+
}
126+
return nil, err
127+
}
128+
userName := string(authSecret.Data[core.BasicAuthUsernameKey])
129+
password := string(authSecret.Data[core.BasicAuthPasswordKey])
130+
131+
druidAuthOpts := druidgo.WithBasicAuth(userName, password)
132+
return &druidAuthOpts, nil
133+
}
134+
135+
func (o *KubeDBClientBuilder) getClientSSLConfig() (*druidgo.ClientOption, error) {
136+
certSecret := &core.Secret{}
137+
err := o.kc.Get(o.ctx, types.NamespacedName{
138+
Namespace: o.db.Namespace,
139+
Name: o.db.GetCertSecretName(olddbapi.DruidClientCert),
140+
}, certSecret)
141+
if err != nil {
142+
if kerr.IsNotFound(err) {
143+
klog.Error(err, "Client certificate secret not found")
144+
return nil, errors.New("client certificate secret is not found")
145+
}
146+
klog.Error(err, "Failed to get client certificate Secret")
147+
return nil, err
148+
}
149+
150+
// get tls cert, clientCA and rootCA for tls config
151+
clientCA := x509.NewCertPool()
152+
rootCA := x509.NewCertPool()
153+
154+
crt, err := tls.X509KeyPair(certSecret.Data[core.TLSCertKey], certSecret.Data[core.TLSPrivateKeyKey])
155+
if err != nil {
156+
klog.Error(err, "Failed to parse private key pair")
157+
return nil, err
158+
}
159+
clientCA.AppendCertsFromPEM(certSecret.Data[dbapi.TLSCACertFileName])
160+
rootCA.AppendCertsFromPEM(certSecret.Data[dbapi.TLSCACertFileName])
161+
162+
httpClient := &http.Client{
163+
Transport: &http.Transport{
164+
TLSClientConfig: &tls.Config{
165+
Certificates: []tls.Certificate{crt},
166+
ClientAuth: tls.RequireAndVerifyClientCert,
167+
ClientCAs: clientCA,
168+
RootCAs: rootCA,
169+
MaxVersion: tls.VersionTLS12,
170+
},
171+
},
172+
}
173+
tlsOpts := druidgo.WithHTTPClient(httpClient)
174+
return &tlsOpts, nil
175+
}
176+
110177
// GetNodesAddress returns DNS for the nodes based on type of the node
111178
func (o *KubeDBClientBuilder) GetNodesAddress() string {
112-
baseUrl := fmt.Sprintf("http://%s-0.%s.%s.svc.cluster.local:%d", o.db.PetSetName(o.nodeRole), o.db.GoverningServiceName(), o.db.Namespace, o.db.DruidNodeContainerPort(o.nodeRole))
179+
var scheme string
180+
if o.db.Spec.EnableSSL {
181+
scheme = "https"
182+
} else {
183+
scheme = "http"
184+
}
185+
186+
baseUrl := fmt.Sprintf("%s://%s-0.%s.%s.svc.cluster.local:%d", scheme, o.db.PetSetName(o.nodeRole), o.db.GoverningServiceName(), o.db.Namespace, o.db.DruidNodeContainerPort(o.nodeRole))
113187
return baseUrl
114188
}

‎go.mod

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@ require (
1818
github.com/go-sql-driver/mysql v1.8.1
1919
github.com/gocql/gocql v1.6.0
2020
github.com/grafadruid/go-druid v0.0.6
21-
github.com/hashicorp/go-retryablehttp v0.7.7
2221
github.com/lib/pq v1.10.7
2322
github.com/michaelklishin/rabbit-hole/v2 v2.16.0
2423
github.com/microsoft/go-mssqldb v1.6.0
@@ -31,9 +30,9 @@ require (
3130
k8s.io/api v0.30.2
3231
k8s.io/apimachinery v0.30.2
3332
k8s.io/klog/v2 v2.130.1
34-
kmodules.xyz/client-go v0.30.17
33+
kmodules.xyz/client-go v0.30.28
3534
kmodules.xyz/custom-resources v0.30.0
36-
kubedb.dev/apimachinery v0.48.1-0.20241003061121-cbe53073e554
35+
kubedb.dev/apimachinery v0.48.1-0.20241025104947-405c179f3f23
3736
sigs.k8s.io/controller-runtime v0.18.4
3837
xorm.io/xorm v1.3.6
3938
)
@@ -79,6 +78,7 @@ require (
7978
github.com/hashicorp/errwrap v1.1.0 // indirect
8079
github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
8180
github.com/hashicorp/go-multierror v1.1.1 // indirect
81+
github.com/hashicorp/go-retryablehttp v0.7.7 // indirect
8282
github.com/hashicorp/go-uuid v1.0.3 // indirect
8383
github.com/imdario/mergo v0.3.16 // indirect
8484
github.com/jcmturner/aescts/v2 v2.0.0 // indirect
@@ -143,7 +143,7 @@ require (
143143
k8s.io/kube-openapi v0.0.0-20240703190633-0aa61b46e8c2 // indirect
144144
k8s.io/utils v0.0.0-20240502163921-fe8a2dddb1d0 // indirect
145145
kmodules.xyz/apiversion v0.2.0 // indirect
146-
kmodules.xyz/monitoring-agent-api v0.30.1 // indirect
146+
kmodules.xyz/monitoring-agent-api v0.30.2 // indirect
147147
kmodules.xyz/offshoot-api v0.30.1 // indirect
148148
kubeops.dev/petset v0.0.7 // indirect
149149
modernc.org/memory v1.5.0 // indirect

0 commit comments

Comments
 (0)