forked from RedHatInsights/insights-operator-utils
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[CCXDEV-12032] Add common Kafka configuration and clowder-related uti…
…lities (RedHatInsights#393) * Add Kafka package with common broker configuration functions * Add clowder-related common functions * Linting and license header * Small refactor of clowder/clowder_test.go
- Loading branch information
Showing
10 changed files
with
603 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,214 @@ | ||
| // Copyright 2024 Red Hat, Inc | ||
| // | ||
| // 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 clowder_test | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "github.com/RedHatInsights/insights-operator-utils/clowder" | ||
| "github.com/RedHatInsights/insights-operator-utils/kafka" | ||
| "github.com/RedHatInsights/insights-operator-utils/postgres" | ||
| api "github.com/redhatinsights/app-common-go/pkg/api/v1" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/tisnik/go-capture" | ||
| "testing" | ||
| ) | ||
|
|
||
| func TestUseDBConfig(t *testing.T) { | ||
| cfg := postgres.StorageConfiguration{} | ||
| expected := postgres.StorageConfiguration{ | ||
| PGUsername: "username", | ||
| PGPassword: "password", | ||
| PGHost: "hostname", | ||
| PGPort: 1234, | ||
| PGDBName: "dbname", | ||
| } | ||
| loadedCfg := api.AppConfig{ | ||
| Database: &api.DatabaseConfig{ | ||
| AdminPassword: "adminpw", | ||
| AdminUsername: "admin", | ||
| Hostname: "hostname", | ||
| Name: "dbname", | ||
| Password: "password", | ||
| Port: 1234, | ||
| RdsCa: nil, | ||
| Username: "username", | ||
| }, | ||
| } | ||
|
|
||
| clowder.UseDBConfig(&cfg, &loadedCfg) | ||
| assert.Equal(t, expected, cfg, "Clowder database config was not used") | ||
| } | ||
|
|
||
| func TestUseClowderTopicsTopicFound(t *testing.T) { | ||
| originalTopicName := "topic1" | ||
| clowderTopicName := "NewTopicName" | ||
| brokerCfg := kafka.BrokerConfiguration{ | ||
| Topic: originalTopicName, | ||
| } | ||
| kafkaTopics := map[string]api.TopicConfig{ | ||
| originalTopicName: { | ||
| Name: clowderTopicName, | ||
| }, | ||
| "topic2": { | ||
| Name: "AnotherTopicName", | ||
| }, | ||
| } | ||
|
|
||
| clowder.UseClowderTopics(&brokerCfg, kafkaTopics) | ||
| assert.Equal(t, clowderTopicName, brokerCfg.Topic, "Clowder topic name was not used") | ||
| } | ||
|
|
||
| func TestUseClowderTopicsTopicNotFound(t *testing.T) { | ||
| originalTopicName := "topic1" | ||
|
|
||
| brokerCfg := kafka.BrokerConfiguration{ | ||
| Topic: originalTopicName, | ||
| } | ||
| kafkaTopics := map[string]api.TopicConfig{ | ||
| "topic2": { | ||
| Name: "AnotherTopicName", | ||
| }, | ||
| } | ||
|
|
||
| output, _ := capture.StandardOutput(func() { | ||
| clowder.UseClowderTopics(&brokerCfg, kafkaTopics) | ||
| }) | ||
| assert.Equal(t, originalTopicName, brokerCfg.Topic, "topic name should not change") | ||
| assert.Contains(t, output, "warning: no kafka mapping found for topic topic1") | ||
| } | ||
|
|
||
| func TestUseBrokerConfigNoKafkaConfig(t *testing.T) { | ||
| brokerCfg := kafka.BrokerConfiguration{} | ||
| loadedConfig := api.AppConfig{} | ||
|
|
||
| output, _ := capture.StandardOutput(func() { | ||
| clowder.UseBrokerConfig(&brokerCfg, &loadedConfig) | ||
| }) | ||
| assert.Contains(t, output, clowder.NoBrokerCfg) | ||
| } | ||
|
|
||
| func TestUseBrokerConfigNoKafkaBrokers(t *testing.T) { | ||
| brokerCfg := kafka.BrokerConfiguration{} | ||
| loadedConfig := api.AppConfig{ | ||
| Kafka: &api.KafkaConfig{}, | ||
| } | ||
|
|
||
| output, _ := capture.StandardOutput(func() { | ||
| clowder.UseBrokerConfig(&brokerCfg, &loadedConfig) | ||
| }) | ||
| assert.Contains(t, output, clowder.NoBrokerCfg) | ||
| } | ||
|
|
||
| func TestUseBrokerConfigNoAuthNoPort(t *testing.T) { | ||
| addr := "test_broker" | ||
| brokerCfg := kafka.BrokerConfiguration{} | ||
| loadedConfig := api.AppConfig{ | ||
| Kafka: &api.KafkaConfig{ | ||
| Brokers: []api.BrokerConfig{ | ||
| { | ||
| Hostname: addr, | ||
| Port: nil, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| clowder.UseBrokerConfig(&brokerCfg, &loadedConfig) | ||
| assert.Equal(t, addr, brokerCfg.Address) | ||
| } | ||
|
|
||
| func TestUseBrokerConfigNoAuth(t *testing.T) { | ||
| brokerCfg := kafka.BrokerConfiguration{} | ||
| port := 12345 | ||
| addr := "test_broker" | ||
| loadedConfig := api.AppConfig{ | ||
| Kafka: &api.KafkaConfig{ | ||
| Brokers: []api.BrokerConfig{ | ||
| { | ||
| Hostname: addr, | ||
| Port: &port, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| clowder.UseBrokerConfig(&brokerCfg, &loadedConfig) | ||
| assert.Equal(t, fmt.Sprintf("%s:%d", addr, port), brokerCfg.Address) | ||
| } | ||
|
|
||
| func TestUseBrokerConfigAuthEnabledNoSasl(t *testing.T) { | ||
| brokerCfg := kafka.BrokerConfiguration{} | ||
| port := 12345 | ||
| addr := "test_broker" | ||
| authType := api.BrokerConfigAuthtypeSasl | ||
| loadedConfig := api.AppConfig{ | ||
| Kafka: &api.KafkaConfig{ | ||
| Brokers: []api.BrokerConfig{ | ||
| { | ||
| Hostname: addr, | ||
| Port: &port, | ||
| Authtype: &authType, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| output, _ := capture.StandardOutput(func() { | ||
| clowder.UseBrokerConfig(&brokerCfg, &loadedConfig) | ||
| }) | ||
|
|
||
| assert.Equal(t, fmt.Sprintf("%s:%d", addr, port), brokerCfg.Address) | ||
| assert.Contains(t, output, clowder.NoSaslCfg) | ||
| } | ||
|
|
||
| func TestUseBrokerConfigAuthEnabledWithSaslConfig(t *testing.T) { | ||
| brokerCfg := kafka.BrokerConfiguration{} | ||
| port := 12345 | ||
| addr := "test_broker" | ||
| saslUsr := "user" | ||
| saslPwd := "pwd" | ||
| saslMechanism := "sasl" | ||
| protocol := "tls" | ||
|
|
||
| authType := api.BrokerConfigAuthtypeSasl | ||
| loadedConfig := api.AppConfig{ | ||
| Kafka: &api.KafkaConfig{ | ||
| Brokers: []api.BrokerConfig{ | ||
| { | ||
| Hostname: addr, | ||
| Port: &port, | ||
| Authtype: &authType, | ||
| Sasl: &api.KafkaSASLConfig{ | ||
| Password: &saslPwd, | ||
| Username: &saslUsr, | ||
| SaslMechanism: &saslMechanism, | ||
| }, | ||
| SecurityProtocol: &protocol, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| output, _ := capture.StandardOutput(func() { | ||
| clowder.UseBrokerConfig(&brokerCfg, &loadedConfig) | ||
| }) | ||
|
|
||
| assert.Equal(t, fmt.Sprintf("%s:%d", addr, port), brokerCfg.Address) | ||
| assert.Contains(t, output, "kafka is configured to use authentication") | ||
| assert.Equal(t, saslUsr, brokerCfg.SaslUsername) | ||
| assert.Equal(t, saslPwd, brokerCfg.SaslPassword) | ||
| assert.Equal(t, saslMechanism, brokerCfg.SaslMechanism) | ||
| assert.Equal(t, protocol, brokerCfg.SecurityProtocol) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| // Copyright 2024 Red Hat, Inc | ||
| // | ||
| // 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 clowder | ||
|
|
||
| var ( | ||
| NoBrokerCfg = noBrokerConfig | ||
| NoSaslCfg = noSaslConfig | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| // Copyright 2024 Red Hat, Inc | ||
| // | ||
| // 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 clowder | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "github.com/RedHatInsights/insights-operator-utils/kafka" | ||
| api "github.com/redhatinsights/app-common-go/pkg/api/v1" | ||
| ) | ||
|
|
||
| // Common constants used for logging and error reporting | ||
| const ( | ||
| noBrokerConfig = "warning: no broker configurations found in clowder config" | ||
| noSaslConfig = "warning: SASL configuration is missing" | ||
| noTopicMapping = "warning: no kafka mapping found for topic %s" | ||
| ) | ||
|
|
||
| // UseBrokerConfig tries to replace parts of the BrokerConfiguration with the values | ||
| // loaded by Clowder | ||
| func UseBrokerConfig(brokerCfg *kafka.BrokerConfiguration, loadedConfig *api.AppConfig) { | ||
| // make sure broker(s) are configured in Clowder | ||
| if loadedConfig.Kafka != nil && len(loadedConfig.Kafka.Brokers) > 0 { | ||
| broker := loadedConfig.Kafka.Brokers[0] | ||
| // port can be empty in api, so taking it into account | ||
| if broker.Port != nil { | ||
| brokerCfg.Address = fmt.Sprintf("%s:%d", broker.Hostname, *broker.Port) | ||
| } else { | ||
| brokerCfg.Address = broker.Hostname | ||
| } | ||
|
|
||
| // SSL config | ||
| if broker.Authtype != nil { | ||
| fmt.Println("kafka is configured to use authentication") | ||
| if broker.Sasl != nil { | ||
| // we are trusting that these values are set and | ||
| // dereferencing the pointers without any check... | ||
| brokerCfg.SaslUsername = *broker.Sasl.Username | ||
| brokerCfg.SaslPassword = *broker.Sasl.Password | ||
| brokerCfg.SaslMechanism = *broker.Sasl.SaslMechanism | ||
| brokerCfg.SecurityProtocol = *broker.SecurityProtocol | ||
|
|
||
| if caPath, err := loadedConfig.KafkaCa(broker); err == nil { | ||
| brokerCfg.CertPath = caPath | ||
| } | ||
| } else { | ||
| fmt.Println(noSaslConfig) | ||
| } | ||
| } | ||
| } else { | ||
| fmt.Println(noBrokerConfig) | ||
| } | ||
| } | ||
|
|
||
| // UseClowderTopics tries to replace the configured topic with the corresponding | ||
| // topic loaded by Clowder | ||
| func UseClowderTopics(configuration *kafka.BrokerConfiguration, kafkaTopics map[string]api.TopicConfig) { | ||
| // Get the correct topic name from clowder mapping if available | ||
| if clowderTopic, ok := kafkaTopics[configuration.Topic]; ok { | ||
| configuration.Topic = clowderTopic.Name | ||
| } else { | ||
| fmt.Printf(noTopicMapping, configuration.Topic) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| // Copyright 2024 Red Hat, Inc | ||
| // | ||
| // 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 clowder | ||
|
|
||
| import ( | ||
| "github.com/RedHatInsights/insights-operator-utils/postgres" | ||
| api "github.com/redhatinsights/app-common-go/pkg/api/v1" | ||
| ) | ||
|
|
||
| // UseDBConfig tries to replace the StorageConfiguration parameters with the | ||
| // values loaded by Clowder | ||
| func UseDBConfig(storageCfg *postgres.StorageConfiguration, loadedConfig *api.AppConfig) { | ||
| storageCfg.PGDBName = loadedConfig.Database.Name | ||
| storageCfg.PGHost = loadedConfig.Database.Hostname | ||
| storageCfg.PGPort = loadedConfig.Database.Port | ||
| storageCfg.PGUsername = loadedConfig.Database.Username | ||
| storageCfg.PGPassword = loadedConfig.Database.Password | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.